Merge origin/master into cli-auto-reconnect; resolve receiveCommands conflict (keep reconnection logic)

Author: miantalha45

distribution/pom.xml

@@ -86,7 +86,6 @@
                                 <descriptor>src/assembly/confignode.xml</descriptor>
                                 <descriptor>src/assembly/cli.xml</descriptor>
                                 <descriptor>src/assembly/library-udf.xml</descriptor>
-                                <descriptor>src/assembly/external-service-impl.xml</descriptor>
                             </descriptors>
                             <finalName>apache-iotdb-${project.version}</finalName>
                         </configuration>

external-service-impl/rest/pom.xml

@@ -153,7 +153,6 @@
         <dependency>
             <groupId>jakarta.ws.rs</groupId>
             <artifactId>jakarta.ws.rs-api</artifactId>
-            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.iotdb</groupId>

external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java

@@ -239,7 +239,7 @@ public Response executeFastLastQueryStatement(
             contentOfQuerySupplier,
             clientSession.getUsername(),
             clientSession.getClientAddress());
-        recordQueries(() -> costTime, contentOfQuerySupplier, t);
+        recordQueries(() -> costTime, contentOfQuerySupplier, t, false);
       }
     }
   }

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

@@ -57,6 +57,7 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -834,4 +835,18 @@ public Process getInstance() {
   public int[] getPortList() {
     return portList;
   }
+
+  public void clearLogContent() throws IOException {
+    Files.newOutputStream(Paths.get(getLogPath()), StandardOpenOption.TRUNCATE_EXISTING).close();
+  }
+
+  public boolean logContains(String content) throws IOException {
+    List<String> lines = Files.readAllLines(Paths.get(getLogPath()), StandardCharsets.UTF_8);
+    for (String line : lines) {
+      if (line.contains(content)) {
+        return true;
+      }
+    }
+    return false;
+  }
 }

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

@@ -26,8 +26,8 @@
 import org.apache.iotdb.isession.SessionDataSet;
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
-import org.apache.iotdb.itbase.category.TableClusterIT;
-import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
 import org.apache.iotdb.rpc.IoTDBConnectionException;
 import org.apache.iotdb.rpc.StatementExecutionException;
 
@@ -87,7 +87,7 @@
 
 @SuppressWarnings("ResultOfMethodCallIgnored")
 @RunWith(IoTDBTestRunner.class)
-@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+@Category({LocalStandaloneIT.class, ClusterIT.class})
 public class IoTDBAlterTimeSeriesTypeIT {
 
   private static final Logger log = LoggerFactory.getLogger(IoTDBAlterTimeSeriesTypeIT.class);

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeIdempotentIT.java

@@ -400,8 +400,6 @@ public void testDropRoleIdempotent() throws Exception {
         Collections.singleton("2,"));
   }
 
