[BugFix] Carry forward untouched indexes into the file-bundling bundle on compaction publish (#76105)

Signed-off-by: sevev <qiangzh95@gmail.com>
(cherry picked from commit 35c68c6)

Author: sevev

fe/fe-core/src/main/java/com/starrocks/transaction/PublishVersionDaemon.java

@@ -44,6 +44,7 @@
 import com.starrocks.catalog.Tablet;
 import com.starrocks.common.Config;
 import com.starrocks.common.ErrorCode;
+import com.starrocks.common.NoAliveBackendException;
 import com.starrocks.common.StarRocksException;
 import com.starrocks.common.ThreadPoolManager;
 import com.starrocks.common.util.FrontendDaemon;
@@ -55,12 +56,15 @@
 import com.starrocks.lake.Utils;
 import com.starrocks.lake.compaction.Quantiles;
 import com.starrocks.metric.MetricRepo;
+import com.starrocks.proto.AggregatePublishVersionRequest;
 import com.starrocks.proto.DeleteTxnLogRequest;
 import com.starrocks.proto.DeleteTxnLogResponse;
 import com.starrocks.proto.TabletStatPB;
 import com.starrocks.proto.TxnInfoPB;
+import com.starrocks.proto.TxnTypePB;
 import com.starrocks.rpc.BrpcProxy;
 import com.starrocks.rpc.LakeService;
+import com.starrocks.rpc.RpcException;
 import com.starrocks.server.GlobalStateMgr;
 import com.starrocks.server.RunMode;
 import com.starrocks.server.WarehouseManager;
@@ -597,6 +601,7 @@ public boolean publishPartitionBatch(Database db, long tableId, PartitionPublish
         //  The mapping is shadow index id -> ShadowIndexTxnBatch
         Map<Long, ShadowIndexTxnBatch> shadowIndexTxnBatches = null;
         Set<Tablet> normalTablets = null;
+        List<Tablet> carryForwardTablets = null;
 
         Locker locker = new Locker();
         locker.lockTablesWithIntensiveDbLock(db.getId(), Lists.newArrayList(tableId), LockType.READ);
@@ -626,6 +631,7 @@ public boolean publishPartitionBatch(Database db, long tableId, PartitionPublish
             }
 
             useAggregatePublish = table.isFileBundling();
+            Set<Long> publishedNormalIndexIds = Sets.newHashSet();
             for (int i = 0; i < transactionStates.size(); i++) {
                 TransactionState txnState = transactionStates.get(i);
                 computeResource = txnState.getComputeResource();
@@ -650,9 +656,21 @@ public boolean publishPartitionBatch(Database db, long tableId, PartitionPublish
                     } else {
                         normalTablets = (normalTablets == null) ? Sets.newHashSet() : normalTablets;
                         normalTablets.addAll(index.getTablets());
+                        publishedNormalIndexIds.add(index.getId());
                     }
                 }
             }
+            // File bundling stores the metadata of ALL tablets of a partition version in a single bundle file.
+            // The batched transactions' index views are point-in-time (for lake compaction snapshotted at txn
+            // begin); if a rollup/MV index became visible after they began, none of them touch it, so publishing
+            // the new version would write a bundle WITHOUT that index's tablets and permanently wedge the
+            // partition. Carry forward every currently-visible NORMAL index none of these transactions touched
+            // so the new bundle stays a complete whole-partition snapshot. See collectFileBundlingCarryForwardTablets.
+            if (useAggregatePublish) {
+                carryForwardTablets = collectFileBundlingCarryForwardTablets(
+                        partition.getLatestMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE),
+                        publishedNormalIndexIds);
+            }
         } finally {
             locker.unLockTablesWithIntensiveDbLock(db.getId(), Lists.newArrayList(tableId), LockType.READ);
         }
