github-actions[bot] commented on code in PR #66778:
URL: https://github.com/apache/doris/pull/66778#discussion_r3784324140
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -457,10 +453,23 @@ public long streamingSplitEstimate(ConnectorSession
session, ConnectorTableHandl
if (snapshot == null) {
return -1;
}
- if (getFormatVersion(table) >= 3) {
- return -1;
+ if (countPushdown && filter.isEmpty()) {
+ boolean netPositionDeletes = sessionBool(session,
IGNORE_ICEBERG_DANGLING_DELETE, false);
+ ManifestDeleteState deleteState =
manifestDeleteState(snapshot.deleteManifests(table.io()));
+ // Keep synchronous planning only while COUNT(*) can still
collapse to one range. Once live deletes
+ // make that impossible, normal file enumeration needs the same
streaming OOM protection as SELECT.
+ if (deleteState == ManifestDeleteState.NONE) {
+ return -1;
+ }
+ if (deleteState != ManifestDeleteState.PRESENT ||
netPositionDeletes) {
+ OptionalLong positionDeleteRows =
livePositionDeleteRowCount(table, snapshot);
Review Comment:
[P2] Keep the v3 fence ahead of the delete-manifest proof
On a format-v3 snapshot with live position deletes and
`ignore_iceberg_dangling_delete=true`, this call opens and walks every live
delete-file entry, then the COUNT block selects synchronous planning. But the
unconditional v3 fence below makes that batch result invariant, and synchronous
`planCountPushdown` calls `livePositionDeleteRowCount` again. The base ordering
fenced v3 before this proof. For large position-delete/deletion-vector
manifests, the estimator therefore adds a full remote metadata scan that cannot
affect its result. Please move the v3 fence ahead of the COUNT block and cover
the flagged v3/live-delete estimator path.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -1241,58 +1251,245 @@ private static Schema pinnedSchema(Table table,
IcebergTableHandle handle) {
}
/**
- * Emit the single collapsed COUNT(*)-pushdown range: the first whole-file
{@link FileScanTask} from
- * {@code scan.planFiles()} carrying the full {@code realCount} via {@code
table_level_row_count} → BE's
- * count reader serves it without opening the data file. Mirrors paimon's
{@code buildCountRange} (one
- * range bearing the summed total). Result-identical to legacy's count
short-circuit even though legacy
- * takes a different shape: legacy byte-splits the count file ({@code
planFileScanTask} →
- * {@code splitFiles} → {@code TableScanUtil.splitFiles}), keeps the first
split task's byte-range for
- * {@code count < 10000}, and {@code assignCountToSplits} distributes the
same total — but under count
- * pushdown BE's count reader never reads the file (the range's
start/length are irrelevant) and sums
- * {@code table_level_row_count} across ranges, so one whole-file range
yields the identical total (and
- * legacy's {@code >10000} parallel multi-split trim is the perf-only
divergence we drop). An empty table
- * (no files) yields no range, so BE gets 0 ranges and COUNT returns 0
(legacy returns empty splits too).
+ * Build a collapsed COUNT(*) range from current manifest-list aggregates.
Summing each data manifest's
+ * added and existing row counts is O(manifests), while only the first
live {@link FileScanTask} is needed as
+ * the representative range. Old manifest lists that omit these aggregates
use the bounded O(files) fallback.
+ * Equality deletes and non-netted position deletes make the optimization
unsafe and tell the caller to
+ * perform a normal scan.
*/
- private List<ConnectorScanRange> planCountPushdown(Table table, TableScan
scan, long realCount,
+ private Optional<List<ConnectorScanRange>> planCountPushdown(Table table,
TableScan scan,
int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer, ConnectorSession session,
Optional<ConnectorExpression> filter) {
- try (CloseableIterable<FileScanTask> tasks =
countPushdownFileScanTasks(scan, session, table, filter)) {
- for (FileScanTask task : tasks) {
- // targetSplitSize = -1: the count-pushdown collapse emits a
single range, so its scheduling
- // weight is irrelevant → PluginDrivenSplit keeps
SplitWeight.standard().
- return Collections.singletonList(buildRange(table,
task.file(), task, formatVersion,
- partitioned, orderedPartitionKeys, zone,
uriNormalizer, realCount, -1, null));
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return Optional.of(Collections.emptyList());
+ }
+
+ boolean netPositionDeletes = sessionBool(session,
IGNORE_ICEBERG_DANGLING_DELETE, false);
+ ManifestDeleteState deleteState =
manifestDeleteState(snapshot.deleteManifests(table.io()));
+ if (deleteState == ManifestDeleteState.PRESENT && !netPositionDeletes)
{
+ return Optional.empty();
+ }
+ OptionalLong positionDeleteRows = deleteState ==
ManifestDeleteState.NONE
+ ? OptionalLong.of(0)
+ : livePositionDeleteRowCount(table, snapshot);
+ if (!positionDeleteRows.isPresent()
+ || (!netPositionDeletes && positionDeleteRows.getAsLong() !=
0)) {
+ return Optional.empty();
+ }
+
+ OptionalLong manifestCount =
liveRowCountFromManifests(snapshot.dataManifests(table.io()));
+ if (manifestCount.isPresent()) {
+ // Compatibility mode nets each live position-delete file's rows
once without trusting summary
+ // counters; it intentionally cannot distinguish dangling entries,
which is why the flag is opt-in.
+ OptionalLong visibleRows = subtractPositionDeleteRows(
+ manifestCount.getAsLong(), netPositionDeletes ?
positionDeleteRows.getAsLong() : 0);
+ if (!visibleRows.isPresent()) {
+ return Optional.empty();
+ }
+ return planManifestCountRange(table, scan,
visibleRows.getAsLong(), formatVersion,
Review Comment:
[P2] Avoid rebuilding the delete index just to choose a COUNT representative
The global proof above has already opened every live delete manifest and
either rejected equality deletes or obtained the position-delete count. This
branch then chooses its first `FileScanTask` through `scan.planFiles()` or the
cache-backed path; both eagerly reread all delete manifests to build a delete
index before yielding that task. The range carries the exact table-level count,
so BE short-circuits without applying the representative task's delete
bindings. Successful flagged counts therefore pay a second full delete-manifest
scan even when batch mode is disabled. Please obtain the representative data
file without constructing unused delete bindings (or reuse the proof's
entries), and test that synchronous flagged count planning opens each delete
manifest only once.
--
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]
