Skip to content

Commit 7558aa7

Browse files
CaideyipiCaideyipi
authored
Fix metadata cache and pipe edge cases (#18147)
1 parent 797319b commit 7558aa7

8 files changed

Lines changed: 177 additions & 9 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
3636
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
3737
import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
38+
import org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTablePrivilegeParseVisitor;
3839
import org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTreePrivilegeParseVisitor;
3940
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractDeleteDataNode;
4041
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
@@ -202,12 +203,14 @@ private void parseAndCollectEvent(final PipeDeleteDataNodeEvent deleteDataEvent)
202203
.process(deleteDataEvent.getDeleteDataNode(), deleteDataEvent.getTablePattern())
203204
.flatMap(
204205
planNode ->
205-
IoTDBSchemaRegionSource.TABLE_PRIVILEGE_PARSE_VISITOR.process(
206-
planNode,
207-
new UserEntity(
208-
Long.parseLong(deleteDataEvent.getUserId()),
209-
deleteDataEvent.getUserName(),
210-
deleteDataEvent.getCliHostname()))))
206+
new PipePlanTablePrivilegeParseVisitor(
207+
deleteDataEvent.isSkipIfNoPrivileges())
208+
.process(
209+
planNode,
210+
new UserEntity(
211+
Long.parseLong(deleteDataEvent.getUserId()),
212+
deleteDataEvent.getUserName(),
213+
deleteDataEvent.getCliHostname()))))
211214
.map(
212215
planNode ->
213216
new PipeDeleteDataNodeEvent(

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -448,7 +448,10 @@ protected void extractProgressReportEvent(final PipeRealtimeEvent event) {
448448
if (lastEvent == null || !(lastEvent.getEvent() instanceof PipeHeartbeatEvent)) {
449449
break;
450450
}
451-
pendingQueue.pollLast();
451+
final PipeRealtimeEvent droppedEvent = (PipeRealtimeEvent) pendingQueue.pollLast();
452+
if (droppedEvent != null) {
453+
droppedEvent.decreaseReferenceCount(PipeRealtimeDataRegionSource.class.getName(), false);
454+
}
452455
}
453456
final Event last = pendingQueue.peekLast();
454457
if (last instanceof PipeRealtimeEvent
@@ -459,6 +462,7 @@ protected void extractProgressReportEvent(final PipeRealtimeEvent event) {
459462
oldEvent
460463
.getProgressIndex()
461464
.updateToMinimumEqualOrIsAfterProgressIndex(event.getProgressIndex()));
465+
event.decreaseReferenceCount(PipeRealtimeDataRegionSource.class.getName(), false);
462466
return;
463467
}
464468
pendingQueue.offer(event);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,7 @@ public class IoTDBSchemaRegionSource extends IoTDBNonDataRegionSource {
9797

9898
// Local for exception
9999
private PipePlanTreePrivilegeParseVisitor treePrivilegeParseVisitor;
100+
private PipePlanTablePrivilegeParseVisitor tablePrivilegeParseVisitor;
100101
private SchemaRegionId schemaRegionId;
101102

102103
private Set<PlanNodeType> listenedTypeSet = new HashSet<>();
@@ -123,6 +124,7 @@ public void customize(
123124
schemaRegionId = new SchemaRegionId(regionId);
124125
listenedTypeSet = SchemaRegionListeningFilter.parseListeningPlanTypeSet(parameters);
125126
treePrivilegeParseVisitor = new PipePlanTreePrivilegeParseVisitor(skipIfNoPrivileges);
127+
tablePrivilegeParseVisitor = new PipePlanTablePrivilegeParseVisitor(skipIfNoPrivileges);
126128

127129
PipeSchemaRegionSourceMetrics.getInstance().register(this);
128130
if (regionId >= 0) {
@@ -294,7 +296,7 @@ protected Optional<PipeWritePlanEvent> trimRealtimeEventByPrivilege(
294296
final Optional<PlanNode> result =
295297
treePrivilegeParseVisitor
296298
.process(((PipeSchemaRegionWritePlanEvent) event).getPlanNode(), userEntity)
297-
.flatMap(planNode -> TABLE_PRIVILEGE_PARSE_VISITOR.process(planNode, userEntity));
299+
.flatMap(planNode -> tablePrivilegeParseVisitor.process(planNode, userEntity));
298300
if (result.isPresent()) {
299301
return Optional.of(
300302
new PipeSchemaRegionWritePlanEvent(result.get(), event.isGeneratedByPipe()));

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/PipePlanTablePrivilegeParseVisitor.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,11 @@
2020
package org.apache.iotdb.db.pipe.source.schemaregion;
2121

2222
import org.apache.iotdb.commons.audit.IAuditEntity;
23+
import org.apache.iotdb.commons.exception.auth.AccessDeniedException;
2324
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
2425
import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName;
2526
import org.apache.iotdb.db.auth.AuthorityChecker;
27+
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
2628
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
2729
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
2830
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
@@ -35,6 +37,17 @@
3537

3638
public class PipePlanTablePrivilegeParseVisitor
3739
implements PlanVisitor<Optional<PlanNode>, IAuditEntity> {
40+
41+
private final boolean skip;
42+
43+
public PipePlanTablePrivilegeParseVisitor() {
44+
this(true);
45+
}
46+
47+
public PipePlanTablePrivilegeParseVisitor(final boolean skip) {
48+
this.skip = skip;
49+
}
50+
3851
@Override
3952
public Optional<PlanNode> visitPlan(final PlanNode node, final IAuditEntity auditEntity) {
4053
return Optional.of(node);
@@ -77,6 +90,10 @@ public Optional<PlanNode> visitDeleteData(
7790
new QualifiedObjectName(node.getDatabaseName(), entry.getTableName()),
7891
auditEntity))
7992
.collect(Collectors.toList());
93+
if (!skip && modEntries.size() != node.getModEntries().size()) {
94+
throw new AccessDeniedException(
95+
DataNodePipeMessages.NOT_HAS_PRIVILEGE_TO_TRANSFER_EVENT + node);
96+
}
8097
return !modEntries.isEmpty()
8198
? Optional.of(
8299
new RelationalDeleteDataNode(node.getPlanNodeId(), modEntries, node.getDatabaseName()))

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -198,7 +198,7 @@ public void rollbackUpdateTable(String database, final String tableName, final S
198198
}
199199
databaseTableMap
200200
.computeIfAbsent(database, k -> new ConcurrentHashMap<>())
201-
.put(tableName, oldTable);
201+
.put(oldName, oldTable);
202202
LOGGER.info(DataNodeSchemaMessages.ROLLBACK_RENAME_OLD_TABLE_SUCCESS, database, oldName);
203203
removeTableFromPreUpdateMap(database, oldName);
204204
}

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipePlanTablePatternParseVisitorTest.java‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,22 @@
1919

2020
package org.apache.iotdb.db.pipe.source;
2121

22+
import org.apache.iotdb.commons.audit.IAuditEntity;
23+
import org.apache.iotdb.commons.audit.UserEntity;
24+
import org.apache.iotdb.commons.exception.auth.AccessDeniedException;
2225
import org.apache.iotdb.commons.path.PartialPath;
2326
import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
2427
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
2528
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
29+
import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName;
30+
import org.apache.iotdb.db.auth.AuthorityChecker;
2631
import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
32+
import org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTablePrivilegeParseVisitor;
2733
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
2834
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
2935
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode;
36+
import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl;
37+
import org.apache.iotdb.db.queryengine.plan.relational.security.AllowAllAccessControl;
3038
import org.apache.iotdb.db.storageengine.dataregion.memtable.DeviceIDFactory;
3139
import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate;
3240
import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry;
@@ -126,4 +134,45 @@ public void testDeleteData() {
126134
tablePattern)
127135
.orElse(null));
128136
}
137+
138+
@Test
139+
public void testDeleteDataPrivilegeRespectsSkipOption() {
140+
final AccessControl oldAccessControl = AuthorityChecker.getAccessControl();
141+
try {
142+
AuthorityChecker.setAccessControl(
143+
new AllowAllAccessControl() {
144+
@Override
145+
public boolean checkCanSelectFromTable4Pipe(
146+
final String userName,
147+
final QualifiedObjectName tableName,
148+
final IAuditEntity auditEntity) {
149+
return "ab".equals(tableName.getObjectName());
150+
}
151+
});
152+
153+
final RelationalDeleteDataNode input =
154+
new RelationalDeleteDataNode(
155+
new PlanNodeId(""),
156+
Arrays.asList(
157+
new TableDeletionEntry(
158+
new DeletionPredicate("ab"), new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE)),
159+
new TableDeletionEntry(
160+
new DeletionPredicate("ac"), new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE))),
161+
"db1");
162+
final UserEntity userEntity = new UserEntity(1, "test", "127.0.0.1");
163+
164+
Assert.assertEquals(
165+
new RelationalDeleteDataNode(
166+
new PlanNodeId(""),
167+
new TableDeletionEntry(
168+
new DeletionPredicate("ab"), new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE)),
169+
"db1"),
170+
new PipePlanTablePrivilegeParseVisitor(true).process(input, userEntity).orElse(null));
171+
Assert.assertThrows(
172+
AccessDeniedException.class,
173+
() -> new PipePlanTablePrivilegeParseVisitor(false).process(input, userEntity));
174+
} finally {
175+
AuthorityChecker.setAccessControl(oldAccessControl);
176+
}
177+
}
129178
}

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
3434
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
3535
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
36+
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
3637
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
3738
import org.apache.iotdb.commons.utils.FileUtils;
3839
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -517,6 +518,37 @@ private void registerTestPipeMeta() throws Exception {
517518
}
518519
}
519520

