[To dev/1.3] Implemented fast last query for on prefix path

iotdb-client/isession/src/main/java/org/apache/iotdb/isession/ISession.java

@@ -25,6 +25,7 @@
 import org.apache.iotdb.isession.util.SystemStatus;
 import org.apache.iotdb.isession.util.Version;
 import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.RedirectException;
 import org.apache.iotdb.rpc.StatementExecutionException;
 import org.apache.iotdb.service.rpc.thrift.TSBackupConfigurationResp;
 import org.apache.iotdb.service.rpc.thrift.TSConnectionInfoResp;
@@ -176,6 +177,9 @@ SessionDataSet executeLastDataQuery(List<String> paths, long lastTime, long time
   SessionDataSet executeLastDataQuery(List<String> paths)
       throws StatementExecutionException, IoTDBConnectionException;
 
+  SessionDataSet executeFastLastDataQueryForOnePrefixPath(final List<String> prefixes)
+      throws IoTDBConnectionException, StatementExecutionException, RedirectException;
+
   SessionDataSet executeLastDataQueryForOneDevice(
       String db, String device, List<String> sensors, boolean isLegalPathNodes)
       throws StatementExecutionException, IoTDBConnectionException;

iotdb-client/isession/src/main/java/org/apache/iotdb/isession/pool/ISessionPool.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.isession.util.SystemStatus;
 import org.apache.iotdb.isession.util.Version;
 import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.RedirectException;
 import org.apache.iotdb.rpc.StatementExecutionException;
 import org.apache.iotdb.service.rpc.thrift.TSBackupConfigurationResp;
 import org.apache.iotdb.service.rpc.thrift.TSConnectionInfoResp;
@@ -486,6 +487,9 @@ SessionDataSetWrapper executeLastDataQueryForOneDevice(
       String db, String device, List<String> sensors, boolean isLegalPathNodes)
       throws StatementExecutionException, IoTDBConnectionException;
 
+  SessionDataSetWrapper executeFastLastDataQueryForOnePrefixPath(final List<String> prefixes)
+      throws IoTDBConnectionException, StatementExecutionException, RedirectException;
+
   SessionDataSetWrapper executeAggregationQuery(
       List<String> paths, List<TAggregationType> aggregations)
       throws StatementExecutionException, IoTDBConnectionException;

iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java

@@ -1046,6 +1046,12 @@ public SessionDataSet executeLastDataQuery(List<String> paths)
     return executeLastDataQuery(paths, time, queryTimeoutInMs);
   }
 
+  @Override
+  public SessionDataSet executeFastLastDataQueryForOnePrefixPath(final List<String> prefixes)
+      throws IoTDBConnectionException, StatementExecutionException, RedirectException {
+    return defaultSessionConnection.executeLastDataQueryForOnePrefixPath(prefixes);
+  }
+
   @Override
   public SessionDataSet executeLastDataQueryForOneDevice(
       String db, String device, List<String> sensors, boolean isLegalPathNodes)

iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java

@@ -45,6 +45,7 @@
 import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq;
 import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp;
 import org.apache.iotdb.service.rpc.thrift.TSFastLastDataQueryForOneDeviceReq;
+import org.apache.iotdb.service.rpc.thrift.TSFastLastDataQueryForOnePrefixPathReq;
 import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq;
 import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsOfOneDeviceReq;
 import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsReq;
@@ -442,6 +443,43 @@ protected SessionDataSet executeRawDataQuery(
         zoneId);
   }
 
+  protected SessionDataSet executeLastDataQueryForOnePrefixPath(final List<String> prefixes)
+      throws StatementExecutionException, IoTDBConnectionException, RedirectException {
+    TSFastLastDataQueryForOnePrefixPathReq req =
+        new TSFastLastDataQueryForOnePrefixPathReq(sessionId, prefixes, statementId);
+    req.setFetchSize(session.fetchSize);
+    req.setEnableRedirectQuery(enableRedirect);
+
+    RetryResult<TSExecuteStatementResp> result =
+        callWithReconnect(
+            () -> {
+              req.setSessionId(sessionId);
+              req.setStatementId(statementId);
+              return client.executeFastLastDataQueryForOnePrefixPath(req);
+            });
+    final TSExecuteStatementResp tsExecuteStatementResp = result.getResult();
+
+    if (result.getRetryAttempts() == 0) {
+      RpcUtils.verifySuccessWithRedirection(tsExecuteStatementResp.getStatus());
+    } else {
+      RpcUtils.verifySuccess(tsExecuteStatementResp.getStatus());
+    }
+
+    return new SessionDataSet(
+        "",
+        tsExecuteStatementResp.getColumns(),
+        tsExecuteStatementResp.getDataTypeList(),
+        tsExecuteStatementResp.columnNameIndexMap,
+        tsExecuteStatementResp.getQueryId(),
+        statementId,
+        client,
+        sessionId,
+        tsExecuteStatementResp.queryResult,
+        tsExecuteStatementResp.isIgnoreTimeStamp(),
+        tsExecuteStatementResp.moreData,
+        zoneId);
+  }
+
   protected Pair<SessionDataSet, TEndPoint> executeLastDataQueryForOneDevice(
       String db, String device, List<String> sensors, boolean isLegalPathNodes, long timeOut)
       throws StatementExecutionException, IoTDBConnectionException {

iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java

@@ -3167,6 +3167,33 @@ public SessionDataSetWrapper executeLastDataQuery(List<String> paths)
     return null;
   }
 
