From c473dbb2adc37fcd77592ff1227a233c08577b78 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 23 Sep 2026 18:11:15 -0600 Subject: [PATCH 1/2] docs: update the user guide for the 1.1.0 release Sweep of the user guide against everything merged since 1.0.0 branched (af534e0fa), per the release preparation step in the release process. Correct statements that no longer match the code: the CI coverage table (only Spark 4.1 is gated before merge since the non-default profiles moved to the nightly run), the Spark 3.x Parquet limitations Comet has long since closed, cast/JSON/regex/array notes that say "falls back" where Comet now routes through the codegen dispatcher, S3 region auto-detection, Iceberg NOT IN pushdown, tokio thread defaults, shuffle and scan metric names, and a Kubernetes example that pointed at a Java 11 / Spark 3.5 image and left Comet disabled for lack of off-heap memory. Remove known-divergence entries that 1.1.0 fixed (boolean to decimal cast, explicit positive timestamp years, the ABFS object store cache key, most of the ANSI error-message items), and document new 1.1.0 behaviour: the experimental in-memory cache, the native Iceberg writer and the WRITE credential access mode, nested hash shuffle keys, BINARY sort-merge join keys, aggregate spill and memory metrics, native TIME functions, Variant-bearing tables whose Variant columns are not read, and the missing expression and plan-node rows. Add an "Upgrading to Comet 1.1.0" section to the upgrade guide, and drop the release-process notes about tag pushes publishing a Docker image, since that workflow was removed in #4241. --- .../contributor-guide/release_process.md | 6 -- .../expressions/_category_template/cast.md | 34 ++++--- .../expressions/_category_template/map.md | 2 +- .../latest/compatibility/expressions/index.md | 3 + .../expressions/spark-3.4/index.md | 3 + .../user-guide/latest/compatibility/index.md | 29 +----- .../latest/compatibility/operators.md | 28 +++++- .../user-guide/latest/compatibility/scans.md | 2 + .../latest/compatibility/spark-versions.md | 33 +++---- docs/source/user-guide/latest/datasources.md | 12 ++- docs/source/user-guide/latest/datatypes.md | 20 ++-- docs/source/user-guide/latest/expressions.md | 69 +++++++------- .../user-guide/latest/iceberg-writes.md | 28 +++--- docs/source/user-guide/latest/iceberg.md | 21 +++-- docs/source/user-guide/latest/installation.md | 22 +++-- docs/source/user-guide/latest/kubernetes.md | 16 +++- docs/source/user-guide/latest/metrics.md | 92 ++++++++++++------- .../user-guide/latest/migration-guide.md | 7 ++ docs/source/user-guide/latest/operators.md | 44 ++++----- docs/source/user-guide/latest/pyarrow-udfs.md | 30 +++--- .../latest/s3-credential-providers.md | 14 ++- .../user-guide/latest/scala_java_udfs.md | 4 +- docs/source/user-guide/latest/source.md | 2 +- docs/source/user-guide/latest/tuning.md | 59 ++++++------ .../latest/understanding-comet-plans.md | 50 ++++++---- 25 files changed, 352 insertions(+), 278 deletions(-) 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..0f3612d57ea 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,9 @@ 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. - 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 +128,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 +147,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..a4cf3b69e74 100644 --- a/docs/source/user-guide/latest/compatibility/operators.md +++ b/docs/source/user-guide/latest/compatibility/operators.md @@ -32,6 +32,27 @@ 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. Enable it with +`spark.comet.exec.inMemoryCache.enabled=true` before the application starts: the value at startup +decides whether Comet sets `spark.sql.cache.serializer` to its Arrow cache serializer, and +because that is a static config the cache format is fixed for the application. Comet does not +replace a `spark.sql.cache.serializer` that the application has already set. Disabling the +setting later only sends cached scans back to Spark's execution path. + +Relations whose schema Comet's Arrow writer does not support are cached in Spark's default +format, and their scans fall back to Spark. Each cached column is stored as its own compressed +Arrow IPC stream, so a scan decodes only the columns it projects. Reads that feed Spark operators +rather than Comet operators still pay a row conversion that Spark's default format avoids, and +can be slower than Spark's cache. + +With `spark.kryo.registrationRequired=true`, also set +`spark.kryo.registrator=org.apache.comet.CometKryoRegistrator` before creating the +`SparkContext`. Otherwise caching fails as soon as a block is serialized, including the disk +half of the default `MEMORY_AND_DISK` storage level. + ## Sampling Comet runs `SampleExec` natively when sampling is performed without replacement, which covers @@ -95,8 +116,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 36d245992eb..f508098019f 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 c2b942961c2..467ae830b3a 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,15 +97,15 @@ functions, and hashing a `CalendarInterval`. 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. | ## Variant -| Type | Status | Notes | -| ------------- | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `VariantType` | 🔜 | Spark 4.0+. Native scan support is tracked by [#4295](https://github.com/apache/datafusion-comet/issues/4295); shredded Parquet read/write by [#3983](https://github.com/apache/datafusion-comet/issues/3983). | +| Type | Status | Notes | +| ------------- | ------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `VariantType` | 🔜 | Spark 4.0+. Reading a Variant column falls back, whether Spark requests it as `VariantType` or, with `spark.sql.variant.pushVariantIntoScan` (enabled by default on Spark 4.1+), as a shredded struct. Parquet and Iceberg scans of tables whose Variant columns the query does not read run natively. Native scan support is tracked by [#4295](https://github.com/apache/datafusion-comet/issues/4295); shredded Parquet read/write by [#3983](https://github.com/apache/datafusion-comet/issues/3983). | ## Other 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 2ffcd08d05d..41bfa37378e 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 @@ -79,7 +79,7 @@ spark.sql.catalog..warehouse=... # Split-operator plan (experimental, off by default) spark.comet.write.iceberg.splitOperator.enabled=true -# Native-write eligibility detection (experimental, off by default; requires the split plan) +# Native iceberg-rust data-file writer (experimental, off by default; requires the split plan) spark.comet.iceberg.write.enabled=true ``` @@ -123,7 +123,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 @@ -132,7 +132,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`. **Most Iceberg write settings are not supported.** Detection is an allowlist: a write is eligible only when its entire effective configuration matches the table below, and anything @@ -166,7 +168,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`. @@ -271,14 +274,13 @@ a data file but not what any reader computes from it: (iceberg-java names files `---`; iceberg-rust uses a process-local counter). - Partition directory names match iceberg-java 1.8+'s `PartitionSpec.partitionToPath` for every - partition type except `float` and `double`, where the value is rendered with Rust's shortest - representation instead of `Float.toString` / `Double.toString` (`f=1` where iceberg-java writes - `f=1.0`). On Iceberg 1.5.x, which the Spark 3.4 profile pins, iceberg-java itself spelled - `timestamp` and `timestamptz` directories with `LocalDateTime.toString()` / - `OffsetDateTime.toString()` (`ts=1969-12-31T23:59:58.500Z`) and left the partition field name - unescaped; Comet uses the 1.8+ spelling on every profile. Distinct partition values still get - distinct directories in all cases, and no reader parses these names — files are resolved through - committed manifests. Iceberg deprecated float and double partitioning in 1.3. + partition type, including `float` and `double`, which follow `Float.toString` / + `Double.toString` (`f=1.0`, `d=1.0E20`). On Iceberg 1.5.x, which the Spark 3.4 profile pins, + iceberg-java itself spelled `timestamp` and `timestamptz` directories with + `LocalDateTime.toString()` / `OffsetDateTime.toString()` (`ts=1969-12-31T23:59:58.500Z`) and left + the partition field name unescaped; Comet uses the 1.8+ spelling on every profile. Distinct + partition values still get distinct directories in all cases, and no reader parses these names — + files are resolved through committed manifests. - File rolling lands on the same row grid as iceberg-java but not necessarily on the same row. Both writers re-check the current file's size against `write.target-file-size-bytes` once every 1000 rows of that file (iceberg-java's `RollingFileWriter.ROWS_DIVISOR`; Comet hands the diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index 2fed6d35034..f5f3c5202a5 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). `client.region` is not auto-detected: when neither it nor `AWS_REGION` is set, 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,16 +180,23 @@ 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) -- Iceberg writes (reads are accelerated, writes use Spark) - 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 transforms falls back) +Writes are not accelerated by default. Comet has an experimental, opt-in native writer for +Iceberg V1 and V2 Parquet tables (`spark.comet.write.iceberg.splitOperator.enabled=true` and +`spark.comet.iceberg.write.enabled=true`); see [Iceberg Writes](iceberg-writes.md). + ### Iceberg UDFs Iceberg ships several `ScalaUDF`s that surface in user queries and maintenance actions: @@ -243,6 +252,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/installation.md b/docs/source/user-guide/latest/installation.md index 9ed0e6bfe3e..c32eacaf672 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. diff --git a/docs/source/user-guide/latest/kubernetes.md b/docs/source/user-guide/latest/kubernetes.md index b91bb93e803..73393dec368 100644 --- a/docs/source/user-guide/latest/kubernetes.md +++ b/docs/source/user-guide/latest/kubernetes.md @@ -21,13 +21,16 @@ ## Comet Docker Images -Run the following command from the root of this repository to build the Comet Docker image, or use a [published -Docker image](https://hub.docker.com/r/apache/datafusion-comet). +Run the following command from the root of this repository to build the Comet Docker image. The image is based on +`apache/spark:4.1.3` and adds the Comet JAR for Spark 4.1 and Scala 2.13 to `$SPARK_HOME/jars`. ```shell docker build -t apache/datafusion-comet -f kube/Dockerfile . ``` +Push the image to a registry that your cluster can pull from, and use that image name in the `image` field of the +examples below. + ## Example Spark Submit The exact syntax will vary depending on the Kubernetes distribution, but an example `spark-submit` command can be @@ -69,14 +72,17 @@ metadata: spec: type: Scala mode: cluster - image: apache/datafusion-comet:$COMET_VERSION-spark3.5.5-scala2.12-java11 + image: apache/datafusion-comet imagePullPolicy: IfNotPresent mainClass: org.apache.spark.examples.SparkPi mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.13-4.1.3.jar sparkConf: - "spark.executor.extraClassPath": "/opt/spark/jars/comet-spark-spark3.5_2.12-$COMET_VERSION.jar" - "spark.driver.extraClassPath": "/opt/spark/jars/comet-spark-spark3.5_2.12-$COMET_VERSION.jar" + "spark.executor.extraClassPath": "/opt/spark/jars/comet-spark-spark4.1_2.13-$COMET_VERSION.jar" + "spark.driver.extraClassPath": "/opt/spark/jars/comet-spark-spark4.1_2.13-$COMET_VERSION.jar" "spark.plugins": "org.apache.spark.CometPlugin" + "spark.memory.offHeap.enabled": "true" + "spark.memory.offHeap.size": "1g" + "spark.executor.memoryOverhead": "1g" "spark.comet.enabled": "true" "spark.comet.exec.enabled": "true" "spark.comet.shuffle.enabled": "true" diff --git a/docs/source/user-guide/latest/metrics.md b/docs/source/user-guide/latest/metrics.md index 5404344e987..ea10b563a5d 100644 --- a/docs/source/user-guide/latest/metrics.md +++ b/docs/source/user-guide/latest/metrics.md @@ -23,11 +23,14 @@ under the License. Comet operators report the following metrics in the Spark SQL UI. -### CometScanExec +### CometBatchScan -| Metric | Description | -| ----------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `scan time` | Total time to scan a Parquet file. This is not comparable to the same metric in Spark because Comet's scan metric is more accurate. Although both Comet and Spark measure the time in nanoseconds, Spark rounds this time to the nearest millisecond per batch and Comet does not. | +| Metric | Description | +| ----------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `scan time` | Time spent reading batches from the wrapped DataSource V2 reader. Comet measures this time in nanoseconds and does not round it to the nearest millisecond per batch. | + +Parquet scans through the DataSource V1 API appear in plans as `CometNativeScan` and report the +native metrics described under [Native Parquet scans](#native-parquet-scans) rather than `scan time`. ### CometIcebergNativeScan @@ -36,12 +39,12 @@ during execution; the planning metrics are Iceberg's own scan-report counters, c Iceberg's Java planner on the driver and surfaced here so they show in the UI as they do for a plain Spark + Iceberg `BatchScan`. -| Metric | Description | -| --------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| `number of output rows` | Rows produced by the scan. | -| `number of bytes scanned` | Bytes read from storage, including data and delete files. | -| `number of file splits processed` | File scan tasks (splits) read by this scan. | -| `scan time` | Time spent in the native scan's record-batch polling, covering the iceberg-rust reader plus Comet's schema adaptation. It excludes time the stream spends waiting between polls, so it is decode/compute time, not end-to-end scan latency. This differs from the `scan time` under `CometScanExec`, which times Parquet file reads. | +| Metric | Description | +| --------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `number of output rows` | Rows produced by the scan. | +| `number of bytes scanned` | Bytes read from storage, including data and delete files. | +| `number of file splits processed` | File scan tasks (splits) read by this scan. | +| `scan time` | Time spent in the native scan's record-batch polling, covering the iceberg-rust reader plus Comet's schema adaptation. It excludes time the stream spends waiting between polls, so it is decode/compute time, not end-to-end scan latency. This differs from the `scan time` under `CometBatchScan`, which times batch reads from a DataSource V2 reader. | The planning metrics below mirror Iceberg's `ScanReport`. They are driver-side values known after scan planning and do not change during execution. @@ -69,6 +72,21 @@ Iceberg's `numDeletes` (deletes applied at read time) is not reported: it is a J counter, and Comet reads natively through iceberg-rust, which exposes no deletes-applied count, so the value would always be 0. +### CometHashAggregate + +Native aggregates with grouping keys report these additional metrics: + +| Metric | Description | +| ------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------- | +| `rows bypassing partial aggregation` | Input rows passed through without partial aggregation. See [Adaptive Partial Aggregation](tuning.md#adaptive-partial-aggregation). | +| `number of spills` | Number of times the aggregate spilled to disk. | +| `total spilled bytes` | Bytes written to aggregate spill files. | +| `number of spilled rows` | Rows written to aggregate spill files. | +| `peak native aggregate memory` | Peak memory used by the native aggregate. | + +Spill bytes from native sorts, aggregates, and sort-merge joins are also added to Spark's task-level +`diskBytesSpilled` metric in every stage, not only in shuffle stages. + ### Hash Joins With `spark.comet.exec.join.dynamicFilter.enabled=true`, native broadcast and shuffled hash joins @@ -93,15 +111,20 @@ reader savings. Existing join, scan, and intervening filter metrics retain their Comet adds some additional metrics: -| Metric | Description | -| ------------------------------- | --------------------------------------------------------------------------- | -| `native shuffle time` | Total time in native code excluding any child operators. | -| `repartition time` | Time to repartition batches. | -| `partition interleaving time` | Time to interleave partitioned batches before writing them. | -| `memory pool time` | Time interacting with memory pool. | -| `encoding and compression time` | Time to encode batches in IPC format and compress using ZSTD. | -| `disk spilled bytes` | Actual bytes written to native shuffle spill files on disk. | -| `memory spilled bytes` | Uncompressed Arrow backing-buffer and partition-index data before spilling. | +| Metric | Description | +| --------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | +| `native shuffle writer time` | Total time in the native shuffle writer, excluding any child operators. | +| `repartition time` | Time to repartition batches. | +| `partition interleaving time` | Time to interleave partitioned batches before writing them. | +| `encoding and compression time` | Time to encode batches in Arrow IPC format and compress them with the configured codec (`spark.comet.shuffle.compression.codec`, default `lz4`). | +| `decoding and decompression time` | Time to decompress and decode shuffle blocks when they are read. | +| `number of spills` | Number of native shuffle spills. | +| `disk spilled bytes` | Actual bytes written to native shuffle spill files on disk. | +| `memory spilled bytes` | Uncompressed Arrow backing-buffer and partition-index data before spilling. | +| `number of input batches` | Batches received by the native shuffle writer. | + +Comet exchanges also report Spark's standard shuffle read and write metrics, including when native +operators read shuffle blocks directly. Disk and memory spilled bytes measure different representations of the same native shuffle spill. Disk spill bytes count the actual bytes written to disk: compressed when shuffle compression is @@ -127,25 +150,24 @@ Here is a guide to some of the native metrics. ### ScanExec -| Metric | Description | -| ----------------- | --------------------------------------------------------------------------------------------------- | -| `elapsed_compute` | Total time spent in this operator, fetching batches from a JVM iterator. | -| `jvm_fetch_time` | Time spent in the JVM fetching input batches to be read by this `ScanExec` instance. | -| `arrow_ffi_time` | Time spent using Arrow FFI to create Arrow batches from the memory addresses returned from the JVM. | +| Metric | Description | +| ----------------- | ------------------------------------------------------------------------ | +| `elapsed_compute` | Total time spent in this operator, fetching batches from a JVM iterator. | +| `cast_time` | Time spent casting columns to the requested data types during the scan. | ### ShuffleWriterExec -| Metric | Description | -| ---------------------- | --------------------------------------------------------------------- | -| `elapsed_compute` | Total time excluding any child operators. | -| `repart_time` | Time to repartition batches. | -| `interleave_time` | Time to interleave partitioned batches before writing them. | -| `ipc_time` | Time to encode batches in IPC format and compress using ZSTD. | -| `mempool_time` | Time interacting with memory pool. | -| `write_time` | Time spent writing bytes to disk. | -| `spill_count` | Number of native shuffle spills. | -| `spilled_bytes` | Actual bytes written to native shuffle spill files on disk. | -| `memory_spilled_bytes` | Uncompressed Arrow backing-buffer and partition-index memory spilled. | +| Metric | Description | +| ---------------------- | --------------------------------------------------------------------------------------- | +| `elapsed_compute` | Total time excluding any child operators. | +| `repart_time` | Time to repartition batches. | +| `interleave_time` | Time to interleave partitioned batches before writing them. | +| `encode_time` | Time to encode batches in Arrow IPC format and compress them with the configured codec. | +| `write_time` | Time spent writing encoded data to its destination. | +| `input_batches` | Number of input batches. | +| `spill_count` | Number of native shuffle spills. | +| `spilled_bytes` | Actual bytes written to native shuffle spill files on disk. | +| `memory_spilled_bytes` | Uncompressed Arrow backing-buffer and partition-index memory spilled. | ### Native Parquet scans diff --git a/docs/source/user-guide/latest/migration-guide.md b/docs/source/user-guide/latest/migration-guide.md index f16ad826067..1de5f884672 100644 --- a/docs/source/user-guide/latest/migration-guide.md +++ b/docs/source/user-guide/latest/migration-guide.md @@ -53,6 +53,13 @@ Treat setting one of these keys as a temporary measure. If you find you cannot s legacy behavior, please open an issue describing your use case so it can be considered before the key is removed. +## Upgrading to Comet 1.1.0 + +Comet `1.1.0` makes no behavior changes that need a `spark.comet.legacy.*` key. + +Comet `1.1.0` requires JDK 17 or later. JDK 11 is no longer supported. See +[Installing Comet](installation.md) for the supported Java, Scala, and Spark versions. + ## Upgrading to Comet 1.0.0 Comet `1.0.0` is the first release under the stable diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md index 2182e7de584..17dd637e1dd 100644 --- a/docs/source/user-guide/latest/operators.md +++ b/docs/source/user-guide/latest/operators.md @@ -49,13 +49,13 @@ omitted from the tables below and may be reconsidered based on demand: ## Scans -| Operator | Status | Notes | -| ----------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| `FileSourceScanExec` | ✅ | Parquet only. Some types and configurations fall back. See [Parquet Scan Compatibility](compatibility/scans.md). | -| `BatchScanExec` | ✅ | Parquet, Apache Iceberg Parquet, and CSV (native) scans. See [Parquet Scan Compatibility](compatibility/scans.md) and the [Iceberg Guide](iceberg.md). | -| `LocalTableScanExec` | ⚠️ | Disabled by default; there is no acceleration advantage and this operator is typically only used in test code. Can be opted into via config ([#4393](https://github.com/apache/datafusion-comet/pull/4393)). | -| `EmptyRelationExec` | ✅ | Spark 4.0 and later. See [Empty Relations](compatibility/operators.md#empty-relations) for native-input support and writer fallback. | -| `InMemoryTableScanExec` | 🔜 | Cached / in-memory table scans fall back today. | +| Operator | Status | Notes | +| ----------------------- | ------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `FileSourceScanExec` | ✅ | Parquet only. Some types and configurations fall back. See [Parquet Scan Compatibility](compatibility/scans.md). | +| `BatchScanExec` | ✅ | Apache Iceberg Parquet scans run natively. Native CSV scans are experimental and disabled by default. DataSource V2 Parquet scans are not accelerated. See [Parquet Scan Compatibility](compatibility/scans.md) and the [Iceberg Guide](iceberg.md). | +| `LocalTableScanExec` | ⚠️ | Disabled by default; there is no acceleration advantage and this operator is typically only used in test code. Can be opted into via config ([#4393](https://github.com/apache/datafusion-comet/pull/4393)). | +| `EmptyRelationExec` | ✅ | Spark 4.0 and later. See [Empty Relations](compatibility/operators.md#empty-relations) for native-input support and writer fallback. | +| `InMemoryTableScanExec` | ⚠️ | Experimental, disabled by default. Set `spark.comet.exec.inMemoryCache.enabled=true` before the application starts so Comet installs its Arrow cache serializer. Relations with unsupported column types stay in Spark's cache format and fall back. See [In-Memory Cache](compatibility/operators.md#in-memory-cache). | ## Projection and filtering @@ -77,11 +77,11 @@ omitted from the tables below and may be reconsidered based on demand: ## Aggregation -| Operator | Status | Notes | -| ------------------------- | ------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `HashAggregateExec` | ✅ | | -| `ObjectHashAggregateExec` | ✅ | Supports a limited set of aggregates, such as `bloom_filter_agg`. Falls back when Comet shuffle is disabled, which would otherwise split the aggregate across Comet and Spark. See the [Tuning Guide](tuning.md). | -| `SortAggregateExec` | 🔜 | Falls back today; Comet currently accelerates hash aggregates. | +| Operator | Status | Notes | +| ------------------------- | ------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `HashAggregateExec` | ✅ | | +| `ObjectHashAggregateExec` | ✅ | Runs the object-buffer aggregates Comet supports, such as `collect_list`, `collect_set`, `percentile`, `approx_percentile`, `mode`, `bloom_filter_agg`, and (Spark 4.0+) `listagg`. Falls back when Comet shuffle is disabled, which would otherwise split the aggregate across Comet and Spark. See [Shuffle](tuning.md#shuffle). | +| `SortAggregateExec` | 🔜 | Falls back today; Comet currently accelerates hash aggregates. | ## Joins @@ -89,7 +89,7 @@ omitted from the tables below and may be reconsidered based on demand: | ----------------------------- | ------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `BroadcastHashJoinExec` | ✅ | | | `ShuffledHashJoinExec` | ✅ | | -| `SortMergeJoinExec` | ✅ | | +| `SortMergeJoinExec` | ✅ | Supports `BINARY` join keys. Nested-type (struct, array, map) and collated-string join keys fall back to Spark. | | `BroadcastNestedLoopJoinExec` | ✅ | Falls back to Spark when the preserved side is broadcast (for example LEFT OUTER with BROADCAST on the left) ([#4429](https://github.com/apache/datafusion-comet/pull/4429)). | ## Exchanges @@ -104,7 +104,7 @@ omitted from the tables below and may be reconsidered based on demand: | Operator | Status | Notes | | ---------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `WindowExec` | ⚠️ | Runs natively and is enabled by default. A broad set of window functions is accelerated; unsupported shapes fall back to Spark. See [window function compatibility](compatibility/operators.md). | -| `WindowGroupLimitExec` | ✅ | Streaming per-partition top-K pushdown for `ROW_NUMBER`, `RANK`, and `DENSE_RANK`. | +| `WindowGroupLimitExec` | ✅ | Spark 3.5 and later. Streaming per-partition top-K pushdown for `ROW_NUMBER`, `RANK`, and `DENSE_RANK`. | ## Generators and set operations @@ -117,16 +117,18 @@ omitted from the tables below and may be reconsidered based on demand: ## Writes -| Operator | Status | Notes | -| ------------------------ | ------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `WriteFilesExec` | ⚠️ | Spark 4.0+. Experimental native Parquet writes, disabled by default (opt-in). Non-partitioned, non-bucketed writes only, and not when `spark.sql.files.maxRecordsPerFile` is set. | -| `DataWritingCommandExec` | ⚠️ | Spark 3.4/3.5 only. Experimental native Parquet writes, disabled by default (opt-in). Replaced by `WriteFilesExec` on Spark 4.0+ and removed with Spark 3.x support. | +| Operator | Status | Notes | +| -------------------------------------------------------------------------------------------------- | ------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `WriteFilesExec` | ⚠️ | Spark 4.0+. Experimental native Parquet writes, disabled by default (opt-in). Non-partitioned, non-bucketed writes only, and not when `spark.sql.files.maxRecordsPerFile` is set. | +| `DataWritingCommandExec` | ⚠️ | Spark 3.4/3.5 only. Experimental native Parquet writes, disabled by default (opt-in). Replaced by `WriteFilesExec` on Spark 4.0+ and removed with Spark 3.x support. | +| `AppendDataExec`, `OverwriteByExpressionExec`, `OverwritePartitionsDynamicExec`, `ReplaceDataExec` | ⚠️ | Apache Iceberg tables only. Experimental, disabled by default. See [Iceberg Writes](iceberg-writes.md). | ## Python and UDF -| Operator | Status | Notes | -| --------------------------------------------------------------------------------------- | ------ | ---------------------------------------------------------------------------------------------------------------------------- | -| `ArrowEvalPythonExec`, `MapInArrowExec`, `MapInPandasExec`, `FlatMapGroupsInPandasExec` | 🔜 | Experimental accelerated PyArrow UDF support is in progress ([#4234](https://github.com/apache/datafusion-comet/pull/4234)). | +| Operator | Status | Notes | +| -------------------------------------------------- | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `MapInArrowExec`, `MapInPandasExec` | ⚠️ | Spark 4.0 and later. Experimental, disabled by default (`spark.comet.exec.pyarrowUDF.enabled`). See [PyArrow UDF Acceleration](pyarrow-udfs.md). | +| `ArrowEvalPythonExec`, `FlatMapGroupsInPandasExec` | 🔜 | Scalar `@pandas_udf` ([#5386](https://github.com/apache/datafusion-comet/issues/5386)) and grouped `applyInPandas` ([#5123](https://github.com/apache/datafusion-comet/issues/5123)) fall back to Spark. | ## See also diff --git a/docs/source/user-guide/latest/pyarrow-udfs.md b/docs/source/user-guide/latest/pyarrow-udfs.md index 5aab9a61b4b..eb704deb6f6 100644 --- a/docs/source/user-guide/latest/pyarrow-udfs.md +++ b/docs/source/user-guide/latest/pyarrow-udfs.md @@ -30,7 +30,7 @@ using the Arrow IPC format. Without Comet, the execution path for these UDFs involves unnecessary data conversions: -1. Comet reads data in Arrow columnar format (via CometScan) +1. Comet reads data in Arrow columnar format (via `CometNativeScan`) 2. Spark inserts a ColumnarToRow transition (converts Arrow to UnsafeRow) 3. The Python runner converts those rows back to Arrow to send to Python 4. Python executes the UDF on Arrow batches @@ -40,8 +40,8 @@ Steps 2 and 3 are redundant since the data starts and ends in Arrow format. ## How Comet Optimizes This -When enabled, Comet detects `PythonMapInArrowExec` / `MapInArrowExec` and `MapInPandasExec` -operators in the physical plan and replaces them with `CometMapInBatchExec`, which: +When enabled, Comet detects `MapInArrowExec` and `MapInPandasExec` operators in the physical plan +and replaces them with `CometMapInBatchExec` (shown as `CometMapInBatch` in plans), which: - Reads Arrow columnar batches directly from the upstream Comet operator - Feeds them to the Python runner without the expensive UnsafeProjection copy @@ -58,18 +58,16 @@ copies that remain. Without Comet's optimization: ``` -PythonMapInArrow / MapInArrow / MapInPandas -+- ColumnarToRow <- Arrow -> 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 33777c497af..4c96488ca93 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. @@ -267,10 +265,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 84a3a16edec..b232284b59c 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 @@ -122,10 +120,8 @@ The valid pool types are: - `fair_unified` (default when `spark.memory.offHeap.enabled=true` is set) - `greedy_unified` -Both pool types are shared across all native execution contexts within the same Spark task. When -Comet executes a shuffle, it runs two native execution contexts concurrently (e.g. one for -pre-shuffle operators and one for the shuffle writer). The shared pool ensures that the combined -memory usage stays within the per-task limit. +Both pool types are shared by every native plan that runs within the same Spark task, so their +combined memory usage stays within the per-task limit. The `fair_unified` pool prevents operators from using more than an even fraction of the available memory (i.e. `pool_size / num_reservations`). This pool works best when you know beforehand @@ -356,13 +352,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 @@ -374,10 +374,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 @@ -388,7 +389,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 @@ -486,26 +487,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 32fe297e67d..18f73cb534f 100644 --- a/docs/source/user-guide/latest/understanding-comet-plans.md +++ b/docs/source/user-guide/latest/understanding-comet-plans.md @@ -276,6 +276,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 @@ -296,20 +299,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 @@ -319,21 +330,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 From d531350c67df88f96e24f0724b0535429b083de3 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 25 Sep 2026 06:23:36 -0600 Subject: [PATCH 2/2] docs: address review feedback on the 1.1.0 user guide Document the Kryo registrator under "Additional Configuration" in the installation guide. Native broadcast needs it with spark.kryo.registrationRequired=true whether or not the in-memory cache is enabled, so link to it from the cache docs rather than describing it as a cache-only setting. Point the operator compatibility cache section and the operators table at the in-memory cache page that landed on main, instead of repeating it, which also drops a stale storage-format claim. List every source Comet checks for the Iceberg S3 region, and link the ANSI integral overflow divergence to #6217. --- .../user-guide/latest/compatibility/index.md | 3 ++- .../latest/compatibility/operators.md | 25 ++++++------------- docs/source/user-guide/latest/iceberg.md | 2 +- .../user-guide/latest/in-memory-cache.md | 5 ++-- docs/source/user-guide/latest/installation.md | 17 +++++++++++++ docs/source/user-guide/latest/operators.md | 14 +++++------ 6 files changed, 38 insertions(+), 28 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index 0f3612d57ea..36cd8cf2b86 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -114,7 +114,8 @@ divergence: - 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. + `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 diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md index a4cf3b69e74..cd779dc081a 100644 --- a/docs/source/user-guide/latest/compatibility/operators.md +++ b/docs/source/user-guide/latest/compatibility/operators.md @@ -35,23 +35,14 @@ 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. Enable it with -`spark.comet.exec.inMemoryCache.enabled=true` before the application starts: the value at startup -decides whether Comet sets `spark.sql.cache.serializer` to its Arrow cache serializer, and -because that is a static config the cache format is fixed for the application. Comet does not -replace a `spark.sql.cache.serializer` that the application has already set. Disabling the -setting later only sends cached scans back to Spark's execution path. - -Relations whose schema Comet's Arrow writer does not support are cached in Spark's default -format, and their scans fall back to Spark. Each cached column is stored as its own compressed -Arrow IPC stream, so a scan decodes only the columns it projects. Reads that feed Spark operators -rather than Comet operators still pay a row conversion that Spark's default format avoids, and -can be slower than Spark's cache. - -With `spark.kryo.registrationRequired=true`, also set -`spark.kryo.registrator=org.apache.comet.CometKryoRegistrator` before creating the -`SparkContext`. Otherwise caching fails as soon as a block is serialized, including the disk -half of the default `MEMORY_AND_DISK` storage level. +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 diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index 93f0e97fa1a..a0a0b521693 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -172,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 not auto-detected: when neither it nor `AWS_REGION` is set, 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. +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 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 c32eacaf672..3dca6dfb65b 100644 --- a/docs/source/user-guide/latest/installation.md +++ b/docs/source/user-guide/latest/installation.md @@ -261,3 +261,20 @@ Some cluster managers may require additional configuration, see