[Enhancement] Record IVM source version/timestamp ranges in MV task run EXTRA_MESSAGE

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

@@ -212,6 +212,16 @@ public TvrVersionRange getTableVersionRange(String dbName, Table table,
         return metadata.getTableVersionRange(dbName, table, startVersion, endVersion);
     }
 
+    @Override
+    public Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
+        ConnectorMetadata metadata = metadataOfTable(table);
+        if (metadata == null) {
+            metadata = metadataOfDb(dbName);
+        }
+
+        return metadata.getVersionCommitTimeMillis(dbName, table, version);
+    }
+
     @Override
     public boolean tableExists(ConnectContext context, String dbName, String tblName) {
         ConnectorMetadata metadata = metadataOfDb(dbName);

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

@@ -194,6 +194,15 @@ default List<TvrTableDeltaTrait> listTableDeltaTraits(String dbName, Table table
         return Lists.newArrayList();
     }
 
+    /**
+     * Commit time of {@code version} (the table's own version space, e.g. an Iceberg snapshot id)
+     * in epoch millis, or empty when it cannot be resolved (unknown/expired version, or a format
+     * with no per-version commit time).
+     */
+    default Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
+        return Optional.empty();
+    }
+
     default boolean tableExists(ConnectContext context, String dbName, String tblName) {
         return listTableNames(context, dbName).contains(tblName);
     }

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

@@ -857,6 +857,12 @@ public List<TvrTableDeltaTrait> listTableDeltaTraits(String dbName, Table table,
         return tvrDeltaTraits;
     }
 
+    @Override
+    public Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
+        Snapshot snapshot = ((IcebergTable) table).getNativeTable().snapshot(version);
+        return snapshot == null ? Optional.empty() : Optional.of(snapshot.timestampMillis());
+    }
+
     @Override
     public TvrVersionRange getTableVersionRange(String dbName, Table table,
                                                 Optional<ConnectorTableVersion> startVersion,

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

@@ -164,6 +164,12 @@ public TvrVersionRange getTableVersionRange(String dbName, Table table,
         return metadata.getTableVersionRange(dbName, table, startVersion, endVersion);
     }
 
+    @Override
+    public Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
+        ConnectorMetadata metadata = metadataOfTable(table);
+        return metadata.getVersionCommitTimeMillis(dbName, table, version);
+    }
+
     @Override
     public List<String> listDbNames(ConnectContext context) {
         return hiveMetadata.listDbNames(context);

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

@@ -16,6 +16,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multimap;
 import com.starrocks.catalog.BaseTableInfo;
@@ -50,7 +51,9 @@
 import com.starrocks.scheduler.mv.BaseTableSnapshotInfo;
 import com.starrocks.scheduler.mv.MVRefreshExecutor;
 import com.starrocks.scheduler.mv.MVRefreshProcessor;
+import com.starrocks.scheduler.persist.MVTaskRunExtraMessage;
 import com.starrocks.server.GlobalStateMgr;
+import com.starrocks.server.MetadataMgr;
 import com.starrocks.sql.StatementPlanner;
 import com.starrocks.sql.analyzer.Analyzer;
 import com.starrocks.sql.analyzer.AnalyzerUtils;
@@ -154,9 +157,50 @@ public ProcessExecPlan getProcessExecPlan(TaskRunContext taskRunContext) throws
         try (Timer ignored = Tracers.watchScope("MVRefreshPrepareRefreshPlan")) {
             insertStmt = prepareRefreshPlan();
         }
+        recordImvSourceRangesOnTaskRun();
         return new ProcessExecPlan(Constants.TaskRunState.SUCCESS, mvContext.getExecPlan(), insertStmt);
     }
 
