fix: reject incompatible decimal precision/scale in native_datafusion scan

[andygrove]

Which issue does this PR close?

Closes #4089.

Rationale for this change

When the native_datafusion scan reads a Parquet column whose physical type is Decimal(p1, s1) under a requested read schema of Decimal(p2, s2) with s2 < s1, the existing schema adapter falls through to Spark's Cast expression. Cast happily truncates fractional digits, producing wrong values silently. Spark's vectorized reader rejects this with SchemaColumnConvertNotSupportedException, and native_iceberg_compat already does the same via TypeUtil.checkParquetType. The native scan should match.

What changes are included in this PR?

native/core/src/parquet/schema_adapter.rs: in replace_with_spark_cast, add a guard before the existing branches that returns DataFusionError::Plan when both physical_type and target_type are Decimal128 and the target scale is smaller than the source scale.

The check is intentionally narrow:

• Only fires on scale narrowing. Precision-only changes with the same scale (e.g. Decimal(5,2) read as Decimal(3,2)) are still allowed and fall through to Spark's Cast, which produces null on per-value overflow. This matches Spark 4.0's parquet-mr fallback behavior exercised by ParquetTypeWideningSuite's parquet decimal type change Decimal(5, 2) -> Decimal(3, 2) overflows with parquet-mr test.
• Does not touch any other type path. The closed prior attempt at broad schema validation ([native_datafusion] [Spark SQL Tests] Schema incompatibility tests expect exceptions that native_datafusion handles gracefully #3311) broke unrelated tests; this one does not.

How are these changes tested?

Added a focused test to ParquetReadSuite: native_datafusion rejects incompatible decimal precision/scale. It writes Decimal(10, 2) data, reads it under Decimal(5, 0) (scale narrowed from 2 to 0), forces spark.comet.scan.impl=native_datafusion and spark.sql.sources.useV1SourceList=parquet, and asserts collect() raises SparkException. Verified against ParquetReadV1Suite (44 tests, all pass; 1 pre-existing test ignored).

The behavior is also covered by the per-impl matrix added in #4087 (decimal(10,2) read as decimal(5,0): native_datafusion), whose assertion will need flipping from "succeeds" to "throws" once that PR merges.

[mbutrovich]

In a case where we expect an exception to be generated anyway, can we catch this at CometScanRule rather than going all the way to serialization and native operators?

[andygrove]

In a case where we expect an exception to be generated anyway, can we catch this at CometScanRule rather than going all the way to serialization and native operators?

I think the issue is that we do not know the types of all the parquet files until runtime?

[mbutrovich]

I think the issue is that we do not know the types of all the parquet files until runtime?

IIRC from looking at this a while back, Spark has read the physical schema already, but thrown it away by the time our Comet rules run with no good way to get it again.

I'm not opposed to handling it this way, just wanted to think through if we could catch it earlier. It's also a fairly uncommon scenario.

[andygrove]

I'm not opposed to handling it this way, just wanted to think through if we could catch it earlier. It's also a fairly uncommon scenario.

I do think this is an edge case that is fairly unlikely IRL because it only happens when the user provides a schema that is incompatible with the file schema

[mbutrovich]

Thanks @andygrove! I double-checked Spark's logic and the scale-narrowing guard looks right for preventing silent truncation.

Spark's isDecimalTypeMatched (line 1169 of ParquetVectorUpdaterFactory.java) requires exact scale match, so scale widening like Decimal(10,2) -> Decimal(10,4) would also be rejected by Spark's vectorized reader but passes through here. The PR description explains why that's fine (lossless cast), which makes sense. Would it be worth noting that in the code comment so future readers don't have to check the PR description?

[andygrove]

Expanded the comment in replace_with_spark_cast to document both the precision-only and scale-widening cases and to reference isDecimalTypeMatched directly. Pushed in fcbec76.

[mbutrovich]

Thanks @andygrove!

[andygrove]

Merged. Thanks @mbutrovich