Support auto-sync of tree-to-table views: detect new measurements and add the matching columns

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java

@@ -52,6 +52,7 @@ public enum TSStatusCode {
   DISPATCH_ERROR(306),
   LICENSE_ERROR(307),
   THRIFT_FRAME_OVERSIZE(308),
+  REPEATED_WITH_EXISTING_TASK(309),
 
   // Client,
   REDIRECTION_RECOMMEND(400),

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java

@@ -614,6 +614,9 @@ private ManagerMessages() {}
   public static final String LOG_GOT_IOEXCEPTION_CONSTRUCT_DATABASE_TREE_49436621 = "Got IOException when construct database Tree";
   public static final String LOG_GOT_IOEXCEPTION_DESERIALIZE_TEMPLATE_INFO_49EE617E = "Got IOException when deserialize template info";
   public static final String MESSAGE_MEASUREMENTS_NOT_FOUND_ARG_CANNOT_AUTO_DETECT_980D7D44 = "Measurements not found for %s, cannot auto detect";
+  public static final String EXCEPTION_THE_DUPLICATED_SOURCE_MEASUREMENT_ARG_IS_UNSUPPORTED_YET_873AF566 = "The duplicated source measurement %s is unsupported yet.";
+  public static final String MESSAGE_MULTIPLE_TYPES_ENCOUNTERED_WHEN_AUTO_DETECTING_TYPE_OF_MEASUREMENT_ARG_PLEASE_CHECK_6EDF48DB = "Multiple types encountered when auto detecting type of measurement '%s', please check";
+  public static final String MESSAGE_THE_MEASUREMENTS_ARG_AND_ARG_SHARE_THE_SAME_LOWER_CASE_WHEN_AUTO_DETECTING_TYPE_PLEASE_CHECK_7225D062 = "The measurements %s and %s share the same lower case when auto detecting type, please check";
   public static final String LOG_FAILED_TAKE_SNAPSHOT_BECAUSE_SNAPSHOT_FILE_ARG_ALREADY_EXIST_EB2A6093 = "Failed to take snapshot, because snapshot file [{}] is already exist.";
   public static final String LOG_FAILED_LOAD_SNAPSHOT_SNAPSHOT_FILE_ARG_NOT_EXIST_8828CFBA = "Failed to load snapshot,snapshot file [{}] is not exist.";
   public static final String LOG_YOU_MAYBE_NEED_RENAME_SIMPLE_DIR_0_0_MANUALLY_2A12C5C9 = "you maybe need to rename the simple dir to 0_0 manually.";

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java

@@ -606,6 +606,9 @@ private ManagerMessages() {}
   public static final String LOG_GOT_IOEXCEPTION_CONSTRUCT_DATABASE_TREE_49436621 = "构建数据库树时发生 IOException";
   public static final String LOG_GOT_IOEXCEPTION_DESERIALIZE_TEMPLATE_INFO_49EE617E = "反序列化模板信息时发生 IOException";
   public static final String MESSAGE_MEASUREMENTS_NOT_FOUND_ARG_CANNOT_AUTO_DETECT_980D7D44 = "未找到 %s 的测点，无法自动检测";
+  public static final String EXCEPTION_THE_DUPLICATED_SOURCE_MEASUREMENT_ARG_IS_UNSUPPORTED_YET_873AF566 = "暂不支持重复的源测点 %s。";
+  public static final String MESSAGE_MULTIPLE_TYPES_ENCOUNTERED_WHEN_AUTO_DETECTING_TYPE_OF_MEASUREMENT_ARG_PLEASE_CHECK_6EDF48DB = "自动检测测点 %s 的类型时遇到多种类型，请检查";
+  public static final String MESSAGE_THE_MEASUREMENTS_ARG_AND_ARG_SHARE_THE_SAME_LOWER_CASE_WHEN_AUTO_DETECTING_TYPE_PLEASE_CHECK_7225D062 = "自动检测类型时，测点 %s 和 %s 的小写形式相同，请检查";
   public static final String LOG_FAILED_TAKE_SNAPSHOT_BECAUSE_SNAPSHOT_FILE_ARG_ALREADY_EXIST_EB2A6093 = "获取快照失败，原因：快照文件 [{}] 已存在。";
   public static final String LOG_FAILED_LOAD_SNAPSHOT_SNAPSHOT_FILE_ARG_NOT_EXIST_8828CFBA = "无法加载快照，快照文件 [{}] 不存在。";
   public static final String LOG_YOU_MAYBE_NEED_RENAME_SIMPLE_DIR_0_0_MANUALLY_2A12C5C9 = "可能需要手动将 simple 目录重命名为 0_0。";

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -1515,4 +1515,7 @@ private ProcedureMessages() {}
       "DataNode setSystemStatus RPC 失败：%s：%s：%s； ";
   public static final String EXCEPTION_CONFIGNODE_TRYSETNODESTATUS_FAILED_ARG_ARG_ARG_C9BA78F4 =
       "ConfigNode trySetNodeStatus 失败：%s：%s：%s； ";
+  public static final String
+      LOG_FAILED_TO_DETECT_THE_TREE_SERIES_UNDER_THE_NEW_VIEW_ARG_ARG_ARG_9544E1D9 =
+          "扫描新建视图 %s.%s 下的树序列失败：%s";
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -265,6 +265,8 @@
 import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
 import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq;
 import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TUpdateTree2TableViewsReq;
+import org.apache.iotdb.confignode.rpc.thrift.TUpdateTree2TableViewsResp;
 import org.apache.iotdb.consensus.common.DataSet;
 import org.apache.iotdb.consensus.exception.ConsensusException;
 import org.apache.iotdb.db.schemaengine.template.TemplateAlterOperationType;
@@ -3378,6 +3380,14 @@ public TSStatus alterOrDropTable(final TAlterOrDropTableReq req) {
     }
   }
 
