Skip to content

Commit 87c8894

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 87c8894

8 files changed

Lines changed: 383 additions & 0 deletions

File tree

‎fe/fe-core/src/main/java/com/starrocks/catalog/IcebergTable.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,7 @@
7474
import org.apache.iceberg.PartitionSpec;
7575
import org.apache.iceberg.Partitioning;
7676
import org.apache.iceberg.Schema;
77+
import org.apache.iceberg.Snapshot;
7778
import org.apache.iceberg.SortDirection;
7879
import org.apache.iceberg.SortField;
7980
import org.apache.iceberg.SortOrder;
@@ -179,6 +180,18 @@ public String getUUID() {
179180
}
180181
}
181182

183+
@Override
184+
public Optional<Long> getVersionCommitTimeMillis(long version) {
185+
try {
186+
Snapshot snapshot = getNativeTable().snapshot(version);
187+
return snapshot == null ? Optional.empty() : Optional.of(snapshot.timestampMillis());
188+
} catch (Exception e) {
189+
LOG.warn("Failed to resolve commit time of snapshot {} for table {}: {}",
190+
version, name, e.getMessage());
191+
return Optional.empty();
192+
}
193+
}
194+
182195
@Override
183196
public List<Column> getPartitionColumns() {
184197
if (partitionColumns == null) {

‎fe/fe-core/src/main/java/com/starrocks/catalog/Table.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -841,6 +841,14 @@ public boolean isTemporal() {
841841
return false;
842842
}
843843

844+
/**
845+
* Commit time in epoch millis of the given TVR version (for lake tables, the snapshot id);
846+
* empty when the table format cannot resolve it.
847+
*/
848+
public Optional<Long> getVersionCommitTimeMillis(long version) {
849+
return Optional.empty();
850+
}
851+
844852
public boolean hasUniqueConstraints() {
845853
List<UniqueConstraint> uniqueConstraint = getUniqueConstraints();
846854
return uniqueConstraint != null;

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

Lines changed: 40 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,6 +51,7 @@
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;
5456
import com.starrocks.sql.StatementPlanner;
5557
import com.starrocks.sql.analyzer.Analyzer;
@@ -154,9 +156,47 @@ public ProcessExecPlan getProcessExecPlan(TaskRunContext taskRunContext) throws
154156
try (Timer ignored = Tracers.watchScope("MVRefreshPrepareRefreshPlan")) {
155157
insertStmt = prepareRefreshPlan();
156158
}
159+
recordImvSourceRangesOnTaskRun();
157160
return new ProcessExecPlan(Constants.TaskRunState.SUCCESS, mvContext.getExecPlan(), insertStmt);
158161
}
159162

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

Lines changed: 131 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,131 @@ 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+
Assertions.assertTrue(mv.getVersionCommitTimeMillis(1L).isEmpty(),
1356+
"tables without version commit times must resolve to empty");
1357+
}
1358+
1359+
/**
1360+
* When IVM planning fails after the TVR deltas were staged and the hybrid processor falls
1361+
* back to PCT, the task run must not keep source ranges from the abandoned IVM attempt.
1362+
*/
1363+
@Test
1364+
public void testImvSourceRangesNotRecordedOnPctFallback() throws Exception {
1365+
String query = "SELECT id, data, date FROM `iceberg0`.`unpartitioned_db`.`t0` as a;";
1366+
MaterializedView mv = createMaterializedViewWithRefreshMode(query, "auto");
1367+
seedTvrBaselineAtVersionZero(mv);
1368+
advanceTableVersionTo(2);
1369+
mockListTableDeltaTraits(ImmutableList.of(
1370+
TvrTableDeltaTrait.ofMonotonic(
1371+
TvrTableDelta.of(TvrVersion.of(0L), TvrVersion.of(2L)),
1372+
TvrDeltaStats.EMPTY)));
1373+
// Fail IVM plan generation after the TVR deltas were staged; the PCT fallback
1374+
// builds its plan from getTaskDefinition() and is unaffected.
1375+
new MockUp<MaterializedView>() {
1376+
@Mock
1377+
public String getIVMTaskDefinition() {
1378+
throw new SemanticException("injected IVM plan failure");
1379+
}
1380+
};
1381+
1382+
MVTaskRunProcessor processor = getMVTaskRunProcessor(mv);
1383+
Assertions.assertInstanceOf(MVHybridRefreshProcessor.class, processor.getMVRefreshProcessor());
1384+
MVHybridRefreshProcessor hybrid = (MVHybridRefreshProcessor) processor.getMVRefreshProcessor();
1385+
Assertions.assertInstanceOf(MVPCTRefreshProcessor.class, hybrid.getCurrentProcessor());
1386+
1387+
MVTaskRunExtraMessage extraMessage =
1388+
processor.getMvTaskRunContext().getStatus().getMvTaskRunExtraMessage();
1389+
Assertions.assertTrue(extraMessage.getImvSourceVersionRange().isEmpty(),
1390+
"PCT fallback must not keep IVM source ranges, got: "
1391+
+ extraMessage.getImvSourceVersionRange());
1392+
Assertions.assertTrue(extraMessage.getImvSourceTimestampRange().isEmpty(),
1393+
"PCT fallback must not keep IVM source timestamps, got: "
1394+
+ extraMessage.getImvSourceTimestampRange());
1395+
}
1396+
1397+
/**
1398+
* Verify commit times of the consumed snapshot range are recorded as imvSourceTimestampRange
1399+
* when the source snapshots are resolvable on the native Iceberg table.
1400+
*/
1401+
@Test
1402+
public void testImvSourceTimestampRangeRecordedOnExtraMessage() throws Exception {
1403+
String query = "SELECT id, data, date FROM `iceberg0`.`partitioned_db`.`t1`";
1404+
MaterializedView mv = createMaterializedViewWithRefreshMode(query, "incremental", "`date`", null);
1405+
1406+
MockIcebergMetadata mockIcebergMetadata =
1407+
(MockIcebergMetadata) connectContext.getGlobalStateMgr().getMetadataMgr()
1408+
.getOptionalMetadata(MockIcebergMetadata.MOCKED_ICEBERG_CATALOG_NAME).get();
1409+
org.apache.iceberg.Table nativeTable = ((IcebergTable) MvUtils.getTableWithIdentifier(
1410+
mv.getBaseTableInfos().get(0)).get()).getNativeTable();
1411+
// Two real Iceberg commits so both range endpoints have resolvable commit times.
1412+
mockIcebergMetadata.addRowsToPartition("partitioned_db", "t1", 10, "date=2020-01-02");
1413+
long startSnapshotId = nativeTable.currentSnapshot().snapshotId();
1414+
mockIcebergMetadata.addRowsToPartition("partitioned_db", "t1", 10, "date=2020-01-03");
1415+
long endSnapshotId = nativeTable.currentSnapshot().snapshotId();
1416+
1417+
Map<BaseTableInfo, TvrVersionRange> tvrMap = mv.getRefreshScheme().getAsyncRefreshContext()
1418+
.getBaseTableInfoTvrVersionRangeMap();
1419+
for (BaseTableInfo info : mv.getBaseTableInfos()) {
1420+
tvrMap.put(info, TvrTableSnapshot.of(startSnapshotId));
1421+
}
1422+
advanceTableVersionTo(endSnapshotId);
1423+
mockListTableDeltaTraits(ImmutableList.of(
1424+
TvrTableDeltaTrait.ofMonotonic(
1425+
TvrTableDelta.of(TvrVersion.of(startSnapshotId), TvrVersion.of(endSnapshotId)),
1426+
TvrDeltaStats.EMPTY)));
1427+
1428+
MVTaskRunProcessor processor = getMVTaskRunProcessor(mv);
1429+
Assertions.assertInstanceOf(MVIVMRefreshProcessor.class, processor.getMVRefreshProcessor());
1430+
1431+
MVTaskRunExtraMessage extraMessage =
1432+
processor.getMvTaskRunContext().getStatus().getMvTaskRunExtraMessage();
1433+
String tableKey = "iceberg0.partitioned_db.t1";
1434+
Assertions.assertEquals(
1435+
Map.of("start", String.valueOf(startSnapshotId), "end", String.valueOf(endSnapshotId)),
1436+
extraMessage.getImvSourceVersionRange().get(tableKey),
1437+
"got: " + extraMessage.getImvSourceVersionRange());
1438+
Assertions.assertEquals(
1439+
Map.of("start", String.valueOf(nativeTable.snapshot(startSnapshotId).timestampMillis()),
1440+
"end", String.valueOf(nativeTable.snapshot(endSnapshotId).timestampMillis())),
1441+
extraMessage.getImvSourceTimestampRange().get(tableKey),
1442+
"got: " + extraMessage.getImvSourceTimestampRange());
1443+
}
1444+
13141445
@Test
13151446
public void testIncrementalFirstRefreshRoutesToHybridForPctBaseline() throws Exception {
13161447
// Empty TVR baseline: factory must route to hybrid so PCT establishes the baseline.

‎fe/fe-core/src/test/java/com/starrocks/scheduler/persist/MVTaskRunExtraMessageTest.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515

1616
package com.starrocks.scheduler.persist;
1717

18+
import com.google.common.collect.ImmutableMap;
1819
import com.google.common.collect.Maps;
1920
import com.google.common.collect.Sets;
2021
import com.starrocks.common.Config;
@@ -112,4 +113,36 @@ public void testMessageWithTooLongBasePartitionsToRefreshMap() {
112113
Assertions.assertTrue(message.getBasePartitionsToRefreshMap().size() ==
113114
Config.max_mv_task_run_meta_message_values_length);
114115
}
116+
117+
@Test
118+
public void testImvSourceRangesShrunkToConfiguredSize() {
119+
Map<String, Map<String, String>> versionRanges = Maps.newHashMap();
120+
Map<String, Map<String, String>> timestampRanges = Maps.newHashMap();
121+
for (int i = 0; i < Config.max_mv_task_run_meta_message_values_length + 10; i++) {
122+
versionRanges.put("catalog.db.table" + i, ImmutableMap.of("start", "1", "end", "2"));
123+
timestampRanges.put("catalog.db.table" + i, ImmutableMap.of("start", "1000", "end", "2000"));
124+
}
125+
MVTaskRunExtraMessage message = new MVTaskRunExtraMessage();
126+
message.setImvSourceVersionRange(versionRanges);
127+
message.setImvSourceTimestampRange(timestampRanges);
128+
Assertions.assertEquals(Config.max_mv_task_run_meta_message_values_length,
129+
message.getImvSourceVersionRange().size());
130+
Assertions.assertEquals(Config.max_mv_task_run_meta_message_values_length,
131+
message.getImvSourceTimestampRange().size());
132+
}
133+
134+
@Test
135+
public void testImvSourceRangesSerializedToJson() {
136+
MVTaskRunExtraMessage message = new MVTaskRunExtraMessage();
137+
message.setImvSourceVersionRange(
138+
ImmutableMap.of("iceberg.db.tbl", ImmutableMap.of("start", "100", "end", "128")));
139+
message.setImvSourceTimestampRange(
140+
ImmutableMap.of("iceberg.db.tbl", ImmutableMap.of("start", "1717999999000", "end", "1718000000000")));
141+
String json = message.toString();
142+
Assertions.assertTrue(json.contains(
143+
"\"imvSourceVersionRange\":{\"iceberg.db.tbl\":{\"start\":\"100\",\"end\":\"128\"}}"), json);
144+
Assertions.assertTrue(json.contains(
145+
"\"imvSourceTimestampRange\":{\"iceberg.db.tbl\":" +
146+
"{\"start\":\"1717999999000\",\"end\":\"1718000000000\"}}"), json);
147+
}
115148
}

0 commit comments

Comments
 (0)