Merge branch 'master' into update_last_cache_in_load

# Conflicts:
#	iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java

Author: jt2594838

LICENSE

@@ -270,3 +270,18 @@ The following files include code modified from Apache Kafka project.
 
 Project page: https://github.com/apache/kafka
 License: https://github.com/apache/kafka/blob/trunk/LICENSE
+
+--------------------------------------------------------------------------------
+
+The following files include code modified from Stream-Lib project.
+
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/Counter.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/DoublyLinkedList.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/ExternalizableUtil.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/ITopK.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/ListNode2.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/Pair.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/StreamSummary.java
+
+Project page: https://github.com/addthis/stream-lib
+License: https://github.com/addthis/stream-lib/blob/master/LICENSE.txt

example/mqtt-customize/src/main/java/org/apache/iotdb/mqtt/server/CustomizedJsonPayloadFormatter.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.db.protocol.mqtt.TreeMessage;
 
 import io.netty.buffer.ByteBuf;
+import org.apache.commons.lang3.NotImplementedException;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -33,7 +34,7 @@
 public class CustomizedJsonPayloadFormatter implements PayloadFormatter {
 
   @Override
-  public List<Message> format(ByteBuf payload) {
+  public List<Message> format(String topic, ByteBuf payload) {
     // Suppose the payload is a json format
     if (payload == null) {
       return Collections.emptyList();
@@ -54,6 +55,12 @@ public List<Message> format(ByteBuf payload) {
     return ret;
   }
 
+  @Override
+  @Deprecated
+  public List<Message> format(ByteBuf payload) {
+    throw new NotImplementedException();
+  }
+
   @Override
   public String getName() {
     // set the value of mqtt_payload_formatter in iotdb-common.properties as the following string:

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

@@ -168,6 +168,7 @@ protected MppJVMConfig initVMConfig() {
         .setMaxHeapSize(EnvUtils.getIntFromSysVar(DATANODE_MAX_HEAP_SIZE, 256, clusterIndex))
         .setMaxDirectMemorySize(
             EnvUtils.getIntFromSysVar(DATANODE_MAX_DIRECT_MEMORY_SIZE, 256, clusterIndex))
+        .setTimezone("Asia/Shanghai")
         .build();
   }
 

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

@@ -107,7 +107,7 @@ private void errorTest(String sql, String errorMessage) {
   @Test
   public void aiNodeConnectionTest() {
     String sql = "SHOW AINODES";
-    String title = "NodeID,Status,RpcAddress,RpcPort";
+    String title = "NodeID,Status,InternalAddress,InternalPort";
     try (Connection connection = EnvFactory.getEnv().getConnection();
         Statement statement = connection.createStatement()) {
 
@@ -176,6 +176,26 @@ public void ModelOperationTest() {
     }
   }
 
+  @Test
+  public void callInferenceTest2() {
+    String sql =
+        "CALL INFERENCE(_holtwinters, \"select s0 from root.AI.data\", predict_length=6, generateTime=true)";
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+      try (ResultSet resultSet = statement.executeQuery(sql)) {
+        ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+        checkHeader(resultSetMetaData, "Time,output0");
+        int count = 0;
+        while (resultSet.next()) {
+          count++;
+        }
+        assertEquals(6, count);
+      }
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+  }
+
   @Test
   public void callInferenceTest() {
     String sql =

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

@@ -970,7 +970,9 @@ public void testLoadWithConvertOnTypeMismatchForTableModel() throws Exception {
       statement.execute(String.format("create database if not exists %s", SchemaConfig.DATABASE_0));
       statement.execute(String.format("use %s", SchemaConfig.DATABASE_0));
       statement.execute(convert2TableSQL(SchemaConfig.TABLE_0, schemaList1, columnCategories));
-      statement.execute(String.format("load '%s'", file.getAbsolutePath()));
+      statement.execute(
+          String.format(
+              "load '%s' with ('database'='%s')", file.getAbsolutePath(), SchemaConfig.DATABASE_0));
       try (final ResultSet resultSet =
           statement.executeQuery(String.format("select count(*) from %s", SchemaConfig.TABLE_0))) {
         if (resultSet.next()) {

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

@@ -165,7 +165,7 @@ public void testSetDefaultSGLevel() throws SQLException {
         statement.execute("INSERT INTO root.fail(timestamp, s1) VALUES (1, 1)");
       } catch (SQLException e) {
         assertEquals(
-            "509: root.fail is not a legal path, because it is no longer than default sg level: 3",
+            "509: An error occurred when executing getDeviceToDatabase():root.fail is not a legal path, because it is no longer than default sg level: 3",
             e.getMessage());
       }
 
@@ -176,6 +176,7 @@ public void testSetDefaultSGLevel() throws SQLException {
         assertTrue(e.getMessage().contains("Illegal defaultStorageGroupLevel: -1, should >= 1"));
       }
 
+      // Failed updates will not change the files.
       // Failed updates will not change the files.
       assertFalse(
           checkConfigFileContains(
@@ -197,6 +198,15 @@ public void testSetDefaultSGLevel() throws SQLException {
       // the default value should take effect
       Assert.assertEquals("root.a", databases.getString(1));
       assertFalse(databases.next());
+
+      // create timeseries with an illegal path
+      try {
+        statement.execute("CREATE TIMESERIES root.db1.s3 WITH datatype=INT32");
+      } catch (SQLException e) {
+        assertEquals(
+            "509: An error occurred when executing getDeviceToDatabase():root.db1 is not a legal path, because it is no longer than default sg level: 3",
+            e.getMessage());
+      }
     }
   }
 }

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipePermissionIT.java

@@ -309,6 +309,49 @@ public void testReceiverPermission() throws Exception {
             TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
             TestUtils.executeNonQueryWithRetry(receiverEnv, "flush");
           });
+
+      // Alter pipe, skip if no privileges
+      try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+          final Statement statement = connection.createStatement()) {
+        statement.execute("alter pipe testPipe modify sink ('skipif'='no-privileges')");
+      } catch (final SQLException e) {
+        e.printStackTrace();
+        fail(e.getMessage());
+      }
+
+      final String dbName2 = "test2";
+
+      // Write some data
+      if (!TableModelUtils.insertData(dbName2, tbName, 0, 100, senderEnv)) {
+        return;
+      }
+
+      // Shall not be transferred
+      TestUtils.assertDataAlwaysOnEnv(
+          receiverEnv, "count databases", "count,", Collections.singleton("2,"), (String) null);
+
+      if (!TestUtils.tryExecuteNonQueryWithRetry(
+          "information_schema",
+          BaseEnv.TABLE_SQL_DIALECT,
+          receiverEnv,
+          "grant insert,create on database test2 to user testUser")) {
+        return;
+      }
+
+      if (!TableModelUtils.insertData(dbName2, tbName, 100, 200, senderEnv)) {
+        return;
+      }
+
+      // Will finally pass
+      TableModelUtils.assertCountData(
+          dbName2,
+          tbName,
+          100,
+          receiverEnv,
+          o -> {
+            TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
+            TestUtils.executeNonQueryWithRetry(receiverEnv, "flush");
+          });
     }
   }
 }

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

@@ -2775,7 +2775,34 @@ public void exceptionTest() {
 
     tableAssertTestFail(
         "select * from table0 asof (tolerance 1s) left join table1 on table0.time<=table1.time",
-        "Tolerance in ASOF JOIN is only support INNER type now",
+        "Tolerance in ASOF JOIN only supports INNER type now",
+        DATABASE_NAME);
+
+    tableAssertTestFail(
+        "select * from table0 asof right join table1 on table0.time<=table1.time",
+        "ASOF JOIN does not support RIGHT type now",
+        DATABASE_NAME);
+
+    tableAssertTestFail(
+        "select * from table0 asof full join table1 on table0.time<=table1.time",
+        "ASOF JOIN does not support FULL type now",
+        DATABASE_NAME);
+  }
+
+  @Test
+  public void aggregationTableScanWithJoinTest() {
+    expectedHeader = new String[] {"date", "_col1", "date", "_col3"};
+    retArray = new String[] {"1970-01-01T00:00:00.000Z,2,1970-01-01T00:00:00.000Z,2,"};
+    // Join may rename the 'time' column, so we need to ensure the correctness of
+    // AggregationTableScan in this case
+    tableResultSetEqualTest(
+        "select * from ("
+            + "select date_bin(1ms,time) as date,count(*)from table0 group by date_bin(1ms,time)) t0 "
+            + "join ("
+            + "select date_bin(1ms,time) as date,count(*)from table1 where time=0 group by date_bin(1ms,time)) t1 "
+            + "on t0.date = t1.date",
+        expectedHeader,
+        retArray,
         DATABASE_NAME);
   }
 

integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBMaintainAuthIT.java

@@ -165,20 +165,10 @@ public void maintainAuthTest() {
         PASSWORD);
 
     // case 12: show queries
-    // user1 with select on information_schema.queries
-    tableAssertTestFail(
-        "SHOW QUERIES",
-        TSStatusCode.NO_PERMISSION.getStatusCode()
-            + ": Access Denied: No permissions for this operation, only root user is allowed",
-        USER_1,
-        PASSWORD);
-    // user2 without select on information_schema.queries
-    tableAssertTestFail(
-        "SHOW QUERIES",
-        TSStatusCode.NO_PERMISSION.getStatusCode()
-            + ": Access Denied: No permissions for this operation, only root user is allowed",
-        USER_2,
-        PASSWORD);
+    // non-root users can access its own queries
+    expectedHeader =
+        new String[] {"query_id", "start_time", "datanode_id", "elapsed_time", "statement", "user"};
+    tableQueryNoVerifyResultTest("show queries", expectedHeader, USER_2, PASSWORD);
 
     // case 13: kill query
     // user2

integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java

@@ -4165,6 +4165,51 @@ public void approxCountDistinctTest() {
         DATABASE_NAME);
   }
 
+  @Test
+  public void approxMostFrequentTest() {
+    String[] expectedHeader = buildHeaders(7);
+    String[] retArray =
+        new String[] {
+          "{\"50000\":8},{\"30.0\":8},{\"55.0\":8},{\"true\":12},{\"0xcafebabe55\":8},{\"1727158540000\":12},{\"20240924\":20},"
+        };
+    tableResultSetEqualTest(
+        "select approx_most_frequent(s2, 1, 10), approx_most_frequent(s3, 1, 10), approx_most_frequent(s4, 1, 10), approx_most_frequent(s5, 1, 10), approx_most_frequent(s8, 1, 10), approx_most_frequent(s9, 1, 10), approx_most_frequent(s10, 1, 10) from table1",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+
+    expectedHeader = new String[] {"time", "province", "_col2"};
+    retArray =
+        new String[] {
+          "2024-09-24T06:15:30.000Z,beijing,{},",
+          "2024-09-24T06:15:31.000Z,beijing,{\"31000\":2},",
+          "2024-09-24T06:15:35.000Z,beijing,{\"35000\":2},",
+          "2024-09-24T06:15:36.000Z,beijing,{},",
+          "2024-09-24T06:15:40.000Z,beijing,{\"40000\":2},",
+          "2024-09-24T06:15:41.000Z,beijing,{},",
+          "2024-09-24T06:15:46.000Z,beijing,{\"46000\":2},",
+          "2024-09-24T06:15:50.000Z,beijing,{\"50000\":4},",
+          "2024-09-24T06:15:51.000Z,beijing,{},",
+          "2024-09-24T06:15:55.000Z,beijing,{},",
+          "2024-09-24T06:15:30.000Z,shanghai,{},",
+          "2024-09-24T06:15:31.000Z,shanghai,{\"31000\":2},",
+          "2024-09-24T06:15:35.000Z,shanghai,{\"35000\":2},",
+          "2024-09-24T06:15:36.000Z,shanghai,{},",
+          "2024-09-24T06:15:40.000Z,shanghai,{\"40000\":2},",
+          "2024-09-24T06:15:41.000Z,shanghai,{},",
+          "2024-09-24T06:15:46.000Z,shanghai,{\"46000\":2},",
+          "2024-09-24T06:15:50.000Z,shanghai,{\"50000\":4},",
+          "2024-09-24T06:15:51.000Z,shanghai,{},",
+          "2024-09-24T06:15:55.000Z,shanghai,{},",
+        };
+
+    tableResultSetEqualTest(
+        "SELECT time,province,approx_most_frequent(s2, 1, 10) from table1 group by 1,2 order by 2,1",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+  }
+
   @Test
   public void exceptionTest() {
     tableAssertTestFail(
@@ -4211,6 +4256,14 @@ public void exceptionTest() {
         "select approx_count_distinct(province, 'test') from table1",
         "701: Second argument of Aggregate functions [approx_count_distinct] should be numberic type and do not use expression",
         DATABASE_NAME);
+    tableAssertTestFail(
+        "select approx_most_frequent(province, -10, 100) from table1",
+        "701: The second and third argument must be greater than 0, but got k=-10, capacity=100",
+        DATABASE_NAME);
+    tableAssertTestFail(
+        "select approx_most_frequent(province, 'test', 100) from table1",
+        "701: The second and third argument of 'approx_most_frequent' function must be numeric literal",
+        DATABASE_NAME);
   }
 
   // ==================================================================