Skip to content

Make the experimental Spark API work as advertised against PySpark 4.2.0 and 3.5.9 - #686

Merged
evertlammerts merged 4 commits into
duckdb:v2.0-cyanopterafrom
evertlammerts:pyspark
Oct 17, 2026
Merged

evertlammerts merged 4 commits into
duckdb:v2.0-cyanopterafrom
evertlammerts:pyspark

Conversation

@evertlammerts

Copy link
Copy Markdown
Member

The experimental Spark API (duckdb.experimental.spark) had drifted a long way from what it claims to be, a drop-in subset of PySpark on DuckDB. A clean install of 1.5.6 could not even import it: _globals.py and _typing.py imported typing_extensions and numpy at module level, neither of which the duckdb package declares (#653). Beyond that, I graded the module against real Spark 4.2.0 on 64 typical snippets and 47 edge cases, and found that 18 of the snippets raised and 12 of the edge cases returned a different value than Spark without any error. Joins with repeated column names were the worst: after a second join, or a withColumn or drop, both columns of a repeated name silently carried the values of the first one.

This PR makes the module do what it says, with a fixed compatibility target: PySpark 4.2.0 and 3.5.9. Where the two versions differ, the module follows 4.2.0 unless it cannot. The rule for everything else is simple. Behavior is mapped mechanically onto DuckDB where that gives Spark's answer; where it does not, the module raises rather than returning a different value; a small number of cases where DuckDB's answer is deliberately kept are documented (38 of them, which I will put on the Spark API docs page). The module does not re-implement Spark semantics: an earlier version of this work had a CSV reader that parsed values the way Spark does, and I took it out again in favor of DuckDB's casts and type detection.

What changed, by area:

  • Imports: no module-level dependency on anything the duckdb package does not declare. pandas, pyarrow and pytz stay optional at the call sites that need them. A test imports the module with each optional package blocked.
  • Names: every function and expression gets Spark's column name, and a frame keeps Spark's names separate from DuckDB's, so repeated names (id, id) survive instead of becoming id, id_1. Columns of a join input are referenced by position, and every name resolution checks for ambiguity and raises AMBIGUOUS_REFERENCE where Spark does.
  • Values: integer casts truncate toward zero like Spark; set operations match struct fields by position; timestamps are collected as epoch microseconds and converted with Spark's own rules, so collect(), toPandas() and show() agree with Spark in any process or session time zone and no longer need pytz; dozens of function-level differences in NULL handling, sequence, regexp_replace, padding, binary concat, and so on.
  • UDFs: register returns a usable callable, a UDF named like a builtin neither shadows nor is replaced by it, struct arguments arrive as Rows.
  • Readers and writers: Spark's save modes (a plain write.csv() no longer overwrites silently), load() by file extension and with a schema (it returned all-NULL rows for a CSV with a schema), _c0 column names, nullValue, header checks against the schema, and option flags that are either honored or raise instead of being accepted and ignored.
  • Input: createDataFrame verifies rows and pandas input against the schema the way Spark does; a list of dicts works (DuckDB Spark API is incompatible with the PySpark API's spark.createDataFrame(list of dict) #183); a statement without a result in spark.sql() gives a frame with an empty schema (Crashes while handling non-select result set (DataFrame) #428; the rest of that issue needs relations without columns in the engine).
  • Errors: Spark's error classes for the cases above, with seven classes added to the module's table from Spark 4.2.0's definitions.

Four binding bugs came out of this work and are fixed in the same change, all confirmed new:

  • StarExpression(exclude=[ColumnExpression("b", "id")]) dropped the qualifier and excluded every id; the name parts now map to column, table, schema and catalog, and an empty name or more than four parts raise ValueError.
  • ColumnExpression("s.a.b.c") with a single dotted string kept only the first, the second to last and the last part on this branch (1.5.6 raised instead), so it bound another struct field. It now keeps every part, as the multi-argument form does.
  • duckdb.struct_type({}) refused the empty struct, a check the engine dropped long ago.
  • A NaN inside a list, struct, map or typed Value parameter became NULL, because the nested conversions dropped the nan_as_null flag the parameter path sets.

Four open community PRs addressed parts of this, and I reimplemented their fixes rather than merging them, because the surrounding code changed under them: #552 (show() signature and the UDF return type bug), #388 (inference for createDataFrame from dicts), #378 (struct() of a list) and #626 (path quoting in load()). The authors are credited: wavebyrd's and Raaif-Yousuf's commits are cherry-picked, and all four are co-authors of the main commit.

What this does not do: the module's coverage of PySpark is unchanged, about a third of the API, with no Window; missing API raises AttributeError or ContributionsAcceptedError as before. The engine's relational API has a bug that this module contains with an extra projection rather than fixes: an aggregate, filter or projection over a limit, order or distinct of a join relation merges into the join's own SELECT beneath the modifier, so join.limit(1).aggregate("count(*)") gives the unlimited count. I will file that separately.

Evidence. The Spark tests now run in three modes: against the module, against real PySpark 4.2.0 and against real PySpark 3.5.9 (USE_ACTUAL_SPARK=true), plus a run in a non-UTC time zone. On this tree: 1813, 1757 and 1745 tests pass, 0 fail (before: 289 tests, with 11 and 15 failures on the two Spark versions). PySpark's own function doctests, run against the module with the names aliased, pass fully for 111 of 185 functions (3 before). The 64 snippets graded against Spark 4.2.0: 48 match, 16 raise, 0 differ silently (45, 18 and silent differences before). The full tests/fast suite passes (8089 tests). The binding fixes were checked under AddressSanitizer and LeakSanitizer against an unmodified build, with no new report.

Resolves #653 and #183, partly #428.

wavebyrd and others added 3 commits October 16, 2026 13:44
The struct() function now unwraps a single list or set argument,
matching the PySpark API behavior and the existing array() function
in this codebase.
load() without a format built its query as select * from {path},
splicing the raw path straight into the SQL text, so any path with a
directory separator in it fails to parse. That is every path except a
bare filename in the working directory, and on Windows the backslash
fails as well.

Quote it as a string literal instead, the same way the rest of the
reader and writer methods hand paths to DuckDB, and escape embedded
single quotes by doubling them.
…2.0 and 3.5.9

Co-authored-by: Simon Kesenci <god@skri.org>
Co-authored-by: Mario Taddeucci <mariotaddeucci@gmx.com>
Co-authored-by: wavebyrd <wavebyrd@gmail.com>
Co-authored-by: Raaif-Yousuf <raaifirfanyousuf@gmail.com>
@evertlammerts
evertlammerts added this pull request to the merge queue Oct 16, 2026
Merged via the queue into duckdb:v2.0-cyanoptera with commit 8796909 Oct 17, 2026
57 of 59 checks passed
@evertlammerts
evertlammerts deleted the pyspark branch October 17, 2026 03:18
evertlammerts added a commit that referenced this pull request Oct 19, 2026
The `Wheel: cp311-win_amd64` release job has failed since #686 landed:

test_numpy_array_literals[array3-ARRAY(5L, -3L)-array(5,
-3)-element_type3-value3]
AssertionError: assert ArrayType(IntegerType(), True) ==
ArrayType(LongType(), True)

The cp311 test group pins `numpy<2` for tensorflow, and numpy 1.x on
Windows uses
int32 as the default integer type (the C `long`). `np.array([5, -3])` is
therefore an
int32 array there, and `lit` types it as `ARRAY<INT>`, which is what
Spark does for an
int32 array as well. The test expected `ARRAY<BIGINT>`, assuming a 64
bit default.

The case now spells `dtype=np.int64`, so it checks the int64 mapping on
every platform.
The int32 mapping is already covered by the neighbouring `np.array([5],
dtype=np.int32)`
case.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

duckdb.experimental.spark fails to import without typing_extensions, which is not a runtime dependency

3 participants