Pipe: Support table pattern matching when sync data between clusters (#…

…13817 following fixup) (#13830)
apache · Oct 18, 2024 · 0d8ad0c · 0d8ad0c
1 parent 4bd629e
commit 0d8ad0c
Show file tree

Hide file tree

Showing 4 changed files with 21 additions and 3 deletions.
diff --git a/...node/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java b/...node/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -120,7 +120,11 @@ private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) th
       return;
     }
 
-    if (!forceTabletFormat && !sourceEvent.shouldParseTimeOrPattern()) {
+    if (!forceTabletFormat
+        && (!sourceEvent.shouldParseTimeOrPattern()
+            || (sourceEvent.isTableModelEvent()
+                && sourceEvent.getTablePattern() == null
+                && !sourceEvent.shouldParseTime()))) {
       collectEvent(sourceEvent);
       return;
     }

diff --git a/...org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventPlainBatch.java b/...org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventPlainBatch.java
@@ -110,6 +110,10 @@ public synchronized void onSuccess() {
     insertNodeBuffers.clear();
     tabletBuffers.clear();
 
+    binaryDataBases.clear();
+    inertNodeDataBases.clear();
+    tabletDataBases.clear();
+
     pipe2BytesAccumulated.clear();
   }
 

diff --git a/.../pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java b/.../pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java
@@ -102,6 +102,8 @@ private Tablet buildNextTablet() throws IOException {
         new Tablet(tableName, columnSchemas, columnTypes, tsBlock.getPositionCount());
     tablet.initBitMaps();
 
+    boolean isAllNull = true;
+
     final TsBlock.TsBlockRowIterator rowIterator = tsBlock.getTsBlockRowIterator();
     while (rowIterator.hasNext()) {
       final Object[] row = rowIterator.next();
@@ -114,11 +116,19 @@ private Tablet buildNextTablet() throws IOException {
       final int rowIndex = tablet.rowSize;
       tablet.addTimestamp(rowIndex, timestamp);
       for (int i = 0, fieldSize = row.length - 1; i < fieldSize; i++) {
-        tablet.addValue(columnNames.get(i), rowIndex, row[i]);
+        final Object value = row[i];
+        tablet.addValue(columnNames.get(i), rowIndex, value);
+        if (value != null && columnTypes.get(i) == Tablet.ColumnType.MEASUREMENT) {
+          isAllNull = false;
+        }
       }
       tablet.rowSize++;
     }
 
+    if (isAllNull) {
+      tablet.rowSize = 0;
+    }
+
     return tablet;
   }
 }
diff --git a/pom.xml b/pom.xml
@@ -166,7 +166,7 @@
         <thrift.version>0.14.1</thrift.version>
         <xz.version>1.9</xz.version>
         <zstd-jni.version>1.5.6-3</zstd-jni.version>
-        <tsfile.version>1.2.0-241018-SNAPSHOT</tsfile.version>
+        <tsfile.version>1.2.0-4e902e0-SNAPSHOT</tsfile.version>
     </properties>
     <!--
     if we claim dependencies in dependencyManagement, then we do not claim