@@ -696,6 +714,10 @@ public boolean publishPartitionBatch(Database db, long tableId, PartitionPublish
                     Utils.publishVersionBatch(publishTablets, txnInfos,
                             startVersion - 1, endVersion, compactionScores, nodeToTablets,
                             computeResource, tabletStats);
+                } else if (CollectionUtils.isNotEmpty(carryForwardTablets)) {
+                    aggregatePublishWithCarryForward(publishTablets, txnInfos, carryForwardTablets,
+                            startVersion - 1, endVersion, nodeToTablets, computeResource, compactionScores,
+                            tabletStats);
                 } else {
                     Utils.aggregatePublishVersion(publishTablets, txnInfos, startVersion - 1, endVersion,
                             compactionScores, nodeToTablets, computeResource, tabletStats);
@@ -1038,6 +1060,7 @@ private boolean publishPartition(@NotNull Database db, @NotNull TableCommitInfo
         ComputeResource computeResource = txnState.getComputeResource();
         List<Tablet> normalTablets = null;
         List<Tablet> shadowTablets = null;
+        List<Tablet> carryForwardTablets = null;
 
         Locker locker = new Locker();
         locker.lockTablesWithIntensiveDbLock(db.getId(), Lists.newArrayList(tableId), LockType.READ);
@@ -1064,6 +1087,7 @@ private boolean publishPartition(@NotNull Database db, @NotNull TableCommitInfo
             }
             baseVersion = partition.getVisibleVersion();
             List<MaterializedIndex> indexes = txnState.getPartitionLoadedIndexes(table.getId(), partition);
+            Set<Long> publishedNormalIndexIds = Sets.newHashSet();
             for (MaterializedIndex index : indexes) {
                 if (!index.visibleForTransaction(txnId)) {
                     LOG.info("Ignored index {} for transaction {}", table.getIndexNameByMetaId(index.getMetaId()), txnId);
@@ -1075,8 +1099,23 @@ private boolean publishPartition(@NotNull Database db, @NotNull TableCommitInfo
                 } else {
                     normalTablets = (normalTablets == null) ? Lists.newArrayList() : normalTablets;
                     normalTablets.addAll(index.getTablets());
+                    publishedNormalIndexIds.add(index.getId());
                 }
             }
+            // File bundling stores the metadata of ALL tablets of a partition version in a single bundle
+            // file (meta/0_<version>.meta). The index set above comes from the transaction's point-in-time
+            // view (for lake compaction it is snapshotted at txn begin). If a materialized-view/rollup
+            // index became visible after this transaction began, that stale view omits it, so publishing
+            // the new version would write a bundle WITHOUT the rollup index's tablets, dropping them from
+            // this version and permanently wedging the partition's publish queue (a later publish reading
+            // those tablets at this version gets a 404). Carry forward every currently-visible NORMAL index
+            // this transaction did not touch, so the new bundle remains a complete whole-partition snapshot.
+            // Only needed for file bundling; without it each tablet keeps its own per-version metadata file.
+            if (useAggregatePublish) {
+                carryForwardTablets = collectFileBundlingCarryForwardTablets(
+                        partition.getLatestMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE),
+                        publishedNormalIndexIds);
+            }
         } finally {
             locker.unLockTablesWithIntensiveDbLock(db.getId(), Lists.newArrayList(tableId), LockType.READ);
         }
@@ -1092,8 +1131,13 @@ private boolean publishPartition(@NotNull Database db, @NotNull TableCommitInfo
                 // Per-tablet stats from the publish response: first-load row counts for statistics
                 // collection, and range-distribution tablet sizes for real-time reshard triggering.
                 Map<Long, TabletStatPB> tabletStats = new HashMap<>();
-                Utils.publishVersion(normalTablets, txnInfo, baseVersion, txnVersion, compactionScores,
-                        computeResource, tabletStats, useAggregatePublish);
+                if (useAggregatePublish && CollectionUtils.isNotEmpty(carryForwardTablets)) {
+                    aggregatePublishWithCarryForward(normalTablets, Lists.newArrayList(txnInfo), carryForwardTablets,
+                            baseVersion, txnVersion, null, computeResource, compactionScores, tabletStats);
+                } else {
+                    Utils.publishVersion(normalTablets, txnInfo, baseVersion, txnVersion, compactionScores,
+                            computeResource, tabletStats, useAggregatePublish);
+                }
 
                 Quantiles quantiles = Quantiles.compute(compactionScores.values());
                 partitionCommitInfo.setCompactionScore(quantiles);
@@ -1120,6 +1164,67 @@ private boolean publishPartition(@NotNull Database db, @NotNull TableCommitInfo
         }
     }
 
+    // For file bundling only: from the partition's currently-visible indexes, return the tablets of every
+    // NORMAL index that this transaction did not touch (i.e. whose index id is not in publishedNormalIndexIds).
+    // These must be carried forward (empty version bump) into the new version's bundle so it stays a complete
+    // whole-partition snapshot; otherwise a transaction with a stale index view (notably lake compaction
+    // whose loaded-index set is snapshotted at txn begin, before a rollup/MV index became visible) would
+    // publish a bundle missing that index and permanently wedge the partition. Returns null when nothing
+    // needs carrying forward (e.g. a normal load already covers every visible index). Caller must hold the
+    // table read lock while obtaining visibleIndexes.
+    static List<Tablet> collectFileBundlingCarryForwardTablets(List<MaterializedIndex> visibleIndexes,
+                                                               Set<Long> publishedNormalIndexIds) {
+        List<Tablet> carryForwardTablets = null;
+        for (MaterializedIndex index : visibleIndexes) {
+            if (index.getState() == MaterializedIndex.IndexState.SHADOW
+                    || publishedNormalIndexIds.contains(index.getId())) {
+                continue;
+            }
+            carryForwardTablets = (carryForwardTablets == null) ? Lists.newArrayList() : carryForwardTablets;
+            carryForwardTablets.addAll(index.getTablets());
+        }
+        return carryForwardTablets;
+    }
+
+    // File bundling only. Publish the touched tablets (this publish's real transactions) together with the
+    // carry-forward tablets (untouched but currently-visible indexes) in ONE aggregate request, so the whole
+    // partition's metadata for the new version lands in a single bundle file (meta/0_<newVersion>.meta). Two
+    // separate aggregate publishes would each truncate-write that same bundle and lose one of the two sets.
+    //
+    // The carry-forward tablets have no txn log for these transactions, so for every real transaction we emit a
+    // matching no-op empty transaction (no_op_publish=true): the BE bypasses log loading and advances the tablet
+    // version without any data change. The count must match the real transactions because the BE requires
+    // new_version == base_version + txns.size() for a multi-transaction batch (a batch of pre-rollup compactions
+    // can span several versions).
+    static void aggregatePublishWithCarryForward(List<Tablet> touchedTablets, List<TxnInfoPB> txnInfos,
+                                                 List<Tablet> carryForwardTablets, long baseVersion,
+                                                 long newVersion, Map<ComputeNode, List<Long>> nodeToTablets,
+                                                 ComputeResource computeResource,
+                                                 Map<Long, Double> compactionScores,
+                                                 Map<Long, TabletStatPB> tabletStats)
+            throws NoAliveBackendException, RpcException {
+        AggregatePublishVersionRequest request = new AggregatePublishVersionRequest();
+        Utils.createSubRequestForAggregatePublish(touchedTablets, txnInfos, baseVersion, newVersion,
+                nodeToTablets, computeResource, request);
+
+        List<TxnInfoPB> carryForwardTxnInfos = Lists.newArrayListWithCapacity(txnInfos.size());
+        for (TxnInfoPB txnInfo : txnInfos) {
+            TxnInfoPB emptyTxnInfo = new TxnInfoPB();
+            emptyTxnInfo.txnId = -1L;
+            emptyTxnInfo.combinedTxnLog = false;
+            emptyTxnInfo.commitTime = txnInfo.commitTime;
+            emptyTxnInfo.txnType = TxnTypePB.TXN_EMPTY;
+            emptyTxnInfo.gtid = txnInfo.gtid;
+            emptyTxnInfo.noOpPublish = true;
+            carryForwardTxnInfos.add(emptyTxnInfo);
+        }
+        // The carry-forward tablets have no txn log to delete on success, so do not thread nodeToTablets here.
+        Utils.createSubRequestForAggregatePublish(carryForwardTablets, carryForwardTxnInfos, baseVersion, newVersion,
+                null, computeResource, request);
+        Utils.sendAggregatePublishVersionRequest(request, baseVersion, computeResource, compactionScores,
+                tabletStats);
+    }
+
     // Per-partition publishPartition phase breakdown for slow outliers.
     // total = executor_queue + db_lock_wait + fe_prep + rpc, where:
     //   executor_queue = CompletableFuture submit -> lambda entry