mbutrovich commented on code in PR #6725:
URL: https://github.com/apache/datafusion-comet/pull/6725#discussion_r4261120897
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -774,15 +779,17 @@ case class CometScanRule(session: SparkSession)
// fields to the table's by field id (apache/iceberg-rust#2617). A
file written before a
// nested field was added (to a struct, or to a struct inside a list
or map) fails the
// native scan with "Incorrect number of arrays for StructArray
fields", and a nested field
- // renamed since the file was written can read back as NULL. The
native read asks for a
- // projected column's full nested type even when Spark prunes it,
taken from the current
- // table schema, or from the scan schema when VERSION AS OF reads a
dropped column (see
- // CometIcebergNativeScan). A FileScanTask does not record which
schema wrote its file, so
- // fall back when any schema in the table's history lacks a nested
field of a projected
- // column or names it differently.
+ // renamed since the file was written can read back as NULL. The
native read usually asks
+ // only for the nested fields Spark's schema pruning keeps, but a scan
can still read with
+ // its full task schema (see IcebergReflection.nativeTaskSchemas),
which for VERSION AS OF
+ // can be an older schema than the table's. So this check walks the
table, scan, and full
+ // task schemas, and falls back even when the changed field is pruned
away. A FileScanTask
+ // does not record which schema wrote its file, so fall back when any
schema in the table's
+ // history lacks a nested field of a projected column or names it
differently.
val nestedFieldsSupported =
try {
- val schemas = Seq(metadata.tableSchema, metadata.scanSchema)
+ val schemas =
+ Seq(metadata.tableSchema, metadata.scanSchema) ++
metadata.fullTaskSchema
Review Comment:
Should this check look at the schema the scan will actually read, rather
than always adding `fullTaskSchema`? With pruning on, a scan whose
`prunedTaskSchema` is defined never reads the full schema, but its full schema
can still trigger the fallback.
Take the unpartitioned table from my reply on the `nativeTaskSchemas`
thread. With pruning on, `SELECT id, c, s.a FROM t VERSION AS OF S` reads the
pruned schema, which holds only `s.a`. At 6213d0576e it ran natively and
matched Spark, and it does on `main` too. At this commit it falls back with
`s.b (added)` because the snapshot's full schema still has `s.b`. The
suggestion below keeps the native read for that query. It still falls back for
the partitioned case from the thread, where `prunedTaskSchema` is `None`, and
for the unpartitioned case with pruning off.
```suggestion
val readSchema = metadata.prunedTaskSchema
.filter(_ =>
CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.get())
.orElse(metadata.fullTaskSchema)
val schemas = Seq(metadata.tableSchema, metadata.scanSchema) ++
readSchema
```
[`CometIcebergNativeScan`](https://github.com/apache/datafusion-comet/blob/8566834663235f9cb47d2f5af3578ed46f57028b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala#L1018-L1024)
makes the same choice during serialization. What do you think about a method
on `CometIcebergNativeScanMetadata` that both call, so the check and the read
can't drift apart? The comment above would then describe the schema the scan
reads instead of "the table, scan, and full task schemas". Could the new
`evo_struct_historical` case also cover the unpartitioned table, to check that
it stays native?
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -1039,12 +1046,21 @@ case class CometScanRule(session: SparkSession)
}
}
+ // Every task needs each partition source and equality-delete key
under one name (see
+ // IcebergReflection.nativeTaskSchemas).
+ val taskSchemaNamesSupported = metadata.fullTaskSchema.isDefined
+ if (!taskSchemaNamesSupported) {
+ fallbackReasons += "Comet's native Iceberg reader cannot name a
partition source or " +
+ "equality-delete key the same way in every task: no schema in the
table's history " +
+ "names it apart from the columns the scan reads"
Review Comment:
This is a new fallback that users can hit. Should it go in the [Current
limitations](https://github.com/apache/datafusion-comet/blob/474b5ece841a7fc6dff37ae04bd6787e654d7be7/docs/source/user-guide/latest/iceberg.md?plain=1#L177-L199)
list in the Iceberg user guide, next to the nested schema evolution entry?
Something like: "Scans where no schema in the table's history gives a partition
source or equality-delete key a name that the scan's other columns don't use."
##########
spark/src/test/resources/sql-tests/iceberg/nested_schema_pruning.sql:
##########
@@ -0,0 +1,135 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- The native Iceberg scan reads only the nested fields that Spark's nested
schema pruning keeps.
+-- These queries check the pruned reads against Spark. CometIcebergNativeSuite
checks that they
+-- scan fewer bytes, and covers deletes, time travel, and files without field
ids.
+
+-- No Iceberg spark-runtime is published for Spark 4.2 yet, and the 4.0
runtime the build reuses
+-- is binary-incompatible with it. See
https://github.com/apache/datafusion-comet/issues/4969.
+-- MaxSparkVersion: 4.1
+
+-- Config: spark.sql.catalog.test_cat=org.apache.iceberg.spark.SparkCatalog
+-- Config: spark.sql.catalog.test_cat.type=hadoop
+-- Config: spark.sql.catalog.test_cat.warehouse=/tmp/comet-iceberg-sql-test
+-- Config: spark.comet.scan.icebergNative.enabled=true
+
+statement
+DROP TABLE IF EXISTS test_cat.db.nested_pruning
+
+statement
+CREATE TABLE test_cat.db.nested_pruning (
+ id INT,
+ s STRUCT<a: INT, pad: STRING, inner: STRUCT<b: INT, pad: STRING>>,
+ items ARRAY<STRUCT<x: INT, pad: STRING>>,
+ m MAP<STRING, STRUCT<v: INT, pad: STRING>>
+) USING iceberg
+
+-- NULL structs, lists, maps, list elements, map values, and fields, so that
the pruned read has
+-- to rebuild the validity of every level from the leaves it keeps.
+statement
+INSERT INTO test_cat.db.nested_pruning
+SELECT
+ CAST(id AS INT),
+ IF(id % 7 = 0, NULL, named_struct(
+ 'a', IF(id % 5 = 0, NULL, CAST(id AS INT)),
+ 'pad', repeat('p', 20),
+ 'inner', IF(id % 3 = 0, NULL, named_struct('b', CAST(id * 2 AS INT),
'pad', repeat('q', 20))))),
+ IF(id % 7 = 0, NULL, array(
+ named_struct('x', CAST(id AS INT), 'pad', repeat('r', 20)),
+ IF(id % 5 = 0, NULL, named_struct('x', CAST(-id AS INT), 'pad',
repeat('s', 20))))),
+ IF(id % 7 = 0, NULL, map('k', IF(id % 3 = 0, NULL,
+ named_struct('v', CAST(id AS INT), 'pad', repeat('t', 20)))))
Review Comment:
The comment above says this data puts NULLs at every level so the pruned
read has to rebuild validity. Should it also have empty lists and maps? An
empty list is valid with no child rows, which is a different case from a NULL
list or a NULL element. For example:
```suggestion
CASE WHEN id % 7 = 0 THEN NULL WHEN id % 11 = 0 THEN array() ELSE array(
named_struct('x', CAST(id AS INT), 'pad', repeat('r', 20)),
IF(id % 5 = 0, NULL, named_struct('x', CAST(-id AS INT), 'pad',
repeat('s', 20)))) END,
CASE WHEN id % 7 = 0 THEN NULL WHEN id % 11 = 0 THEN map() ELSE map('k',
IF(id % 3 = 0, NULL,
named_struct('v', CAST(id AS INT), 'pad', repeat('t', 20)))) END
```
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -2284,6 +2324,373 @@ class CometIcebergNativeSuite
}
}
+ // Spark's nested schema pruning reaches the native scan through the scan
schema, so only the
+ // nested fields a query uses are read and the wide `pad` strings beside
them are skipped.
+ // sql-tests/iceberg/nested_schema_pruning.sql checks the results. One data
file holds every
+ // row, so each `pad` column chunk is larger than iceberg-rust's 1 MiB read
coalescing, which
+ // would otherwise merge the reads of the kept chunks across the skipped
ones.
+ test("nested schema pruning reads only the nested fields the query uses") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "test_cat.db.nested_pruning"
+ spark.sql(s"""
+ CREATE TABLE $table (
+ id INT,
+ s STRUCT<a: INT, pad: STRING>,
+ items ARRAY<STRUCT<x: INT, pad: STRING>>,
+ m MAP<STRING, STRUCT<v: INT, pad: STRING>>
+ ) USING iceberg
+ """)
+ spark.sql(s"""
+ INSERT INTO $table
+ SELECT
+ CAST(id AS INT),
+ named_struct('a', CAST(id AS INT), 'pad', pad),
+ array(named_struct('x', CAST(id AS INT), 'pad', pad)),
+ map('k', named_struct('v', CAST(id AS INT), 'pad', pad))
+ FROM (
+ SELECT id, concat_ws('', transform(array('a', 'b', 'c', 'd'),
+ salt -> sha2(concat(CAST(id AS STRING), salt), 256))) AS pad
+ FROM range(0, 20000, 1, 1))
+ """)
+
+ def bytesScanned(pruneNestedFields: Boolean): Long = {
+ var bytes = 0L
+ withSQLConf(
+ CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key ->
+ pruneNestedFields.toString) {
+ val df = spark.sql(s"SELECT sum(s.a), count(items.x),
count(m['k'].v) FROM $table")
+ df.collect()
+ val scans =
collectIcebergNativeScans(df.queryExecution.executedPlan)
+ assert(scans.length == 1, s"expected one native scan, got
${scans.length}")
+ bytes = scans.head.metrics("bytes_scanned").value
+ }
+ bytes
+ }
+ val prunedBytes = bytesScanned(pruneNestedFields = true)
+ val fullBytes = bytesScanned(pruneNestedFields = false)
+ assert(
+ prunedBytes * 4 < fullBytes,
+ s"pruned read should skip the pad fields: pruned=$prunedBytes,
full=$fullBytes")
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+
+ // A pruned task schema still needs the columns iceberg-rust uses beyond the
projection: the
+ // partition source and the equality-delete key when the query projects
neither. Tasks with
+ // deletes read with the pruned schema too, and every task of a scan reads
with the same one.
+ // VERSION AS OF reads of columns dropped, renamed, or swapped after the
snapshot use the
+ // snapshot's names.
+ test("nested schema pruning with deletes, partitions, and time travel") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ // Also checks that every task, tasks with deletes included, reads
with one schema that
+ // leaves out the pruned `pad` fields.
+ def checkPrunedNativeScan(query: String): Unit = {
Review Comment:
Could `checkPrunedNativeScan` also check the answer with
`spark.comet.scan.icebergNative.nestedSchemaPruning.enabled` set to `false`?
Could the queries in the equality-delete naming test do the same? With the
config off, every task now reads `fullTaskSchema`, which uses the snapshot's
names, so the renamed and swapped columns read in this test are the full-schema
half of #6822. Right now nothing checks the config-off results for them. If
these cases cover everything left in #6822 on `main`, should the PR description
say it fixes #6822, leaving the `branch-1.1` part to a backport?
--
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]
