[Load] Isolate reused TsFile parser memory from Pipe pool

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

@@ -28,8 +28,6 @@
 import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tsfile.parser.table.TsFileInsertionEventTableParser;
 import org.apache.iotdb.db.pipe.metric.overview.PipeTsFileToTabletsMetrics;
-import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
-import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
 import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
 import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
@@ -64,14 +62,16 @@
   protected final PipeInsertionEvent sourceEvent; // used to report progress
 
   // mods entry
-  protected PipeMemoryBlock allocatedMemoryBlockForModifications;
+  protected TsFileInsertionEventParserMemoryBlock allocatedMemoryBlockForModifications;
   protected PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer> currentModifications;
 
   protected long parseStartTimeNano = -1;
   protected boolean parseStartTimeRecorded = false;
   protected boolean parseEndTimeRecorded = false;
 
-  protected final PipeMemoryBlock allocatedMemoryBlockForTablet;
+  protected final TsFileInsertionEventParserMemoryBlock allocatedMemoryBlockForTablet;
+
+  protected final TsFileInsertionEventParserMemoryManager memoryManager;
 
   protected TsFileSequenceReader tsFileSequenceReader;
 
@@ -90,6 +90,36 @@
       final boolean skipIfNoPrivileges,
       final PipeInsertionEvent sourceEvent,
       final boolean isWithMod) {
+    this(
+        tsFile,
+        pipeName,
+        creationTime,
+        treePattern,
+        tablePattern,
+        startTime,
+        endTime,
+        pipeTaskMeta,
+        entity,
+        skipIfNoPrivileges,
+        sourceEvent,
+        isWithMod,
+        TsFileInsertionEventParserMemoryManager.pipe());
+  }
+
+  protected TsFileInsertionEventParser(
+      final File tsFile,
+      final String pipeName,
+      final long creationTime,
+      final TreePattern treePattern,
+      final TablePattern tablePattern,
+      final long startTime,
+      final long endTime,
+      final PipeTaskMeta pipeTaskMeta,
+      final IAuditEntity entity,
+      final boolean skipIfNoPrivileges,
+      final PipeInsertionEvent sourceEvent,
+      final boolean isWithMod,
+      final TsFileInsertionEventParserMemoryManager memoryManager) {
     this.pipeName = pipeName;
     this.creationTime = creationTime;
     this.entity = entity;
@@ -106,9 +136,9 @@
 
     this.pipeTaskMeta = pipeTaskMeta;
     this.sourceEvent = sourceEvent;
+    this.memoryManager = memoryManager;
 
-    this.allocatedMemoryBlockForTablet =
-        PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+    this.allocatedMemoryBlockForTablet = memoryManager.forceAllocateForTabletWithRetry(0);
 
     LOGGER.debug(
         DataNodePipeMessages.TSFILE_HAS_INITIALIZED_PIPENAME_CREATION_TIME_PATTERN,
@@ -180,7 +210,7 @@
   protected void releaseTabletMemoryBlock() {
     if (allocatedMemoryBlockForTablet != null
         && allocatedMemoryBlockForTablet.getMemoryUsageInBytes() > 0) {
-      PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForTablet, 0);
+      allocatedMemoryBlockForTablet.forceResize(0);
     }
   }
 

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

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.event.common.tsfile.parser;
+
+/**
+ * Memory block used by a tsfile parser.
+ *
+ * <p>The parser is shared by Pipe and Load. Keeping the block behind this small interface allows
+ * the same parsing code to use the owning subsystem's memory pool instead of hard-coding the Pipe
+ * pool.
+ */
+public interface TsFileInsertionEventParserMemoryBlock extends AutoCloseable {
+
+  long getMemoryUsageInBytes();
+
+  void forceResize(long newSizeInBytes);
+
+  @Override
+  void close();
+}

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

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.event.common.tsfile.parser;
+
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
+
+/** Allocates parser working memory from the pool owned by the caller. */
+public interface TsFileInsertionEventParserMemoryManager {
+
+  TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(long sizeInBytes);
+
+  TsFileInsertionEventParserMemoryBlock forceAllocate(long sizeInBytes);
+
+  static TsFileInsertionEventParserMemoryManager pipe() {
+    return PipeHolder.INSTANCE;
+  }
+
+  final class PipeHolder {
+    private static final TsFileInsertionEventParserMemoryManager INSTANCE =
+        new TsFileInsertionEventParserMemoryManager() {
+          @Override
+          public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(
+              final long sizeInBytes) {
+            return new PipeBlock(
+                PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(sizeInBytes));
+          }
+
+          @Override
+          public TsFileInsertionEventParserMemoryBlock forceAllocate(final long sizeInBytes) {
+            return new PipeBlock(PipeDataNodeResourceManager.memory().forceAllocate(sizeInBytes));
+          }
+        };
+  }
+
+  final class PipeBlock implements TsFileInsertionEventParserMemoryBlock {
+    private final PipeMemoryBlock delegate;
+
+    private PipeBlock(final PipeMemoryBlock delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public long getMemoryUsageInBytes() {
+      return delegate.getMemoryUsageInBytes();
+    }
+
+    @Override
+    public void forceResize(final long newSizeInBytes) {
+      PipeDataNodeResourceManager.memory().forceResize(delegate, newSizeInBytes);
+    }
+
+    @Override
+    public void close() {
+      delegate.close();
+    }
+  }
+}

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

@@ -35,9 +35,10 @@
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool;
 import org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParser;
+import org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
+import org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryManager;
 import org.apache.iotdb.db.pipe.event.common.tsfile.parser.util.ModsOperationUtil;
 import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
-import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
 import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
 import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
@@ -74,7 +75,7 @@
   private static final Logger LOGGER =
       LoggerFactory.getLogger(TsFileInsertionEventQueryParser.class);
 
-  private final PipeMemoryBlock allocatedMemoryBlock;
+  private final TsFileInsertionEventParserMemoryBlock allocatedMemoryBlock;
   private final TsFileReader tsFileReader;
 
   private final Iterator<Map.Entry<IDeviceID, List<String>>> deviceMeasurementsMapIterator;
@@ -166,6 +167,39 @@
       final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
       final boolean isWithMod)
       throws IOException, IllegalPathException {
+    this(
+        pipeName,
+        creationTime,
+        tsFile,
+        pattern,
+        startTime,
+        endTime,
+        pipeTaskMeta,
+        sourceEvent,
+        entity,
+        skipIfNoPrivileges,
+        deviceIsAlignedMap,
+        deviceMeasurementsMapOverride,
+        isWithMod,
+        TsFileInsertionEventParserMemoryManager.pipe());
+  }
+
+  public TsFileInsertionEventQueryParser(
+      final String pipeName,
+      final long creationTime,
+      final File tsFile,
+      final TreePattern pattern,
+      final long startTime,
+      final long endTime,
+      final PipeTaskMeta pipeTaskMeta,
+      final PipeInsertionEvent sourceEvent,
+      final IAuditEntity entity,
+      final boolean skipIfNoPrivileges,
+      final Map<IDeviceID, Boolean> deviceIsAlignedMap,
+      final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
+      final boolean isWithMod,
+      final TsFileInsertionEventParserMemoryManager memoryManager)
+      throws IOException, IllegalPathException {
     super(
         tsFile,
         pipeName,
@@ -178,16 +212,16 @@
         entity,
         skipIfNoPrivileges,
         sourceEvent,
-        isWithMod);
+        isWithMod,
+        memoryManager);
 
     try {
       currentModifications =
           isWithMod
               ? ModsOperationUtil.loadModificationsFromTsFile(tsFile)
               : PatternTreeMapFactory.getModsPatternTreeMap();
       allocatedMemoryBlockForModifications =
-          PipeDataNodeResourceManager.memory()
-              .forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
+          memoryManager.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
 
       final PipeTsFileResourceManager tsFileResourceManager = PipeDataNodeResourceManager.tsfile();
       final Map<IDeviceID, List<String>> deviceMeasurementsMap;
@@ -248,8 +282,7 @@
         memoryRequiredInBytes +=
             PipeMemoryWeightUtil.memoryOfIDeviceID2StrList(deviceMeasurementsMap);
       }
-      allocatedMemoryBlock =
-          PipeDataNodeResourceManager.memory().forceAllocate(memoryRequiredInBytes);
+      allocatedMemoryBlock = memoryManager.forceAllocate(memoryRequiredInBytes);
 
       final Iterator<Map.Entry<IDeviceID, List<String>>> iterator =
           deviceMeasurementsMap.entrySet().iterator();
@@ -325,6 +358,25 @@
       final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
       final boolean isWithMod)
       throws IOException, IllegalPathException {
+    this(
+        tsFile,
+        pattern,
+        startTime,
+        endTime,
+        deviceMeasurementsMapOverride,
+        isWithMod,
+        TsFileInsertionEventParserMemoryManager.pipe());
+  }
+
+  public TsFileInsertionEventQueryParser(
+      final File tsFile,
+      final TreePattern pattern,
+      final long startTime,
+      final long endTime,
+      final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
+      final boolean isWithMod,
+      final TsFileInsertionEventParserMemoryManager memoryManager)
+      throws IOException, IllegalPathException {
     this(
         null,
         0,
@@ -338,7 +390,8 @@
         false,
         null,
         deviceMeasurementsMapOverride,
-        isWithMod);
+        isWithMod,
+        memoryManager);
   }
 
   private Map<IDeviceID, List<String>> filterDeviceMeasurementsMapByPattern(

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

@@ -23,9 +23,8 @@
 import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool;
+import org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
 import org.apache.iotdb.db.pipe.event.common.tsfile.parser.util.ModsOperationUtil;
-import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
-import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
 import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
 import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
@@ -70,7 +69,7 @@ public class TsFileInsertionEventQueryParserTabletIterator implements Iterator<T
 
   private final QueryDataSet queryDataSet;
 
-  private final PipeMemoryBlock allocatedBlockForTablet;
+  private final TsFileInsertionEventParserMemoryBlock allocatedBlockForTablet;
 
   // Maintain sorted mods list and current index for each measurement
   private final List<ModsOperationUtil.ModsInfo> measurementModsList;
@@ -83,7 +82,7 @@ public class TsFileInsertionEventQueryParserTabletIterator implements Iterator<T
       final IDeviceID deviceId,
       final List<String> measurements,
       final IExpression timeFilterExpression,
-      final PipeMemoryBlock allocatedBlockForTablet,
+      final TsFileInsertionEventParserMemoryBlock allocatedBlockForTablet,
       final PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer> currentModifications,
       final TabletStringInternPool tabletStringInternPool)
       throws IOException {
@@ -172,8 +171,7 @@ private Tablet buildNextTablet() throws IOException {
                 // Used for tree model
                 deviceIdString, schemas, rowCountAndMemorySize.getLeft());
         if (allocatedBlockForTablet.getMemoryUsageInBytes() < rowCountAndMemorySize.getRight()) {
-          PipeDataNodeResourceManager.memory()
-              .forceResize(allocatedBlockForTablet, rowCountAndMemorySize.getRight());
+          allocatedBlockForTablet.forceResize(rowCountAndMemorySize.getRight());
         }
         this.rowRecord = null; // Clear the saved first row
         isFirstRow = false;