diff --git a/docs/source/contributor-guide/release_process.md b/docs/source/contributor-guide/release_process.md index e490cbec259..842ace9dcdb 100644 --- a/docs/source/contributor-guide/release_process.md +++ b/docs/source/contributor-guide/release_process.md @@ -287,9 +287,6 @@ git tag 0.13.0-rc1 git push apache 0.13.0-rc1 ``` -Note that pushing a release candidate tag will trigger a GitHub workflow that will build a Docker image and publish -it to GitHub Container Registry at https://github.com/apache/datafusion-comet/pkgs/container/datafusion-comet - ### Publishing Documentation In `docs` directory: @@ -468,9 +465,6 @@ git tag 0.13.0 git push apache 0.13.0 ``` -Note that pushing a release tag will trigger a GitHub workflow that will build a Docker image and publish -it to GitHub Container Registry at https://github.com/apache/datafusion-comet/pkgs/container/datafusion-comet - Reply to the vote thread to close the vote and announce the release. The announcement email should include: - The release version diff --git a/docs/source/user-guide/latest/compatibility/expressions/_category_template/cast.md b/docs/source/user-guide/latest/compatibility/expressions/_category_template/cast.md index 9d44b603017..e2ace6c1219 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/_category_template/cast.md +++ b/docs/source/user-guide/latest/compatibility/expressions/_category_template/cast.md @@ -22,17 +22,23 @@ under the License. Cast operations in Comet fall into three levels of support: - **C (Compatible)**: The results match Apache Spark -- **I (Incompatible)**: The results may match Apache Spark for some inputs, but there are known issues where some inputs - will result in incorrect results or exceptions. The query stage will fall back to Spark by default. Setting - `spark.comet.expression.Cast.allowIncompatible=true` will allow all incompatible casts to run natively in Comet, but this is not - recommended for production use. -- **U (Unsupported)**: Comet does not provide a native version of this cast expression and the query stage will fall back to - Spark. +- **I (Incompatible)**: Comet's native cast may produce different results from Apache Spark for some inputs. By + default, Comet instead runs Spark's own code-generated cast inside the Comet pipeline, so the results match Spark. + Setting `spark.comet.expression.Cast.allowIncompatible=true` selects the native cast, but this is not recommended + for production use. +- **U (Unsupported)**: Comet does not provide a native version of this cast, so it runs Spark's own code-generated + cast inside the Comet pipeline. - **N/A**: Spark does not support this cast. -## ANSI Mode Fallback +Incompatible and unsupported casts fall back to Spark only when Comet's codegen dispatcher cannot handle them (for +example, casts involving `VariantType`) or when `spark.comet.exec.scalaUDF.codegen.enabled=false`. -Cast will fall back to Spark in some cases when ANSI mode is enabled. This can be enabled by setting `spark.comet.expression.Cast.allowIncompatible=true`. See the [Comet Supported Expressions Guide](../../../expressions.md) for more information on this configuration setting. +## ANSI Mode Support + +Enabling ANSI mode does not by itself make a cast fall back to Spark or require +`spark.comet.expression.Cast.allowIncompatible=true`: every cast that Spark permits in ANSI mode has the same support +level as in legacy mode. See the [ANSI Mode](#ansi-mode) table below for per-pair support, and the +[Comet Supported Expressions Guide](../../../expressions.md) for more information on the `allowIncompatible` setting. ## Whitespace Trimming in Casts from String @@ -143,20 +149,12 @@ Comet's native `CAST(date AS STRING)` is compatible with Spark. Years below 1000 zero-padded to four digits (e.g. year 999 renders as `0999-01-01`). Years above 9999 are rendered without truncation. The cast is timezone-independent. -## String to TimestampNTZ - -Comet's native `CAST(string AS TIMESTAMP_NTZ)` implementation matches Apache Spark's behavior. -Unlike `CAST(string AS TIMESTAMP)`, this cast is timezone-independent: any timezone offset in -the input string (e.g. `+08:00`, `Z`, `UTC`) is silently discarded, and the local date-time -components are preserved as-is. Time-only strings (e.g. `T12:34:56`, `12:34`) produce `NULL`. -The result is always a wall-clock timestamp with no timezone conversion or DST adjustment. - ## Decimal with Negative Scale to String Casting a `DecimalType` with a negative scale to `StringType` is marked as incompatible when `spark.sql.legacy.allowNegativeScaleOfDecimal` is `false` (the default). When that config is -disabled, Spark cannot create negative-scale decimals, so Comet falls back to avoid running -native execution on unexpected inputs. +disabled, Spark cannot create negative-scale decimals, so by default Comet runs Spark's +code-generated cast instead of the native one. When `spark.sql.legacy.allowNegativeScaleOfDecimal=true`, the cast is compatible. Comet matches Spark's behavior of using Java `BigDecimal.toString()` semantics, which produces scientific diff --git a/docs/source/user-guide/latest/compatibility/expressions/_category_template/map.md b/docs/source/user-guide/latest/compatibility/expressions/_category_template/map.md index 9368ae68d69..2acbfa88a63 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/_category_template/map.md +++ b/docs/source/user-guide/latest/compatibility/expressions/_category_template/map.md @@ -26,7 +26,7 @@ keys, in `try_element_at`, and in other contexts where map ordering must be dete runs `MapSort` natively, so map shuffle and group-by-on-map stay on Comet under Spark 4.0. When `spark.comet.exec.strictFloatingPoint=true`, `MapSort` falls back to Spark for maps whose -keys contain `Float` or `Double` (consistent with `SortOrder` and `SortArray`). Arrow's sort uses +keys contain `Float` or `Double`. Arrow's sort uses IEEE total ordering for floating-point, which differs from Spark's `Double.compare` semantics for `NaN` and `-0.0`. diff --git a/docs/source/user-guide/latest/compatibility/expressions/index.md b/docs/source/user-guide/latest/compatibility/expressions/index.md index 35746991951..8d28eee9f5d 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/index.md +++ b/docs/source/user-guide/latest/compatibility/expressions/index.md @@ -32,6 +32,9 @@ Spark 4.0 Spark 4.1 ``` +Spark 3.4 support is deprecated and will be removed in a future release; see +[Spark Version Compatibility](../spark-versions.md#spark-34). + Expressions that are not 100% Spark-compatible fall back to Spark by default, except those with a JVM codegen-dispatch path, which stay in Comet's native pipeline and match Spark exactly. Set `spark.comet.expression.EXPRNAME.allowIncompatible=true`, where `EXPRNAME` is diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/index.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/index.md index 40fe54c4eab..0550dfb0c66 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/index.md +++ b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/index.md @@ -27,6 +27,9 @@ expression class name, to run Comet's native implementation despite its differen from Spark. See the [Comet Supported Expressions Guide](../../../expressions.md) for more information on this configuration setting. +Spark 3.4 support is deprecated and will be removed in a future release; see +[Spark Version Compatibility](../../spark-versions.md#spark-34). + ```{toctree} :maxdepth: 1 diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index b1e3216161d..36cd8cf2b86 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -68,7 +68,8 @@ which document their per-expression configs and the specific differences to expe This is distinct from expressions that have **no** codegen-dispatch path: there, the incompatible cases fall back to Spark by default, and `allowIncompatible=true` runs the native -(incompatible) path instead. `cast` is the main example; see the +(incompatible) path instead. Aggregate functions such as `mode` are the main example, because the +codegen dispatcher covers only scalar expressions; see the [expression reference](../expressions.md) for which expressions have incompatible cases. ## Strings with non-UTF-8 bytes @@ -111,12 +112,10 @@ Code that catches `SparkException` and only asserts on message substrings is una inspects the exception class, `getCondition()`, or the parameterised error class will observe divergence: -- Byte / Short `Add`, `Subtract`, and `Multiply` overflow raises `ARITHMETIC_OVERFLOW` where Spark - 4.1 raises `BINARY_ARITHMETIC_OVERFLOW`, and Long overflow surfaces as `"integer overflow"` - rather than `"long overflow"`. `Abs` uses Rust type names (`Int8`, `Int64`, ...) in the message - instead of Spark's SQL type names, and the scalar path of `UnaryMinus` on Byte / Short emits a - malformed message. The `try_` suggestion is omitted from all of these - ([#5071](https://github.com/apache/datafusion-comet/issues/5071)). +- Byte / Short `Add`, `Subtract`, and `Multiply` overflow raises `ARITHMETIC_OVERFLOW` (for + example `byte overflow`) where Spark raises `BINARY_ARITHMETIC_OVERFLOW`, and integral + `ARITHMETIC_OVERFLOW` messages omit Spark's `try_` suggestion + ([#6217](https://github.com/apache/datafusion-comet/issues/6217)). - Wide-decimal arithmetic overflow, decimal divide-by-zero, and decimal-to-decimal cast overflow raise raw Arrow errors that bypass `SparkErrorConverter` and surface as `CometNativeException` rather than `SparkArithmeticException` with the proper error class and query context @@ -130,18 +129,11 @@ The following native paths silently return values that differ from Spark for edg Most also have entries in the per-category expression pages linked above; they are collected here so users hunting an unexpected value have a single place to check: -- `CAST(boolean AS DECIMAL(p, s))` where `10^s` exceeds the target precision (e.g. - `DECIMAL(1, 1)`) throws `NUMERIC_VALUE_OUT_OF_RANGE` regardless of the eval mode. Spark returns - `NULL` under legacy and try mode, and only throws under ANSI - ([#5068](https://github.com/apache/datafusion-comet/issues/5068)). - `CAST(string AS timestamp)` and `CAST(string AS timestamp_ntz)` trim Unicode whitespace. Spark trims only the bytes `0x00`-`0x20` and `0x7F`, so a value padded with an ASCII control byte parses in Spark and returns `NULL` in Comet, while a value padded with non-ASCII whitespace such as `U+3000` returns `NULL` in Spark and parses in Comet ([#5149](https://github.com/apache/datafusion-comet/issues/5149)). -- **Explicit positive timestamp years:** Spark accepts strings such as `+7528` as the start - of that year, while Comet's native string-to-timestamp cast returns NULL in non-ANSI mode - ([#5716](https://github.com/apache/datafusion-comet/issues/5716)). - Native `RANGE` window frames with an explicit `PRECEDING` / `FOLLOWING` offset diverge from Spark when the boundary arithmetic overflows for `DATE` or `DECIMAL` `ORDER BY` columns ([#5022](https://github.com/apache/datafusion-comet/issues/5022)). @@ -156,13 +148,3 @@ so users hunting an unexpected value have a single place to check: `spark.sql.codegen.fallback` or when the generated code exceeds `spark.sql.codegen.hugeMethodLimit`, where an intermediate overflow that later cancels out returns `NULL` (or raises under ANSI) in Spark but the recovered value in Comet. - -## Object store cache - -When Comet's native scan reads Parquet files, it caches one object store instance per -`(scheme + host + port, hadoop-config-hash)` key. For `abfss://container@account.dfs.core.windows.net/...` -URLs, the container lives in the URL userinfo, not the host, so two containers in one storage -account currently collide on the same cache entry. Within a single executor process, reading from -a second container after a first can be served by the first container's store instance and return -its data. S3, GCS, and HDFS are unaffected because their bucket / host lives in the URL host -component. Tracked by [#4993](https://github.com/apache/datafusion-comet/issues/4993). diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md index 1e1103955b7..cd779dc081a 100644 --- a/docs/source/user-guide/latest/compatibility/operators.md +++ b/docs/source/user-guide/latest/compatibility/operators.md @@ -32,6 +32,18 @@ operator restrictions and aggregate buffer compatibility checks still apply. Parquet writes whose input plans contain an empty relation use Spark's writer to preserve readable empty output files and their schema metadata. +## In-Memory Cache + +Comet can store cached relations (`df.cache()`, `CACHE TABLE`) in Arrow format and scan them +natively. This is experimental and disabled by default; see [In-Memory Cache](../in-memory-cache.md) +for how to enable it. Comet does not replace a `spark.sql.cache.serializer` that the application +has already set. Relations whose schema Comet's Arrow writer does not support are cached in +Spark's default format, and their scans fall back to Spark. Reads that feed Spark operators rather +than Comet operators can be slower than Spark's cache. + +With Kryo and `spark.kryo.registrationRequired=true`, Comet needs its Kryo registrator whether or +not the cache is enabled; see [Kryo serialization](../installation.md#kryo-serialization). + ## Sampling Comet runs `SampleExec` natively when sampling is performed without replacement, which covers @@ -95,8 +107,11 @@ runs natively; it is controlled by `spark.comet.exec.windowGroupLimit.enabled` ( **Known incompatibilities:** -- Signed-zero ordering (`-0.0` vs `+0.0`) diverges from Spark's `RankLimitIterator`; see - [floating-point ordering](./floating-point.md#ordering-signed-zero-00-vs-00). +- Floating-point values nested in array or struct `ORDER BY` keys are compared with Arrow's raw + total ordering, so ranks can differ from Spark when the data mixes `-0.0` and `+0.0` or more + than one NaN representation ([#5507](https://github.com/apache/datafusion-comet/issues/5507)). + Scalar `FLOAT` and `DOUBLE` keys are normalized and match Spark; see + [floating-point ordering](./floating-point.md). ## Round-Robin Partitioning diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index d479e75ffd8..57363125331 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -48,6 +48,8 @@ The following features are not supported and cause Comet to fall back to Spark: does not replicate. By default Comet falls back to Spark in this case. Set `spark.comet.scan.allowDisabledParquetVectorizedReader=true` to opt in to running the Comet Parquet scan regardless. +- A read schema that repeats a Parquet field id, at the top level or within a struct, when + `spark.sql.parquet.fieldId.read.enabled=true`. The following limitation may produce incorrect results without falling back to Spark: diff --git a/docs/source/user-guide/latest/compatibility/spark-versions.md b/docs/source/user-guide/latest/compatibility/spark-versions.md index 9992dc4f1f0..516d08b899a 100644 --- a/docs/source/user-guide/latest/compatibility/spark-versions.md +++ b/docs/source/user-guide/latest/compatibility/spark-versions.md @@ -33,21 +33,17 @@ Spark 3.4.3 is supported with Java 17 and Scala 2.12/2.13. ```{warning} Spark 3.4 support is deprecated as of the 1.0.0 release and will be removed in a future release. Comet continues to build and publish Spark 3.4 binaries in the meantime, but Apache Spark's own -SQL test suite is no longer run against Spark 3.4 on every change: it runs only when a contributor +SQL test suite no longer runs against Spark 3.4 automatically: it runs only when a contributor opts a pull request into it. Regressions specific to Spark 3.4 are therefore more likely to reach a release than on the other supported versions. We recommend moving to Spark 3.5 or later. ``` ### Known Limitations -- **Reading `TimestampLTZ` as `TimestampNTZ`**: Spark 3.4 raises an error for this operation - (SPARK-36182), but Comet's Parquet scan silently returns the raw UTC value instead. - See [Parquet Compatibility](scans.md#parquet-scan-limitations) for details. - -- **Unsupported Parquet type conversions**: Spark 3.4 raises schema incompatibility errors for - certain type mismatches (e.g., reading INT32 as BIGINT, decimal precision changes), but Comet's - Comet's Parquet scan may not detect these and could return unexpected values. - See [Parquet Compatibility](scans.md#parquet-scan-limitations) for details. +- **Extra `SparkException` layer in Parquet schema mismatch errors**: when a Parquet read is + rejected because a file's type cannot be converted to the requested type, the error's cause chain + has one more `SparkException` layer than Spark's own reader produces. + See [Parquet Compatibility](scans.md#schema-mismatch-handling) for details. ## Spark 3.5 @@ -55,18 +51,14 @@ Spark 3.5.9 is supported with Java 17 and Scala 2.12/2.13. ### Known Limitations -- **Reading `TimestampLTZ` as `TimestampNTZ`**: Spark 3.5 raises an error for this operation - (SPARK-36182), but Comet's Parquet scan silently returns the raw UTC value instead. - See [Parquet Compatibility](scans.md#parquet-scan-limitations) for details. - -- **Unsupported Parquet type conversions**: Spark 3.5 raises schema incompatibility errors for - certain type mismatches (e.g., reading INT32 as BIGINT, decimal precision changes), but Comet's - Comet's Parquet scan may not detect these and could return unexpected values. - See [Parquet Compatibility](scans.md#parquet-scan-limitations) for details. +- **Extra `SparkException` layer in Parquet schema mismatch errors**: when a Parquet read is + rejected because a file's type cannot be converted to the requested type, the error's cause chain + has one more `SparkException` layer than Spark's own reader produces. + See [Parquet Compatibility](scans.md#schema-mismatch-handling) for details. ## Spark 4.0 -Spark 4.0.4 is supported with Java 17 and Scala 2.13. +Spark 4.0.4 is supported with Java 17/21 and Scala 2.13. ### Known Limitations @@ -93,6 +85,7 @@ Spark 4.1.3 is supported with Java 17/21 and Scala 2.13. Spark 4.2.0 is provided as experimental support with Java 17 and Scala 2.13. ```{warning} -Spark 4.2 support is experimental. Comet tests run in CI for this version, but the Spark SQL tests -do not yet. It is intended for early evaluation only and should not be used in production. +Spark 4.2 support is experimental. Comet's own test suites run nightly for this version, but +Apache Spark's SQL test suite does not run against it yet. It is intended for early evaluation only +and should not be used in production. ``` diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index eff37b0a1ee..197b31f7dfa 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -29,7 +29,8 @@ Arrow format, allowing the Comet pipeline to take over after that, but the proce ### Apache Iceberg -Comet accelerates Iceberg scans of Parquet files. See the [Iceberg Guide] for more information. +Comet accelerates Iceberg scans of Parquet files and has an experimental, opt-in native Iceberg writer. +See the [Iceberg Guide] and [Iceberg Writes](iceberg-writes.md) for more information. [Iceberg Guide]: iceberg.md @@ -37,7 +38,8 @@ Comet accelerates Iceberg scans of Parquet files. See the [Iceberg Guide] for mo Comet provides experimental Rust-based CSV scan support. When `spark.comet.scan.csv.v2.enabled` is enabled, CSV files are read in Rust for improved performance. This feature is experimental and performance benefits are -workload-dependent. +workload-dependent. Only Spark's DataSource V2 CSV scan is accelerated, and Spark reads CSV through the V1 API by +default, so also remove `csv` from `spark.sql.sources.useV1SourceList`. Alternatively, when `spark.comet.convert.csv.enabled` is enabled, data from Spark's CSV reader is immediately converted into Arrow format, allowing the Comet pipeline to take over after that. @@ -308,6 +310,10 @@ credential providers and options documented above also apply to alias-scheme URL translation feeds the native Iceberg scan; see [Object store configuration (S3)](iceberg.md#object-store-configuration-s3) in the Iceberg guide. +A native Parquet scan whose alias-scheme paths span more than one bucket falls back to Spark. Alias +schemes apply to native scans only: a native Iceberg write to an alias-scheme location falls back to +iceberg-java. + ### Examples The following examples demonstrate how to configure S3 access using different authentication methods. @@ -344,7 +350,7 @@ Comet's S3 support has the following limitations: 1. **Partial Hadoop S3A configuration support**: Not all Hadoop S3A configurations are currently supported. Only the configurations listed in the tables above are translated and applied to the underlying `object_store` crate. -2. **Custom credential providers**: Custom implementations of AWS credential providers are not supported. The implementation only supports the standard credential providers listed in the table above. We are planning to add support for custom credential providers through a JNI-based adapter that will allow calling Java credential providers from Rust code. See [#1829](https://github.com/apache/datafusion-comet/issues/1829) for more details. +2. **Custom credential providers**: Custom credential provider classes named in `fs.s3a.aws.credentials.provider` are not supported; only the standard providers listed in the table above are. To route credential requests through your own Java code, implement Comet's `CometS3CredentialProvider` SPI; see [S3 Credential Providers](s3-credential-providers.md). Broader Hadoop S3A integration is tracked in [#1829](https://github.com/apache/datafusion-comet/issues/1829). ## Azure diff --git a/docs/source/user-guide/latest/datatypes.md b/docs/source/user-guide/latest/datatypes.md index 5d3e6e736c0..1b619e67cad 100644 --- a/docs/source/user-guide/latest/datatypes.md +++ b/docs/source/user-guide/latest/datatypes.md @@ -71,12 +71,12 @@ the tables below and may be reconsidered based on demand: ## Datetime -| Type | Status | Notes | -| ------------------ | ------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `DateType` | ✅ | | -| `TimestampType` | ✅ | | -| `TimestampNTZType` | ✅ | | -| `TimeType` | ⚠️ | Spark 4.1+. Native serialization is in place; some operators (sort, shuffle, min/max) are still being wired up ([#4288](https://github.com/apache/datafusion-comet/issues/4288)). | +| Type | Status | Notes | +| ------------------ | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `DateType` | ✅ | | +| `TimestampType` | ✅ | | +| `TimestampNTZType` | ✅ | | +| `TimeType` | ⚠️ | Spark 4.1+. Native serialization is in place; some operators (sort, min/max) are still being wired up ([#4288](https://github.com/apache/datafusion-comet/issues/4288)). | ## Interval @@ -97,7 +97,7 @@ functions. Remaining work is tracked by | Type | Status | Notes | | ------------ | ------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `StructType` | ✅ | Empty structs (no fields) fall back. | +| `StructType` | ✅ | Empty structs (no fields) and structs with duplicate field names fall back. | | `ArrayType` | ✅ | | | `MapType` | ✅ | Hash aggregate group keys cannot contain a `MapType` (transitively): Arrow's row format used by DataFusion's grouped hash aggregate does not support `Map`, so such groupings fall back. | @@ -113,8 +113,9 @@ Direct projection requires explicit configuration on every supported Spark versi Parquet timestamp inference settings. Support for Spark's whole-value pushdown rewrite is tracked by [#5519](https://github.com/apache/datafusion-comet/issues/5519). Nested Variant columns, pushed-down Variant field extraction, expressions, writes, shuffle and spill, Python operators, encrypted -files, and Iceberg scans fall back to Spark. Spark also handles columnar-to-row conversion of -the native scan output and strict reads with `allowReadingShredded=false`. Broader +files, and Iceberg scans that read a Variant column fall back to Spark. Iceberg scans of tables +whose Variant columns the query does not read run natively. Spark also handles columnar-to-row +conversion of the native scan output and strict reads with `allowReadingShredded=false`. Broader support is tracked by [#4295](https://github.com/apache/datafusion-comet/issues/4295) and [#3983](https://github.com/apache/datafusion-comet/issues/3983). diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 700d027d0da..1328f9029ec 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -30,9 +30,9 @@ Some ✅ Supported expressions have specific incompatible cases that are not run Those cases must be opted into per expression with `spark.comet.expression.EXPRNAME.allowIncompatible=true` (where `EXPRNAME` is the Spark expression class name, for example `Cast`). There is no global opt-in. By default such a case -either falls back to Spark (for example `cast`) or, when the expression has a Spark-compatible -codegen-dispatch implementation, runs through that instead (for example the regex and JSON -families). See [Native and codegen-dispatch implementations](compatibility/index.md#native-and-codegen-dispatch-implementations) +either falls back to Spark (for example the `mode` aggregate) or, when the expression has a +Spark-compatible codegen-dispatch implementation, runs through that instead (for example `cast` +and the regex and JSON families). See [Native and codegen-dispatch implementations](compatibility/index.md#native-and-codegen-dispatch-implementations) for how Comet chooses. Most expressions can also be disabled with `spark.comet.expression.EXPRNAME.enabled=false`, where @@ -62,18 +62,18 @@ The Implementation column is auto-generated from the serde definitions in `Query ## Not currently planned Comet focuses acceleration on mainstream relational, string, datetime, math, and collection -expressions. The following function families are **not currently planned** for native acceleration (they are not on the 1.0 roadmap): specialized functionality with narrow real-world analytics use and high implementation cost. They fall back to Spark and may be reconsidered based on demand: +expressions. The following function families are **not currently planned** for native acceleration (they are not on the current roadmap): specialized functionality with narrow real-world analytics use and high implementation cost. They fall back to Spark and may be reconsidered based on demand: - **Probabilistic sketches and approximate top-k** (`kll_sketch_*`, `hll_*`, `theta_*`, `count_min_sketch`, `bitmap_*`, `approx_top_k*`): specialized data structures with exact-correctness traps. - **Geospatial** (`st_*`): brand-new Spark 4.1 functionality, specialized. - **Avro / Protobuf codecs** (`from_avro`, `to_avro`, `from_protobuf`, `to_protobuf`, `schema_of_avro`): format conversion belongs at the IO layer, not expression evaluation. -- **JVM reflection** (`java_method`, `reflect`): niche, and they invoke arbitrary JVM methods (a security concern). +- **JVM reflection** (`java_method`, `reflect`, `try_reflect`): niche, and they invoke arbitrary JVM methods (a security concern). - **UTF-8 validation** (`is_valid_utf8`, `make_valid_utf8`, `validate_utf8`, `try_validate_utf8`): niche Spark 4.x string-validation helpers. - **Miscellaneous niche** (`histogram_numeric`, `version`, `sentences`, `quote`): low-value or specialized functions with little benefit from native acceleration. The file-metadata functions `input_file_name`, `input_file_block_start`, and `input_file_block_length` depend on scan-internal per-row file information rather than the expression layer; their support status is covered in the [scan compatibility guide](compatibility/scans.md). -Note that `median` and `mode` are planned: they are mainstream exact aggregates. `approx_count_distinct` is supported because Comet ports Spark's `HyperLogLogPlusPlus` exactly, so its result is bit-identical to Spark. +Note that `median` and `mode` are supported: they are mainstream exact aggregates. `mode` runs natively only with `spark.comet.expression.Mode.allowIncompatible=true`. `approx_count_distinct` is supported because Comet ports Spark's `HyperLogLogPlusPlus` exactly, so its result is bit-identical to Spark. The tables below list every Spark built-in expression with its current status. @@ -114,8 +114,9 @@ The tables below list every Spark built-in expression with its current status. | `median` | ✅ | — | Rewrites to `percentile(col, 0.5)` and runs natively for supported percentile inputs | | `min` | ✅ | Native | | | `min_by` | ✅ | Native | Value and ordering must be fixed-length types | -| `mode` | ✅ | Native | `mode(col)` only; Spark breaks ties non-deterministically, so Comet returns the smallest tied value and falls back by default, opt-in via allowIncompatible ([#3970](https://github.com/apache/datafusion-comet/issues/3970)) | +| `mode` | ✅ | Native | `mode(col)` only; Spark breaks ties non-deterministically, so Comet returns the smallest tied value and falls back by default, opt-in via allowIncompatible | | `percentile` | ✅ | Native | Single literal percentage on numeric input runs natively; array of percentages and a frequency argument fall back to Spark | +| `percentile_approx` | ✅ | Native | Alias of `approx_percentile`; same restrictions apply | | `percentile_cont` | ✅ | — | Spark 4.0+ `WITHIN GROUP (ORDER BY ...)`; ascending only runs natively, `DESC` falls back to Spark | | `percentile_disc` | 🔜 | — | Percentile aggregate | | `regr_avgx` | ✅ | — | Native: Spark rewrites to `Average` (tests in [#4551](https://github.com/apache/datafusion-comet/pull/4551)) | @@ -150,14 +151,14 @@ The tables below list every Spark built-in expression with its current status. | `array` | ✅ | Native | | | `array_append` | ✅ | Native | | | `array_compact` | ✅ | — | | -| `array_contains` | ✅ | Native | NaN/signed-zero handling may differ ([details](compatibility/floating-point.md)) | +| `array_contains` | ✅ | Native | Float/double element arrays route through the JVM codegen dispatcher by default; the native path is opt-in via allowIncompatible | | `array_distinct` | ✅ | Native | NaN/signed-zero handling may differ ([details](compatibility/floating-point.md)) | | `array_except` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the incompatible native path is opt-in via allowIncompatible ([details](compatibility/expressions/array.md)) | | `array_insert` | ✅ | Native | | | `array_intersect` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the incompatible native path is opt-in via allowIncompatible ([details](compatibility/expressions/array.md)) | | `array_join` | ✅ | Hybrid | Native for literal or column delimiter and null replacement; other cases and non-UTF8_BINARY collations use the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | | `array_max` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | -| `array_min` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) | +| `array_min` | ✅ | Native | May return `-0.0` where Spark returns `0.0` when both zeros are present | | `array_position` | ✅ | Native | Binary/struct/map/null elements fall back | | `array_prepend` | ✅ | — | | | `array_remove` | ✅ | Native | | @@ -171,7 +172,7 @@ The tables below list every Spark built-in expression with its current status. | `sequence` | ✅ | Hybrid | Integral types run natively; date/timestamp sequences use codegen dispatch | | `shuffle` | ✅ | Native | Binary/struct/map elements fall back | | `slice` | ✅ | Native | Native ([#4149](https://github.com/apache/datafusion-comet/pull/4149)) | -| `sort_array` | ✅ | Hybrid | Nested struct/null arrays fall back | +| `sort_array` | ✅ | Hybrid | Struct, nested-array, and null elements run natively; other element types (for example intervals), and floating-point elements when `spark.comet.exec.strictFloatingPoint=true`, route through the JVM codegen dispatcher | --- @@ -200,8 +201,8 @@ The tables below list every Spark built-in expression with its current status. | --- | --- | --- | --- | | `array_size` | ✅ | — | | | `cardinality` | ✅ | Native | | -| `concat` | ✅ | Hybrid | Binary/array children fall back | -| `reverse` | ✅ | Hybrid | Binary-element arrays fall back (Incompatible) ([details](compatibility/expressions/array.md)) | +| `concat` | ✅ | Hybrid | Binary/array children and non-UTF8_BINARY collations route through the JVM codegen dispatcher | +| `reverse` | ✅ | Hybrid | Arrays with binary, struct, or map elements, and collated strings, route through the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) | | `size` | ✅ | Native | | --- @@ -229,7 +230,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | Function | Status | Implementation | Notes | | --- | --- | --- | --- | -| `cast` | ✅ | Native | Some casts fall back; float-to-decimal is opt-in ([details](compatibility/expressions/cast.md)) | +| `cast` | ✅ | Native | Casts without a native path (for example boolean to decimal) route through the JVM codegen dispatcher ([details](compatibility/expressions/cast.md)) | --- @@ -278,7 +279,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `make_date` | ✅ | Native | | | `make_dt_interval` | ✅ | Codegen dispatch | | | `make_interval` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; intervals outside Arrow's nanosecond range are tracked by [#5279](https://github.com/apache/datafusion-comet/issues/5279); the native path is opt-in via allowIncompatible ([details](compatibility/expressions/datetime.md)) | -| `make_time` | 🔜 | — | Spark 4.1 TIME type; tracked by [#4288](https://github.com/apache/datafusion-comet/issues/4288) | +| `make_time` | ✅ | — | Spark 4.1+; requires `spark.sql.timeType.enabled=true`, which Spark leaves off by default. Runs natively; remaining TIME type work is tracked by [#4288](https://github.com/apache/datafusion-comet/issues/4288) | | `make_timestamp` | ✅ | Hybrid | | | `make_timestamp_ltz` | ✅ | — | 2-arg TIME form falls back | | `make_timestamp_ntz` | ✅ | — | 2-arg TIME form falls back | @@ -301,7 +302,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `timestampadd` | ✅ | — | Reached through the grammar rather than the function registry; runs through codegen dispatch | | `timestampdiff` | ✅ | — | Reached through the grammar rather than the function registry; runs through codegen dispatch | | `to_date` | ✅ | — | Rewrites to `Cast` (or `Cast(GetTimestamp)` with a format) before Comet sees the plan | -| `to_time` | 🔜 | — | Spark 4.1 TIME type; tracked by [#4288](https://github.com/apache/datafusion-comet/issues/4288) | +| `to_time` | ✅ | — | Spark 4.1+; requires `spark.sql.timeType.enabled=true`, which Spark leaves off by default. The one-argument form runs natively | | `to_timestamp` | ✅ | — | Rewrites to `Cast` (or `GetTimestamp` with a format) before Comet sees the plan | | `to_timestamp_ltz` | ✅ | — | Rewrites to `to_timestamp` (`TimestampType`) | | `to_timestamp_ntz` | ✅ | — | Rewrites to `to_timestamp` (`TimestampNTZType`) | @@ -310,8 +311,10 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `trunc` | ✅ | Hybrid | | | `try_make_interval` | ✅ | — | Rewrites to `MakeInterval`; same support as `make_interval` (Spark 4.0+) | | `try_make_timestamp` | ✅ | — | | +| `try_make_timestamp_ltz` | ✅ | — | Same support as `try_make_timestamp` (Spark 4.0+) | +| `try_make_timestamp_ntz` | ✅ | — | Same support as `try_make_timestamp` (Spark 4.0+) | | `try_to_date` | ✅ | — | Rewrites to `Cast`/`GetTimestamp` before Comet sees the plan; same support as `to_date` | -| `try_to_time` | 🔜 | — | Spark 4.1 TIME type; tracked by [#4288](https://github.com/apache/datafusion-comet/issues/4288) | +| `try_to_time` | ✅ | — | Same support as `to_time` | | `try_to_timestamp` | ✅ | — | Rewrites to `Cast`/`GetTimestamp` before Comet sees the plan; same support as `to_timestamp` | | `unix_date` | ✅ | Native | | | `unix_micros` | ✅ | Codegen dispatch | | @@ -329,7 +332,8 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci ## generator_funcs `explode`, `explode_outer`, `posexplode`, and `posexplode_outer` are supported via -`CometExplodeExec` (operator-level, not expression-level). Enabled by default via +`CometExplodeExec` (operator-level, not expression-level) for array input; map input falls back +to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enabled by default via `spark.comet.exec.explode.enabled`. | Function | Status | Implementation | Notes | @@ -362,13 +366,13 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | Function | Status | Implementation | Notes | | --- | --- | --- | --- | -| `from_json` | ✅ | Hybrid | Falls back by default; opt-in via allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#from_json)) | -| `get_json_object` | ✅ | Hybrid | Some inputs need allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#get_json_object)) | -| `json_array_length` | ✅ | Hybrid | Single-quoted/trailing JSON needs allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#json_array_length)) | +| `from_json` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the native path (supported schemas only) is opt-in via allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#from_json)) | +| `get_json_object` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the native path is opt-in via allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#get_json_object)) | +| `json_array_length` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the native path, which differs for single-quoted JSON, unescaped control characters, and trailing content, is opt-in via allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#json_array_length)) | | `json_object_keys` | ✅ | Codegen dispatch | | | `json_tuple` | 🔜 | — | [#3160](https://github.com/apache/datafusion-comet/issues/3160) | | `schema_of_json` | ✅ | Codegen dispatch | | -| `to_json` | ✅ | Hybrid | Options and map/array inputs fall back ([audit](../../contributor-guide/expression-audits/json_funcs.md#to_json)) | +| `to_json` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default, including options and map/array inputs; the native path is opt-in via allowIncompatible ([audit](../../contributor-guide/expression-audits/json_funcs.md#to_json)) | --- @@ -401,7 +405,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `map_contains_key` | ✅ | — | | | `map_entries` | ✅ | Native | | | `map_from_arrays` | ✅ | Native | | -| `map_from_entries` | ✅ | Hybrid | BinaryType key/value falls back (Incompatible) ([details](compatibility/expressions/map.md)) | +| `map_from_entries` | ✅ | Hybrid | BinaryType keys/values and `spark.sql.mapKeyDedupPolicy=LAST_WIN` route through the JVM codegen dispatcher ([details](compatibility/expressions/map.md)) | | `map_keys` | ✅ | Native | | | `map_values` | ✅ | Native | | | `str_to_map` | ✅ | Hybrid | | @@ -462,7 +466,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `radians` | ✅ | Native | | | `rand` | ✅ | Native | | | `randn` | ✅ | Native | | -| `random` | ✅ | Native | Alias for `rand` (Spark 4.0+); seed must be a literal | +| `random` | ✅ | Native | Alias for `rand`; seed must be a literal | | `randstr` | ✅ | Native | Random string (Spark 4.0+); length and seed must be literals | | `rint` | ✅ | Native | | | `round` | ✅ | Hybrid | Float/double inputs route through the JVM codegen dispatcher; other types run natively | @@ -541,9 +545,9 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `like` | ✅ | Hybrid | | | `not` | ✅ | Native | | | `or` | ✅ | Native | | -| `regexp` | ✅ | Hybrid | In-subset literals run natively; others fall back by default ([details](compatibility/regex.md)) | -| `regexp_like` | ✅ | Hybrid | In-subset literals run natively; others fall back by default ([details](compatibility/regex.md)) | -| `rlike` | ✅ | Hybrid | In-subset literals run natively; others fall back by default ([details](compatibility/regex.md)) | +| `regexp` | ✅ | Hybrid | In-subset literal patterns run natively; other patterns route through the JVM codegen dispatcher ([details](compatibility/regex.md)) | +| `regexp_like` | ✅ | Hybrid | In-subset literal patterns run natively; other patterns route through the JVM codegen dispatcher ([details](compatibility/regex.md)) | +| `rlike` | ✅ | Hybrid | In-subset literal patterns run natively; other patterns route through the JVM codegen dispatcher ([details](compatibility/regex.md)) | --- @@ -587,12 +591,12 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `overlay` | ✅ | Codegen dispatch | | | `position` | ✅ | Codegen dispatch | | | `printf` | ✅ | Codegen dispatch | | -| `regexp_count` | ✅ | — | Runs natively (rewrites to `size(regexp_extract_all(...))`) | +| `regexp_count` | ✅ | — | Rewrites to `size(regexp_extract_all(...))`; `regexp_extract_all` routes through the JVM codegen dispatcher by default | | `regexp_extract` | ✅ | Native | | | `regexp_extract_all` | ✅ | Native | | | `regexp_instr` | ✅ | Codegen dispatch | Routed through the JVM codegen dispatcher | | `regexp_replace` | ✅ | Hybrid | | -| `regexp_substr` | ✅ | — | Runs natively (rewrites to `nullif(regexp_extract(...), '')`) | +| `regexp_substr` | ✅ | — | Rewrites to `nullif(regexp_extract(...), '')`; `regexp_extract` routes through the JVM codegen dispatcher by default | | `repeat` | ✅ | Native | | | `replace` | ✅ | Hybrid | | | `right` | ✅ | Native | | @@ -606,16 +610,16 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `substr` | ✅ | Native | | | `substring` | ✅ | Native | | | `substring_index` | ✅ | Native | | -| `to_binary` | ✅ | — | Hex and base64 forms accelerated natively; the `utf-8` form rewrites to `encode` and runs via codegen dispatch on Spark 4.0+ | +| `to_binary` | ✅ | — | The hex form runs natively; the base64 and `utf-8` forms route through the JVM codegen dispatcher | | `to_char` | ✅ | Codegen dispatch | | | `to_number` | ✅ | Codegen dispatch | | | `to_varchar` | ✅ | Codegen dispatch | | | `translate` | ✅ | Hybrid | Codegen dispatch by default: DataFusion's `translate` iterates over Unicode graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion sentinel, so the native path is opt-in via allowIncompatible | | `trim` | ✅ | Native | | -| `try_to_binary` | ✅ | — | Runs natively (rewrites to `try_eval(to_binary(...))`) | +| `try_to_binary` | ✅ | — | Rewrites to `try_eval(to_binary(...))`, which routes through the JVM codegen dispatcher | | `try_to_number` | ✅ | Codegen dispatch | Routed through the JVM codegen dispatcher | | `ucase` | ✅ | Hybrid | | -| `unbase64` | ✅ | Codegen dispatch | | +| `unbase64` | ✅ | Codegen dispatch | Column or literal input runs natively; other child expressions, and the strict form used by `to_binary(str, 'base64')`, route through the JVM codegen dispatcher | | `upper` | ✅ | Hybrid | | --- @@ -634,6 +638,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | Function | Status | Implementation | Notes | | --- | --- | --- | --- | | `parse_url` | ✅ | Native | | +| `try_parse_url` | ✅ | — | Spark 4.0+; runs natively and returns NULL for an invalid URL | | `try_url_decode` | ✅ | — | | | `url_decode` | ✅ | — | | | `url_encode` | ✅ | — | | @@ -694,6 +699,8 @@ Comet also accelerates a number of Catalyst expressions that have no Spark SQL f - **Accessor expressions (subscript and field access, not functions):** struct field access (`col.field`), array element access (`arr[i]`), and map value access (`map[key]`). - **Internal decimal arithmetic:** `CheckOverflow`, `MakeDecimal`, and `UnscaledValue`, which the analyzer inserts around decimal operations. - **User-defined functions:** Scala UDFs registered through the DataFrame or SQL API. +- **DataSource V2 catalog functions:** Iceberg's system functions `bucket`, `truncate`, `years`, `months`, `days`, and `hours` (for example `system.bucket(16, id)`) run natively; see [Iceberg system functions](iceberg.md#iceberg-system-functions). +- **Lowered built-ins:** Spark lowers some built-in functions to `StaticInvoke` or `Invoke` calls. Those without a native mapping run through the JVM codegen dispatcher when their input and output types are supported. - **Structural expressions:** aliases, attribute references, literals, sort orders, and `CASE WHEN`. This list is illustrative, not exhaustive: the per-function tables are not the complete set of expressions Comet can accelerate. diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md index 74712b73147..c66ba422646 100644 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ b/docs/source/user-guide/latest/iceberg-writes.md @@ -44,7 +44,7 @@ and it is the foundation for the second toggle: when `spark.comet.iceberg.write.enabled=true` and the write passes the eligibility check below, the `IcebergWrite` operator's per-task Parquet write is delegated to [iceberg-rust](https://github.com/apache/iceberg-rust) via Comet's native execution pipeline -([#5308](https://github.com/apache/datafusion-comet/issues/5308)). +([#5361](https://github.com/apache/datafusion-comet/pull/5361)). ## How the native write works @@ -127,7 +127,7 @@ trade-off, only no plan change. ## Native Parquet write eligibility When `spark.comet.iceberg.write.enabled=true` -([#5308](https://github.com/apache/datafusion-comet/issues/5308)), the `IcebergWrite` operator's +([#5361](https://github.com/apache/datafusion-comet/pull/5361)), the `IcebergWrite` operator's per-task Parquet write is delegated to [iceberg-rust](https://github.com/apache/iceberg-rust). The native writer must produce the same outcome as iceberg-java — the same Parquet features, statistics, and manifest metadata — so a write is only eligible when every table property it @@ -136,7 +136,9 @@ feeding the write is fully Comet-native. For a partitioned table that plan inclu distribution and local sort Iceberg requests on its partition transforms; those stay native because the transforms themselves have native implementations (see [Iceberg system functions](iceberg.md)). Ineligible writes run through iceberg-java unchanged, -with the reason reported as a fall-back reason in Comet's extended EXPLAIN output. +with the reason reported as a fall-back reason in Comet's extended EXPLAIN output. A write that +runs natively shows `CometIcebergWrite` under `IcebergCommit` in the physical plan; an ineligible +write keeps `IcebergWrite`. The native writer reads its input as Arrow batches from a Comet operator, so the write's input must itself run in Comet. A write whose input is a local relation, such as `INSERT ... VALUES` or @@ -177,7 +179,8 @@ Within the namespaces that shape data-file bytes — `write.parquet.*` and `parq everything not listed above must be absent: unvetted `write.parquet.*` keys (e.g. `bloom-filter-max-bytes`, `stats-enabled.column.*`, keys added by future Iceberg versions), any `parquet.*` table property (including `parquet.enable.dictionary`), and any `parquet.*` -key in the session Hadoop configuration (with `HadoopFileIO`-backed output those reach +key in the session Hadoop configuration other than the reader-only +`parquet.hadoop.vectored.io.enabled` (with `HadoopFileIO`-backed output those reach iceberg-java's writer but not the native one). Also gated explicitly: any `encryption.*` key, `write.object-storage.enabled=true`, `write.location-provider.impl`, and `io-impl`. diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index 3bd7f8e23a9..a0a0b521693 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -26,7 +26,7 @@ then serialized to Comet's native execution engine (see [PR #2528](https://github.com/apache/datafusion-comet/pull/2528)). The example below uses Spark's package downloader to retrieve Comet $COMET_VERSION and Iceberg -1.8.1, but Comet has been tested with Iceberg 1.5, 1.7, 1.8, 1.9, 1.10, and 1.11. The native Iceberg +1.8.1, but Comet has been tested with Iceberg 1.5, 1.8, 1.9, 1.10, and 1.11. The native Iceberg reader is enabled by default. To disable it, set `spark.comet.scan.icebergNative.enabled=false`. The example uses the Spark 3.5 / Scala 2.12 build of Comet; substitute the Comet artifact @@ -97,7 +97,7 @@ The native Iceberg reader supports the following features: - Equality and comparison predicates (`=`, `!=`, `>`, `>=`, `<`, `<=`) - Logical operators (`AND`, `OR`) - NULL checks (`IS NULL`, `IS NOT NULL`) on primitive columns -- `IN` and `NOT IN` list operations +- `IN` list operations (`NOT IN` is applied after the scan) - `BETWEEN` operations NULL checks on struct, array, and map columns still use native scans and return correct @@ -156,6 +156,8 @@ scala> spark.sql("SELECT * FROM rest_cat.db.test_table").show() The native reader has its own Rust object store client and does not go through Iceberg's JVM FileIO, neither `S3FileIO` nor the older Hadoop S3A filesystem. It configures that client from the catalog's `s3.*` properties (the same keys `S3FileIO` reads), from `spark.hadoop.fs.s3a.*` settings, or, for a scheme opted into `spark.hadoop.fs.comet.s3Compliant.schemes`, from vendor-style `fs...*` keys (see [S3-Compliant Filesystem Schemes](datasources.md#s3-compliant-filesystem-schemes)). That third source is translated into the same `fs.s3a.*` shape as the second before it reaches the reader. S3 configuration therefore reaches the native reader through one of these three channels. +Per-bucket `fs.s3a.bucket..*` settings apply to the bucket that holds the table's data and delete files. The native reader uses one object-store configuration per scan, so a scan whose data or delete files span more than one S3 bucket falls back to Spark. + For a custom S3-compatible endpoint, configure the catalog with the endpoint, path-style access, region, and credentials (Hive shown): ```shell @@ -170,7 +172,7 @@ For a custom S3-compatible endpoint, configure the catalog with the endpoint, pa --conf spark.sql.catalog.s3_cat.s3.secret-access-key=... ``` -These `s3.*` storage properties are not specific to the Hive catalog shown here. When `s3.access-key-id` / `s3.secret-access-key` are omitted, credentials come from the standard AWS chain (environment variables, instance profiles, and so on). `client.region` is auto-detected for AWS but should be set for non-AWS endpoints. If your REST catalog vends temporary credentials, the native reader does not consume them automatically, and wiring that requires the credential provider bridge. See Iceberg's [S3 FileIO](https://iceberg.apache.org/docs/latest/aws/#s3-fileio) docs for the full property list, and [S3 Credential Providers](s3-credential-providers.md) for vended or per-request credentials. +These `s3.*` storage properties are not specific to the Hive catalog shown here. When `s3.access-key-id` / `s3.secret-access-key` are omitted, credentials come from the standard AWS chain (environment variables, instance profiles, and so on). The region is not auto-detected: when neither the catalog (`client.region` or `s3.region`) nor the executor environment (`AWS_REGION` or `AWS_DEFAULT_REGION`) supplies one, Comet uses `us-east-1`, so set it for AWS buckets in any other region. If your REST catalog vends temporary credentials, the native reader does not consume them automatically, and wiring that requires the credential provider bridge. See Iceberg's [S3 FileIO](https://iceberg.apache.org/docs/latest/aws/#s3-fileio) docs for the full property list, and [S3 Credential Providers](s3-credential-providers.md) for vended or per-request credentials. ### Current limitations @@ -178,10 +180,14 @@ The following scenarios will fall back to the JVM Iceberg reader: - Iceberg table spec v4 or newer - v3 tables with columns that declare an initial default value -- v3 column types the native reader cannot read (`variant`, `geometry`, `geography`, `unknown`) +- v3 column types the native reader cannot read (`geometry`, `geography`, `unknown`), and + `variant` columns the query reads (on Spark 4.0+, a table whose `variant` columns are not + projected is read natively) - Encrypted tables with 192-bit data keys (no AES-192-GCM in the underlying crypto) - Delete files in a format other than Parquet or Puffin (Avro or ORC positional/equality deletes) - Tables backed by Avro or ORC data files (only Parquet is accelerated) +- Scans whose data or delete files span more than one S3 bucket (the native reader uses one + object-store configuration per scan) - Tables partitioned on `BINARY` or `DECIMAL` (with precision >28) columns - Scans with residual filters using `truncate`, `bucket`, `year`, `month`, `day`, or `hour` transform functions (partition pruning still works, but row-level filtering of these @@ -245,6 +251,6 @@ the expression fall back to Spark. ### Task input metrics -The native Iceberg reader populates Spark's task-level `inputMetrics.bytesRead` (visible in the Spark UI Stages tab) using the `bytes_read` counter from iceberg-rust's `ScanMetrics`. This counter includes bytes read from both data files and delete files. +The native Iceberg reader populates Spark's task-level `inputMetrics.bytesRead` (visible in the Spark UI Stages tab) using the `bytes_read` counter from iceberg-rust's `ScanMetrics`. This counter includes bytes read from both data files and delete files. The scan's SQL metrics, including Iceberg's planning counters, are listed in the [Metrics Guide](metrics.md#cometicebergnativescan). Iceberg Java does not explicitly report `bytesRead` to Spark's task input metrics. On the iceberg Java path, any `bytesRead` value comes from Hadoop's filesystem-level I/O counters, not from Iceberg itself. Because Comet's native reader and the Hadoop filesystem use different counting mechanisms, the exact byte counts will differ between the two paths. diff --git a/docs/source/user-guide/latest/in-memory-cache.md b/docs/source/user-guide/latest/in-memory-cache.md index c37fa2dc32d..8e39f00fe5e 100644 --- a/docs/source/user-guide/latest/in-memory-cache.md +++ b/docs/source/user-guide/latest/in-memory-cache.md @@ -159,7 +159,7 @@ Spark serializes a cached batch with `spark.serializer` whenever the block leave `_SER` storage levels, replication, cross-executor fetches, and the disk half of the default `MEMORY_AND_DISK`. So an ordinary `df.cache()` that spills is enough to reach it. -If you run with `spark.kryo.registrationRequired=true`, register Comet's classes: +If you run Kryo with `spark.kryo.registrationRequired=true`, register Comet's classes: ``` spark.serializer=org.apache.spark.serializer.KryoSerializer @@ -171,7 +171,8 @@ Comet cannot set `spark.kryo.registrator` for you the way it sets `spark.sql.cac `KryoSerializer` reads it when `SparkEnv` builds the serializer, which happens before any plugin runs. Without it, caching fails with a "Class is not registered" error that does not name this feature. Comet's driver plugin warns at startup when it sees Kryo, `registrationRequired`, and no -registrator. +registrator. Native broadcast needs the same registrator even when the cache is disabled; see +[Kryo serialization](installation.md#kryo-serialization). ## Limitations diff --git a/docs/source/user-guide/latest/installation.md b/docs/source/user-guide/latest/installation.md index 9ed0e6bfe3e..3dca6dfb65b 100644 --- a/docs/source/user-guide/latest/installation.md +++ b/docs/source/user-guide/latest/installation.md @@ -49,29 +49,31 @@ Comet requires JDK 17 or later. JDK 11 is no longer supported as of the 1.1.0 re ```{warning} Spark 3.4 support is deprecated as of the 1.0.0 release and will be removed in a future release. -Apache Spark's own SQL test suite is no longer run against Spark 3.4 on every change; it runs only -on demand. We recommend moving to Spark 3.5 or later. +Apache Spark's own SQL test suite no longer runs against Spark 3.4 automatically; it runs only on +demand. We recommend moving to Spark 3.5 or later. ``` | Spark Version | Java Version | Scala Version | Comet Tests in CI | Spark SQL Tests in CI | | ------------- | ------------ | ------------- | ----------------- | --------------------- | -| 3.4.3 | 17 | 2.12/2.13 | Yes | On demand | -| 3.5.9 | 17 | 2.12/2.13 | Yes | Yes | -| 4.0.4 | 17/21 | 2.13 | Yes | Yes | -| 4.1.3 | 17/21 | 2.13 | Yes | Yes | +| 3.4.3 | 17 | 2.12/2.13 | Nightly | On demand | +| 3.5.9 | 17 | 2.12/2.13 | Nightly | Nightly | +| 4.0.4 | 17/21 | 2.13 | Nightly | Nightly | +| 4.1.3 | 17/21 | 2.13 | Before merge | Before merge | Note that we do not test the full matrix of supported Java and Scala versions in CI for every Spark version. -"On demand" in the table above means the suite is not run automatically before a change is merged. -A contributor can still run it against an individual pull request, but Spark 3.4 is no longer -covered by default. +"Before merge" in the table above means the suite must pass before a change is merged. "Nightly" means +the suite runs once a day against the `main` branch, so a regression it finds is caught after the change +has been merged rather than before. "On demand" means the suite does not run automatically at all. A +contributor can still run it against an individual pull request, but Spark 3.4 is no longer covered by +default. Experimental support is provided for the following versions of Apache Spark and is intended for development/testing use only and should not be used in production yet. | Spark Version | Java Version | Scala Version | Comet Tests in CI | Spark SQL Tests in CI | | ------------- | ------------ | ------------- | ----------------- | --------------------- | -| 4.2.0 | 17 | 2.13 | Yes | No | +| 4.2.0 | 17 | 2.13 | Nightly | No | Note that Comet may not fully work with proprietary forks of Apache Spark such as the Spark versions offered by Cloud Service Providers. @@ -259,3 +261,20 @@ Some cluster managers may require additional configuration, see Row copy - +- CometNativeExec <- Arrow batch - +- CometScan +MapInArrow / MapInPandas ++- CometColumnarToRow <- Arrow -> Row copy + +- CometNativeScan <- Arrow batch ``` With the optimization enabled: ``` CometMapInBatch <- Arrow batch in/out, Python runner attached -+- CometNativeExec - +- CometScan ++- CometNativeScan ``` ## Configuration @@ -95,7 +93,7 @@ worker. Both confs can be set independently. | PySpark API | Spark Plan Node | Supported | | -------------------------------- | --------------------------- | --------- | -| `df.mapInArrow(func, schema)` | `PythonMapInArrowExec` | Yes | +| `df.mapInArrow(func, schema)` | `MapInArrowExec` | Yes | | `df.mapInPandas(func, schema)` | `MapInPandasExec` | Yes | | `@pandas_udf` (scalar) | `ArrowEvalPythonExec` | Not yet | | `df.applyInPandas(func, schema)` | `FlatMapGroupsInPandasExec` | Not yet | @@ -144,17 +142,15 @@ You should see: ``` CometMapInBatch ... -+- CometNativeExec ... - +- CometScan ... ++- CometNativeScan parquet ... ``` Instead of the unoptimized plan: ``` -PythonMapInArrow ... -+- ColumnarToRow - +- CometNativeExec ... - +- CometScan ... +MapInArrow ... ++- CometColumnarToRow + +- CometNativeScan parquet ... ``` When AQE is enabled (the Spark default) and the query contains a shuffle, the @@ -163,7 +159,7 @@ running an action will show the unoptimized plan: ``` AdaptiveSparkPlan isFinalPlan=false -+- PythonMapInArrow ... ++- MapInArrow ... +- CometExchange ... ``` diff --git a/docs/source/user-guide/latest/s3-credential-providers.md b/docs/source/user-guide/latest/s3-credential-providers.md index df3d76005bb..e97e22c4eb7 100644 --- a/docs/source/user-guide/latest/s3-credential-providers.md +++ b/docs/source/user-guide/latest/s3-credential-providers.md @@ -19,7 +19,7 @@ under the License. # S3 Credential Providers -Comet's native S3 readers normally fetch credentials from the standard AWS credential chain (static keys, instance profiles, environment variables, etc.). Some clusters use a vendor-managed mechanism instead, where credentials are issued per request based on a JWT or per S3 path. For those clusters, Comet supports loading a vendor-supplied bridge class that routes every native credential request through the vendor's Java code. +Comet's native S3 readers and native Iceberg writer normally fetch credentials from the standard AWS credential chain (static keys, instance profiles, environment variables, etc.). Some clusters use a vendor-managed mechanism instead, where credentials are issued per request based on a JWT or per S3 path. For those clusters, Comet supports loading a vendor-supplied bridge class that routes every native credential request through the vendor's Java code. ## Do I need this? @@ -102,7 +102,7 @@ Without the config set, no credential-related log lines appear at startup; nativ ## Iceberg: explicit S3 region required -With the bridge configured, Comet wires a custom credential loader into `iceberg-storage-opendal`. `opendal`'s built-in S3 region auto-detection only runs when no custom loader is configured, so on the bridge path the region (and endpoint for non-AWS) must be set explicitly on the Spark catalog: +`iceberg-storage-opendal` does not auto-detect a bucket's region, with or without the bridge. When neither the catalog (`s3.region` or `client.region`) nor the executor environment (`AWS_REGION` / `AWS_DEFAULT_REGION`) supplies a region, Comet uses `us-east-1`. That suits most non-AWS S3-compatible services but fails for AWS buckets in other regions, so set the region (and the endpoint for non-AWS) explicitly on the Spark catalog: ``` spark.sql.catalog..s3.region = us-east-1 @@ -110,8 +110,6 @@ spark.sql.catalog..s3.endpoint = https://... (non-AWS only) spark.sql.catalog..s3.path-style-access = true (path-style endpoints only) ``` -If you hit `region is missing. Please find it by S3::detect_region() or set them in env`, this is the missing config. - ## Writing a bridge Comet's native scan paths (`object_store` for raw Parquet, `opendal` via `iceberg-rust` for Iceberg) bypass Hadoop S3A entirely. The standard `AWSCredentialsProvider.getCredentials()` has no path argument, so vendors that issue per-path STS credentials cannot expose them through it. The `CometS3CredentialProvider` SPI fills that gap. @@ -296,10 +294,10 @@ public final class IcebergRESTVendedS3Provider implements CometS3CredentialProvi ### Access mode -| Value | Used for | -| ------- | -------------------------------------------------------------------------- | -| `READ` | All native scan paths (raw Parquet, Iceberg). Comet today only sends READ. | -| `WRITE` | Reserved for future native write paths. | +| Value | Used for | +| ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `READ` | All native scan paths (raw Parquet, Iceberg). | +| `WRITE` | The native Iceberg writer (see [Iceberg Writes](iceberg-writes.md)). If the configured provider fails to initialize, the write fails rather than falling back to the default credential chain. | A `WRITE` credential is not implicitly read-capable. Vendors that need read-during-write workflows include the required read permissions in the IAM policy attached to their `WRITE` credentials. diff --git a/docs/source/user-guide/latest/scala_java_udfs.md b/docs/source/user-guide/latest/scala_java_udfs.md index 3a55982bbcf..43a30374578 100644 --- a/docs/source/user-guide/latest/scala_java_udfs.md +++ b/docs/source/user-guide/latest/scala_java_udfs.md @@ -34,7 +34,7 @@ This feature is enabled by default. Set `spark.comet.exec.scalaUDF.codegen.enabl ## Supported - User functions registered via `udf(...)`, `spark.udf.register(...)` (Scala or Java functional interfaces), or SQL `CREATE FUNCTION ... AS 'com.example.MyUDF'`. -- Scalar input/output types: `Boolean`, `Byte`, `Short`, `Int`, `Long`, `Float`, `Double`, `Decimal`, `String`, `Binary`, `Date`, `Timestamp`, `TimestampNTZ`. +- Scalar input/output types: `Boolean`, `Byte`, `Short`, `Int`, `Long`, `Float`, `Double`, `Decimal`, `String`, `Binary`, `Date`, `Timestamp`, `TimestampNTZ`, `YearMonthInterval`, `DayTimeInterval`, `CalendarInterval`. - Complex input/output types with arbitrary nesting: `ArrayType`, `StructType`, `MapType`. - Composition with other Catalyst expressions inside the argument tree (e.g. `myUdf(upper(s))` runs as one unit in the Comet pipeline). - Higher-order functions (`transform`, `filter`, `exists`, `aggregate`, `zip_with`, `map_filter`, `map_zip_with`, etc.) inside the argument tree. @@ -45,7 +45,7 @@ This feature is enabled by default. Set `spark.comet.exec.scalaUDF.codegen.enabl - Table UDFs and generators. - Python `@udf` and Pandas `@pandas_udf`. - Hive `GenericUDF` and `SimpleUDF`. -- `CalendarIntervalType`, `NullType`, and `UserDefinedType` arguments and return types. UDT-typed columns fall back to Spark; to keep execution in the Comet pipeline, store and read the underlying representation directly (e.g. write MLlib `Vector` outputs as `Struct, values: Array>` rather than `VectorUDT`). +- `NullType` and `UserDefinedType` arguments and return types. UDT-typed columns fall back to Spark; to keep execution in the Comet pipeline, store and read the underlying representation directly (e.g. write MLlib `Vector` outputs as `Struct, values: Array>` rather than `VectorUDT`). - Trees whose total nested-field count (output plus all input columns the UDF tree references) exceeds `spark.sql.codegen.maxFields` (default 100). Comet refuses these at plan time and the operator falls back to Spark. When a UDF is rejected, the reason surfaces through Comet's standard fallback diagnostics; the query still runs on Spark. diff --git a/docs/source/user-guide/latest/source.md b/docs/source/user-guide/latest/source.md index 841c4e20848..2384cac32d1 100644 --- a/docs/source/user-guide/latest/source.md +++ b/docs/source/user-guide/latest/source.md @@ -26,7 +26,7 @@ It is sometimes preferable to build from source for a specific platform. This documentation is for the current development version of Comet. Published source releases are only available for released versions. -To use this version of Comet, see the following section on building from the GitHubn repository. +To use this version of Comet, see the following section on building from the GitHub repository. diff --git a/docs/source/user-guide/latest/tuning.md b/docs/source/user-guide/latest/tuning.md index 429befc6482..7fcb8ec6109 100644 --- a/docs/source/user-guide/latest/tuning.md +++ b/docs/source/user-guide/latest/tuning.md @@ -23,13 +23,11 @@ Comet provides some tuning options to help you get the best performance from you ## Configuring Tokio Runtime -Comet uses a global tokio runtime per executor process using tokio's defaults of one worker thread per core and a -maximum of 512 blocking threads. These values can be overridden using the environment variables `COMET_WORKER_THREADS` -and `COMET_MAX_BLOCKING_THREADS`. - -It is recommended that `COMET_WORKER_THREADS` be set to the number of executor cores. This may not be necessary -in some environments, such as Kubernetes, where the number of cores allocated to a pod will already be equal to the -number of executor cores. +Comet uses a global tokio runtime per executor process. By default it starts one worker thread per executor core +(`spark.executor.cores`, or the thread count of `local[N]` and `local[*]` masters) and allows up to 512 blocking +threads, which is tokio's default. If `spark.executor.cores` is not set outside local mode, Comet starts a single +worker thread. These values can be overridden using the environment variables `COMET_WORKER_THREADS` and +`COMET_MAX_BLOCKING_THREADS`. ## Adaptive Partial Aggregation @@ -432,13 +430,17 @@ back to Spark for shuffle operations. #### Native Shuffle Comet provides a fully native shuffle implementation, which generally provides the best performance. Native shuffle -supports `HashPartitioning`, `RangePartitioning` and `SinglePartitioning` but currently only supports primitive type -partitioning keys. Columns that are not partitioning keys may contain complex types like maps, structs, and arrays. +supports `HashPartitioning`, `RangePartitioning`, and `SinglePartition`, plus `RoundRobinPartitioning` when enabled +(see [Round-Robin Partitioning](compatibility/operators.md#round-robin-partitioning)). Range partitioning keys must be +scalar types. Hash partitioning keys must be scalar types unless +`spark.comet.shuffle.native.partitioning.hash.nested.enabled=true`, which also admits struct, array, and (Spark 4.0 +and later) map keys. That setting is disabled by default until the performance of the nested hashing paths has been +measured. Columns that are not partitioning keys may contain complex types like maps, structs, and arrays. #### Columnar (JVM) Shuffle Comet Columnar shuffle is JVM-based and supports `HashPartitioning`, `RoundRobinPartitioning`, `RangePartitioning`, and -`SinglePartitioning`. This shuffle implementation supports complex data types as partitioning keys. +`SinglePartition`. This shuffle implementation supports complex data types as partitioning keys. By default, Comet will convert a Spark `ShuffleExchangeExec` to columnar shuffle even when the shuffle's child is a non-Comet (Spark) plan. The benefit is that the next query stage can start as native Comet execution, since the @@ -450,10 +452,11 @@ on Spark. #### Automatic Revert to Spark Shuffle -When a Comet columnar shuffle ends up between two non-Comet operators (for example, a partial/final hash aggregate -pair that Comet could not convert), Comet reverts it to Spark's built-in shuffle. Keeping columnar shuffle between -two row-based operators would add `row -> Arrow -> shuffle -> Arrow -> row` conversions with no Comet consumer on -either side to benefit from columnar output. +When a Comet columnar shuffle ends up between a partial and a final aggregate that Comet could not convert (both +remain Spark `HashAggregateExec` or `ObjectHashAggregateExec` operators), Comet reverts it to Spark's built-in shuffle. +Keeping columnar shuffle between the two row-based aggregates would add `row -> Arrow -> shuffle -> Arrow -> row` +conversions with no Comet consumer on either side to benefit from columnar output. Other shuffles between non-Comet +operators are not reverted. This shifts the affected shuffles from Comet's off-heap memory pool back to the JVM execution memory pool. Clusters tuned for a small JVM heap may see `ExternalSorter` spills on queries where this revert fires. Shuffle I/O may also @@ -464,7 +467,7 @@ Each revert is logged at `INFO` level on the driver as `Reverting Comet columnar This optimization is enabled by default and can be disabled by setting `spark.comet.shuffle.revertRedundantColumnar.enabled=false`, in which case Comet will keep the columnar shuffle -even when both its parent and child are non-Comet operators. +even when both of those aggregates run on Spark. ### Remote Shuffle with Celeborn @@ -562,26 +565,28 @@ repeated columnar-to-row and row-to-columnar conversions that dominate stage run `spark.comet.exec.transitionRevert.enabled=true` to have Comet revert the entire stage to Spark row execution when the number of columnar-to-row transitions exceeds `spark.comet.exec.transitionRevert.maxTransitions` (default `2`). This trades native execution of a small -subset of operators for eliminating conversion overhead across the stage. +subset of operators for eliminating conversion overhead across the stage. A stage is not reverted when it holds a +native aggregate whose intermediate buffer Spark cannot exchange with Comet across a stage boundary, because +reverting it would split that aggregate between the two engines. -### Entire-Plan Fallback for Wide or Deeply Nested Schemas +### Wide or Deeply Nested Schemas The cost of each conversion also grows sharply with schema shape: for wide or deeply nested schemas, columnar-to-row conversion is especially expensive because the conversion work scales with the number of columns and nested fields. If profiling shows these conversions dominating a query over such a schema, set `spark.comet.exec.transitionRevert.enabled=true` and lower -`spark.comet.exec.transitionRevert.maxTransitions` (default `2`) to `1`. Note that this causes the entire -plan to fall back to Spark row-based execution — Comet removes its native operators rather than running a +`spark.comet.exec.transitionRevert.maxTransitions` (default `2`) to `1`. Note that this reverts every stage that +exceeds the threshold to Spark row-based execution — Comet removes the stage's native operators rather than running a mix of native and fallback operators joined by repeated conversions — which can be cheaper than paying the expensive conversions again and again. ## Metrics Overhead -Comet exposes rich native operator metrics for observability (see [Metrics](metrics.md)), but they are -disabled by default because traversing the Spark plan on every task adds measurable overhead, and metrics -require an external sink (for example Prometheus) to be useful. Enable them with -`spark.comet.metrics.enabled=true` when you have a metrics sink configured. This setting must be applied -before the `SparkSession` is created. +The SQL metrics described in [Metrics](metrics.md) are always collected. Setting `spark.comet.metrics.enabled=true` +additionally publishes plan-coverage counters (`operators.native`, `operators.spark`, `queries.planned`, +`transitions`, and `acceleration.ratio`) through Spark's metrics system under the `comet` source. It is disabled by +default because it walks every executed plan on the driver after each query, and the counters are only useful with an +external sink (for example Prometheus) configured. This setting must be applied before the `SparkSession` is created. ## Explain Plan diff --git a/docs/source/user-guide/latest/understanding-comet-plans.md b/docs/source/user-guide/latest/understanding-comet-plans.md index 6d2b3615103..beb5c6ebd14 100644 --- a/docs/source/user-guide/latest/understanding-comet-plans.md +++ b/docs/source/user-guide/latest/understanding-comet-plans.md @@ -301,6 +301,9 @@ by role. Names match what is shown in the plan output. | `CometNativeScan` | Parquet scan that runs entirely in Rust via DataFusion. | | `CometIcebergNativeScan` | Iceberg Parquet scan that runs entirely in Rust via DataFusion. | | `CometCsvNativeScan` | CSV scan that runs entirely in Rust via DataFusion (experimental). | +| `CometInMemoryTableScan` | JVM-side scan of a table cached in Comet's Arrow format (experimental, disabled by default). | +| `CometLocalTableScan` | JVM-side scan of a `LocalTableScanExec` (disabled by default). | +| `CometEmptyRelation` | Empty native input that replaces `EmptyRelationExec` (Spark 4.0 and later). | ### Native Rust Operators @@ -321,20 +324,28 @@ consecutively in a plan, they execute as a single fused block. | `CometBroadcastHashJoin` | `BroadcastHashJoinExec` | | `CometBroadcastNestedLoopJoin` | `BroadcastNestedLoopJoinExec` | | `CometSortMergeJoin` | `SortMergeJoinExec` | -| `CometWindow` | `WindowExec` | +| `CometWindowExec` | `WindowExec` | +| `CometWindowGroupLimitExec` | `WindowGroupLimitExec` (Spark 3.5 and later) | +| `CometSample` | `SampleExec` (sampling without replacement) | | `CometTakeOrderedAndProject` | `TakeOrderedAndProjectExec` | +| `CometWriteFiles` | `WriteFilesExec` (Spark 4.0 and later, experimental native Parquet writes) | +| `CometNativeWrite` | `DataWritingCommandExec` (Spark 3.x, experimental native Parquet writes) | +| `CometIcebergWrite` | `IcebergWrite` (experimental native Iceberg data-file writes) | ### JVM-Side Operators These keep their data on the JVM but participate in the Comet pipeline. -| Node | Notes | -| ------------------------ | ----------------------------------------------------------------------------------- | -| `CometUnion` | JVM-side union of Comet inputs. The Rust side reads each branch as a separate scan. | -| `CometCoalesce` | JVM-side partition coalesce. | -| `CometCollectLimit` | JVM-side collect limit, equivalent to `CollectLimitExec`. | -| `CometBroadcastExchange` | Broadcast exchange producing serialized Arrow batches that the consumer can decode. | -| `CometSubqueryBroadcast` | Companion to `CometBroadcastExchange` for dynamic partition pruning subqueries. | +| Node | Notes | +| ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------- | +| `CometUnion` | JVM-side union of Comet inputs. The Rust side reads each branch as a separate scan. | +| `CometCoalesce` | JVM-side partition coalesce. | +| `CometCollectLimit` | JVM-side collect limit, equivalent to `CollectLimitExec`. | +| `CometBroadcastExchange` | Broadcast exchange producing serialized Arrow batches that the consumer can decode. | +| `CometSubqueryBroadcast` | Companion to `CometBroadcastExchange` for dynamic partition pruning subqueries. | +| `CometMapInBatch` | Runs `mapInArrow` / `mapInPandas` Python UDFs on Comet's Arrow batches (experimental). See [PyArrow UDF Acceleration](pyarrow-udfs.md). | +| `IcebergWrite` | Executor-side Iceberg data-file write in Comet's split-operator Iceberg write plan (experimental). See [Iceberg Writes](iceberg-writes.md). | +| `IcebergCommit` | Driver-side commit for Comet's split-operator Iceberg write plan (experimental). | ### Shuffle Operators @@ -344,21 +355,26 @@ use: - **`CometExchange`** is the **native shuffle** path. The child must already be a Comet operator producing columnar Arrow batches; the node calls `executeColumnar()` on its child and the partition, encode, and compress - steps run in Rust. Hash and range partitioning **keys** must be - primitive types because the Rust hashing and ordering do not support complex - types, but the data columns themselves can include `StructType`, - `ArrayType`, and `MapType` since batches are serialized via the Arrow IPC - writer. + steps run in Rust. Range partitioning **keys** must be scalar types. Hash + partitioning keys must also be scalar types unless + `spark.comet.shuffle.native.partitioning.hash.nested.enabled=true`, which + admits struct, array, and (Spark 4.0 and later) map keys. The data columns + themselves can include `StructType`, `ArrayType`, and `MapType` since + batches are serialized via the Arrow IPC writer. - **`CometColumnarExchange`** is the **JVM columnar shuffle** path. It accepts either Spark row-based input or Comet columnar input, which makes it the fallback when the child is not a Comet operator or when a hash/range key - type is not supported by native shuffle (for example, collated strings). It - is still preferred over Spark's native shuffle when Comet shuffle is - enabled. + type is not supported by native shuffle (for example, a struct or array + hash key while nested hash keys are disabled). It is still preferred over + Spark's native shuffle when Comet shuffle is enabled. Keys with a + non-default string collation are supported by neither path and use Spark's + shuffle. Both paths support the same set of partitioning schemes (`HashPartitioning`, `RangePartitioning`, `RoundRobinPartitioning`, -`SinglePartition`) and both can carry complex types in data columns. +`SinglePartition`) and both can carry complex types in data columns. Native +round-robin partitioning is disabled by default; see +[Round-Robin Partitioning](compatibility/operators.md#round-robin-partitioning). The choice between the two is automatic. See the [Tuning Guide shuffle section](tuning.md#shuffle) for how to enable Comet