521+
@Test
522+
public void testProgressReportExtractionReleasesDroppedEvents() {
523+
final TestRealtimeDataRegionSource source = new TestRealtimeDataRegionSource();
524+
525+
final PipeRealtimeEvent heartbeatEvent =
526+
PipeRealtimeEventFactory.createRealtimeEvent(dataRegion1, false);
527+
heartbeatEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER);
528+
source.extractHeartbeatForTest(heartbeatEvent);
529+
Assert.assertEquals(1, heartbeatEvent.getEvent().getReferenceCount());
530+
531+
final PipeRealtimeEvent progressEvent = createProgressReportRealtimeEvent();
532+
progressEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER);
533+
source.extractProgressReportEventForTest(progressEvent);
534+
535+
Assert.assertEquals(0, heartbeatEvent.getEvent().getReferenceCount());
536+
Assert.assertEquals(1, progressEvent.getEvent().getReferenceCount());
537+
Assert.assertEquals(1, source.getEventCount());
538+
539+
final PipeRealtimeEvent mergedProgressEvent = createProgressReportRealtimeEvent();
540+
mergedProgressEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER);
541+
source.extractProgressReportEventForTest(mergedProgressEvent);
542+
543+
Assert.assertEquals(0, mergedProgressEvent.getEvent().getReferenceCount());
544+
Assert.assertEquals(1, progressEvent.getEvent().getReferenceCount());
545+
Assert.assertEquals(1, source.getEventCount());
546+
547+
final Event queuedEvent = source.pollForTest();
548+
Assert.assertSame(progressEvent, queuedEvent);
549+
((EnrichedEvent) queuedEvent).clearReferenceCount(TEST_REFERENCE_HOLDER);
550+
}
551+
520552
private void removeTestPipeMeta() throws Exception {
521553
final PipeMetaKeeper pipeMetaKeeper = getPipeMetaKeeper();
522554
pipeMetaKeeper.acquireWriteLock();
@@ -545,6 +577,47 @@ private void releaseSuppliedEvent(final Event event) {
545577
}
546578
}
547579

