andygrove commented on PR #5638:
URL:
https://github.com/apache/datafusion-comet/pull/5638#issuecomment-5533914393
_LLM-assisted: this reply and the changes it describes were written with
Claude Code._
@sunchao the P1 is fixed and both claims are gone.
**The compile break** (f2f0c70). `SQLHelper.withSQLConf` is `(pairs:
(String, String)*)(f: => Unit): Unit` on 3.4 and 3.5, so a method typed
`Array[Row]` cannot return the wrapper. The helper now collects into a local
inside the configuration scope and returns it after the block. I ran
`test-compile` locally on `-Pspark-3.4`, `-Pspark-3.5` and the default 4.1
profile; all three are clean.
**The claims.** The scaladoc no longer says the two arms scan identically.
It now says that enabling Comet also replaces the Parquet scan and the
projection with native operators, that each ratio is therefore a query-level
scan-plus-projection result, and that it does not isolate the transform. The
sentence about the ratios being a floor on the expression-level gain is out of
the PR description, which @jordepic had also asked me to rewrite in plainer
English, so I rewrote the whole thing. What is left of the Parquet point is
only what it should be: keeping the data in Parquet keeps the Iceberg reader
out of the Spark arm and lets both arms read the same files.
**The run.** I re-ran on the current head rather than reusing the numbers,
so the table in the description is from this run. One reproduction note worth
recording: my first attempt reported Comet at 0.1x-0.3x across the board,
because `spark/target/classes/org/apache/comet/darwin/aarch64/libcomet.dylib`
still held the debug library from an earlier test run and the resources plugin
skips the copy when the destination is newer than the source. `make
benchmark-<class>` does not hit this because its `release` prerequisite
rebuilds the Rust library first. The numbers below are with the 110 MB release
library in place, verified by size.
Apple M3 Max, macOS 26.6.2, OpenJDK 17.0.10, Spark 4.1, Scala 2.13, `cargo
build --release`, `-Prelease -Pspark-4.1`, 1,048,576 rows,
`SPARK_GENERATE_BENCHMARK_FILES=1`, best of five, as `no-null / with-null`:
```
bucket(int) 1.2X / 1.4X truncate(int) 1.1X / 1.4X
years(date) 2.0X / 2.2X
bucket(long) 1.2X / 1.4X truncate(long) 1.2X / 1.4X
years(ts) 2.5X / 2.5X
bucket(dec) 3.6X / 3.2X truncate(str_dict) 1.6X / 1.8X
months(date) 2.0X / 2.4X
bucket(str_dict) 1.6X / 1.7X truncate(str) 1.2X / 1.3X
months(ts) 2.6X / 2.5X
bucket(str) 1.3X / 1.3X truncate(bin) 1.4X / 1.4X
days(date) 1.1X / 1.4X
bucket(bin) 1.4X / 1.3X days(ts)
2.9X / 2.8X
bucket(date) 1.2X / 1.5X
hours(ts) 2.9X / 2.8X
bucket(ts) 1.6X / 1.7X
```
These differ from the ones I posted before by at most 0.2x in either
direction, which is run-to-run drift on a laptop with other work on it, not a
change in the code.
**Warnings.** None. The results file contains no `WARNING` block, so neither
the not-fully-native check nor the folded-away check fired for any of the 40
cases. All 40 row-by-row comparisons passed as well; a mismatch throws before
the case is timed, so the run completing is the assertion.
**Executed plans.** Captured during the timed run by printing
`stripAQEPlan(df.queryExecution.executedPlan)` for both arms of each case.
Operator chains for all 40:
<details>
<summary>40 cases, Spark plan vs Comet plan</summary>
```
case Spark Comet
------------------------- ------------------------------------
----------------------------------------------------
bucket(int) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(int, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(long) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(long, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(dec) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(dec, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(str_dict) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(str_dict, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(str) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(str, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(bin) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(bin, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(date) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(date, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(ts) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
bucket(ts, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(int) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(int, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(long) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(long, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(str_dict) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(str_dict, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(str) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(str, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(bin) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
truncate(bin, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
years(date) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
years(date, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
years(ts) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
years(ts, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
months(date) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
months(date, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
months(ts) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
months(ts, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
days(date) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
days(date, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
days(ts) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
days(ts, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
hours(ts) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
hours(ts, nulls) Project -> ColumnarToRow -> FileScan
CometColumnarToRow -> CometProject -> CometNativeScan
```
</details>
Every Comet plan is `CometColumnarToRow -> CometProject -> CometNativeScan`,
so no case fell back. Six representative pairs in full:
<details>
<summary>Full plans</summary>
### bucket(dec)
Spark:
```
*(1) Project
[static_invoke(org.apache.iceberg.spark.functions.BucketFunction$BucketDecimal.invoke(16,
c_dec#31)) AS
static_invoke(org.apache.iceberg.spark.functions.BucketFunction$BucketDecimal.invoke(16,
c_dec))#1047]
+- *(1) ColumnarToRow
+- FileScan parquet [c_dec#31] Batched: true, DataFilters: [], Format:
Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>], PartitionFilters:
[], PushedFilters: [], ReadSchema: struct<c_dec:decimal(38,10)>
```
Comet:
```
*(1) CometColumnarToRow
+- CometProject
[static_invoke(org.apache.iceberg.spark.functions.BucketFunction$BucketDecimal.invoke(16,
c_dec))#1051],
[static_invoke(org.apache.iceberg.spark.functions.BucketFunction$BucketDecimal.invoke(16,
c_dec#31)) AS
static_invoke(org.apache.iceberg.spark.functions.BucketFunction$BucketDecimal.invoke(16,
c_dec))#1051]
+- CometNativeScan parquet [c_dec#31] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>],
PartitionFilters: [], PushedFilters: [], ReadSchema:
struct<c_dec:decimal(38,10)>
```
### truncate(str_dict)
Spark:
```
*(1) Project
[static_invoke(org.apache.iceberg.spark.functions.TruncateFunction$TruncateString.invoke(4,
c_str_dict#33)) AS
static_invoke(org.apache.iceberg.spark.functions.TruncateFunction$TruncateString.invoke(4,
c_str_dict))#4493]
+- *(1) ColumnarToRow
+- FileScan parquet [c_str_dict#33] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>],
PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c_str_dict:string>
```
Comet:
```
*(1) CometColumnarToRow
+- CometProject
[static_invoke(org.apache.iceberg.spark.functions.TruncateFunction$TruncateString.invoke(4,
c_str_dict))#4497],
[static_invoke(org.apache.iceberg.spark.functions.TruncateFunction$TruncateString.invoke(4,
c_str_dict#33)) AS
static_invoke(org.apache.iceberg.spark.functions.TruncateFunction$TruncateString.invoke(4,
c_str_dict))#4497]
+- CometNativeScan parquet [c_str_dict#33] Batched: true, DataFilters:
[], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>],
PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c_str_dict:string>
```
### years(ts)
Spark:
```
*(1) Project
[static_invoke(org.apache.iceberg.spark.functions.YearsFunction$TimestampToYearsFunction.invoke(c_ts#41))
AS
static_invoke(org.apache.iceberg.spark.functions.YearsFunction$TimestampToYearsFunction.invoke(c_ts))#5855]
+- *(1) ColumnarToRow
+- FileScan parquet [c_ts#41] Batched: true, DataFilters: [], Format:
Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>], PartitionFilters:
[], PushedFilters: [], ReadSchema: struct<c_ts:timestamp>
```
Comet:
```
*(1) CometColumnarToRow
+- CometProject
[static_invoke(org.apache.iceberg.spark.functions.YearsFunction$TimestampToYearsFunction.invoke(c_ts))#5859],
[static_invoke(org.apache.iceberg.spark.functions.YearsFunction$TimestampToYearsFunction.invoke(c_ts#41))
AS
static_invoke(org.apache.iceberg.spark.functions.YearsFunction$TimestampToYearsFunction.invoke(c_ts))#5859]
+- CometNativeScan parquet [c_ts#41] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>],
PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c_ts:timestamp>
```
### months(date)
Spark:
```
*(1) Project
[static_invoke(org.apache.iceberg.spark.functions.MonthsFunction$DateToMonthsFunction.invoke(c_date#39))
AS
static_invoke(org.apache.iceberg.spark.functions.MonthsFunction$DateToMonthsFunction.invoke(c_date))#6318]
+- *(1) ColumnarToRow
+- FileScan parquet [c_date#39] Batched: true, DataFilters: [], Format:
Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>], PartitionFilters:
[], PushedFilters: [], ReadSchema: struct<c_date:date>
```
Comet:
```
*(1) CometColumnarToRow
+- CometProject
[static_invoke(org.apache.iceberg.spark.functions.MonthsFunction$DateToMonthsFunction.invoke(c_date))#6322],
[static_invoke(org.apache.iceberg.spark.functions.MonthsFunction$DateToMonthsFunction.invoke(c_date#39))
AS
static_invoke(org.apache.iceberg.spark.functions.MonthsFunction$DateToMonthsFunction.invoke(c_date))#6322]
+- CometNativeScan parquet [c_date#39] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>],
PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c_date:date>
```
### days(ts, nulls)
Spark:
```
*(1) Project
[static_invoke(org.apache.iceberg.spark.functions.DaysFunction$TimestampToDaysFunction.invoke(c_ts_n#42))
AS
static_invoke(org.apache.iceberg.spark.functions.DaysFunction$TimestampToDaysFunction.invoke(c_ts_n))#8144]
+- *(1) ColumnarToRow
+- FileScan parquet [c_ts_n#42] Batched: true, DataFilters: [], Format:
Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>], PartitionFilters:
[], PushedFilters: [], ReadSchema: struct<c_ts_n:timestamp>
```
Comet:
```
*(1) CometColumnarToRow
+- CometProject
[static_invoke(org.apache.iceberg.spark.functions.DaysFunction$TimestampToDaysFunction.invoke(c_ts_n))#8148],
[static_invoke(org.apache.iceberg.spark.functions.DaysFunction$TimestampToDaysFunction.invoke(c_ts_n#42))
AS
static_invoke(org.apache.iceberg.spark.functions.DaysFunction$TimestampToDaysFunction.invoke(c_ts_n))#8148]
+- CometNativeScan parquet [c_ts_n#42] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>],
PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c_ts_n:timestamp>
```
### hours(ts)
Spark:
```
*(1) Project
[static_invoke(org.apache.iceberg.spark.functions.HoursFunction$TimestampToHoursFunction.invoke(c_ts#41))
AS
static_invoke(org.apache.iceberg.spark.functions.HoursFunction$TimestampToHoursFunction.invoke(c_ts))#8370]
+- *(1) ColumnarToRow
+- FileScan parquet [c_ts#41] Batched: true, DataFilters: [], Format:
Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>], PartitionFilters:
[], PushedFilters: [], ReadSchema: struct<c_ts:timestamp>
```
Comet:
```
*(1) CometColumnarToRow
+- CometProject
[static_invoke(org.apache.iceberg.spark.functions.HoursFunction$TimestampToHoursFunction.invoke(c_ts))#8374],
[static_invoke(org.apache.iceberg.spark.functions.HoursFunction$TimestampToHoursFunction.invoke(c_ts#41))
AS
static_invoke(org.apache.iceberg.spark.functions.HoursFunction$TimestampToHoursFunction.invoke(c_ts))#8374]
+- CometNativeScan parquet [c_ts#41] Batched: true, DataFilters: [],
Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:<tmp>],
PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c_ts:timestamp>
```
</details>
The other 34 pairs differ only in the function class, the column name and
the expression ids. I can post them all if you want them verbatim.
@jordepic on your third point: #5651 has landed and is merged into this
branch, so the multi-byte string column is back in the partitioned write test's
partition spec (8c6a064), together with the `Long.MinValue` row that had been
excluded for the `truncate(1000, i64)` spec that replaced it. The suite passes
locally, but that is macOS with a UTF-8 platform charset, where the failure
never reproduced in the first place. The Linux `[scans]` jobs on this push are
the run that actually confirms your diagnosis.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