-  // Table model
-
   private void testIdempotent(
       final List<String> beforeSqlList,
       final String testSql,

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

@@ -0,0 +1,105 @@
+/*
+ * 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.relational.it.query.recent;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.IOException;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareData;
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData;
+import static org.apache.iotdb.db.it.utils.TestUtils.resultSetEqualTest;
+import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class})
+public class IoTDBDebugQueryIT {
+  private static final String DATABASE_NAME = "test_db";
+  private static final String[] createTableSqls =
+      new String[] {
+        "CREATE DATABASE " + DATABASE_NAME,
+        "USE " + DATABASE_NAME,
+        "create table table1(device string tag, value int32 field)",
+        "insert into table1(time,device,value) values(2020-01-01 00:00:01.000,'d1',1)",
+        "FLUSH",
+      };
+  private static final String[] createTreeSqls =
+      new String[] {
+        "create timeseries root.test.departments.department_id TEXT",
+        "create timeseries root.test.departments.dep_name TEXT",
+        "insert into root.test.departments(time, department_id, dep_name) values(1, 'D001', '研发部')",
+        "FLUSH",
+      };
+  private static DataNodeWrapper dataNodeWrapper;
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    EnvFactory.getEnv().initClusterEnvironment();
+    dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapperList().get(0);
+    prepareTableData(createTableSqls);
+    prepareData(createTreeSqls);
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  @Test
+  public void tableTest() throws IOException {
+    // clear log content to reduce lines spanned in logContains check
+    dataNodeWrapper.clearLogContent();
+
+    String[] expectedHeader = new String[] {"time", "device", "value"};
+    String[] retArray = new String[] {"2020-01-01T00:00:01.000Z,d1,1,"};
+    String sql = "debug select time,device,value from table1";
+    tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
+    assertTrue(dataNodeWrapper.logContains("Cache miss: table1.d1"));
+    assertTrue(dataNodeWrapper.logContains(sql));
+  }
+
+  @Test
+  public void treeTest() throws IOException {
+    // clear log content to reduce lines spanned in logContains check
+    dataNodeWrapper.clearLogContent();
+
+    String[] expectedHeader =
+        new String[] {
+          "Time", "root.test.departments.department_id", "root.test.departments.dep_name"
+        };
+    String[] retArray = new String[] {"1,D001,研发部,"};
+    String sql = "debug select department_id, dep_name from root.test.departments";
+    resultSetEqualTest(sql, expectedHeader, retArray);
+
+    assertTrue(dataNodeWrapper.logContains("Cache miss: root.test.departments"));
+    assertTrue(dataNodeWrapper.logContains(sql));
+  }
+}

iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java

@@ -141,7 +141,6 @@ private static boolean matchesSessionOrStatementFailure(String msg) {
   static final int CODE_ERROR = 1;
 
   static final String ISO8601_ARGS = "disableISO8601";
-  static final List<String> AGGREGRATE_TIME_LIST = new ArrayList<>();
   static final String RPC_COMPRESS_ARGS = "c";
   private static final String RPC_COMPRESS_NAME = "rpcCompressed";
   static final String TIMEOUT_ARGS = "timeout";

iotdb-client/cli/src/main/java/org/apache/iotdb/cli/Cli.java

@@ -237,11 +237,11 @@ private static void executeSql(CliContext ctx) throws TException {
       connection.setQueryTimeout(queryTimeout);
       properties = connection.getServerProperties();
       timestampPrecision = properties.getTimestampPrecision();
-      AGGREGRATE_TIME_LIST.addAll(properties.getSupportedTimeAggregationOperations());
       processCommand(ctx, execute, connection);
       ctx.exit(lastProcessStatus);
     } catch (SQLException e) {
-      ctx.getPrinter().println(IOTDB_ERROR_PREFIX + "Can't execute sql because" + e.getMessage());
+      ctx.getPrinter()
+          .println(IOTDB_ERROR_PREFIX + ": Can't execute sql because " + e.getMessage());
       ctx.exit(CODE_ERROR);
     }
   }

iotdb-client/cli/src/main/java/org/apache/iotdb/tool/tsfile/ImportTsFileRemotely.java

@@ -164,7 +164,7 @@ public void sendHandshake() {
                 "Handshake error with target server ip: %s, port: %s, because: %s.",
                 client.getIpAddress(), client.getPort(), resp.getStatus()));
       } else {
-        client.setTimeout(PipeConfig.getInstance().getPipeConnectorTransferTimeoutMs());
+        client.setTimeout(PipeConfig.getInstance().getPipeSinkTransferTimeoutMs());
         IOT_PRINTER.println(
             String.format(
                 "Handshake success. Target server ip: %s, port: %s",
@@ -232,7 +232,7 @@ public void doTransfer(final File tsFile, final File modFile) throws PipeExcepti
 
   private void transferFilePieces(final File file, final boolean isMultiFile)
       throws PipeException, IOException {
-    final int readFileBufferSize = PipeConfig.getInstance().getPipeConnectorReadFileBufferSize();
+    final int readFileBufferSize = PipeConfig.getInstance().getPipeSinkReadFileBufferSize();
     final byte[] readBuffer = new byte[readFileBufferSize];
     long position = 0;
     try (final RandomAccessFile reader = new RandomAccessFile(file, "r")) {
@@ -299,10 +299,9 @@ private void initClient() {
       this.client =
           new IoTDBSyncClient(
               new ThriftClientProperty.Builder()
-                  .setConnectionTimeoutMs(
-                      PipeConfig.getInstance().getPipeConnectorHandshakeTimeoutMs())
+                  .setConnectionTimeoutMs(PipeConfig.getInstance().getPipeSinkHandshakeTimeoutMs())
                   .setRpcThriftCompressionEnabled(
-                      PipeConfig.getInstance().isPipeConnectorRPCThriftCompressionEnabled())
+                      PipeConfig.getInstance().isPipeSinkRPCThriftCompressionEnabled())
                   .build(),
               getEndPoint().getIp(),
               getEndPoint().getPort(),