+    /**
+     * Record the staged TVR version range and snapshot commit times per base table on the task
+     * run's extra message, surfaced via information_schema.task_runs.EXTRA_MESSAGE.
+     * Must stay after prepareRefreshPlan(): recording earlier leaves stale ranges on the task
+     * run when the hybrid processor falls back to PCT on an IVM planning failure.
+     */
+    private void recordImvSourceRangesOnTaskRun() {
+        updateTaskRunStatus(status -> {
+            Map<String, Map<String, String>> versionRanges = Maps.newHashMap();
+            Map<String, Map<String, String>> timestampRanges = Maps.newHashMap();
+            for (BaseTableSnapshotInfo snapshotInfo : snapshotBaseTables.values()) {
+                TvrVersionRange delta = ((TvrTableSnapshotInfo) snapshotInfo).getTvrSnapshot();
+                if (delta == null) {
+                    continue;
+                }
+                BaseTableInfo baseTableInfo = snapshotInfo.getBaseTableInfo();
+                String tableFullName = baseTableInfo.getReadableString();
+                // TvrVersion.toString() renders the MIN/MAX sentinels as "MIN"/"MAX"
+                versionRanges.put(tableFullName, ImmutableMap.of(
+                        "start", delta.from().toString(),
+                        "end", delta.to().toString()));
+                timestampRanges.put(tableFullName,
+                        resolveCommitTimeRange(baseTableInfo.getDbName(), snapshotInfo.getBaseTable(), delta));
+            }
+            MVTaskRunExtraMessage extraMessage = status.getMvTaskRunExtraMessage();
+            extraMessage.setImvSourceVersionRange(versionRanges);
+            extraMessage.setImvSourceTimestampRange(timestampRanges);
+        });
+    }
+
+    private static Map<String, String> resolveCommitTimeRange(String dbName, Table table, TvrVersionRange delta) {
+        Map<String, String> commitTimes = Maps.newLinkedHashMap();
+        MetadataMgr metadataMgr = GlobalStateMgr.getCurrentState().getMetadataMgr();
+        delta.start().flatMap(version -> metadataMgr.getVersionCommitTimeMillis(dbName, table, version))
+                .ifPresent(time -> commitTimes.put("start", String.valueOf(time)));
+        delta.end().flatMap(version -> metadataMgr.getVersionCommitTimeMillis(dbName, table, version))
+                .ifPresent(time -> commitTimes.put("end", String.valueOf(time)));
+        return commitTimes;
+    }
+
     @Override
     public Constants.TaskRunState execProcessExecPlan(TaskRunContext taskRunContext,
                                                       ProcessExecPlan processExecPlan,

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

@@ -82,6 +82,13 @@ public class MVTaskRunExtraMessage implements Writable {
     @SerializedName("pinnedSnapshotIdMap")
     private Map<String, Long> pinnedSnapshotIdMap = Maps.newHashMap();
 
+    // For IVM refreshes: per base table ("catalog.db.tbl"), the consumed TVR version range
+    // {start, end} and the matching snapshot commit times in epoch millis (empty when unresolvable).
+    @SerializedName("imvSourceVersionRange")
+    private Map<String, Map<String, String>> imvSourceVersionRange = Maps.newHashMap();
+    @SerializedName("imvSourceTimestampRange")
+    private Map<String, Map<String, String>> imvSourceTimestampRange = Maps.newHashMap();
+
     public MVTaskRunExtraMessage() {
     }
 
@@ -222,6 +229,24 @@ public void setPinnedSnapshotIdMap(Map<String, Long> pinnedSnapshotIdMap) {
                 Config.max_mv_task_run_meta_message_values_length);
     }
 
+    public Map<String, Map<String, String>> getImvSourceVersionRange() {
+        return imvSourceVersionRange;
+    }
+
+    public void setImvSourceVersionRange(Map<String, Map<String, String>> imvSourceVersionRange) {
+        this.imvSourceVersionRange = MvUtils.shrinkToSize(imvSourceVersionRange,
+                Config.max_mv_task_run_meta_message_values_length);
+    }
+
+    public Map<String, Map<String, String>> getImvSourceTimestampRange() {
+        return imvSourceTimestampRange;
+    }
+
+    public void setImvSourceTimestampRange(Map<String, Map<String, String>> imvSourceTimestampRange) {
+        this.imvSourceTimestampRange = MvUtils.shrinkToSize(imvSourceTimestampRange,
+                Config.max_mv_task_run_meta_message_values_length);
+    }
+
     @Override
     public String toString() {
         return GsonUtils.GSON.toJson(this);

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

@@ -573,6 +573,11 @@ public TvrVersionRange getTableVersionRange(String dbName, Table table,
                 .orElse(TvrTableSnapshot.empty());
     }
 
+    public Optional<Long> getVersionCommitTimeMillis(String dbName, Table table, long version) {
+        Optional<ConnectorMetadata> connectorMetadata = getOptionalMetadata(table.getCatalogName());
+        return connectorMetadata.flatMap(metadata -> metadata.getVersionCommitTimeMillis(dbName, table, version));
+    }
+
     public Optional<Database> getDatabase(ConnectContext context, BaseTableInfo baseTableInfo) {
         if (baseTableInfo.isInternalCatalog()) {
             return Optional.ofNullable(getDb(baseTableInfo.getDbId()));

fe/fe-core/src/test/java/com/starrocks/connector/CatalogConnectorMetadataTest.java

@@ -34,6 +34,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -290,4 +291,25 @@ void testAcquireTvrSnapshotDelegatesToChild(@Mocked ConnectorMetadata connectorM
         TvrTableSnapshot actual = catalogConnectorMetadata.acquireTvrSnapshot("test_db", table, mvId);
         assertEquals(expected, actual);
     }
+
+    @Test
+    void testGetVersionCommitTimeMillisDelegatesToChild(@Mocked ConnectorMetadata connectorMetadata,
+                                                        @Mocked Table table) {
+        Optional<Long> expected = Optional.of(1781000000000L);
+        new Expectations() {
+            {
+                connectorMetadata.getVersionCommitTimeMillis("test_db", table, 42L);
+                result = expected;
+                times = 1;
+            }
+        };
+
+        CatalogConnectorMetadata catalogConnectorMetadata = new CatalogConnectorMetadata(
+                connectorMetadata,
+                informationSchemaMetadata,
+                metaMetadata
+        );
+
+        assertEquals(expected, catalogConnectorMetadata.getVersionCommitTimeMillis("test_db", table, 42L));
+    }
 }

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

@@ -244,6 +244,23 @@ public void testListDatabaseNames(@Mocked IcebergCatalog icebergCatalog) {
         Assertions.assertEquals(expectResult, metadata.listDbNames(connectContext));
     }
 
+    @Test
+    public void testGetVersionCommitTimeMillis() {
+        IcebergHiveCatalog icebergHiveCatalog = new IcebergHiveCatalog(CATALOG_NAME, new Configuration(), DEFAULT_CONFIG);
+        IcebergMetadata metadata = new IcebergMetadata(CATALOG_NAME, HDFS_ENVIRONMENT, icebergHiveCatalog,
+                Executors.newSingleThreadExecutor(), Executors.newSingleThreadExecutor(), null);
+        mockedNativeTableA.newAppend().appendFile(FILE_A).commit();
+        mockedNativeTableA.refresh();
+        Snapshot snapshot = mockedNativeTableA.currentSnapshot();
+        IcebergTable table = new IcebergTable(1, "tableA", CATALOG_NAME, CATALOG_NAME, "iceberg_db",
+                "tableA", "", Lists.newArrayList(), mockedNativeTableA, Maps.newHashMap());
+
+        Assertions.assertEquals(Optional.of(snapshot.timestampMillis()),
+                metadata.getVersionCommitTimeMillis("iceberg_db", table, snapshot.snapshotId()));
+        Assertions.assertEquals(Optional.empty(),
+                metadata.getVersionCommitTimeMillis("iceberg_db", table, snapshot.snapshotId() + 1));
+    }
+
     @Test
     public void testGetDB(@Mocked IcebergHiveCatalog icebergHiveCatalog) {
         String db = "db";

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

@@ -53,6 +53,7 @@
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.types.Types;
@@ -643,6 +644,12 @@ public TvrTableSnapshot getCurrentTvrSnapshot(String dbName, com.starrocks.catal
         return TvrTableSnapshot.of(TvrVersion.of(1L));
     }
 
+    @Override
+    public Optional<Long> getVersionCommitTimeMillis(String dbName, com.starrocks.catalog.Table table, long version) {
+        Snapshot snapshot = ((IcebergTable) table).getNativeTable().snapshot(version);
+        return snapshot == null ? Optional.empty() : Optional.of(snapshot.timestampMillis());
+    }
+
     // ConnectorMetadata's default returns TvrTableSnapshot.empty(), which leaves the planned scan
     // pinned to the MIN snapshot (0 partitions, no data) -- low-cardinality dict collection and the
     // group-by min/max rule both then no-op. Mirror production's IcebergMetadata.getTableVersionRange:

fe/fe-core/src/test/java/com/starrocks/connector/unified/UnifiedMetadataTest.java

@@ -57,6 +57,7 @@
 
 import java.util.HashMap;
 import java.util.List;
+import java.util.Optional;
 
 import static com.starrocks.catalog.Table.TableType.DELTALAKE;
 import static com.starrocks.catalog.Table.TableType.HIVE;
@@ -579,4 +580,23 @@ public void testAcquireTvrSnapshotRoutesByTableType(@Mocked IcebergTable iceberg
         TvrTableSnapshot actual = unifiedMetadata.acquireTvrSnapshot("test_db", icebergTable, mvId);
         assertEquals(expected, actual);
     }
+
+    @Test
+    public void testGetVersionCommitTimeMillisRoutesByTableType(@Mocked IcebergTable icebergTable) {
+        Optional<Long> expected = Optional.of(1781000000000L);
+        new Expectations() {
+            {
+                icebergTable.getType();
+                result = ICEBERG;
+                minTimes = 1;
+            }
+            {
+                icebergMetadata.getVersionCommitTimeMillis("test_db", icebergTable, 42L);
+                result = expected;
+                times = 1;
+            }
+        };
+
+        assertEquals(expected, unifiedMetadata.getVersionCommitTimeMillis("test_db", icebergTable, 42L));
+    }
 }