fix: preserve Spark semantics for dictionary-encoded Parquet inputs and reject dictionary targets

[peterxcli]

Which issue does this PR close?

Closes #5096.

Rationale for this change

Comet can receive Arrow dictionary arrays as physical Parquet input before schema adaptation. Spark, however, cannot express Dictionary as a SQL cast target. Spark infers a Parquet schema from its own metadata or the Parquet physical schema, while arrow-rs also honors ARROW:schema; an unannotated Parquet BINARY column can therefore reach Comet as Dictionary(Int32, Binary) while Spark expects StringType when binaryAsString is enabled.

That reachable source path must cast dictionary values with Spark semantics before Arrow unpacks the dictionary. The previous implementation conflated reachable dictionary sources with unreachable dictionary targets, hard-coded Int32 dictionary keys, and duplicated Arrow's dictionary handling in the Parquet conversion path.

What changes are included in this PR?

• Reject Dictionary cast targets with a clear internal error because Spark cannot produce them.
• For reachable string and binary dictionary sources, cast the dictionary values with Spark semantics, replace the values through Arrow's key-agnostic dictionary API, and use cast_with_options to unpack the result.
• Remove the hand-written Parquet dictionary branch and rely on the existing Arrow can_cast_types / cast_with_options fallback.
• Add Spark source links documenting the cast-target and Parquet-schema invariants.
• Add a native Parquet SQL regression that writes Dictionary(Int32, Binary) metadata and invalid UTF-8, then verifies Comet matches Spark's replacement-character behavior.

How are these changes tested?

• cargo test --manifest-path native/Cargo.toml -p datafusion-comet-spark-expr --lib (595 passed)
• cargo test --manifest-path native/Cargo.toml -p datafusion-comet --lib (135 passed, 4 HDFS tests ignored)
• cargo clippy --manifest-path native/Cargo.toml -p datafusion-comet-spark-expr -p datafusion-comet --lib --tests -- -D warnings
• cargo fmt --manifest-path native/Cargo.toml --all -- --check
• Focused native Parquet SQL regression with the default Spark 4.1 profile
• Focused native Parquet SQL regression with -Pspark-3.5
• ./mvnw spotless:check -DskipTests -Dskip-rat
• git diff --check

As a negative control, removing the source-dictionary handler makes the SQL regression fail: Spark returns f�o, while Comet returns null.

[andygrove]

Thanks for picking this up. Inverting the order so the Spark cast owns the value semantics and Arrow owns the dictionary construction is the right shape, and deleting the parquet branch is a nice cleanup.

I spent some time verifying the two preconditions the issue asked about, and both hold up:

• can_cast_types(Dictionary(_, value_type), to) in arrow-cast 58.4 delegates straight to can_cast_types(value_type, to), and the dict-to-dict arm delegates to the value types. So the can_cast_types fallback in parquet_convert_array covers exactly the same set of combinations the old recursion covered through its own arrow fallback. No combination that used to convert now falls through to _ => Ok(array).
• None of the Comet-specific arms in parquet_convert_array has Utf8 or LargeUtf8 as its from-type. They are Struct, List, Timestamp(us, None), Map, and FixedSizeBinary(16). Since the deleted branch restricted values to Utf8/LargeUtf8, the recursion could only ever reach identity or the arrow fallback, so nothing Spark-specific is lost by deleting it.

Dropping the spark_cast_postprocess call on the new return also looks safe. All four of its arms require to_type to be Int64 or Utf8, so with a Dictionary target it was always the identity.

I also liked the test data choice in test_cast_to_dictionary_deduplicates_casted_values. "0.2" and "." both cast to 0, matching Spark's UTF8String.toInt truncation behavior, so it is a real post-cast collision rather than a contrived one.

A few things I would like to work through before this merges.

The new path supports fewer dictionary value types than the old one

arrow::compute::cast reaches cast_to_dictionary (arrow-cast-58.4.0/src/cast/dictionary.rs:178), and that function has no arm for Boolean. It also has none for Null or any nested type. Those all land on _ => Err("Unsupported output type for dictionary packing: {dict_value_type}").

Under the old code, cast_array(StringArray, Dictionary(Int32, Boolean)) worked. dict_from_values wrapped the strings into Dictionary(Int32, Utf8), the recursion then hit the from-dictionary arm, and spark_cast_utf8_to_boolean was applied to the values. Under the new code the same call Spark-casts to Boolean correctly and then fails at the packing step.

What makes this awkward to guard against is that can_cast_types disagrees with cast here. can_cast_types(Boolean, Dictionary(Int32, Boolean)) returns true, because the (_, Dictionary(_, value_type)) arm just recurses on the value type. So a caller cannot probe for support first.

