github-actions[bot] commented on code in PR #66521:
URL: https://github.com/apache/doris/pull/66521#discussion_r3763596135
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java:
##########
@@ -1898,6 +1911,9 @@ private void startStreamingSplit() {
LOG.warn("Failed to close streaming split source for
{}", handle, ce);
}
}
+ List<ConnectorScanProfile> scanProfiles =
onPluginClassLoader(scanProvider,
Review Comment:
[P1] Drain the reporter before signaling split completion
Both `finishSchedule()` and `setException()` notify split consumers before
this `finally` closes the source and drains its report. There is consequently
no happens-before edge from metric insertion to query/profile completion: the
producer can be descheduled after the terminal signal and the final profile can
be pushed without these metrics. Cancellation is worse because
`SplitAssignment.stop()` neither closes this local source nor awaits the
unretained future. Make close + collection part of the producer completion
barrier before publishing success/error, and make stop own/close the active
source while preserving the original scan error.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -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));
Review Comment:
[P1] Preserve metrics on the manifest-cache streaming path
When `meta.cache.iceberg.manifest.enable=true`, `streamingFileScanTasks`
returns the custom `cacheBackedFileScanTasks` iterable. That iterator reads
manifests and builds `BaseFileScanTask`s directly; neither it nor its `close()`
ever calls `scan.planFiles()`, which is the operation that fires this attached
reporter. The scan therefore returns all splits but `collectScanProfiles()`
drains nothing, so the release-note behavior still disappears for a supported
batch path. Please publish equivalent metrics for the cache-backed enumeration
(including its fallback/error semantics) and cover a cache-enabled stream with
an exactly-one-profile assertion.
##########
fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeScanProfileTest.java:
##########
@@ -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 + ")";
Review Comment:
[P2] Make the concurrency regression deterministic
This test does not force the pre-fix interleaving. The 32 common-pool tasks
have no ready/start barrier or test seam that makes two writers both observe
the group as absent before either adds it; if the pool has one worker, or the
first short task completes lookup/add before the next lookup, the old racy
implementation also reports 32 children and the test passes. Please use a
controlled latch/barrier or an instrumented lookup/add seam so the test
reliably fails without the new synchronization.
##########
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());
Review Comment:
[P1] Keep profiles from repeated scans of the same table
Serializing group creation does not prevent loss when two scan nodes have
the same child name. Iceberg derives `scanLabel` only from
`scanReport.tableName()`, so a self-join or two `UNION ALL` branches over the
same table both produce `Table Scan (db.t)`. `RuntimeProfile.addChild` then
removes and replaces the existing same-named child, leaving only one branch's
metrics. Please give each scan occurrence a stable unique identity (or
otherwise preserve duplicate labels), and add a same-label
sequential/concurrent test rather than only `t0` through `t31`.
##########
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
Review Comment:
[P1] Preserve the API-4 collection lifecycle
Before this change, this public SPI contract guaranteed that
`collectScanProfiles` was called immediately after `planScan` on that same
thread. FE now also calls an existing override after `streamSplits` on the
schedule executor, on a path that never called `planScan`. An API-4 external
connector may implement streaming and eager diagnostics while validly depending
on prior `planScan` state or planning-thread context; the unchanged version
gate will still load it and FE can now call it in an unsupported lifecycle.
Please add a separate opt-in streaming-profile hook with a safe default, or
treat the lifecycle/threading expansion as an incompatible API revision and
bump/adapt the plugin contract.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -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));
Review Comment:
[P1] Avoid the shared formatter on concurrent report callbacks
This makes `IcebergScanProfileReporter.report()` reachable concurrently from
separate streaming scan-node tasks, but every reporter in the plugin
classloader shares the static `DecimalFormat BYTES_FORMAT`.
`NumberFormat`/`DecimalFormat` require separate instances or external
synchronization, so concurrent byte-counter rendering can corrupt profile
strings or throw while the source is closing. Use per-call or thread-local
formatting and add a parallel reporter test; the new concurrency test exercises
only RuntimeProfile insertion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
