[fix](fe) Report Iceberg scan metrics in batch mode

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java

@@ -245,10 +245,10 @@ public class IcebergScanPlanProvider implements ConnectorScanPlanProvider {
     private final IcebergFormatCache formatCache;
 
     // FIX-SCAN-METRICS: per-query stash of the iceberg SDK scan diagnostics captured by the attached
-    // IcebergScanProfileReporter during planScan, keyed by session queryId. fe-core drains it
-    // (collectScanProfiles) right after planScan on the same thread; releaseReadTransaction reclaims any entry
-    // a thrown planScan left behind. Attached only on the synchronous data/count path (never streaming or
-    // system-table, which fe-core never drains), so the value list is appended single-threaded.
+    // IcebergScanProfileReporter during eager or streaming planning, keyed by session queryId. fe-core drains it
+    // (collectScanProfiles) right after planScan or after closing the streaming split source;
+    // releaseReadTransaction reclaims any entry left behind by a failed planning path. Never attached to system
+    // tables, whose serialized-task planning does not contribute these scan metrics.
     private final ConcurrentHashMap<String, List<ConnectorScanProfile>> scanProfileStash = new ConcurrentHashMap<>();
 
     // Test-only gate for the PERF-11 per-file memo: how many times computePerFileInvariants actually ran across
@@ -484,6 +484,12 @@ public ConnectorSplitSource streamSplits(ConnectorSession session, ConnectorTabl
         IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
         Table table = resolveTable(session, iceHandle);
         TableScan scan = buildScan(table, iceHandle, filter, session);
+        // Match the eager planScan path: planFiles() emits its ScanReport when the streaming source closes.
+        // The engine drains the queryId-keyed profile after closing that source, so batch and non-batch scans
+        // expose the same Iceberg scan metrics without relying on a thread-local query context here.
+        if (session != null) {
+            scan = scan.metricsReporter(new IcebergScanProfileReporter(session.getQueryId(), scanProfileStash));
+        }
         int formatVersion = getFormatVersion(table);
         List<String> orderedPartitionKeys = IcebergPartitionUtils.getIdentityPartitionColumns(table);
         ZoneId zone = resolveSessionZone(session);

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanProfileReporter.java

@@ -43,10 +43,9 @@
  * behavior in the plugin architecture: it is a self-contained port (the connector cannot import fe-core, so
  * {@code DebugUtil}'s time/byte formatters are inlined and Guava is avoided).
  *
- * <p>The SDK invokes {@link #report} on CLOSE of the {@code planFiles} iterable, which the connector performs
- * synchronously on the planScan thread — so a fresh reporter is created per scan bound to that scan's queryId,
- * and attached ONLY on the synchronous data/count path (never the streaming or system-table path, which fe-core
- * never drains).</p>
+ * <p>The SDK invokes {@link #report} on CLOSE of the {@code planFiles} iterable. A fresh reporter is created per
+ * scan and bound to that scan's queryId; fe-core drains eager scans after {@code planScan} and streaming scans
+ * after closing their split source.</p>
  */
 public class IcebergScanProfileReporter implements MetricsReporter {
     /**

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java

@@ -25,6 +25,7 @@
 import org.apache.doris.connector.spi.pushdown.ConnectorComparison;
 import org.apache.doris.connector.spi.pushdown.ConnectorExpression;
 import org.apache.doris.connector.spi.pushdown.ConnectorLiteral;
+import org.apache.doris.connector.spi.scan.ConnectorScanProfile;
 import org.apache.doris.connector.spi.scan.ConnectorScanRange;
 import org.apache.doris.connector.spi.scan.ConnectorScanRequest;
 import org.apache.doris.connector.spi.scan.ConnectorSplitSource;
@@ -1812,6 +1813,37 @@ public void streamSplitsProducesOneLazyRangePerFile() throws IOException {
                 "streaming must yield one range per file");
     }
 
+    @Test
+    public void streamSplitsCollectsSameScanMetricsAsPlanScan() throws IOException {
+        Table table = threeFileTable();
+        IcebergTableHandle handle = new IcebergTableHandle("db1", "t1");
+        ConnectorSession session = emptySession();
+        IcebergScanPlanProvider provider = providerOver(table);
+
+        provider.planScan(session, ConnectorScanRequest.builder(handle, Collections.emptyList()).build());
+        List<ConnectorScanProfile> eagerProfiles = provider.collectScanProfiles(session);
+        Assertions.assertEquals(1, eagerProfiles.size());
+
+        List<ConnectorScanRange> streamedRanges = drain(provider.streamSplits(
+                session, handle, Collections.emptyList(), Optional.empty(), -1L));
+        Assertions.assertEquals(3, streamedRanges.size());
+        List<ConnectorScanProfile> streamingProfiles = provider.collectScanProfiles(session);
+        Assertions.assertEquals(1, streamingProfiles.size(),
+                "closing the streaming source must publish its Iceberg ScanReport");
+
+        ConnectorScanProfile eager = eagerProfiles.get(0);
+        ConnectorScanProfile streaming = streamingProfiles.get(0);
+        Assertions.assertEquals(eager.getGroupName(), streaming.getGroupName());
+        Assertions.assertEquals(eager.getScanLabel(), streaming.getScanLabel());
+        Assertions.assertEquals(eager.getMetrics().keySet(), streaming.getMetrics().keySet());
+        for (Map.Entry<String, String> entry : eager.getMetrics().entrySet()) {
+            if (!"planning".equals(entry.getKey())) {
+                Assertions.assertEquals(entry.getValue(), streaming.getMetrics().get(entry.getKey()),
+                        "batch and non-batch metric values must match for " + entry.getKey());
+            }
+        }
+    }
+
     @Test
     public void streamSplitsRewriteScopeSkipsUnscopedFilesViaLookahead() throws IOException {
         // A rewrite scope keeps only f1 + f3; the source's look-ahead must skip f2 in hasNext(). MUTATION:

fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProvider.java

@@ -282,12 +282,13 @@ default OptionalLong scannedPartitionCount(List<ConnectorScanRange> scanRanges)
      * {@link ConnectorScanProfile} groups the engine writes into the query's profile execution summary.
      *
      * <p>The default returns an empty list (connector reports nothing). A connector that wants scan
-     * diagnostics harvests them from its SDK during {@code planScan} (the paimon SDK exposes a metric
-     * registry, the iceberg SDK a metrics reporter), stashes them keyed by {@link ConnectorSession#getQueryId()},
-     * and drains them here — mirroring the per-query queryId stashes this SPI already uses (read-transaction
-     * release, rewritable-delete supply). The engine calls this immediately after {@code planScan} on the
-     * same thread, so the harvest is complete; the connector must also drop its stash on
-     * {@link #releaseReadTransaction} to reclaim any entry a thrown {@code planScan} left behind.</p>
+     * diagnostics harvests them from its SDK during {@code planScan} or streaming split generation (the paimon
+     * SDK exposes a metric registry, the iceberg SDK a metrics reporter), stashes them keyed by
+     * {@link ConnectorSession#getQueryId()}, and drains them here — mirroring the per-query queryId stashes this
+     * SPI already uses (read-transaction release, rewritable-delete supply). The engine calls this immediately
+     * after {@code planScan}, or after closing a streaming split source, so the harvest is complete; the connector
+     * must also drop its stash on {@link #releaseReadTransaction} to reclaim an entry left by a failed planning
+     * path.</p>
      *
      * @param session the current session (its queryId keys the connector's per-query stash)
      * @return this scan's diagnostics, or an empty list (the default) to contribute nothing to the profile

fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java

@@ -416,11 +416,7 @@ static void writeScanProfilesInto(RuntimeProfile executionSummary, List<Connecto
             return;
         }
         for (ConnectorScanProfile profile : profiles) {
-            RuntimeProfile group = executionSummary.getChildMap().get(profile.getGroupName());
-            if (group == null) {
-                group = new RuntimeProfile(profile.getGroupName());
-                executionSummary.addChild(group, true);
-            }
+            RuntimeProfile group = getOrCreateScanProfileGroup(executionSummary, profile.getGroupName());
             RuntimeProfile scan = new RuntimeProfile(profile.getScanLabel());
             for (Map.Entry<String, String> entry : profile.getMetrics().entrySet()) {
                 scan.addInfoString(entry.getKey(), entry.getValue());
@@ -429,6 +425,19 @@ static void writeScanProfilesInto(RuntimeProfile executionSummary, List<Connecto
         }
     }
 
+    private static RuntimeProfile getOrCreateScanProfileGroup(RuntimeProfile executionSummary, String groupName) {
+        // Multiple streaming scan nodes in one query can finish concurrently. Serialize the compound child-map
+        // lookup/add so one callback cannot replace the group created by another and discard its scan profile.
+        synchronized (executionSummary) {
+            RuntimeProfile group = executionSummary.getChildMap().get(groupName);
+            if (group == null) {
+                group = new RuntimeProfile(groupName);
+                executionSummary.addChild(group, true);
+            }
+            return group;
+        }
+    }
+
     @Override
     public String getNodeExplainString(String prefix, TExplainLevel detailLevel) {
         StringBuilder output = new StringBuilder();
@@ -1870,6 +1879,10 @@ private void startStreamingSplit() {
         pinRewriteFileScope();
         final ConnectorTableHandle handle = currentHandle;
         final ConnectorScanPlanProvider scanProvider = resolveScanProvider();
+        // ConnectContext is thread-local and is unavailable on the schedule executor. Capture this query's
+        // execution summary before dispatch so streaming metrics are written to the same profile as eager scans.
+        SummaryProfile summaryProfile = SummaryProfile.getSummaryProfile(ConnectContext.get());
+        final RuntimeProfile executionSummary = summaryProfile == null ? null : summaryProfile.getExecutionSummary();
         Executor scheduleExecutor = Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
         CompletableFuture.runAsync(() -> {
             ConnectorSplitSource source = null;
@@ -1898,6 +1911,9 @@ private void startStreamingSplit() {
                         LOG.warn("Failed to close streaming split source for {}", handle, ce);
                     }
                 }
+                List<ConnectorScanProfile> scanProfiles = onPluginClassLoader(scanProvider,
+                        () -> scanProvider.collectScanProfiles(connectorSession));
+                writeScanProfilesInto(executionSummary, scanProfiles);
             }
         }, scheduleExecutor);
     }

fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProfileTest.java

@@ -23,10 +23,13 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * FIX-SCAN-METRICS — guards {@link PluginDrivenScanNode#writeScanProfilesInto}, the connector-agnostic
@@ -87,4 +90,22 @@ public void sharesGroupAcrossScans() {
         Assertions.assertEquals("3", group.getChildMap().get("Table Scan (db.a)").getInfoString("data_files"));
         Assertions.assertEquals("5", group.getChildMap().get("Table Scan (db.b)").getInfoString("data_files"));
     }
+
+    @Test
+    public void concurrentStreamingScansShareOneGroup() {
+        RuntimeProfile summary = new RuntimeProfile("Execution Summary");
+        List<CompletableFuture<Void>> writes = new ArrayList<>();
+        for (int i = 0; i < 32; i++) {
+            String label = "Table Scan (db.t" + i + ")";
+            writes.add(CompletableFuture.runAsync(() -> PluginDrivenScanNode.writeScanProfilesInto(
+                    summary, Collections.singletonList(
+                            profile("Iceberg Scan Metrics", label, "data_files", "1")))));
+        }
+        CompletableFuture.allOf(writes.toArray(new CompletableFuture[0])).join();
+
+        RuntimeProfile group = summary.getChildMap().get("Iceberg Scan Metrics");
+        Assertions.assertNotNull(group);
+        Assertions.assertEquals(32, group.getChildMap().size(),
+                "concurrent batch scans must not replace the shared profile group");
+    }
 }