+  @Override
+  public TUpdateTree2TableViewsResp updateTree2TableViews(TUpdateTree2TableViewsReq req) {
+    final TSStatus status = confirmLeader();
+    return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+        ? procedureManager.addColumnForTree2TableViews(req)
+        : new TUpdateTree2TableViewsResp(status);
+  }
+
   @Override
   public TDeleteTableDeviceResp deleteDevice(final TDeleteTableDeviceReq req) {
     final TSStatus status = confirmLeader();

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java

@@ -168,6 +168,8 @@
 import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq;
 import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq;
 import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TUpdateTree2TableViewsReq;
+import org.apache.iotdb.confignode.rpc.thrift.TUpdateTree2TableViewsResp;
 import org.apache.iotdb.consensus.common.DataSet;
 import org.apache.iotdb.rpc.TSStatusCode;
 
@@ -919,6 +921,8 @@ TPermissionInfoResp login(
 
   TSStatus alterOrDropTable(final TAlterOrDropTableReq req);
 
+  TUpdateTree2TableViewsResp updateTree2TableViews(TUpdateTree2TableViewsReq req);
+
   TDeleteTableDeviceResp deleteDevice(final TDeleteTableDeviceReq req);
 
   TShowTableResp showTables(final String database, final boolean isDetails);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java

@@ -39,6 +39,7 @@
 import org.apache.iotdb.commons.path.PathPatternTree;
 import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta;
 import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
+import org.apache.iotdb.commons.schema.table.AlterOrDropTableOperationType;
 import org.apache.iotdb.commons.schema.table.TableNodeStatus;
 import org.apache.iotdb.commons.schema.table.TreeViewSchema;
 import org.apache.iotdb.commons.schema.table.TsTable;
@@ -169,6 +170,9 @@
 import org.apache.iotdb.confignode.rpc.thrift.TRemoveRegionReq;
 import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq;
 import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TUpdateTree2TableViewFailure;
+import org.apache.iotdb.confignode.rpc.thrift.TUpdateTree2TableViewsReq;
+import org.apache.iotdb.confignode.rpc.thrift.TUpdateTree2TableViewsResp;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.db.exception.BatchProcessException;
 import org.apache.iotdb.rpc.RpcUtils;
@@ -215,6 +219,15 @@ public class ProcedureManager {
   private static final String PROCEDURE_TIMEOUT_MESSAGE =
       "Timed out to wait for procedure return. The procedure is still running.";
 
+  /** One addColumnForTree2TableViews call waits at most this long for its procedures to finish. */
+  private static final long ADD_VIEW_COLUMN_WAIT_TIME_OUT_MS = 30_000L;
+
+  /** How many times a request which overlapped with another task is submitted again. */
+  private static final int OVERLAP_RETRY_ROUNDS = 4;
+
+  /** How long a retry round waits for the overlapping task to finish before submitting again. */
+  private static final long OVERLAP_RETRY_WAIT_MS = 2_000L;
+
   private final ConfigManager configManager;
   private ProcedureExecutor<ConfigNodeProcedureEnv> executor;
   private ProcedureScheduler scheduler;
@@ -2268,6 +2281,159 @@ public TSStatus alterTableAddColumn(final TAlterOrDropTableReq req) {
                 false));
   }
 
+  public TUpdateTree2TableViewsResp addColumnForTree2TableViews(TUpdateTree2TableViewsReq req) {
+    if (req == null || !req.isSetViewInfos()) {
+      return new TUpdateTree2TableViewsResp(RpcUtils.getStatus(TSStatusCode.ILLEGAL_PARAMETER));
+    }
+
+    final List<TUpdateTree2TableViewFailure> failedViews = new ArrayList<>();
+    final List<Pair<TAlterOrDropTableReq, Procedure<ConfigNodeProcedureEnv>>> procedures =
+        new ArrayList<>();
+    // The requests which overlapped with another task on the same table, together with the status
+    // of the last submission attempt, see processOverlappingTask.
+    final List<Pair<TAlterOrDropTableReq, TSStatus>> overlappingReqs = new ArrayList<>();
+
+    // Submit every view before waiting, so procedures for different views can run concurrently.
+    for (final TAlterOrDropTableReq addColumnReq : req.getViewInfos()) {
+      if (!isAddViewColumnReq(addColumnReq)) {
+        failedViews.add(illegalAddViewColumnFailure(addColumnReq));
+        continue;
+      }
+
+      final TSStatus submissionStatus = submitAddViewColumn(addColumnReq, procedures, failedViews);
+      if (submissionStatus.getCode() == TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode()) {
+        overlappingReqs.add(new Pair<>(addColumnReq, submissionStatus));
+      }
+    }
+
+    processOverlappingTask(overlappingReqs, procedures, failedViews);
+
+    final long waitDeadline = System.currentTimeMillis() + ADD_VIEW_COLUMN_WAIT_TIME_OUT_MS;
+    for (final Pair<TAlterOrDropTableReq, Procedure<ConfigNodeProcedureEnv>> procedure :
+        procedures) {
+      final long remainingWaitTime = Math.max(0, waitDeadline - System.currentTimeMillis());
+      final TSStatus status = waitingProcedureFinished(procedure.getRight(), remainingWaitTime);
+      if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+        failedViews.add(
+            new TUpdateTree2TableViewFailure(
+                procedure.getLeft().database, procedure.getLeft().tableName, status));
+      }
+    }
+
+    final TSStatus status = failedViews.isEmpty() ? StatusUtils.OK : StatusUtils.INTERNAL_ERROR;
+    final TUpdateTree2TableViewsResp response = new TUpdateTree2TableViewsResp(status);
+    if (!failedViews.isEmpty()) {
+      response.setFailedViews(failedViews);
+    }
+    return response;
+  }
+
+  /** Whether the request carries everything an add-column operation on a view needs. */
+  private static boolean isAddViewColumnReq(final @Nullable TAlterOrDropTableReq addColumnReq) {
+    return Objects.nonNull(addColumnReq)
+        && addColumnReq.isSetDatabase()
+        && addColumnReq.isSetTableName()
+        && addColumnReq.isSetQueryId()
+        && addColumnReq.isSetOperationType()
+        && addColumnReq.getOperationType()
+            == AlterOrDropTableOperationType.ADD_COLUMN.getTypeValue()
+        && addColumnReq.isSetUpdateInfo()
+        && addColumnReq.isSetIsView()
+        && addColumnReq.isIsView();
+  }
+
+  private static TUpdateTree2TableViewFailure illegalAddViewColumnFailure(
+      final TAlterOrDropTableReq addColumnReq) {
+    return new TUpdateTree2TableViewFailure(
+        Objects.isNull(addColumnReq) ? "" : Objects.toString(addColumnReq.getDatabase(), ""),
+        Objects.isNull(addColumnReq) ? "" : Objects.toString(addColumnReq.getTableName(), ""),
+        RpcUtils.getStatus(TSStatusCode.ILLEGAL_PARAMETER));
+  }
+
+  /**
+   * Submits the add-column procedure of one view and settles everything but an overlap with another
+   * task on the same table.
+   */
+  private TSStatus submitAddViewColumn(
+      final TAlterOrDropTableReq addColumnReq,
+      final List<Pair<TAlterOrDropTableReq, Procedure<ConfigNodeProcedureEnv>>> procedures,
+      final List<TUpdateTree2TableViewFailure> failedViews) {
+    final AddViewColumnProcedure procedure;
+    procedure =
+        new AddViewColumnProcedure(
+            addColumnReq.database,
+            addColumnReq.tableName,
+            addColumnReq.queryId,
+            TsTableColumnSchemaUtil.deserializeColumnSchemaList(addColumnReq.updateInfo),
+            false);
+
+    final Pair<TSStatus, Long> submissionResult =
+        executeWithoutDuplicateWithoutWaiting(
+            addColumnReq.database,
+            null,
+            addColumnReq.tableName,
+            null,
+            addColumnReq.queryId,
+            ProcedureType.ADD_VIEW_COLUMN_PROCEDURE,
+            procedure);
+    final TSStatus status = submissionResult.getLeft();
+    if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+      procedures.add(new Pair<>(addColumnReq, procedure));
+    } else if (status.getCode() == TSStatusCode.REPEATED_WITH_EXISTING_TASK.getStatusCode()) {
+      final Procedure<ConfigNodeProcedureEnv> repeatedProcedure =
+          executor.getProcedures().get(submissionResult.getRight());
+      // null means the procedure could have finished at this moment
+      if (Objects.isNull(repeatedProcedure)) {
+        failedViews.add(
+            new TUpdateTree2TableViewFailure(
+                addColumnReq.database,
+                addColumnReq.tableName,
+                RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR)));
+      } else {
+        procedures.add(new Pair<>(addColumnReq, repeatedProcedure));
+      }
+    } else if (status.getCode() != TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode()) {
+      failedViews.add(
+          new TUpdateTree2TableViewFailure(addColumnReq.database, addColumnReq.tableName, status));
+    }
+    return status;
+  }
+
+  /**
+   * Retries the requests which overlapped with another task on the same table, A request which
+   * still overlaps after the last round is reported as a failure, with the status of its last
+   * attempt.
+   */
+  private void processOverlappingTask(
+      final List<Pair<TAlterOrDropTableReq, TSStatus>> overlappingReqs,
+      final List<Pair<TAlterOrDropTableReq, Procedure<ConfigNodeProcedureEnv>>> procedures,
+      final List<TUpdateTree2TableViewFailure> failedViews) {
+    List<Pair<TAlterOrDropTableReq, TSStatus>> pending = overlappingReqs;
+    for (int round = 0; round < OVERLAP_RETRY_ROUNDS && !pending.isEmpty(); round++) {
+      try {
+        Thread.sleep(OVERLAP_RETRY_WAIT_MS);
+      } catch (final InterruptedException e) {
+        Thread.currentThread().interrupt();
+        break;
+      }
+      final List<Pair<TAlterOrDropTableReq, TSStatus>> stillOverlapping =
+          new ArrayList<>(pending.size());
+      for (final Pair<TAlterOrDropTableReq, TSStatus> request : pending) {
+        final TSStatus status = submitAddViewColumn(request.getLeft(), procedures, failedViews);
+        if (status.getCode() == TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode()) {
+          stillOverlapping.add(new Pair<>(request.getLeft(), status));
+        }
+      }
+      pending = stillOverlapping;
+    }
+
+    for (final Pair<TAlterOrDropTableReq, TSStatus> request : pending) {
+      failedViews.add(
+          new TUpdateTree2TableViewFailure(
+              request.getLeft().database, request.getLeft().tableName, request.getRight()));
+    }
+  }
+
   public TSStatus alterTableSetProperties(final TAlterOrDropTableReq req) {
     final boolean isView = req.isSetIsView() && req.isIsView();
     return executeWithoutDuplicate(
@@ -2606,6 +2772,36 @@ public TSStatus executeWithoutDuplicate(
     return waitingProcedureFinished(procedureToWait);
   }
 
+  public Pair<TSStatus, Long> executeWithoutDuplicateWithoutWaiting(
+      final String database,
+      final TsTable table,
+      final String tableName,
+      final @Nullable String newName,
+      final String queryId,
+      final ProcedureType thisType,
+      final Procedure<ConfigNodeProcedureEnv> procedure) {
+    final long procedureId;
+    synchronized (this) {
+      final Pair<Long, Boolean> procedureIdDuplicatePair =
+          checkDuplicateTableTask(database, table, tableName, newName, queryId, thisType);
+      procedureId = procedureIdDuplicatePair.getLeft();
+      if (procedureId == -1) {
+        if (Boolean.TRUE.equals(procedureIdDuplicatePair.getRight())) {
+          TSStatus status =
+              RpcUtils.getStatus(
+                  TSStatusCode.OVERLAP_WITH_EXISTING_TASK,
+                  "Some other task is operating table with same name.");
+          return new Pair<>(status, -1L);
+        }
+        this.executor.submitProcedure(procedure);
+        return new Pair<>(StatusUtils.OK, procedure.getProcId());
+      } else {
+        TSStatus status = RpcUtils.getStatus(TSStatusCode.REPEATED_WITH_EXISTING_TASK, "");
+        return new Pair<>(status, procedureId);
+      }
+    }
+  }
+
   public Pair<Long, Boolean> checkDuplicateTableTask(
       final @Nonnull String database,
       final TsTable table,