Repository navigation
Conversation
github-actions
Bot
added
enhancement
rich7420
added
run-all-spark-profiles
rich7420
force-pushed
the
feat/5994-native-wide-decimal-hash
branch
from
September 28, 2026 14:50
e4408ce to
3b22912
Compare
rich7420
force-pushed
the
feat/5994-native-wide-decimal-hash
branch
from
October 2, 2026 17:21
27faa50 to
36902da
Compare
* 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.
…t credentials out (apache#6059) * fix: let explicit Hadoop Azure auth outrank ambient AZURE_* environment variables The native Azure store seeded its builder from the environment and object_store's fixed credential order let an ambient bearer token, account key or webhook token file outrank or hybridise the identity the job configured through fs.azure keys. With a Hadoop auth mechanism configured, or a provider type naming MsiTokenProvider, no environment variable is consulted at all, so the configured identity can neither be replaced nor redirected. Without one, the environment applies as before, keeping AKS workload identity working. * fix: validate every Hadoop Azure auth mechanism and borrow only the token file for a named identity Reject Hadoop configuration the native scan cannot honour instead of letting object_store fall through to the Azure CLI or managed identity: a partial OAuth principal, an explicit auth.type the translated keys do not satisfy, a provider class other than the MSI, workload identity or client credentials providers (matched by exact simple name), the untranslated SAS provider, key provider, refresh token and user name or password keys, and blank values. Translate fs.azure.sas.fixed.token below the container-scoped SAS key, take the account from the host lowercased, and when Hadoop names only an identity borrow AZURE_FEDERATED_TOKEN_FILE alone, erroring when no token file exists. Every error carries one prefix and names the key, never a credential value. * fix: read only the Hadoop keys of the auth mechanism the account selects Hadoop's `AbfsConfiguration.getAuthType` resolves the account-scoped `fs.azure.account.auth.type` over the global one, and `AzureBlobFileSystemStore.initializeClient` then reads only that mechanism's keys. The native scan translated and validated every credential key instead, so an account that selects `SharedKey` while a global OAuth secret remains in the configuration was rejected for missing the OAuth client id and tenant. An explicit auth type now decides which keys are read: only the selected mechanism's keys are translated into the builder and validated, and the other mechanisms' keys, provider class and unsupported-mechanism keys are ignored as Hadoop ignores them, blank or not. Incomplete active credentials are still rejected, and no auth type at all still reads every key. * fix: apply Hadoop's default SharedKey when an account key is set without an auth type `AbfsConfiguration.getAuthType` falls back to `SharedKey` when no `fs.azure.account.auth.type` is set, and `initializeClient` then reads only the account key. The native scan read every mechanism's keys in that case, so a valid account key beside an unused OAuth client secret was rejected for the missing client id and tenant. An account key entry without an auth type now selects `SharedKey`: the OAuth and SAS keys beside it are neither translated nor validated, a blank key is still reported, `fs.azure.account.keyprovider` is still rejected, and the environment contributes nothing. With neither an auth type nor a key, every mechanism is read as before, so OAuth-only and environment-only configurations keep working. * fix: keep the Azure account as written and accept blank MSI client id and tenant Hadoop reads the account from the raw URL authority and appends it to each account-scoped key as written, so a lowercased account missed the key Hadoop finds and fell through to the environment. extract_account now keeps the host label as is. Under MsiTokenProvider Hadoop accepts an empty client id and tenant, which is how a system-assigned identity is configured. Those two keys are treated as absent for that provider only, so the builder proceeds to the managed identity endpoint with no client id. A blank client id under any other provider is still an error naming the key. * fix: read only the OAuth keys the selected Azure provider reads A shared Hadoop configuration can carry OAuth settings for several providers. Hadoop reads only the keys of the provider class it selects, but the native scan validated and translated every fs.azure.account.oauth2 key. A complete client credentials setup was rejected over an unused password or a blank MSI endpoint, and a leftover tenant, secret or token file could change the identity the store used. The provider's key list is now resolved once and both validation and translation read only those keys. With no provider class the rules are as before. An explicit SimpleKeyProvider, Hadoop's own default, is accepted with an account key and rejected without one. Other key providers are still rejected, and the class name must match exactly because Hadoop loads it untrimmed. * fix: follow Hadoop's defaults for blank Azure keys and prefer the fixed SAS token Under MsiTokenProvider, Hadoop reads fs.azure.account.oauth2.msi.endpoint and msi.authority through getTrimmedPasswordString(key, default), so a blank value means the IMDS endpoint and https://login.microsoftonline.com/. The native scan rejected the blank key while the driver read fine. WorkloadIdentityTokenProvider defaults msi.authority and token.file the same way. Blank values of those keys now count as unset under the class that defaults them, and stay errors without a provider class. The token file falls to AZURE_FEDERATED_TOKEN_FILE as an unset one does, and no ambient endpoint or authority host stands in for the blank key. ABFS reads fs.azure.sas.fixed.token and never the container-scoped fs.azure.sas.<container>.<account> key, which belongs to WASB, so a job that set both listed files with one token and read them natively with the other. The fixed token now comes first and the container-scoped key is the fallback. Whichever key is selected must be non-blank. The tenant id section of the Azure docs now says the provider class decides which of msi.tenant and client.endpoint is read. * fix: resolve Azure SAS settings the way Hadoop ABFS does A global fs.azure.sas.token.provider.type now counts only when the global auth type (SharedKey when unset) matches the account's, as in Hadoop's getTokenProviderClass. A configuration with a global OAuth auth type, an account-scoped SAS auth type and an unused global provider class was rejected even though Hadoop accepts it. The SAS token lookup now probes the container-scoped fixed token fs.azure.sas.fixed.token.<container>.<host> first, under the URL's own host with any explicit port, as Hadoop 3.4.2 and later do. Then come the account-scoped and global fixed tokens and the WASB container key. A blank container-scoped token is an error. The user guide notes that older Hadoop does not read that key and that the native scan reads container-scoped forms only for the SAS fixed token. * fix: trim the Azure SAS token before handing it to object_store Hadoop reads the fixed token with getTrimmedPasswordString and drops a leading '?'. The native scan checked the trimmed value for blankness but passed the original string on, so a padded token such as ' ?sig=...' reached object_store with the '?' after the whitespace and the parsed credential had a '?sig' key. Trim whichever SAS key was selected before translating it. * fix: read the container-scoped SAS token only when the driver's Hadoop does Hadoop 3.4.2 added a container-scoped lookup to AbfsConfiguration (containerConf), so fs.azure.sas.fixed.token.<container>.<host> is read before the account-level token. Hadoop 3.4.1 and older never read it, and Spark 3.4, 3.5 and 4.0 ship those versions. Reading it unconditionally broke or changed scans that work there: a blank container key next to a valid global token failed the scan, and a different value replaced the token Hadoop uses. NativeConfig now checks whether the hadoop-azure on the classpath has containerConf and passes the answer to native code as comet.azure.containerScopedSasToken in the object store options for abfs and abfss. The native scan reads and validates the container-scoped key only when it is true; otherwise it follows Hadoop 3.4.1 (account-scoped fixed token, then global) with the WASB key as its own fallback. When the key is present but not read, ambient AZURE_* credentials stay out, and if it is the only credential the scan fails with an error naming the key instead of building a store that would fall back to managed identity. Hadoop fails on that configuration too. * fix: send Azure client-credential token requests to the configured endpoint host Hadoop's ClientCredsTokenProvider posts to the full fs.azure.account.oauth2.client.endpoint. The native store took only the tenant from it, so object_store built the token URL on login.microsoftonline.com and a configured authentication proxy or sovereign authority was bypassed. When the tenant comes from the endpoint, a client secret is set and no token file is, the endpoint's origin is now translated to object_store's authority host as well. Without a secret, or with a token file (which object_store prefers for workload identity), the endpoint is not a token URL and its host receives no other credential. * fix: keep a client endpoint's path prefix in the Azure token URL An endpoint mounted under a path, such as https://auth-proxy.example/aad/<tenant>/oauth2/v2.0/token, gave its first segment as the tenant and only its origin as the authority host, so object_store requested /aad/oauth2/v2.0/token and dropped the tenant. The tenant is now the segment just before oauth2, and the authority host is the origin plus every segment ahead of it, so object_store requests the configured URL. Endpoints without a later oauth2 segment keep the first segment as the tenant, and standard and v1 endpoints are unchanged. * fix: take the Azure token endpoint's tenant from before its last oauth2 segment An endpoint whose path prefix itself contains oauth2, such as /gateway/oauth2/<tenant>/oauth2/v2.0/token, gave gateway as the tenant because the first oauth2 segment decided the split, so object_store dropped the real tenant from the token URL. Use the last oauth2 segment, which is the one in the token suffix; endpoints without a later oauth2 segment still take the first segment as the tenant. * fix: read Azure account-scoped keys under the URL host alone, as Hadoop does AbfsConfiguration.accountConf builds one scoped key, <key>.<host>, from the URL authority as written, port included. The native scan also probed the other endpoint suffix and the bare account name, and once the lookup chose the auth mechanism those extra forms changed the answer: a SharedKey auth type scoped to the blob host overrode a global OAuth for a dfs URL, and an override scoped to host:443 was dropped. Read only Hadoop's form, then the global key, for every account-scoped setting, the container SAS keys and the WASB fallback included. * fix: read Azure credential keys in Hadoop's container-scoped form Hadoop 3.4.2 reads the OAuth keys and the SAS fixed token through getPasswordString, which probes <key>.<container>.<host> before the account-scoped and global forms. The native scan read that form only for the SAS fixed token, so a client endpoint set per container was missed once the provider class decided which keys are read, and a configuration that works on Hadoop failed. Every key Hadoop reads that way now takes the container form first when the driver's hadoop-azure has containerConf, under the probe the SAS token already used. The account key and the mechanism keys stay account-then-global, as in Hadoop. On older Hadoop a container-scoped credential key that is the only credential is an error naming the key rather than a fall-through to the environment, unless a workload identity is named. * feat: resolve Azure ABFS authentication through Hadoop on the driver The native scan reimplemented the ABFS driver's fs.azure.* lookup rules, and every review round found another place where the two disagreed: scoping by container, account or host, the auth-type guard for provider classes, trimming, blank values, and what each hadoop-azure release changed. NativeConfig now asks the classpath's AbfsConfiguration the questions AzureBlobFileSystemStore.initializeClient asks, through its public getters, and forwards the answers: comet.azure.* markers for the outcome, auth type, provider class and any error, plus the resolved values under the global key names. Mechanisms the native scan cannot run are declined by name, never instantiated. Any reflection failure or Hadoop exception becomes an error marker; nothing falls through. Only the selected account's resolved values travel in the plan, instead of every fs.azure.* key. hadoop-azure is a test dependency pinned to each profile's Hadoop line so the resolver runs against 3.3.4, 3.4.1, 3.4.2 and 3.5.0 in the suite. * refactor: build the native Azure store from the driver's resolved authentication The native scan no longer reads fs.azure.* settings by Hadoop's lookup rules. The driver resolves the mechanism and its values through the ABFS driver and forwards them with comet.azure.* markers, so this side only maps them: none reads the environment, resolved applies exactly the keys the mechanism owns, declined and error fail with the driver's verdict, and a missing or unknown marker fails as well. The scoping, capability probe, unread-key guard, provider-class filtering, blank checks and environment policy that reimplemented Hadoop are gone. * docs: describe Azure authentication as resolved by Hadoop on the driver The Azure section now explains that the driver asks the ABFS driver which mechanism and values apply, lists which mechanisms the native scan can build and what happens for the rest, states when the environment is read, and records the changes from 1.1. The account scoped lookup order, the endpoint suffix limitation and the WASB container key, which no longer apply, are gone. * test: pin the Azure resolver against hadoop-azure and the real FileSystem AbfsAuthParitySuite runs 137 configurations through the resolver and compares each with an expectation recorded for the hadoop-azure version on the classpath (3.3.4, 3.4.1, 3.4.2 or 3.5.0, chosen by capability probes), with hand-written expectations for the cases where versions differ or Hadoop's rules surprise. Every case also goes through FileSystem.newInstance offline and the outcome class and root cause must agree. Record mode writes the expectation file for the running version and refuses a partial run. Writing the cross-check found one gap: SimpleKeyProvider reads the account key through a configuration without a container, so a key set as fs.azure.account.key.null.<account> initializes the FileSystem while the resolver reported no configuration and would have sent the native scan to the environment. The resolver now probes the key the same way. * fix: let unused unreadable ABFS keys pass the configuration check Deciding whether an account has ABFS authentication configured read every auth key through Hadoop. A key the selected mechanism never uses, such as an OAuth secret with a cyclic ${...} reference next to a valid SharedKey, threw there and failed the scan although hadoop-azure initializes fine. A key whose read throws inside Hadoop now counts as configured. The chosen mechanism reads only its own keys, so a broken key it needs still fails in Hadoop's words. * fix: fall back for ABFS scans that span containers or accounts The native scan resolved ABFS authentication for the first file's container and account and applied it to every file, so a scan over two accounts with different keys sent the first account's key to the second. Matching credentials are not enough either: the native scan builds one object store per file partition from that partition's first file and keeps only each file's path, while Spark packs files from different roots into one partition. A scan whose ABFS files span more than one container or account, or that mixes abfs/abfss with another scheme, now falls back to Spark for both the Parquet and the CSV native scans. The reason is also tagged on the Spark scan that stays in the plan, and a declined CSV scan now falls back to that scan instead of keeping the Comet wrapper. AbfsScanFallbackSuite reads two fake accounts and two containers end to end through a local abfss filesystem. AbfsAuthParitySuite now creates java.io.tmpdir before using it, since a fresh checkout may not have it. * perf: stop reading ABFS credentials once the account is known to be configured Deciding whether an account has ABFS authentication read every auth key through Hadoop's getPassword, only so a later failure message could be redacted against all of them. With a credential provider configured, each read builds and loads the keystore, so a plain SharedKey scan made 48 lookups where Hadoop makes 3, about half a second of driver planning per scan. The check now stops at the first key that configures the account, reading the plain keys first. The remaining credentials are read only when resolution fails, before the error message and the driver log are redacted. Also make the multi-account fallback tests pass the strict Scala warnings job.
…me (apache#6754) * fix: fall back from native V2 scans when the plan reads input_file_name The native Iceberg and CSV V2 scans do not set InputFileBlockHolder, so input_file_name, input_file_block_start and input_file_block_length returned empty values. Fall back to Spark's reader as the V1 path does. Closes apache#6707 * refactor: reuse readsInputFileBlock in the V2 scan check apache#6703 added the shared helper for the same predicate. * test: cover the native CSV V2 scan's input_file_name fallback Document the fallback in the CSV section of datasources.md. * test: move the Iceberg input_file_name test to a SQL fixture Match the full fallback reason there and in the CSV test.
…apache#6841) * fix: accept all BMP Unicode decimal digits in cast(string as decimal) Spark casts a string to decimal with `new java.math.BigDecimal(s.trim)`, which accepts any char for which `Character.isDigit(char)` is true, in both the mantissa and the exponent. Comet only normalized fullwidth digits, so strings such as '٣' (U+0663 ARABIC-INDIC DIGIT THREE) or '1e३' returned NULL, or raised CAST_INVALID_INPUT under ANSI, where Spark returns a value. Normalize every non-ASCII Nd digit in the Basic Multilingual Plane to ASCII before parsing. Supplementary-plane digits stay rejected, since BigDecimal sees them as surrogate pairs. The pure-ASCII fast path is unchanged. Integral, floating-point and date casts are left alone: Spark only accepts ASCII digits there. * perf: keep ASCII and fullwidth digits off the table lookup Normalizing every char through the Nd table slowed down already supported fullwidth and mixed ASCII/fullwidth input. Scan bytes instead, copying ASCII directly and mapping the fullwidth UTF-8 encoding in place as before, and only decode a char for the table lookup otherwise. The output buffer is preallocated, since it is never longer than the input. Add fullwidth, mixed ASCII/fullwidth and Arabic-Indic cases to the cast_from_string benchmark. * fix: drop removed SparkCastOptions argument in the new decimal benchmark apache#6135 removed the third argument of SparkCastOptions::new on main.
…apache#6827) * fix: preserve CASE WHEN struct field names from the first THEN branch * test: keep CASE WHEN tests inline for focused review * test: cover COALESCE struct field names * test: cover nested and null CASE WHEN struct branches
… memory (apache#6867) * fix: keep one row per key in a native PartialMerge that runs short of memory Comet ran Spark's PartialMerge aggregate as DataFusion Partial, which emits the groups it holds when it runs short of memory and starts again. In Spark's single-distinct plan the stage above the PartialMerge counts its rows, so count(DISTINCT) counted every copy of a key. Run a PartialMerge as DataFusion Single instead, which spills. MergeAsPartial evaluates to a struct of its state, and a projection unpacks it into the state columns that the Partial plan emitted. SpillReplayPool also covers SingleHashAggregateStream, whose spill replay starves the same way as the final aggregates' do (apache#6254). Closes apache#6857. * Emit PartialMerge states a column at a time for row-only aggregates MergeAsPartial now wraps an inner aggregate that has no GroupsAccumulator (first, last and any_value without an ordering) in a GroupsAccumulatorAdapter itself. Under Single mode, DataFusion otherwise wrapped the wrapper's row accumulators and evaluated every group to a one-row struct. Also write the planner's two output wrappers the same way, check the planner test's output with assert_batches_sorted_eq and cover first there, and drop two settings that the memory-pressure tests don't read.
…ezone (apache#6868) Native plans must label every TimestampType value Timestamp(µs, "UTC") and every TimestampNTZType value with no timezone. The label never changes the values, so checkSparkAnswer cannot see a wrong one, and the casts at ScanExec and the shuffle writer's SchemaAlignExec repair it silently. Add the internal test-only config spark.comet.testing.strictTimestampLabels.enabled, which CometTestBase turns on. With it, CometExecIterator checks each batch a native plan returns against the plan's Spark output types, ScanExec fails instead of casting a relabelled input timestamp, and the native shuffle writer fails planning instead of aligning one.
…6005) * fix: route wide decimal hash shuffle keys through Spark-compatible hashing * test: retain safe aggregate buffers after wide decimal shuffle fallback * Preserve one-partition decimal shuffle and cover mixed joins * docs: fix native shuffle table formatting * test: avoid implicit widening in wide decimal benchmark * refactor: drop the wide-decimal shuffle guards that this rule makes unreachable apache#6564 and apache#6607 added narrower versions of the same restriction: the typed Dataset check in nativeShuffleFailureReasons and the decimal half of hashesDifferentlyFromSpark. supportedHashPartitioningDataType now declines every multi-partition hash key with a decimal wider than 18 digits, so neither can fire. The typed Dataset check only ran when no other reason had been found, and convertsInputForNativeShuffle checks nativeShuffleFailureReasons before hashesDifferentlyFromSpark. The tests those PRs added in CometTypedDatasetSuite and CometShuffleInputConversionSuite assert the routing and still pass. * test: cover a decimal join with a Spark-bucketed input A table bucketed on a DECIMAL(38, 0) key needs no exchange in a join, so only the other input is shuffled, and its partitions have to match the buckets Spark wrote. Without the hash key restriction, native shuffle put the rows in other partitions and the join returned 128 of 1000 rows, with both scans native and no Spark shuffle in the plan. DECIMAL(18, 0) stays on native shuffle and matches. --------- Co-authored-by: Andy Grove <agrove@apache.org>
apache#6814) * fix: preserve empty Iceberg partitions after dynamic pruning * test: simplify empty Iceberg scan credential regression * test: cover single-split Iceberg joins after DPP
Part of apache#6627. Preserve Java regex semantics and dispatcher routing while replacing 18 Scala tests, including multi-batch extraction coverage.
…orders NaN as Spark does (apache#6866) * refactor: stop normalizing float divisors now that percentile_approx orders NaN as Spark does CometExecRule wrapped the divisor of a FLOAT or DOUBLE Divide or Remainder in a Project or Filter in NormalizeNaNAndZero. Comparisons don't need it, since the native comparison builder follows Spark's ordering, including the check for a zero divisor. It stayed only because it also canonicalized a NaN quotient: 1.0D / (-d) of a NaN d is a NaN with the sign bit set, and percentile_approx ordered that NaN below every other value (apache#6519). apache#6829 fixed percentile_approx, so the rule can go. nan_divisor.sql keeps its percentile_approx query, which now checks the sign-bit NaN quotient directly, along with the hash and extrema queries for the earlier consumers. * test: regenerate TPC-DS plans without the divisor normalization Without NormalizeNaNAndZero around float divisors, the 11 approved plans that hold a float division in a Project or Filter count one native expression fewer, and q78 no longer lists knownfloatingpointnormalized and normalizenanandzero among its dispatched expressions. Regenerated on Spark 3.4, 3.5, 4.0, 4.1 and 4.2; 4.1 and 4.2 need no override.
Part of apache#6634. Preserve scalar input types, hash child order and dictionary configurations while replacing 18 Scala tests with SQL fixtures.
* test: add SQL error-class parity checks * docs: clarify SQL fixture locations
…ache#6382) * fix: restore localReads assertion in AdaptiveQueryExecSuite diffs * fix: restore localReads assertion in Spark 3.4.3 diff
…titioning (apache#6268) * fix: do not resolve DPP subqueries when AQE reads a native scan's partitioning AQE validates a stage by reading outputPartitioning after CoalesceShufflePartitions changes it, which is before Comet converts the DPP placeholders. CometNativeScanExec answered with its post-pruning split count, which ran the SubqueryAdaptiveBroadcastExec placeholder and threw. A non-bucketed scan now reports UnknownPartitioning(0), as FileSourceScanExec does, and CometNativeExec sizes the native RDD from perPartitionData at execution. Part of apache#6133. * fix: do not resolve DPP subqueries when AQE reads an Iceberg scan's partitioning CometIcebergNativeScanExec counted its tasks in outputPartitioning, which runs BatchScanExec.inputRDD and translateRuntimeFilterV2 on a DPP subquery that has no result yet while AQE validates a stage. It now reports UnknownPartitioning(0), as BatchScanExec does on Spark 3.5 and later, and CometNativeExec sizes the native RDD from the scan's tasks at execution. * docs: say a plan-data scan's outputPartitioning must not run its DPP subqueries AQE reads outputPartitioning while it optimizes a stage, before the adaptive DPP placeholders are converted, so a scan that computes it from perPartitionData runs a placeholder subquery. Say so in the CometScanWithPlanData doc, and point a scan that cannot report a real partitioning otherwise at UnknownPartitioning(0), as CometNativeScanExec does for a non-bucketed scan. * test: count a native scan's partitions through perPartitionData A non-bucketed CometNativeScanExec now reports UnknownPartitioning(0), as FileSourceScanExec does, so the coalesced-scan metrics test cannot read the split count from outputPartitioning. perPartitionData holds the count the native RDD is sized from, which is what the precondition meant to check.
* docs: prefer behavior-named parameters over Spark versions When behavior varies between Spark versions, resolve the version in Scala and pass native code a parameter that describes the behavior. Add the guidance to the contributor guide and the expression skills. * docs: address review on behavior-named parameters Reconcile the review skill's shim paragraph with the new rule: the version check may live in a shim or a named predicate in Scala, and protobuf and native code see only the behavior. Extend the rule to native function names, point to RegrSparkVersions for fixes that land in patch releases, ask comments to cite the Spark JIRA, state the fork benefit as it is today, and replace the removed isSpark33Plus helper. The two existing version-named names are tracked in apache#6891.
* test: run the S3 suites against an available MinIO image Docker Hub no longer serves minio/minio, so every suite built on CometS3TestBase aborted in beforeAll. Use cgr.dev/chainguard/minio, declared as a substitute for minio/minio so MinIOContainer accepts it. The suites also relied on an AWS region from the environment. Give the bucket setup client an explicit region, and add client.region to the REST catalog vended properties, since Iceberg's JVM S3FileIO reads that key rather than s3.region. * test: pin the MinIO test image by digest Chainguard rebuilds the latest tag from its MinIO fork, and newer releases tighten request checks, so a rebuild could break the S3 suites with no change on our side. Pin the multi-arch index digest of RELEASE.2026-09-22T19-25-18Z so every run uses the same server.
…ntegration (apache#6888) Only comet-spark is useful to a consumer. Its published pom is flattened and shade-reduced, so it has no <parent> and no comet-common dependency, and comet-common's six classes are inside the shaded jar. The parent pom is installed as the raw root pom.xml with unresolved ${...} placeholders in its artifactId, and comet-spark-integration only checks the jar. The nightly snapshot workflow passed -Dmaven.deploy.skip=false, so it uploaded all four modules for each of the five variants. Release staging uploads whatever `install` leaves in its temporary repository, which includes comet-common and the parent pom. Skip deploying every module except comet-spark, drop the override from the snapshot workflow, and have publish-to-maven.sh upload only comet-spark-spark* directories. Installing is unchanged: builds run from inside spark/ (make benchmark-*, the TPC-H and TPC-DS data generation in pr_build_linux.yml) resolve comet-common from the local repository.
…native Iceberg writer (apache#6773) * feat: close fanout partitions early when the memory pool refuses the native Iceberg writer A native fanout write keeps a file open for every partition a task writes to, and its buffers count against the task's memory pool, so a task over many partitions could fail where iceberg-java's on-heap writer succeeded. When the pool refuses the writer's reservation, it now writes out and closes the partitions holding the most memory, in their open file or in rows not yet handed to it, until what is left fits. A closed partition's next rows open a new file with the same writer properties, so the write finishes with more, smaller files instead of failing. A files_closed_early metric counts them. Writes that keep one file open still fail when that file outgrows the pool. * docs: describe how a native fanout write closes partitions early under memory pressure * test: compare the early-closing fanout write with its source rows The control write is native too, so it could lose the same rows. Also fix a stale FanoutWriter mention in the user guide and a missing noun in the FanoutFiles doc comment. * refactor: keep each fanout partition's feed, writer and memory in one record FanoutPartitions holds every partition in first-seen order, each with its feed, its open writer, the OpenFileMemory child its files report to, and the properties its files are written with. That replaces FanoutFiles and FanoutFeeds and their join by key, the seen counter, PartitionProperties::get and the partition field threaded through the metered builder. InnerWriter::reserve now owns the close-until-it-fits loop, ranking partitions by index without cloning keys and tracking pending bytes without rescanning. WriteMetrics gets a constructor, the Rust memory tests share their setup, and the Scala test captures the write with captureWrite. * docs: say the pool cannot fail a fanout write, and fix a stale comment Everything a fanout write reserves belongs to one of its partitions, so once every partition holding memory is closed the resize is down to nothing, which the pool cannot refuse. The contributor guide said a fanout write could still fail with nothing left to close; only an unpartitioned or clustered write can. FanoutPartition's key is used to open every file of the partition, not only to write held rows out at close. * test: check that a fanout write short of memory closes the partitions holding the most No test pinned the order in which a write the pool refuses closes its partitions: closing every partition at once, or closing them smallest first or in the order they were first seen, passed every test. The new test has 16 or 64 partitions take turns through 32 batches in an eighth of the memory the write needs, and asserts fewer than six files per partition. Closing the largest first writes between four and five, closing all at once seven, and the other two orders seventeen. The user guide now says how many files to expect, and that Iceberg's rewrite_data_files compacts them. * test: give the close-order test's count for 64 partitions too Closing partitions smallest first or in first-seen order writes 17 files per partition with 16 partitions and 20 with 64; the comment only gave the first. * refactor: drop the expect from a fanout partition's write and check its counters in debug builds - `FanoutPartition::write` takes the writer it opens from `Option::insert`, so it no longer needs an `expect`. - `OpenFileMemory::child` asserts that it is called on the task's total, since a share counts only one level up. - `FanoutPartitions::write` asserts that `held_bytes` is still the sum of what the feeds hold back, and the field's doc names the places that keep it so. - `by_memory` says why it ranks partitions by index. - `payload_rows_from` takes its regions by value instead of copying them. * test: check the files_closed_early count in the fanout early-close tests The Rust tests passed `WriteMetrics::default()` and never read it back, so a miscount in `reserve`'s return went unnoticed. `write_reserving_from` now takes the metrics, and `fanout_write_in_part_of_the_pool` gives its tight write a real `WriteMetrics` and checks `closed_early <= files <= closed_early + partitions`: no file rolls at the target size, so every file was closed either early or when the task ended, at most one per partition. Counting one file fewer or one more per refusal fails `a_fanout_write_short_of_memory_closes_the_partitions_holding_the_most`. * fix: pass on a resize error that is not the pool refusing the fanout writer `reserve` treated any error from `try_resize` as the pool refusing, and closed partitions until the resize fit. A pool backed by Spark can also fail a resize with an exception Spark threw, such as a spill of another consumer that failed or the `InterruptedException` of a task killed while it waited for memory. That comes back as `External` (a Java throwable) or `Execution` (any other Comet error), while every pool reports a refusal as `ResourcesExhausted`. Once enough partitions had closed, the resize was a shrink that never called Spark, so the write carried on and the exception was lost. Both `try_resize` calls now retry only on `ResourcesExhausted` and pass anything else back, which fails the task. The new test's pool refuses twice, so two partitions close early, then fails the third resize with an `External` or an `Execution` error. It asserts that the write returns that error and deletes every file it created, the two closed early included. The contributor guide now says a refusal from the pool cannot fail a fanout write, rather than the pool, and the comment on `reserve` no longer says an open file holds at most a row group, which is wrong on S3 and GCS. * docs: say a fanout partition stays in FanoutPartitions, not open, until the task ends The comment on `held_bytes` still said every partition stays open, which stopped being true once partitions could close early. What makes the held-back rows share one limit is that every partition, and its feed, stays in `partitions`.
…or map is NULL (apache#6716) * fix: skip an array or map function's later arguments where the array or map is NULL Spark's array_contains, array_position, array_remove, arrays_overlap, array_union, slice, m[k] and a[i] are null-intolerant BinaryExpressions or TernaryExpressions, which return NULL for a NULL array or map without evaluating the other arguments. The native functions evaluated every argument over the whole batch, so an argument that raises, such as an ANSI cast of a malformed string, failed a query that Spark completes, and a nondeterministic argument saw rows Spark never gives it. Add a native NullShortCircuit wrapper that evaluates the first argument over the whole batch and each later one only for the rows where no earlier argument is NULL, as ArrayInsert already does for its own arguments. A null_short_circuit flag on ScalarFunc and ListExtract enables it, set through QueryPlanSerde.withNullShortCircuit. The planner leaves a function unwrapped when every later argument can be evaluated for any row. Closes apache#6613. * Simplify NullShortCircuit and skip masking for infallible arguments Track the skipped rows with NullBuffer::union and collect an argument's columns with collect_columns. Keep each argument's batch field on the argument itself instead of in a parallel list. Evaluate a later argument that can be evaluated for any row over the whole batch, as literals already were, and build a projection only for the arguments that are masked. Leave a column whose index does not change unrewritten. Point the guide and the proto comment at withNullShortCircuit instead of repeating its conditions. * Evaluate a shared subexpression for every row, as Spark's generated code does Spark's subexpression elimination evaluates a subexpression that an operator's expressions share for every row, before the expressions that use it, in projections and aggregations. A shared ANSI cast therefore raises in Spark even on rows whose array is NULL. Before serializing an operator, ask Spark's EquivalentExpressions which subexpressions its expressions share, and leave unmasked any call whose later argument holds one, so it raises there as well. * Evaluate a masked argument through one copy, so shuffle keeps its state A masked argument was evaluated through its rebuilt copy, which reads its columns from a batch of just those columns, only for a batch with a NULL in an earlier argument, and through the original for every other batch. Rebuilding an argument whose column indices change gives it new state, so array_union(a, shuffle(b)) restarted shuffle's random sequence each time evaluation moved between the two, and drew different permutations from Spark's. The rebuilt copy now evaluates a masked argument for every batch. A unit test evaluates array_union(a, shuffle(b)) over a batch without a NULL array, one with, and one without again, against the unwrapped function, which draws the same permutations as Spark when b is NULL wherever a is. A Spark 4.0+ SQL fixture reads 8200 rows from one file, so that only the scan's second batch has NULL arrays. Both fail with the previous evaluation. * fix: count LEGACY and TRY string-to-integer casts as infallible, and find shared subexpressions lazily Cast::is_infallible now accepts a cast from Utf8 or LargeUtf8 to an integer in LEGACY or TRY mode, which returns NULL for a string it cannot parse. Comet's parsers for those modes never raise. With ANSI off, the native planner then leaves array_contains(ids, CAST(s AS INT)) unwrapped, and CASE keeps its eager path for such a cast. withSharedSubexpressions now runs EquivalentExpressions on the operator's first canShortCircuitNulls call instead of for every converted operator. canShortCircuitNulls skips it entirely when every later argument is a leaf, which subexpression elimination never shares. array_null_short_circuit_non_ansi.sql runs the array shapes with ANSI off. The casts return NULL there, while a slice start of 0 and a rand argument still depend on the short-circuit. The shuffle fixture pins spark.comet.batchSize, and the map fixture covers map_from_arrays, which its CASE WHEN guards already protect. * perf: keep CASE lazy for string-parsing casts, and give NullShortCircuit its own check Counting a LEGACY or TRY string-to-integer cast as infallible also put CASE and IF on their eager path, which parses every row for each branch that some row chooses. With a selective predicate, that made IF about 8x slower and a three-branch CASE about 3x slower. Cast::is_infallible keeps its original set: casts that cannot fail and cost little more than reading their input. Cast::cannot_fail adds the LEGACY and TRY string-to-integer casts. case_when.rs exposes both checks. CASE uses is_infallible, so it chooses the path it chose before. NullShortCircuit uses cannot_fail, since it masks an argument only to keep it from failing on rows that Spark skips. The conditional bench gains IF and CASE shapes whose branches parse strings. * fix: find shared subexpressions where shuffle keys and TakeOrderedAndProject are serialized The native shuffle writer serializes its partitioning keys, and CometTakeOrderedAndProjectExec its sort keys and projection, when the plan runs, outside the operator conversion that sets the shared subexpressions. canShortCircuitNulls took the missing set as an empty one, so a cast that two partitioning keys share was skipped on the rows whose array is NULL, where Spark's projection evaluates it for every row and raises. canShortCircuitNulls now skips no argument outside withSharedSubexpressions. The shuffle writer calls it with the partitioning keys, which Spark evaluates together in one projection, and CometTakeOrderedAndProjectExec with its own expressions. * fix: scope a TakeOrderedAndProject's sort keys and projection apart, as Spark evaluates them 3090d85 gave CometTakeOrderedAndProjectExec's sort keys and projection one shared set. Spark's ordering evaluates each sort key on its own, without subexpression elimination, and its projection runs separately over the rows the sort keeps. So a cast in two sort keys, or in a sort key and the projection, raised in Comet where Spark returns the rows. The top-K plans now use an empty set, and the projection its projection list.
…e task has ended (apache#6813) * refactor: share the missing task entry retry between the shuffle allocator and the native pool apache#6403 added a retry for Spark's missing task entry (SPARK-59444) to CometUnifiedShuffleMemoryAllocator, and apache#6310 added its own copy to CometTaskMemoryManager.acquireMemory. Move both into MissingTaskEntryRetry under org.apache.comet. CometTaskMemoryManager passes its task id, which keeps apache#6310's exact "key not found: <id>" match; the shuffle allocator keeps apache#6403's prefix match. After the retries run out the native pool still gets a zero grant and the shuffle allocator still throws SparkOutOfMemoryError, and the warning now carries the last exception for both. The fixtures and the waiting-request harness of both suites move to TaskMemoryTestUtils, and CometTaskMemoryManagerSuite keeps one race test next to the refusal and rethrow tests. memory_management.md describes the retry once for both call sites, says what a zero grant does to try_grow and grow, and drops the stale apache#6304 reference. Co-authored-by: Isaac <no-reply@databricks.com> * fix: don't retry a missing task entry once the task has ended releasePlan drops a plan's stream when the task ends, while a Tokio worker may still wait in Spark's acquireMemory for the memory that stream holds. When the release empties the task's balance, the worker loses the task's entry. Before apache#6310 it failed at once; with the retry it registered the ended task again and, below its minimum share, waited for other tasks to release memory while holding the TaskMemoryManager monitor that the executor's cleanUpAllAllocatedMemory needs. MissingTaskEntryRetry now refuses without retrying when the task has completed or been killed: CometTaskMemoryManager returns a zero grant and the shuffle allocator throws SparkOutOfMemoryError. Both callers pass the TaskContext they were created in, since native threads have none of their own. Co-authored-by: Isaac <no-reply@databricks.com> --------- Co-authored-by: Isaac <no-reply@databricks.com>
… join test (apache#6902) With AQE on, once DPP prunes one shuffled input, AQEPropagateEmptyRelation replaces the join with an empty relation. Spark 4.0 plans that as EmptyRelationExec, which Comet converts to CometEmptyRelationExec, but Spark 3.5 has no EmptyRelation node and plans an empty LocalTableScanExec. Check for the node each version plans. Closes apache#6901.
rich7420
added
run-spark-4.1-tests
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
17 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?
Addresses apache#5994. Fork CI mirror of upstream #6548.
Rationale for this change
Spark hashes decimals with precision above 18 using the minimal signed big-endian bytes of the unscaled value. Comet currently falls back for SQL
hashandxxhash64, while native shuffle hashes wide decimal keys as fixed-width little-endian bytes. The shuffle mismatch can lose matching rows when one join input uses native shuffle and the other uses Spark-compatible shuffle.This fixes the shared native hash encoding. apache#6005 addresses the shuffle mismatch by avoiding native shuffle for wide decimal keys. This PR instead keeps native shuffle and makes its partition assignment match Spark. If apache#6005 lands first, its guard will need to be reconciled with this implementation.
What changes are included in this PR?
hashandxxhash64for wide decimals, including arrays, structs and maps. Keep wide decimals on Comet's xxhash64 kernel because the pinned DataFusion kernel uses a different encoding.How are these changes tested?
Merged upstream main
93ce5c347into published feature head4198ffeb6, preserving its history. The only manual conflict resolution combines imports inCometNativeShuffleSuite.scala; both the feature's mixed-shuffle checks and main's AQE/reuse checks remain. The decimal kernels are unchanged by this update. New merge head:bfaad41d8.Fresh local verification on the integrated tree (Apple M3 Pro, JDK 21, Spark 4.1.3):
Performance evidence was collected on feature head
27faa50ed; the subsequent main updates do not change the decimal kernels. All 36 Criterion array cases improved against the Spark-compatible implementation before the typed-list optimization: Murmur3 5.0–14.2x and xxhash64 2.1–15.8x across the three list types, 2/32 elements, sliced inputs and no/sparse/dense nulls. Paired scalar remeasurement found no significant slowdown.Two whole-query runs consumed every result through a checksum and verified Spark parity, native expression routing and no JVM codegen dispatch. With 1,048,576 Parquet rows, native array Murmur3 averaged 85/86 ms versus 159/176 ms with Spark hash fallback; native array xxhash64 averaged 89/83 ms versus 167/168 ms with fallback. These are local workload measurements.
CI is running on current head
bfaad41d8: upstream CI and fork CI. The fork retainsrun-all-spark-profiles,run-spark-4.1-testsandrun-benchmark-checkto request all Comet Spark profiles, the Spark 4.1 SQL shards and benchmark compile/lint. The same selected profile/SQL coverage passed on published parent4198ffeb6; that is historical evidence, not the current-head verdict. Keep this PR Draft until current-head CI completes.