Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wanpengfei-git merged 1 commit into from
Jul 10, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 107 additions & 2 deletions fe/fe-core/src/main/java/com/starrocks/transaction/PublishVersionDaemon.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();
Expand All @@ -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);
}
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand All @@ -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);
}
Expand All @@ -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);
Expand All @@ -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
Expand Down
Loading
Loading