Reading Parquet with an explicit schema silently narrows values instead of failing like Spark

[davidlghellin]

Description

When reading Parquet with an explicit StructType, Spark only allows a small set of safe upcasts between the declared type and the type stored in the file. Any other conversion is rejected at read time with FAILED_READ_FILE.PARQUET_COLUMN_DATA_TYPE_MISMATCH.

Sail performs the cast instead, and returns silently converted values. For narrowing conversions this is silent data loss: 9.5 is read back as 9, and the integers 1, 2, 3 are read back as true, true, true.

This matters most in the setup where it was found: no catalog at all, reading files directly with a declared schema. There the StructType is the only type contract there is, so a cast that Spark treats as a hard error goes completely unnoticed.

Steps to reproduce

Parquet file written with id INT, score DOUBLE, values (1, 9.5), (2, 7.25), (3, NULL).

import os
import tempfile

from pyspark.sql import SparkSession
from pyspark.sql.types import DoubleType, IntegerType, StructField, StructType

remote = os.environ.get("SPARK_REMOTE", "sc://localhost:50051")
if remote == "local":
    # PySpark's builder treats SPARK_REMOTE as a Connect URL, so it has to go
    # before we can ask for a local master.
    os.environ.pop("SPARK_REMOTE", None)
    spark = SparkSession.builder.master("local[2]").getOrCreate()
    engine = "SPARK JVM"
else:
    spark = SparkSession.builder.remote(remote).getOrCreate()
    engine = "SAIL"

path = os.path.join(tempfile.mkdtemp(), "t.parquet")
spark.createDataFrame(
    [(1, 9.5), (2, 7.25)],
    StructType([StructField("id", IntegerType()), StructField("score", DoubleType())]),
).write.parquet(path)

print(f"===== {engine} =====")
print("file schema:    ", spark.read.parquet(path).schema.simpleString())

# The file stores score as DOUBLE. Declare it as INT on purpose.
narrowed = StructType([StructField("score", IntegerType())])
print("declared schema:", narrowed.simpleString())

try:
    rows = spark.read.schema(narrowed).parquet(path).collect()
    print(f"RESULT: {rows}   <-- 9.5 and 7.25 were truncated, no error raised")
except Exception as e:  # noqa: BLE001
    first = " ".join(str(e).split())[:200]
    print(f"RAISED: {type(e).__name__}: {first}")

Run against Sail (server on port 50051):

SPARK_REMOTE="sc://localhost:50051" hatch run python repro_parquet_schema_cast.py

Run against Spark JVM:

source "$HOME/.sdkman/bin/sdkman-init.sh" && sdk use java 17.0.9-tem SPARK_REMOTE="local" hatch run python  repro_parquet_schema_cast.py

===== SAIL =====
file schema:     struct<id:int,score:double>
declared schema: struct<score:int>
RESULT: [Row(score=9), Row(score=7)]   <-- 9.5 and 7.25 were truncated, no error raised

===== SPARK JVM =====
file schema:     struct<id:int,score:double>
declared schema: struct<score:int>
.....
RAISED: Py4JJavaError: An error occurred while calling o54.collectToPython. : org.apache.spark.SparkException: [FAILED_READ_FILE.PARQUET_COLUMN_DATA_TYPE_MISMATCH] Encountered error while reading file file:///var/folders/p4