diff --git a/dev/ci/compute-changes.py b/dev/ci/compute-changes.py index 0e44fd1ed15..42d4b859a37 100644 --- a/dev/ci/compute-changes.py +++ b/dev/ci/compute-changes.py @@ -156,7 +156,6 @@ "spark/src/main/scala/org/apache/comet/vector/**", "spark/src/main/scala/org/apache/spark/sql/comet/CometMapInBatchExec.scala", "spark/src/main/scala/org/apache/spark/sql/comet/shims/MapInBatchInfo.scala", - "spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala", "spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala", "spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala", "spark/src/main/spark-4.0/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala", @@ -205,7 +204,6 @@ "common/src/main/**", "common/pom.xml", "spark/src/main/**", - "!spark/src/main/spark-3.4/**", "!spark/src/main/spark-4.0/**", "!spark/src/main/spark-4.1/**", "!spark/src/main/spark-4.2/**", @@ -233,7 +231,6 @@ "common/src/main/**", "common/pom.xml", "spark/src/main/**", - "!spark/src/main/spark-3.4/**", "!spark/src/main/spark-3.5/**", "!spark/src/main/spark-3.x/**", "!spark/src/main/spark-4.1/**", @@ -261,7 +258,6 @@ "common/src/main/**", "common/pom.xml", "spark/src/main/**", - "!spark/src/main/spark-3.4/**", "!spark/src/main/spark-3.5/**", "!spark/src/main/spark-3.x/**", "!spark/src/main/spark-4.0/**", @@ -289,7 +285,6 @@ "common/src/main/**", "common/pom.xml", "spark/src/main/**", - "!spark/src/main/spark-3.4/**", "!spark/src/main/spark-3.5/**", "!spark/src/main/spark-3.x/**", "!spark/src/main/spark-4.0/**", diff --git a/docs/source/about/faq.md b/docs/source/about/faq.md index a06435e2c3b..9937e1ea277 100644 --- a/docs/source/about/faq.md +++ b/docs/source/about/faq.md @@ -90,7 +90,7 @@ for one platform only. ### Why do I get a `ClassNotFoundException` for `CometShuffleManager`? -Spark could not find the Comet jar when it created the shuffle manager. On Spark 3.4 and 3.5, +Spark could not find the Comet jar when it created the shuffle manager. On Spark 3.5, executors create the shuffle manager before they load jars passed with `--jars` or `--packages`, so the jar must be on the startup classpath. Either copy it into `$SPARK_HOME/jars` on every machine, or set `spark.driver.extraClassPath` and `spark.executor.extraClassPath` to its path, as the diff --git a/docs/source/contributor-guide/adding_a_new_expression.md b/docs/source/contributor-guide/adding_a_new_expression.md index d699db08845..a2841f396ff 100644 --- a/docs/source/contributor-guide/adding_a_new_expression.md +++ b/docs/source/contributor-guide/adding_a_new_expression.md @@ -291,7 +291,7 @@ In addition to `getSupportLevel`, which governs runtime planning decisions, the - `getIncompatibleReasons(): Seq[String]` - Reasons the expression may produce different results than Spark. - `getUnsupportedReasons(): Seq[String]` - Reasons the expression, or certain usages of it, may not be supported by Comet. -These methods do not affect runtime behavior. They are called by `GenerateDocs` (`spark/src/main/scala/org/apache/comet/GenerateDocs.scala`) when building the user-facing Compatibility Guide pages. The Markdown templates live in `docs/source/user-guide/latest/compatibility/expressions/_category_template/` (for example, `math.md`, `datetime.md`, `array.md`, `aggregate.md`, `struct.md`); `docs/build.sh` copies them into per-Spark-version subdirectories (`spark-3.4/`, `spark-3.5/`, `spark-4.0/`, `spark-4.1/`) and runs `GenerateDocs` once per profile, so the published page reflects the expressions registered for that Spark version. Each reason is rendered as a bullet in the corresponding page. +These methods do not affect runtime behavior. They are called by `GenerateDocs` (`spark/src/main/scala/org/apache/comet/GenerateDocs.scala`) when building the user-facing Compatibility Guide pages. The Markdown templates live in `docs/source/user-guide/latest/compatibility/expressions/_category_template/` (for example, `math.md`, `datetime.md`, `array.md`, `aggregate.md`, `struct.md`); `docs/build.sh` copies them into per-Spark-version subdirectories (`spark-3.5/`, `spark-4.0/`, `spark-4.1/`) and runs `GenerateDocs` once per profile, so the published page reflects the expressions registered for that Spark version. Each reason is rendered as a bullet in the corresponding page. `GenerateDocs` also rewrites the **Implementation** column of every table in [`docs/source/user-guide/latest/expressions.md`](../user-guide/latest/expressions.md) based on the serde's trait mixins: `NativeOptInAvailable` (or its subtype `CodegenDispatchFallback`) is reported as `Hybrid`, a plain `CometCodegenDispatch` is reported as `Codegen dispatch`, and anything else is reported as `Native`. Rows whose function name is not registered in Spark's `FunctionRegistry` (or has no matching serde) get an em-dash placeholder. diff --git a/docs/source/contributor-guide/iceberg-spark-tests.md b/docs/source/contributor-guide/iceberg-spark-tests.md index baa887defbd..d36b75bd8a6 100644 --- a/docs/source/contributor-guide/iceberg-spark-tests.md +++ b/docs/source/contributor-guide/iceberg-spark-tests.md @@ -181,8 +181,7 @@ records one of three writers: - `jvm`: Comet's split operator planned the write but kept Iceberg's JVM writer (`IcebergWriteExec`). The line includes the reasons Comet recorded for not converting it. - `spark`: Spark's own V2 write operator ran the write, so Comet's split operator never saw it. Examples - are `WriteToDataSourceV2` for a streaming micro-batch, and on Spark 3.4 both `WriteDelta` for - merge-on-read and the CTAS and RTAS execs, which write the table themselves. + are `WriteToDataSourceV2` for a streaming micro-batch. `dev/ci/summarize-iceberg-writes.py` turns these records into a table on the job's summary page. It shows the count and share of each writer, the most common fallback reasons, and the Spark write diff --git a/docs/source/contributor-guide/iceberg-writes.md b/docs/source/contributor-guide/iceberg-writes.md index 01da96ff8ef..0fad3c35370 100644 --- a/docs/source/contributor-guide/iceberg-writes.md +++ b/docs/source/contributor-guide/iceberg-writes.md @@ -74,13 +74,13 @@ IcebergCommit driver: collect task commit messages, BatchWrite.c +- scans, projects, exchanges, sorts; inside AQE with or without the split ``` -| Component | Location | Role | -| ------------------------------------------------------------------------------------------------------------------------------------ | -------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `IcebergWriteStrategy` | `spark/src/main/scala/org/apache/comet/iceberg/` | Planner strategy. Matches `AppendData`, `OverwriteByExpression`, `OverwritePartitionsDynamic`, `ReplaceData` (and Iceberg's own `ReplaceIcebergData`, which Iceberg 1.5.2 plans on Spark 3.4), plus Spark 3.5+ Iceberg `WriteDelta` under the testing split flag. | -| `IcebergWriteLogical` | same | Logical anchor for the writer, so AQE re-plans re-emit only the writer and not a second committer. | -| `IcebergWriteExec` | `spark/src/main/scala/org/apache/spark/sql/comet/` | JVM writer. Runs iceberg-java's `DataWriter` or, for `WriteDelta`, `DeltaWriter` per task and returns the serialized `WriterCommitMessage` as one binary row. | -| `IcebergCommitExec` | same | Driver committer. A `V2CommandExec`, so `run()` is memoized and the commit happens once. | -| `IcebergReplaceDataShim`, `IcebergDeltaLogicalShim`, `IcebergDeltaWriterShim`, `IcebergRefreshCacheShim`, `IcebergDriverMetricsShim` | `spark/src/main/spark-*/org/apache/comet/iceberg/` | Version differences: operation-coded `ReplaceData` / `WriteDelta` rows, cache refresh by name on 4.1+, driver metric reporting. | +| Component | Location | Role | +| ------------------------------------------------------------------------------------------------------------------------------------ | -------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `IcebergWriteStrategy` | `spark/src/main/scala/org/apache/comet/iceberg/` | Planner strategy. Matches `AppendData`, `OverwriteByExpression`, `OverwritePartitionsDynamic`, `ReplaceData` plus Iceberg `WriteDelta` under the testing split flag. | +| `IcebergWriteLogical` | same | Logical anchor for the writer, so AQE re-plans re-emit only the writer and not a second committer. | +| `IcebergWriteExec` | `spark/src/main/scala/org/apache/spark/sql/comet/` | JVM writer. Runs iceberg-java's `DataWriter` or, for `WriteDelta`, `DeltaWriter` per task and returns the serialized `WriterCommitMessage` as one binary row. | +| `IcebergCommitExec` | same | Driver committer. A `V2CommandExec`, so `run()` is memoized and the commit happens once. | +| `IcebergReplaceDataShim`, `IcebergDeltaLogicalShim`, `IcebergDeltaWriterShim`, `IcebergRefreshCacheShim`, `IcebergDriverMetricsShim` | `spark/src/main/spark-*/org/apache/comet/iceberg/` | Version differences: operation-coded `ReplaceData` / `WriteDelta` rows, cache refresh by name on 4.1+, driver metric reporting. | Things to know before changing this layer: @@ -97,12 +97,12 @@ Things to know before changing this layer: - **Commit-coordinator writes are not intercepted.** Iceberg's current `SparkWrite` and position-delta write do not ask for one; the checks in `buildTwoOp` and `buildDeltaTwoOp` are defensive. -- **WriteDelta stays JVM-backed.** Spark 3.5+ merge-on-read commands are intercepted by the split +- **WriteDelta stays JVM-backed.** Spark 3.5 and later merge-on-read commands are intercepted by the split plan only when the testing flag `spark.comet.write.iceberg.splitOperator.enabled` is on, and then `IcebergWriteExec` delegates their rows to Iceberg's JVM `DeltaWriter`; the native Iceberg writer is explicitly declined. Since the split plan gives such a write nothing over Spark's operator, - `spark.comet.write.iceberg.enabled` alone leaves it on Spark's plan, as does Spark 3.4. -- **What is not intercepted:** streaming writes and CTAS/RTAS on Spark 3.4. Those keep Spark's plan. + `spark.comet.write.iceberg.enabled` alone leaves it on Spark's plan. +- **What is not intercepted:** streaming writes. Those keep Spark's plan. On Spark 4.1+, `IcebergWriteSummaryShim` finds either Spark's `MergeRowsExec` or `CometMergeRowsExec` in the executed query and forwards its eight action counters to diff --git a/docs/source/contributor-guide/shim_pattern.svg b/docs/source/contributor-guide/shim_pattern.svg index 0f966c7a667..6f1e19bee32 100644 --- a/docs/source/contributor-guide/shim_pattern.svg +++ b/docs/source/contributor-guide/shim_pattern.svg @@ -6,8 +6,8 @@ Shim Pattern: Per-Version Spark API Bridging - Spark 3.4, 3.5, and 4.0 all have slightly different QueryExecutionErrors APIs. - Comet ships a separate ShimSparkErrorConverter implementation for each version. + Spark 3.5 and 4.x have slightly different QueryExecutionErrors APIs. + Comet ships a separate ShimSparkErrorConverter implementation for each. @@ -16,53 +16,40 @@ Parses JSON → calls convertErrorType() - - - + + - spark-3.4 - spark-3.5 - spark-4.0 + spark-3.5 + spark-4.x - - ShimSparkErrorConverter - spark-3.4 edition - Uses Spark 3.4 API signatures - (no SQLQueryContext in some calls) - - - ShimSparkErrorConverter - spark-3.5 edition - Uses Spark 3.5 API signatures - (takes SQLQueryContext directly) - - - ShimSparkErrorConverter - spark-4.0 edition - Uses Spark 4.0 API signatures - (new functionName param in some) + + ShimSparkErrorConverter + spark-3.5 edition + Uses Spark 3.5 API signatures + (takes SQLQueryContext directly) + + + ShimSparkErrorConverter + spark-4.x edition + Uses Spark 4.x API signatures (4.0, 4.1, 4.2) + (new functionName param in some) - - - + + - - QueryExecutionErrors - Spark 3.4 internal API - - - QueryExecutionErrors - Spark 3.5 internal API + + QueryExecutionErrors + Spark 3.5 internal API - - QueryExecutionErrors - Spark 4.0 internal API + + QueryExecutionErrors + Spark 4.x internal API - Example difference: "BinaryArithmeticOverflow" — Spark 3.x does NOT take functionName param, Spark 4.0 does. + Example difference: "BinaryArithmeticOverflow" — Spark 3.5 does NOT take functionName param, Spark 4.x does. diff --git a/docs/source/contributor-guide/sql_error_propagation.md b/docs/source/contributor-guide/sql_error_propagation.md index 59e139f0c90..eb253466b64 100644 --- a/docs/source/contributor-guide/sql_error_propagation.md +++ b/docs/source/contributor-guide/sql_error_propagation.md @@ -405,8 +405,8 @@ def convertToSparkException(e: CometQueryExecutionException): Throwable = { ### `ShimSparkErrorConverter` calls the real Spark API -Because Spark's `QueryExecutionErrors` API changes between Spark versions (3.4, 3.5, and the 4.x line), -there is a separate implementation per branch (in `spark-3.4/`, `spark-3.5/`, and `spark-4.x/`, which is +Because Spark's `QueryExecutionErrors` API changes between Spark versions (3.5 and the 4.x line), +there is a separate implementation per branch (in `spark-3.5/` and `spark-4.x/`, which is shared by Spark 4.0, 4.1, and 4.2). ![Shim pattern for per-version Spark API bridging](./shim_pattern.svg) diff --git a/docs/source/user-guide/latest/compatibility/floating-point.md b/docs/source/user-guide/latest/compatibility/floating-point.md index d185e254b74..e43f602ea9a 100644 --- a/docs/source/user-guide/latest/compatibility/floating-point.md +++ b/docs/source/user-guide/latest/compatibility/floating-point.md @@ -118,7 +118,7 @@ natively on Spark 4.0.5+, 4.1.4+, and 4.2+. These releases treat `-0.0` and `0.0 representation, as one value in these functions, including inside nested arrays and structs, and return the normalized value. Spark 4.2.0 normalizes the arguments in the plan (SPARK-54918), and 4.0.5, 4.1.4, and 4.2.1 normalize while these functions evaluate (SPARK-59602). Comet normalizes -the elements the same way. Spark 3.4, 3.5, 4.0.0 to 4.0.4, and 4.1.0 to 4.1.3 keep positive and +the elements the same way. Spark 3.5, 4.0.0 to 4.0.4, and 4.1.0 to 4.1.3 keep positive and negative zero distinct in flat arrays, so on those versions these functions fall back to Spark. Other element types remain native on every version. diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index a24cc12a8ce..7c98b3fb93a 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -136,7 +136,7 @@ Comet puts the files of each store into separate partitions, so such a scan can number of partitions than Spark would use. A native Parquet scan packs each store's files on their own, so `spark.sql.files.maxPartitionNum` applies to each store's files separately. A native CSV scan that would split a partition falls back to Spark when Spark planned the operators above it for -the scan's own partitioning, as Spark 3.4 does for a scan of one partition. +the scan's own partitioning. Such a scan forwards the object store settings of every scheme it reads, so its files can mix schemes such as `s3a://`, `gs://` and `hdfs://`. It falls back to Spark when it reads one bucket @@ -458,7 +458,7 @@ Azure scans discover Root CA Certificates the same way S3 scans do. See [Root CA ### Authentication is resolved by Hadoop -For each `abfs` or `abfss` path, the Spark driver asks the hadoop-azure library on its classpath which authentication mechanism applies to that container and account, and which values it needs. The question goes to the same `AbfsConfiguration` the ABFS `FileSystem` uses, so every Hadoop rule holds: account-scoped keys (`.`), container-scoped keys (`..`, hadoop-azure 3.4.2 and later), credential providers behind `hadoop.security.credential.provider.path` (a JCEKS keystore, for example) and `${...}` substitution behave exactly as they do for Spark's own reads. This holds for the hadoop-azure matching each Spark line's Hadoop: 3.3.4 with Spark 3.4 and 3.5, 3.4.1 with 4.0, 3.4.2 with 4.1, 3.5.0 with 4.2. +For each `abfs` or `abfss` path, the Spark driver asks the hadoop-azure library on its classpath which authentication mechanism applies to that container and account, and which values it needs. The question goes to the same `AbfsConfiguration` the ABFS `FileSystem` uses, so every Hadoop rule holds: account-scoped keys (`.`), container-scoped keys (`..`, hadoop-azure 3.4.2 and later), credential providers behind `hadoop.security.credential.provider.path` (a JCEKS keystore, for example) and `${...}` substitution behave exactly as they do for Spark's own reads. This holds for the hadoop-azure matching each Spark line's Hadoop: 3.3.4 with Spark 3.5, 3.4.1 with 4.0, 3.4.2 with 4.1, 3.5.0 with 4.2. The native scan receives the resolved values and builds the `object_store` client from them. It reads no `fs.azure.*` key of its own. A scan whose files span more than one ABFS container or account, or that mixes `abfs`/`abfss` paths with other schemes, falls back to Spark's own scan. diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index cf8baac011d..58ba641a2f8 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -769,4 +769,4 @@ still falls back when needed. ## See also - [Comet Compatibility Guide](compatibility/index.md) - known incompatibilities and edge cases for supported expressions. -- [Expression Audits (contributor guide)](../../contributor-guide/expression-audits/index.md) - per-version (Spark 3.4 / 3.5 / 4.0 / 4.1) audit notes for audited expressions. +- [Expression Audits (contributor guide)](../../contributor-guide/expression-audits/index.md) - per-version (Spark 3.5 / 4.0 / 4.1) audit notes for audited expressions. diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index f788ab1da9c..8618ac105c2 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -411,14 +411,13 @@ summary-aware writer commit contract. For copy-on-write row-level DML, the mechanism differs by Spark version: on Spark 4.0+ the analyzer emits operation-coded rows that Comet's writer dispatches through `ReplaceData`'s -projections, while on Spark 3.4/3.5 the rewritten rows are written as a plain row stream. The +projections, while on Spark 3.5 the rewritten rows are written as a plain row stream. The supported set of operations is the same either way. Merge-on-read uses Spark's `WriteDelta`, and Iceberg's JVM `DeltaWriter` writes its rows under either plan, so it keeps Spark's stock write plan. On Spark 3.5+ the testing setting `spark.comet.write.iceberg.splitOperator.enabled` plans it as the split plan, with the same driver commit and reporting path, but `CometIcebergWriteExec` is never used for position-delta rows. -Spark 3.4 always leaves `WriteDelta` on Spark's stock write plan. On Spark 4.1+ the split plan matches two further stock-Spark behaviours: MERGE metrics are forwarded to the writer's commit (Iceberg 1.11+ records them in the snapshot summary), and @@ -438,8 +437,6 @@ The rewrite is skipped — and the write runs through Spark's stock combined ope - the write is neither an Iceberg `SparkWrite` nor a supported Iceberg position-delta write; - the table uses merge-on-read (`WriteDelta`), unless the testing setting `spark.comet.write.iceberg.splitOperator.enabled` is on with Spark 3.5+; -- the statement is CTAS / RTAS on Spark 3.4, where the staged exec writes inline; on Spark - 3.5+ those statements re-plan their inner append, which is intercepted normally; - the write is a streaming write (`writeStream` into an Iceberg table); - the write requires Spark's commit coordinator, which Comet's per-task commit protocol does not use; diff --git a/docs/source/user-guide/latest/in-memory-cache.md b/docs/source/user-guide/latest/in-memory-cache.md index c1bbb479c42..ef73632b925 100644 --- a/docs/source/user-guide/latest/in-memory-cache.md +++ b/docs/source/user-guide/latest/in-memory-cache.md @@ -34,10 +34,6 @@ $SPARK_HOME/bin/spark-shell \ --conf spark.comet.exec.inMemoryCache.enabled=true ``` -On Spark 3.4, a union of a shuffle and a relation cached in Comet's format keeps every shuffle -partition. Spark 3.4 has no hook for the rule through which Comet lets AQE coalesce the shuffle -partitions of a union that Comet runs. - It has to be set before the `SparkContext` starts. Comet's driver plugin chooses `spark.sql.cache.serializer` once, while the context is initializing, so an application keeps the cache format it started with however the config is set afterwards. The plugin installs Comet's diff --git a/docs/source/user-guide/latest/installation.md b/docs/source/user-guide/latest/installation.md index 434b9b0d07b..05e17ef325b 100644 --- a/docs/source/user-guide/latest/installation.md +++ b/docs/source/user-guide/latest/installation.md @@ -205,7 +205,7 @@ WARN CometExecRule: Comet cannot execute some parts of this plan natively (set s +- LocalTableScan [COMET: Native support for operator LocalTableScanExec is disabled. Set spark.comet.exec.localTableScan.enabled=true to enable it.] ``` -On Spark 3.4 and 3.5 the native writer replaces the whole write command rather than just the +On Spark 3.5 the native writer replaces the whole write command rather than just the per-task write, so the same message appears on `Execute InsertIntoHadoopFsRelationCommand` and names `DataWritingCommandExec`. diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md index 2e1a49d23a1..969319e0c11 100644 --- a/docs/source/user-guide/latest/operators.md +++ b/docs/source/user-guide/latest/operators.md @@ -137,7 +137,7 @@ combined with `OR`, which Spark plans as an existence `BroadcastNestedLoopJoinEx | 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. See [Parquet](parquet.md#writing). | -| `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. See [Parquet](parquet.md#writing). | +| `DataWritingCommandExec` | ⚠️ | Spark 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. See [Parquet](parquet.md#writing). | | Iceberg writes: `AppendDataExec`, `OverwriteByExpressionExec`, `OverwritePartitionsDynamicExec`, `ReplaceDataExec` | ⚠️ | Disabled by default (opt-in). `spark.comet.write.iceberg.enabled=true` plans the write as `IcebergWrite` and `IcebergCommit` and writes eligible data files natively (`CometIcebergWrite`). Other writes use Iceberg's Java writer. Covers `INSERT INTO`, `INSERT OVERWRITE`, and copy-on-write `DELETE` / `UPDATE` / `MERGE`. Merge-on-read writes (`WriteDeltaExec`) fall back. See [Iceberg Writes](iceberg.md#writes). | | `MergeRowsExec` | ⚠️ | Spark 3.5+. Experimental, disabled by default (`spark.comet.exec.mergeRows.enabled`). On Spark 4.1+, stock V2 writers retain Spark MergeRowsExec; Comet's split Iceberg path can run it natively while preserving MergeSummary. See [MERGE INTO](compatibility/operators.md#merge-into-mergerowsexec). | diff --git a/docs/source/user-guide/latest/parquet.md b/docs/source/user-guide/latest/parquet.md index 04132cf8e47..22b8fe71a4d 100644 --- a/docs/source/user-guide/latest/parquet.md +++ b/docs/source/user-guide/latest/parquet.md @@ -113,8 +113,8 @@ reads that fail where Spark succeeds are listed in ### Missing Nested Fields -When all requested leaf fields under a nested column are missing from a Parquet file, Spark 3.4 -through 4.0 return NULL for the entire missing subtree, including nested arrays and maps. Comet +When all requested leaf fields under a nested column are missing from a Parquet file, Spark 3.5 +and 4.0 return NULL for the entire missing subtree, including nested arrays and maps. Comet follows this behavior. On Spark 4.1+, the default (`false`) preserves the file's parent nullness. Set `spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing=true` for legacy behavior. A field that exists in the file but contains only NULL values is not a missing field. @@ -172,7 +172,7 @@ SchemaColumnConvertNotSupportedException`) instead of the one-level chain Spark' - **Spark 3.x: `TIMESTAMP_MILLIS` overflow past the first batch is wrapped in a file-read error**. When a Parquet `TIMESTAMP_MILLIS` value overflows on conversion to microseconds, Spark's reader - throws `ArithmeticException: long overflow`. Spark 3.4 and 3.5 wrap a reader error in a + throws `ArithmeticException: long overflow`. Spark 3.5 wraps a reader error in a `SparkException` that names the file only while reading the first batch of each file split. For the vectorized reader, that batch is the first `spark.sql.parquet.columnarReaderBatchSize` rows (4096 by default), or the first row group if that is smaller. An overflow in a later batch fails @@ -315,7 +315,7 @@ for production, so do not turn it off to avoid an entry on this page. With `spark.comet.write.parquet.enabled=true`: -- **The `EXCEPTION` rebase mode is ignored.** On Spark 3.4 and 3.5, whose default rebase mode is +- **The `EXCEPTION` rebase mode is ignored.** On Spark 3.5, whose default rebase mode is `EXCEPTION`, the native writer writes dates before October 15, 1582 and timestamps before January 1, 1900 that Spark refuses to write ([#7047](https://github.com/apache/datafusion-comet/issues/7047)). With @@ -326,7 +326,7 @@ With `spark.comet.write.parquet.enabled=true`: as written by Spark 3.0 or later and reads the values back unchanged under every `spark.sql.parquet.datetimeRebaseModeInRead` setting. Files written by a Comet build that predates that key have no version, and Spark then fails to read their pre-1582 values under its - default read mode on 3.4 and 3.5, or reads them shifted by several days under `LEGACY`. + default read mode on 3.5, or reads them shifted by several days under `LEGACY`. - **`spark.sql.parquet.outputTimestampType` and `spark.sql.parquet.writeLegacyFormat` are ignored.** Timestamps are always written as `INT64` microseconds, never as Spark's default `INT96`, so with `TIMESTAMP_MILLIS` they keep the sub-millisecond digits that Spark drops. Spark's @@ -335,7 +335,7 @@ With `spark.comet.write.parquet.enabled=true`: `spark.sql.parquet.inferTimestampNTZ.enabled=false` it reads `TIMESTAMP_NTZ` columns as `TIMESTAMP` ([#7048](https://github.com/apache/datafusion-comet/issues/7048), [#3425](https://github.com/apache/datafusion-comet/issues/3425)). -- **Spark 3.4 and 3.5: `INSERT INTO` leaves the target stale.** Comet skips the steps Spark runs +- **Spark 3.5: `INSERT INTO` leaves the target stale.** Comet skips the steps Spark runs after a write: file index refresh, cache refresh, partition refresh and statistics update. Later queries can return the table's previous contents until `REFRESH TABLE` ([#6315](https://github.com/apache/datafusion-comet/issues/6315)). diff --git a/docs/source/user-guide/latest/pyarrow-udfs.md b/docs/source/user-guide/latest/pyarrow-udfs.md index 48848ecd6e3..39846b2b6bb 100644 --- a/docs/source/user-guide/latest/pyarrow-udfs.md +++ b/docs/source/user-guide/latest/pyarrow-udfs.md @@ -287,7 +287,7 @@ on the unoptimized path. `org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager` and enable `spark.comet.shuffle.enabled=true` at session startup. With a vanilla Spark `Exchange` in the plan the data leaves the shuffle as rows and the optimization cannot fire. -- Spark 4.0 or newer is required. On Spark 3.4 and 3.5 the optimization is a no-op even when +- Spark 4.0 or newer is required. On Spark 3.5 the optimization is a no-op even when enabled; vanilla `PythonMapInArrowExec` / `MapInPandasExec` handle the operation. The Spark 3.5 `PythonArrowInput` trait has a different contract than 4.x and a separate implementation has not been written. Track 3.5 support as a future follow-on if there is user demand. diff --git a/docs/source/user-guide/latest/s3-credential-providers.md b/docs/source/user-guide/latest/s3-credential-providers.md index c32dcd75d04..d33ce766929 100644 --- a/docs/source/user-guide/latest/s3-credential-providers.md +++ b/docs/source/user-guide/latest/s3-credential-providers.md @@ -245,7 +245,7 @@ Publish a real `expirationEpochMillis` when you have one. On both paths Comet re On the Parquet/`object_store` path, `object_store` signs a request once and sends the same signature on every retry, for up to 3 minutes by default, and it does not retry a 403. The five minutes Comet leaves before an expiry cover those retries. A credential you return with less time left than that is used for the request that asked for it and not kept. -The built-in adapters report an expiry when the AWS SDK exposes one, which it does on the Spark 4.x builds (SDK v2). On the Spark 3.4 and 3.5 builds (SDK v1) they cannot, and report `0`. +The built-in adapters report an expiry when the AWS SDK exposes one, which it does on the Spark 4.x builds (SDK v2). On the Spark 3.5 build (SDK v1) they cannot, and report `0`. ### Returned fields diff --git a/spark/src/main/scala/org/apache/comet/GenerateDocs.scala b/spark/src/main/scala/org/apache/comet/GenerateDocs.scala index f36487080d3..06a34c907f8 100644 --- a/spark/src/main/scala/org/apache/comet/GenerateDocs.scala +++ b/spark/src/main/scala/org/apache/comet/GenerateDocs.scala @@ -144,7 +144,7 @@ object GenerateDocs { * Args: * - args(0): user guide root directory (e.g. `docs/source/user-guide/latest/`). * - args(1) (optional): per-Spark-version subdirectory for compatibility pages (e.g. - * `spark-3.4`). When omitted, compat pages are written to `compatibility/expressions/` + * `spark-3.5`). When omitted, compat pages are written to `compatibility/expressions/` * directly, preserving the legacy flat layout used by released-version doc trees. */ def main(args: Array[String]): Unit = { diff --git a/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergDeltaLogicalShim.scala b/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergDeltaLogicalShim.scala deleted file mode 100644 index afbf7d60b70..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergDeltaLogicalShim.scala +++ /dev/null @@ -1,27 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.iceberg - -import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan - -/** WriteDelta interception starts with Spark 3.5. */ -private[iceberg] object IcebergDeltaLogicalShim extends IcebergDeltaLogicalShimApi { - override def extract(plan: LogicalPlan): Option[DeltaLogicalFields] = None -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala b/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala deleted file mode 100644 index 4d66a374663..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala +++ /dev/null @@ -1,40 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.iceberg - -import org.apache.spark.sql.connector.catalog.TableCatalog -import org.apache.spark.sql.execution.SparkPlan -import org.apache.spark.sql.execution.datasources.v2.{AtomicCreateTableAsSelectExec, AtomicReplaceTableAsSelectExec, CreateTableAsSelectExec, ReplaceTableAsSelectExec} - -/** - * Spark 3.4: CTAS and RTAS write the table from the create or replace exec itself, through - * `TableWriteExecHelper.writeWithV2`, so the write never appears as a write node of its own. - */ -private[iceberg] object IcebergTableAsSelectShim { - - /** The catalog `plan` writes through, when `plan` is a CTAS or RTAS exec. */ - def writeCatalog(plan: SparkPlan): Option[TableCatalog] = plan match { - case p: CreateTableAsSelectExec => Some(p.catalog) - case p: AtomicCreateTableAsSelectExec => Some(p.catalog) - case p: ReplaceTableAsSelectExec => Some(p.catalog) - case p: AtomicReplaceTableAsSelectExec => Some(p.catalog) - case _ => None - } -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala deleted file mode 100644 index f983a6cb39e..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala +++ /dev/null @@ -1,78 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.catalyst.expressions.aggregate.Sum - -import org.apache.comet.expressions.CometEvalMode -import org.apache.comet.serde.{CometAggregateExpressionSerde, CometEncode, CometExpressionSerde, CometStringDecode} -import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr} - -/** - * `CometExprShim` acts as a shim for parsing expressions from different Spark versions. - */ -trait CometExprShim { - protected def evalMode(c: Cast): CometEvalMode.Value = - CometEvalModeUtil.fromSparkEvalMode(c.evalMode) - - def binaryOutputStyle: BinaryOutputStyle = BinaryOutputStyle.HEX_DISCRETE - - // Spark 3.4 ships Jackson 2.14, which does not impose the default numeric-token length limit, - // but it can run with a newer Jackson, so choose the native variant from the runtime version. - def getJsonObjectNativeFunctionName: String = - if (CometExprShim.jacksonLimitsNumberLength) "get_json_object" - else "get_json_object_spark34" - - def sparkVersionSpecificStringExpressions - : Map[Class[_ <: Expression], CometExpressionSerde[_]] = - Map(classOf[StringDecode] -> CometStringDecode, classOf[Encode] -> CometEncode) - def sparkVersionSpecificMathExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = - Map.empty - def sparkVersionSpecificMiscExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = - Map.empty - def sparkVersionSpecificMapExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = - Map.empty - def sparkVersionSpecificAggregates: Map[Class[_], CometAggregateExpressionSerde[_]] = - Map.empty - - def sparkVersionSpecificExprToProtoInternal( - expr: Expression, - inputs: Seq[Attribute], - binding: Boolean): Option[Expr] = None -} - -object CometExprShim { - private lazy val jacksonLimitsNumberLength: Boolean = { - // Jackson 2.15 added StreamReadConstraints' default 1000-digit number limit - val v = com.fasterxml.jackson.core.json.PackageVersion.VERSION - v.getMajorVersion > 2 || (v.getMajorVersion == 2 && v.getMinorVersion >= 15) - } -} - -object CometEvalModeUtil { - def fromSparkEvalMode(evalMode: EvalMode.Value): CometEvalMode.Value = evalMode match { - case EvalMode.LEGACY => CometEvalMode.LEGACY - case EvalMode.TRY => CometEvalMode.TRY - case EvalMode.ANSI => CometEvalMode.ANSI - } - - def sumEvalMode(s: Sum): EvalMode.Value = s.evalMode -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimBatchReader.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimBatchReader.scala deleted file mode 100644 index 17b60e0e5a1..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimBatchReader.scala +++ /dev/null @@ -1,37 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.spark.paths.SparkPath -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.execution.datasources.PartitionedFile - -object ShimBatchReader { - - def newPartitionedFile(partitionValues: InternalRow, file: String): PartitionedFile = - PartitionedFile( - partitionValues, - SparkPath.fromPathString(file), - -1, // -1 means we read the entire file - -1, - Array.empty[String], - 0, - 0) -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometBroadcastExchangeExec.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometBroadcastExchangeExec.scala deleted file mode 100644 index 93eee60923d..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometBroadcastExchangeExec.scala +++ /dev/null @@ -1,51 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.spark.SparkContext -import org.apache.spark.network.util.JavaUtils -import org.apache.spark.sql.execution.exchange.BroadcastExchangeLike -import org.apache.spark.sql.internal.SQLConf - -import org.apache.comet.shims.ShimCometBroadcastExchangeExec.SPARK_MAX_BROADCAST_TABLE_SIZE - -trait ShimCometBroadcastExchangeExec { - - def setJobGroupOrTag(sc: SparkContext, broadcastExchange: BroadcastExchangeLike): Unit = { - // Setup a job group here so later it may get cancelled by groupId if necessary. - sc.setJobGroup( - broadcastExchange.runId.toString, - s"broadcast exchange (runId ${broadcastExchange.runId})", - interruptOnCancel = true) - } - - def cancelJobGroup(sc: SparkContext, broadcastExchange: BroadcastExchangeLike): Unit = { - sc.cancelJobGroup(broadcastExchange.runId.toString) - } - - def maxBroadcastTableBytes(conf: SQLConf): Long = { - JavaUtils.byteStringAsBytes(conf.getConfString(SPARK_MAX_BROADCAST_TABLE_SIZE, "8GB")) - } - -} - -object ShimCometBroadcastExchangeExec { - val SPARK_MAX_BROADCAST_TABLE_SIZE = "spark.sql.maxBroadcastTableSize" -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometMergeRows.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometMergeRows.scala deleted file mode 100644 index 488e9734347..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometMergeRows.scala +++ /dev/null @@ -1,35 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan -import org.apache.spark.sql.execution.SparkPlan - -import org.apache.comet.serde.CometOperatorSerde - -/** - * Spark 3.4 predates the core `MergeRowsExec` implementation used by newer Spark versions. - * Nothing is registered here, so MERGE continues through the version-specific JVM path. - */ -object ShimCometMergeRows { - val nativeExecs: Map[Class[_ <: SparkPlan], CometOperatorSerde[_]] = Map.empty - - def withNativeMergeSummary(query: LogicalPlan): LogicalPlan = query -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometSparkSessionExtensions.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometSparkSessionExtensions.scala deleted file mode 100644 index 55e3e1cf8a3..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometSparkSessionExtensions.scala +++ /dev/null @@ -1,60 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.spark.sql.SparkSessionExtensions -import org.apache.spark.sql.catalyst.rules.Rule -import org.apache.spark.sql.execution.{QueryExecution, SparkPlan} - -trait ShimCometSparkSessionExtensions { - - protected val EXTENDED_EXPLAIN_PROVIDERS_KEY = "spark.sql.extendedExplainProviders" - - def supportsExtendedExplainInfo(qe: QueryExecution): Boolean = { - try { - qe.getClass.getDeclaredMethod( - "extendedExplainInfo", - classOf[String => Unit], - classOf[SparkPlan]) - } catch { - case _: NoSuchMethodException | _: SecurityException => return false - } - true - } - - // injectQueryStageOptimizerRule not available on Spark 3.4. - // CometPlanAdaptiveDynamicPruningFilters and CometReuseSubquery are not registered. - // On 3.4, Spark's PlanAdaptiveDynamicPruningFilters handles SABs directly (converting to - // TrueLiteral when it can't find BroadcastHashJoinExec, disabling DPP). - def injectQueryStageOptimizerRuleShim( - extensions: SparkSessionExtensions, - rule: Rule[SparkPlan]): Unit = {} - - // Registers a queryStagePrepRule only on Spark < 3.5. The 3.5+ variants no-op this shim. - // Used by CometSpark34AqeDppFallbackRule, which is a correctness workaround for AQE DPP on - // Spark 3.4 where injectQueryStageOptimizerRule (and therefore - // CometPlanAdaptiveDynamicPruningFilters) is unavailable. See - // CometSpark34AqeDppFallbackRule's class docstring for details. - def injectPreSpark35QueryStagePrepRuleShim( - extensions: SparkSessionExtensions, - rule: Rule[SparkPlan]): Unit = { - extensions.injectQueryStagePrepRule(_ => rule) - } -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometWindowGroupLimit.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometWindowGroupLimit.scala deleted file mode 100644 index b338a32b05f..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometWindowGroupLimit.scala +++ /dev/null @@ -1,40 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.spark.sql.comet.CometWindowGroupLimitExec.Fields -import org.apache.spark.sql.execution.SparkPlan - -/** - * Spark 3.4 does not have `WindowGroupLimitExec` (introduced by SPARK-37099 in Spark 3.5). This - * no-op shim keeps the shared Comet code compiling against Spark 3.4 while ensuring the operator - * is never registered for conversion on this profile. - */ -object ShimCometWindowGroupLimit { - - /** - * The `WindowGroupLimitExec` class object on Spark 3.5+, or `None` on Spark 3.4. - * `CometExecRule` uses this to conditionally register the serde in `nativeExecs`. - */ - def windowGroupLimitClass: Option[Class[_ <: SparkPlan]] = None - - /** Extract WGL fields when `op` is a `WindowGroupLimitExec` (Spark 3.5+). */ - def extract(op: SparkPlan): Option[Fields] = None -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimFileFormat.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimFileFormat.scala deleted file mode 100644 index 6906a39fbfc..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimFileFormat.scala +++ /dev/null @@ -1,67 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.hadoop.fs.Path -import org.apache.spark.sql.catalyst.expressions.Literal -import org.apache.spark.sql.execution.datasources.{FileFormat, PartitionedFile, RowIndexUtil} -import org.apache.spark.sql.types.{DataType, StructType} - -object ShimFileFormat { - - // A name for a temporary column that holds row indexes computed by the file format reader - // until they can be placed in the _metadata struct. - val ROW_INDEX_TEMPORARY_COLUMN_NAME: String = FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME - - def findRowIndexColumnIndexInSchema(sparkSchema: StructType): Int = - RowIndexUtil.findRowIndexColumnIndexInSchema(sparkSchema) - - // Spark 3.4 has no per-format metadata extractor concept (added in Spark 3.5, SPARK-43868); - // it derives these values inline in FileFormat.updateMetadataInternalRow. Replicate that - // logic here so callers can use the same extractor-map shape across all Spark versions. - private val baseMetadataExtractors: Map[String, PartitionedFile => Any] = Map( - FileFormat.FILE_PATH -> { pf: PartitionedFile => - // Use `new Path(Path.toString)` as a form of canonicalization - new Path(pf.filePath.toPath.toString).toUri.toString - }, - FileFormat.FILE_NAME -> { pf: PartitionedFile => - pf.filePath.toUri.getRawPath.split("/").lastOption.getOrElse("") - }, - FileFormat.FILE_SIZE -> { pf: PartitionedFile => pf.fileSize }, - FileFormat.FILE_BLOCK_START -> { pf: PartitionedFile => pf.start }, - FileFormat.FILE_BLOCK_LENGTH -> { pf: PartitionedFile => pf.length }, - // The modificationTime from the file has millisecond granularity, but the TimestampType for - // `file_modification_time` has microsecond granularity. - FileFormat.FILE_MODIFICATION_TIME -> { pf: PartitionedFile => pf.modificationTime * 1000 }) - - def fileConstantMetadataExtractors( - fileFormat: FileFormat): Map[String, PartitionedFile => Any] = - baseMetadataExtractors - - // dataType is unused on this Spark version; getFileConstantMetadataColumnValue only gained - // a dataType parameter in Spark 4.2 (SPARK-56931). Accepted here so callers can pass it - // uniformly across Spark versions. - def getFileConstantMetadataColumnValue( - name: String, - file: PartitionedFile, - extractors: Map[String, PartitionedFile => Any], - dataType: DataType): Literal = - Literal(extractors(name)(file)) -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSQLConf.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSQLConf.scala deleted file mode 100644 index 3ab5e5cbaca..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSQLConf.scala +++ /dev/null @@ -1,36 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy - -trait ShimSQLConf { - protected val LEGACY = LegacyBehaviorPolicy.LEGACY - protected val CORRECTED = LegacyBehaviorPolicy.CORRECTED - - /** - * Reads `spark.sql.execution.arrow.useLargeVarTypes`. Spark 3.4 has no typed accessor for this - * conf, so read by raw key. The conf only governs the destination Arrow IPC root width on Spark - * 4.x, so the value returned here matters only to callers that look it up explicitly. - */ - protected def arrowUseLargeVarTypes(conf: SQLConf): Boolean = - conf.getConfString("spark.sql.execution.arrow.useLargeVarTypes", "false").toBoolean -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSubqueryBroadcast.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSubqueryBroadcast.scala deleted file mode 100644 index 20663bb05ba..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSubqueryBroadcast.scala +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.spark.sql.catalyst.expressions.Expression -import org.apache.spark.sql.execution.{SparkPlan, SubqueryAdaptiveBroadcastExec, SubqueryBroadcastExec} - -trait ShimSubqueryBroadcast { - - def getSubqueryBroadcastIndices(sab: SubqueryAdaptiveBroadcastExec): Seq[Int] = { - Seq(sab.index) - } - - def getSubqueryBroadcastExecIndices(sub: SubqueryBroadcastExec): Seq[Int] = { - Seq(sub.index) - } - - def createSubqueryBroadcastExec( - name: String, - indices: Seq[Int], - buildKeys: Seq[Expression], - child: SparkPlan): SubqueryBroadcastExec = { - assert(indices.length == 1, s"Multi-index DPP not supported on Spark 3.4: indices=$indices") - SubqueryBroadcastExec(name, indices.head, buildKeys, child) - } -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometInMemoryTableScanExec.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometInMemoryTableScanExec.scala deleted file mode 100644 index b5d7421b7ee..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometInMemoryTableScanExec.scala +++ /dev/null @@ -1,25 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.spark.sql.comet.shims - -import org.apache.spark.sql.execution.LeafExecNode - -// Spark 3.4 has no InMemoryTableScanLike or table-cache query stages. -trait ShimCometInMemoryTableScanExec extends LeafExecNode diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometScanExec.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometScanExec.scala deleted file mode 100644 index 7d8ba9f0f8b..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometScanExec.scala +++ /dev/null @@ -1,83 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.spark.sql.comet.shims - -import org.apache.hadoop.fs.{FileStatus, Path} -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression} -import org.apache.spark.sql.execution.{FileSourceScanExec, PartitionedFileUtil} -import org.apache.spark.sql.execution.datasources._ -import org.apache.spark.sql.execution.datasources.parquet.ParquetOptions -import org.apache.spark.sql.sources.Filter -import org.apache.spark.sql.types.StructType - -import org.apache.comet.shims.ShimFileFormat - -trait ShimCometScanExec { - def wrapped: FileSourceScanExec - - lazy val fileConstantMetadataColumns: Seq[AttributeReference] = - wrapped.fileConstantMetadataColumns - - protected def newFileScanRDD( - fsRelation: HadoopFsRelation, - readFunction: PartitionedFile => Iterator[InternalRow], - filePartitions: Seq[FilePartition], - readSchema: StructType, - options: ParquetOptions): FileScanRDD = new FileScanRDD( - fsRelation.sparkSession, - readFunction, - filePartitions, - readSchema, - fileConstantMetadataColumns, - options) - - protected def isNeededForSchema(sparkSchema: StructType): Boolean = { - // TODO: remove after PARQUET-2161 becomes available in Parquet (tracked in SPARK-39634) - ShimFileFormat.findRowIndexColumnIndexInSchema(sparkSchema) >= 0 - } - - protected def getPartitionedFile(f: FileStatus, p: PartitionDirectory): PartitionedFile = - PartitionedFileUtil.getPartitionedFile(f, f.getPath, p.values) - - protected def splitFiles( - sparkSession: SparkSession, - file: FileStatus, - filePath: Path, - isSplitable: Boolean, - maxSplitBytes: Long, - partitionValues: InternalRow): Seq[PartitionedFile] = - PartitionedFileUtil.splitFiles( - sparkSession, - file, - filePath, - isSplitable, - maxSplitBytes, - partitionValues) - - protected def getPushedDownFilters( - relation: HadoopFsRelation, - dataFilters: Seq[Expression]): Seq[Filter] = { - val supportNestedPredicatePushdown = DataSourceUtils.supportNestedPredicatePushdown(relation) - dataFilters.flatMap(DataSourceStrategy.translateFilter(_, supportNestedPredicatePushdown)) - } - -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometShuffleMaterialization.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometShuffleMaterialization.scala deleted file mode 100644 index 26ba626a665..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometShuffleMaterialization.scala +++ /dev/null @@ -1,44 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.spark.sql.comet.shims - -import org.apache.spark.sql.SparkSession - -/** - * Restores the session context captured before shuffle materialization leaves the caller thread. - */ -trait ShimCometShuffleMaterialization { - private val activeSession = SparkSession.getActiveSession - - protected def withCapturedSession[T](body: => T): T = { - val previousSession = SparkSession.getActiveSession - activeSession match { - case Some(session) => SparkSession.setActiveSession(session) - case None => SparkSession.clearActiveSession() - } - try body - finally { - previousSession match { - case Some(session) => SparkSession.setActiveSession(session) - case None => SparkSession.clearActiveSession() - } - } - } -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometUnionExec.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometUnionExec.scala deleted file mode 100644 index 57db09d5f38..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometUnionExec.scala +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.spark.sql.comet.shims - -import scala.reflect.ClassTag - -import org.apache.spark.SparkContext -import org.apache.spark.rdd.RDD -import org.apache.spark.sql.catalyst.plans.physical.Partitioning - -object ShimCometUnionExec { - - /** - * Unions a sequence of RDDs while preserving the declared output partitioning. Before Spark - * 4.1, [[org.apache.spark.sql.execution.UnionExec]] always reports - * [[org.apache.spark.sql.catalyst.plans.physical.UnknownPartitioning]], so this shim simply - * concatenates partitions via `SparkContext.union`. The partitioning-aware path is only needed - * on Spark 4.1+ (see SPARK-52921). - */ - def unionRDDs[T: ClassTag]( - sc: SparkContext, - rdds: Seq[RDD[T]], - outputPartitioning: Partitioning): RDD[T] = { - sc.union(rdds) - } -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala deleted file mode 100644 index 70978c5c3d8..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala +++ /dev/null @@ -1,451 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.spark.sql.comet.shims - -import java.io.FileNotFoundException - -import scala.util.matching.Regex - -import org.apache.spark.{QueryContext, SparkDateTimeException, SparkException} -import org.apache.spark.sql.catalyst.trees.SQLQueryContext -import org.apache.spark.sql.errors.QueryExecutionErrors -import org.apache.spark.sql.execution.datasources.{DataSourceUtils, SchemaColumnConvertNotSupportedException} -import org.apache.spark.sql.types._ -import org.apache.spark.unsafe.types.UTF8String - -object ShimSparkErrorConverter { - val ObjectLocationPattern: Regex = "Object at location (.+?) not found".r -} - -/** - * Spark 3.4 implementation for converting error types to proper Spark exceptions. - * - * Handles all error cases using the Spark 3.4 QueryExecutionErrors API. Differs from the 3.5 - * version in 4 places where the API changed between 3.4 and 3.5. - */ -trait ShimSparkErrorConverter { - - private def sqlCtx(context: Array[QueryContext]): SQLQueryContext = - context.headOption.map(_.asInstanceOf[SQLQueryContext]).getOrElse(null) - - private def parseFloatLiteral(value: String): Float = { - value.toLowerCase match { - case "inf" | "+inf" | "infinity" | "+infinity" => Float.PositiveInfinity - case "-inf" | "-infinity" => Float.NegativeInfinity - case "nan" | "+nan" | "-nan" => Float.NaN - case _ => value.toFloat - } - } - - private def parseDoubleLiteral(value: String): Double = { - val normalized = value.toLowerCase.stripSuffix("d") - normalized match { - case "inf" | "+inf" | "infinity" | "+infinity" => Double.PositiveInfinity - case "-inf" | "-infinity" => Double.NegativeInfinity - case "nan" | "+nan" | "-nan" => Double.NaN - case _ => normalized.toDouble - } - } - - def convertErrorType( - errorType: String, - errorClass: String, - params: Map[String, Any], - context: Array[QueryContext], - summary: String): Option[Throwable] = { - val _ = (errorClass, summary) - - errorType match { - - case "DivideByZero" => - Some(QueryExecutionErrors.divideByZeroError(sqlCtx(context))) - - case "RemainderByZero" => - Some(QueryExecutionErrors.divideByZeroError(sqlCtx(context))) - - case "IntervalDividedByZero" => - Some(QueryExecutionErrors.intervalDividedByZeroError(sqlCtx(context))) - - case "BinaryArithmeticOverflow" => - // Spark 3.x does not take functionName parameter - Some( - QueryExecutionErrors.binaryArithmeticCauseOverflowError( - params("value1").toString.toShort, - params("symbol").toString, - params("value2").toString.toShort)) - - case "ArithmeticOverflow" => - val fromType = params("fromType").toString - val functionName = params.get("functionName").map(_.toString).getOrElse("") - Some( - QueryExecutionErrors - .arithmeticOverflowError(fromType + " overflow", functionName, sqlCtx(context))) - - case "IntegralDivideOverflow" => - Some(QueryExecutionErrors.overflowInIntegralDivideError(sqlCtx(context))) - - case "DecimalSumOverflow" => - // Spark 3.x takes SQLQueryContext, not QueryContext - Some(QueryExecutionErrors.overflowInSumOfDecimalError(sqlCtx(context))) - - case "NumericValueOutOfRange" => - // Use Java BigDecimal to avoid Scala BigDecimal's DECIMAL128 MathContext rewriting. - val decimal = Decimal(new java.math.BigDecimal(params("value").toString)) - Some( - QueryExecutionErrors.cannotChangeDecimalPrecisionError( - decimal, - params("precision").toString.toInt, - params("scale").toString.toInt, - sqlCtx(context))) - - case "DatetimeOverflow" => - Some( - new SparkException( - errorClass = "DATETIME_OVERFLOW", - messageParameters = params.map { case (k, v) => (k, v.toString) }, - cause = null)) - - case "InvalidArrayIndex" => - Some( - QueryExecutionErrors.invalidArrayIndexError( - params("indexValue").toString.toInt, - params("arraySize").toString.toInt, - sqlCtx(context))) - - case "InvalidElementAtIndex" => - Some( - QueryExecutionErrors.invalidElementAtIndexError( - params("indexValue").toString.toInt, - params("arraySize").toString.toInt, - sqlCtx(context))) - - case "InvalidIndexOfZero" => - Some(QueryExecutionErrors.invalidIndexOfZeroError(sqlCtx(context))) - - case "InvalidBitmapPosition" => - // invalidBitmapPositionError does not exist in Spark 3.4 - Some( - new SparkException( - errorClass = "INVALID_BITMAP_POSITION", - messageParameters = params.map { case (k, v) => (k, v.toString) }, - cause = null)) - - case "DuplicatedMapKey" => - Some(QueryExecutionErrors.duplicateMapKeyFoundError(params("key"))) - - case "NullMapKey" => - Some(QueryExecutionErrors.nullAsMapKeyNotAllowedError()) - - case "MapKeyValueDiffSizes" => - Some(QueryExecutionErrors.mapDataKeyArrayLengthDiffersFromValueArrayLengthError()) - - case "ExceedMapSizeLimit" => - Some(QueryExecutionErrors.exceedMapSizeLimitError(params("size").toString.toInt)) - - case "CollectionSizeLimitExceeded" => - // createArrayWithElementsExceedLimitError takes (count: Any) in Spark 3.4; pass the - // decimal string through since the reported length can exceed Long range. - Some( - QueryExecutionErrors.createArrayWithElementsExceedLimitError( - params("numElements").toString)) - - case "SequenceIllegalBoundaries" => - // Spark 3.x codegen throws a plain IllegalArgumentException for sequence boundaries. - Some( - new IllegalArgumentException( - s"Illegal sequence boundaries: ${params("start")} to ${params("stop")} " + - s"by ${params("step")}")) - - case "SequenceBatchTooLarge" => - // Comet-specific per-batch limit for native `sequence`. Point the user at - // spark.comet.batchSize since Spark itself has no equivalent guard. - Some( - new SparkException( - "Comet's native `sequence` kernel cannot materialize a batch with " + - s"${params("totalElements")} total elements: it exceeds the per-batch " + - "limit or the allocator refused the reservation. Lower " + - "`spark.comet.batchSize` so fewer rows are grouped per batch.", - null)) - - case "Internal" => - Some(SparkException.internalError(params("message").toString)) - - case "NotNullAssertViolation" => - Some( - QueryExecutionErrors.foundNullValueForNotNullableFieldError( - params("fieldName").toString)) - - case "ValueIsNull" => - Some( - QueryExecutionErrors.fieldCannotBeNullError( - params.getOrElse("rowIndex", 0).toString.toInt, - params("fieldName").toString)) - - case "CannotParseTimestamp" => - Some( - QueryExecutionErrors.ansiDateTimeParseError(new Exception(params("message").toString))) - - case "IllegalDayOfWeek" => - Some( - QueryExecutionErrors - .ansiIllegalArgumentError(s"Illegal input for day of week: ${params("string")}")) - - case "DatetimeFieldOutOfBounds" => - Some( - QueryExecutionErrors.ansiDateTimeError( - new java.time.DateTimeException(params("rangeMessage").toString))) - - case "InvalidFractionOfSecond" => - Some(QueryExecutionErrors.invalidFractionOfSecondError()) - - case "CastInvalidValue" => - val str = UTF8String.fromString(params("value").toString) - val targetType = getDataType(params("toType").toString) - Some( - QueryExecutionErrors - .invalidInputInCastToNumberError(targetType, str, sqlCtx(context))) - - case "InvalidInputInCastToDatetime" => - val expression = - s"'${params("value").toString.replace("\\", "\\\\").replace("'", "\\'")}'" - val sourceType = s""""${params("fromType").toString}"""" - val targetType = s""""${params("toType").toString}"""" - Some( - new SparkDateTimeException( - errorClass = "CAST_INVALID_INPUT", - messageParameters = Map( - "expression" -> expression, - "sourceType" -> sourceType, - "targetType" -> targetType, - "ansiConfig" -> "\"spark.sql.ansi.enabled\""), - context = context, - summary = summary)) - - case "CastOverFlow" => - val fromType = getDataType(params("fromType").toString) - val toType = getDataType(params("toType").toString) - val valueStr = params("value").toString - - val typedValue: Any = fromType match { - case _: DecimalType => - val cleanStr = if (valueStr.endsWith("BD")) valueStr.dropRight(2) else valueStr - Decimal(cleanStr) - case ByteType => - val cleanStr = if (valueStr.endsWith("T")) valueStr.dropRight(1) else valueStr - cleanStr.toByte - case ShortType => - val cleanStr = if (valueStr.endsWith("S")) valueStr.dropRight(1) else valueStr - cleanStr.toShort - case IntegerType => valueStr.toInt - case LongType => - val cleanStr = if (valueStr.endsWith("L")) valueStr.dropRight(1) else valueStr - cleanStr.toLong - case FloatType => parseFloatLiteral(valueStr) - case DoubleType => parseDoubleLiteral(valueStr) - case StringType => UTF8String.fromString(valueStr) - case _ => valueStr - } - - Some(QueryExecutionErrors.castingCauseOverflowError(typedValue, fromType, toType)) - - case "CannotParseDecimal" => - Some(QueryExecutionErrors.cannotParseDecimalError()) - - case "InvalidUtf8String" => - // invalidUTF8StringError does not exist in Spark 3.x; use generic fallback - Some( - new SparkException( - errorClass = "INVALID_UTF8_STRING", - messageParameters = params.map { case (k, v) => (k, v.toString) }, - cause = null)) - - case "UnexpectedPositiveValue" => - Some( - QueryExecutionErrors.unexpectedValueForStartInFunctionError( - params("parameterName").toString)) - - case "UnexpectedNegativeValue" => - Some( - QueryExecutionErrors.unexpectedValueForLengthInFunctionError( - params("parameterName").toString)) - - case "InvalidRegexGroupIndex" => - // invalidRegexGroupIndexError does not exist in Spark 3.4 - Some( - new SparkException( - errorClass = "INVALID_REGEX_GROUP_INDEX", - messageParameters = params.map { case (k, v) => (k, v.toString) }, - cause = null)) - - case "InvalidUrl" => - Some( - QueryExecutionErrors.invalidUrlError( - UTF8String.fromString(params("url").toString), - new java.net.URISyntaxException(params("url").toString, "Invalid URL"))) - - case "DatatypeCannotOrder" => - // orderedOperationUnsupportedByDataTypeError takes DataType in Spark 3.4, not String - Some( - QueryExecutionErrors.orderedOperationUnsupportedByDataTypeError( - getDataType(params("dataType").toString))) - - case "ScalarSubqueryTooManyRows" => - // multipleRowScalarSubqueryError was renamed to multipleRowSubqueryError in Spark 3.x - Some(QueryExecutionErrors.multipleRowSubqueryError(sqlCtx(context))) - - case "IntervalArithmeticOverflowWithSuggestion" => - // Spark 3.x uses a single intervalArithmeticOverflowError method - Some( - QueryExecutionErrors.intervalArithmeticOverflowError( - "Interval arithmetic overflow", - params.get("functionName").map(_.toString).getOrElse(""), - sqlCtx(context))) - - case "IntervalArithmeticOverflowWithoutSuggestion" => - Some( - QueryExecutionErrors - .intervalArithmeticOverflowError("Interval arithmetic overflow", "", sqlCtx(context))) - - case "DuplicateFieldCaseInsensitive" => - Some( - QueryExecutionErrors.foundDuplicateFieldInCaseInsensitiveModeError( - params("requiredFieldName").toString, - params("matchedOrcFields").toString)) - - case "DuplicateFieldByFieldId" => - // Mirror Spark's `ParquetReadSupport.matchIdField` which calls - // `foundDuplicateFieldInFieldIdLookupModeError` when more than one Parquet field - // shares an id requested by the read schema. - Some( - QueryExecutionErrors.foundDuplicateFieldInFieldIdLookupModeError( - params("requiredId").toString.toInt, - params("matchedFields").toString)) - - case "ParquetMissingFieldIds" => - // Mirror Spark's `ParquetReadSupport.inferSchema`, which throws a plain - // `RuntimeException` (not a SparkException) when the read schema requests field - // ids and the file carries none. - Some( - new RuntimeException( - "Spark read schema expects field Ids, but Parquet file schema doesn't " + - "contain any field Ids. Please remove the field ids from Spark schema or " + - "ignore missing ids by setting " + - "`spark.sql.parquet.fieldId.read.ignoreMissing = true`")) - - case "ParquetSchemaConvert" => - // Mirror Spark's FileScanRDD: wrap the SchemaColumnConvertNotSupportedException - // in a SparkException whose message is "Parquet column cannot be converted in - // file ...". The native side may not have the file path; an empty path - // still produces a message that contains "Parquet column cannot be converted in - // file" (which is what Spark's own SQL tests assert). - val column = params("column").toString - val physicalType = params("physicalType").toString - val logicalType = params("sparkType").toString - val filePath = params.get("filePath").map(_.toString).getOrElse("") - val cause = - new SchemaColumnConvertNotSupportedException(column, physicalType, logicalType) - Some( - QueryExecutionErrors.unsupportedSchemaColumnConvertError( - filePath, - column, - logicalType, - physicalType, - cause)) - - case "FileNotFound" => - val msg = params("message").toString - // Extract file path from native error message and format like Hadoop's - // FileNotFoundException: "File does not exist" - val path = ShimSparkErrorConverter.ObjectLocationPattern - .findFirstMatchIn(msg) - .map(_.group(1)) - .getOrElse(msg) - Some( - QueryExecutionErrors.readCurrentFileNotFoundError( - new FileNotFoundException(s"File $path does not exist"))) - - case "ParquetTimestampOverflow" => - // Spark 3.4's FileScanRDD wraps a reader error in cannotReadFilesError only while - // nextIterator reads the first batch of a file split; an overflow in a later batch - // reaches the user as the bare ArithmeticException. Comet wraps the overflow from every - // batch, as Spark 4.x does, instead of mirroring those batch boundaries. The difference - // is declared in the Parquet compatibility guide (compatibility/scans.md). - val filePath = params.get("filePath").map(_.toString).getOrElse("") - Some( - QueryExecutionErrors - .cannotReadFilesError(new ArithmeticException("long overflow"), filePath)) - - case "ReadAncientDatetime" => - // Spark raises this unwrapped, not as FAILED_READ_FILE. The helper picks the rebase - // config for the format and throws on a format it does not know. - Some(DataSourceUtils.newRebaseExceptionInRead(params("format").toString)) - - case "CannotReadFile" => - // A per-file read failure of a readable-but-broken file (corrupt/truncated parquet, - // object_store, IO) classified by typed DataFusionError variant on the native side. Wrap - // in the FAILED_READ_FILE SparkException Spark itself produces when its own parquet reader - // fails. (A genuinely-missing file is "FileNotFound" above.) `filePath` is filled by - // SparkErrorConverter from the per-task file list when the native error carried none. - val message = params.get("message").map(_.toString).getOrElse("") - val filePath = params.get("filePath").map(_.toString).getOrElse("") - Some(QueryExecutionErrors.cannotReadFilesError(new SparkException(message), filePath)) - - case _ => - None - } - } - - private def getDataType(typeName: String): DataType = { - typeName.toUpperCase match { - case "BYTE" | "TINYINT" => ByteType - case "SHORT" | "SMALLINT" => ShortType - case "INT" | "INTEGER" => IntegerType - case "LONG" | "BIGINT" => LongType - case "FLOAT" | "REAL" => FloatType - case "DOUBLE" => DoubleType - case "DECIMAL" => DecimalType.SYSTEM_DEFAULT - case "STRING" | "VARCHAR" => StringType - case "BINARY" => BinaryType - case "BOOLEAN" => BooleanType - case "DATE" => DateType - case "TIMESTAMP" => TimestampType - case _ => - try { - DataType.fromDDL(typeName) - } catch { - case _: Exception => - // fromDDL rejects types that are syntactically invalid in SQL DDL, such as - // DECIMAL(p,s) with a negative scale (valid when allowNegativeScaleOfDecimal=true). - // Parse those manually rather than silently falling back to StringType. - if (typeName.toUpperCase.startsWith("DECIMAL(") && typeName.endsWith(")")) { - val inner = typeName.substring("DECIMAL(".length, typeName.length - 1) - val parts = inner.split(",") - if (parts.length == 2) { - try { - DataTypes.createDecimalType(parts(0).trim.toInt, parts(1).trim.toInt) - } catch { - case _: Exception => StringType - } - } else StringType - } else StringType - } - } - } -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimTaskMetrics.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimTaskMetrics.scala deleted file mode 100644 index 5b2a5fb5bf5..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimTaskMetrics.scala +++ /dev/null @@ -1,29 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.spark.sql.comet.shims - -import org.apache.spark.executor.TaskMetrics -import org.apache.spark.util.AccumulatorV2 - -object ShimTaskMetrics { - - def getTaskAccumulator(taskMetrics: TaskMetrics): Option[AccumulatorV2[_, _]] = - taskMetrics.externalAccums.lastOption -} diff --git a/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala index 59d8c7f251b..4b522347568 100644 --- a/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala +++ b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala @@ -30,15 +30,13 @@ import org.apache.spark.sql.vectorized.ColumnarBatch /** * Spark 3.x stub for the PyArrow UDF acceleration support. * - * The columnar runner introduced in #4234 only targets Spark 4.0+. On Spark 3.4 / 3.5 the - * matchers return `None`, the rewrite does not fire, and vanilla Spark handles `mapInArrow` / + * The columnar runner introduced in #4234 only targets Spark 4.0+. On Spark 3.5 the matchers + * return `None`, the rewrite does not fire, and vanilla Spark handles `mapInArrow` / * `mapInPandas` unchanged. The runner factory throws; it is never called because the matchers * always return `None`. 3.x support can be added later if there is user demand. * - * Shared across spark-3.4 and spark-3.5 because both are identical: 3.4 lacks the modern - * `ArrowPythonRunner` constructor and `arrowUseLargeVarTypes`, and 3.5's `PythonArrowInput` trait - * has a different contract (`writeIteratorToArrowStream` one-shot vs 4.x's - * `writeNextBatchToArrowStream` batch-at-a-time), so neither version can host the columnar input + * 3.5's `PythonArrowInput` trait has a different contract (`writeIteratorToArrowStream` one-shot + * vs 4.x's `writeNextBatchToArrowStream` batch-at-a-time), so it cannot host the columnar input * implementation without a separate rewrite. */ trait ShimCometMapInBatch { diff --git a/spark/src/test/spark-3.4/org/apache/comet/exec/CometMergeRowsSuite.scala b/spark/src/test/spark-3.4/org/apache/comet/exec/CometMergeRowsSuite.scala deleted file mode 100644 index b97c0e5bb5a..00000000000 --- a/spark/src/test/spark-3.4/org/apache/comet/exec/CometMergeRowsSuite.scala +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.exec - -import org.apache.spark.sql.CometTestBase - -import org.apache.comet.shims.ShimCometMergeRows - -/** Spark 3.4 compatibility coverage for the version-specific MergeRows registration shim. */ -class CometMergeRowsSuite extends CometTestBase { - test("Spark 3.4 leaves native MergeRows execution unregistered") { - assert(ShimCometMergeRows.nativeExecs.isEmpty) - } -} diff --git a/spark/src/test/spark-3.4/org/apache/comet/shims/ShimCometTPCHQuerySuite.scala b/spark/src/test/spark-3.4/org/apache/comet/shims/ShimCometTPCHQuerySuite.scala deleted file mode 100644 index f19f03ff2fd..00000000000 --- a/spark/src/test/spark-3.4/org/apache/comet/shims/ShimCometTPCHQuerySuite.scala +++ /dev/null @@ -1,30 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.comet.shims - -import org.apache.spark.sql.{SparkSession, SQLQueryTestHelper} - -trait ShimCometTPCHQuerySuite extends SQLQueryTestHelper { - protected def getNormalizedQueryExecutionResult( - session: SparkSession, - sql: String): (String, Seq[String]) = { - getNormalizedResult(session, sql) - } -} diff --git a/spark/src/test/spark-3.4/org/apache/spark/sql/ShimCometTestBase.scala b/spark/src/test/spark-3.4/org/apache/spark/sql/ShimCometTestBase.scala deleted file mode 100644 index ce56eeb4364..00000000000 --- a/spark/src/test/spark-3.4/org/apache/spark/sql/ShimCometTestBase.scala +++ /dev/null @@ -1,61 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.spark.sql - -import org.apache.spark.SparkConf -import org.apache.spark.sql.catalyst.expressions.{Expression, MakeDecimal} -import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan - -trait ShimCometTestBase { - type SparkSessionType = SparkSession - - def createSparkSessionWithExtensions(conf: SparkConf): SparkSessionType = { - SparkSession - .builder() - .config(conf) - .master("local[1]") - .withExtensions(new org.apache.comet.CometSparkSessionExtensions) - .getOrCreate() - } - - def datasetOfRows(spark: SparkSession, plan: LogicalPlan): DataFrame = { - Dataset.ofRows(spark, plan) - } - - def getColumnFromExpression(expr: Expression): Column = { - new Column(expr) - } - - def extractLogicalPlan(df: DataFrame): LogicalPlan = { - df.logicalPlan - } - - def createMakeDecimalColumn(child: Expression, precision: Int, scale: Int): Column = { - new Column(MakeDecimal(child, precision, scale)) - } - - def createMakeDecimalColumn( - child: Expression, - precision: Int, - scale: Int, - nullOnOverflow: Boolean): Column = { - new Column(MakeDecimal(child, precision, scale, nullOnOverflow)) - } -}