suxiaogang223 commented on code in PR #66521:
URL: https://github.com/apache/doris/pull/66521#discussion_r3765100529
##########
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:
Thanks, this is a valid observation, but it is outside the intentionally
narrow scope of this PR. This change restores batch/non-batch parity for the
Iceberg SDK `planFiles()` reporter path: when the non-batch path emits a
ScanReport, batch mode now preserves it too. Manifest-cache enumeration
bypasses `planFiles()` and currently emits no equivalent reporter metrics in
either mode, so supporting it requires separate cache hit/miss instrumentation
and fallback semantics. I narrowed the PR description and release note
accordingly; this should be tracked separately.
##########
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:
This is a valid existing profile-key collision, but it is not introduced by
batch-mode metric collection: both eager and streaming profiles derive `Table
Scan (db.table)`, and RuntimeProfile replaces duplicate child names. Fixing it
requires a stable per-scan-node identity/naming contract across all scan modes.
That is broader than this PR’s batch-mode parity scope, so I documented the
boundary and will leave the naming change to a follow-up.
##########
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:
Fixed in daaae7c1f91. The streaming source owner is registered before async
dispatch, so cancellation closes both active sources and sources created after
stop. Source close, profile collection, and profile insertion now happen before
success/error is published. The original scan failure remains primary when
profile collection also fails, with the profile failure suppressed. Added
completion-ordering and cancellation ownership tests.
##########
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:
Fixed in daaae7c1f91. Byte formatting now uses a per-call DecimalFormat
instead of shared mutable state, and a parallel reporter test covers concurrent
formatting.
--
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]
