Merge branch 'master' into fast-iot-ci

apache · Oct 24, 2024 · 3a14c1b · 3a14c1b
2 parents 7bc70f1 + 4e79200
commit 3a14c1b
Show file tree

Hide file tree

Showing 65 changed files with 1,794 additions and 633 deletions.
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java
@@ -65,6 +65,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -954,7 +955,7 @@ public void testInsertTabletWithTTL()
         Assert.assertTrue(record.getFields().get(0).getLongV() > timeLowerBound);
         count++;
       }
-      Assert.assertTrue(count > 0 && count < 4);
+      Assert.assertEquals(2, count);
     }
   }
 
@@ -1041,6 +1042,39 @@ public void testInsertUnsequenceData()
     }
   }
 
+  @Test
+  public void testInsertAllNullRow() throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement st1 = connection.createStatement()) {
+      st1.execute("use \"test\"");
+      st1.execute("create table table5(d1 string id, s1 int32 measurement, s2 int32 measurement)");
+
+      st1.execute("insert into table5(time, d1,s1,s2) values(1,'a',1,null)");
+      // insert all null row
+      st1.execute("insert into table5(time, d1,s1,s2) values(2,'a',null,null)");
+
+      ResultSet rs1 = st1.executeQuery("select * from table5");
+      assertTrue(rs1.next());
+      assertEquals("1", rs1.getString("s1"));
+      assertNull(rs1.getString("s2"));
+      assertTrue(rs1.next());
+      assertNull(rs1.getString("s1"));
+      assertNull(rs1.getString("s2"));
+      assertFalse(rs1.next());
+
+      st1.execute("flush");
+
+      rs1 = st1.executeQuery("select * from table5");
+      assertTrue(rs1.next());
+      assertEquals("1", rs1.getString("s1"));
+      assertNull(rs1.getString("s2"));
+      assertTrue(rs1.next());
+      assertNull(rs1.getString("s1"));
+      assertNull(rs1.getString("s2"));
+      assertFalse(rs1.next());
+    }
+  }
+
   private List<Integer> checkHeader(
       ResultSetMetaData resultSetMetaData, String expectedHeaderStrings, int[] expectedTypes)
       throws SQLException {

diff --git a/...rc/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java b/...rc/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java
diff --git a/...va/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java b/...va/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java
@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
 import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor;
+import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalPhantomReferenceCleaner;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
@@ -49,6 +50,9 @@ public class PipeConfigNodeRuntimeAgent implements IService {
   private final PipePeriodicalJobExecutor pipePeriodicalJobExecutor =
       new PipePeriodicalJobExecutor();
 
+  private final PipePeriodicalPhantomReferenceCleaner pipePeriodicalPhantomReferenceCleaner =
+      new PipePeriodicalPhantomReferenceCleaner();
+
   @Override
   public synchronized void start() {
     PipeConfig.getInstance().printAllConfigs();
@@ -65,6 +69,10 @@ public synchronized void start() {
     // Start periodical job executor
     pipePeriodicalJobExecutor.start();
 
+    if (PipeConfig.getInstance().getPipeEventReferenceTrackingEnabled()) {
+      pipePeriodicalPhantomReferenceCleaner.start();
+    }
+
     isShutdown.set(false);
     LOGGER.info("PipeRuntimeConfigNodeAgent started");
   }
@@ -159,4 +167,9 @@ private void report(
   public void registerPeriodicalJob(String id, Runnable periodicalJob, long intervalInSeconds) {
     pipePeriodicalJobExecutor.register(id, periodicalJob, intervalInSeconds);
   }
+
+  public void registerPhantomReferenceCleanJob(
+      String id, Runnable periodicalJob, long intervalInSeconds) {
+    pipePeriodicalPhantomReferenceCleaner.register(id, periodicalJob, intervalInSeconds);
+  }
 }
diff --git a/...che/iotdb/confignode/manager/pipe/resource/ref/PipeConfigNodePhantomReferenceManager.java b/...che/iotdb/confignode/manager/pipe/resource/ref/PipeConfigNodePhantomReferenceManager.java
@@ -29,7 +29,7 @@ public PipeConfigNodePhantomReferenceManager() {
     super();
 
     PipeConfigNodeAgent.runtime()
-        .registerPeriodicalJob(
+        .registerPhantomReferenceCleanJob(
             "PipePhantomReferenceManager#gcHook()",
             // NOTE: lambda CAN NOT be replaced with method reference
             () -> super.gcHook(),

diff --git a/...tanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java b/...tanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java
@@ -26,6 +26,7 @@
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
 import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor;
+import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalPhantomReferenceCleaner;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
@@ -62,6 +63,9 @@ public class PipeDataNodeRuntimeAgent implements IService {
   private final PipePeriodicalJobExecutor pipePeriodicalJobExecutor =
       new PipePeriodicalJobExecutor();
 
+  private final PipePeriodicalPhantomReferenceCleaner pipePeriodicalPhantomReferenceCleaner =
+      new PipePeriodicalPhantomReferenceCleaner();
+
   //////////////////////////// System Service Interface ////////////////////////////
 
   public synchronized void preparePipeResources(
@@ -87,6 +91,10 @@ public synchronized void start() throws StartupException {
         PipeConfig.getInstance().getPipeStuckRestartIntervalSeconds());
     pipePeriodicalJobExecutor.start();
 
+    if (PipeConfig.getInstance().getPipeEventReferenceTrackingEnabled()) {
+      pipePeriodicalPhantomReferenceCleaner.start();
+    }
+
     isShutdown.set(false);
   }
 
@@ -225,4 +233,9 @@ public void stopPeriodicalJobExecutor() {
   public void clearPeriodicalJobExecutor() {
     pipePeriodicalJobExecutor.clear();
   }
+
+  public void registerPhantomReferenceCleanJob(
+      String id, Runnable periodicalJob, long intervalInSeconds) {
+    pipePeriodicalPhantomReferenceCleaner.register(id, periodicalJob, intervalInSeconds);
+  }
 }
diff --git a/...ore/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/...ore/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
@@ -118,11 +118,11 @@ protected void createPipeTask(
       throws IllegalPathException {
     if (pipeTaskMeta.getLeaderNodeId() == CONFIG.getDataNodeId()) {
       final PipeParameters extractorParameters = pipeStaticMeta.getExtractorParameters();
+      final DataRegionId dataRegionId = new DataRegionId(consensusGroupId);
       final boolean needConstructDataRegionTask =
-          StorageEngine.getInstance()
-                  .getAllDataRegionIds()
-                  .contains(new DataRegionId(consensusGroupId))
-              && DataRegionListeningFilter.shouldDataRegionBeListened(extractorParameters);
+          StorageEngine.getInstance().getAllDataRegionIds().contains(dataRegionId)
+              && DataRegionListeningFilter.shouldDataRegionBeListened(
+                  extractorParameters, dataRegionId);
       final boolean needConstructSchemaRegionTask =
           SchemaEngine.getInstance()
                   .getAllSchemaRegionIds()

diff --git a/...tanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeBuilder.java b/...tanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeBuilder.java
@@ -64,9 +64,11 @@ public Map<Integer, PipeTask> build() throws IllegalPathException {
 
       if (pipeTaskMeta.getLeaderNodeId() == CONFIG.getDataNodeId()) {
         final PipeParameters extractorParameters = pipeStaticMeta.getExtractorParameters();
+        final DataRegionId dataRegionId = new DataRegionId(consensusGroupId);
         final boolean needConstructDataRegionTask =
-            dataRegionIds.contains(new DataRegionId(consensusGroupId))
-                && DataRegionListeningFilter.shouldDataRegionBeListened(extractorParameters);
+            dataRegionIds.contains(dataRegionId)
+                && DataRegionListeningFilter.shouldDataRegionBeListened(
+                    extractorParameters, dataRegionId);
         final boolean needConstructSchemaRegionTask =
             schemaRegionIds.contains(new SchemaRegionId(consensusGroupId))
                 && !SchemaRegionListeningFilter.parseListeningPlanTypeSet(extractorParameters)

diff --git a/...ache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferTabletBinaryReqV2.java b/...ache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferTabletBinaryReqV2.java
@@ -27,13 +27,16 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement;
 
 import org.apache.tsfile.utils.PublicBAOS;
 import org.apache.tsfile.utils.ReadWriteIOUtils;
 
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.Objects;
 
 public class PipeTransferTabletBinaryReqV2 extends PipeTransferTabletBinaryReq {
@@ -71,6 +74,16 @@ public InsertBaseStatement constructStatement() {
 
     // Table model
     statement.setWriteToTable(true);
+    if (statement instanceof InsertRowsStatement) {
+      List<InsertRowStatement> rowStatements =
+          ((InsertRowsStatement) statement).getInsertRowStatementList();
+      if (rowStatements != null && !rowStatements.isEmpty()) {
+        for (InsertRowStatement insertRowStatement : rowStatements) {
+          insertRowStatement.setWriteToTable(true);
+          insertRowStatement.setDatabaseName(dataBaseName);
+        }
+      }
+    }
     statement.setDatabaseName(dataBaseName);
     return statement;
   }

diff --git a/.../iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferTabletInsertNodeReqV2.java b/.../iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferTabletInsertNodeReqV2.java
@@ -28,13 +28,16 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement;
 
 import org.apache.tsfile.utils.PublicBAOS;
 import org.apache.tsfile.utils.ReadWriteIOUtils;
 
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.Objects;
 
 public class PipeTransferTabletInsertNodeReqV2 extends PipeTransferTabletInsertNodeReq {
@@ -71,6 +74,16 @@ public InsertBaseStatement constructStatement() {
 
     // Table model
     statement.setWriteToTable(true);
+    if (statement instanceof InsertRowsStatement) {
+      List<InsertRowStatement> rowStatements =
+          ((InsertRowsStatement) statement).getInsertRowStatementList();
+      if (rowStatements != null && !rowStatements.isEmpty()) {
+        for (InsertRowStatement insertRowStatement : rowStatements) {
+          insertRowStatement.setWriteToTable(true);
+          insertRowStatement.setDatabaseName(dataBaseName);
+        }
+      }
+    }
     statement.setDatabaseName(dataBaseName);
     return statement;
   }

diff --git a/...java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/...java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -163,7 +163,7 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa
       PipeDataNodeResourceManager.wal().pin(walEntryHandler);
       if (Objects.nonNull(pipeName)) {
         PipeDataNodeRemainingEventAndTimeMetrics.getInstance()
-            .increaseInsertionEventCount(pipeName + "_" + creationTime);
+            .increaseTabletEventCount(pipeName + "_" + creationTime);
       }
       return true;
     } catch (final Exception e) {
@@ -196,7 +196,7 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
     } finally {
       if (Objects.nonNull(pipeName)) {
         PipeDataNodeRemainingEventAndTimeMetrics.getInstance()
-            .decreaseInsertionEventCount(pipeName + "_" + creationTime);
+            .decreaseTabletEventCount(pipeName + "_" + creationTime);
       }
     }
   }

diff --git a/...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
@@ -171,7 +171,7 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa
                 PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet));
     if (Objects.nonNull(pipeName)) {
       PipeDataNodeRemainingEventAndTimeMetrics.getInstance()
-          .increaseInsertionEventCount(pipeName + "_" + creationTime);
+          .increaseTabletEventCount(pipeName + "_" + creationTime);
     }
     return true;
   }
@@ -180,7 +180,7 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa
   public boolean internallyDecreaseResourceReferenceCount(final String holderMessage) {
     if (Objects.nonNull(pipeName)) {
       PipeDataNodeRemainingEventAndTimeMetrics.getInstance()
-          .decreaseInsertionEventCount(pipeName + "_" + creationTime);
+          .decreaseTabletEventCount(pipeName + "_" + creationTime);
     }
     allocatedMemoryBlock.close();
 

diff --git a/...rc/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/DataRegionListeningFilter.java b/...rc/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/DataRegionListeningFilter.java
@@ -19,9 +19,13 @@
 
 package org.apache.iotdb.db.pipe.extractor.dataregion;
 
+import org.apache.iotdb.commons.consensus.DataRegionId;
 import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.pipe.agent.task.PipeTask;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
+import org.apache.iotdb.db.storageengine.StorageEngine;
 import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 
@@ -57,12 +61,35 @@ public class DataRegionListeningFilter {
     }
   }
 
-  public static boolean shouldDataRegionBeListened(PipeParameters parameters)
-      throws IllegalPathException {
+  public static boolean shouldDataRegionBeListened(
+      PipeParameters parameters, DataRegionId dataRegionId) throws IllegalPathException {
     final Pair<Boolean, Boolean> insertionDeletionListeningOptionPair =
         parseInsertionDeletionListeningOptionPair(parameters);
-    return insertionDeletionListeningOptionPair.getLeft()
-        || insertionDeletionListeningOptionPair.getRight();
+    final boolean hasSpecificListeningOption =
+        insertionDeletionListeningOptionPair.getLeft()
+            || insertionDeletionListeningOptionPair.getRight();
+    if (!hasSpecificListeningOption) {
+      return false;
+    }
+
+    final DataRegion dataRegion = StorageEngine.getInstance().getDataRegion(dataRegionId);
+    if (dataRegion == null) {
+      return true;
+    }
+
+    final String databaseRawName = dataRegion.getDatabaseName();
+    final String databaseTreeModel =
+        databaseRawName.startsWith("root.") ? databaseRawName : "root." + databaseRawName;
+    final String databaseTableModel =
+        databaseRawName.startsWith("root.") ? databaseRawName.substring(5) : databaseRawName;
+
+    final TreePattern treePattern = TreePattern.parsePipePatternFromSourceParameters(parameters);
+    final TablePattern tablePattern = TablePattern.parsePipePatternFromSourceParameters(parameters);
+
+    return treePattern.isTreeModelDataAllowedToBeCaptured()
+            && treePattern.mayOverlapWithDb(databaseTreeModel)
+        || tablePattern.isTableModelDataAllowedToBeCaptured()
+            && tablePattern.matchesDatabase(databaseTableModel);
   }
 
   public static Pair<Boolean, Boolean> parseInsertionDeletionListeningOptionPair(

diff --git a/...c/main/java/org/apache/iotdb/db/pipe/metric/PipeDataNodeRemainingEventAndTimeMetrics.java b/...c/main/java/org/apache/iotdb/db/pipe/metric/PipeDataNodeRemainingEventAndTimeMetrics.java
@@ -129,16 +129,16 @@ public void register(final IoTDBSchemaRegionExtractor extractor) {
     }
   }
 
-  public void increaseInsertionEventCount(final String pipeID) {
+  public void increaseTabletEventCount(final String pipeID) {
     remainingEventAndTimeOperatorMap
         .computeIfAbsent(pipeID, k -> new PipeDataNodeRemainingEventAndTimeOperator())
-        .increaseInsertionEventCount();
+        .increaseTabletEventCount();
   }
 
-  public void decreaseInsertionEventCount(final String pipeID) {
+  public void decreaseTabletEventCount(final String pipeID) {
     remainingEventAndTimeOperatorMap
         .computeIfAbsent(pipeID, k -> new PipeDataNodeRemainingEventAndTimeOperator())
-        .decreaseInsertionEventCount();
+        .decreaseTabletEventCount();
   }
 
   public void increaseTsFileEventCount(final String pipeID) {

diff --git a/.../main/java/org/apache/iotdb/db/pipe/metric/PipeDataNodeRemainingEventAndTimeOperator.java b/.../main/java/org/apache/iotdb/db/pipe/metric/PipeDataNodeRemainingEventAndTimeOperator.java
@@ -44,7 +44,7 @@ class PipeDataNodeRemainingEventAndTimeOperator extends PipeRemainingOperator {
   private final Set<IoTDBSchemaRegionExtractor> schemaRegionExtractors =
       Collections.newSetFromMap(new ConcurrentHashMap<>());
 
-  private final AtomicInteger insertionEventCount = new AtomicInteger(0);
+  private final AtomicInteger tabletEventCount = new AtomicInteger(0);
   private final AtomicInteger tsfileEventCount = new AtomicInteger(0);
   private final AtomicInteger heartbeatEventCount = new AtomicInteger(0);
 
@@ -58,12 +58,12 @@ class PipeDataNodeRemainingEventAndTimeOperator extends PipeRemainingOperator {
 
   //////////////////////////// Remaining event & time calculation ////////////////////////////
 
-  void increaseInsertionEventCount() {
-    tsfileEventCount.incrementAndGet();
+  void increaseTabletEventCount() {
+    tabletEventCount.incrementAndGet();
   }
 
-  void decreaseInsertionEventCount() {
-    tsfileEventCount.decrementAndGet();
+  void decreaseTabletEventCount() {
+    tabletEventCount.decrementAndGet();
   }
 
   void increaseTsFileEventCount() {
@@ -84,6 +84,7 @@ void decreaseHeartbeatEventCount() {
 
   long getRemainingEvents() {
     return tsfileEventCount.get()
+        + tabletEventCount.get()
         + heartbeatEventCount.get()
         + schemaRegionExtractors.stream()
             .map(IoTDBSchemaRegionExtractor::getUnTransferredEventCount)
@@ -105,7 +106,7 @@ long getRemainingEvents() {
     final double invocationValue = collectInvocationHistogram.getMean();
     // Do not take heartbeat event into account
     final double totalDataRegionWriteEventCount =
-        tsfileEventCount.get() * Math.max(invocationValue, 1) + insertionEventCount.get();
+        tsfileEventCount.get() * Math.max(invocationValue, 1) + tabletEventCount.get();
 
     dataRegionCommitMeter.updateAndGet(
         meter -> {