This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 7558aa7ef80 Fix metadata cache and pipe edge cases (#18147)
7558aa7ef80 is described below
commit 7558aa7ef8012f94a082dbb51b55ef2906f60b75
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 8 17:15:04 2026 +0800
Fix metadata cache and pipe edge cases (#18147)
---
.../agent/task/connection/PipeEventCollector.java | 15 +++--
.../realtime/PipeRealtimeDataRegionSource.java | 6 +-
.../schemaregion/IoTDBSchemaRegionSource.java | 4 +-
.../PipePlanTablePrivilegeParseVisitor.java | 17 +++++
.../db/schemaengine/table/DataNodeTableCache.java | 2 +-
.../PipePlanTablePatternParseVisitorTest.java | 49 +++++++++++++++
.../db/pipe/source/PipeRealtimeExtractTest.java | 73 ++++++++++++++++++++++
.../schemaengine/table/DataNodeTableCacheTest.java | 20 ++++++
8 files changed, 177 insertions(+), 9 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
index 2324a3d2707..5ecd81d92b4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -35,6 +35,7 @@ import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
+import
org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTablePrivilegeParseVisitor;
import
org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTreePrivilegeParseVisitor;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractDeleteDataNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
@@ -202,12 +203,14 @@ public class PipeEventCollector implements EventCollector
{
.process(deleteDataEvent.getDeleteDataNode(),
deleteDataEvent.getTablePattern())
.flatMap(
planNode ->
-
IoTDBSchemaRegionSource.TABLE_PRIVILEGE_PARSE_VISITOR.process(
- planNode,
- new UserEntity(
- Long.parseLong(deleteDataEvent.getUserId()),
- deleteDataEvent.getUserName(),
- deleteDataEvent.getCliHostname()))))
+ new PipePlanTablePrivilegeParseVisitor(
+ deleteDataEvent.isSkipIfNoPrivileges())
+ .process(
+ planNode,
+ new UserEntity(
+
Long.parseLong(deleteDataEvent.getUserId()),
+ deleteDataEvent.getUserName(),
+ deleteDataEvent.getCliHostname()))))
.map(
planNode ->
new PipeDeleteDataNodeEvent(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
index 62136877fc1..0d03df454ed 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -448,7 +448,10 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
if (lastEvent == null || !(lastEvent.getEvent() instanceof
PipeHeartbeatEvent)) {
break;
}
- pendingQueue.pollLast();
+ final PipeRealtimeEvent droppedEvent = (PipeRealtimeEvent)
pendingQueue.pollLast();
+ if (droppedEvent != null) {
+
droppedEvent.decreaseReferenceCount(PipeRealtimeDataRegionSource.class.getName(),
false);
+ }
}
final Event last = pendingQueue.peekLast();
if (last instanceof PipeRealtimeEvent
@@ -459,6 +462,7 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
oldEvent
.getProgressIndex()
.updateToMinimumEqualOrIsAfterProgressIndex(event.getProgressIndex()));
+
event.decreaseReferenceCount(PipeRealtimeDataRegionSource.class.getName(),
false);
return;
}
pendingQueue.offer(event);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java
index ecb27f1b266..8e33be13bdb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java
@@ -97,6 +97,7 @@ public class IoTDBSchemaRegionSource extends
IoTDBNonDataRegionSource {
// Local for exception
private PipePlanTreePrivilegeParseVisitor treePrivilegeParseVisitor;
+ private PipePlanTablePrivilegeParseVisitor tablePrivilegeParseVisitor;
private SchemaRegionId schemaRegionId;
private Set<PlanNodeType> listenedTypeSet = new HashSet<>();
@@ -123,6 +124,7 @@ public class IoTDBSchemaRegionSource extends
IoTDBNonDataRegionSource {
schemaRegionId = new SchemaRegionId(regionId);
listenedTypeSet =
SchemaRegionListeningFilter.parseListeningPlanTypeSet(parameters);
treePrivilegeParseVisitor = new
PipePlanTreePrivilegeParseVisitor(skipIfNoPrivileges);
+ tablePrivilegeParseVisitor = new
PipePlanTablePrivilegeParseVisitor(skipIfNoPrivileges);
PipeSchemaRegionSourceMetrics.getInstance().register(this);
if (regionId >= 0) {
@@ -294,7 +296,7 @@ public class IoTDBSchemaRegionSource extends
IoTDBNonDataRegionSource {
final Optional<PlanNode> result =
treePrivilegeParseVisitor
.process(((PipeSchemaRegionWritePlanEvent) event).getPlanNode(),
userEntity)
- .flatMap(planNode ->
TABLE_PRIVILEGE_PARSE_VISITOR.process(planNode, userEntity));
+ .flatMap(planNode -> tablePrivilegeParseVisitor.process(planNode,
userEntity));
if (result.isPresent()) {
return Optional.of(
new PipeSchemaRegionWritePlanEvent(result.get(),
event.isGeneratedByPipe()));
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/PipePlanTablePrivilegeParseVisitor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/PipePlanTablePrivilegeParseVisitor.java
index 4737b18cc76..f6c8b17bf67 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/PipePlanTablePrivilegeParseVisitor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/PipePlanTablePrivilegeParseVisitor.java
@@ -20,9 +20,11 @@
package org.apache.iotdb.db.pipe.source.schemaregion;
import org.apache.iotdb.commons.audit.IAuditEntity;
+import org.apache.iotdb.commons.exception.auth.AccessDeniedException;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName;
import org.apache.iotdb.db.auth.AuthorityChecker;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
@@ -35,6 +37,17 @@ import java.util.stream.Collectors;
public class PipePlanTablePrivilegeParseVisitor
implements PlanVisitor<Optional<PlanNode>, IAuditEntity> {
+
+ private final boolean skip;
+
+ public PipePlanTablePrivilegeParseVisitor() {
+ this(true);
+ }
+
+ public PipePlanTablePrivilegeParseVisitor(final boolean skip) {
+ this.skip = skip;
+ }
+
@Override
public Optional<PlanNode> visitPlan(final PlanNode node, final IAuditEntity
auditEntity) {
return Optional.of(node);
@@ -77,6 +90,10 @@ public class PipePlanTablePrivilegeParseVisitor
new QualifiedObjectName(node.getDatabaseName(),
entry.getTableName()),
auditEntity))
.collect(Collectors.toList());
+ if (!skip && modEntries.size() != node.getModEntries().size()) {
+ throw new AccessDeniedException(
+ DataNodePipeMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_EVENT + node);
+ }
return !modEntries.isEmpty()
? Optional.of(
new RelationalDeleteDataNode(node.getPlanNodeId(), modEntries,
node.getDatabaseName()))
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
index 142e3553715..d8a7e434a9f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
@@ -198,7 +198,7 @@ public class DataNodeTableCache implements ITableCache {
}
databaseTableMap
.computeIfAbsent(database, k -> new ConcurrentHashMap<>())
- .put(tableName, oldTable);
+ .put(oldName, oldTable);
LOGGER.info(DataNodeSchemaMessages.ROLLBACK_RENAME_OLD_TABLE_SUCCESS,
database, oldName);
removeTableFromPreUpdateMap(database, oldName);
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipePlanTablePatternParseVisitorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipePlanTablePatternParseVisitorTest.java
index a4da1fc2fdf..283ca6e4090 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipePlanTablePatternParseVisitorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipePlanTablePatternParseVisitorTest.java
@@ -19,14 +19,22 @@
package org.apache.iotdb.db.pipe.source;
+import org.apache.iotdb.commons.audit.IAuditEntity;
+import org.apache.iotdb.commons.audit.UserEntity;
+import org.apache.iotdb.commons.exception.auth.AccessDeniedException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName;
+import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
+import
org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTablePrivilegeParseVisitor;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
+import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl;
+import
org.apache.iotdb.db.queryengine.plan.relational.security.AllowAllAccessControl;
import org.apache.iotdb.db.storageengine.dataregion.memtable.DeviceIDFactory;
import
org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate;
import
org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry;
@@ -126,4 +134,45 @@ public class PipePlanTablePatternParseVisitorTest {
tablePattern)
.orElse(null));
}
+
+ @Test
+ public void testDeleteDataPrivilegeRespectsSkipOption() {
+ final AccessControl oldAccessControl = AuthorityChecker.getAccessControl();
+ try {
+ AuthorityChecker.setAccessControl(
+ new AllowAllAccessControl() {
+ @Override
+ public boolean checkCanSelectFromTable4Pipe(
+ final String userName,
+ final QualifiedObjectName tableName,
+ final IAuditEntity auditEntity) {
+ return "ab".equals(tableName.getObjectName());
+ }
+ });
+
+ final RelationalDeleteDataNode input =
+ new RelationalDeleteDataNode(
+ new PlanNodeId(""),
+ Arrays.asList(
+ new TableDeletionEntry(
+ new DeletionPredicate("ab"), new
TimeRange(Long.MIN_VALUE, Long.MAX_VALUE)),
+ new TableDeletionEntry(
+ new DeletionPredicate("ac"), new
TimeRange(Long.MIN_VALUE, Long.MAX_VALUE))),
+ "db1");
+ final UserEntity userEntity = new UserEntity(1, "test", "127.0.0.1");
+
+ Assert.assertEquals(
+ new RelationalDeleteDataNode(
+ new PlanNodeId(""),
+ new TableDeletionEntry(
+ new DeletionPredicate("ab"), new TimeRange(Long.MIN_VALUE,
Long.MAX_VALUE)),
+ "db1"),
+ new PipePlanTablePrivilegeParseVisitor(true).process(input,
userEntity).orElse(null));
+ Assert.assertThrows(
+ AccessDeniedException.class,
+ () -> new PipePlanTablePrivilegeParseVisitor(false).process(input,
userEntity));
+ } finally {
+ AuthorityChecker.setAccessControl(oldAccessControl);
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
index 1b714a80009..fc4a15a885b 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
@@ -33,6 +33,7 @@ import
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import
org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
import
org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -517,6 +518,37 @@ public class PipeRealtimeExtractTest {
}
}
+ @Test
+ public void testProgressReportExtractionReleasesDroppedEvents() {
+ final TestRealtimeDataRegionSource source = new
TestRealtimeDataRegionSource();
+
+ final PipeRealtimeEvent heartbeatEvent =
+ PipeRealtimeEventFactory.createRealtimeEvent(dataRegion1, false);
+ heartbeatEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER);
+ source.extractHeartbeatForTest(heartbeatEvent);
+ Assert.assertEquals(1, heartbeatEvent.getEvent().getReferenceCount());
+
+ final PipeRealtimeEvent progressEvent =
createProgressReportRealtimeEvent();
+ progressEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER);
+ source.extractProgressReportEventForTest(progressEvent);
+
+ Assert.assertEquals(0, heartbeatEvent.getEvent().getReferenceCount());
+ Assert.assertEquals(1, progressEvent.getEvent().getReferenceCount());
+ Assert.assertEquals(1, source.getEventCount());
+
+ final PipeRealtimeEvent mergedProgressEvent =
createProgressReportRealtimeEvent();
+ mergedProgressEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER);
+ source.extractProgressReportEventForTest(mergedProgressEvent);
+
+ Assert.assertEquals(0, mergedProgressEvent.getEvent().getReferenceCount());
+ Assert.assertEquals(1, progressEvent.getEvent().getReferenceCount());
+ Assert.assertEquals(1, source.getEventCount());
+
+ final Event queuedEvent = source.pollForTest();
+ Assert.assertSame(progressEvent, queuedEvent);
+ ((EnrichedEvent) queuedEvent).clearReferenceCount(TEST_REFERENCE_HOLDER);
+ }
+
private void removeTestPipeMeta() throws Exception {
final PipeMetaKeeper pipeMetaKeeper = getPipeMetaKeeper();
pipeMetaKeeper.acquireWriteLock();
@@ -545,6 +577,47 @@ public class PipeRealtimeExtractTest {
}
}
+ private PipeRealtimeEvent createProgressReportRealtimeEvent() {
+ final ProgressReportEvent progressReportEvent = new
ProgressReportEvent(null, 0, null);
+ progressReportEvent.bindProgressIndex(MinimumProgressIndex.INSTANCE);
+ return PipeRealtimeEventFactory.createRealtimeEvent(progressReportEvent);
+ }
+
+ private static class TestRealtimeDataRegionSource extends
PipeRealtimeDataRegionSource {
+
+ private void extractHeartbeatForTest(final PipeRealtimeEvent event) {
+ extractHeartbeat(event);
+ }
+
+ private void extractProgressReportEventForTest(final PipeRealtimeEvent
event) {
+ extractProgressReportEvent(event);
+ }
+
+ private Event pollForTest() {
+ return pendingQueue.directPoll();
+ }
+
+ @Override
+ protected void doExtract(final PipeRealtimeEvent event) {
+ // Not needed in this reference-counting unit test.
+ }
+
+ @Override
+ public Event supply() {
+ return pendingQueue.directPoll();
+ }
+
+ @Override
+ public boolean isNeedListenToTsFile() {
+ return false;
+ }
+
+ @Override
+ public boolean isNeedListenToInsertNode() {
+ return false;
+ }
+ }
+
private static class NoTsFileRealtimeDataRegionSource extends
PipeRealtimeDataRegionSource {
private final AtomicInteger observedTsFileEventCount = new
AtomicInteger(0);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
index 4b33991e3d7..2e3717cf233 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
@@ -91,6 +91,26 @@ public class DataNodeTableCacheTest {
}
}
+ @Test
+ public void rollbackRenameTableRestoresOldName() {
+ final DataNodeTableCache cache = DataNodeTableCache.getInstance();
+ final String newTableName = "table2";
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ try {
+ cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, createTable(TABLE_NAME),
null);
+ cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+
+ cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE,
createTable(newTableName), TABLE_NAME);
+ cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, newTableName,
TABLE_NAME);
+
+ Assert.assertEquals(
+ TABLE_NAME, cache.getTable(TABLE_CACHE_TEST_DATABASE,
TABLE_NAME).getTableName());
+ Assert.assertNull(cache.getTable(TABLE_CACHE_TEST_DATABASE,
newTableName, false));
+ } finally {
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ }
+ }
+
private Semaphore getFetchTableSemaphore(final DataNodeTableCache cache)
throws Exception {
final Field field =
DataNodeTableCache.class.getDeclaredField("fetchTableSemaphore");
field.setAccessible(true);
