This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 142d3b14fce Pipe: Cache table-model pattern matches by table (#18648)
142d3b14fce is described below
commit 142d3b14fce9d2dc7787777b52ddb2a1a1a1a5d2
Author: shuwenwei <[email protected]>
AuthorDate: Mon Sep 21 12:08:53 2026 +0800
Pipe: Cache table-model pattern matches by table (#18648)
---
.../org/apache/iotdb/db/auth/AuthorityChecker.java | 6 +-
.../matcher/CachedSchemaPatternMatcher.java | 41 ++++++-----
.../pattern/CachedSchemaPatternMatcherTest.java | 83 ++++++++++++++++++++++
3 files changed, 110 insertions(+), 20 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
index 0593db10840..951444d2028 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
@@ -131,13 +131,15 @@ public class AuthorityChecker {
}
public static boolean invalidateCache(String username, String roleName) {
+ final boolean invalidated =
+ authorityFetcher.get().getAuthorCache().invalidateCache(username,
roleName);
PipeInsertionDataNodeListener.getInstance().invalidateAllCache();
- return authorityFetcher.get().getAuthorCache().invalidateCache(username,
roleName);
+ return invalidated;
}
public static void invalidateAllCache() {
- PipeInsertionDataNodeListener.getInstance().invalidateAllCache();
authorityFetcher.get().getAuthorCache().invalidAllCache();
+ PipeInsertionDataNodeListener.getInstance().invalidateAllCache();
}
public static User getUser(String username) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
index e9d2e98c5d0..d8f427868f0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
@@ -61,7 +61,7 @@ public class CachedSchemaPatternMatcher implements
PipeDataRegionMatcher {
// Use full cache to avoid queue stuck and block insertion
protected final Map<IDeviceID, Set<PipeRealtimeDataRegionSource>>
deviceToSourcesCache;
- protected final Map<Pair<String, IDeviceID>,
Set<PipeRealtimeDataRegionSource>>
+ protected final Map<Pair<String, String>, Set<PipeRealtimeDataRegionSource>>
databaseAndTableToSourcesCache;
public CachedSchemaPatternMatcher() {
@@ -102,7 +102,7 @@ public class CachedSchemaPatternMatcher implements
PipeDataRegionMatcher {
public void invalidateCache() {
lock.writeLock().lock();
try {
- // Will invalidate device cache
+ // The table-model cache also depends on access control, so it must be
invalidated separately.
databaseAndTableToSourcesCache.clear();
} finally {
lock.writeLock().unlock();
@@ -144,6 +144,11 @@ public class CachedSchemaPatternMatcher implements
PipeDataRegionMatcher {
return new Pair<>(matchedSources,
findUnmatchedSources(matchedSources));
}
+ // tableNames is also used for privilege checks on table-model TsFile
events, so it must be
+ // complete even after every source has already matched.
+ final boolean isTableModelTsFileEvent =
+ event.getEvent() instanceof PipeTsFileInsertionEvent
+ && ((PipeTsFileInsertionEvent)
event.getEvent()).isTableModelEvent();
final Set<String> tableNames = new HashSet<>();
for (final Map.Entry<IDeviceID, String[]> entry :
event.getSchemaInfo().entrySet()) {
final IDeviceID deviceID = entry.getKey();
@@ -154,16 +159,17 @@ public class CachedSchemaPatternMatcher implements
PipeDataRegionMatcher {
|| deviceID.getTableName().equals(PATH_ROOT)) {
matchTreeModelEvent(deviceID, entry.getValue(), matchedSources);
} else {
- tableNames.add(deviceID.getTableName());
- matchTableModelEvent(
- event.getEvent() instanceof PipeInsertionEvent
- ? ((PipeInsertionEvent)
event.getEvent()).getTableModelDatabaseName()
- : null,
- deviceID,
- matchedSources);
+ final String tableName = deviceID.getTableName();
+ if (tableNames.add(tableName) && matchedSources.size() <
sources.size()) {
+ final String tableModelDatabaseName =
+ event.getEvent() instanceof PipeInsertionEvent
+ ? ((PipeInsertionEvent)
event.getEvent()).getTableModelDatabaseName()
+ : null;
+ matchTableModelEvent(tableModelDatabaseName, tableName,
matchedSources);
+ }
}
- if (matchedSources.size() == sources.size()) {
+ if (matchedSources.size() == sources.size() &&
!isTableModelTsFileEvent) {
break;
}
}
@@ -171,7 +177,7 @@ public class CachedSchemaPatternMatcher implements
PipeDataRegionMatcher {
if (event.getEvent() instanceof PipeTsFileInsertionEvent) {
final PipeTsFileInsertionEvent tsFileInsertionEvent =
(PipeTsFileInsertionEvent) event.getEvent();
- if (tsFileInsertionEvent.isTableModelEvent()) {
+ if (isTableModelTsFileEvent) {
tsFileInsertionEvent.setTableNames(tableNames);
} else {
tsFileInsertionEvent.setTreeSchemaMap(event.getSchemaInfo());
@@ -273,7 +279,7 @@ public class CachedSchemaPatternMatcher implements
PipeDataRegionMatcher {
protected void matchTableModelEvent(
final String databaseName,
- final IDeviceID tableName,
+ final String tableName,
final Set<PipeRealtimeDataRegionSource> matchedSources) {
// this would not happen
if (databaseName == null) {
@@ -294,7 +300,7 @@ public class CachedSchemaPatternMatcher implements
PipeDataRegionMatcher {
}
protected Set<PipeRealtimeDataRegionSource> filterSourcesByDatabaseAndTable(
- final Pair<String, IDeviceID> databaseNameAndTableName) {
+ final Pair<String, String> databaseNameAndTableName) {
final Set<PipeRealtimeDataRegionSource> filteredSources = new HashSet<>();
for (final PipeRealtimeDataRegionSource source : sources) {
@@ -317,21 +323,20 @@ public class CachedSchemaPatternMatcher implements
PipeDataRegionMatcher {
}
private boolean matchesTablePattern(
- final TablePattern tablePattern, final Pair<String, IDeviceID>
databaseNameAndTableName) {
+ final TablePattern tablePattern, final Pair<String, String>
databaseNameAndTableName) {
return Objects.isNull(tablePattern)
|| (tablePattern.isTableModelDataAllowedToBeCaptured()
&& tablePattern.matchesDatabase(databaseNameAndTableName.getLeft())
- &&
tablePattern.matchesTable(databaseNameAndTableName.getRight().getTableName()));
+ && tablePattern.matchesTable(databaseNameAndTableName.getRight()));
}
private boolean notFilteredByAccess(
- final UserEntity userEntity, final Pair<String, IDeviceID>
databaseNameAndTableName) {
+ final UserEntity userEntity, final Pair<String, String>
databaseNameAndTableName) {
return AuthorityChecker.getAccessControl()
.checkCanSelectFromTable4Pipe(
userEntity.getUsername(),
new QualifiedObjectName(
- databaseNameAndTableName.getLeft(),
- databaseNameAndTableName.getRight().getTableName()),
+ databaseNameAndTableName.getLeft(),
databaseNameAndTableName.getRight()),
userEntity);
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
index 61d5232a3d8..832bae39352 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
@@ -25,6 +25,8 @@ import
org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvi
import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.epoch.TsFileEpoch;
@@ -39,12 +41,18 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -73,6 +81,35 @@ public class CachedSchemaPatternMatcherTest {
}
}
+ private static class CountingCachedSchemaPatternMatcher extends
CachedSchemaPatternMatcher {
+
+ private final int sourcesToAddOnEachMatch;
+ private int tableMatchCount;
+
+ private CountingCachedSchemaPatternMatcher(final int
sourcesToAddOnEachMatch) {
+ this.sourcesToAddOnEachMatch = sourcesToAddOnEachMatch;
+ }
+
+ @Override
+ protected void matchTableModelEvent(
+ final String databaseName,
+ final String tableName,
+ final Set<PipeRealtimeDataRegionSource> matchedSources) {
+ ++tableMatchCount;
+ int addedSourceCount = 0;
+ for (final PipeRealtimeDataRegionSource source : sources) {
+ matchedSources.add(source);
+ if (++addedSourceCount >= sourcesToAddOnEachMatch) {
+ break;
+ }
+ }
+ }
+
+ private int getTableMatchCount() {
+ return tableMatchCount;
+ }
+ }
+
private CachedSchemaPatternMatcher matcher;
private ExecutorService executorService;
private List<PipeRealtimeDataRegionSource> extractors;
@@ -178,6 +215,52 @@ public class CachedSchemaPatternMatcherTest {
future.get();
}
+ @Test
+ public void testTableModelMatchesEachTableOncePerEvent() throws Exception {
+ final CountingCachedSchemaPatternMatcher countingMatcher =
+ new CountingCachedSchemaPatternMatcher(1);
+ countingMatcher.register(new PipeRealtimeDataRegionFakeSource());
+ countingMatcher.register(new PipeRealtimeDataRegionFakeSource());
+
+ final PipeInsertionEvent insertionEvent =
Mockito.mock(PipeInsertionEvent.class);
+ Mockito.when(insertionEvent.getTableModelDatabaseName()).thenReturn("db");
+ final Map<IDeviceID, String[]> schemaInfo = new LinkedHashMap<>();
+ schemaInfo.put(new StringArrayDeviceID("table1", "tag1"), new String[0]);
+ schemaInfo.put(new StringArrayDeviceID("table1", "tag2"), new String[0]);
+
+ Assert.assertEquals(
+ 1,
+ countingMatcher
+ .match(new MockedPipeRealtimeEvent(insertionEvent, null,
schemaInfo))
+ .getLeft()
+ .size());
+ Assert.assertEquals(1, countingMatcher.getTableMatchCount());
+ }
+
+ @Test
+ public void
testMultiTableTsFileCollectsAllTableNamesAfterAllSourcesMatched() throws
Exception {
+ final CountingCachedSchemaPatternMatcher countingMatcher =
+ new CountingCachedSchemaPatternMatcher(Integer.MAX_VALUE);
+ countingMatcher.register(new PipeRealtimeDataRegionFakeSource());
+ countingMatcher.register(new PipeRealtimeDataRegionFakeSource());
+
+ final PipeTsFileInsertionEvent tsFileInsertionEvent =
+ Mockito.mock(PipeTsFileInsertionEvent.class);
+ Mockito.when(tsFileInsertionEvent.isTableModelEvent()).thenReturn(true);
+
Mockito.when(tsFileInsertionEvent.getTableModelDatabaseName()).thenReturn("db");
+ final Map<IDeviceID, String[]> schemaInfo = new LinkedHashMap<>();
+ schemaInfo.put(new StringArrayDeviceID("table1", "tag1"), new String[0]);
+ schemaInfo.put(new StringArrayDeviceID("table2", "tag2"), new String[0]);
+
+ countingMatcher.match(new MockedPipeRealtimeEvent(tsFileInsertionEvent,
null, schemaInfo));
+
+ final ArgumentCaptor<Set<String>> tableNamesCaptor =
ArgumentCaptor.forClass(Set.class);
+
Mockito.verify(tsFileInsertionEvent).setTableNames(tableNamesCaptor.capture());
+ Assert.assertEquals(
+ new HashSet<>(Arrays.asList("table1", "table2")),
tableNamesCaptor.getValue());
+ Assert.assertEquals(1, countingMatcher.getTableMatchCount());
+ }
+
public static class PipeRealtimeDataRegionFakeSource extends
PipeRealtimeDataRegionSource {
public PipeRealtimeDataRegionFakeSource() {
