fix: reject a file without field ids at any depth whether or not id matching is on

[dwsmith1983]

Which issue does this PR close?

Closes #5936.

Rationale for this change

Spark's ParquetReadSupport.getRequestedSchema raises "Spark read schema expects field Ids, but Parquet file schema doesn't contain any field Ids" whenever the requested schema carries a parquet.field.id at any depth and the file carries none at any depth, unless spark.sql.parquet.fieldId.read.ignoreMissing is set. It does this on every read, for both readers, without consulting spark.sql.parquet.fieldId.read.enabled. The same code is on branch-3.5 and branch-4.0.

The native scan ran that check only when the read flag was on and only over root fields, and it ran it inside the name remap, which a case-sensitive session with the flag off never reaches. Two differences followed. A read schema with ids over a file without ids returned rows where Spark raises. A file whose ids sit only on nested fields was rejected where Spark reads it and null-fills the unmatched fields.

What changes are included in this PR?

• any_nested_field_has_id in parquet_support.rs walks struct children, map keys and values and every list representation, mirroring Spark's ParquetUtils.hasFieldIds and containsFieldIds.
• The check moves to the top of SparkPhysicalExprAdapterFactory::create, before the remap, and reads ignore_missing_field_id and the two recursive predicates only. remap_physical_schema no longer takes ignore_missing_field_id. Id matching itself is unchanged and still gated on the read flag and root ids, as Spark's clipParquetGroupFields gates it per struct level.
• The Iceberg scan sets ignore_missing_field_id, since Iceberg resolves columns by id itself and its reader supplies the ids on every field. That path could not reach the check before and cannot now.

No JVM change: root ids and nested ids already reach the native requested schema regardless of the read flag.

How are these changes tested?

Rust tests in schema_adapter.rs drive the real scan path: a read schema with ids over a file without ids is rejected with the read flag off, at the root and when the only id is on a struct child; a file whose ids sit only below the root is read with the unmatched root null-filled; ignoreMissing still suppresses the rejection; a read with no ids on either side is unchanged. A test in parquet_support.rs covers the predicate at every depth and for every list type. The first three scan tests fail before the change.

ParquetReadSuite gains both halves of Spark's ParquetFieldIdIOSuite."global read/write flag should work correctly": with the read flag off, a read schema with ids over a file without ids raises in Spark and in Comet, and reads by name under ignoreMissing; with the read flag off and a file that carries ids, differently named fields read as nulls with no error. A Rust test pins the same non-raising case through the scan. A second test reads a file whose ids sit only on nested fields with the flag on and checks the answer and the native operator against Spark. Both fail on main.

ParquetReadV1Suite, CometNativeReaderSuite, CometJoinSuite and CometScanRuleSuite pass on Spark 3.5. CometJoinSuite."Broadcast coalescing falls back for array field metadata mismatch" reads an id-bearing schema over an id-free file with ignoreMissing set, so it is the one existing test whose result depends on the new rule being suppressed by the conf, and it still passes.

Spark's own ParquetFieldIdIOSuite exercises this path and runs in the Spark SQL job rather than the pull request tier, and the Iceberg scan is touched, so this needs the run-spark-4.1-tests and run-iceberg-tests labels from a maintainer.

[andygrove]

The read of ParquetReadSupport.getRequestedSchema is right. I checked 3.5.8 and 4.1.1 and the check really is !ignoreMissingIds && !containsFieldIds(fileSchema) && hasFieldIds(requestedSchema), with no reference to PARQUET_FIELD_ID_READ_ENABLED, and both helpers really are recursive. Moving it ahead of the remap is the right shape, and the recursive predicate is a clear improvement over the root-only one.

There is a problem with which schema the logical half of the check runs against. It makes Comet raise on a read that Spark serves, and I have a reproducer that passes on 0e60a0c79 and fails here. Details inline on schema_adapter.rs.

Locally on the default profile I get 195 passed for cargo test -p datafusion-comet --lib parquet:: and 71 passed for ParquetReadV1Suite, including all three of the new tests, so the new coverage holds on 4.1 as well as on 3.5.

CI has not run anything yet. Every workflow on the head commit is still sitting at action_required, so there is no signal on this beyond what I ran locally.

[andygrove]

One thing about which schema the logical half of this check runs against. create gets DataFusion's logical_file_schema, and that resolves to TableSchema::file_schema(), which init_datasource_exec builds from data_schema whenever the projection resolves by name. Spark's catalystRequestedSchema comes from SPARK_ROW_REQUESTED_SCHEMA, which ParquetFileFormat.setupHadoopConf sets to requiredSchema, so Spark is looking at the pruned schema. That makes a read schema carrying an id on a column the query never projects raise here where Spark reads it.

I think this is also why it never misfired before the change. CometNativeScan.scala:265 sends useFieldId = PARQUET_FIELD_ID_READ_ENABLED && ParquetUtils.hasFieldIds(scan.requiredSchema), so the old should_match_by_id gate was already carrying Spark's exact predicate over the pruned schema, and dropping the gate dropped that with it.

Here is what I ran. It passes on 0e60a0c79 and fails on this branch, with the read flag both off and on:

val writeSchema = new StructType().add("a", IntegerType).add("b", IntegerType)
// two rows written, no field ids anywhere in the file
val readSchema = new StructType()
  .add("a", IntegerType, true, withId(1))
  .add("b", IntegerType, true)
spark.read.schema(readSchema).parquet(path).select("b").collect()

Spark gives [2], [4]. Comet raises FAILED_READ_FILE.NO_HINT caused by RuntimeException: Spark read schema expects field Ids, but Parquet file schema doesn't contain any field Ids.

Would it work to compute the logical half at plan time from required_schema, which init_datasource_exec already has in scope? Something like spark_parquet_options.requested_schema_has_field_ids = any_nested_field_has_id(required_schema.fields()) next to the existing use_field_id assignment, then testing that flag here instead of walking logical_file_schema. That also saves re-walking the read schema on every file open. If you would rather match Spark exactly than reimplement hasFieldIds in Rust, the JVM already computes ParquetUtils.hasFieldIds(scan.requiredSchema) at CometNativeScan.scala:266 and it could ride over on the proto instead.

Could you also add the pruned-projection case to ParquetReadSuite? It is the one shape none of the current tests cover, and it is the one that regressed.

[andygrove]

Agreed this path should not be subject to the check, and setting the flag keeps Iceberg behaving exactly as it does on main, which is the safe choice.

If you take the required_schema approach from the other comment, the new flag defaults to false in SparkParquetOptions::new, so the check is already inert here and this line stops doing anything. Worth either dropping it then or reworking the comment so it does not read as load-bearing when it is not.

[andygrove]

field_holds_id looks at the metadata on list element fields and on map key, value and entries fields. Spark's hasFieldIds cannot see any of those, because ArrayType and MapType recurse into elementType and keyType / valueType, and only a StructField can carry the metadata. On the read path the serde never populates them, since schema2Proto(scan.requiredSchema) leaves parentField as None and nestedParquetFieldId then returns None, so today the two agree. It is only the file-side walk that needs to reach those nodes, to match containsFieldIds over the raw MessageType.

Could you note in the doc comment that the element and key/value checks are there for the physical schema? Otherwise a later change that starts emitting collection ids on the read path turns this into a false positive with nothing to warn against it.