Repository navigation
Conversation
rich7420
added
run-all-spark-profiles
github-actions
Bot
added
enhancement
rich7420
force-pushed
the
feat/2837-native-map-explode
branch
from
September 28, 2026 14:50
76768f2 to
9601e81
Compare
Cover JVM-fed maps, nullable Boolean batch boundaries, all-NULL/empty outer batches, native map producers, and stored floating-point keys. Require native generators and matching Spark results and schemas. Benchmark map variants and nested map/array inputs with sinks that consume keys and positions. Remove the unrelated Parquet field-ID assertion change so it can be reviewed independently.
Keep input arrays as temporary placeholders until generated columns replace them. Build take indices only when a carried column consumes them. Cover recursive unnest with and without carried columns and both drop and outer null handling. Preserve the existing bounded map and array execution path.
rich7420
marked this pull request as ready for review
October 3, 2026 16:36
* feat: convert OneRowRelation to Arrow by default Turn on spark.comet.convert.oneRowRelation.enabled by default. The row a query without a FROM clause reads has no columns, so converting it costs almost nothing, and it lets the operators above it run in Comet. Where the parent cannot run in Comet, EliminateRedundantTransitions drops the conversion again, so no extra transition is added. On Spark 4.2 the TPC-DS q77a golden now records the remaining blocker, the struct-typed scalar subquery result the projection above each OneRowRelation reads (apache#5834), instead of the unsupported leaf. * fix: handle expressions exposed by converting OneRowRelation by default Folded array literals are UnsafeArrayData and have no backing array, so serialize them through toArray. might_contain accepts an untyped NULL for either argument. parse_url raises Spark's INVALID_URL error under ANSI instead of a generic native error. An Expand with no output columns falls back to Spark. * fix: decline codegen dispatch for expressions that contain a subquery * test: adapt Spark explain tests to the Comet plan for OneRowRelation queries
…eException (apache#6243) * fix: rethrow the exception a JVM input throws instead of a CometNativeException A native plan pulls each batch of a JVM input through the Arrow C stream exported by CometArrowStream.stream. When the JVM producer throws while building a batch, Arrow Java's exported stream catches the throwable and hands native only its text, so the task failed with a CometNativeException ("C Data interface error: ...") instead of the exception Spark throws. For inputs of Arrow-backed batches this happened past the first batch only, because schema reconciliation reads the first one on the JVM. Row inputs through CometSparkToColumnarExec hit it on every batch. CometArrowStream.stream now wraps each exported reader to keep the first throwable its loadNextBatch throws, keyed by the exported stream until the task completes. When executePlan fails, CometExecIterator rethrows that throwable if one of its input streams recorded one. Closes apache#6234. * fix: drop unneeded string interpolator flagged by scalafix * refactor: match the recorded input failure once, share the failed-write test helper CometExecIterator looked up the recorded input failure twice, in the guard and in the body of the catch case. An extractor does it once, and the comment at the catch site shrinks to a line now that CometArrowStream.inputFailure carries the explanation. The native-versus-JVM failed-write scaffolding in CometIcebergWriteActionSuite was copied three times. failedWrites runs a write against a native-writer table and a JVM-writer table, checks that the native run went through CometIcebergWriteExec, and returns both failures.
… review skills (apache#6340) * docs: check timezone handling against the contributor guide in the PR review skills Point the review skills at docs/source/contributor-guide/timezones.md (added in apache#6337) when a PR deals with timestamps or the session timezone. - review-comet-pr: read the page up front, and a cross-cutting "Timestamps and timezones" check covering the UTC label invariant, where the timezone comes from, Etc/UTC, NTZ values, tests, and which parts of the page a PR can make stale. - review-comet-expression-pr: the page in the guide table, serde and kernel checklist items, the page's testing guidance, and a common finding for mislabelled timestamp results. - review-comet-ffi-pr: timestamps cross the boundary unconverted and new producers label with NATIVE_TIMEZONE. - review-comet-iceberg-write-pr: timestamp partition values are UTC, and iceberg-rust's years/months depend on the +00:00 relabel. * docs: update the timezone review notes for CometTimeZone and the Iceberg partition kernels
) * feat: charge native write buffers to the task's memory pool Neither native writer reserved what it buffers, so a native Parquet or Iceberg write held its row groups in memory that no budget accounted for. A fanout Iceberg write multiplies that by the number of partitions a task writes, and when it outgrew the executor's memory overhead the container was killed. Both writers now register one memory consumer per task and resize it after every batch. ParquetWriterExec reserves parquet-rs's estimate of its in-progress row group. The Iceberg writer wraps iceberg-rust's ParquetWriterBuilder so that every open file reports what it holds, capped at the row-group size because iceberg-rust exposes only the bytes written plus the in-progress row group, and adds the rows its pacers hold back. Neither writer can spill, so a resize the pool refuses fails the task with a native out-of-memory error that Spark can retry. Closes apache#5648. Closes apache#6115. * fix: release an Iceberg file's reservation once its row groups leave memory A data file's share was its written size capped at one row group, so every fanout file that had flushed stayed charged a whole row group until it closed. A local file keeps nothing it has flushed, so a pool of eight 128 KiB row groups failed a twelve-partition write whose files held almost nothing. Each file's OutputFile is now wrapped so that its writer counts the bytes that leave memory on their way to storage, and the file reports what it has written less those. A local file releases a row group as it is handed on. S3 and GCS do not: OpenDAL collects what a file flushes into parts of at least 5 MiB and holds each part until the next one is complete or the file closes, so a file there keeps what it has flushed since its last upload charged.
) Bumps [rustls](https://github.com/rustls/rustls) from 0.23.44 to 0.23.45. - [Release notes](https://github.com/rustls/rustls/releases) - [Changelog](https://github.com/rustls/rustls/blob/main/CHANGELOG.md) - [Commits](rustls/rustls@v/0.23.44...v/0.23.45) --- updated-dependencies: - dependency-name: rustls dependency-version: 0.23.45 dependency-type: indirect ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
apache#5956) * feat: Use DataFusion date_trunc for compatible trunc and date_trunc paths Reuse the upstream physical date_trunc for scalar formats while keeping Comet per-row format, dictionary, and session-timezone handling, with a fallback outside DataFusion's TimestampNanosecond range. * perf: retain direct Date32 truncation kernel * fix: preserve timezone semantics for minute truncation * fix: take the earlier midnight for non-UTC coarse truncation * fix: align timestamp truncation with Spark DST semantics and optimize UTC * fix: preserve timestamp truncation nulls and stabilize DST tests * fix: truncate extreme UTC timestamps with integer calendar arithmetic * fix: handle dictionary and fine-grained timestamp truncation boundaries * perf: Avoid unnecessary dictionary key scans in timestamp truncation
…he codegen dispatcher (apache#6714) * perf: skip Spark's encoders for boxed primitive Scala UDF values in the codegen dispatcher A typed Scala UDF converts a boxed primitive parameter (`java.lang.Long`, ...) through its input encoder and a boxed primitive result through its output encoder, each a projection per row. The kernel now drops those encoders when their expression is exactly the boxing or unboxing call, so `ScalaUDF` passes the value through `CatalystTypeConverters` unchanged, as Spark 4 does for a Java UDF. Every other encoder is left alone. Closes apache#6706. * Compare boxed encoders with the trees Spark's builders produce Check each encoder by building the expected tree with Spark's own `DeserializerBuildHelper` and `SerializerBuildHelper` builders and comparing with `==`, in place of field-by-field matching. Drop parameter encoders with `filterNot` like the result's, and always copy the `ScalaUDF`, since an equal copy leaves its parents untouched. In the tests, register the shared UDFs once, pass vector classes directly, drop two no-op casts, and run the mixed-parameter calls as one query.
…patcher (apache#6471) * fix: Route collated array element comparisons through the codegen dispatcher array_contains, arrays_overlap, array_distinct and array_union compare array elements by raw bytes in their native kernels, so they returned wrong results for strings with a non-UTF8_BINARY collation (for example UTF8_LCASE or UTF8_BINARY_RTRIM) when a dispatched collated cast fed them. Report these cases as Incompatible and mix in CodegenDispatchFallback so they run Spark's collation-aware comparison through the JVM codegen dispatcher by default. ArrayDistinct gets its own serde for this. * docs: Mark collation-routed array functions as Hybrid in expressions.md
…apache#6730) * perf: convert Spark's unsafe rows to Arrow straight from their memory ArrowWriter's row path read every row through Spark's generic getters. A string went through a UTF8String and Arrow's per-value setter, each array, map and struct allocated a view, and every element took a generic write with two or three virtual or interface calls. Unsafe rows now take a path of their own, and unsafe arrays are written in bulk: - Fixed-width values are read from their slot at the vector's width, and arrays of them are copied in one block, with the validity inverted from the array's null words. - Strings and binaries are copied from the offset and size in their slot, short ones a word at a time. A negative size or an offset overflow is rejected before anything is written or allocated, for a value and for every element of an array. - Decimals up to 18 digits are written from the unscaled long, and wider ones are range-checked in place. - Arrays, maps and structs point one reused view at each value. Closes apache#6721. * Simplify the unsafe-row writers - Use Arrow's fillEmpties and setValueLengthSafe in place of hand-written hole filling and buffer growth, and share one variable-width overflow check. - Read the fixed-width vector through its field in every per-value method, and have the base defaults delegate to write. - Move the columnar code StringWriter and BinaryWriter shared into their base. - Write unsafe arrays of decimals up to 18 digits from the unscaled long, range-checked as UnsafeArrayData.getDecimal checks it. - Mark map entries valid with one masked write rather than a write per bit, and call setOne only for long runs. - Share the tests' reject, root-comparison and fill helpers. * test: check that a decimal past its precision in an unsafe array still fails Reads an unsafe array of decimals at a narrower precision than it was written with, holding 10^p or -10^p, for both the unscaled long and the unscaled bytes. * test: write an unsafe array of each primitive type Each of the twelve vector types whose unsafe array elements are copied in one block now reaches that copy, at every element width. Only int and long arrays did before. * fix: give each nested value a view of its own again The array, map and struct writers pointed one reused unsafe view at every value they read. Each view then kept the last row it read reachable after that row's values were copied to Arrow, along with the row's memory, until the writer was dropped. They now read each value through Spark's getters, which allocate a view per value as the generic path does. That keeps about nine tenths of the gain on nested types. A test walks the writers' fields after each row and checks that the row's memory is not reachable from them. * refactor: share the variable-width growth step between the append paths writeVariableWidth, writeDictionaryVariableWidth and the unsafe array path each filled the offsets of values skipped since the last one set and grew the validity and offset buffers. They now call reserveValues, next to reserveData. A new test appends a dictionary-encoded string field as a column after its struct took the row path, which the dictionary path's growth step had no test for. The array of each primitive type no longer repeats the two that nestedTypes lists.
…pache#6740) * fix: match Spark 4.2 date_trunc overflow for SECOND and MILLISECOND Spark 4.2.0 (SPARK-56663) raises `long overflow` when SECOND or MILLISECOND truncation falls below the smallest timestamp, where earlier versions wrap the Long subtraction. The serde now tells the native kernel which behavior to follow, and the wrapping fixture from apache#5956 is split into a 4.1-and-earlier file and a 4.2 file. * perf: keep values-only truncation for safe non-UTC SECOND/MILLISECOND dictionaries SECOND and MILLISECOND ignore the timezone, so when they do not wrap they need key masking only for a value near the lower bound. Previously a non-UTC dictionary fell through to the unconditional non-UTC masking branch and scanned every key.
…pache#6778) * docs: add an FAQ page with an answer on vectorized Java/Scala UDFs The About section gets a Frequently Asked Questions page. Its first question is whether Comet plans an API for vectorized Java/Scala UDFs like the Rust UDF API. The answer gives what the prototype in apache#6697 found: for simple functions the vectorized form saved about 3 ns per row once the code generator's overheads were removed, which did not justify the API's cost to users. It also says where a vectorized function could still pay off, and points to apache#6694 for those use cases. * docs: shorten the vectorized UDF FAQ answer
* fix: harden Iceberg delete-file reflection handling * test: cover Iceberg delete-file serde failures --------- Co-authored-by: Andy Grove <agrove@apache.org>
… is too large (apache#6384) * perf: keep the sort-merge join when the forced hash join's build side is too large With spark.comet.exec.forceShuffledHashJoin on, RewriteJoin turned every eligible SortMergeJoinExec into a shuffled hash join and read the build side's size estimate only to pick the build side. DataFusion's hash join has to hold that side in memory, so one large join failed the task while the rest of the query would have gained from the rewrite. The rule now rewrites only when the build side's estimate is under a limit and otherwise keeps the SortMergeJoinExec, which still runs natively, recording the reason as plan info. The limit is Spark's own rule for choosing a shuffled hash join, spark.sql.autoBroadcastJoinThreshold times the initial shuffle partition count, with Spark's 10 MB default threshold standing in when broadcasts are disabled. spark.comet.exec.forceShuffledHashJoin.maxBuildSize sets a fixed limit, and a non-positive value removes it, which is what the hash join benchmark configurations and the macOS benchmarking guide now pin so they keep measuring every join. * fix: restore the sort a kept sort-merge join needs after a hash join rewrite When RewriteJoin keeps a sort-merge join above a join it has already turned into a hash join on the same key, the kept join reads unsorted input. EnsureRequirements placed no sort between the two because the inner sort-merge join's ordering satisfied the outer one, and the rewrite then removed the inner join's sorts. The native sort-merge join does not check its input ordering, so matches were silently dropped. After the rewrite, add a local SortExec wherever a child no longer satisfies its parent's required ordering, the same check EnsureRequirements makes. This covers joins kept by the build-size limit and the LeftSemi and ExistenceJoin joins the rewrite already keeps, which lost rows the same way. * test: cover a window over a join the hash join rewrite reordered A window partitioned by the join key gets no sort of its own, because the sort-merge join's output ordering already satisfies it. Once RewriteJoin turns that join into a hash join, the window needs the ordering restored, or the native window fails its partition-ordering assertion. The test runs with AQE on and off and checks the join became a hash join.
…#4746) * fix: honor Spark’s commit protocol in Spark 3.x native writes * ci: register legacy native write commit protocol tests * refactor: drop the native Parquet writer's work_dir file naming Both native writers now take the exact file name from Spark's commit protocol, so nothing sets work_dir, job_id or task_attempt_id any more. Remove the Some(work_dir) arm from ParquetWriterExec::execute, reserve the three proto fields, and correct the comments that still described the Spark 3.x writer naming its own files. * fix: handle empty inputs in the Spark 3.x native writer Mirror CometWriteFilesExec and FileFormatWriter: swap a zero-partition child for a single empty partition so the output still carries a schema (SPARK-23271), and only write a file from partition 0 or a partition that has rows. Move createTaskContext to the companion object so the task closure no longer captures the exec node and its child subtree. The SPARK-23271 and empty-partition tests now run on Spark 3.x too. * fix: match Spark 3.x write admission and cover empty scans * docs: say native writes over an empty relation stay native With the empty-relation guard gone from CometDataWritingCommand, the user guide and the CometWriteFilesExec comment still described Parquet writes over an empty relation falling back to Spark's writer. Both native write paths now swap a zero-partition input for one empty partition, so such a write stays native and still leaves a readable schema-only file. * fix: take the Spark 3.x native write codec from the job configuration The Spark 3.x task named its file after the codec in the job configuration, because getFileExtension reads it from there. The native writer used the codec that convert resolved at planning time. The two agree today only because both copy the precedence rule of ParquetOptions. Read the codec with CodecConfig.from(taskContext) next to setOutputPath, as CometWriteFilesExec.executeTask does, so the file name and the footer agree by construction. A new CometNativeWriteSuite test sets a job codec that differs from the planned one and checks the name and the footer. The mixed-case codec test now checks the file name on Spark 3.x too. * test: make the injected cleanup failure throw from SQLMetric.add too Since apache#6416, CometMetricNode reports a sum metric with add instead of set. The test metric threw only from set, so the two cleanup-failure tests in CometNativeWriteSuite stopped injecting their failure and failed on Spark 3.4 and 3.5. * ci: pre-compile Spark SQL test classes in the manual writer workflow The Parquet writer test step caps the sbt heap at 1 GB for the forked test JVM, but this workflow compiles Spark first, and sbt ran out of heap while it compiled sql/core. Compile the SQL test classes in an earlier step with a 3 GB heap. This branch had the same step in July, before a history rewrite dropped it. * test: build CometNativeWriteExec with the merged constructor The merge with main brings in two callers written against the old constructor. apache#5957's RevertNativeForTransitionHeavyStagesSuite relied on the removed committer and jobTrackerID defaults, and apache#6247's write_reserving_from helper still passed the removed work_dir, job_id and task_attempt_id arguments to ParquetWriterExec::try_new. CleanupFailingNativeWriteExec also needs apache#5957's originalPlan argument. --------- Co-authored-by: Andy Grove <agrove@apache.org>
… the task's entry (apache#6403) * fix: retry a shuffle page allocation that Spark failed after dropping the task's entry Spark's ExecutionMemoryPool registers a task's memoryForTask entry once, before its wait loop, and reads it on every pass, while releaseMemory removes the entry when the task's balance reaches zero. When the JVM shuffle allocator's page or pointer array request was parked below the task's minimum share and Comet's native memory consumer released the task's last bytes, the request woke to NoSuchElementException "key not found: <taskAttemptId>" and failed the task, because the shuffle writers only expect SparkOutOfMemoryError. CometUnifiedShuffleMemoryAllocator now retries allocate and allocateArray when Spark throws that exception from ExecutionMemoryPool. The retry registers the task again and waits as the first attempt would have. Nothing is counted or allocated before the exception, so retrying cannot leak. After three attempts it refuses the request with the same SparkOutOfMemoryError it uses for a refused page, with the last exception as its cause. The Spark issue is SPARK-59827. * docs: say which callers the page allocation retry covers Only the shuffle allocator's callers are guarded; Spark's own operators in the task can still hit the removed entry until Spark re-registers a waiting task (apache/spark#59103). * docs: point the shuffle allocator retry at SPARK-59444 apache/spark#59103 (SPARK-59827) was closed as a duplicate of apache/spark#58747 (SPARK-59444), which fixes the same race in ExecutionMemoryPool, so link that issue and PR instead.
Co-authored-by: Andy Grove <agrove@apache.org>
* docs: expand the FAQ and refresh the Gluten comparison * docs: correct the shuffle manager and PyArrow UDF answers in the FAQ When spark.shuffle.manager does not name Comet's shuffle manager and spark.comet.shuffle.enabled is true (the default), Comet disables itself instead of only running shuffles in Spark. Say so, and name the setting that keeps Comet enabled. State that the mapInArrow and mapInPandas acceleration needs Spark 4.0 or later, and name spark.comet.exec.pyarrowUDF.enabled.
…he operands (apache#6691) * perf: compare floats in Spark's order without normalizing copies of the operands Since apache#6447 every native comparison of FLOAT or DOUBLE operands follows Spark's SQL ordering, but it got there by normalizing each operand that is not a literal into a copy and running Arrow's kernels on the copies. For arrays and structs, <=>, <, <=, > and >= deep-copied both operands in every batch. A new SparkComparison expression compares in place instead. Flat floats go through branch-free kernels over the value buffers, column against column or against a scalar, for =, <>, <, <=, >, >=, <=> and IS DISTINCT FROM. Lists and structs use spark_comparator for the ordering operators and spark_equality for the null-safe ones. The nested comparisons, nested = and <> included, now compare only the rows where neither side is null, iterating the validity bitmap, rather than testing both null buffers on every row. Planning is unchanged: a literal is still normalized while the plan is built, a float column compared with a literal in a scan's data filters is still a plain BinaryExpr so that Parquet pruning recognizes it, and operands of other types, such as dictionary-encoded floats, keep the normalize-then-compare path. * fix: split float comparison bitmaps with as_chunks for clippy 1.99 CI's rust-test job lints with the current stable toolchain, Rust 1.99.0, whose clippy adds chunks_exact_to_as_chunks. It rejected the three chunks_exact(64) calls in pack and pack_pairs under -D warnings, so the job failed before any test ran. as_chunks::<64>() yields the same 64-value chunks as arrays, which also drops the try_into().unwrap() conversions. * perf: keep CASE WHEN eager when a branch compares floats `is_infallible` in `case_when.rs` accepted the `BinaryExpr` of normalized operands that a float comparison used to be, but not the `SparkComparison` that replaced it. A CASE WHEN or IF with a float comparison in a later WHEN or in a branch value therefore fell back from eager to lazy evaluation. `SparkComparison::is_infallible` now reports a comparison of two operands of the same type as unable to fail, which covers every flat Float32 or Float64 comparison. The nested comparators reject a dictionary-encoded leaf against a plain one, which the logical type check in `spark_comparison` lets through, so a nested comparison counts only when the types are equal. The `is_infallible` test builds comparisons through `spark_comparison`, so a change in what the planner builds shows up there. The contributor guide describes comparing in place and lists the float comparison kernels.
apache#6809) Co-authored-by: Isaac <no-reply@databricks.com>
apache#6715) * perf: run Spark's null guard for a Scala UDF inside the codegen kernel Spark's HandleNullInputsForUDF rule wraps a Scala UDF that has a primitive parameter over a nullable input as `if (isnull(a) or ...) null else f(knownnotnull(a), ...)`. The serde sent only the UDF to the codegen dispatcher and ran the If natively, as a CASE that filters the batch for each branch and merges the results, which cost more than the call it protects. CometIf now recognizes that exact shape, including the `false` form that ReplaceNullWithFalseInPredicate leaves in predicates, and sends the whole If to the dispatcher, so the null check is a branch in the kernel's loop. When the dispatcher cannot take the whole guard, the If is converted natively as before and no fallback reason is recorded against it. * fix: keep a null guard native when an expression in it is disabled Converting the guard natively checks spark.comet.expression.<name>.enabled on the UDF and on every node of the predicate, and a disabled one takes the projection to Spark. Dispatching the guard whole skipped those checks, so with ScalaUDF, IsNull, Or, or an expression in a guarded argument disabled, the projection stayed in Comet. emitNullGuardDispatch now declines the guard in that case, and the native conversion records the same fallback reason as before. * test: bind the dispatched null guard in the Scala UDF dispatcher benchmark CometScalaUDFDispatchBenchmark, added by apache#6709, said that Spark's null check around a UDF with a primitive parameter runs natively around the dispatched call, and its JVM layer cases bound the ScalaUDF alone as the subtree the serde would dispatch. The check now runs in the kernel with the call. The scaladoc says so, and the JVM layer cases bind the projected expression, which for a primitive parameter is the check together with the call. * perf: check a repeated argument once when matching a UDF's null guard BooleanSimplification drops a repeated isnull from the predicate that HandleNullInputsForUDF builds, so with a column at two primitive positions, f(a, a) optimizes to if (isnull(a)) null else f(knownnotnull(a), knownnotnull(a)). The predicate rebuilt for the match kept every repeat, so those guards stayed native. It is now rebuilt from an ExpressionSet of the guarded arguments. ExpressionSet keeps every nondeterministic argument, so a guard over one still stays native. The first null guard test also covers addBoth(a, a), addThree(a, b, a) and addBoth(a, 1L), the last with an argument that cannot be null.
…el (apache#6354) * fix: truncate per-row date_trunc formats with the literal-format kernel * fix: pass the SECOND/MILLISECOND overflow policy through the format-column path * fix: handle a NULL format and a literal timestamp with a format column in date_trunc
…-point mode (apache#6735) * feat: run nullable nested float sort keys natively in strict floating-point mode Strict floating-point mode declined a sort or window key that nests floats in an array or struct whose type can hold a null element or field, because the native sort placed such a null by the key's null order (apache#6476) and a RANGE frame ordered it above every other value (apache#6477). apache#6683 and apache#6684 now decline exactly those shapes in every mode: ASC NULLS LAST or DESC NULLS FIRST on such a key, and a RANGE frame that has to find a row's peers over one. What strict mode still declined beyond them, the default null orders, TopK, ranks, rank limits, ROWS frames and RANGE frames bounded only by the partition, already matches Spark, so the strict-only decline goes. The strict fixture now runs those shapes natively under strict mode, with a null element, a -0.0 and a NaN with the sign bit set among the keys, and keeps expect_fallback for the shapes the general guards decline. CometWindowExecSuite expects a native rank limit in strict mode, and the nested strict sort test in CometExpressionSuite adds a key that can hold a null. The floating-point compatibility page, the operator tuning page and the spark.comet.exec.strictFloatingPoint doc no longer describe a strict exception. * test: expect native strict sort over nullable nested float keys * test: sort a real null element in the strict nested float sort test The IF key in the strict nested sort test never produced a null, because id runs from 0 to 4. It now nulls row 4's element, so the test checks that a null element sorts first in strict mode, ahead of the signed zeros and NaN payloads, with that row's bits check skipped. Also update two comments that still described the strict-only decline: nested_float_order_keys.sql and CometWindowGroupLimitExec now say that strict mode runs these keys natively, and that CometSortOrder declines ASC NULLS LAST and DESC NULLS FIRST on them in every mode.
* perf: project cached batches by buffer selection, prune on collated strings Follow-up to apache#5051, applying items from apache#5487. Replace the per-column Arrow IPC stream layout of `CometCachedBatch` with a single encapsulated IPC record batch message per cached batch, carrying no Schema message and no end-of-stream marker. The reader rebuilds the schema from the cached relation's attributes, so a wide relation no longer repeats the same schema bytes once per cached batch. Compression moves from a whole-payload Spark codec to Arrow's per-buffer IPC compression. That is what makes projection cheap: the message metadata records every buffer's offset and length in the body, so `CachedBatchIpc.readProjected` copies out only the byte ranges of the columns a scan selected and decompresses just those. This subsumes the separate "drop the schema message" item, since there is no longer a per-column stream to frame. Dictionary-encoded columns are decoded before being stored: a payload with no schema message cannot describe a dictionary encoding. The codec defaults to zstd, and lz4 is deliberately not offered. Arrow's lz4 is commons-compress's pure-Java implementation, unrelated to the JNI-accelerated lz4-java behind `spark.io.compression.codec`. Over a 200k-row six-column relation it measured 205s to write against 347ms for zstd, while also producing larger output, so no workload prefers it. zstd also beats storing batches uncompressed on both axes (347ms and 2 MiB against 1743ms and 13 MiB), because the bytes it saves cost more to copy and store than compressing them costs. Decompression is done here rather than left to `VectorLoader`, which leaks: `VectorLoader.loadBuffers` collects a field's decompressed buffers into a local list and releases them only after the whole field loads, so a buffer that fails to decompress strands every buffer of that field decompressed before it. A string column reaches this, its offsets buffer decompressing before its data buffer throws. Also track statistics bounds for collated string columns, comparing with the collation's own ordering through a new `CometTypeShim.compareStrings`. Matching the bare `StringType` object excluded collated columns, which then got null bounds and no pruning. Benchmark over a 5M-row six-column relation, keeping the cached scan native against falling back to a Spark cache scan and converting: 1.3x on a repeated scan, 1.3x on a narrow projection and 2.3x on a full projection. * refactor: use Spark's interpreted ordering for bounds, hoist projection layout Cleanup pass over the cache format change. No behaviour change. Drop the `compareStrings` shim in favour of `TypeUtils.getInterpretedOrdering`. That method is public with the same signature on every supported Spark version, and on Spark 4 it resolves a `StringType` through `CollationFactory.fetchCollation(collationId).comparator` -- the comparison the shim was reaching for. So the collation awareness comes from Spark itself and the shim, its Spark 3.x stub and the hand-rolled per-type `compare` all go. The ordering is now resolved once per column per partition rather than being re-dispatched on the `DataType` twice per row. Build the projection's index layout once per partition instead of per batch. The node, buffer and variadic index arithmetic is a pure function of the cached schema and the selected columns, but it walks every field of the relation, so recomputing it per batch made the bookkeeping O(total columns) against O(selected columns) of useful work -- worst in the wide-relation, narrow-projection case the format exists for. `CachedBatchIpc.Projection` now holds that layout and the projected schema, and owns the whole decode; `ProjectedBatch` is left with ownership only. This also puts the projected schema next to the code that packs buffers in the same order, an invariant that previously spanned two files unstated. Smaller cleanups: use Arrow's `DataSizeRoundingUtil.roundUpTo8Multiple` rather than open-coding IPC body alignment; size the serialization buffer from the record batch's known body length instead of growing from 32 bytes; resolve decompressors once instead of per batch; share the dictionary lookup guard between `Utils.combineDictionaryProviders` and the cache writer; read the codec config through one helper carrying the driver-vs-executor rationale; and collapse the duplicated compressed-buffer predicate and scramble loop in the test helper. Corrects two `Utils` scaladocs that still described the per-column stream format this change replaced. Benchmark and codec figures in the docs re-measured against the current code. * fix: relocate the arrow-compression service file when shading arrow-compression ships META-INF/services/org.apache.arrow.vector.compression.CompressionCodec$Factory. The shade plugin copies it verbatim without a ServicesResourceTransformer, so the jar declared a provider for Spark's own unshaded Arrow interface while naming a class that exists here only under the relocated package. Every ServiceLoader lookup Spark's Arrow made then failed with a ServiceConfigurationError, which took CompressionCodec.Factory's static initializer down with it and broke unrelated Arrow IPC reads, including mapInArrow. Add ServicesResourceTransformer so the service file name and its contents are both relocated. arrow-compression is the only bundled artifact that ships one. Also drop an unused NonFatal import that scalafix flagged. * test: drop the cache leak test that depends on zstd corruption detection "releases its vectors when a column fails part way through" zeroed the last 16 bytes of a compressed buffer and required the read to fail. Whether that fails is a property of the zstd runtime, not of Comet: the cached payload is byte-identical across Spark versions, but Comet takes zstd-jni from Spark rather than from arrow-compression, and 1.5.5 (Spark 3.4, 3.5) decodes that frame while 1.5.7 (Spark 4.x) reports it corrupt. So the test passed on 4.x and failed on 3.4 and 3.5. The scenario it claimed to cover is also unreachable: CachedBatchIpc decompresses every selected buffer before VectorLoader runs, so no content corruption can fail part way through the load. The two remaining leak tests corrupt a frame from its header onwards, which every zstd release rejects, and already cover a failure at a column's first buffer and a failure after an earlier buffer of the same column decoded. Records the constraint on scramble so a future test does not reach for a tail-only corruption again, and drops the now unused truncateColumn helper and the dictionary fixture's payload argument. * feat: enable Comet's in-memory cache by default Flip spark.comet.exec.inMemoryCache.enabled to true so cached tables are stored and scanned in Comet's Arrow format without an opt-in. CometDriverPlugin.maybeSetCacheSerializer read the config out of SparkConf with a hardcoded false default, so flipping the ConfigEntry alone would have left the serializer uninstalled unless the user set the key explicitly. It now falls back to the entry's own default, matching how the plugin reads spark.comet.metrics.enabled. Stacked on apache#5543. * test: cover nested columns in the cached-batch projection tests and benchmark Addresses review feedback asking whether nested data should be tested and benchmarked. Nested columns were already round-tripped, but only under a full projection, which cannot see the part of the format that is nontrivial for them. A flat column always owns one field node and two or three buffers; a nested one owns a run as long as its subtree, and selecting every column covers the whole sequence however it is partitioned. So the buffer-span arithmetic was only exercised in the one shape where getting it wrong does not show. Adds two tests over a six-column relation whose middle four columns are a struct, an array, a map and a struct wrapping an array: - Each column takes its turn as the sole projection while the other five are corrupted, so a run computed short or long is caught by reaching into a corrupted neighbour. - Values are compared against the uncached query across single-column, paired and out-of-order projections. Row counts cannot catch a window that is misaligned but still decompresses, and out-of-order is the case a full projection cannot stand in for. The per-column statistics test now runs over the nested relation too, since a nested column's recorded size is the sum of its whole subtree. Both new tests fail if fieldNodeCount stops recursing into children. In the benchmark, adds the three projection widths over a relation of struct columns, and asserts the width each case claims. That assertion caught the existing "full projection (6 of 6 columns)" case reading three: count() over a non-nullable column is rewritten to count(1) by NullPropagation, which prunes the column out of the scan, and only k, s1 and s2 were nullable -- and those only incidentally, because Remainder can divide by zero. Every column of both relations is now nullable so count(c) genuinely reads c, and the documented numbers are regenerated. Array and map columns are left out of the benchmark deliberately: the baseline arm needs Spark's cache scan to bridge into Comet operators, and CometSparkToColumnarExec declines ArrayType and MapType, so for those the arm does not exist and the two cases stop measuring the same boundary. The docs say so rather than leaving it to be rediscovered. * fix: drop a redundant string interpolator flagged by scalafix RedundantSyntax * review: check the cached layout, and address the rest of the review Reader-side: a cached payload carries no schema, so `Projection` derived every node and buffer window from `Utils.toArrowSchema(cacheAttributes)` with nothing checking the writer had produced that layout. `load` now compares `nodesLength()`/`buffersLength()` against the totals `selectedRange` already computes, before any unchecked `batch.buffers(j)`. Writer-side: `isArrowBacked` accepts a `FixedSizeBinaryVector` for a `BinaryType` column, which is two buffers where the reader rebuilds three, and it answers for the top-level vector only -- so a struct of large strings passes it and is stored with 64-bit offsets. `matchesReaderLayout` compares the batch's Arrow types against the reader's recursively, and a batch that disagrees takes the conversion path instead. A dictionary column's field carries the index type, so the dictionary's field is what is compared. Also: an unrecognized body-compression byte is rejected rather than read as plain bytes, `fieldVariadicCount` and the variadic plumbing are gone (the length check covers view vectors, which the counts would not have), `columnSizes` no longer re-walks each column's subtree, the write codec is a case class rather than a bare tuple, the per-partition `Projection` is lazy so a row-count-only read never builds it, `hydrateDictionaries` is `decodeDictionaries`, `Projection` takes an `IndexedSeq`, and the stale `readProjected` links and some over-long comments are fixed. Tests: the two projection tests become one parameterized over both relations, caching once and restoring the payload between columns instead of re-caching; the two leak tests become one with two corruption points. New tests cover the reader's layout check and the writer declining a fixed-size-binary batch. * review: own the write-side compression buffers, fix the activation example Compressing through VectorUnloader leaks on the failure path: appendNodes retains each input buffer and accumulates the compressed ones into a list local to getRecordBatch, so a buffer that fails to compress strands that retain and leaves every buffer compressed before it reachable from nothing. Closing the input batch afterwards undoes neither. Unload plain and compress in CachedBatchIpc.compressed instead, mirroring what decompressed already does on the read side, so every allocation stays reachable from an error path that owns it. The docs enabled the cache with spark.conf.set, which cannot work: the driver plugin picks spark.sql.cache.serializer while the SparkContext is initializing. Show it as a startup --conf. Also drops a redundant s interpolator that the scalafix lint rejected. * fix: write cached batches to the schema width, not the batch width ArrowWriter.writeColumns drove its loop from the input ColumnarBatch's width while indexing the writer's fields, which come from the schema the batch is written under. That assumed every producer hands over a batch exactly as wide as the schema. Iceberg's vectorized reader does not. BatchDeleteFilter.filterBatch reads with the delete filter's requiredSchema, which carries _pos after the projected columns when a data file has position deletes, and trims the extras back only when the file also has equality deletes. A merge-on-read UPDATE writes position deletes and no equality deletes, so the extra column survives into the batch, and caching such a relation failed with ArrayIndexOutOfBoundsException inside the write loop. Drive the loop from the writer's fields instead, which writes exactly the columns the schema describes: the extras are trailing, the same prefix Iceberg keeps when it does trim. A batch narrower than the schema is a genuine contract violation and is now refused with a message naming both widths. Closes apache#6087. * fix: report a cached relation's decoded size to the planner, not its compressed size Comet's cache format reported each batch's sizeInBytes as its stored, compressed payload. Spark's planner reads the sum of those as the size of a materialized cached relation, for the broadcast threshold and the shuffled hash join build side among others, and both of Spark's own cache formats report decoded sizes there. With the default zstd codec a cached relation could look several times smaller than in Spark's format and be broadcast where Spark would shuffle it. Record each column's decoded Arrow size, measured before compression as Spark's ArrowCachedBatchSerializer does, and let CometCachedBatch inherit sizeInBytes from SimpleMetricsCachedBatch as the sum of those. * fix: keep Spark's cache scan for a relation whose cached plan records observed metrics Spark collects Dataset.observe metrics after a query with CollectMetricsExec.collect, which reaches the metrics recorded inside a cached plan only through an InMemoryTableScanExec over it. With Comet's native cache scan in its place those metrics were lost: QueryExecution.observedMetrics came back empty, Observation.get returned an empty map on Spark 3.5 and later, and on Spark 3.4 it never returned. Keep Spark's scan for such a relation, with a fallback reason. The data stays in Comet's format and is read through the existing fallback path. The check walks the cached plan the way CollectMetricsExec.collect does, through subqueries, adaptive plans and nested caches. * test: install Comet's cache serializer in the Spark SQL test sessions The Spark SQL diffs turn Comet on through SharedSparkSession and TestHive rather than by loading CometPlugin, and the plugin is what installs Comet's cache serializer. So even with spark.comet.exec.inMemoryCache.enabled on by default, every Spark SQL suite cached in Spark's own format and none of them exercised Comet's. Set spark.sql.cache.serializer in both places when Comet is enabled, as the plugin would, and adapt the tests that assume Spark's cache scan: - plan checks that only need a cache scan to be present also accept CometInMemoryTableScanExec, reading its originalPlan where the test uses the scan's relation; - PartitionBatchPruningSuite checks its answers as before, but reads the test-only accumulators only from Spark's scan, which Comet's lacks; - CacheTableInKryoSuite registers Comet's classes with CometKryoRegistrator, as Comet asks of any application that sets spark.kryo.registrationRequired; - tests of Spark internals that Comet's scan replaces are tagged IgnoreComet: exact cache size estimates, the union's columnar support, subquery reuse through the scan's predicates, and AQE coalescing under a union (apache#6454). Each diff was regenerated from a clone of its Spark tag with the existing diff applied, after checking that it round-tripped unchanged. * fix: coalesce shuffle partitions under Comet unions the way Spark does Spark's CoalesceShufflePartitions coalesces each child of a UnionExec as its own group, and from Spark 4.0 each child of a CartesianProductExec, BroadcastHashJoinExec or BroadcastNestedLoopJoinExec too. It matches those classes, so a CometUnionExec fell through to the case that coalesces only when every leaf below it is an exchange stage. A union with a scan or a table-cache stage in one branch kept every partition of the shuffles in the others. Add CometCoalesceShufflePartitions, a query-stage optimizer rule that runs after Spark's. For a Comet operator whose Spark original is one of those, and whose shuffle stages no AQE rule has touched, it rebuilds the Spark operators over the Comet children, runs Spark's own rule over that, and swaps the Comet operators back in. It extends AQEShuffleReadRule, so AQE gates and validates it the way it does Spark's rule. Spark 3.4 has no hook for query-stage optimizer rules, so the fix applies from Spark 3.5. Closes apache#6454. * test: fix three Spark SQL test adaptations for Comet's cache format - CachedBatchSerializerNoUnwrapSuite sets its own cache serializer, but Spark keeps the first one it loads for the rest of the JVM, so the suite ran with Comet's. Clear it before and after the suite, as CachedBatchSerializerSuite does (4.1.3, 4.2.0). - The SPARK-19993 helper counted the subqueries in a Comet cache scan's original plan, which subquery reuse never reaches, on top of the same subqueries in the filter above the scan. - SPARK-35332's AQE check reads the cached plan from the SQL UI's plan tree, which shows it only under Spark's own cache scan (apache#6463). Skip it under Comet (4.0.4 and later). * fix: keep Spark's cache format when Kryo would reject Comet's or Comet disables itself CometDriverPlugin installed ArrowCachedBatchSerializer whenever Comet, its native execution and the cache config were enabled at startup. Two more startup settings decide whether that format can work: - With spark.kryo.registrationRequired=true and no CometKryoRegistrator, Kryo rejects CometCachedBatch, so caching failed with "Class is not registered" as soon as Spark serialized a cached block, where Spark's own format works. The plugin only warned. It now keeps Spark's format, and the warning says so. - With Comet shuffle enabled but neither of Comet's shuffle managers configured, Comet disables itself, so every cache was stored in Comet's format with only Spark operators to read it. The plugin now keeps Spark's format there too. The plugin's boolean config reads also honor a deprecated alternative key, such as spark.comet.exec.shuffle.enabled, as a session does. * docs: describe the in-memory cache default in the upgrade guide Add a 1.2.0 section to the upgrade guide for the new default of spark.comet.exec.inMemoryCache.enabled: what changes, how to keep Spark's cache format, and when Comet keeps it without being asked. The operator and compatibility pages still called the feature disabled by default, and the cache guide still said an application that started with the default kept Spark's format. * test: benchmark the in-memory cache against Spark's format with AQE on The published comparisons either read the same Comet-written cache both ways, with spark.comet.sparkToColumnar.enabled turned on, or read both formats with Comet off. Neither is what an application gets from the feature's default. Add cases with Comet and AQE on and Comet's other settings at their defaults, reading a relation cached in Spark's format and in Comet's, with Comet operators above the cache scan and with a Spark operator above it. * feat: record why Spark scans a relation cached in Comet's format when Comet is off spark.sql.cache.serializer is static, so a relation cached in Comet's format keeps it for the life of the application. A session that turns Comet or its native execution off then reads it with Spark's InMemoryTableScanExec, which is slower than reading Spark's own format (apache#5485). CometExecRule records a fallback reason for the other ways Spark ends up scanning a cache, but it returned before looking at the plan in these two cases, so nothing explained them. Record a reason on each such scan, naming the cause and the startup setting that keeps caches in Spark's format. * docs: say an application keeps the cache format it started with The sentence described a session started with the default, which keeps Spark's format only while the default is off. * test: format the in-memory cache benchmark * fix: bind the fallback reason result for the strict Scala warnings build -Ywarn-value-discard rejects the node that withFallbackReason returns, discarded as the last expression of a case. * docs: compare the in-memory cache with Spark's format as an application gets it Add the benchmark's adaptive results to the cache guide: with Comet and AQE on and Comet's other settings at their defaults, Comet's format is as fast or faster than Spark's in every shape but a read of three long columns, which zstd decompression makes about 10% slower. That includes a Spark operator above the native scan. The guide said that, without the feature, a CometSparkColumnarToColumnar converts each batch of Spark's cache scan for Comet. That takes spark.comet.sparkToColumnar.enabled, which is off by default, so by default the operators above Spark's scan run on Spark. The limitation that Spark operators read Comet's format slowly applies to Spark's own cache scan, not to Spark operators above Comet's. * fix: count Kryo registrations however the application made them The gate kept Spark's cache format whenever spark.kryo.registrator did not list CometKryoRegistrator, even where the application had registered Comet's cached batch another way, such as spark.kryo.classesToRegister. Comet's format works there, while Spark's does not before 4.1, which registers DefaultCachedBatch itself only from then on, so the gate broke caches that worked. Ask a Kryo instance built from the application's conf which of Comet's classes it has registered, and keep Spark's format only if Comet's cached batch is not among them. If Kryo cannot be built, fall back to looking for CometKryoRegistrator in spark.kryo.registrator. The startup warning uses the same check and names the missing classes. * docs: describe the Kryo condition by registration in the upgrade guide The plugin now keeps Spark's cache format when Kryo has not registered Comet's cached batch, however the application registers classes, rather than when spark.kryo.registrator does not list CometKryoRegistrator. * docs: name spark.comet.convert.inMemoryCache.enabled for converting Spark's cache scan spark.comet.sparkToColumnar.enabled is deprecated for in-memory cached tables since apache#6602, which gave the conversion its own config. * test: accept Comet's cache scan in BroadcastJoinSuite's cached-relation tests Since apache#6415, suites that build their own SparkSession run Comet, and with the cache on by default BroadcastJoinSuite's cached relations are read by CometInMemoryTableScanExec. SPARK-23214 and SPARK-37742 count Spark's cache scan and broadcast hash join, so count Comet's as well. * fix: coalesce with the whole stage in view and keep partitionings above Comet unions Address review of the CometCoalesceShufflePartitions rule: - Stand in only for Comet operators whose output partitioning is unknown. From Spark 4.1 a union of co-partitioned children reports their partitioning, and an aggregate above reads it without a shuffle. Coalescing one branch made Comet's aggregate, which states no required distribution, return each key twice. - Hand Spark's rule the whole stage rather than the subtree below each Comet operator, so it sees a Cartesian product or nested loop join above a union and uses its smaller target size there, and divides the minimum partition count over every group. Leave a stage that already has a read over a shuffle, which Spark's rule would coalesce again. - Copy the Spark original when withNewChildren hands it back, so a union that AQE plans again over its stages is coalesced too. - Drop the CartesianProductExec case, which has no Comet counterpart, and document the rule in CometSparkSessionExtensions. * test: run Spark's SPARK-42101 union coalescing test with Comet again The test failed with Comet's cache format because AQE did not coalesce the shuffled branch of a union that Comet ran (apache#6454), which apache#6459 fixes. * fix: coalesce Comet union branches in stages that already hold an AQE shuffle read CometCoalesceShufflePartitions returned a stage unchanged when it held any AQEShuffleReadExec, because Spark's rule asserts on reads that are not skew splits. Below a Cartesian product, Spark's rule coalesces the shuffle on the other side by itself first, so the read it leaves there kept the shuffled branch of a Comet union at every partition. Hide each read behind a leaf while Spark's rule runs, and put it back afterwards. Spark's rule then leaves alone the shuffles it would coalesce together with a read one, and coalesces the rest of the stage. * docs: say Spark's cache scan can read Comet's format more slowly The fused reader from apache#5859 reads Comet's format about as fast as Spark's own, and faster for the narrowest reads, so the upgrade note should not say Spark's scan is always slower. * test: leave Comet's cache scan out of the DPP suite's subquery check Since apache#6577, CometInMemoryTableScanExec exposes Spark's own InMemoryTableScanExec as its one subquery, so that Spark's UI draws the cached plan below it. With the cache on by default, the DPP suite's 'filtering ratio policy fallback' caches its dimension table, and checkPartitionPruningPredicate requires every subquery of an adaptive plan to contain an AdaptiveSparkPlanExec, which Spark's scan does not. Skip it there in every diff, since the query never runs it as a subquery. * feat: keep Comet's cache format opt-in on Spark 3.4 Spark 3.4 has no hook for query-stage optimizer rules, so CometCoalesceShufflePartitions cannot run there, and AQE cannot coalesce the shuffled branch of a Comet union whose other branch reads a cache in Comet's format (apache#6454). spark.comet.exec.inMemoryCache.enabled now defaults to true only from Spark 3.5, and the plugin, which follows the entry's default, keeps Spark's format on 3.4 unless it is set. The 3.4 Spark SQL diff goes back to main's: its sessions install Comet's serializer only as the plugin would, which on 3.4 it no longer does by default, so its cache test adaptations are not needed there. * docs: rewrap the Spark 3.4 cache default sentences * test: run Spark's SPARK-35332 AQE cache test with Comet again The diffs skipped CachedTableSuite's 'SPARK-35332: Make cache plan disable configs configurable - check AQE' on Spark 4.0 and later, because SparkPlanInfo drew the cached plan only below Spark's own InMemoryTableScanExec. Since apache#6577 it is drawn below CometInMemoryTableScanExec too, so the test finds the cached plan. The test's check that the cached plan was coalesced, which the diffs already adapt, looked for AQEShuffleRead only below a columnar-to-row transition, where it is when Comet's plan is cached in Spark's format. In Comet's columnar format the cached plan ends in the read itself, as in Spark, so the check accepts it there too. The test passes on Spark 4.0, 4.1 and 4.2 with either format, and with Comet off. * docs: stop describing the in-memory cache as experimental It is enabled by default from Spark 3.5. The operator table now marks InMemoryTableScanExec as supported, and the plan-node table no longer says the cache scan is disabled by default. * test: skip only the format-dependent checks in the Spark SQL diffs SPARK-36120, SPARK-33687, SPARK-22673 and SPARK-43376 now run with Comet's cache format and skip only the size or subquery-reuse assertions that depend on it. The union test in CoalesceShufflePartitionsSuite runs its coalescing check again on Spark 3.5 and later, where CometCoalesceShufflePartitions coalesces the shuffles on the union's join side. From Spark 3.5, CometTestSettings also makes Comet's cache serializer the default of the sql and hive test JVMs, next to the shuffle manager. Spark keeps the first cache serializer it loads for the rest of the JVM, so a suite that caches first through a session of its own left the suites after it in Spark's format. * test: time building the cache in Spark's format in the cache benchmark runBuildBenchmark caches the same relation in Spark's format and in Comet's, with each codec, from Comet operators and from Spark operators, and reports the footprint of each. The cache guide has the results. * test: skip only the columnar checks of SPARK-37371 under Comet Comet runs the union of two cached relations as a CometUnion over its own cache scans, so the checks on Spark's scans and union are skipped there. The union with a local relation stays Spark's non-columnar UnionExec, and both answers are checked, under Comet as well. --------- Co-authored-by: test <a@b.c>
apache#6599) * feat: run array_contains on float elements natively with Spark's equality A Comet kernel compares float elements as Spark's genEqual does, with -0.0 equal to 0.0 and all NaNs equal, at any depth, so these calls no longer go through the codegen dispatcher. Closes apache#6520 * fix: keep nested floats and eagerly evaluated values on the dispatcher Route only flat FLOAT and DOUBLE arrays to spark_array_contains. Nested float elements, and a value other than a literal or column read over a nullable array, stay on the codegen dispatcher, which is faster for nested elements, keeps string collations, and skips the value for a null array as Spark does. * test: import EDGE_VALUES from float_semantics in array_contains tests Main moved the constant there, so the crate's unit tests failed to compile. --------- Co-authored-by: Andy Grove <agrove@apache.org>
…ceberg writes (apache#6808) apache#6441 makes a native Iceberg write to S3 fall back when the Hadoop configuration has an fs.s3a.* setting the native writer doesn't support. Spark 4.1 sets fs.s3a.committer.magic.enabled and fs.s3a.committer.name for every application when spark-hadoop-cloud is on the classpath (SPARK-47618), and many deployments set fs.s3a.experimental.input.fadvise, fs.s3a.bulk.delete.page.size or fs.s3a.committer.threads cluster-wide. Any of them makes every native S3 write fall back. None of these settings affects an Iceberg data-file write: the native writer doesn't go through S3A, fadvise is a read hint, and Iceberg commits through table metadata rather than a Hadoop output committer. Ignore them like the Spark-seeded read settings already in IgnoredHadoopS3Keys.
…Spark 3.x native writer (apache#6821) * fix: run InsertIntoHadoopFsRelationCommand's post-write steps in the Spark 3.x native writer On Spark 3.x CometNativeWriteExec replaces the whole InsertIntoHadoopFsRelationCommand, but only ran the job/task commit lifecycle. After the job commits it now refreshes the table's file index, recaches cached plans that read the output path, and updates or clears the table stats, as the command does. It also rejects duplicate output column names up front, which the command checks before writing. Co-authored-by: Isaac <no-reply@databricks.com> * test: assert the duplicate-column write is native Spark's own writer rejects the same columns, so a fallback would keep the test green without exercising the native writer's check. Capture the plan of the failed write as well, and assert it carried a native write. Co-authored-by: Isaac <no-reply@databricks.com> --------- Co-authored-by: Isaac <no-reply@databricks.com>
…#6695) * refactor: move the plan entry point logic into core functions Separate the native logic of createPlan, setShufflePartitionPusher, executePlan and releasePlan from their JNI argument handling: - create_execution_context builds the execution context from plain values (PlanInputs) and a memory pool factory. prepare_plan_settings parses the Spark configs and initializes the Tokio runtime. - register_shuffle_partition_pusher checks the plan state before building the pusher. - execute_plan returns the plan's next batch; the export still exports it. - release_plan stops and drops the plan. These plans call back into the JVM, so the JVM objects stay JNI global references in the execution context (PlanJvmRefs), and the steps that need an Env are passed in as callbacks: the memory pool factory, the pusher factory, and PublishMetrics. ExecutionContext::metrics becomes an Option, which is always set by createPlan, so a plan can be created and run without a JVM in unit tests. Exception classes and messages are unchanged. * Test that release_plan returns the memory of a plan holding a reservation The lifecycle test ran a RangeScan, which never reserves memory, so its check that nothing stays reserved after release_plan could not fail. Run a descending sort over the range instead, release it after its first batch while it still holds its buffered input, and check that the plan holds a reservation before the release and none after. Do the same when the final metrics publish fails. Draining a plan to the end stays covered by a separate test.
* refactor: remove unused cast compatibility flag * refactor: drop unused parquet allow_incompat option The flag has no remaining reader after the cast-option cleanup, and the unused CometConf import fails scalafix. * fix: preserve DPP filters when reverting native scans * Fix SparkCastOptions constructor calls * refactor: drop the DPP scan reversion from the cast cleanup * fix: drop the removed cast-option argument from case-when calls * fix: drop the removed parquet-option argument from the page-index test * fix: drop the removed cast-option argument from the cast_nested bench apache#6648 added this bench after the last merge of main into this branch, and it still passed the third argument that this PR removes from SparkCastOptions::new. --------- Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: Andy Grove <agrove@apache.org>
…'s entry (apache#6310) Spark removes a task's execution pool entry when its balance reaches zero, and an acquire that was parked in the pool then throws NoSuchElementException when it wakes up. Releases take no task monitor, so another native thread or a JVM consumer of the same task can empty the balance while a native acquire waits. The error is not a memory refusal, so the operator could not spill and the task failed. CometTaskMemoryManager now recognises that one exception and calls Spark again. A missing entry means the task held nothing from Spark, so the failed call has no grant to reconcile, and the retry registers the task again and waits for its share as the first call would have. After three attempts it returns a zero grant, which the native side treats as a refusal and spills. Closes apache#6224.
…conversion (apache#6824) * Builder appending refactoring * Formatting, small refactoring and adding docstrings * Added docstring regarding alignment * Fixed the fallback in an unaligned but NULLABLE case, that was failing the CI * Changes fixed * Fixed cargo formatting * Remade the building process using arrow buffers with no hot-spots in Vec<> loops * Formatting fix * Added docstring about runtime check of alignment * Addressed the review comments * Fixed formatting * Added docstring regarding endianess * test: cover a timestamp array with a timezone in the append round trip The round-trip test built every builder with its type's default data type, so the timestamp case never had a timezone. append_array panics when the array's data type, timezone included, differs from the builder's. The test now builds the builder with the column's data type and adds a timestamp column with a timezone, which fails if the bitmap path drops the timezone. * perf: append a nullable array with nulls in bulk only from 64 elements Building the validity and value buffers takes four allocations per array, which cost as much as appending 40 to 55 elements one by one, so the bulk path made lists of shorter arrays with a null up to 4.8 times slower than the per-element loop. MIN_BULK_NULLABLE_APPEND_ELEMENTS keeps those arrays on the loop. No array length is slower than on main now, and arrays from 64 elements keep the speedup. - The timestamp appender takes the timezone by reference, so the list path doesn't clone it for every row. - process_primitive_lists is main's single arm again, with an optional argument list. - The SAFETY comments are back, and the alignment comment points at SparkUnsafeObject. - The partial-byte, all-null and all-valid tests run at lengths on both sides of the cutoff, from an 8-byte-aligned copy. The round trip covers 63 and 64 elements with nulls, and its "no nulls and long enough" case now has 8 elements. A new test appends several arrays into one builder, so a validity buffer lands at an offset that isn't a whole byte. * test: cover the list path's timezone and make the bulk cutoff measurable in-tree - timestamp_list_column_keeps_its_timezone drives a Timestamp(Microsecond, Some("UTC")) list column through append_columns, so append_list_column_batch must forward the timezone. With &None in its place, append_array's data type assertion fails. CometColumnarShuffleSuite gets an array<timestamp> case with 70-element arrays that hold a null, which fails the same way end to end. - The impl_append_to_builder doc states that `timezone` must be the builder's, and when a mismatch panics. The has_null() comment covers the validity buffer path too. - array_element_append gets a list_with_one_null group: 8192 list rows of 32, 64 and 128 elements with one null each, for i32 and i64, appended as the list path appends them. Its doc comment shows how to compare the bulk path with the per-element loop, and MIN_BULK_NULLABLE_APPEND_ELEMENTS cites it instead of an out-of-tree harness. * test: read the bench's array words with as_chunks Clippy 1.99 flags chunks_exact with a constant chunk size (chunks_exact_to_as_chunks), so the rust-test job failed under -D warnings before any Rust test ran. The buffer's length is a multiple of 8, so as_chunks::<8>() leaves no remainder. --------- Co-authored-by: sandudb <alexandrdomenti@gmail.com> Co-authored-by: Matt Butrovich <mbutrovich@users.noreply.github.com>
…l arithmetic (apache#5216) * perf: skip DecimalPrecision.promote's rewrite when there is no decimal arithmetic QueryPlanSerde.exprToProto runs DecimalPrecision.promote over every expression it converts. The rule is a transformUp, so it walks and rebuilds every node even when nothing matches, which is the common case. Guard it with a cheap traversal that looks for arithmetic producing a decimal, and only run the rewrite when one exists. * ci: register DecimalPrecisionSuite in the PR build workflows * docs: describe transformUp's real cost in the promote guard comment
… evaluation (apache#6777) Registering a native UDF with a type Comet has no native representation for, such as a user-defined type, succeeded, and every call then fell back to Spark, which cannot evaluate it. CometNativeUDF.register now refuses such a signature before it loads the library. Spark also evaluates a call itself while planning a query: over local data, in a filter on partition columns, and to sample the keys of a global sort. The error a native UDF raises there said that Comet had not taken the operator holding the call, which sent users to the explain output for a fallback that never happened. The error, the scaladoc and the Rust UDF guide now list those cases, and the guide says to sort on a column holding a UDF's result rather than on the call. Split out of apache#6697.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
12 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
CI mirror of upstream apache/datafusion-comet#6107, addressing apache#2837. This fork PR does not close or merge the upstream issue.
Rationale for this change
Map inputs to
explodeandposexplodecurrently fall back to Spark.What changes are included in this PR?
Support maps in all four explode variants by sharing their entry buffers with the existing unnest path. Preserve field metadata, entry order and outer NULL rows. Add regression tests and update support documentation.
How are these changes tested?
Local verification passed 18 Rust tests and 52 targeted tests on each of Spark 3.5.9 and 4.1.3, covering native execution, NULL/empty maps, nested inputs, sliced arrays and batch boundaries.
Upstream CI passed at
9601e81ba, including Required Checks. The corresponding fork run is not fully green: Spark 3.4 scans faileda file without ids next to a file with ids is checked on its ownin ParquetReadSuite (544 passed, one failed in that job). Its missing-field-ID error assertion failed. The relationship to this map-explode change has not been established; do not treat the run as a complete cross-profile pass until the failure is classified and resolved.