github-actions[bot] commented on code in PR #70410:
URL: https://github.com/apache/doris/pull/70410#discussion_r4545082518
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -653,9 +650,48 @@ private CloseableIterable<FileScanTask>
streamingFileScanTasks(TableScan scan, C
}
}
scanContext.checkCancelled();
+ if (!isManifestCacheEnabled() && canPlanManifestsWithoutCache(scan,
session)) {
+ return TableScanUtil.splitFiles(
+ manifestFileScanTasks(scan, session, table, filter, null,
readControl, readHandle, scanContext),
Review Comment:
[P1] Project row predicates before pruning delete manifests
For a cache-disabled streaming scan, a format-v2 unpartitioned table with at
least two small data manifests and a delete manifest now reaches this branch
for `WHERE id = 1`. The eager delete loop passes that row expression to
`ManifestEvaluator.forPartitionFilter` (line 3286), which [binds directly
against the partition
struct](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.12.0/api/src/main/java/org/apache/iceberg/expressions/ManifestEvaluator.java);
the empty struct has no `id`, so planning throws `ValidationException`.
Previously the SDK projected the predicate first; this new branch has no
fallback. Use `forRowFilter` or an inclusive partition projection, and cover
non-partition/transformed-partition predicates with deletes.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -653,9 +650,48 @@ private CloseableIterable<FileScanTask>
streamingFileScanTasks(TableScan scan, C
}
}
scanContext.checkCancelled();
+ if (!isManifestCacheEnabled() && canPlanManifestsWithoutCache(scan,
session)) {
+ return TableScanUtil.splitFiles(
+ manifestFileScanTasks(scan, session, table, filter, null,
readControl, readHandle, scanContext),
+ sliceSize);
+ }
return TableScanUtil.splitFiles(sdkStreamingFileScanTasks(scan,
scanContext), sliceSize);
}
+ private boolean canPlanManifestsWithoutCache(TableScan scan,
ConnectorSession session) {
+ if (!SchemaAwareDataTableScan.supportsLocalManifestPlanning(scan)) {
+ return false;
+ }
+ ConnectorTaskScheduler scheduler = manifestReadScheduler(session);
+ // Standalone callers have no engine-owned workers/lifetime. Keep the
SDK's executor and scan
+ // semantics rather than silently turning its parallel planning into
inline serial reads.
+ if (scheduler == null || scheduler == ConnectorTaskScheduler.DIRECT) {
+ return false;
+ }
+ Snapshot snapshot = scan.snapshot();
+ return snapshot != null &&
canBufferManifestFiles(snapshot.dataManifests(scan.table().io()),
+ manifestReadConcurrency(session));
+ }
+
+ static boolean canBufferManifestFiles(List<ManifestFile> manifests, int
window) {
+ // A single manifest already streams without ParallelIterable in the
SDK. Preserve early delivery
+ // within large manifests too: our prefetcher retains whole parsed
manifests, whereas SDK consumers
+ // can take the first file before its manifest finishes. Bound the
window to the SDK's default
+ // 30,000-file planning buffer; unknown counts retain the streaming
SDK path.
+ if (manifests.size() <= 1) {
+ return false;
+ }
+ int filesPerManifest = MAX_UNCACHED_MANIFEST_WINDOW_FILES / window;
+ for (ManifestFile manifest : manifests) {
+ Integer added = manifest.addedFilesCount();
+ Integer existing = manifest.existingFilesCount();
Review Comment:
[P2] Count deleted entries before choosing whole-manifest loading
For an eligible uncached streaming scan, this guard admits a data manifest
with one live file and tens of thousands of `DELETED` entries because only
added+existing is checked. Such manifests occur after deletes/overwrites, and
[ManifestReader still traverses those
entries](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/ManifestReader.java).
`IcebergManifestReader` drains the entire reader before returning, and the
first required manifest is loaded synchronously, so even `LIMIT 1` can wait for
that large manifest instead of consuming an early live entry through SDK
streaming. Include deleted entries (with conservative unknown-count handling),
or a physical-size bound, in the fallback decision.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -653,9 +650,48 @@ private CloseableIterable<FileScanTask>
streamingFileScanTasks(TableScan scan, C
}
}
scanContext.checkCancelled();
+ if (!isManifestCacheEnabled() && canPlanManifestsWithoutCache(scan,
session)) {
Review Comment:
[P2] Preserve table metrics reporting when bypassing SDK planning
For native `BaseTable` streaming scans with multiple small manifests, this
new cache-disabled branch never invokes `TableScan.planFiles()`. That bypasses
[SnapshotScan's ScanEvent and close-time MetricsReporter
callback](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/SnapshotScan.java),
which the old `withFileIO` SDK path preserved. Configured table reporters,
including REST-catalog scan reporting, therefore stop receiving these scans.
Preserve the native reporting lifecycle around local planning and add a
streaming reporter test; the existing historical reporter test exercises only
`planScan`. This concerns table/catalog reporting, independently of Doris's
synchronous query-profile reporter.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -3251,10 +3286,10 @@ private CloseableIterable<FileScanTask>
cacheBackedFileScanTasks(TableScan scan,
if (!ManifestEvaluator.forPartitionFilter(filterExpr, spec,
caseSensitive).eval(manifest)) {
continue;
}
- ManifestCacheValue value =
manifestCache.getCachedManifestValue(manifest, statsQueryId);
+ ManifestCacheValue value = cached.apply(manifest);
if (value == null) {
try {
- value = deleteReadContext.read(manifest, m ->
manifestCacheGet(m, table, statsQueryId));
+ value = deleteReadContext.read(manifest, loader);
Review Comment:
[P2] Keep uncached delete-manifest reads parallel
The new uncached streaming route reaches this eager loop, where every
`deleteReadContext.read` finishes before the next manifest starts. Previously
SDK `DataTableScan` passed its executor through `ManifestGroup` to
[DeleteFileIndex's parallel delete-manifest
loader](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java).
An unfiltered streaming scan with two small data manifests and many remote
delete manifests passes the new guard but now performs all those reads serially
before producing any split, regardless of `iceberg_manifest_load_concurrency`.
Use bounded parallel delete loading, or retain SDK planning for this case until
it is supported.
--
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]
