Comet 1.1.0 Release (September)

[andygrove]

What is the problem the feature request solves?

Let's use this issue to plan the next major release.

Issues and PRs tagged for 1.1.0: https://github.com/apache/datafusion-comet/milestone/21

Describe the potential solution

No response

Additional context

No response

[andygrove]

Proposed priorities for 1.1.0: correctness first

I went through the open backlog to work out what we should resolve before cutting 1.1.0, focusing on correctness bugs and real performance regressions rather than new features. The milestone currently has 61 open issues, but most of the critical correctness bugs are not on it: of the 15 open issues labelled priority:critical, only 5 carry the 1.1.0 milestone. The ranking below ignores the milestone and orders by user impact and how close a fix is.

Tier 1: silent wrong results with an approved fix

These just need to land.

Issue	Problem	Fix PR	Status
#5824	Semi/anti joins and explode_outer share a reused exchange and return duplicated rows	#5828	Approved, CI green
#5783	Native scan returns N times the rows for a struct with duplicate field names	#5786	Approved, needs rebase
#5701	array_distinct / array_union diverge on -0.0 on Spark versions without SPARK-54918	#5750	Approved, CI green
#5366	to_time rejects inputs Spark accepts (already on the 1.0.1 milestone)	#5738	Approved, one failing check

Tier 2: silent wrong results, fix needs review or does not exist yet

This is where I would put reviewer time before anything else in the milestone.

