comphead commented on code in PR #6116:
URL: https://github.com/apache/datafusion-comet/pull/6116#discussion_r4099362673
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -102,6 +102,11 @@ pub(crate) fn init_datasource_exec(
);
spark_parquet_options.use_field_id = use_field_id;
spark_parquet_options.ignore_missing_field_id = ignore_missing_field_id;
+ // Spark runs its missing-id check against the pruned read schema it hands
the reader, not
+ // the full data schema that DataFusion later passes the schema adapter,
so the answer is
+ // taken from `required_schema` here, once per scan, and handed to the
reader factory below.
+ spark_parquet_options.requested_schema_has_field_ids =
+ any_nested_field_has_id(required_schema.fields());
Review Comment:
Following up on @andygrove's alternative in the first review: now that the
file side walks the raw footer, this Arrow walk only answers the requested
side. That is exactly `ParquetUtils.hasFieldIds(scan.requiredSchema)`, which
`CometNativeScan.scala:266` already computes for `use_field_id`.
Sending `!IGNORE_MISSING_PARQUET_FIELD_ID && hasFieldIds(requiredSchema)` as
one proto bool, in place of `ignore_missing_field_id` (field 16, whose only
native reader is now this check), would:
- remove `any_nested_field_has_id`, `field_holds_id`, `list_element_field`
and their test, about 100 lines
- match Spark by construction. The walk here also counts ids on list
elements and map keys and values, which Spark's `hasFieldIds` cannot see. It
agrees today only because the serde never sets them.
If you prefer to keep it native,
`required_schema.flattened_fields().iter().any(|f| field_id(f).is_some())` does
the same walk, and `shuffle_block_writer.rs` already uses that pattern. The
`LargeList`, `FixedSizeList`, `ListView` and `LargeListView` arms can't be
reached either, because the serde only emits `List`, `Map` and `Struct`.
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -194,7 +199,11 @@ pub(crate) fn init_datasource_exec(
scan_io_source,
parquet_source.metrics(),
)
- .with_spark_variant_schema(projects_variant),
+ .with_spark_variant_schema(projects_variant)
+ .with_missing_field_id_check(
+ spark_parquet_options.requested_schema_has_field_ids,
+ spark_parquet_options.ignore_missing_field_id,
+ ),
Review Comment:
After this PR, `SparkParquetOptions.requested_schema_has_field_ids` and
`SparkParquetOptions.ignore_missing_field_id` are written and read back only
inside this function. The adapter branch that read `ignore_missing_field_id` is
gone. Both fields are still cloned into every `CometCastColumnExpr` and take
part in its `PartialEq`/`Hash`.
The factory and the reader also each store two bools that are only ever read
as `requested && !ignore`. Suggest passing a single `require_field_ids: bool`
through a one-argument builder, like `with_spark_variant_schema`, and dropping
both options fields. This applies whether or not the proto change above happens.
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -191,13 +204,40 @@ impl EagerPageIndexReaderFactory {
metadata_cache,
scan_io_metrics,
spark_variant_schema: false,
+ requested_schema_has_field_ids: false,
+ ignore_missing_field_id: false,
}
}
pub fn with_spark_variant_schema(mut self, enabled: bool) -> Self {
self.spark_variant_schema = enabled;
self
}
+
+ /// Arm Spark's missing field id check. A file whose Parquet schema
carries no field id
+ /// is refused on open when `requested_schema_has_field_ids` is set and
+ /// `ignore_missing_field_id` is not. Both default to off, so a factory
that never calls
+ /// this reads every file.
+ pub fn with_missing_field_id_check(
+ mut self,
+ requested_schema_has_field_ids: bool,
+ ignore_missing_field_id: bool,
+ ) -> Self {
+ self.requested_schema_has_field_ids = requested_schema_has_field_ids;
+ self.ignore_missing_field_id = ignore_missing_field_id;
+ self
+ }
+}
+
+/// True when `node` or any node under it carries a field id, the way Spark's
+/// `containsFieldIds` answers it over the raw Parquet schema, message root
included.
+fn parquet_schema_has_field_ids(node: &ParquetType) -> bool {
Review Comment:
Naming nit: `parquet_schema_has_field_ids` (raw Parquet schema),
`schema_has_field_ids` (root fields only, `schema_adapter.rs:83`) and
`any_nested_field_has_id` (Arrow, recursive) sound alike but check different
depths. Spark's names would make the difference visible: `contains_field_ids`
here, after `containsFieldIds`, and `any_root_field_has_id` for the root-only
gate.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -2152,6 +2153,295 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
}
+
+ // Spark's `ParquetReadSupport` checks for missing file ids before it looks
at
+ // `fieldId.read.enabled`, so the error is raised with id matching off as
well. With
+ // `ignoreMissing` set, both engines fall back to matching by name and read
real values.
+ test("read schema with field ids raises on a file without ids when id
matching is off") {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ val readSchema = new StructType().add("a", IntegerType, true,
withId(1))
+ val writeSchema = new StructType().add("a", IntegerType, true)
+ val writeData = Seq(Row(100), Row(200))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData),
writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ def readCause(): Throwable = intercept[SparkException] {
+ spark.read.schema(readSchema).parquet(dir.getCanonicalPath).collect()
+ }.getCause
+ withClue("Spark with Comet disabled") {
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ assertMissingIdsException(readCause())
+ }
+ }
+ withClue("Comet") {
+ assertMissingIdsException(readCause())
+ }
+
+ withSQLConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key -> "true") {
+
checkSparkAnswerAndOperator(spark.read.schema(readSchema).parquet(dir.getCanonicalPath))
+ }
+ }
+ }
+ }
+
+ // Spark's `containsFieldIds` walks the whole file schema, so ids that sit
only on struct
+ // children count. The root field whose id the file lacks is null filled
rather than rejected.
+ test("a file whose field ids are only on nested fields reads without a
missing-id error") {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ withTempPath { dir =>
+ val nested = StructType(Seq(StructField("a", IntegerType, nullable =
true, withId(11))))
+ val writeSchema = new StructType().add("s", nested, true)
+ val readSchema = new StructType()
+ .add("s", nested, true)
+ .add("missing", IntegerType, true, withId(7))
+ val writeData = Seq(Row(Row(1)), Row(Row(2)))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData),
writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+
checkSparkAnswerAndOperator(spark.read.schema(readSchema).parquet(dir.getCanonicalPath))
+ }
+ }
+ }
+
+ // Second half of Spark `ParquetFieldIdIOSuite.test("global read/write flag
should work
+ // correctly")`: the file carries ids but the read flag is off, so columns
resolve by name
+ // only. None of the read names exist in the file, so every value is null
and nothing raises.
+ test("field ids in the file are ignored when id matching is off") {
Review Comment:
This is the second half of Spark's own "global read/write flag should work
correctly", which already runs under Comet in the Spark SQL jobs, since no diff
ignores `ParquetFieldIdIOSuite`. The file carries ids, so the new check never
fires, and I'd expect this to pass on `main` too. Suggest dropping it.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -2152,6 +2153,295 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
}
+
+ // Spark's `ParquetReadSupport` checks for missing file ids before it looks
at
+ // `fieldId.read.enabled`, so the error is raised with id matching off as
well. With
+ // `ignoreMissing` set, both engines fall back to matching by name and read
real values.
+ test("read schema with field ids raises on a file without ids when id
matching is off") {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ val readSchema = new StructType().add("a", IntegerType, true,
withId(1))
+ val writeSchema = new StructType().add("a", IntegerType, true)
+ val writeData = Seq(Row(100), Row(200))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData),
writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ def readCause(): Throwable = intercept[SparkException] {
+ spark.read.schema(readSchema).parquet(dir.getCanonicalPath).collect()
+ }.getCause
+ withClue("Spark with Comet disabled") {
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ assertMissingIdsException(readCause())
+ }
+ }
+ withClue("Comet") {
+ assertMissingIdsException(readCause())
+ }
+
+ withSQLConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key -> "true") {
+
checkSparkAnswerAndOperator(spark.read.schema(readSchema).parquet(dir.getCanonicalPath))
+ }
+ }
+ }
+ }
+
+ // Spark's `containsFieldIds` walks the whole file schema, so ids that sit
only on struct
+ // children count. The root field whose id the file lacks is null filled
rather than rejected.
+ test("a file whose field ids are only on nested fields reads without a
missing-id error") {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ withTempPath { dir =>
+ val nested = StructType(Seq(StructField("a", IntegerType, nullable =
true, withId(11))))
+ val writeSchema = new StructType().add("s", nested, true)
+ val readSchema = new StructType()
+ .add("s", nested, true)
+ .add("missing", IntegerType, true, withId(7))
+ val writeData = Seq(Row(Row(1)), Row(Row(2)))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData),
writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+
checkSparkAnswerAndOperator(spark.read.schema(readSchema).parquet(dir.getCanonicalPath))
+ }
+ }
+ }
+
+ // Second half of Spark `ParquetFieldIdIOSuite.test("global read/write flag
should work
+ // correctly")`: the file carries ids but the read flag is off, so columns
resolve by name
+ // only. None of the read names exist in the file, so every value is null
and nothing raises.
+ test("field ids in the file are ignored when id matching is off") {
+ withSQLConf(
+ SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true",
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ val readSchema = new StructType()
+ .add("some", IntegerType, true, withId(1))
+ .add("other", StringType, true, withId(2))
+ .add("name", StringType, true, withId(3))
+ val writeSchema = new StructType()
+ .add("a", IntegerType, true, withId(1))
+ .add("rand1", StringType, true, withId(2))
+ .add("rand2", StringType, true, withId(3))
+ val writeData = Seq(Row(100, "text", "txt"), Row(200, "more", "mr"))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData),
writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ val df = spark.read.schema(readSchema).parquet(dir.getCanonicalPath)
+ checkSparkAnswerAndOperator(df)
+ checkAnswer(df, Row(null, null, null) :: Row(null, null, null) :: Nil)
+ }
+ }
+ }
+
+ // Spark checks for missing file ids against the pruned read schema, so an
id on a column
+ // the query never projects does not reject a file without ids, whatever the
read flag says.
+ // The same read with both columns projected still raises.
+ test("field ids on an unprojected column do not reject a file without ids") {
+ Seq("false", "true").foreach { readEnabled =>
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> readEnabled) {
+ withTempPath { dir =>
+ val writeSchema = new StructType().add("a", IntegerType).add("b",
IntegerType)
+ val readSchema = new StructType()
+ .add("a", IntegerType, true, withId(1))
+ .add("b", IntegerType, true)
+ val writeData = Seq(Row(1, 2), Row(3, 4))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData),
writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ withClue(s"read flag $readEnabled") {
+ val pruned =
spark.read.schema(readSchema).parquet(dir.getCanonicalPath).select("b")
+ checkSparkAnswerAndOperator(pruned)
+ checkAnswer(pruned, Row(2) :: Row(4) :: Nil)
+
+ val cause = intercept[SparkException] {
+
spark.read.schema(readSchema).parquet(dir.getCanonicalPath).collect()
+ }.getCause
+ assert(
+ cause.isInstanceOf[RuntimeException] &&
+ cause.getMessage.contains("Parquet file schema doesn't contain
any field Ids"),
+ cause)
+ }
+ }
+ }
+ }
+ }
+
+ // Spark writes timestamps as INT96 by default. The reader coerces INT96 to
microseconds and
+ // rebuilds every container field without its metadata on the way, so the id
on `s` is gone
+ // from the Arrow schema the adapter sees. The missing-id check reads the
Parquet schema, where
+ // the id still is, so the file reads and `s` resolves by name. Resolving
`s` by id is a
+ // matter for the remap, which still works from the coerced schema, so id
matching stays off.
+ test("field ids on a struct holding a timestamp survive the missing-id
check") {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ val nested = new StructType().add("a", IntegerType).add("ts",
TimestampType)
+ val schema = new StructType().add("s", nested, true, withId(1))
+ val ts = Timestamp.valueOf("2020-01-01 00:00:00")
+ val writeData = Seq(Row(Row(1, ts)), Row(Row(2, ts)))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData), schema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ val df = spark.read.schema(schema).parquet(dir.getCanonicalPath)
+ checkSparkAnswerAndOperator(df)
+ checkAnswer(df, Row(Row(1, ts)) :: Row(Row(2, ts)) :: Nil)
+ }
+ }
+ }
+
+ // Spark checks each file on its own. A directory holding one file with ids
and one without
+ // raises on the second, and with `ignoreMissing` the file without ids reads
as nulls because
+ // no root field of it carries the requested id.
+ test("a file without ids next to a file with ids is checked on its own") {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ withTempPath { dir =>
+ val idSchema = new StructType().add("x", IntegerType, true, withId(1))
+ val plainSchema = new StructType().add("a", IntegerType, true)
+ val readSchema = new StructType().add("a", IntegerType, true,
withId(1))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(Seq(Row(100),
Row(200))), idSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(Seq(Row(1),
Row(2))), plainSchema)
+ .write
+ .mode("append")
+ .parquet(dir.getCanonicalPath)
+
+ val cause = intercept[SparkException] {
+ spark.read.schema(readSchema).parquet(dir.getCanonicalPath).collect()
+ }.getCause
+ assertMissingIdsException(cause)
+
+ withSQLConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key -> "true") {
+ val df = spark.read.schema(readSchema).parquet(dir.getCanonicalPath)
+ checkSparkAnswerAndOperator(df)
+ checkAnswer(df, Row(100) :: Row(200) :: Row(null) :: Row(null) ::
Nil)
+ }
+ }
+ }
+ }
+
+ // Nested schema pruning hands the reader only the struct children a query
touches, so an id
+ // on `s.a` rejects a file without ids when `s.a` is read and plays no part
when only `s.b` is.
+ test("field ids on a pruned struct child are checked only when that child is
read") {
+ Seq("false", "true").foreach { readEnabled =>
+ withSQLConf(
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> readEnabled,
+ SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "true") {
+ withTempPath { dir =>
+ val writeSchema = new StructType()
+ .add("s", new StructType().add("a", IntegerType).add("b",
IntegerType), true)
+ val readSchema = new StructType().add(
+ "s",
+ StructType(
+ Seq(
+ StructField("a", IntegerType, nullable = true, withId(11)),
+ StructField("b", IntegerType, nullable = true))),
+ true)
+ spark
+ .createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(Row(1, 2)), Row(Row(3,
4)))),
+ writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ withClue(s"read flag $readEnabled") {
+ val cause = intercept[SparkException] {
+
spark.read.schema(readSchema).parquet(dir.getCanonicalPath).select("s.a").collect()
+ }.getCause
+ assertMissingIdsException(cause)
+
+ val onlyB =
spark.read.schema(readSchema).parquet(dir.getCanonicalPath).select("s.b")
+ checkSparkAnswerAndOperator(onlyB)
+ checkAnswer(onlyB, Row(2) :: Row(4) :: Nil)
+ }
+ }
+ }
+ }
+ }
+
+ // Id zero is an id like any other, so a read schema carrying it expects ids
in the file.
+ test("field id zero on a root field expects ids in the file") {
+ Seq("false", "true").foreach { readEnabled =>
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> readEnabled) {
+ withTempPath { dir =>
+ val readSchema = new StructType().add("a", IntegerType, true,
withId(0))
+ val writeSchema = new StructType().add("a", IntegerType, true)
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(Seq(Row(1),
Row(2))), writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ withClue(s"read flag $readEnabled") {
+ def readCause(): Throwable = intercept[SparkException] {
+
spark.read.schema(readSchema).parquet(dir.getCanonicalPath).collect()
+ }.getCause
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ assertMissingIdsException(readCause())
+ }
+ assertMissingIdsException(readCause())
+ }
+ }
+ }
+ }
+ }
+
+ // A count reads no columns, so the pruned read schema carries no ids and a
file without ids
+ // is counted rather than rejected.
+ test("count over an id-bearing schema reads a file without ids") {
+ Seq("false", "true").foreach { readEnabled =>
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> readEnabled) {
+ withTempPath { dir =>
+ val readSchema = new StructType().add("a", IntegerType, true,
withId(1))
+ val writeSchema = new StructType().add("a", IntegerType, true)
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(Seq(Row(1),
Row(2))), writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ withClue(s"read flag $readEnabled") {
+ val counted =
+
spark.read.schema(readSchema).parquet(dir.getCanonicalPath).selectExpr("count(*)")
+ checkSparkAnswerAndOperator(counted)
+ checkAnswer(counted, Row(2L) :: Nil)
+ }
+ }
+ }
+ }
+ }
+
+ // Spark raises a plain `RuntimeException` for a read schema with ids over a
file without any.
+ // How many `SparkException` layers sit above it varies with the Spark
version. Spark 4 wraps
+ // a reader failure in a `FAILED_READ_FILE` `SparkException`, and on 4.0 and
4.1 that wrapper
+ // is the exception `collect()` raises, so the `RuntimeException` is its
direct cause. The
+ // Comet shim follows Spark, so walk the cause chain instead of counting the
layers above it.
+ private def assertMissingIdsException(cause: Throwable): Unit = {
Review Comment:
This reimplements `CometTestBase.causeChain`, and the `readCause()` plus
`COMET_ENABLED=false` pairs at L2172 and L2392 reimplement
`checkSparkAnswerMaybeThrows`.
Spark's own assertion,
`intercept[SparkException]{...}.getCause.isInstanceOf[RuntimeException]`, is
identical on 3.4 through 4.1. The existing port at L2119 and the new test at
L2266 already use it, so the chain walk shouldn't be needed. I'd expect the
direct form to hold for Comet on every profile, but CI here only ran 4.1.
Either way, one assertion style for this error would be good.
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -45,6 +45,13 @@
//!
//! Filed upstream as apache/datafusion#23978. Revert this once the opener
merges its deferred
//! page-index load back into `FileMetadataCache` instead of bypassing it.
+//!
+//! The reader also carries Spark's missing field id check, because the footer
is first at hand
+//! here. Spark's `ParquetReadSupport` refuses to open a file whose raw schema
carries no field
+//! id when the requested schema carries one, unless `ignoreMissing` is set,
and it walks the
+//! raw `MessageType` to decide. The Arrow schema the schema adapter sees
later cannot stand in
+//! for that walk: the INT96 coercion rebuilds container fields without their
metadata, and an
+//! id on a `list` or `key_value` group, or on the message root, never reaches
an Arrow field.
Review Comment:
Two things on the module doc:
- The revert note above now undersells what a revert would remove. This
factory also hosts the missing-id check and the Variant footer rewrite, so any
revert has to keep a `get_metadata` hook. Worth saying so here.
- The INT96 part of the rationale is a DataFusion bug, fixed upstream by
apache/datafusion#24790 (merged 2026-09-07, not in 55.1.0), so it goes stale at
the next DataFusion bump. The lasting reason is that ids on repeated
`list`/`key_value` groups and on the message root never reach Arrow. I'd lead
with that and cite apache/datafusion#24790 once. I'd also expect that bump to
fix #6131, but I haven't checked.
The same explanation is repeated in about a dozen places:
`parquet_support.rs:111-114`, `:428-435` and `:440-450`,
`parquet_exec.rs:105-107`, `schema_adapter.rs:79-82`, `:211-212` and
`:891-895`, `errors.rs:651-655`, and `ParquetReadSuite.scala:2279-2283`. This
doc plus a short comment at the check would be enough.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -890,6 +887,12 @@ 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.
+ //
+ // The check that a file carries field ids at all, which Spark's
`ParquetReadSupport`
+ // runs before anything else, lives in
`EagerPageIndexReader::get_metadata`, where the
+ // raw Parquet schema is at hand. By the time the schemas reach this
point the INT96
+ // coercion may have dropped the ids from container fields, so
`physical_file_schema`
+ // cannot answer that question.
Review Comment:
This comment, and the ones at L79-82 and L211-212, describe code that now
lives in another file. I'd drop them, or keep a single line on
`schema_has_field_ids` saying it is root-only on purpose.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -2959,6 +2968,595 @@ mod test {
Ok(())
}
+ /// The message every rejected read carries, from
`SparkError::ParquetMissingFieldIds`.
+ const MISSING_IDS: &str = "Parquet file schema doesn't contain any field
Ids";
Review Comment:
Most of these scan tests have a Scala twin in this PR, and the Scala side
also compares against Spark:
| Rust test | `ParquetReadSuite` |
|---|---|
| `missing_file_field_ids_rejected_when_id_matching_disabled` | L2160 |
| `nested_logical_field_ids_rejected_when_file_has_none` | L2340, reject
half |
| `nested_file_field_ids_satisfy_the_missing_id_check` | L2193 |
| `missing_file_field_ids_allowed_when_ignore_missing_is_set` | L2119,
`ignoreMissing` half |
| `file_ids_ignored_when_id_matching_disabled` | L2216 |
| `field_ids_on_an_unprojected_column_do_not_reject_a_file_without_ids` |
L2246 |
| `struct_id_hidden_by_int96_coercion_satisfies_the_missing_id_check` |
L2284 |
Also:
- Every read without ids already covers `no_logical_field_ids_never_rejects`.
- `struct_with_int96_and_no_ids_is_rejected` repeats the first test with an
INT96 file, which the raw-schema walk ignores.
- The `use_field_id` and `skip_arrow_metadata` loops vary inputs the check
never reads.
Suggest keeping `parquet_schema_has_field_ids_sees_ids_on_any_node` (plus a
`key_value` case) and the `errors.rs` unit test. The repeated-group cases could
move to Scala: `createParquetWriter(MessageTypeParser.parseMessageType("...
repeated group list = 5 { ... }"))` (`CometTestBase.scala:790`) can write them,
and then they are compared against Spark too.
Any scan test that stays would fit better in
`parquet_exec/field_id_tests.rs`, next to `variant_tests.rs`. It could call the
existing `init_test_scan` (`parquet_exec.rs:434`) instead of adding
`PlannerScan` and `scan_file_via_planner`. `write_parquet` also duplicates the
writer inside `scan_parquet`.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -2152,6 +2153,295 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
}
+
+ // Spark's `ParquetReadSupport` checks for missing file ids before it looks
at
+ // `fieldId.read.enabled`, so the error is raised with id matching off as
well. With
+ // `ignoreMissing` set, both engines fall back to matching by name and read
real values.
+ test("read schema with field ids raises on a file without ids when id
matching is off") {
Review Comment:
This is the existing port at L2119 with the read flag flipped, and L2379
repeats it with `withId(0)`. Parameterizing L2119 over `fieldId.read.enabled`,
and adding an id-0 schema to its `Seq(readSchema, ...)` loop, would cover both
and let these two tests go.
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -498,6 +544,20 @@ impl AsyncFileReader for EagerPageIndexReader {
}
let metadata = metadata?;
+ // Spark's `ParquetReadSupport` refuses to open a file that
carries no field ids when
+ // the requested schema carries some, unless `ignoreMissing` is
set, and it walks the
+ // raw `MessageType` to decide. The same walk runs here over the
footer's schema. The
+ // error keeps its Spark type through `ParquetError::External`,
which the JNI layer
+ // unwraps, so the JVM sees the same exception Spark raises.
+ if require_file_field_ids
+ && !parquet_schema_has_field_ids(
+ metadata.file_metadata().schema_descr().root_schema(),
+ )
+ {
+ return Err(ParquetError::External(Box::new(
+ SparkError::ParquetMissingFieldIds,
+ )));
+ }
Review Comment:
Optional: the file is known here (`object_meta.location`), but
`ParquetMissingFieldIds` carries no path, so the 4.x shim calls
`cannotReadFilesError(cause, "")` and the message names no file. Carrying
`filePath`, as `ParquetSchemaConvert` does, would give Spark's message shape.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -2152,6 +2153,295 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
}
+
+ // Spark's `ParquetReadSupport` checks for missing file ids before it looks
at
+ // `fieldId.read.enabled`, so the error is raised with id matching off as
well. With
+ // `ignoreMissing` set, both engines fall back to matching by name and read
real values.
+ test("read schema with field ids raises on a file without ids when id
matching is off") {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ val readSchema = new StructType().add("a", IntegerType, true,
withId(1))
+ val writeSchema = new StructType().add("a", IntegerType, true)
+ val writeData = Seq(Row(100), Row(200))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData),
writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ def readCause(): Throwable = intercept[SparkException] {
+ spark.read.schema(readSchema).parquet(dir.getCanonicalPath).collect()
+ }.getCause
+ withClue("Spark with Comet disabled") {
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ assertMissingIdsException(readCause())
+ }
+ }
+ withClue("Comet") {
+ assertMissingIdsException(readCause())
+ }
+
+ withSQLConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key -> "true") {
+
checkSparkAnswerAndOperator(spark.read.schema(readSchema).parquet(dir.getCanonicalPath))
+ }
+ }
+ }
+ }
+
+ // Spark's `containsFieldIds` walks the whole file schema, so ids that sit
only on struct
+ // children count. The root field whose id the file lacks is null filled
rather than rejected.
+ test("a file whose field ids are only on nested fields reads without a
missing-id error") {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ withTempPath { dir =>
+ val nested = StructType(Seq(StructField("a", IntegerType, nullable =
true, withId(11))))
+ val writeSchema = new StructType().add("s", nested, true)
+ val readSchema = new StructType()
+ .add("s", nested, true)
+ .add("missing", IntegerType, true, withId(7))
+ val writeData = Seq(Row(Row(1)), Row(Row(2)))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData),
writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+
checkSparkAnswerAndOperator(spark.read.schema(readSchema).parquet(dir.getCanonicalPath))
+ }
+ }
+ }
+
+ // Second half of Spark `ParquetFieldIdIOSuite.test("global read/write flag
should work
+ // correctly")`: the file carries ids but the read flag is off, so columns
resolve by name
+ // only. None of the read names exist in the file, so every value is null
and nothing raises.
+ test("field ids in the file are ignored when id matching is off") {
+ withSQLConf(
+ SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true",
+ SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "false") {
+ withTempPath { dir =>
+ val readSchema = new StructType()
+ .add("some", IntegerType, true, withId(1))
+ .add("other", StringType, true, withId(2))
+ .add("name", StringType, true, withId(3))
+ val writeSchema = new StructType()
+ .add("a", IntegerType, true, withId(1))
+ .add("rand1", StringType, true, withId(2))
+ .add("rand2", StringType, true, withId(3))
+ val writeData = Seq(Row(100, "text", "txt"), Row(200, "more", "mr"))
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(writeData),
writeSchema)
+ .write
+ .mode("overwrite")
+ .parquet(dir.getCanonicalPath)
+
+ val df = spark.read.schema(readSchema).parquet(dir.getCanonicalPath)
+ checkSparkAnswerAndOperator(df)
+ checkAnswer(df, Row(null, null, null) :: Row(null, null, null) :: Nil)
+ }
+ }
+ }
+
+ // Spark checks for missing file ids against the pruned read schema, so an
id on a column
+ // the query never projects does not reject a file without ids, whatever the
read flag says.
+ // The same read with both columns projected still raises.
+ test("field ids on an unprojected column do not reject a file without ids") {
Review Comment:
L2246, L2340 and L2407 all test one rule, that the check uses the pruned
schema, and write six files between them. One file with `a, b, s struct<a, b>`
and ids on `a` and `s.a` covers all three: `select b`, `select s.b` and
`count(*)` read, while `select a` and `select s.a` raise.
The `readEnabled` loops can go as well. For the pruned reads `use_field_id`
is false either way, so the native plan is the same. The raising reads never
consult the flag, and flag independence is covered once at L2160 (or at L2119
if that gets parameterized).
Separately, `checkAnswer` right after `checkSparkAnswerAndOperator` runs the
query twice more, here and in most of the new tests. Fixed expected rows are
only worth it where they are the point, like the null-fill at L2332.
--
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]
