andygrove commented on code in PR #6116:
URL: https://github.com/apache/datafusion-comet/pull/6116#discussion_r4075661959
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -852,6 +849,21 @@ impl PhysicalExprAdapterFactory for
SparkPhysicalExprAdapterFactory {
// to the original physical names. This is necessary because
downstream code
// (reassign_expr_columns) looks up columns by name in the actual
stream schema,
// which uses the original physical file column names.
+ //
+ // Before any of that, mirror the eager check in Spark's
`ParquetReadSupport`: a read
+ // schema that carries field ids at any depth may not read a file that
carries none at
+ // any depth, unless `ignoreMissing` is set. Spark applies this check
whether or not
+ // `fieldId.read.enabled` is on, so it runs before the id matching
gate below and does
+ // not depend on the remap.
+ if !self.parquet_options.ignore_missing_field_id
Review Comment:
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:
```scala
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.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -414,6 +414,39 @@ fn field_id(field: &arrow::datatypes::Field) ->
Option<i32> {
.and_then(|v| v.parse::<i32>().ok())
}
+/// True when a field in `fields`, at any nesting depth, carries a Parquet
field id. Spark's
+/// `containsFieldIds` walks the whole file schema the same way, and
`ParquetUtils.hasFieldIds`
+/// walks the read schema. The root-only `schema_has_field_ids` in the schema
adapter stays as the
+/// gate for id matching, which only ever renames root fields.
+pub(crate) fn any_nested_field_has_id(fields: &Fields) -> bool {
Review Comment:
`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.
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -230,7 +230,13 @@ impl IcebergScanExec {
let scan_metrics = scan_result.metrics().clone();
let stream = scan_result.stream();
- let spark_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC",
false);
+ let mut spark_options = SparkParquetOptions::new(EvalMode::Legacy,
"UTC", false);
+ // Iceberg resolves columns by id itself and its reader supplies the
ids, so the
+ // missing-id check that guards plain Parquet reads does not apply
here. The same holds
+ // for a migrated table read through a name mapping: the reader still
resolves the
+ // columns itself and hands back a schema of its own, so the Spark
check has nothing to
+ // say there either.
+ spark_options.ignore_missing_field_id = true;
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
