Skip to content

Commit 8c261bb

Browse files
Youngwbclaude
Youngwb
and
claude
committed
[Enhancement] Record IVM source version/timestamp ranges in MV task run EXTRA_MESSAGE
Record, for each IVM (incremental) MV refresh, the TVR version range consumed from every base table and the matching Iceberg snapshot commit times into information_schema.task_runs.EXTRA_MESSAGE: - imvSourceVersionRange: {"catalog.db.tbl": {"start": "100", "end": "128"}} - imvSourceTimestampRange: same shape, values are snapshot commit times in epoch millis; endpoints that cannot be resolved (MIN/MAX sentinel versions, expired/unknown snapshots, metadata errors) are omitted, so a table degrades to {} instead of failing the refresh. The ranges are recorded right after MVIVMRefreshProcessor stages the per-table TVR deltas, through the existing best-effort updateTaskRunStatus() path, so recording can never fail a refresh. PCT runs leave both fields empty. Both setters cap entries via MvUtils.shrinkToSize like the other EXTRA_MESSAGE maps. These fields feed the IMV_SOURCE_* columns of the planned MV refresh jobs system table. Co-Authored-By: Claude <noreply@anthropic.com> Signed-off-by: Youngwb <yangwenbo_mailbox@163.com>
1 parent 2d26da3 commit 8c261bb

12 files changed

Lines changed: 407 additions & 0 deletions

File tree

‎fe/fe-core/src/main/java/com/starrocks/connector/CatalogConnectorMetadata.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -212,6 +212,16 @@ public TvrVersionRange getTableVersionRange(String dbName, Table table,
212212
return metadata.getTableVersionRange(dbName, table, startVersion, endVersion);
213213
}
214214

215+
@Override
216+
public Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
217+
ConnectorMetadata metadata = metadataOfTable(table);
218+
if (metadata == null) {
219+
metadata = metadataOfDb(dbName);
220+
}
221+
222+
return metadata.getVersionCommitTimeMillis(dbName, table, version);
223+
}
224+
215225
@Override
216226
public boolean tableExists(ConnectContext context, String dbName, String tblName) {
217227
ConnectorMetadata metadata = metadataOfDb(dbName);

‎fe/fe-core/src/main/java/com/starrocks/connector/ConnectorMetadata.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -194,6 +194,15 @@ default List<TvrTableDeltaTrait> listTableDeltaTraits(String dbName, Table table
194194
return Lists.newArrayList();
195195
}
196196

197+
/**
198+
* Commit time of {@code version} (the table's own version space, e.g. an Iceberg snapshot id)
199+
* in epoch millis, or empty when it cannot be resolved (unknown/expired version, or a format
200+
* with no per-version commit time).
201+
*/
202+
default Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
203+
return Optional.empty();
204+
}
205+
197206
default boolean tableExists(ConnectContext context, String dbName, String tblName) {
198207
return listTableNames(context, dbName).contains(tblName);
199208
}

‎fe/fe-core/src/main/java/com/starrocks/connector/iceberg/IcebergMetadata.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -857,6 +857,12 @@ public List<TvrTableDeltaTrait> listTableDeltaTraits(String dbName, Table table,
857857
return tvrDeltaTraits;
858858
}
859859

