Pipe: Optimize the TsFile memory control mechanism of the disassembly table model (apache#14890)

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java

@@ -53,7 +53,6 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.time.LocalDate;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -64,8 +63,6 @@
 
 public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
 
-  private static final LocalDate EMPTY_DATE = LocalDate.of(1000, 1, 1);
-
   private final long startTime;
   private final long endTime;
   private final Filter filter;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java

@@ -24,28 +24,30 @@
 import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParser;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
 import org.apache.iotdb.pipe.api.exception.PipeException;
 
-import org.apache.tsfile.file.metadata.TableSchema;
 import org.apache.tsfile.read.TsFileSequenceReader;
-import org.apache.tsfile.read.controller.CachedChunkLoaderImpl;
-import org.apache.tsfile.read.controller.MetadataQuerierByFileImpl;
-import org.apache.tsfile.read.query.executor.TableQueryExecutor;
 import org.apache.tsfile.write.record.Tablet;
 
 import java.io.File;
 import java.io.IOException;
 import java.util.Iterator;
-import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Objects;
 
 public class TsFileInsertionEventTableParser extends TsFileInsertionEventParser {
 
-  private final TableQueryExecutor tableQueryExecutor;
+  private final long startTime;
+  private final long endTime;
+  private final TablePattern tablePattern;
 
-  private final Iterator<Map.Entry<String, TableSchema>> filteredTableSchemaIterator;
+  private final PipeMemoryBlock allocatedMemoryBlockForBatchData;
+  private final PipeMemoryBlock allocatedMemoryBlockForChunk;
+  private final PipeMemoryBlock allocatedMemoryBlockForChunkMeta;
+  private final PipeMemoryBlock allocatedMemoryBlockForTableSchemas;
 
   public TsFileInsertionEventTableParser(
       final File tsFile,
@@ -58,16 +60,20 @@ public TsFileInsertionEventTableParser(
     super(null, pattern, startTime, endTime, pipeTaskMeta, sourceEvent);
 
     try {
+      this.allocatedMemoryBlockForChunk =
+          PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+      this.allocatedMemoryBlockForBatchData =
+          PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+      this.allocatedMemoryBlockForChunkMeta =
+          PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+      this.allocatedMemoryBlockForTableSchemas =
+          PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+
+      this.startTime = startTime;
+      this.endTime = endTime;
+      this.tablePattern = pattern;
+
       tsFileSequenceReader = new TsFileSequenceReader(tsFile.getPath(), true, true);
-      filteredTableSchemaIterator =
-          tsFileSequenceReader.getTableSchemaMap().entrySet().stream()
-              .filter(entry -> Objects.isNull(pattern) || pattern.matchesTable(entry.getKey()))
-              .iterator();
-      tableQueryExecutor =
-          new TableQueryExecutor(
-              new MetadataQuerierByFileImpl(tsFileSequenceReader),
-              new CachedChunkLoaderImpl(tsFileSequenceReader),
-              TableQueryExecutor.TableQueryOrdering.DEVICE);
     } catch (final Exception e) {
       close();
       throw e;
@@ -79,29 +85,35 @@ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
     return () ->
         new Iterator<TabletInsertionEvent>() {
 
-          private TsFileInsertionEventTableParserTabletIterator tabletIterator = null;
+          private TsFileInsertionEventTableParserTabletIterator tabletIterator;
 
           @Override
           public boolean hasNext() {
-            while (tabletIterator == null || !tabletIterator.hasNext()) {
-              if (!filteredTableSchemaIterator.hasNext()) {
-                close();
-                return false;
-              }
-
-              final Map.Entry<String, TableSchema> entry = filteredTableSchemaIterator.next();
-
-              try {
+            try {
+              if (tabletIterator == null) {
                 tabletIterator =
                     new TsFileInsertionEventTableParserTabletIterator(
-                        tableQueryExecutor, entry.getKey(), entry.getValue(), startTime, endTime);
-              } catch (final Exception e) {
+                        tsFileSequenceReader,
+                        entry ->
+                            Objects.isNull(tablePattern)
+                                || tablePattern.matchesTable(entry.getKey()),
+                        allocatedMemoryBlockForTablet,
+                        allocatedMemoryBlockForBatchData,
+                        allocatedMemoryBlockForChunk,
+                        allocatedMemoryBlockForChunkMeta,
+                        allocatedMemoryBlockForTableSchemas,
+                        startTime,
+                        endTime);
+              }
+              if (!tabletIterator.hasNext()) {
                 close();
-                throw new PipeException("failed to create TsFileInsertionDataTabletIterator", e);
+                return false;
               }
+              return true;
+            } catch (Exception e) {
+              close();
+              throw new PipeException("Error while parsing tsfile insertion event", e);
             }
-
-            return true;
           }
 
           @Override
@@ -174,4 +186,25 @@ public TabletInsertionEvent next() {
           }
         };
   }
+
+  @Override
+  public void close() {
+    super.close();
+
+    if (allocatedMemoryBlockForBatchData != null) {
+      allocatedMemoryBlockForBatchData.close();
+    }
+
+    if (allocatedMemoryBlockForChunk != null) {
+      allocatedMemoryBlockForChunk.close();
+    }
+
+    if (allocatedMemoryBlockForChunkMeta != null) {
+      allocatedMemoryBlockForChunkMeta.close();
+    }
+
+    if (allocatedMemoryBlockForTableSchemas != null) {
+      allocatedMemoryBlockForTableSchemas.close();
+    }
+  }
 }