comphead commented on code in PR #5732:
URL: https://github.com/apache/datafusion-comet/pull/5732#discussion_r4020380184
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -628,6 +628,13 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
None
} else {
operation match {
+ // iceberg-rust has accessors only for primitive fields, not
containers. Keep
+ // the post-scan filter without sending residuals that would
warn on every task.
+ case IS_NULL | IS_NOT_NULL | NOT_NULL
+ if attribute.dataType.isInstanceOf[ArrayType] ||
+ attribute.dataType.isInstanceOf[MapType] ||
+ attribute.dataType.isInstanceOf[StructType] =>
+ None
Review Comment:
**1. (major) This is a second mechanism for "drop residuals on this
column".**
Two lines above, `icebergExprToProto` already has a per-column drop gate:
`pageIndexUnsupportedColumns`, computed once per scan from the table schema.
This new check adds a parallel, per-node, per-operation gate and re-implements
`DataTypeSupport.isComplexType` inline as three `isInstanceOf`.
Preferred: add `struct<` / `list<` / `map<` to
`IcebergReflection.pageIndexUnsupportedColumns` (rename it to something like
`residualUnsupportedColumns`) and delete this `case` entirely. One gate,
evaluated once per scan instead of once per predicate node, and it covers
*every* operation on a container rather than only the three null ops. It also
uses the same Iceberg-type-string convention this file already relies on.
Minimum, if you would rather keep the Spark-side check: `case IS_NULL |
IS_NOT_NULL | NOT_NULL if isComplexType(attribute.dataType) => None`.
Two questions while you are here:
- The whole correctness argument is that Spark retains the filter above the
scan. That holds because Iceberg only drops a post-scan filter when the
predicate exactly selects partitions, and a container can never be a partition
column - but nothing in the code states that invariant, and it is the
load-bearing assumption for this PR. Worth one sentence in this comment.
- Only the three null ops are gated. Whole-struct equality and
`array_contains` are not pushed by Iceberg Java today, so no container residual
reaches the match for them. Is that a guarantee or an observation? Gating by
column removes the question.
##########
native/spark-expr/src/array_funcs/get_array_struct_fields.rs:
##########
@@ -86,23 +93,23 @@ impl PhysicalExpr for GetArrayStructFields {
}
fn nullable(&self, input_schema: &Schema) -> DataFusionResult<bool> {
- Ok(self.list_field(input_schema)?.is_nullable()
- || self.child_field(input_schema)?.is_nullable())
+ self.child.nullable(input_schema)
}
fn evaluate(&self, batch: &RecordBatch) -> DataFusionResult<ColumnarValue>
{
let child_value =
self.child.evaluate(batch)?.into_array(batch.num_rows())?;
+ let field = self.child_field(batch.schema().as_ref())?;
Review Comment:
**2. `child_field` hoisted above the dispatch.**
Two small consequences:
- For a non-list child this now fails with `"Unexpected data type in
GetArrayStructFields"` instead of the intended `"Unexpected child type for
ListExtract: {data_type:?}"` below, because `child_field` -> `list_field` hits
the same non-list value first. Error-message regression on an internal error,
but easy to avoid.
- `child_field` clones a `Field` (name `String` + metadata map) on every
batch, even when the nullability it computes already matches `fields[ordinal]`.
Suggest computing it inside the two list arms, and `Arc::clone` when
`list_field.is_nullable() || field.is_nullable() == field.is_nullable()`.
##########
native/spark-expr/src/array_funcs/get_array_struct_fields.rs:
##########
@@ -171,3 +178,99 @@ impl Display for GetArrayStructFields {
)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::Int32Array;
+ use arrow::buffer::{NullBuffer, OffsetBuffer};
+ use arrow::datatypes::Field;
+ use datafusion::physical_expr::expressions::Column;
+
+ fn check_nullability<O: OffsetSizeTrait>() {
Review Comment:
**3a. Some of this belongs in the existing SQL-file harness.**
`spark/src/test/resources/sql-tests/expressions/array/get_array_struct_fields.sql`
already exists and today only covers a NULL *row*, not a NULL *element*. The
null-struct-element case (`array(NULL, named_struct('a', 2))`) is two lines
appended there and gives end-to-end Spark-vs-Comet coverage of the
`child_with_parent_nulls` union.
The *required*-field case genuinely needs the Iceberg schema API (Spark DDL
writes every parquet field optional), so that one has to stay in Scala.
The nullability-metadata assertions here cannot be expressed in SQL, so
keeping a Rust test is right. Minor:
`assert_eq!(expr.nullable(&schema).unwrap(), list_nullable)` is invariant
across the two inner loops, so it re-runs 4x for no added coverage.
##########
spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala:
##########
@@ -237,6 +238,114 @@ class CometFuzzIcebergSuite extends CometFuzzIcebergBase {
}
}
+ test("filter pushdown - IS NULL/IS NOT NULL on nested fuzz columns stays
native") {
+ val df = spark.table(icebergTableName)
+ val complexColumns = df.schema.fields.filter(f =>
isComplexType(f.dataType)).map(_.name)
+ assert(complexColumns.nonEmpty, "expected complex columns in the fuzz
schema")
+
+ for (name <- complexColumns; predicate <- Seq(col(name).isNull,
col(name).isNotNull)) {
+ withClue(predicate.toString) {
+ val (_, cometPlan) = checkSparkAnswer(df.where(predicate))
+ assert(collectIcebergNativeScans(cometPlan).length == 1, s"$cometPlan")
+ }
+ }
+ }
+
+ test("filter pushdown - IS NULL/IS NOT NULL on list, map and struct columns
stays native") {
+ val tableName = "hadoop_catalog.db.null_check_test"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $tableName (
+ id INT, l ARRAY<STRUCT<a: INT>>, m MAP<STRING, STRUCT<a: INT>>, s
STRUCT<a: INT>
+ ) USING iceberg
+ """)
+ // Container nullness is distinct from emptiness, null elements and null
struct fields.
+ spark.sql(s"""
+ INSERT INTO $tableName VALUES
+ (1, array(named_struct('a', 1)), map('k', named_struct('a', 1)),
named_struct('a', 1)),
+ (2, NULL, NULL, NULL),
+ (3, array(), map(), named_struct('a', NULL)),
+ (4, array(NULL), map('k', NULL), named_struct('a', NULL)),
+ (5, array(named_struct('a', NULL)), map('k', named_struct('a',
NULL)), named_struct('a', NULL))
+ """)
+ for (column <- Seq("l", "m", "s"); predicate <- Seq("IS NULL", "IS NOT
NULL")) {
+ val query = s"SELECT id FROM $tableName WHERE $column $predicate"
+ withClue(query) {
+ val (_, cometPlan) = checkSparkAnswer(query)
+ val expected = if (predicate == "IS NULL") Seq(Row(2)) else Seq(1,
3, 4, 5).map(Row(_))
+ checkAnswer(spark.sql(query), expected)
+ val scans = collectIcebergNativeScans(cometPlan)
+ assert(scans.length == 1, s"$cometPlan")
+ // Planning commonData leaks manifest streams on Iceberg versions
before 1.8.0.
+ if (!isIcebergVersionLessThan("1.8.0")) {
+ val common =
OperatorOuterClass.IcebergScanCommon.parseFrom(scans.head.commonData)
+ assert(
+ common.getResidualPoolCount == 0,
+ s"unexpected complex-column residual: $query")
+ }
+ }
+ }
+
+ // Spark infers IS NOT NULL below ordinary generators, but not outer
generators.
+ // Compare complete rows to preserve null elements and distinguish null
struct parents
+ // from non-null structs with null fields. Native scanning does not
imply residual pushdown.
Review Comment:
**5. Comment does not match the code.** It says "Compare complete rows ...
distinguish null struct parents from non-null structs with null fields", but
the loop only calls `checkSparkAnswer` and excludes `s`. Either narrow the
comment to what the loop does (ordinary vs outer generator, and that native
scanning does not imply residual pushdown) or add the struct case.
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -2214,8 +2214,31 @@ class CometIcebergNativeSuite
}
}
+ test("complex type null residuals are not serialized") {
+ import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType,
StructType}
+ import org.apache.comet.serde.operator.CometIcebergNativeScan
+ import org.apache.spark.sql.catalyst.expressions.AttributeReference
Review Comment:
**7. Missing `assume(icebergAvailable, "Iceberg not available in
classpath")`.**
Every other test in this suite opens with it; this one calls
`Expressions.isNull` directly, so if the Iceberg runtime is absent it errors
instead of cancelling.
Also: hoist the three imports to the file header (the current order also
interleaves `org.apache.comet` between two `org.apache.spark` imports, which
would not survive scalafmt at the top level), and wrap the loop assert in
`withClue` so a failure names the offending dataType/predicate.
##########
spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala:
##########
@@ -237,6 +238,114 @@ class CometFuzzIcebergSuite extends CometFuzzIcebergBase {
}
}
+ test("filter pushdown - IS NULL/IS NOT NULL on nested fuzz columns stays
native") {
+ val df = spark.table(icebergTableName)
+ val complexColumns = df.schema.fields.filter(f =>
isComplexType(f.dataType)).map(_.name)
+ assert(complexColumns.nonEmpty, "expected complex columns in the fuzz
schema")
+
+ for (name <- complexColumns; predicate <- Seq(col(name).isNull,
col(name).isNotNull)) {
+ withClue(predicate.toString) {
+ val (_, cometPlan) = checkSparkAnswer(df.where(predicate))
+ assert(collectIcebergNativeScans(cometPlan).length == 1, s"$cometPlan")
+ }
+ }
+ }
+
+ test("filter pushdown - IS NULL/IS NOT NULL on list, map and struct columns
stays native") {
+ val tableName = "hadoop_catalog.db.null_check_test"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $tableName (
+ id INT, l ARRAY<STRUCT<a: INT>>, m MAP<STRING, STRUCT<a: INT>>, s
STRUCT<a: INT>
+ ) USING iceberg
+ """)
+ // Container nullness is distinct from emptiness, null elements and null
struct fields.
+ spark.sql(s"""
+ INSERT INTO $tableName VALUES
+ (1, array(named_struct('a', 1)), map('k', named_struct('a', 1)),
named_struct('a', 1)),
+ (2, NULL, NULL, NULL),
+ (3, array(), map(), named_struct('a', NULL)),
+ (4, array(NULL), map('k', NULL), named_struct('a', NULL)),
+ (5, array(named_struct('a', NULL)), map('k', named_struct('a',
NULL)), named_struct('a', NULL))
+ """)
+ for (column <- Seq("l", "m", "s"); predicate <- Seq("IS NULL", "IS NOT
NULL")) {
+ val query = s"SELECT id FROM $tableName WHERE $column $predicate"
+ withClue(query) {
+ val (_, cometPlan) = checkSparkAnswer(query)
+ val expected = if (predicate == "IS NULL") Seq(Row(2)) else Seq(1,
3, 4, 5).map(Row(_))
+ checkAnswer(spark.sql(query), expected)
+ val scans = collectIcebergNativeScans(cometPlan)
+ assert(scans.length == 1, s"$cometPlan")
+ // Planning commonData leaks manifest streams on Iceberg versions
before 1.8.0.
+ if (!isIcebergVersionLessThan("1.8.0")) {
+ val common =
OperatorOuterClass.IcebergScanCommon.parseFrom(scans.head.commonData)
+ assert(
+ common.getResidualPoolCount == 0,
+ s"unexpected complex-column residual: $query")
+ }
+ }
+ }
+
+ // Spark infers IS NOT NULL below ordinary generators, but not outer
generators.
+ // Compare complete rows to preserve null elements and distinguish null
struct parents
+ // from non-null structs with null fields. Native scanning does not
imply residual pushdown.
+ for (column <- Seq("l", "m"); generator <- Seq("explode",
"explode_outer")) {
+ val query = s"SELECT id, $generator($column) FROM $tableName"
+ withClue(query) {
+ val (_, cometPlan) = checkSparkAnswer(query)
+ assert(collectIcebergNativeScans(cometPlan).length == 1,
s"$cometPlan")
+ }
+ }
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $tableName")
+ }
+ }
+ test("filter pushdown - required field projection preserves null array
elements") {
+ import org.apache.iceberg.Schema
+ import org.apache.iceberg.catalog.TableIdentifier
+ import org.apache.iceberg.spark.SparkCatalog
+ import org.apache.iceberg.types.Types
Review Comment:
**6. This test is in the wrong suite.**
It builds an explicit Iceberg table through the Java API with a hand-written
schema, so it is not a fuzz test. It belongs in `CometIcebergNativeSuite` next
to `withTempIcebergDir`, which already has the helpers for exactly this.
Also:
- the `filter pushdown - ` name prefix is wrong; this tests projection
nullability;
- `WHERE l IS NOT NULL` is incidental to the nullability fix and can go;
- imports should be at the file header rather than in the method body,
matching the rest of the suite;
- missing blank line between this test and the one above.
##########
spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala:
##########
@@ -237,6 +238,114 @@ class CometFuzzIcebergSuite extends CometFuzzIcebergBase {
}
}
+ test("filter pushdown - IS NULL/IS NOT NULL on nested fuzz columns stays
native") {
+ val df = spark.table(icebergTableName)
+ val complexColumns = df.schema.fields.filter(f =>
isComplexType(f.dataType)).map(_.name)
+ assert(complexColumns.nonEmpty, "expected complex columns in the fuzz
schema")
+
+ for (name <- complexColumns; predicate <- Seq(col(name).isNull,
col(name).isNotNull)) {
+ withClue(predicate.toString) {
+ val (_, cometPlan) = checkSparkAnswer(df.where(predicate))
+ assert(collectIcebergNativeScans(cometPlan).length == 1, s"$cometPlan")
+ }
+ }
+ }
Review Comment:
**4. Three layers now assert the same thing.**
This test, the `null_check_test` test below, and the four updated
`checkIcebergNativeScan` cases in `CometIcebergNativeSuite` all assert
"container null check keeps one native scan". Suggest one layer per concern:
- fuzz sweep (this test) for generated shapes - arrays of structs, structs
of arrays;
- a SQL file for row correctness (see the comment below);
- `CometIcebergNativeSuite` for the residual-pool / serde gate.
As written this test is the most redundant of the three, since the
schema-driven sweep asserts strictly less than the hand-built table does.
##########
docs/source/user-guide/latest/iceberg.md:
##########
@@ -98,6 +98,13 @@ The native Iceberg reader supports the following features:
- `IN` and `NOT IN` list operations
- `BETWEEN` operations
+Native scanning does not imply that every predicate is evaluated inside
iceberg-rust.
+List, map, and struct NULL checks can use native scans. Their residuals are
omitted
+before native planning; the retained post-scan filter enforces them. Empty
collections
+and collections containing null elements are non-null, matching Spark. A
conjunction
+containing one of these residuals currently loses native row-group pruning for
primitive conjuncts;
+safe partial pruning is tracked in
[#5883](https://github.com/apache/datafusion-comet/issues/5883).
+
Review Comment:
**9. (nit)** This paragraph is inserted between the feature bullet list and
the next `**Partitioning:**` heading, so it reads as an orphaned block inside
the list. Consider moving it below the whole feature list, or under its own
small heading such as "Predicate pushdown vs native scanning".
Wording nit: "Their residuals are omitted before native planning" - they are
omitted at serialization time, in `CometIcebergNativeScan`, which is a bit more
precise.
##########
spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala:
##########
@@ -237,6 +238,114 @@ class CometFuzzIcebergSuite extends CometFuzzIcebergBase {
}
}
+ test("filter pushdown - IS NULL/IS NOT NULL on nested fuzz columns stays
native") {
+ val df = spark.table(icebergTableName)
+ val complexColumns = df.schema.fields.filter(f =>
isComplexType(f.dataType)).map(_.name)
+ assert(complexColumns.nonEmpty, "expected complex columns in the fuzz
schema")
+
+ for (name <- complexColumns; predicate <- Seq(col(name).isNull,
col(name).isNotNull)) {
+ withClue(predicate.toString) {
+ val (_, cometPlan) = checkSparkAnswer(df.where(predicate))
+ assert(collectIcebergNativeScans(cometPlan).length == 1, s"$cometPlan")
+ }
+ }
+ }
+
+ test("filter pushdown - IS NULL/IS NOT NULL on list, map and struct columns
stays native") {
+ val tableName = "hadoop_catalog.db.null_check_test"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $tableName (
+ id INT, l ARRAY<STRUCT<a: INT>>, m MAP<STRING, STRUCT<a: INT>>, s
STRUCT<a: INT>
+ ) USING iceberg
+ """)
+ // Container nullness is distinct from emptiness, null elements and null
struct fields.
+ spark.sql(s"""
+ INSERT INTO $tableName VALUES
+ (1, array(named_struct('a', 1)), map('k', named_struct('a', 1)),
named_struct('a', 1)),
+ (2, NULL, NULL, NULL),
+ (3, array(), map(), named_struct('a', NULL)),
+ (4, array(NULL), map('k', NULL), named_struct('a', NULL)),
+ (5, array(named_struct('a', NULL)), map('k', named_struct('a',
NULL)), named_struct('a', NULL))
+ """)
+ for (column <- Seq("l", "m", "s"); predicate <- Seq("IS NULL", "IS NOT
NULL")) {
+ val query = s"SELECT id FROM $tableName WHERE $column $predicate"
+ withClue(query) {
+ val (_, cometPlan) = checkSparkAnswer(query)
+ val expected = if (predicate == "IS NULL") Seq(Row(2)) else Seq(1,
3, 4, 5).map(Row(_))
+ checkAnswer(spark.sql(query), expected)
+ val scans = collectIcebergNativeScans(cometPlan)
+ assert(scans.length == 1, s"$cometPlan")
Review Comment:
**3b. This maps directly onto the SQL-file harness.**
`spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql`
shows the pattern, including per-file `-- Config: spark.sql.catalog....` lines
for the Iceberg catalog. A new `sql-tests/iceberg/complex_null_checks.sql`
would cover the `l`/`m`/`s` x `IS NULL`/`IS NOT NULL` matrix, and `query` runs
`checkSparkAnswerAndOperator`, which is a *stronger* assertion than
`assert(scans.length == 1)` because it also pins the retained filter as native.
Only the residual-pool assertion needs to stay in Scala.
Also: `checkSparkAnswer(query)` followed by `checkAnswer(spark.sql(query),
expected)` executes each of these six queries three times. The explicit
expected rows are worth keeping (they pin "empty collection is not null"
independently of Spark), but they would come for free in the SQL file.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -1023,7 +969,7 @@ case class CometScanRule(session: SparkSession)
defaultValuesSupported && schemaTypesSupported &&
encryptionKeyLengthSupported &&
taskValidation.allParquet && allSupportedFilesystems &&
allLocationsOpenable &&
metadataSchemeSupported && partitionTypesSupported &&
unifiedPartitionTypeSupported &&
- complexTypePredicatesSupported && transformFunctionsSupported &&
+ transformFunctionsSupported &&
Review Comment:
**8. (nit)** Leftover from removing `complexTypePredicatesSupported &&` -
`transformFunctionsSupported &&` now sits alone on its own line. Fold it onto
the line above.
--
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]
