[Subscription] Release uncommitted tablet payloads with WAL replay

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

@@ -2706,4 +2706,6 @@ private DataNodePipeMessages() {}
       "Failed to read OPC UA server operation limits, use defaults: maxNodesPerWrite={}, maxNodesPerNodeManagement={}";
   public static final String MESSAGE_ARG_SUBSCRIPTION_ENTRY_REQUIRES_ARG_BYTES_ABOVE_THE_CURRENT_PER_QUEUE_MAXIMUM_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_SUBSCRIPTION_MATERIALIZATION_MEMORY_WAL_PROGRESS_HAS_NOT_ADVANCED_AFCBC7FC =
       "[%s] Subscription entry requires %d bytes, above the current per-queue maximum %d bytes (DataNode budget %d bytes). Reduce the write batch/field size or increase subscription materialization memory. WAL progress has not advanced.";
+  public static final String LOG_SUBSCRIPTION_FAILED_TO_REPLAY_DELIVERY_BATCH_ARG_FROM_WAL_RETAINING_UNCOMMITTED_STATE_0654EACD =
+      "Subscription failed to replay delivery batch {} from WAL; retaining uncommitted state";
 }

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

@@ -2531,4 +2531,6 @@ private DataNodePipeMessages() {}
       "读取 OPC UA 服务器操作限制失败，使用默认值：maxNodesPerWrite={}，maxNodesPerNodeManagement={}";
   public static final String MESSAGE_ARG_SUBSCRIPTION_ENTRY_REQUIRES_ARG_BYTES_ABOVE_THE_CURRENT_PER_QUEUE_MAXIMUM_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_SUBSCRIPTION_MATERIALIZATION_MEMORY_WAL_PROGRESS_HAS_NOT_ADVANCED_AFCBC7FC =
       "[%s] 订阅 entry 需要 %d 字节，超过当前队列可用上限 %d 字节（DataNode 预算 %d 字节）。请减小写入 batch/字段大小或增加订阅物化内存。WAL 进度未推进。";
+  public static final String LOG_SUBSCRIPTION_FAILED_TO_REPLAY_DELIVERY_BATCH_ARG_FROM_WAL_RETAINING_UNCOMMITTED_STATE_0654EACD =
+      "Subscription 无法从 WAL 重放交付批次 {}；保留未提交状态";
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java

@@ -104,6 +104,18 @@ public String getDatabaseName() {
     return databaseName;
   }
 
+  /** Freezes the column projection used by a delivered batch for later WAL replay. */
+  ConsensusLogToTabletConverter snapshotForReplay() {
+    return new ConsensusLogToTabletConverter(
+        treePattern, tablePattern, getColumnFilterMatcher(), databaseName);
+  }
+
+  boolean hasSameReplayProjection(final ConsensusLogToTabletConverter other) {
+    return treePattern == other.treePattern
+        && tablePattern == other.tablePattern
+        && fallbackColumnFilterMatcher == other.fallbackColumnFilterMatcher;
+  }
+
   boolean isTableModel() {
     return Objects.nonNull(tablePattern);
   }
@@ -126,6 +138,16 @@ static String safeDeviceIdForLog(final InsertNode node) {
    * DataRegionStateMachine.grabPlanNode()}.
    */
   static InsertNode deserializeToInsertNode(final IndexedConsensusRequest indexedRequest) {
+    return deserializeToInsertNode(indexedRequest, false);
+  }
+
+  /** WAL replay must wait until every fragment, including the seal, is readable. */
+  static InsertNode deserializeCompleteInsertNode(final IndexedConsensusRequest indexedRequest) {
+    return deserializeToInsertNode(indexedRequest, true);
+  }
+
+  private static InsertNode deserializeToInsertNode(
+      final IndexedConsensusRequest indexedRequest, final boolean requireComplete) {
     final List<SearchNode> searchNodes = new ArrayList<>();
     PlanNode nonSearchNode = null;
 
@@ -151,6 +173,9 @@ static InsertNode deserializeToInsertNode(final IndexedConsensusRequest indexedR
             indexedRequest.getSearchIndex(),
             e.getMessage(),
             e);
+        if (requireComplete) {
+          return null;
+        }
         continue;
       }
 
@@ -173,6 +198,9 @@ static InsertNode deserializeToInsertNode(final IndexedConsensusRequest indexedR
     }
 
     if (!searchNodes.isEmpty()) {
+      if (requireComplete && !searchNodes.get(searchNodes.size() - 1).isLastFragment()) {
+        return null;
+      }
       final PlanNode merged = searchNodes.get(0).merge(searchNodes);
       if (merged instanceof InsertNode) {
         final InsertNode mergedInsert = (InsertNode) merged;