Uh oh!
There was an error while loading. Please reload this page.
fix: coalesce should return correct datatype - #168
Merged
Conversation
viirya
commented
Mar 5, 2024
Comment on lines
+37
to
+48
| test("coalesce should return correct datatype") { | ||
| Seq(true, false).foreach { dictionaryEnabled => | ||
| withTempDir { dir => | ||
| val path = new Path(dir.toURI.toString, "test.parquet") | ||
| makeParquetFileAllTypes(path, dictionaryEnabled = dictionaryEnabled, 10000) | ||
| withParquetTable(path.toString, "tbl") { | ||
| checkSparkAnswerAndOperator( | ||
| "SELECT coalesce(cast(_18 as date), cast(_19 as date), _20) FROM tbl") | ||
| } | ||
| } | ||
| } | ||
| } |
MemberAuthor
There was a problem hiding this comment.
Due to the issue apache/datafusion#9458, the return type and the actual output array is different in DataFusion coalesce function:
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 2.0 failed 1 times, most recent failure: Lost task 0.0 in stage 2.0 (TID 2) (192.168.86.44 executor driver): org.apache.comet.CometNativeException
: Arrow error: Invalid argument error: column types must match schema types, expected Utf8 but found Date32 at column index 0 at org.apache.comet.Native.executePlan(Native Method) at org.apache.comet.CometExecIterator.executeNative(CometExecIterator.scala:65)
at org.apache.comet.CometExecIterator.getNextBatch(CometExecIterator.scala:111)
at org.apache.comet.CometExecIterator.hasNext(CometExecIterator.scala:126)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.columnartorow_nextBatch_0$(Unknown Source)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)
viirya
commented
Mar 5, 2024
Comment on lines
+1504
to
+1506
| // TODO: Remove this once we have new DataFusion release which includes | ||
| // the fix: https://github.com/apache/arrow-datafusion/pull/9459 | ||
| castToProto(None, a.dataType, childExpr) |
MemberAuthor
There was a problem hiding this comment.
This is a workaround for now before we have new DataFusion release that includes the fix: apache/datafusion#9459
viirya
commented
Mar 5, 2024
MemberAuthor
cc @sunchao |
viirya
commented
Mar 5, 2024
MemberAuthor
Merged. Thanks. |
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 20, 2026
…ntains helper New comprehensive feature-coverage suite that asserts both engagement (Comet plan exec present) AND correctness for each special feature the contrib supports: - Deletion Vectors (DV-bearing tables after DELETE) - Row tracking (unmaterialised _metadata.row_id from baseRowId) - Synthetic columns (_metadata.row_index downstream consumption) - input_file_name() (per-task FileBlockHolder threading, 2-file table -> distinct path per row) - Complex types (struct + array + map round-trip) - Aggregation (count/sum over Delta with native scan asserted) - Self-join (native scan in both join sides) - Time travel by timestamp (separate from versionAsOf in NativeSuite) Adds `assertNativePlanContains(df, "ExecName"...)` helper in CometDeltaTestBase: a hard guard against silent-disengagement bugs that asserts specific simple class names appear in the executed plan after collect. Fails with full plan in the message. Current: 4/8 features pass (complex types, aggregation, self-join, timestamp time-travel). The 4 failures (DV / row tracking / synthetic / input_file_name) are tracked under apache#168 + new follow-ups -- they expose contrib-side bugs that were hidden until the reflection bridges were fixed. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 20, 2026
forceApply was needed earlier in the debug sequence (before the DeltaIntegration $-fix landed) so AQE would wrap simple plans without exchanges -- otherwise Comet's QueryStagePrepRule never fired. With the contrib actually engaging now, Delta's PreprocessTableWithDVsStrategy produces plans AQE elects to wrap naturally for every Delta read. forceApply additionally triggered Spark internal asserts inside AdaptiveSparkPlanExec.setLogicalLinkForNewQueryStage on column-mapping-rewritten plans (4 of 5 CometDeltaColumnMappingSuite tests). Removing it: NativeSuite 12/12 still pass, ColumnMappingSuite 3/5 pass (was 0/5), FeaturesSuite 4/8 (unchanged). Remaining failures (DV / row tracking / synthetic / input_file_name) are real contrib-side bugs around special-column engagement; tracked under apache#168. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 20, 2026
…Format got rewritten Delta's `PreprocessTableWithDVs` strategy and other internal rewriters turn `DeltaParquetFileFormat`+DeltaFileIndex into plain `ParquetFileFormat`+`PreparedDeltaFileIndex` (or similar batch indexes) before our rule sees the plan. The contrib's `isDeltaFileFormat` check returned false on that shape, so `transformV1IfDelta` declined and the DV-bearing scan went to Spark's vanilla reader. Two-spot fix: - `transformV1IfDelta` (line 83): accept either Delta fileFormat OR Delta-internal FileIndex (`isBatchFileIndex` already covers Tahoe* + PreparedDeltaFileIndex + CdcAddFileIndex variants). - `collectDeltaScanBelow` (line 108): same broader acceptance for the DV-strip helper that walks under a `Project(Filter(DV-pattern))` wrapper looking for the underlying Delta scan. Partial fix: opens the gate so more code paths reach the contrib path, but the DV-strip-then-rewrite chain still has additional issues downstream (DV test still fails after this change). Tracked further under apache#168. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
…leIndexes Same shape as the earlier collectDeltaScanBelow fix (e2ae4c6): the helper walks under a DV-filter wrapper looking for the Delta scan, but only accepted scans where `relation.fileFormat` is DeltaParquetFileFormat. Many DV-rewritten plans have `ParquetFileFormat` with a Delta-internal FileIndex (PreparedDeltaFileIndex etc.) instead. Accept those too. Partial fix -- DV strip still has a downstream issue where the rebuilt scan doesn't fastEquals the original `scanExec` so the per-scan rewrite loop in transformV1 picks the wrong target. Tracked under apache#168. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
…ing enabled) `_metadata.row_index` is only populated on Delta tables with row tracking enabled. The previous test wrote without enableRowTracking and then tried to extract row_index from _metadata, which threw an AnalysisException. Fix the test to enable row tracking and add an explicit assertion that Comet engages. (The test still fails because of a separate contrib bug -- the contrib doesn't recognize this row-tracking pattern when Delta's strategy has already rewritten the scan to plain parquet over PreparedDeltaFileIndex. Tracked under apache#168 alongside DV and input_file_name.) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
CometNativeScanExec, CometBatchScanExec, CometScanExec, and CometIcebergNativeScanExec all do: scanExec.logicalLink.foreach(replacement.setLogicalLink) The contrib's CometDeltaNativeScan.createExec was missing this single line, leaving the resulting CometDeltaNativeScanExec without a logical link. AdaptiveSparkPlanExec.setLogicalLinkForNewQueryStage asserts every new query-stage node has a logicalLink set; without this, AQE plans containing a CometDeltaNativeScanExec hit the assertion when AQE wraps a stage that includes one. Symptoms unblocked: - Running multiple test suites together (NativeSuite + ColumnMappingSuite + FeaturesSuite) previously yielded 4/25 pass. With this fix: 19/25 pass mixed (same as individual runs). - spark.sql.adaptive.forceApply could not be used previously without triggering Spark internal asserts on column-mapping rewrites; this is a separate stability issue but the underlying cause is the same. The remaining 6 failures (DV * 3, row-tracking, synthetic, input_file_name) are pattern-match gaps in the contrib's recognition of Delta's strategy-rewritten plans -- tracked under apache#168. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
…a constant Two small but related cleanups: 1. CometDeltaNativeScan: when building projection_vector for the parquet ParquetSource, skip any required column that resolves to neither file-data nor partition-schema. Previously such columns produced -1 which wrapped to u64::MAX on the native side and crashed inside DataFusion's `FileScanConfigBuilder::with_projection_indices` with "index out of bounds: the len is N but the index is 18446744073709551615". 2. DeltaReflection: add `TmpMetadataRowIndexColumnName = "_tmp_metadata_row_index"` constant to mirror Delta's `DeltaParquetFileFormat.TMP_METADATA_ROW_INDEX_COLUMN_NAME`. Used in plans Delta builds for `_metadata.row_index` reads from row-tracking tables. Currently a constant only -- the contrib does not yet synthesize this column natively (full support requires per-row metadata synthesis including file_path / file_name / file_size / etc. from FileScanConfig, which is significant native work). Failure mode for the remaining 6 tests has shifted from "index out of bounds crash" (now fixed) to "Output column count mismatch" -- a real correctness issue where the contrib doesn't synthesize the Spark `_metadata.*` / Delta `_tmp_metadata_row_index` virtual columns that Delta's strategies expect downstream. Tracked under apache#168. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
Delta plans that read `_metadata.row_index` from row-tracking-enabled tables expose the row index as `_tmp_metadata_row_index` (the Delta-internal physical name from `DeltaParquetFileFormat.TMP_METADATA_ROW_INDEX_COLUMN_NAME`) rather than the canonical `__delta_internal_row_index` the contrib previously emitted. End-to-end support: proto: new `DeltaScanCommon.row_index_column_alias` (field 23). When empty defaults to `__delta_internal_row_index`. native (synthetic_columns.rs): `DeltaSyntheticColumnsExec::new` takes an extra `row_index_column_name: &str` arg, stores it on the struct, threads it through `build_output_schema`. `with_new_children` preserves the stored name on re-construction. native (contrib_delta_scan.rs dispatcher): reads `common.row_index_column_alias`, falls back to `ROW_INDEX_COLUMN_NAME` when empty, passes to the `DeltaSyntheticColumnsExec::new` call. Scala (CometDeltaNativeScan): emit_row_index now fires for either `__delta_internal_row_index` OR `_tmp_metadata_row_index` in the scan's required schema. When the latter, set `row_index_column_alias` on the proto so the native synthesis produces a column with the matching name (without renaming or projection). `isSynthetic` also recognises the alternate name. Effect on failing tests: column count check moved from "expected 3, got 1" to "expected 3, got 2" -- one additional virtual column now gets correctly synthesized. The remaining gap (1 column short on these plans) is Spark `_metadata.*` virtual columns (file_path / file_name / file_size / etc.) that the contrib does not yet natively populate from FileScanConfig; that's substantial additional native parquet work tracked under apache#168. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
…columns Delta's planning strategies inject Spark's `_metadata.*` virtual columns as flat top-level columns in the FileScan output: `file_path`, `file_name`, `file_size`, `file_block_start`, `file_block_length`, `file_modification_time`. The contrib previously only knew about its four canonical synthetic columns (`__delta_internal_row_index`, `__delta_internal_is_row_deleted`, `row_id`, `row_commit_version`), so Delta DV / row-tracking / `_metadata`-consuming plans hit "Output column count mismatch" crashes downstream. End-to-end support across all three layers: proto: new `DeltaScanCommon.metadata_column_names = 24` (ordered list of names to emit), and `DeltaScanTask.modification_time = 10` (epoch millis for `file_modification_time` synthesis). native (synthetic_columns.rs): - New `TaskMetadata` struct carries per-task constants (file_path, file_size, byte_range, modification_time) - `metadata_field(name)` returns the Arrow Field for each `_metadata.*` virtual column with Spark-matching dtypes: `file_path`/`file_name` Utf8, `file_size`/`file_block_start`/ `file_block_length` Int64, `file_modification_time` Timestamp(Microsecond, "UTC") - `DeltaSyntheticColumnsExec::new` takes additional `metadata_column_names: Vec<String>` and `task_metadata_by_partition: Vec<TaskMetadata>` - `build_output_schema` appends metadata columns after the 4 canonical synthetics - `augment()` synthesizes per-batch arrays for each: file_path and file_name (basename) from `task.file_path`, file_size from `task.file_size`, file_block_start/length from byte_range or file_size, file_modification_time as `TimestampMicrosecondArray` with "UTC" timezone (ms -> us conversion) native (contrib_delta_scan.rs dispatcher): - `need_per_file_groups` and `need_synthetics` both include `!metadata_column_names.is_empty()` - Builds `task_metadata_per_group` parallel to `deleted_indexes_per_group` / `base_row_ids_per_group` - Passes `metadata_column_names` and `task_metadata_per_group` to `DeltaSyntheticColumnsExec::new` Scala (CometDeltaNativeScan + DeltaReflection): - Detect `_metadata.*` names in `scan.requiredSchema`; add each present name to the proto via `addMetadataColumnNames` - `ExtractedAddFile` carries `modificationTime`, populated from `AddFile.modificationTime` via reflection in `extractBatchAddFiles` - `buildTaskListFromAddFiles` sets `task.modificationTime` from the AddFile - `needsSyntheticEmit` accounts for metadata-only emit paths - Metadata column names added to `isSynthetic` recognition so they get skipped from the parquet `projection_vector` (otherwise the old `-1` -> u64::MAX wrap would crash native again) Effect: contrib now natively populates `_metadata.*` virtual columns that Delta's strategies require downstream. Test status holds at 19/25 contrib Scala tests passing; remaining 6 failures shifted to distinct contrib-side bugs (Int64-nullable schema mismatch on row tracking, SQL planning failure on certain `_metadata.*` consumption, filter test still empty) tracked under apache#168 -- the architectural gap that the previous "expected 3, got 1" message reflected is now closed at the contrib's emit layer. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
…_order
Two related fixes to the synthetic-column emission:
Native: `row_index` column was emitted as UInt64 but Spark's
`_metadata.row_index` is `LongType` (signed Int64). Comet's
type-mapping layer rejects `Int(64, false)` (UInt64) with
"Unsupported data type". Fix by emitting Int64Array with i64 values
cast from the row-position counter.
Scala: `syntheticEmitOrder` lookup used the canonical
`__delta_internal_row_index` name even when an alias was set via
`row_index_column_alias`. For row-tracking plans that surface
`_tmp_metadata_row_index` instead, the `finalOutputIndices`
computation then failed an assertion ("synthetic column
'_tmp_metadata_row_index' in required_schema but no emit flag is
set"). Fix by using the emitted name (alias if present, canonical
otherwise) in the lookup, and append `metadata_column_names` to
the order so reorder indices land correctly.
Test status: 20/25 contrib Scala tests now pass (was 19/25). The
"synthetic: native scan engages when row tracking is enabled" test
now passes end-to-end. Remaining 5 failures (3 DV variants, row
tracking with `_row-id-col-<UUID>` materialised column, input_file_name)
are distinct issues tracked under apache#168.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
…utput Spark appends `_metadata.file_path` (and siblings) to `scan.output` even when `scan.requiredSchema` doesn't carry them, so downstream operators that bind to those attributes by name (e.g. Delta's PreprocessTableWithDVs reads `_metadata.file_path` off the scan output) get attribute-resolution failures when the wrapped exec only emits the required-schema fields. This patch: 1. Detects `_metadata.*` virtual columns from `scan.output` in addition to `scan.requiredSchema` and includes them in `requiredSchemaFields` so the wrapped exec output schema carries them. 2. Extends `metadataColumnNamesEmitted` filter to the same union, so the native side knows to synthesise the columns it now reports. 3. Switches `data_filters` pushdown to bind against the non-synthetic prefix of `scan.requiredSchema` instead of `scan.output`. Filters were binding against `scan.output` which now carries extra attributes, causing `Bound` indices to misalign with the stripped `required_schema` the native side decodes against. Status: removes the "Output column count mismatch: expected 3, got 2" crash on DV-bearing tables after DELETE. Data correctness still off (test expects 13 rows, gets 1 -- distinct issue with how the DELETE predicate gets re-pushed to the SELECT scan; tracked under apache#168). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
…row-id cols Adds three new native-synthesised column kinds so plans that read `_metadata.row_id` on row-tracking-enabled Delta tables work without falling back: - `base_row_id`: per-file Int64 constant pulled from `DeltaScanTask.base_row_id` (`AddFile.baseRowId`). Threaded through `TaskMetadata.base_row_id` and emitted from a per-partition `task_metadata_by_partition`. - `_row-id-col-<uuid>`: Delta's materialised row-id column. When the parquet file doesn't carry it (unmaterialised case), emit all-null Int64 so Delta's `GenerateRowIDs` Project falls back to `base_row_id + row_index`. - `_row-commit-version-col-<uuid>`: analogous to materialised row-id. Detection is by name in Scala -- prefix match against scan.output / scan.requiredSchema. `metadataColumnNamesEmitted` now walks `scan.output` in order so the wrapped exec's output layout matches what the upstream Project expects without a final reorder. Test status: 22/25 contrib Scala tests pass (was 20/25). The "row tracking: unmaterialised _metadata.row_id synthesised from baseRowId" test now passes end-to-end. Two failures remain (DV-after-DELETE returning 1 row instead of 13; input_file_name returning Set()), tracked under apache#168. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
When the scan emits synthetic columns (row_index, _metadata.*, base_row_id, materialised row-id) AND the file has a deletion vector, we previously chose ONE wrapper: synthetic-emit OR DV-filter, never both. That made the DV-DELETE / DV-in-use SELECT paths return un-filtered rows whenever any synthetic was requested (e.g. when scan.output added _metadata.file_path). Chain the two execs: synthetic emission runs first so row_index reflects the file's physical position; DV filter then drops deleted rows -- the emitted columns ride along with the kept rows. SKIPPED when emit_is_row_deleted is on (UPDATE/DELETE/MERGE writers consume the flag and need every row). Test status: 21/25 contrib Scala tests pass (was 20/25 at session start). Three failures remain (DV-DELETE returning 1 row, DV-in-use same, CM+DV Int32/Int8 assertion, input_file_name returning empty Set) -- all tracked under apache#168. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
…Type Delta declares `__delta_internal_is_row_deleted` as `ByteType` (Spark = JVM `Byte` = Arrow `Int8`). We were emitting it as `Int32`, which made DataFusion's interval-propagator panic with `Only intervals with the same data type are intersectable, lhs:Int32, rhs:Int8` whenever the upstream `Filter(__delta_internal_is_row_deleted = 0)` was processed against a literal that Spark types as Byte. The failure surfaced on the column- mapping + DV combined test path, where stats propagation runs over both the column-mapping rewriter output AND the synthetic-column filter. Drop the column from `Int32Array` to `Int8Array`; values stay 0/1. Test status: 24/25 contrib Scala tests pass (was 23/25). "column mapping + deletion vectors combined" now passes. The remaining failure is the DV-in-use second-DELETE freshness issue, where Delta materializeable DV-descriptor cache returns the first DELETE's cardinality after a second DELETE on the same file -- tracked separately under apache#168. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
schenksj added a commit
to schenksj/datafusion-comet
that referenced
this pull request
May 21, 2026
`preparedScan.files` returns the AddFile list captured at FileIndex
construction time. When the same logical path is read multiple times
across DML statements, the cached snapshot can hand back stale DV
descriptors. Try `matchingFiles(Nil, Nil)` first -- it asks Delta for
the live snapshot's matching files and picks up fresh DV info -- and
only fall back to `preparedScan.files` when the live call isn't
available.
Test status: still 24/25 contrib Scala tests pass. The remaining
failure ("deletion vectors: accelerates DV-in-use tables via native DV
filter") is the *second* DELETE on the same file failing to reflect in
the read; root-cause analysis (this session) shows Delta keeps the same
on-disk DV descriptor for the second DELETE -- vanilla works because
PreprocessTableWithDVs computes is_row_deleted at runtime via a UDF
that consults the LATEST in-memory DV bitmap. Tracked under apache#168.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Closes#167.
Rationale for this change
What changes are included in this PR?
How are these changes tested?