feat(load): route decoded tsfile pieces through consensus

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -1160,6 +1160,9 @@ public final class DataNodeQueryMessages {
       "Start load TsFile {} locally.";
   public static final String LOAD_ALL_FAILED_TSFILES_ARE_CONVERTED_TO_TABLETS =
       "Load: all failed TsFiles are converted to tablets and inserted.";
+  public static final String
+      LOG_LOAD_FAILED_TO_LOAD_SOME_TSFILES_BY_CONVERTING_THEM_INTO_TABLETS_FAILED_TSFILES_ARG_7D9DB9C3 =
+          "Load: failed to load some TsFiles by converting them into tablets. Failed TsFiles: %s";
 
   // --- Plan / Statement ---
 
@@ -2682,6 +2685,8 @@ public final class DataNodeQueryMessages {
       "LoadTsFileScheduler: Region migration was detected during loading TsFile {}, will convert to insertion to avoid data loss";
   public static final String LOAD_TSFILE_ARG_SUCCESSFULLY_LOAD_PROCESS_ARG_ARG =
       "Load TsFile {} Successfully, load process [{}/{}]";
+  public static final String LOG_LOAD_BATCH_FINISHED_DELETING_ARG_SOURCE_TSFILES_AFTER_LOAD_D5EE56E9 =
+      "LOAD batch finished, deleting %d source TsFiles after load";
   public static final String CAN_NOT_LOAD_TSFILE_ARG_LOAD_PROCESS_ARG_ARG =
       "Can not Load TsFile {}, load process [{}/{}]";
   public static final String LOAD_TSFILE_S_FAILED_WILL_TRY_TO_CONVERT_TO_TABLETS_AND_INSERT_FAILED_TSFILES_ARG =
@@ -2692,6 +2697,8 @@ public final class DataNodeQueryMessages {
       "Parse or send TsFile %s error.";
   public static final String DISPATCH_ONE_PIECE_TO_REPLICASET_ARG_ERROR_RESULT_STATUS_CODE_ARG =
       "Dispatch one piece to ReplicaSet {} error. Result status code {}. ";
+  public static final String LOG_LOAD_CONSENSUS_SUBMIT_TRANSIENT_FAILURE_RETRY_D7E1D9A6 =
+      "Transient failure while submitting LOAD consensus {} (load {}) to {}, will retry ({}/{}): {}";
   public static final String RESULT_STATUS_MESSAGE_ARG_DISPATCH_PIECE_NODE_ERROR_PERCENT_NARG =
       "Result status message {}. Dispatch piece node error:%n{}";
   public static final String SUB_STATUS_CODE_ARG_SUB_STATUS_MESSAGE_ARG =
@@ -3793,6 +3800,7 @@ private DataNodeQueryMessages() {}
   public static final String EXCEPTION_THE_SECOND_ARGUMENT_OF_PERCENTILE_FUNCTION_PERCENTAGE_MUST_BE_A_DOUBLE_LITERAL_D9464B46 = "The second argument of 'percentile' function percentage must be a double literal";
   public static final String EXCEPTION_DATA_TYPE_MISMATCH_FOR_MEASUREMENT_ARGARGARG_TYPE_IN_TSFILE_ARG_TYPE_IN_IOTDB_ARG_C5BA7DBD = "Data type mismatch for measurement %s%s%s, type in TsFile: %s, type in IoTDB: %s";
   public static final String MESSAGE_FAILED_TO_RELEASE_EXTERNAL_TSFILE_QUERY_RESOURCE_712EE978 = "Failed to release external TsFile query resource";
+  public static final String EXCEPTION_UNKNOWN_LOADTSFILECONSENSUSOP_ORDINAL_ARG_62848FC2 = "Unknown LoadTsFileConsensusOp ordinal: ";
   public static final String EXCEPTION_OUTER_QUERY_TIMEOUT_EXCEEDED_BEFORE_IOTDBLOCAL_QUERY_STARTS_800BFA63 = "Outer query timeout exceeded before IoTDBLocal query starts";
   public static final String MESSAGE_FAILED_TO_CLOSE_UDF_RESULT_SET_AT_INDEX_ARG_A293B7EC = "Failed to close UDF result set at index {}";
   public static final String EXCEPTION_INTERNAL_QUERY_EXECUTION_NOT_FOUND_62642542 = "Internal query execution not found";
@@ -3812,5 +3820,7 @@ private DataNodeQueryMessages() {}
   public static final String EXCEPTION_VISIBLEALIASES_IS_NULL_630B27F1 = "visibleAliases is null";
   public static final String EXCEPTION_HAS_NO_PERMISSION_TO_EXECUTE_ARG_BECAUSE_ONLY_THE_SUPERUSER_CAN_ALTER_HIM_HERSELF_C5902893 =
       "Has no permission to execute %s, because only the superuser can alter him/herself.";
+  public static final String EXCEPTION_LOAD_CONSENSUS_INVALID_PIECE_REF_F3498507 =
+      "Invalid LOAD consensus piece ref: path %s, offset %d, size %d";
 
 }

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java

@@ -29,6 +29,83 @@ private StorageEngineMessages() {}
   // ======================== StorageEngine ========================
 
   public static final String FAIL_TO_RECOVER_WAL = "Fail to recover wal.";
+  public static final String LOG_LOAD_CONSENSUS_WRITE_TO_REGION_ARG_VIA_PROTOCOL_ARG_EBB55042 =
+      "Write LOAD consensus node to region {} via protocol {}";
+  public static final String LOG_LOAD_CONSENSUS_WRITE_TO_REGION_ARG_VIA_PEER_ARG_FAILED_TRYING_NEXT_REPLICA_ARG_39217580 =
+      "LOAD consensus write to region {} via peer {} failed, trying next replica: {}";
+  public static final String LOG_LOAD_CONSENSUS_REFRESH_REPLICA_SET_FAILED_7C244C63 =
+      "Failed to refresh LOAD consensus replica set for region {}, using cached set: {}";
+  public static final String MESSAGE_LOAD_CONSENSUS_PIECE_CHECKSUM_MISMATCH_CF261675 =
+      "LOAD consensus piece checksum mismatch, loadId: %s, pieceIndex: %d";
+  public static final String MESSAGE_LOAD_CONSENSUS_WAL_FLUSH_FAILED_8BE1375A =
+      "Failed to flush LOAD consensus WAL entry";
+  public static final String MESSAGE_LOAD_CONSENSUS_RATIS_NOT_SUPPORTED_D371E344 =
+      "LOAD consensus is not supported on Ratis";
+  public static final String EXCEPTION_LOAD_CONSENSUS_STAGED_FILE_EOF_8743387D =
+      "Unexpected end of file when reading staged piece %s at offset %s.";
+  public static final String MESSAGE_LOAD_CONSENSUS_PIECE_NOT_CONTINUOUS_AFTER_FAILOVER_D6FFAC6C =
+      "LOAD piece %d of load %s cannot be applied because the previous pieces have not all been "
+          + "applied (the staged state may have been lost by a leader failover).";
+  public static final String MESSAGE_LOAD_CONSENSUS_PIECE_DATA_MISSING_AFTER_PULL_8269CB0B =
+      "LOAD piece %d data of load %s is still missing after pulling from the write node.";
+  public static final String LOG_LOAD_CONSENSUS_FORWARD_PIECE_FAILED_34F9EBE7 =
+      "Failed to forward LOAD piece {} of load {} to follower {}: {}";
+  public static final String LOG_LOAD_CONSENSUS_PULL_PIECE_FAILED_AFB003D5 =
+      "Failed to pull LOAD piece {} of load {} from write node {}: {}";
+  public static final String MESSAGE_LOAD_CONSENSUS_PULL_WITHOUT_SOURCE_ENDPOINT_3B20D9E9 =
+      "LOAD pull request has no source endpoint.";
+  public static final String MESSAGE_LOAD_CONSENSUS_PULL_WITHOUT_RETAINED_PIECE_AD3C9D4F =
+      "LOAD piece %d of load %s is not retained on the write node.";
+  public static final String MESSAGE_LOAD_CONSENSUS_PULL_PUSH_BACK_FAILED_1A90C2B9 =
+      "Failed to push LOAD piece %d of load %s back to %s: %s";
+  public static final String LOG_LOAD_CONSENSUS_ABORT_MARKER_FAILED_6A218023 =
+      "Failed to log the LOAD ABORT marker of load {}: {}";
+  public static final String LOG_LOAD_CONSENSUS_COMMIT_MARKER_FAILED_5353A4E5 =
+      "Failed to log the LOAD COMMIT marker of load {}: {}";
+  public static final String EXCEPTION_LOAD_CONSENSUS_PIECE_DATA_MISSING_OR_CHECKSUM_MISMATCH_AFTER_PULL_35F4972E =
+      "LOAD task %s piece %d data is missing or its checksum mismatches after pull.";
+  public static final String LOG_LOAD_CONSENSUS_RETAINED_PIECE_READ_FAILED_0659D19B =
+      "Failed to read retained LOAD piece {} of load {} from {}: {}";
+  public static final String LOG_LOAD_CONSENSUS_RETAINED_PIECE_WRITE_FAILED_99697608 =
+      "Failed to write retained LOAD piece {} of load {} to {}: {}";
+  public static final String LOG_LOAD_CONSENSUS_APPLIED_PIECE_RESTORE_FAILED_5BC74BBA =
+      "Failed to restore the applied LOAD piece entry {} of load {}.";
+  public static final String EXCEPTION_LOAD_CONSENSUS_STAGED_FILE_NOT_CONTINUOUS_F9408C19 =
+      "Staged file %s of load %s is not continuous: expected offset %d but current file length is %d.";
+  public static final String MESSAGE_LOAD_CONSENSUS_PREPARE_WITHOUT_STAGED_DATA_FE8ADC37 =
+      "Cannot prepare load %s because no staged data exists on this node.";
+  public static final String MESSAGE_LOAD_CONSENSUS_ABORT_MARKER_FAILED_16343CF5 =
+      "LOAD abort marker write failed for load %s, staged data is kept for retry: %s";
+  public static final String MESSAGE_LOAD_CONSENSUS_COMMIT_MARKER_FAILED_184F9E59 =
+      "LOAD commit marker write failed for load %s, staged data is kept for retry: %s";
+  public static final String LOG_LOAD_CONSENSUS_RECOVERED_TASK_02824CE6 =
+      "Recovered in-progress LOAD task {} from disk; staged data is kept until COMMIT or ABORT.";
+  public static final String LOG_LOAD_CONSENSUS_RECOVER_TASK_META_FAILED_C39E04BB =
+      "Failed to recover the task meta of LOAD task {}: {}";
+  public static final String LOG_LOAD_CONSENSUS_TASK_META_WRITE_FAILED_5D2420BF =
+      "Failed to persist the task meta of LOAD task {} to {}: {}";
+  public static final String LOG_LOAD_CONSENSUS_TERMINAL_MARKER_WRITE_FAILED_4D6D7433 =
+      "Failed to write the terminal marker of LOAD task {} to {}: {}";
+  public static final String LOG_LOAD_CONSENSUS_RECOVER_TASK_UNRESUMABLE_A159436C =
+      "Cannot resume LOAD task {} from disk: its durable task meta is missing or corrupt; the next "
+          + "command will re-create it.";
+  public static final String EXCEPTION_LOAD_CONSENSUS_STAGED_FILE_SHORT_WRITE_E7392FAD =
+      "Staged file %s of load %s was not fully written: expected %d bytes at offset %d, wrote %d";
+  public static final String EXCEPTION_LOAD_CONSENSUS_PROGRESS_SERIALIZE_FAILED_28EFD091 =
+      "Failed to serialize LOAD progress index for time partition %d";
+  public static final String MESSAGE_LOAD_CONSENSUS_PREPARE_VERIFICATION_FAILED_B3865A82 =
+      "LOAD prepare verification failed for load %s: expected %d pieces with checksum %d, found %d pieces with checksum %d";
+  public static final String EXCEPTION_LOAD_CONSENSUS_STAGED_FILE_INCOMPLETE_1CDE954B =
+      "Staged file %s of load %s is incomplete and cannot be committed.";
+  public static final String LOG_LOAD_CONSENSUS_SNAPSHOT_TAKEN_09A7DD4C =
+      "Snapshotted %d in-progress LOAD task(s) with %d staged file(s) for region %s into %s.";
+  public static final String LOG_LOAD_CONSENSUS_SNAPSHOT_RESTORED_90ABC1BF =
+      "Restored %d in-progress LOAD task(s) with %d staged file(s) from snapshot %s.";
+  public static final String EXCEPTION_LOAD_CONSENSUS_SNAPSHOT_RESTORE_FAILED_F8C29C64 =
+      "Failed to restore LOAD snapshot from %s: %s";
+  public static final String EXCEPTION_LOAD_TSFILE_ALIGNED_VALUE_CHUNK_TIME_CHUNK_EEB00760 =
+      "Cannot attach value chunk of measurement %s in file %s: expected exactly one buffered "
+          + "aligned time chunk, found %d.";
   public static final String STORAGE_ENGINE_FAILED_TO_SET_UP = "Storage engine failed to set up.";
   public static final String SEQ_MEMTABLE_FLUSH_CHECK_THREAD_STARTED = "start sequence memtable timed flush check thread successfully.";
   public static final String UNSEQ_MEMTABLE_FLUSH_CHECK_THREAD_STARTED = "start unsequence memtable timed flush check thread successfully.";
@@ -477,6 +554,8 @@ private StorageEngineMessages() {}
   public static final String CANNOT_CREATE_TSFILE_FOR_WRITING = "Can not create TsFile {} for writing.";
   public static final String CLOSE_TSFILE_IO_WRITER_ERROR = "Close TsFileIOWriter {} error.";
   public static final String CLOSE_MODIFICATION_FILE_ERROR = "Close ModificationFile {} error.";
+  public static final String EXCEPTION_TABLE_ARG_ARG_DOES_NOT_EXIST_WHEN_APPLYING_LOAD_CHUNK_DATA_IT_MAY_HAVE_BEEN_DROPPED_AFTER_THE_LOAD_WAS_ANALYZED_DDB35F93 =
+      "Table '%s.%s' does not exist when applying LOAD chunk data. It may have been dropped after the LOAD was analyzed.";
   public static final String TASK_DIR_NOT_EMPTY_SKIP_DELETE = "Task dir {} is not empty, skip deleting.";
   public static final String LOAD_CLEANUP_TASK_CANCELED = "Load cleanup task {} is canceled.";
   public static final String LOAD_CLEANUP_TASK_STARTS = "Load cleanup task {} starts.";

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -1141,6 +1141,9 @@ public final class DataNodeQueryMessages {
       "开始本地加载 TsFile {}。";
   public static final String LOAD_ALL_FAILED_TSFILES_ARE_CONVERTED_TO_TABLETS =
       "加载：所有失败的 TsFile 已转换为 Tablet 并插入。";
+  public static final String
+      LOG_LOAD_FAILED_TO_LOAD_SOME_TSFILES_BY_CONVERTING_THEM_INTO_TABLETS_FAILED_TSFILES_ARG_7D9DB9C3 =
+          "加载：部分 TsFile 通过转换为 Tablet 仍加载失败。失败的 TsFile：%s";
 
   // --- Plan / Statement ---
 
@@ -3159,6 +3162,8 @@ public final class DataNodeQueryMessages {
       "LoadTsFileScheduler：加载 TsFile {} 期间检测到 Region 迁移，将转换为 insertion 以避免数据丢失。";
   public static final String LOAD_TSFILE_ARG_SUCCESSFULLY_LOAD_PROCESS_ARG_ARG =
       "成功加载 TsFile {}，加载进度 [{}/{}]";
+  public static final String LOG_LOAD_BATCH_FINISHED_DELETING_ARG_SOURCE_TSFILES_AFTER_LOAD_D5EE56E9 =
+      "LOAD 批次结束，删除 %d 个源 TsFile";
   public static final String CAN_NOT_LOAD_TSFILE_ARG_LOAD_PROCESS_ARG_ARG =
 
       "无法加载 TsFile {}，加载进度 [{}/{}]。";
@@ -3174,6 +3179,8 @@ public final class DataNodeQueryMessages {
   public static final String DISPATCH_ONE_PIECE_TO_REPLICASET_ARG_ERROR_RESULT_STATUS_CODE_ARG =
 
       "分发 TsFile 片段到 ReplicaSet {} 出错。结果状态码 {}。 ";
+  public static final String LOG_LOAD_CONSENSUS_SUBMIT_TRANSIENT_FAILURE_RETRY_D7E1D9A6 =
+      "提交 LOAD 共识 {}（load {}）到 {} 时遇到瞬时失败，将重试（{}/{}）：{}";
   public static final String RESULT_STATUS_MESSAGE_ARG_DISPATCH_PIECE_NODE_ERROR_PERCENT_NARG =
 
       "结果状态消息 {}。分发片段节点出错：%n{}";
@@ -4549,6 +4556,7 @@ private DataNodeQueryMessages() {}
   public static final String EXCEPTION_THE_SECOND_ARGUMENT_OF_PERCENTILE_FUNCTION_PERCENTAGE_MUST_BE_A_DOUBLE_LITERAL_D9464B46 = "'percentile' 函数的第二个参数 percentage 必须是 double 字面量";
   public static final String EXCEPTION_DATA_TYPE_MISMATCH_FOR_MEASUREMENT_ARGARGARG_TYPE_IN_TSFILE_ARG_TYPE_IN_IOTDB_ARG_C5BA7DBD = "测点 %s%s%s 的数据类型不匹配，TsFile 中类型：%s，IoTDB 中类型：%s";
   public static final String MESSAGE_FAILED_TO_RELEASE_EXTERNAL_TSFILE_QUERY_RESOURCE_712EE978 = "释放外部 TsFile 查询资源失败";
+  public static final String EXCEPTION_UNKNOWN_LOADTSFILECONSENSUSOP_ORDINAL_ARG_62848FC2 = "未知的 LoadTsFileConsensusOp 序号：";
   public static final String EXCEPTION_OUTER_QUERY_TIMEOUT_EXCEEDED_BEFORE_IOTDBLOCAL_QUERY_STARTS_800BFA63 = "在 IoTDBLocal 查询开始前，外层查询已超时";
   public static final String MESSAGE_FAILED_TO_CLOSE_UDF_RESULT_SET_AT_INDEX_ARG_A293B7EC = "关闭索引 {} 处的 UDF 结果集失败";
   public static final String EXCEPTION_INTERNAL_QUERY_EXECUTION_NOT_FOUND_62642542 = "未找到内部查询执行";
@@ -4569,5 +4577,7 @@ private DataNodeQueryMessages() {}
   public static final String EXCEPTION_VISIBLEALIASES_IS_NULL_630B27F1 = "visibleAliases 不能为空";
   public static final String EXCEPTION_HAS_NO_PERMISSION_TO_EXECUTE_ARG_BECAUSE_ONLY_THE_SUPERUSER_CAN_ALTER_HIM_HERSELF_C5902893 =
       "无权执行 %s，因为只有超级用户可以修改其自身。";
+  public static final String EXCEPTION_LOAD_CONSENSUS_INVALID_PIECE_REF_F3498507 =
+      "无效的 LOAD 共识分片引用：路径 %s，偏移 %d，大小 %d";
 
 }