Could you add a test for StringArray -> Dictionary(Int32, Boolean) so we pin down what the behavior is either way? None of the existing tests cover a boolean dictionary target, which is why CI is green on this. If it does regress, one option is to keep constructing the dictionary manually when Arrow cannot pack the value type.

The value cast is threaded with cast options but the packing call is not

Would you mind using cast_with_options(&values, to_type, &native_cast_options) instead of the bare arrow::compute::cast? Every other arrow call in cast_array threads the options through, and the bare cast silently picks up CastOptions::default(). I do not think it changes results today, since values already has the target value type and the inner cast inside pack_numeric_to_dictionary is an identity, but it is the one call in the function that drops the eval mode.

The surviving from-dictionary arm still hand-rolls, and it ignores the requested key type

This one predates the PR, but it is the same function and the same class of bug you just fixed on the parquet side, so I would rather not leave it floating.

The Dictionary(Int32, Utf8 | LargeUtf8 | Binary | LargeBinary) arm in cast_array is hardcoded to DictionaryArray::<Int32Type>. For to_type = Dictionary(Int16, Int32) it returns Dictionary(Int32, Int32), so the returned DataType does not match what was requested. After this PR, cast_array honours the requested key type when the source is plain and ignores it when the source is a string or binary dictionary, which is a confusing split.

Arrow's dictionary_to_dictionary_cast already handles both the key cast and the value cast, and it errors cleanly when keys do not fit rather than silently returning the wrong type, so this arm may be able to collapse the same way. If that is out of scope here, could you file a tracking issue and link it from this PR? There is also an Arc::new(casted_dictionary.clone()) in that arm cloning something already owned, if you end up touching it.

Test coverage

The two new tests are well targeted at what changed. Two gaps I would like closed:

• Could you add a case with a non-Int32 key type on the target, say Dictionary(Int8, Utf8)? Your parquet test covers Int32 -> Int16 keys but the cast test does not, and that is exactly where the key-type question above lives.
• A case for a non-string source would be good too, for example Int32Array -> Dictionary(Int32, Int64). The old path could not reach that, so it is new behavior worth pinning down.

One question on scope

It would help if the description said whether any real query plan can request a Dictionary cast target. The doc comment on spark_cast_postprocess says "Spark cannot specify Dictionary as to_type", and the required schemas that reach SparkSchemaAdapter come from Spark types, which never carry dictionaries. If that is right, the nested-dictionary bug you fixed on the parquet side is not user-reachable today and this is defensive cleanup, which is worth stating so reviewers know why there is no SQL-level test. If there is a reachable path, an end-to-end test would be valuable.

[peterxcli]

@andygrove Thanks for the detailed review, especially for verifying that the generic Parquet fallback fully covers the deleted branch. I pushed a follow-up in 9242ea3ab.

After tracing the Spark and Comet paths, I ended up separating dictionary sources from dictionary targets.

Dictionary targets

I confirmed that a real Spark plan cannot request a Dictionary cast target. Spark has no Dictionary SQL type, Comet’s protobuf type system cannot serialize one, and the normal scan/shuffle boundaries expose dictionary value types to downstream expressions. I also checked the historical CASE regression at runtime; its native coercion is now Utf8 + Utf8 → Utf8, not Dictionary.

I therefore removed target packing instead of adding Boolean, non-Int32 key, and non-string target coverage. cast_array now rejects every Dictionary target explicitly, and the former target-success test is now a rejection test. This also removes the bare arrow::compute::cast packing call, so there is no longer a call that drops native_cast_options.

With Dictionary targets unsupported, there is no remaining requested-key mismatch to track separately.

Dictionary sources

The source-Dictionary path is reachable during native Parquet schema adaptation. Spark ignores ARROW:schema, while arrow-rs honors it before adapting the physical array to Spark’s requested logical type.

I rewrote that handler to use as_any_dictionary and with_values, so it is no longer hard-coded to Int32 keys and no longer clones a newly owned dictionary. It Spark-casts the dictionary values first, then uses Arrow’s cast_with_options with native_cast_options to unpack the result.

SQL coverage

I added an end-to-end native Parquet test. It writes an unannotated Parquet BINARY column with ARROW:schema declaring Dictionary(Int32, Binary), enables Spark’s binaryAsString, and includes invalid UTF-8.

As a negative control, removing the source-Dictionary handler makes the test fail: Spark returns f�o, while Comet returns null. With the handler, the results match. I ran it with Spark 4.1 and Spark 3.5.

The hand-written Parquet dictionary branch remains deleted, and the synthetic Parquet Dictionary-target test was removed because its target cannot originate from a Spark schema.