860+
@Override
861+
public Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
862+
Snapshot snapshot = ((IcebergTable) table).getNativeTable().snapshot(version);
863+
return snapshot == null ? Optional.empty() : Optional.of(snapshot.timestampMillis());
864+
}
865+
860866
@Override
861867
public TvrVersionRange getTableVersionRange(String dbName, Table table,
862868
Optional<ConnectorTableVersion> startVersion,

‎fe/fe-core/src/main/java/com/starrocks/connector/unified/UnifiedMetadata.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -164,6 +164,12 @@ public TvrVersionRange getTableVersionRange(String dbName, Table table,
164164
return metadata.getTableVersionRange(dbName, table, startVersion, endVersion);
165165
}
166166

167+
@Override
168+
public Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
169+
ConnectorMetadata metadata = metadataOfTable(table);
170+
return metadata.getVersionCommitTimeMillis(dbName, table, version);
171+
}
172+
167173
@Override
168174
public List<String> listDbNames(ConnectContext context) {
169175
return hiveMetadata.listDbNames(context);

‎fe/fe-core/src/main/java/com/starrocks/scheduler/mv/ivm/MVIVMRefreshProcessor.java‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
import com.google.common.annotations.VisibleForTesting;
1818
import com.google.common.base.Joiner;
19+
import com.google.common.collect.ImmutableMap;
1920
import com.google.common.collect.Maps;
2021
import com.google.common.collect.Multimap;
2122
import com.starrocks.catalog.BaseTableInfo;
@@ -50,7 +51,9 @@
5051
import com.starrocks.scheduler.mv.BaseTableSnapshotInfo;
5152
import com.starrocks.scheduler.mv.MVRefreshExecutor;
5253
import com.starrocks.scheduler.mv.MVRefreshProcessor;
54+
import com.starrocks.scheduler.persist.MVTaskRunExtraMessage;
5355
import com.starrocks.server.GlobalStateMgr;
56+
import com.starrocks.server.MetadataMgr;
5457
import com.starrocks.sql.StatementPlanner;
5558
import com.starrocks.sql.analyzer.Analyzer;
5659
import com.starrocks.sql.analyzer.AnalyzerUtils;
@@ -154,9 +157,50 @@ public ProcessExecPlan getProcessExecPlan(TaskRunContext taskRunContext) throws
154157
try (Timer ignored = Tracers.watchScope("MVRefreshPrepareRefreshPlan")) {
155158
insertStmt = prepareRefreshPlan();
156159
}
160+
recordImvSourceRangesOnTaskRun();
157161
return new ProcessExecPlan(Constants.TaskRunState.SUCCESS, mvContext.getExecPlan(), insertStmt);
158162
}
159163

