Investigate mixed-type duplicate Parquet roots: Spark field-ID reads return anomalous values

[erikbogado-nstech]

Compatibility boundary: mixed physical types under duplicate root names

Related: #5884, #5654, #5786. This tracks a specific mixed-type boundary separately from nested duplicate rejection and malformed row-count handling.

On Linux x86_64 with JDK 17.0.20.1, Spark 4.1.3, vectorized Parquet reader, case-sensitive resolution, field-ID reads enabled, spark.sql.parquet.filterPushdown=false:

• Valid file: root a: INT64 (id=1) followed by a: INT32 (id=2). Each physical column contains [1,3,null]; one row group, three rows, three values per column.
• Read each ID separately as a renamed BIGINT: ID1 returns [1,3,null]; ID2 returns [1,0,null]. Reproduced with Spark batch sizes 4096 and 1.
• The earlier local Comet first-wins experiment returned [1,3,null] for both aliases. That was an unpublished experimental candidate based on fix: match Spark's duplicate field and field id semantics in parquet field lookup #5654 commit 1fa518b54ea68de77577dc61be74a470e181c51b, Comet 1.1.0-SNAPSHOT, DataFusion 55.1.0 / parquet 59.3.0, with local alias/first-root changes. This is not a claim about released Comet or current fix: match Spark's duplicate field and field id semantics in parquet field lookup #5654 head.
• Expressed as paired results: Spark [1,1],[3,0],[null,null] versus candidate Comet [1,1],[3,3],[null,null]. Spark pairs were assembled from two separate single-ID reads, not a simultaneous two-column Spark read.
• Non-vectorized Spark reads throw FAILED_READ_FILE.NO_HINT in this probe; switching readers is not a demonstrated workaround.

Normal first-physical-column selection plus numeric conversion would yield [1,3,null] for both IDs. The observed zero is anomalous; neither byte-compatible replication nor a corruption root cause is established or accepted.

Reproduce without a binary attachment

Generate the fixture with this standalone Rust program. Cargo.toml:

[package]
name = "mixed-root-repro"
version = "0.1.0"
edition = "2021"
[dependencies]
arrow = "=59.3.0"
parquet = { version = "=59.3.0", features = ["arrow"] }

src/main.rs (same ArrowWriter/default-properties construction as the observed fixture):

use std::{collections::HashMap, fs::File, sync::Arc};
use arrow::{array::{ArrayRef, Int32Array, Int64Array},
    datatypes::{DataType, Field, Schema}, record_batch::RecordBatch};
use parquet::arrow::ArrowWriter;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let field = |ty, id: &str| Field::new("a", ty, true).with_metadata(
        HashMap::from([("PARQUET:field_id".to_string(), id.to_string())]));
    let schema = Arc::new(Schema::new(vec![
        field(DataType::Int64, "1"), field(DataType::Int32, "2")]));
    let columns: Vec<ArrayRef> = vec![
        Arc::new(Int64Array::from(vec![Some(1), Some(3), None])),
        Arc::new(Int32Array::from(vec![Some(1), Some(3), None]))];
    let mut writer = ArrowWriter::try_new(File::create("mixed-root.parquet")?,
        schema.clone(), None)?;
    writer.write(&RecordBatch::try_new(schema, columns)?)?;
    writer.close()?;
    Ok(())
}

Run cargo run, then in Spark 4.1.3 spark-shell with Comet disabled:

import org.apache.spark.sql.types._
spark.conf.set("spark.comet.enabled", "false")
spark.conf.set("spark.sql.caseSensitive", "true")
spark.conf.set("spark.sql.parquet.fieldId.read.enabled", "true")
spark.conf.set("spark.sql.parquet.filterPushdown", "false")
spark.conf.set("spark.sql.parquet.enableVectorizedReader", "true")
spark.conf.set("spark.sql.parquet.columnarReaderBatchSize", "4096")
for (id <- Seq(1L, 2L)) {
  val schema = new StructType(Array(StructField("x", LongType, true,
    new MetadataBuilder().putLong("parquet.field.id", id).build())))
  println(s"id=$id: " + spark.read.schema(schema)
    .parquet("mixed-root.parquet").collect().toSeq)
}

Upstream Spark/parquet-java investigation

Spark uses JIRA, not GitHub issues. No upstream Spark ticket has been filed by this investigation yet; please link any existing exact upstream report here.

Relevant Spark 4.1.3 source: checkColumn obtains the file descriptor by name path; initColumnReader constructs the decoder from the requested descriptor.

An isolated probe read the raw Thrift footer: INT64/id1 then INT32/id2, each with three values. By contrast, parquet-java's converted ColumnChunkMetaData reported INT32 for both columns, and MessageType.getColumns() returned the INT32/id2 descriptor twice. Name-based descriptors therefore cannot establish physical leaf identity for this file.

A local decoder experiment selected fileSchema.getType(0) into a one-field MessageType, set that requested schema on a fresh ParquetFileReader, then constructed Spark's VectorizedColumnReader with that selected INT64 descriptor. An assertion checked [1,3,null] and passed. This demonstrates a viable physical-descriptor seam, not a complete Spark SQL fix or proof of every step causing the anomalous result.

Next: trace descriptor/page association end to end, establish upstream intended behavior, and retain the failing Spark/Comet comparison. Do not silently turn the anomalous zero into Comet acceptance. #5786's green nested-only patch is independent of this experiment.

[ErikBPF]

Update: the investigation now includes SQL-level regression coverage, beyond the earlier decoder probe. The parquet-java fix is published as apache/parquet-java#3796, with reproduction tracked in apache/parquet-java#3795. Its current-master port passes 112 targeted tests plus Spotless and package checks.

The isolated Spark candidate selects the first physical root descriptor while retaining requested field IDs and suppresses ambiguous predicate pushdown without dropping logical filters. Targeted release-based owner checks pass: 8 duplicate-root tests and 101 filter tests, with scalastyle clean. No Spark PR has been opened yet: JIRA tracking, a current-master port and a normal published parquet dependency pin remain prerequisites.

These are bounded experimental and targeted-module results, not released fixes or full-reactor certification. Simultaneous aliases and duplicate nested, encrypted or non-vectorized reads remain outside validated scope. Anomalous Spark values are not adopted as desired behavior.