Added name check to avoid directory attack for object type

.github/workflows/cluster-it-1c1d1a.yml

@@ -59,5 +59,5 @@ jobs:
         uses: actions/upload-artifact@v4
         with:
           name: cluster-log-ainode-${{ matrix.os }}
-          path: integration-test/target/ainode-logs
+          path: integration-test/target/*-logs
           retention-days: 30

dependencies.json

@@ -32,13 +32,13 @@
     "com.nimbusds:nimbus-jose-jwt",
     "com.nimbusds:oauth2-oidc-sdk",
     "com.sun.istack:istack-commons-runtime",
-    "com.timecho.ratis:ratis-client",
-    "com.timecho.ratis:ratis-common",
-    "com.timecho.ratis:ratis-grpc",
-    "com.timecho.ratis:ratis-metrics-api",
-    "com.timecho.ratis:ratis-proto",
-    "com.timecho.ratis:ratis-server",
-    "com.timecho.ratis:ratis-server-api",
+    "org.apache.ratis:ratis-client",
+    "org.apache.ratis:ratis-common",
+    "org.apache.ratis:ratis-grpc",
+    "org.apache.ratis:ratis-metrics-api",
+    "org.apache.ratis:ratis-proto",
+    "org.apache.ratis:ratis-server",
+    "org.apache.ratis:ratis-server-api",
     "com.zaxxer:HikariCP",
     "commons-cli:commons-cli",
     "commons-codec:commons-codec",

integration-test/src/assembly/mpp-test.xml

@@ -63,7 +63,7 @@
         </fileSet>
         <fileSet>
             <outputDirectory>lib</outputDirectory>
-            <directory>${project.basedir}/../iotdb-core/ainode/dist/</directory>
+            <directory>${project.basedir}/../iotdb-core/ainode/dist/ainode/</directory>
             <fileMode>0755</fileMode>
         </fileSet>
     </fileSets>

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java

@@ -580,8 +580,8 @@ public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
   }
 
   @Override
-  public CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
-    setProperty("default_database_level", String.valueOf(defaultStorageGroupLevel));
+  public CommonConfig setDefaultDatabaseLevel(int defaultDatabaseLevel) {
+    setProperty("default_database_level", String.valueOf(defaultDatabaseLevel));
     return this;
   }
 

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java

@@ -602,9 +602,9 @@ public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
   }
 
   @Override
-  public CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
-    dnConfig.setDefaultStorageGroupLevel(defaultStorageGroupLevel);
-    cnConfig.setDefaultStorageGroupLevel(defaultStorageGroupLevel);
+  public CommonConfig setDefaultDatabaseLevel(int defaultDatabaseLevel) {
+    dnConfig.setDefaultDatabaseLevel(defaultDatabaseLevel);
+    cnConfig.setDefaultDatabaseLevel(defaultDatabaseLevel);
     return this;
   }
 

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AINodeWrapper.java

@@ -25,9 +25,9 @@
 import org.apache.tsfile.external.commons.io.file.PathUtils;
 import org.slf4j.Logger;
 
-import java.io.BufferedWriter;
 import java.io.File;
-import java.io.FileWriter;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.LinkOption;
@@ -37,6 +37,7 @@
 import java.nio.file.StandardCopyOption;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Properties;
 import java.util.stream.Stream;
 
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.AI_NODE_NAME;
@@ -62,15 +63,19 @@ public class AINodeWrapper extends AbstractNodeWrapper {
   public static final String CACHE_BUILT_IN_MODEL_PATH = "/data/ainode/models/weights";
 
   private void replaceAttribute(String[] keys, String[] values, String filePath) {
-    try (BufferedWriter writer = new BufferedWriter(new FileWriter(filePath, true))) {
-      for (int i = 0; i < keys.length; i++) {
-        String line = keys[i] + "=" + values[i];
-        writer.newLine();
-        writer.write(line);
-      }
+    Properties props = new Properties();
+    try (FileInputStream in = new FileInputStream(filePath)) {
+      props.load(in);
+    } catch (IOException e) {
+      logger.warn("Failed to load existing AINode properties from {}, because: ", filePath, e);
+    }
+    for (int i = 0; i < keys.length; i++) {
+      props.setProperty(keys[i], values[i]);
+    }
+    try (FileOutputStream out = new FileOutputStream(filePath)) {
+      props.store(out, "Updated by AINode integration-test env");
     } catch (IOException e) {
-      logger.error(
-          "Failed to set attribute for AINode in file: {} because {}", filePath, e.getMessage());
+      logger.error("Failed to save properties to {}, because:", filePath, e);
     }
   }
 

integration-test/src/main/java/org/apache/iotdb/itbase/constant/TestConstant.java

@@ -168,13 +168,12 @@ public static String recordToInsert(TSRecord record) {
   }
 
   public static String getTestTsFilePath(
-      String logicalStorageGroupName,
-      long VirtualStorageGroupId,
+      String logicalDatabaseName,
+      long VirtualDatabaseId,
       long TimePartitionId,
       long tsFileVersion) {
     String filePath =
-        String.format(
-            TEST_TSFILE_PATH, logicalStorageGroupName, VirtualStorageGroupId, TimePartitionId);
+        String.format(TEST_TSFILE_PATH, logicalDatabaseName, VirtualDatabaseId, TimePartitionId);
     String fileName =
         System.currentTimeMillis()
             + FilePathUtils.FILE_NAME_SEPARATOR
@@ -184,11 +183,8 @@ public static String getTestTsFilePath(
   }
 
   public static String getTestTsFileDir(
-      String logicalStorageGroupName, long VirtualStorageGroupId, long TimePartitionId) {
+      String logicalDatabaseName, long VirtualDatabaseId, long TimePartitionId) {
     return String.format(
-        TestConstant.TEST_TSFILE_PATH,
-        logicalStorageGroupName,
-        VirtualStorageGroupId,
-        TimePartitionId);
+        TestConstant.TEST_TSFILE_PATH, logicalDatabaseName, VirtualDatabaseId, TimePartitionId);
   }
 }

integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java

@@ -186,7 +186,7 @@ CommonConfig setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
 
   CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled);
 
-  default CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
+  default CommonConfig setDefaultDatabaseLevel(int defaultDatabaseLevel) {
     return this;
   }
 

integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestResultSet.java

@@ -781,8 +781,12 @@ public Ref getRef(int columnIndex) {
   }
 
   @Override
-  public Blob getBlob(int columnIndex) {
-    throw new UnsupportedOperationException();
+  public Blob getBlob(int columnIndex) throws SQLException {
+    RequestDelegate<Blob> delegate = createLocalRequestDelegate();
+    for (ResultSet rs : resultSets) {
+      delegate.addRequest(() -> rs.getBlob(columnIndex));
+    }
+    return delegate.requestAllAndCompare();
   }
 
   @Override
@@ -806,8 +810,12 @@ public Ref getRef(String columnLabel) {
   }
 
   @Override
-  public Blob getBlob(String columnLabel) {
-    throw new UnsupportedOperationException();
+  public Blob getBlob(String columnLabel) throws SQLException {
+    RequestDelegate<Blob> delegate = createLocalRequestDelegate();
+    for (ResultSet rs : resultSets) {
+      delegate.addRequest(() -> rs.getBlob(columnLabel));
+    }
+    return delegate.requestAllAndCompare();
   }
 
   @Override

integration-test/src/test/java/org/apache/iotdb/ainode/it/AINodeConcurrentInferenceIT.java

@@ -90,33 +90,6 @@ private static void prepareDataForTableModel() throws SQLException {
     }
   }
 
-  //  @Test
-  public void concurrentCPUCallInferenceTest() throws SQLException, InterruptedException {
-    concurrentCPUCallInferenceTest("timer_xl");
-    concurrentCPUCallInferenceTest("sundial");
-  }
-
-  private void concurrentCPUCallInferenceTest(String modelId)
-      throws SQLException, InterruptedException {
-    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TREE_SQL_DIALECT);
-        Statement statement = connection.createStatement()) {
-      final int threadCnt = 4;
-      final int loop = 10;
-      final int predictLength = 96;
-      statement.execute(String.format("LOAD MODEL %s TO DEVICES 'cpu'", modelId));
-      checkModelOnSpecifiedDevice(statement, modelId, "cpu");
-      concurrentInference(
-          statement,
-          String.format(
-              "CALL INFERENCE(%s, 'SELECT s FROM root.AI', predict_length=%d)",
-              modelId, predictLength),
-          threadCnt,
-          loop,
-          predictLength);
-      statement.execute(String.format("UNLOAD MODEL %s FROM DEVICES 'cpu'", modelId));
-    }
-  }
-
   //  @Test
   public void concurrentGPUCallInferenceTest() throws SQLException, InterruptedException {
     concurrentGPUCallInferenceTest("timer_xl");
@@ -150,39 +123,6 @@ private void concurrentGPUCallInferenceTest(String modelId)
   String forecastUDTFSql =
       "SELECT forecast(s, 'MODEL_ID'='%s', 'PREDICT_LENGTH'='%d') FROM root.AI";
 
-  @Test
-  public void concurrentCPUForecastTest() throws SQLException, InterruptedException {
-    concurrentCPUForecastTest("timer_xl", forecastUDTFSql);
-    concurrentCPUForecastTest("sundial", forecastUDTFSql);
-    concurrentCPUForecastTest("timer_xl", forecastTableFunctionSql);
-    concurrentCPUForecastTest("sundial", forecastTableFunctionSql);
-  }
-
-  private void concurrentCPUForecastTest(String modelId, String selectSQL)
-      throws SQLException, InterruptedException {
-    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
-        Statement statement = connection.createStatement()) {
-      final int threadCnt = 4;
-      final int loop = 10;
-      final int predictLength = 96;
-      statement.execute(String.format("LOAD MODEL %s TO DEVICES 'cpu'", modelId));
-      checkModelOnSpecifiedDevice(statement, modelId, "cpu");
-      long startTime = System.currentTimeMillis();
-      concurrentInference(
-          statement,
-          String.format(selectSQL, modelId, predictLength),
-          threadCnt,
-          loop,
-          predictLength);
-      long endTime = System.currentTimeMillis();
-      LOGGER.info(
-          String.format(
-              "Model %s concurrent inference %d reqs (%d threads, %d loops) in CPU takes time: %dms",
-              modelId, threadCnt * loop, threadCnt, loop, endTime - startTime));
-      statement.execute(String.format("UNLOAD MODEL %s FROM DEVICES 'cpu'", modelId));
-    }
-  }
-
   @Test
   public void concurrentGPUForecastTest() throws SQLException, InterruptedException {
     concurrentGPUForecastTest("timer_xl", forecastUDTFSql);
@@ -221,7 +161,7 @@ private void checkModelOnSpecifiedDevice(Statement statement, String modelId, St
       throws SQLException, InterruptedException {
     Set<String> targetDevices = ImmutableSet.copyOf(device.split(","));
     LOGGER.info("Checking model: {} on target devices: {}", modelId, targetDevices);
-    for (int retry = 0; retry < 20; retry++) {
+    for (int retry = 0; retry < 200; retry++) {
       Set<String> foundDevices = new HashSet<>();
       try (final ResultSet resultSet =
           statement.executeQuery(String.format("SHOW LOADED MODELS '%s'", device))) {

integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java

@@ -106,7 +106,7 @@ public void tearDown() {
   @Test
   public void testPartitionInfoSnapshot() throws Exception {
     final String sg = "root.sg";
-    final int storageGroupNum = 10;
+    final int databaseNum = 10;
     final int seriesPartitionSlotsNum = 10;
     final int timePartitionSlotsNum = 10;
 
@@ -118,18 +118,18 @@ public void testPartitionInfoSnapshot() throws Exception {
 
       Set<TCQEntry> expectedCQEntries = createCQs(client);
 
-      for (int i = 0; i < storageGroupNum; i++) {
-        String storageGroup = sg + i;
-        TDatabaseSchema storageGroupSchema = new TDatabaseSchema(storageGroup);
-        TSStatus status = client.setDatabase(storageGroupSchema);
+      for (int i = 0; i < databaseNum; i++) {
+        String database = sg + i;
+        TDatabaseSchema databaseSchema = new TDatabaseSchema(database);
+        TSStatus status = client.setDatabase(databaseSchema);
         assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
 
         for (int j = 0; j < seriesPartitionSlotsNum; j++) {
           TSeriesPartitionSlot seriesPartitionSlot = new TSeriesPartitionSlot(j);
 
           // Create SchemaPartition
           ByteBuffer patternTree =
-              generatePatternTreeBuffer(new String[] {storageGroup + ".d" + j + ".s"});
+              generatePatternTreeBuffer(new String[] {database + ".d" + j + ".s"});
           TSchemaPartitionReq schemaPartitionReq = new TSchemaPartitionReq(patternTree);
           TSchemaPartitionTableResp schemaPartitionTableResp =
               client.getOrCreateSchemaPartitionTable(schemaPartitionReq);
@@ -139,10 +139,8 @@ public void testPartitionInfoSnapshot() throws Exception {
               schemaPartitionTableResp.getStatus().getCode());
           Assert.assertNotNull(schemaPartitionTableResp.getSchemaPartitionTable());
           assertEquals(1, schemaPartitionTableResp.getSchemaPartitionTableSize());
-          Assert.assertNotNull(
-              schemaPartitionTableResp.getSchemaPartitionTable().get(storageGroup));
-          assertEquals(
-              1, schemaPartitionTableResp.getSchemaPartitionTable().get(storageGroup).size());
+          Assert.assertNotNull(schemaPartitionTableResp.getSchemaPartitionTable().get(database));
+          assertEquals(1, schemaPartitionTableResp.getSchemaPartitionTable().get(database).size());
 
           for (int k = 0; k < timePartitionSlotsNum; k++) {
             TTimePartitionSlot timePartitionSlot =
@@ -151,9 +149,9 @@ public void testPartitionInfoSnapshot() throws Exception {
             // Create DataPartition
             Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap =
                 new HashMap<>();
-            partitionSlotsMap.put(storageGroup, new HashMap<>());
+            partitionSlotsMap.put(database, new HashMap<>());
             partitionSlotsMap
-                .get(storageGroup)
+                .get(database)
                 .put(
                     seriesPartitionSlot,
                     new TTimeSlotList()
@@ -167,19 +165,18 @@ public void testPartitionInfoSnapshot() throws Exception {
                 dataPartitionTableResp.getStatus().getCode());
             Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
             assertEquals(1, dataPartitionTableResp.getDataPartitionTableSize());
-            Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable().get(storageGroup));
-            assertEquals(
-                1, dataPartitionTableResp.getDataPartitionTable().get(storageGroup).size());
+            Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable().get(database));
+            assertEquals(1, dataPartitionTableResp.getDataPartitionTable().get(database).size());
             Assert.assertNotNull(
                 dataPartitionTableResp
                     .getDataPartitionTable()
-                    .get(storageGroup)
+                    .get(database)
                     .get(seriesPartitionSlot));
             assertEquals(
                 1,
                 dataPartitionTableResp
                     .getDataPartitionTable()
-                    .get(storageGroup)
+                    .get(database)
                     .get(seriesPartitionSlot)
                     .size());
           }