perf: dedupe Iceberg residuals and delete files in native scan serde#4982
perf: dedupe Iceberg residuals and delete files in native scan serde#4982mbutrovich wants to merge 13 commits into
Conversation
|
@sandugood are you able to test this PR branch to see if it resolves the issue for you? Thanks for reporting! I also added some INFO level instrumentation that might help us track it down if it's still failing. |
|
Testing it right away |
Thank you so much! The new logging I added looks like this: |
|
The error still persists. Here are the logs: So at one point the number of tasks just explodes and exceeds the 2GB threshold. Here is the error stacktrace: |
|
Thanks @sandugood, that log is exactly what we needed. The instrumentation pins it, and it is not the delete files: The Two things stand out. A single residual is normally a small reduced predicate, so ~250 KB each means the scan's filter is very large, and 13,695 distinct residuals means each task carries its own copy. The residual is only used for row-group pruning on the native side; the post-scan filter still enforces correctness, so we can safely bound or skip it. To make sure the fix targets the real shape: what does the pushed-down filter on the source scan look like? The residual comes from Iceberg's per-file |
You've nailed down exactly the case - it is a large static filter: |
I've looked at this a bit and a few problems with Iceberg serde pop out at me for residuals: 1) dedupe doesn't really work because they all get unique I can try to come up with a cleaner way to do this. I'm afraid it ended up with an inefficient implementation when trying to POC the Iceberg scan. But we wanted correctness first, and then once it was correct we could optimize. This is the optimize step :) |
- proto: IcebergLiteral/IcebergDecimal/IcebergPredicate + IcebergPredicateOperator (mirrors iceberg-rust); PartitionValue unified onto IcebergLiteral; residual_pool → repeated IcebergPredicate. - Rust: iceberg_predicate_to_predicate + iceberg_literal_to_datum, partition_value_to_literal rewrite, rewrite_not() before bind, old Spark-expr conversion cluster deleted. - Scala: icebergExprToProto + proto builders + predicateLiteralToProto, partitionValueToProto/icebergLiteralToProto, residual site + pool type, old conversion cluster deleted, imports cleaned (ByteString/Base64 imported). Tests: - New predicate fuzz test in CometFuzzIcebergSuite (sample-from-data + Column API over primitive columns: comparisons, !=, explicit NOT, IN/NOT IN, IS [NOT] NULL, AND/OR) — the coverage that would have caught the NOT bug. - Residual dedup test tightened: asserts one pool entry shared by many task references (no longer vacuously passable), logInfo removed. - Delete-file dedup test: logInfo removed.
mbutrovich
changed the title
| } | ||
| } | ||
|
|
||
| test("Iceberg temporal types written as INT96") { |
There was a problem hiding this comment.
Iceberg's writer doesn't write INT96. The only time they can end up in a Parquet data file is migration, which "migration - INT96 timestamp" covers in CometIcebergNativeSuite. These ParquetOutputTimestampType tests in CometFuzzIcebergSuite are a waste.
| assert(testedBinary) | ||
| } | ||
|
|
||
| test("regexp_replace") { |
There was a problem hiding this comment.
decode and regexp_replace are exercise elsewhere and this is is redundant.
|
@sandugood once CI goes green, do you mind trying this PR again? I refactored the serde code and added some new tests. Local testing looked good, but I want to make sure that CI goes green before you try it on your table. Thank you in advance! |
Sure |
|
Yeah, I think there is still some refactor needed to push the CI to green |
… column index and tests
Thanks for trying again. I think I cleaned up the failures. Could you try a clean build? That failure looks like proto wasn't rebuilt. Thanks again for all of your help so far @sandugood! |
|
Yeah, the proto seemed to be the issue |
Wonderful! |
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Review assisted by an LLM (Claude); a human reviewed the findings before posting.
Nice fix, and the instrumentation-driven diagnosis via @sandugood's logs is a great artifact of the PR. All CI green and reporter confirms the workload unblocks. A few small things I noticed while reading through.
Stale documentation in planner.rs
The block comment at native/core/src/execution/planner.rs:4094-4106 still describes the old conversion path:
// Predicates are converted through Spark expressions rather than directly from
// Iceberg Java to Iceberg Rust. This leverages Comet's existing expression
// serialization infrastructure, which handles hundreds of expression types.
//
// Conversion path:
// Iceberg Expression (Java) -> Spark Catalyst Expression -> Protobuf -> Iceberg Predicate (Rust)
That path is exactly what this PR removes. Also, line 4108 has a leftover one-line docstring /// Converts a protobuf Spark expression to an Iceberg predicate for row-group filtering. above the new (correct) multiline docstring on iceberg_predicate_to_predicate.
Nested-column gap in pageIndexUnsupportedColumns
IcebergReflection.pageIndexUnsupportedColumns iterates schema.columns(), which returns top-level fields only. If Iceberg ever emits a residual on a nested field (name arrives as a dotted path like struct.decimal_field), the gate would miss it and let a FIXED_LEN_BYTE_ARRAY predicate through. The docstring already flags this as top-level. Do we know whether Iceberg's ResidualEvaluator can produce nested references from a Spark push-down? If so it might be worth a follow-up test.
Missed pushdown for ByteType/ShortType
predicateLiteralToProto in CometIcebergNativeScan.scala handles Boolean/Integer/Long/Float/Double/Date/String/Timestamp/TimestampNTZ but not ByteType or ShortType. The old extract_literal_as_datum widened ByteVal/ShortVal to Datum::int. On Iceberg 1.6.0+ (which supports byte/short natively, and the fuzz base only excludes them for older versions), filters on a byte or short column now silently skip pushdown. Correctness is fine, but it's a small regression on that surface. Two extra cases widening to int_val would restore parity.
Similarly, BinaryType falls through to None. iceberg-rust supports BYTE_ARRAY in the page index (only FIXED_LEN_BYTE_ARRAY is the problem this PR is guarding), so binary predicates could in principle push. Deliberate exclusion or worth revisiting?
Fuzz test observation, not a defect
filter pushdown - fuzz predicates over primitive columns asserts results match Spark and that the scan stays native. Both hold when predicateLiteralToProto returns None — the scan just skips pushdown and CometFilter picks it up. So the byte/short/binary case above wouldn't fail this test. Worth knowing when relying on it as regression coverage.
Test coverage strengths worth calling out
The dedup regressions are well-designed. delete file pool does not duplicate shared delete files counts raw-byte occurrences (structure-agnostic) and guards totalReferences > distinctPaths.size against a vacuous pass. residual pool dedups a shared non-partition filter to one entry similarly guards tasksWithResidual > 1. The three FIXED_LEN_BYTE_ARRAY tests all set write.parquet.row-group-size-bytes = 100 to force the page index into existence, which is what actually exercises the failure surface. NOT over a partially supported AND does not drop rows is exactly the right anchor for the whole-or-nothing rule.
Nice bonus: float literals no longer widen to double (Datum::double(*v as f64) → Datum::float(*v)) and date/timestamp literals push with the correct type. Those were real latent issues on the old path.
Nothing here is a block. Mostly the stale planner.rs comment and the byte/short/binary pushdown gaps are worth a look.
|
Thanks for the careful read, @andygrove. All four addressed in the latest commit. Stale Byte/Short pushdown. Restored. Nested-column gate. I verified the concern does not reach the gate. Spark does emit nested references as dotted paths, but Comet reads the residual's column name ( Binary. This one turned out to be the opposite of a missed optimization. iceberg-rust's page-index evaluator accepts Fuzz coverage. Good catch that the existing test cannot see pushdown. I added |
# Conflicts: # native/core/src/execution/planner.rs
3 participants
Which issue does this PR close?
Closes #4944.
Rationale for this change
CometIcebergNativeScanExecbroadcasts an Iceberg scan's shared data as a singleIcebergScanCommonprotobuf. On a large merge-on-read plan this crashed the driver:toByteArrayallocatesnew byte[getSerializedSize()], andgetSerializedSize()returnsint, so a message over 2 GiB wraps to a negative length. Per-pool instrumentation on the reporter's plan showed the dominant contributor was theresidual_poolat ~3.43 GB (13,695 residuals, one per task, zero deduplication), not the delete files. Two serde issues let pools grow with the number of tasks; this PR fixes both so the common message scales with the number of distinct values rather than references.Residual pool (the confirmed cause)
Residuals were serialized through the general Spark expression serde (
exprToProto), which stamps every node with a freshexpr_id(from a global counter) and aquery_context(the full SQL text) for native ANSI error reporting. Neither is used on the residual path: a residual only drives row-group pruning, and the post-scanCometFilterenforces correctness. The per-nodeexpr_idmade every task's residual serialize to distinct bytes, so structurally identical residuals never deduplicated and the pool grew with the task count.Delete files
Each
DeleteFileListheld full copies of everyIcebergDeleteFilein a task's delete set. Under Iceberg's default partition delete granularity one delete file applies to many data files, and staggered sequence numbers give each task a different subset of the same files, so a shared delete file was re-serialized once per distinct set. Iceberg's ownDeleteFileIndexstores each delete file once and references it many times; the serde did not preserve that sharing.What changes are included in this PR?
Replace the residual's Spark-expression round-trip with a purpose-built Iceberg-native serde, and intern delete files by index. The Iceberg path no longer depends on the Spark expression format at all.
operator.proto): addIcebergLiteral(a typed primitive-valueoneof, shared by partition values and predicate literals) andIcebergPredicate(mirroring iceberg-rust'sPredicateandPredicateOperator).residual_poolbecomesrepeated IcebergPredicate;PartitionValueunifies ontoIcebergLiteral. Adddelete_file_pooland makeDeleteFileListholduint32indices into it.CometIcebergNativeScan.scala): serialize residuals directly toIcebergPredicate(noexpr_id/query_context); a sharedicebergLiteralToProtobuilds both partition values and predicate literals; delete files are interned into the flat pool by path.planner.rs):iceberg_predicate_to_predicatemaps the proto onto iceberg-rustReferencebuilders (withrewrite_not()before bind, since the scan evaluators require negation-normal form); delete files resolve from the flat pool by index.Two rules govern what a residual pushes. A residual is only a pruning hint, so anything unrepresentable is safely omitted and left to
CometFilter:page_index_evaluator.rs), so any predicate on adecimal,uuid, orfixedcolumn fails the scan, including theIS NOT NULLthat Iceberg adds for every filtered column. These columns are identified by their Iceberg type from the table schema, since Spark surfacesuuidasStringTypeandfixedasBinaryTypeand cannot distinguish them from ordinary string/binary columns.NOT: De Morgan makesNOT(A AND B)equal toNOT A OR NOT B, where droppingBstrengthens the predicate and would wrongly prune. Rather than track polarity across arbitrary nesting, any unconvertible leaf elides the entire residual.Other intended changes:
float,date, andtimestamp[tz]literals now push with the correctDatumtype (previously widened or dropped), so pruning is more precise.NOT_INis still not pushed (inherently unprunable from column stats). Correctness is unchanged in all cases; the general Spark expression serde used by every other operator is untouched. Unrecognized ops or literal types degrade to no-pushdown, so the format stays correct across Iceberg library versions and new primitive types (geometry, variant, nanosecond timestamps) are additive.serializePartitionsalso logs per-pool byte sizes beforetoByteArray, so a future overflow names its dominant pool directly.How are these changes tested?
CometIcebergNativeSuite(residual-filter, transform/decimal partition, positional/equality/MOR delete, andNOT/comparison filter tests) passes unchanged, confirming the refactor is behavior-preserving on results.CometFuzzIcebergSuite: per primitive column it samples real values and builds predicates through the typedColumnAPI (comparisons,!=, explicitNOT,IN/NOT IN,IS [NOT] NULL,AND/OR) against Spark, exercising every operator and type through the new serde.decimal,uuid, andfixedcolumns confirm the scan stays native and returns correct results when the page-index gate drops the predicate.NOT over a partially supported AND does not drop rowsconfirms a residual mixing a supported and an unsupported leaf is elided rather than partially pushed.