580+
private PipeRealtimeEvent createProgressReportRealtimeEvent() {
581+
final ProgressReportEvent progressReportEvent = new ProgressReportEvent(null, 0, null);
582+
progressReportEvent.bindProgressIndex(MinimumProgressIndex.INSTANCE);
583+
return PipeRealtimeEventFactory.createRealtimeEvent(progressReportEvent);
584+
}
585+
586+
private static class TestRealtimeDataRegionSource extends PipeRealtimeDataRegionSource {
587+
588+
private void extractHeartbeatForTest(final PipeRealtimeEvent event) {
589+
extractHeartbeat(event);
590+
}
591+
592+
private void extractProgressReportEventForTest(final PipeRealtimeEvent event) {
593+
extractProgressReportEvent(event);
594+
}
595+
596+
private Event pollForTest() {
597+
return pendingQueue.directPoll();
598+
}
599+
600+
@Override
601+
protected void doExtract(final PipeRealtimeEvent event) {
602+
// Not needed in this reference-counting unit test.
603+
}
604+
605+
@Override
606+
public Event supply() {
607+
return pendingQueue.directPoll();
608+
}
609+
610+
@Override
611+
public boolean isNeedListenToTsFile() {
612+
return false;
613+
}
614+
615+
@Override
616+
public boolean isNeedListenToInsertNode() {
617+
return false;
618+
}
619+
}
620+
548621
private static class NoTsFileRealtimeDataRegionSource extends PipeRealtimeDataRegionSource {
549622

550623
private final AtomicInteger observedTsFileEventCount = new AtomicInteger(0);

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,26 @@ public void commitAfterRollbackUpdateTableIsIgnored() {
9191
}
9292
}
9393

94+
@Test
95+
public void rollbackRenameTableRestoresOldName() {
96+
final DataNodeTableCache cache = DataNodeTableCache.getInstance();
97+
final String newTableName = "table2";
98+
cache.invalid(TABLE_CACHE_TEST_DATABASE);
99+
try {
100+
cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, createTable(TABLE_NAME), null);
101+
cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
102+
103+
cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, createTable(newTableName), TABLE_NAME);
104+
cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, newTableName, TABLE_NAME);
105+
106+
Assert.assertEquals(
107+
TABLE_NAME, cache.getTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME).getTableName());
108+
Assert.assertNull(cache.getTable(TABLE_CACHE_TEST_DATABASE, newTableName, false));
109+
} finally {
110+
cache.invalid(TABLE_CACHE_TEST_DATABASE);
111+
}
112+
}
113+
94114
private Semaphore getFetchTableSemaphore(final DataNodeTableCache cache) throws Exception {
95115
final Field field = DataNodeTableCache.class.getDeclaredField("fetchTableSemaphore");
96116
field.setAccessible(true);

0 commit comments

Comments
 (0)