+  @Override
+  public SessionDataSetWrapper executeFastLastDataQueryForOnePrefixPath(final List<String> prefixes)
+      throws IoTDBConnectionException, StatementExecutionException {
+    for (int i = 0; i < RETRY; i++) {
+      ISession session = getSession();
+      try {
+        SessionDataSet resp = session.executeFastLastDataQueryForOnePrefixPath(prefixes);
+        SessionDataSetWrapper wrapper = new SessionDataSetWrapper(resp, session, this);
+        occupy(session);
+        return wrapper;
+      } catch (IoTDBConnectionException e) {
+        // TException means the connection is broken, remove it and get a new one.
+        LOGGER.warn("executeLastDataQuery failed", e);
+        cleanSessionAndMayThrowConnectionException(session, i, e);
+      } catch (StatementExecutionException | RuntimeException e) {
+        putBack(session);
+        throw e;
+      } catch (Throwable e) {
+        LOGGER.error(EXECUTE_LASTDATAQUERY_ERROR, e);
+        putBack(session);
+        throw new RuntimeException(e);
+      }
+    }
+    // never go here
+    return null;
+  }
+
   @Override
   public SessionDataSetWrapper executeLastDataQueryForOneDevice(
       String db, String device, List<String> sensors, boolean isLegalPathNodes)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/FastLastHandler.java

@@ -0,0 +1,73 @@
+/*
+ * 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.db.protocol.rest.v2.handler;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.db.protocol.rest.v2.model.ExecutionStatus;
+import org.apache.iotdb.db.protocol.rest.v2.model.PrefixPathList;
+import org.apache.iotdb.db.protocol.session.IClientSession;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.service.rpc.thrift.TSLastDataQueryReq;
+
+import javax.ws.rs.core.Response;
+
+import java.util.ArrayList;
+import java.util.Collections;
+
+public class FastLastHandler {
+
+  public static TSLastDataQueryReq createTSLastDataQueryReq(
+      IClientSession clientSession, PrefixPathList prefixPathList) {
+    TSLastDataQueryReq req = new TSLastDataQueryReq();
+    req.sessionId = clientSession.getId();
+    req.paths =
+        Collections.singletonList(String.join(".", prefixPathList.getPrefixPaths()) + ".**");
+    req.time = Long.MIN_VALUE;
+    req.setLegalPathNodes(true);
+    return req;
+  }
+
+  public static Response buildErrorResponse(TSStatusCode statusCode) {
+    return Response.ok()
+        .entity(
+            new org.apache.iotdb.db.protocol.rest.model.ExecutionStatus()
+                .code(statusCode.getStatusCode())
+                .message(statusCode.name()))
+        .build();
+  }
+
+  public static Response buildExecutionStatusResponse(TSStatus status) {
+    return Response.ok()
+        .entity(new ExecutionStatus().code(status.getCode()).message(status.getMessage()))
+        .build();
+  }
+
+  public static void setupTargetDataSet(
+      org.apache.iotdb.db.protocol.rest.v2.model.QueryDataSet dataSet) {
+    dataSet.addExpressionsItem("Timeseries");
+    dataSet.addExpressionsItem("Value");
+    dataSet.addExpressionsItem("DataType");
+    dataSet.addDataTypesItem("TEXT");
+    dataSet.addDataTypesItem("TEXT");
+    dataSet.addDataTypesItem("TEXT");
+    dataSet.setValues(new ArrayList<>());
+    dataSet.setTimestamps(new ArrayList<>());
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/RequestValidationHandler.java

@@ -20,6 +20,7 @@
 import org.apache.iotdb.db.protocol.rest.v2.model.ExpressionRequest;
 import org.apache.iotdb.db.protocol.rest.v2.model.InsertRecordsRequest;
 import org.apache.iotdb.db.protocol.rest.v2.model.InsertTabletRequest;
+import org.apache.iotdb.db.protocol.rest.v2.model.PrefixPathList;
 import org.apache.iotdb.db.protocol.rest.v2.model.SQL;
 
 import org.apache.commons.lang3.Validate;
@@ -40,6 +41,13 @@ public static void validateSQL(SQL sql) {
     }
   }
 
+  public static void validatePrefixPaths(PrefixPathList prefixPathList) {
+    Objects.requireNonNull(prefixPathList.getPrefixPaths(), "prefix_paths should not be null");
+    if (prefixPathList.getPrefixPaths().isEmpty()) {
+      throw new IllegalArgumentException("prefix_paths should not be empty");
+    }
+  }
+
   public static void validateInsertTabletRequest(InsertTabletRequest insertTabletRequest) {
     Objects.requireNonNull(insertTabletRequest.getTimestamps(), "timestamps should not be null");
     Objects.requireNonNull(insertTabletRequest.getIsAligned(), "is_aligned should not be null");