164+
/**
165+
* Record the staged TVR version range and snapshot commit times per base table on the task
166+
* run's extra message, surfaced via information_schema.task_runs.EXTRA_MESSAGE.
167+
* Must stay after prepareRefreshPlan(): recording earlier leaves stale ranges on the task
168+
* run when the hybrid processor falls back to PCT on an IVM planning failure.
169+
*/
170+
private void recordImvSourceRangesOnTaskRun() {
171+
updateTaskRunStatus(status -> {
172+
Map<String, Map<String, String>> versionRanges = Maps.newHashMap();
173+
Map<String, Map<String, String>> timestampRanges = Maps.newHashMap();
174+
for (BaseTableSnapshotInfo snapshotInfo : snapshotBaseTables.values()) {
175+
TvrVersionRange delta = ((TvrTableSnapshotInfo) snapshotInfo).getTvrSnapshot();
176+
if (delta == null) {
177+
continue;
178+
}
179+
BaseTableInfo baseTableInfo = snapshotInfo.getBaseTableInfo();
180+
String tableFullName = baseTableInfo.getReadableString();
181+
// TvrVersion.toString() renders the MIN/MAX sentinels as "MIN"/"MAX"
182+
versionRanges.put(tableFullName, ImmutableMap.of(
183+
"start", delta.from().toString(),
184+
"end", delta.to().toString()));
185+
timestampRanges.put(tableFullName,
186+
resolveCommitTimeRange(baseTableInfo.getDbName(), snapshotInfo.getBaseTable(), delta));
187+
}
188+
MVTaskRunExtraMessage extraMessage = status.getMvTaskRunExtraMessage();
189+
extraMessage.setImvSourceVersionRange(versionRanges);
190+
extraMessage.setImvSourceTimestampRange(timestampRanges);
191+
});
192+
}
193+
194+
private static Map<String, String> resolveCommitTimeRange(String dbName, Table table, TvrVersionRange delta) {
195+
Map<String, String> commitTimes = Maps.newLinkedHashMap();
196+
MetadataMgr metadataMgr = GlobalStateMgr.getCurrentState().getMetadataMgr();
197+
delta.start().flatMap(version -> metadataMgr.getVersionCommitTimeMillis(dbName, table, version))
198+
.ifPresent(time -> commitTimes.put("start", String.valueOf(time)));
199+
delta.end().flatMap(version -> metadataMgr.getVersionCommitTimeMillis(dbName, table, version))
200+
.ifPresent(time -> commitTimes.put("end", String.valueOf(time)));
201+
return commitTimes;
202+
}
203+
160204
@Override
161205
public Constants.TaskRunState execProcessExecPlan(TaskRunContext taskRunContext,
162206
ProcessExecPlan processExecPlan,

‎fe/fe-core/src/main/java/com/starrocks/scheduler/persist/MVTaskRunExtraMessage.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,13 @@ public class MVTaskRunExtraMessage implements Writable {
8282
@SerializedName("pinnedSnapshotIdMap")
8383
private Map<String, Long> pinnedSnapshotIdMap = Maps.newHashMap();
8484

85+
// For IVM refreshes: per base table ("catalog.db.tbl"), the consumed TVR version range
86+
// {start, end} and the matching snapshot commit times in epoch millis (empty when unresolvable).
87+
@SerializedName("imvSourceVersionRange")
88+
private Map<String, Map<String, String>> imvSourceVersionRange = Maps.newHashMap();
89+
@SerializedName("imvSourceTimestampRange")
90+
private Map<String, Map<String, String>> imvSourceTimestampRange = Maps.newHashMap();
91+
8592
public MVTaskRunExtraMessage() {
8693
}
8794

@@ -222,6 +229,24 @@ public void setPinnedSnapshotIdMap(Map<String, Long> pinnedSnapshotIdMap) {
222229
Config.max_mv_task_run_meta_message_values_length);
223230
}
224231

232+
public Map<String, Map<String, String>> getImvSourceVersionRange() {
233+
return imvSourceVersionRange;
234+
}
235+
236+
public void setImvSourceVersionRange(Map<String, Map<String, String>> imvSourceVersionRange) {
237+
this.imvSourceVersionRange = MvUtils.shrinkToSize(imvSourceVersionRange,
238+
Config.max_mv_task_run_meta_message_values_length);
239+
}
240+
241+
public Map<String, Map<String, String>> getImvSourceTimestampRange() {
242+
return imvSourceTimestampRange;
243+
}
244+
245+
public void setImvSourceTimestampRange(Map<String, Map<String, String>> imvSourceTimestampRange) {
246+
this.imvSourceTimestampRange = MvUtils.shrinkToSize(imvSourceTimestampRange,
247+
Config.max_mv_task_run_meta_message_values_length);
248+
}
249+
225250
@Override
226251
public String toString() {
227252
return GsonUtils.GSON.toJson(this);

‎fe/fe-core/src/main/java/com/starrocks/server/MetadataMgr.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -573,6 +573,11 @@ public TvrVersionRange getTableVersionRange(String dbName, Table table,
573573
.orElse(TvrTableSnapshot.empty());
574574
}
575575

576+
public Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
577+
Optional<ConnectorMetadata> connectorMetadata = getOptionalMetadata(table.getCatalogName());
578+
return connectorMetadata.flatMap(metadata -> metadata.getVersionCommitTimeMillis(dbName, table, version));
579+
}
580+
576581
public Optional<Database> getDatabase(ConnectContext context, BaseTableInfo baseTableInfo) {
577582
if (baseTableInfo.isInternalCatalog()) {
578583
return Optional.ofNullable(getDb(baseTableInfo.getDbId()));

‎fe/fe-core/src/test/java/com/starrocks/connector/iceberg/MockIcebergMetadata.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
import org.apache.iceberg.DataFiles;
5454
import org.apache.iceberg.PartitionSpec;
5555
import org.apache.iceberg.Schema;
56+
import org.apache.iceberg.Snapshot;
5657
import org.apache.iceberg.Table;
5758
import org.apache.iceberg.TableMetadata;
5859
import org.apache.iceberg.types.Types;
@@ -643,6 +644,12 @@ public TvrTableSnapshot getCurrentTvrSnapshot(String dbName, com.starrocks.catal
643644
return TvrTableSnapshot.of(TvrVersion.of(1L));
644645
}
645646

647+
@Override
648+
public Optional<Long> getVersionCommitTimeMillis(String dbName, com.starrocks.catalog.Table table, long version) {
649+
Snapshot snapshot = ((IcebergTable) table).getNativeTable().snapshot(version);
650+
return snapshot == null ? Optional.empty() : Optional.of(snapshot.timestampMillis());
651+
}
652+
646653
// ConnectorMetadata's default returns TvrTableSnapshot.empty(), which leaves the planned scan
647654
// pinned to the MIN snapshot (0 partitions, no data) -- low-cardinality dict collection and the
648655
// group-by min/max rule both then no-op. Mirror production's IcebergMetadata.getTableVersionRange:

‎fe/fe-core/src/test/java/com/starrocks/scheduler/mv/ivm/IVMBasedMvRefreshProcessorIcebergTest.java‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,14 @@
1717
import com.google.common.collect.ImmutableList;
1818
import com.starrocks.catalog.BaseTableInfo;
1919
import com.starrocks.catalog.Database;
20+
import com.starrocks.catalog.IcebergTable;
2021
import com.starrocks.catalog.MaterializedView;
2122
import com.starrocks.common.AnalysisException;
2223
import com.starrocks.common.tvr.TvrDeltaStats;
2324
import com.starrocks.common.tvr.TvrTableDelta;
2425
import com.starrocks.common.tvr.TvrTableDeltaTrait;
2526
import com.starrocks.common.tvr.TvrTableSnapshot;
27+
import com.starrocks.common.tvr.TvrVersion;
2628
import com.starrocks.common.tvr.TvrVersionRange;
2729
import com.starrocks.connector.iceberg.MockIcebergMetadata;
2830
import com.starrocks.load.loadv2.IVMInsertLoadTxnCallback;
@@ -34,10 +36,14 @@
3436
import com.starrocks.scheduler.mv.pct.MVPCTRefreshProcessor;
3537
import com.starrocks.scheduler.persist.MVTaskRunExtraMessage;
3638
import com.starrocks.server.GlobalStateMgr;
39+
import com.starrocks.sql.analyzer.SemanticException;
3740
import com.starrocks.sql.ast.KeysType;
41+
import com.starrocks.sql.optimizer.rule.transformation.materialization.MvUtils;
3842
import com.starrocks.sql.plan.ExecPlan;
3943
import com.starrocks.sql.plan.PlanTestBase;
4044
import com.starrocks.thrift.TExplainLevel;
45+
import mockit.Mock;
46+
import mockit.MockUp;
4147
import org.junit.jupiter.api.Assertions;
4248
import org.junit.jupiter.api.BeforeAll;
4349
import org.junit.jupiter.api.MethodOrderer.MethodName;
@@ -1311,6 +1317,129 @@ public void testPinnedSnapshotIdMapRecordedOnExtraMessage() throws Exception {
13111317
"pinnedSnapshotIdMap value should match the PCT-synced snapshot id");
13121318
}
13131319

1320+
/**
1321+
* Verify the TVR version range consumed per base table is recorded on the task run's extra
1322+
* message so it is visible via information_schema.task_runs.EXTRA_MESSAGE.
1323+
*/
1324+
@Test
1325+
public void testImvSourceVersionRangeRecordedOnExtraMessage() throws Exception {
1326+
String query = "SELECT id, data, date FROM `iceberg0`.`unpartitioned_db`.`t0` as a;";
1327+
MaterializedView mv = createMaterializedViewWithRefreshMode(query, "incremental");
1328+
seedTvrBaselineAtVersionZero(mv);
1329+
// Synthetic version: no Iceberg snapshot with this id exists on the mock native table,
1330+
// so commit-time resolution must degrade to an empty per-table map.
1331+
advanceTableVersionTo(999999L);
1332+
mockListTableDeltaTraits(ImmutableList.of(
1333+
TvrTableDeltaTrait.ofMonotonic(
1334+
TvrTableDelta.of(TvrVersion.of(0L), TvrVersion.of(999999L)),
1335+
TvrDeltaStats.EMPTY)));
1336+
1337+
MVTaskRunProcessor processor = getMVTaskRunProcessor(mv);
1338+
Assertions.assertInstanceOf(MVIVMRefreshProcessor.class, processor.getMVRefreshProcessor());
1339+
1340+
MVTaskRunExtraMessage extraMessage =
1341+
processor.getMvTaskRunContext().getStatus().getMvTaskRunExtraMessage();
1342+
Map<String, Map<String, String>> versionRanges = extraMessage.getImvSourceVersionRange();
1343+
Assertions.assertEquals(Map.of("start", "0", "end", "999999"),
1344+
versionRanges.get("iceberg0.unpartitioned_db.t0"),
1345+
"imvSourceVersionRange should record the consumed TVR range, got: " + versionRanges);
1346+
Assertions.assertTrue(extraMessage.toString().contains("\"imvSourceVersionRange\""),
1347+
"extra message JSON should contain imvSourceVersionRange: " + extraMessage);
1348+
1349+
Map<String, String> timestampRange =
1350+
extraMessage.getImvSourceTimestampRange().get("iceberg0.unpartitioned_db.t0");
1351+
Assertions.assertNotNull(timestampRange,
1352+
"imvSourceTimestampRange should have an entry per staged base table");
1353+
Assertions.assertTrue(timestampRange.isEmpty(),
1354+
"unresolvable snapshot ids should degrade to an empty map, got: " + timestampRange);
1355+
}
1356+
1357+
/**
1358+
* When IVM planning fails after the TVR deltas were staged and the hybrid processor falls
1359+
* back to PCT, the task run must not keep source ranges from the abandoned IVM attempt.
1360+
*/
1361+
@Test
1362+
public void testImvSourceRangesNotRecordedOnPctFallback() throws Exception {
1363+
String query = "SELECT id, data, date FROM `iceberg0`.`unpartitioned_db`.`t0` as a;";
1364+
MaterializedView mv = createMaterializedViewWithRefreshMode(query, "auto");
1365+
seedTvrBaselineAtVersionZero(mv);
1366+
advanceTableVersionTo(2);
1367+
mockListTableDeltaTraits(ImmutableList.of(
1368+
TvrTableDeltaTrait.ofMonotonic(
1369+
TvrTableDelta.of(TvrVersion.of(0L), TvrVersion.of(2L)),
1370+
TvrDeltaStats.EMPTY)));
1371+
// Fail IVM plan generation after the TVR deltas were staged; the PCT fallback
1372+
// builds its plan from getTaskDefinition() and is unaffected.
1373+
new MockUp<MaterializedView>() {
1374+
@Mock
1375+
public String getIVMTaskDefinition() {
1376+
throw new SemanticException("injected IVM plan failure");
1377+
}
1378+
};
1379+
1380+
MVTaskRunProcessor processor = getMVTaskRunProcessor(mv);
1381+
Assertions.assertInstanceOf(MVHybridRefreshProcessor.class, processor.getMVRefreshProcessor());
1382+
MVHybridRefreshProcessor hybrid = (MVHybridRefreshProcessor) processor.getMVRefreshProcessor();
1383+
Assertions.assertInstanceOf(MVPCTRefreshProcessor.class, hybrid.getCurrentProcessor());
1384+
1385+
MVTaskRunExtraMessage extraMessage =
1386+
processor.getMvTaskRunContext().getStatus().getMvTaskRunExtraMessage();
1387+
Assertions.assertTrue(extraMessage.getImvSourceVersionRange().isEmpty(),
1388+
"PCT fallback must not keep IVM source ranges, got: "
1389+
+ extraMessage.getImvSourceVersionRange());
1390+
Assertions.assertTrue(extraMessage.getImvSourceTimestampRange().isEmpty(),
1391+
"PCT fallback must not keep IVM source timestamps, got: "
1392+
+ extraMessage.getImvSourceTimestampRange());
1393+
}
1394+
1395+
/**
1396+
* Verify commit times of the consumed snapshot range are recorded as imvSourceTimestampRange
1397+
* when the source snapshots are resolvable on the native Iceberg table.
1398+
*/
1399+
@Test
1400+
public void testImvSourceTimestampRangeRecordedOnExtraMessage() throws Exception {
1401+
String query = "SELECT id, data, date FROM `iceberg0`.`partitioned_db`.`t1`";
1402+
MaterializedView mv = createMaterializedViewWithRefreshMode(query, "incremental", "`date`", null);
1403+
1404+
MockIcebergMetadata mockIcebergMetadata =
1405+
(MockIcebergMetadata) connectContext.getGlobalStateMgr().getMetadataMgr()
1406+
.getOptionalMetadata(MockIcebergMetadata.MOCKED_ICEBERG_CATALOG_NAME).get();
1407+
org.apache.iceberg.Table nativeTable = ((IcebergTable) MvUtils.getTableWithIdentifier(
1408+
mv.getBaseTableInfos().get(0)).get()).getNativeTable();
1409+
// Two real Iceberg commits so both range endpoints have resolvable commit times.
1410+
mockIcebergMetadata.addRowsToPartition("partitioned_db", "t1", 10, "date=2020-01-02");
1411+
long startSnapshotId = nativeTable.currentSnapshot().snapshotId();
1412+
mockIcebergMetadata.addRowsToPartition("partitioned_db", "t1", 10, "date=2020-01-03");
1413+
long endSnapshotId = nativeTable.currentSnapshot().snapshotId();
1414+
1415+
Map<BaseTableInfo, TvrVersionRange> tvrMap = mv.getRefreshScheme().getAsyncRefreshContext()
1416+
.getBaseTableInfoTvrVersionRangeMap();
1417+
for (BaseTableInfo info : mv.getBaseTableInfos()) {
1418+
tvrMap.put(info, TvrTableSnapshot.of(startSnapshotId));
1419+
}
1420+
advanceTableVersionTo(endSnapshotId);
1421+
mockListTableDeltaTraits(ImmutableList.of(
1422+
TvrTableDeltaTrait.ofMonotonic(
1423+
TvrTableDelta.of(TvrVersion.of(startSnapshotId), TvrVersion.of(endSnapshotId)),
1424+
TvrDeltaStats.EMPTY)));
1425+
1426+
MVTaskRunProcessor processor = getMVTaskRunProcessor(mv);
1427+
Assertions.assertInstanceOf(MVIVMRefreshProcessor.class, processor.getMVRefreshProcessor());
1428+
1429+
MVTaskRunExtraMessage extraMessage =
1430+
processor.getMvTaskRunContext().getStatus().getMvTaskRunExtraMessage();
1431+
String tableKey = "iceberg0.partitioned_db.t1";
1432+
Assertions.assertEquals(
1433+
Map.of("start", String.valueOf(startSnapshotId), "end", String.valueOf(endSnapshotId)),
1434+
extraMessage.getImvSourceVersionRange().get(tableKey),
1435+
"got: " + extraMessage.getImvSourceVersionRange());
1436+
Assertions.assertEquals(
1437+
Map.of("start", String.valueOf(nativeTable.snapshot(startSnapshotId).timestampMillis()),
1438+
"end", String.valueOf(nativeTable.snapshot(endSnapshotId).timestampMillis())),
1439+
extraMessage.getImvSourceTimestampRange().get(tableKey),
1440+
"got: " + extraMessage.getImvSourceTimestampRange());
1441+
}
1442+
13141443
@Test
13151444
public void testIncrementalFirstRefreshRoutesToHybridForPctBaseline() throws Exception {
13161445
// Empty TVR baseline: factory must route to hybrid so PCT establishes the baseline.

0 commit comments

Comments
 (0)