diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java index 024a036df11158..239b9b99535a6e 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java +++ b/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> 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 orderedPartitionKeys = IcebergPartitionUtils.getIdentityPartitionColumns(table); ZoneId zone = resolveSessionZone(session); diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanProfileReporter.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanProfileReporter.java index 5ee788d2135f28..ceeee24a321046 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanProfileReporter.java +++ b/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). * - *

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).

+ *

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.

*/ public class IcebergScanProfileReporter implements MetricsReporter { /** diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java index 426f89c7d427b6..42e61377ee90ff 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java +++ b/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 eagerProfiles = provider.collectScanProfiles(session); + Assertions.assertEquals(1, eagerProfiles.size()); + + List streamedRanges = drain(provider.streamSplits( + session, handle, Collections.emptyList(), Optional.empty(), -1L)); + Assertions.assertEquals(3, streamedRanges.size()); + List 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 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: diff --git a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProvider.java b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProvider.java index d1e5569b3fec90..d7b515af0c33ce 100644 --- a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProvider.java +++ b/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 scanRanges) * {@link ConnectorScanProfile} groups the engine writes into the query's profile execution summary. * *

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.

+ * 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.

* * @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 diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java index ec13b7e04b8a10..7197213cc61a96 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java @@ -416,11 +416,7 @@ static void writeScanProfilesInto(RuntimeProfile executionSummary, List entry : profile.getMetrics().entrySet()) { scan.addInfoString(entry.getKey(), entry.getValue()); @@ -429,6 +425,19 @@ static void writeScanProfilesInto(RuntimeProfile executionSummary, List { ConnectorSplitSource source = null; @@ -1898,6 +1911,9 @@ private void startStreamingSplit() { LOG.warn("Failed to close streaming split source for {}", handle, ce); } } + List scanProfiles = onPluginClassLoader(scanProvider, + () -> scanProvider.collectScanProfiles(connectorSession)); + writeScanProfilesInto(executionSummary, scanProfiles); } }, scheduleExecutor); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProfileTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProfileTest.java index 49f22484a9e033..6cc1cb003dbe3a 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProfileTest.java +++ b/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> 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"); + } }