[IOTDB-4070] Memory control for Query Operators (apache#6999)

Author: JackieTien97

integration-test/src/main/java/org/apache/iotdb/it/env/DataNodeWrapper.java

@@ -58,6 +58,8 @@ protected void updateConfig(Properties properties) {
     if (this.targetConfigNode != null) {
       properties.setProperty(IoTDBConstant.TARGET_CONFIG_NODES, this.targetConfigNode);
     }
+    properties.setProperty("max_tsblock_size_in_bytes", "1024");
+    properties.setProperty("page_size_in_byte", "1024");
   }
 
   @Override

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBNestedQueryIT.java

@@ -371,6 +371,7 @@ public void testInvalidNestedBuiltInAggregation() {
         statement.executeQuery(query);
       } catch (SQLException e) {
         Assert.assertTrue(
+            e.getMessage(),
             e.getMessage()
                 .contains("The argument of the aggregation function must be a time series."));
       }

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSameMeasurementsDifferentTypesIT.java

@@ -47,7 +47,7 @@
 @Category({LocalStandaloneIT.class, ClusterIT.class})
 public class IoTDBSameMeasurementsDifferentTypesIT {
 
-  private static BaseConfig tsFileConfig = ConfigFactory.getConfig();
+  private static final BaseConfig tsFileConfig = ConfigFactory.getConfig();
   private static int maxNumberOfPointsInPage;
   private static int pageSizeInByte;
   private static int groupSizeInByte;
@@ -88,10 +88,6 @@ private static void insertData() {
     try (Connection connection = EnvFactory.getEnv().getConnection();
         Statement statement = connection.createStatement()) {
 
-      for (String sql : TestConstant.createSql) {
-        statement.execute(sql);
-      }
-
       statement.execute("SET STORAGE GROUP TO root.fans");
       statement.execute("CREATE TIMESERIES root.fans.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE");
       statement.execute("CREATE TIMESERIES root.fans.d1.s0 WITH DATATYPE=INT64, ENCODING=RLE");
@@ -127,29 +123,27 @@ public void selectAllTest() {
       ResultSet resultSet1 = statement1.executeQuery(selectSql);
       int cnt1 = 0;
       while (resultSet1.next() && cnt1 < 5) {
-        StringBuilder builder = new StringBuilder();
-        builder
-            .append(resultSet1.getString(TestConstant.TIMESTAMP_STR))
-            .append(",")
-            .append(resultSet1.getString("root.fans.d0.s0"))
-            .append(",")
-            .append(resultSet1.getString("root.fans.d1.s0"));
-        Assert.assertEquals(retArray[cnt1], builder.toString());
+        String ans =
+            resultSet1.getString(TestConstant.TIMESTAMP_STR)
+                + ","
+                + resultSet1.getString("root.fans.d0.s0")
+                + ","
+                + resultSet1.getString("root.fans.d1.s0");
+        Assert.assertEquals(retArray[cnt1], ans);
         cnt1++;
       }
 
       statement2.setFetchSize(10);
       ResultSet resultSet2 = statement2.executeQuery(selectSql);
       int cnt2 = 0;
       while (resultSet2.next()) {
-        StringBuilder builder = new StringBuilder();
-        builder
-            .append(resultSet2.getString(TestConstant.TIMESTAMP_STR))
-            .append(",")
-            .append(resultSet2.getString("root.fans.d0.s0"))
-            .append(",")
-            .append(resultSet2.getString("root.fans.d1.s0"));
-        Assert.assertEquals(retArray[cnt2], builder.toString());
+        String ans =
+            resultSet2.getString(TestConstant.TIMESTAMP_STR)
+                + ","
+                + resultSet2.getString("root.fans.d0.s0")
+                + ","
+                + resultSet2.getString("root.fans.d1.s0");
+        Assert.assertEquals(retArray[cnt2], ans);
         cnt2++;
       }
       Assert.assertEquals(9, cnt2);
@@ -158,14 +152,13 @@ public void selectAllTest() {
       // function,
       // and the cursor has been moved to the next position, so we should fetch that value first.
       do {
-        StringBuilder builder = new StringBuilder();
-        builder
-            .append(resultSet1.getString(TestConstant.TIMESTAMP_STR))
-            .append(",")
-            .append(resultSet1.getString("root.fans.d0.s0"))
-            .append(",")
-            .append(resultSet1.getString("root.fans.d1.s0"));
-        Assert.assertEquals(retArray[cnt1], builder.toString());
+        String ans =
+            resultSet1.getString(TestConstant.TIMESTAMP_STR)
+                + ","
+                + resultSet1.getString("root.fans.d0.s0")
+                + ","
+                + resultSet1.getString("root.fans.d1.s0");
+        Assert.assertEquals(retArray[cnt1], ans);
         cnt1++;
       } while (resultSet1.next());
       // Although the statement2 has the same sql as statement1, they shouldn't affect each other.

integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBSortedShowTimeseriesIT.java

@@ -189,6 +189,7 @@ public void showTimeseriesOrderByHeatTest1() throws ClassNotFoundException {
         count++;
       }
       assertEquals(retArray1.size(), count);
+      resultSet.close();
 
       resultSet = statement.executeQuery("show LATEST timeseries");
       count = 0;
@@ -214,6 +215,7 @@ public void showTimeseriesOrderByHeatTest1() throws ClassNotFoundException {
         count++;
       }
       assertEquals(retArray2.size(), count);
+      resultSet.close();
 
     } catch (Exception e) {
       e.printStackTrace();
@@ -266,6 +268,7 @@ public void showTimeseriesOrderByHeatWithLimitTest() {
         count++;
       }
       assertEquals(retSet.size(), count);
+      resultSet.close();
 
     } catch (Exception e) {
       e.printStackTrace();
@@ -313,6 +316,7 @@ public void showTimeseriesOrderByHeatWithWhereTest() {
         count++;
       }
       assertEquals(retArray.length, count);
+      resultSet.close();
 
     } catch (Exception e) {
       e.printStackTrace();

integration/src/test/java/org/apache/iotdb/db/integration/IoTDBManageTsFileResourceIT.java

@@ -54,8 +54,7 @@
 public class IoTDBManageTsFileResourceIT {
   private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
   private TsFileResourceManager tsFileResourceManager = TsFileResourceManager.getInstance();
-  private double prevTimeIndexMemoryProportion;
-  private double prevTimeIndexMemoryThreshold;
+  private long prevTimeIndexMemoryThreshold;
   private int prevCompactionThreadNum;
 
   private static String[] unSeqSQLs =
@@ -89,16 +88,14 @@ public class IoTDBManageTsFileResourceIT {
   @Before
   public void setUp() throws ClassNotFoundException {
     EnvironmentUtils.envSetUp();
-    prevTimeIndexMemoryProportion = CONFIG.getTimeIndexMemoryProportion();
+    prevTimeIndexMemoryThreshold = CONFIG.getAllocateMemoryForTimeIndex();
     prevCompactionThreadNum = CONFIG.getConcurrentCompactionThread();
     Class.forName(Config.JDBC_DRIVER_NAME);
   }
 
   @After
   public void tearDown() throws Exception {
     EnvironmentUtils.cleanEnv();
-    prevTimeIndexMemoryThreshold =
-        prevTimeIndexMemoryProportion * CONFIG.getAllocateMemoryForRead();
     tsFileResourceManager.setTimeIndexMemoryThreshold(prevTimeIndexMemoryThreshold);
     CONFIG.setConcurrentCompactionThread(prevCompactionThreadNum);
   }

server/src/assembly/resources/conf/iotdb-datanode.properties

@@ -448,10 +448,6 @@ timestamp_precision=ms
 # Datatype: double
 # flush_proportion=0.4
 
-# Ratio of read memory allocated for timeIndex, 0.2 by default
-# Datatype: double
-# time_index_memory_proportion=0.2
-
 # Ratio of write memory allocated for buffered arrays, 0.6 by default
 # Datatype: double
 # buffered_arrays_memory_proportion=0.6
@@ -485,6 +481,10 @@ timestamp_precision=ms
 # Datatype: int
 # io_task_queue_size_for_flushing=10
 
+# If true, we will estimate each query's possible memory footprint before executing it and deny it if its estimated memory exceeds current free memory
+# Datatype: bool
+# enable_query_memory_estimation=true
+
 ####################
 ### Upgrade Configurations
 ####################
@@ -646,9 +646,9 @@ timestamp_precision=ms
 # Datatype: boolean
 # meta_data_cache_enable=true
 
-# Read memory Allocation Ratio: BloomFilterCache, ChunkCache, TimeSeriesMetadataCache, memory used for constructing QueryDataSet and Free Memory Used in Query.
-# The parameter form is a:b:c:d:e, where a, b, c, d and e are integers. for example: 1:1:1:1:1 , 1:100:200:300:400
-# chunk_timeseriesmeta_free_memory_proportion=1:100:200:300:400
+# Read memory Allocation Ratio: BloomFilterCache : ChunkCache : TimeSeriesMetadataCache : Coordinator : Operators : DataExchange : timeIndex in TsFileResourceList : others.
+# The parameter form is a:b:c:d:e:f:g:h, where a, b, c, d, e, f, g and h are integers. for example: 1:1:1:1:1:1:1:1 , 1:100:200:50:200:200:200:50
+# chunk_timeseriesmeta_free_memory_proportion=1:100:200:50:200:200:200:50
 
 ####################
 ### LAST Cache Configuration

server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -137,19 +137,13 @@ public class IoTDBConfig {
   private long allocateMemoryForRead = Runtime.getRuntime().maxMemory() * 3 / 10;
 
   /** Memory allocated for the mtree */
-  private long allocateMemoryForSchema = Runtime.getRuntime().maxMemory() * 1 / 10;
-
-  /** Memory allocated for the read process besides cache */
-  private long allocateMemoryForReadWithoutCache = allocateMemoryForRead * 300 / 1001;
+  private long allocateMemoryForSchema = Runtime.getRuntime().maxMemory() / 10;
 
   private volatile int maxQueryDeduplicatedPathNum = 1000;
 
   /** Ratio of memory allocated for buffered arrays */
   private double bufferedArraysMemoryProportion = 0.6;
 
-  /** Memory allocated proportion for timeIndex */
-  private double timeIndexMemoryProportion = 0.2;
-
   /** Flush proportion for system */
   private double flushProportion = 0.4;
 
@@ -477,6 +471,24 @@ public class IoTDBConfig {
   /** Memory allocated for chunk cache in read process */
   private long allocateMemoryForChunkCache = allocateMemoryForRead * 100 / 1001;
 
+  /** Memory allocated for operators */
+  private long allocateMemoryForCoordinator = allocateMemoryForRead * 50 / 1001;
+
+  /** Memory allocated for operators */
+  private long allocateMemoryForOperators = allocateMemoryForRead * 200 / 1001;
+
+  /** Memory allocated for operators */
+  private long allocateMemoryForDataExchange = allocateMemoryForRead * 200 / 1001;
+
+  /** Memory allocated proportion for timeIndex */
+  private long allocateMemoryForTimeIndex = allocateMemoryForRead * 200 / 1001;
+
+  /**
+   * If true, we will estimate each query's possible memory footprint before executing it and deny
+   * it if its estimated memory exceeds current free memory
+   */
+  private boolean enableQueryMemoryEstimation = true;
+
   /** Whether to enable Last cache */
   private boolean lastCacheEnable = true;
 
@@ -1358,6 +1370,10 @@ void setConcurrentSubRawQueryThread(int concurrentSubRawQueryThread) {
     this.concurrentSubRawQueryThread = concurrentSubRawQueryThread;
   }
 
+  public long getMaxBytesPerQuery() {
+    return allocateMemoryForDataExchange / concurrentQueryThread;
+  }
+
   public int getRawQueryBlockingQueueCapacity() {
     return rawQueryBlockingQueueCapacity;
   }
@@ -1688,14 +1704,6 @@ public void setBufferedArraysMemoryProportion(double bufferedArraysMemoryProport
     this.bufferedArraysMemoryProportion = bufferedArraysMemoryProportion;
   }
 
-  public double getTimeIndexMemoryProportion() {
-    return timeIndexMemoryProportion;
-  }
-
-  public void setTimeIndexMemoryProportion(double timeIndexMemoryProportion) {
-    this.timeIndexMemoryProportion = timeIndexMemoryProportion;
-  }
-
   public double getFlushProportion() {
     return flushProportion;
   }
@@ -1744,14 +1752,6 @@ void setAllocateMemoryForRead(long allocateMemoryForRead) {
     this.allocateMemoryForRead = allocateMemoryForRead;
   }
 
-  public long getAllocateMemoryForReadWithoutCache() {
-    return allocateMemoryForReadWithoutCache;
-  }
-
-  public void setAllocateMemoryForReadWithoutCache(long allocateMemoryForReadWithoutCache) {
-    this.allocateMemoryForReadWithoutCache = allocateMemoryForReadWithoutCache;
-  }
-
   public boolean isEnableExternalSort() {
     return enableExternalSort;
   }
@@ -1963,6 +1963,46 @@ public void setAllocateMemoryForChunkCache(long allocateMemoryForChunkCache) {
     this.allocateMemoryForChunkCache = allocateMemoryForChunkCache;
   }
 
+  public long getAllocateMemoryForCoordinator() {
+    return allocateMemoryForCoordinator;
+  }
+
+  public void setAllocateMemoryForCoordinator(long allocateMemoryForCoordinator) {
+    this.allocateMemoryForCoordinator = allocateMemoryForCoordinator;
+  }
+
+  public long getAllocateMemoryForOperators() {
+    return allocateMemoryForOperators;
+  }
+
+  public void setAllocateMemoryForOperators(long allocateMemoryForOperators) {
+    this.allocateMemoryForOperators = allocateMemoryForOperators;
+  }
+
+  public long getAllocateMemoryForDataExchange() {
+    return allocateMemoryForDataExchange;
+  }
+
+  public void setAllocateMemoryForDataExchange(long allocateMemoryForDataExchange) {
+    this.allocateMemoryForDataExchange = allocateMemoryForDataExchange;
+  }
+
+  public long getAllocateMemoryForTimeIndex() {
+    return allocateMemoryForTimeIndex;
+  }
+
+  public void setAllocateMemoryForTimeIndex(long allocateMemoryForTimeIndex) {
+    this.allocateMemoryForTimeIndex = allocateMemoryForTimeIndex;
+  }
+
+  public boolean isEnableQueryMemoryEstimation() {
+    return enableQueryMemoryEstimation;
+  }
+
+  public void setEnableQueryMemoryEstimation(boolean enableQueryMemoryEstimation) {
+    this.enableQueryMemoryEstimation = enableQueryMemoryEstimation;
+  }
+
   public boolean isLastCacheEnabled() {
     return lastCacheEnable;
   }