simplify process

Author: shizy818

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/CteScanOperator.java

@@ -30,7 +30,6 @@
 import org.apache.iotdb.db.utils.cte.CteDataStore;
 import org.apache.iotdb.db.utils.cte.MemoryReader;
 
-import org.apache.tsfile.common.conf.TSFileDescriptor;
 import org.apache.tsfile.read.common.block.TsBlock;
 import org.apache.tsfile.utils.RamUsageEstimator;
 import org.slf4j.Logger;
@@ -45,9 +44,6 @@ public class CteScanOperator extends AbstractSourceOperator {
   private final int dataStoreRefCount;
   private CteDataReader dataReader;
 
-  private final long maxReturnSize =
-      TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes();
-
   public CteScanOperator(
       OperatorContext operatorContext, PlanNodeId sourceId, CteDataStore dataStore) {
     this.operatorContext = operatorContext;
@@ -96,7 +92,8 @@ public long calculateMaxPeekMemory() {
 
   @Override
   public long calculateMaxReturnSize() {
-    return maxReturnSize;
+    // The returned object is a reference to TsBlock in CteDataReader
+    return RamUsageEstimator.NUM_BYTES_OBJECT_REF;
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java

@@ -93,8 +93,8 @@
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.MigrateRegion;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Parameter;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.PipeStatement;
-import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Query;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Prepare;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Query;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ReconstructRegion;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.RelationalAuthorStatement;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.RemoveAINode;
@@ -160,7 +160,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.BlockingDeque;
-import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingDeque;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/CteMaterializer.java

@@ -250,6 +250,17 @@ private List<Integer> adjustColumnIndexMapping(List<Integer> originalIndexList)
       return originalIndexList;
     }
 
+    boolean isSequential = true;
+    for (int i = 0; i < originalIndexList.size(); i++) {
+      if (originalIndexList.get(i) != i) {
+        isSequential = false;
+        break;
+      }
+    }
+    if (isSequential) {
+      return originalIndexList;
+    }
+
     // Create LinkedHashMap to maintain value-position mapping
     Map<Integer, Integer> valueToPositionMap = new LinkedHashMap<>();
     IntStream.range(0, originalIndexList.size())

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java

@@ -256,7 +256,7 @@ private RelationPlan processNamedQuery(Table table, Query namedQuery, Scope scop
     }
 
     if (namedQuery.isMaterialized() && namedQuery.isDone()) {
-      RelationPlan materializedCtePlan = processMaterializedCte(table, scope);
+      RelationPlan materializedCtePlan = processMaterializedCte(table, namedQuery, scope);
       if (materializedCtePlan != null) {
         return materializedCtePlan;
       }
@@ -265,8 +265,8 @@ private RelationPlan processNamedQuery(Table table, Query namedQuery, Scope scop
     return processRegularCte(table, namedQuery, scope);
   }
 
-  private RelationPlan processMaterializedCte(Table table, Scope scope) {
-    CteDataStore dataStore = queryContext.getCteDataStore(table);
+  private RelationPlan processMaterializedCte(Table table, Query query, Scope scope) {
+    CteDataStore dataStore = query.getCteDataStore();
     if (dataStore == null) {
       return null;
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/cte/CteDataStore.java

@@ -53,7 +53,7 @@ public CteDataStore(TableSchema tableSchema, List<Integer> columnIndex2TsBlockCo
 
   public boolean addTsBlock(TsBlock tsBlock) {
     IoTDBConfig iotConfig = IoTDBDescriptor.getInstance().getConfig();
-    long bytesSize = tsBlock.getRetainedSizeInBytes();
+    long bytesSize = tsBlock.getSizeInBytes();
     int rows = tsBlock.getPositionCount();
     if (bytesSize + cachedBytes >= iotConfig.getCteBufferSize()
         || rows + cachedRows >= iotConfig.getMaxRowsInCteBuffer()) {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/CteScanOperatorTest.java

@@ -29,7 +29,6 @@
 import org.apache.iotdb.db.utils.cte.CteDataStore;
 
 import com.google.common.collect.ImmutableList;
-import org.apache.tsfile.common.conf.TSFileDescriptor;
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.read.common.block.TsBlock;
 import org.apache.tsfile.read.common.block.TsBlockBuilder;
@@ -40,6 +39,7 @@
 import org.apache.tsfile.read.common.type.StringType;
 import org.apache.tsfile.read.common.type.TimestampType;
 import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.RamUsageEstimator;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -157,11 +157,9 @@ public void testMemory() throws Exception {
 
     // maxPeekMemory + maxReturnSize + retainedSize
     long maxPeekMemory = cteScanOperator.calculateMaxPeekMemory();
-    assertEquals(
-        TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes(), maxPeekMemory);
+    assertEquals(RamUsageEstimator.NUM_BYTES_OBJECT_REF, maxPeekMemory);
     long maxReturnSize = cteScanOperator.calculateMaxReturnSize();
-    assertEquals(
-        TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes(), maxReturnSize);
+    assertEquals(RamUsageEstimator.NUM_BYTES_OBJECT_REF, maxReturnSize);
     long retainedSize = cteScanOperator.calculateRetainedSizeAfterCallingNext();
     assertEquals(0L, retainedSize);