• AVG returns NULL from empty native partials (Incorrect AVG result from empty native partial buffers #5418, Unsafe native partial aggregates survive child-triggered final fallback #5419). A plain avg(x) ... WHERE ... with a Spark final aggregate can return NULL. PRs fix: preserve Spark AVG buffer and overflow semantics #5420 and fix: revert unsafe partial aggregates after final fallback #5421 are open. Match Spark ObjectHashAggregate decimal AVG buffer semantics #5509 is the decimal AVG follow-on under ObjectHashAggregateExec.
• ABFS object store cache key drops the container (Object store cache key drops the ABFS container, so two containers in one storage account share a store instance #4993). Two Azure containers in one storage account share a store instance. PR fix: include ABFS container in object store cache key #5053 has no reviewer yet. Cross-container data mixing is the worst outcome on this list.
• Four serdes give wrong answers for a nondeterministic child (size, arrays_zip, map_from_arrays and array_append return wrong answers for a nondeterministic child #5781). size, arrays_zip, map_from_arrays and array_append serialize the child twice. No PR and unassigned, but feat: address remaining issues for CreateArray #5766 already fixed the same shape for element_at, so it should be a small change.
• Nested Parquet schema evolution silently NULLs or parses strings (Nested (struct/list/map) Parquet schema-evolution conversions bypass Spark's type-conversion rules: silent NULLs, string parsing, and a native panic #5671). Conversions Spark rejects at top level are performed inside structs. PR fix: apply Spark's Parquet conversion rules to nested struct/list/map fields #5681 is open.
• Wide date-to-timestamp casts wrap or panic (Date-to-timestamp casts can overflow or panic for wide dates #5456). PR fix: make wide date-to-timestamp casts safe #5457 is open.
• array_min / array_max signed-zero ties ([Bug] array_min and array_max differ from Spark on signed-zero ties #5401). PR fix: match Spark ordering in native array extrema #5403 is open.
• get_json_object duplicate keys (Native implementation of get_json_object returns last value for duplicate keys, Spark returns first #4947). PR fix: get_json_object returns first value for duplicate keys to match Spark #4971 is open.
• Nested float ordering and rank (Match Spark ordering and rank semantics for floating values nested in arrays and structs #5507). Assigned, no PR yet.
• CometHashAggregateExec and AQEPropagateEmptyRelation (CometHashAggregateExec doesn't participate in Spark's AQEPropagateEmptyRelation optimization #4412). Known to cause wrong results. On the 1.0.1 milestone, no PR.
• Map null-key rejection and mapKeyDedupPolicy ([Bug] map_from_arrays / map_from_entries do not enforce null-key rejection or spark.sql.mapKeyDedupPolicy #4680, str_to_map does not honour mapKeyDedupPolicy=LAST_WIN #5857). PRs fix: enforce null-key rejection and mapKeyDedupPolicy in native map construction #5854 and fix: validate map constructor row lengths and null short-circuiting #5846 are open.

Tier 3: crashes and panics

• Native panic casting to negative-scale decimal when spark.sql.legacy.allowNegativeScaleOfDecimal=true #5013 negative-scale decimal panic. PR fix: fall back to Spark for negative-scale decimal casts and arithmetic #5050 is open.
• timestamp_trunc panics on DST-transition timestamps in a DST timezone #5633 timestamp_trunc panics on DST transitions. Assigned.
• bug: slice crashes on arrays with a non-nullable element field #5743 slice crashes on non-nullable element arrays. Assigned.
• Native collect_set(struct) aggregate crashes on spill emit #5239 collect_set(struct) aborts on spill. Root cause is GroupedHashAggregateStream::emit throws ArrowError: column types must match schema types datafusion#24069; worth checking whether the DataFusion 55 bump already fixed it.
• Null CalendarInterval literal throws native exception instead of evaluating to null #5058 and Hashing a CalendarInterval value fails with "Unsupported data type in hasher: Interval(MonthDayNano)" #5059 CalendarInterval null literal and hashing. PRs fix: support null calendar interval literals #5133 and fix: support CalendarIntervalType hashing #5135 have been open since July.
• Native shuffle accepts structs with duplicate field names, then fails importing the batch back to the JVM #5605 native shuffle accepts duplicate struct field names then fails on import. Same family as Native Parquet scan multiplies rows for a struct with duplicate field names #5783 and Duplicate field ids inside a struct are not validated when the file schema equals the requested schema and no predicate is pushed #5801.
• Sporadic SIGSEGV in macOS [scans] workflow in CI #5023 sporadic SIGSEGV in the macOS scans CI job. Unassigned; a real segfault that should at least get a root cause before release.
• revertToSpark erases CometIcebergWriteExec / CometNativeWriteExec because originalPlan is the node's own child #5719 revertToSpark drops the write node. Only affects the opt-in native write paths, so lower priority.

Memory and performance regressions

Only two items look like genuine regressions rather than long-standing gaps.

• Fair pool caps a task at pool_size / num_consumers ([EPIC] Memory pool and accounting audit sweep #5212, finding 1). Users hit OOM well below their configured off-heap size. PRs fix: enforce fair memory share per reservation #5466, fix: account fair-pool memory across sibling reservations #5847 and perf: stop holding the fair pool lock across blocking memory calls #5613 target it and all need review.
• DataFusion 55 TPC-H off-heap bump from 2g to 4g (Establish whether the DataFusion 55 TPC-H off-heap bump is real, and changelog it if so #5704). If this is real it is a user-visible memory regression that belongs in the changelog. If it is an accounting artifact of the peak_mem_used change, the bump should be reverted. Cheap to settle either way.

Not proposed as release blockers: #4988 only matters when row-filter pushdown is opted in, #3873 and #4614 are long-standing, #5112 predates 1.0.0, and #5834 and #4949 are Spark 4.2 fallbacks. #5795 (element_at on maps slower than Spark) already has PR #5806.

Milestone housekeeping

1. I have added the critical issues that had no milestone to 1.1.0: Comet join and explode operators omit joinType / outer from equals, so exchange reuse returns wrong results #5824, Native Parquet scan multiplies rows for a struct with duplicate field names #5783, Checked TIMESTAMP_MILLIS overflow for nested fields and nested-predicate scans is blocked on DataFusion nested-field pruning #5553, Date-to-timestamp casts can overflow or panic for wide dates #5456, Unsafe native partial aggregates survive child-triggered final fallback #5419, Incorrect AVG result from empty native partial buffers #5418, [Bug] array_min and array_max differ from Spark on signed-zero ties #5401, A Rust UDF silently answers calls to an ordinary Scala UDF registered under the same name #5295.
2. I have moved the sort-aware Iceberg read work (Sort-aware Iceberg reads in Comet via a per-partition streaming merge #5323 and Sort-aware Iceberg reads: per-partition streaming merge + report ordering and grouping (v1) #5337 through Bound memory when a partition has many files #5343) plus Codify Iceberg version compatibility #4871 and Parquet reader: Add support for reading row index metadata columns #3432 to the 1.2.0 milestone, since they are enhancements rather than fixes. Process items such as Drop support for JDK 11 #4844, Drop support for Spark 3.4 #4329, Add a docs-review Claude skill to enforce style guide and documentation quality #4420 and Establish nomenclature style guide and audit operator names for clarity #4419 are still on 1.1.0 and could also move if nobody objects.

Happy to adjust the ordering if others see different priorities.

[andygrove]

What is already on main for 1.1.0

218 commits have landed on main since branch-1.0 was cut at af534e0 (74 fix, 45 feat, 21 perf, plus chores, tests, docs and CI). Highlights:

Dependencies

• DataFusion 55.0 and Arrow/Parquet 59.2 (deps: bump DataFusion 55.0 and Arrow/Parquet 59.2 #5262). Follow-ups tracked in [EPIC] DataFusion 55 upgrade follow-ups #5709.

Headline features

• Celeborn remote shuffle: native map-side push, reader and planning (feat: add RSS partition writer and task-owned JNI callback (1/n) #5473 through feat: enable native-only Celeborn shuffle planning (8/n) #5537).
• Native Iceberg V2 writer via iceberg-rust, with split writer/committer operators (feat: implement native Iceberg V2 writer via iceberg-rust #5361, feat: Optionally split the Iceberg V2 write operator into distinct writer and committer operations #4658, feat: detect Iceberg V2 writes and emit fall-back reasons #5298). Both write flags still default off.
• Native dynamic filter pushdown from hash joins into Parquet scans (feat: native dynamic filter pushdown for hash join into Parquet scans #5699).
• Aggregates: max_by/min_by (feat: support max_by and min_by aggregate expressions #4817), mode (feat: support native aggregate function mode #4782), the regr_* family (feat: implement regr_slope, regr_intercept, regr_r2, regr_sxx, regr_syy, regr_sxy aggregates #4775), WindowGroupLimitExec (feat: support WindowGroupLimitExec #4870).
• Expressions: explode_outer (feat: support explode_outer #5192), make_interval (feat: add make_interval support (codegen dispatch + native) #5039), spark_sequence (feat: add native spark_sequence kernel for integral element types #5614), unbase64 (feat: native spark_unbase64 kernel #5451), concat_ws with arrays (feat: support native concat_ws with string arrays #5725), Iceberg system functions (feat: Support Iceberg system functions (bucket, truncate, years/months/days/hours) natively #5638), plus codegen-dispatch routing for encode, timestampadd/timestampdiff, translate, to_csv, lpad/rpad, round on floats and unrecognized Invoke/StaticInvoke.
• Scan: S3-compatible filesystems (feat: support S3 compliant filesystems #5314), _metadata constant columns (feat: support _metadata constant columns in native Parquet scan #5237), unprojected Spark 4 VARIANT columns (feat: Support native scans with unprojected Spark 4 VARIANT columns #5377), unicode case-sensitive field names (feat: support unicode case sensitive field names for reading parquet #5602), scan I/O and read-amplification metrics (feat: expose native Parquet scan I/O and read-amplification metrics #5453).
• Experimental in-memory cache, disabled by default (feat: add experimental native support for in-memory cache, disabled by default #5051).

Notable correctness fixes

• Exchange reuse returned wrong aggregate results (fix: preserve aggregate result identity during exchange reuse #5470).
• Silent overflow reading Parquet TIMESTAMP_MILLIS (fix: prevent silent overflow when reading Parquet TIMESTAMP_MILLIS values #5177, fix: check nested TIMESTAMP_MILLIS overflow in unfiltered scans #5740).
• Float sort and window rank keys not normalized (fix: normalize scalar float sort and window rank keys #5469); struct null mask ignored when hashing fields (fix: apply the parent struct's null mask before hashing its fields #5754).
• Decimal to float/double cast rounding (fix: correctly rounded decimal to double/float cast matching BigDecimal.doubleValue/floatValue #5684); float to integral ANSI bounds (fix: match Spark's ANSI bound check for float/double to integral casts #5683); string to timestamp segment rules (fix: align string to timestamp parsing with Spark's segment rules #5682).
• Native shuffle 2GB task serialization OOM on many partitions (fix: Native shuffle fails with a 2GB task serialization OOM on jobs with many partitions #5392); memory leak after failed Arrow import (fix: prevent memory leak after failed Arrow vector import #5539).
• Memory and spill metrics were wrong or missing across shuffle, aggregate and non-shuffle stages (fix: report native shuffle write metrics accurately #5370, fix: expose native memory usage to Spark #5408, fix: report native child spill metrics in shuffle tasks #5445, fix: track peak memory before JVM shuffle spill #5463, fix: report native operator spill metrics in Spark task metrics for non-shuffle stages #5497, fix: report bounded shuffle allocator memory usage #5516, fix: report native shuffle read metrics #5554, fix: make native shuffle spill metrics independent of input batching #5628).

Performance

• collect_list/collect_set GroupsAccumulator (perf: give collect_list and collect_set a native GroupsAccumulator #5803), explode slicing instead of gathering (perf: slice the child instead of gathering it when unnesting #5667), shuffle scratch and IPC context reuse (perf: reuse Arrow IPC compression context across shuffle blocks #5038, perf: reuse per-partition scratch in the shuffle write path #5568), faster ArrowWriter (perf: improve ArrowWriter performance for fixed-length vectors #5046, perf: bulk copy fixed-width columns in ArrowWriter #5442), vectorized integer-to-decimal cast (perf: vectorize integer-to-decimal cast #4939), spark_floor (perf: optimize spark_floor (up to 4x faster) #4911), spark_base64 (perf: optimize spark_base64 in spark-expr #4885).

Full list: git log af534e0..main.

[dwsmith1983]

Two of the "no PR" items in the priorities list have PRs now: #5867 for #5781 (the four serdes decline a nondeterministic child and dispatch, with the map-key lookups from #5580 in the same change) and #5866 for #5605 (duplicate struct field names declined in the shared type gate before Java Arrow sees them, plus the native shuffle predicate). #5654 covers the field-id side of the same duplicate-name family as #5783 and closes #5801.

[dwsmith1983]

@andygrove could we try and target having the delta native contrib #5365 completed for the release of 1.1.0?

[andygrove]

@andygrove could we try and target having the delta native contrib #5365 completed for the release of 1.1.0?

Could you add a comment on the 1.1.0 tracking issue?

edit: nm, I was replying on my phone - did not realize this comment was already on this issue

[dwsmith1983]

With the end-of-month target, could #5867 (closes #5781), #5654 (closes #5801) and #5365 go on the 1.1.0 milestone? All three are approved by sunchao and waiting on CI approval.

[andygrove]

With the end-of-month target, could #5867 (closes #5781), #5654 (closes #5801) and #5365 go on the 1.1.0 milestone? All three are approved by sunchao and waiting on CI approval.

Sure, I've tagged them all with the 1.1.0 milestone and approved CI.