comphead commented on code in PR #5732:
URL: https://github.com/apache/datafusion-comet/pull/5732#discussion_r4038568154
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -628,6 +629,12 @@ 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.
+ // Containers cannot be partition columns, so Iceberg cannot
remove their null
+ // checks from the post-scan filter through exact partition
selection.
+ case IS_NULL | IS_NOT_NULL | NOT_NULL if
isComplexType(attribute.dataType) =>
Review Comment:
**Altitude.** This is the same veto as the `pageIndexUnsupportedColumns`
gate ten lines above
("iceberg-rust cannot bind a residual on this column at all"), re-expressed
one level lower and
narrowed to three operators. Folding it into that `if` is less code and
makes the property
operator-independent:
```scala
// Two reasons iceberg-rust cannot bind a residual on this column
at all, so drop every
// predicate over it (including a unary IS [NOT] NULL) and let the
post-scan CometFilter
// enforce it: the page index rejects the column's type, or the
column is a container and
// iceberg-rust has accessors only for primitive fields.
Containers cannot be partition
// columns, so Iceberg never removes their null checks from the
post-scan filter through
// exact partition selection.
if (pageIndexUnsupportedColumns.contains(columnName) ||
isComplexType(attribute.dataType)) {
None
} else {
operation match {
case IS_NULL => Some(unaryPredicate(columnName,
IcebergPredicateOperator.IsNull))
...
```
Behaviour for null checks is identical. What changes is that `=` / `<` /
`IN` on a complex attribute
becomes *stated* to be unpushable instead of falling out of
`predicateLiteralToProto`'s
`case _ => return None` default arm. Today those are covered by accident,
and adding one Spark type
to that match would silently stop covering them.
`attribute` is already in scope at the `if`, so there is no plumbing.
Keeping the check on
`attribute.dataType` rather than moving it into
`IcebergReflection.pageIndexUnsupportedColumns` is
also the right choice: that set is built from `metadata.tableSchema`, the
*current* schema, so it
would miss a dropped column read under `VERSION AS OF`, while the scan
output attribute is always
present.
##########
native/spark-expr/src/array_funcs/get_array_struct_fields.rs:
##########
@@ -86,23 +95,24 @@ 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)
Review Comment:
This is the right answer, but it is the one line in the file a reader is
most likely to trip over,
because both sibling expressions carry the `|| field.is_nullable()` term
that is deliberately absent
here: `GetStructField::nullable` (`get_struct_field.rs:73-83`) and
`ListExtract::nullable`
(`array_funcs/list_extract.rs:123-135`) both spell out why. One line
matching that convention:
```suggestion
// The output is a list, so it is null exactly where the input list
is. The elements'
// nullability is not part of this flag: it rides in the `List`
field `data_type` returns.
self.child.nullable(input_schema)
```
##########
native/spark-expr/src/array_funcs/get_array_struct_fields.rs:
##########
@@ -86,23 +95,24 @@ 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())?;
match child_value.data_type() {
DataType::List(_) => {
+ let field = self.child_field(batch.schema().as_ref())?;
let list_array = as_list_array(&child_value)?;
- get_array_struct_fields(list_array, self.ordinal)
+ get_array_struct_fields(list_array, self.ordinal, field)
}
DataType::LargeList(_) => {
+ let field = self.child_field(batch.schema().as_ref())?;
let list_array = as_large_list_array(&child_value)?;
- get_array_struct_fields(list_array, self.ordinal)
+ get_array_struct_fields(list_array, self.ordinal, field)
}
data_type => Err(DataFusionError::Internal(format!(
"Unexpected child type for ListExtract: {data_type:?}"
Review Comment:
Two small things in this block.
`self.child_field(...)` is now called once per match arm; the two arms are
otherwise identical modulo
the downcast, so it can be hoisted. And `"Unexpected child type for
ListExtract"` is pre-existing
copy-paste from `array_funcs/list_extract.rs:201` -- the other two error
arms in this same file say
`GetArrayStructFields` (lines 61 and 79). Worth fixing here since the new
test below asserts on that
exact string, which makes the wrong name load-bearing.
`batch.schema_ref()` also avoids the `Arc` clone/drop that `batch.schema()`
does per batch.
```suggestion
// Derived from the schema, not from the values, so the array always
carries the field
// `data_type` declared for it.
let field = self.child_field(batch.schema_ref())?;
match child_value.data_type() {
DataType::List(_) => {
get_array_struct_fields(as_list_array(&child_value)?,
self.ordinal, field)
}
DataType::LargeList(_) => {
get_array_struct_fields(as_large_list_array(&child_value)?,
self.ordinal, field)
}
data_type => Err(DataFusionError::Internal(format!(
"Unexpected child type for GetArrayStructFields:
{data_type:?}"
```
(With the hoist, a non-list child errors out of `list_field` first, so the
new test's assertion needs
to become `.contains("GetArrayStructFields: Int32")`.)
##########
native/spark-expr/src/array_funcs/get_array_struct_fields.rs:
##########
@@ -171,3 +181,114 @@ 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>() {
+ for list_nullable in [true, false] {
+ for element_nullable in [true, false] {
+ for field_nullable in [false, true] {
+ let field = Arc::new(Field::new("a", DataType::Int32,
field_nullable));
+ let values = Arc::new(StructArray::new(
+ vec![field].into(),
+ vec![Arc::new(Int32Array::from(vec![1, 99]))],
+ element_nullable.then(|| NullBuffer::from(vec![true,
false])),
+ ));
+ let list = Arc::new(GenericListArray::<O>::new(
+ Arc::new(Field::new(
+ "element",
+ values.data_type().clone(),
+ element_nullable,
+ )),
+ OffsetBuffer::from_lengths([1, 1, 0, 0]),
+ values,
+ list_nullable.then(|| NullBuffer::from(vec![true,
true, true, false])),
+ ));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "l",
+ list.data_type().clone(),
+ list_nullable,
+ )]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::<GenericListArray<O>>::clone(&list)],
+ )
+ .unwrap();
+ let expr =
GetArrayStructFields::new(Arc::new(Column::new("l", 0)), 0);
+ assert_eq!(expr.nullable(&schema).unwrap(), list_nullable);
+ let result =
expr.evaluate(&batch).unwrap().into_array(4).unwrap();
+ assert_eq!(expr.data_type(&schema).unwrap(),
*result.data_type());
+ let result = result
+ .as_any()
+ .downcast_ref::<GenericListArray<O>>()
+ .unwrap();
+ let output_field = match result.data_type() {
+ DataType::List(field) | DataType::LargeList(field) =>
field,
+ _ => unreachable!(),
+ };
+ assert_eq!(
+ output_field.is_nullable(),
+ element_nullable || field_nullable
+ );
+ let values = result
+ .values()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(values.value(0), 1);
+ assert_eq!(values.is_null(1), element_nullable);
+ assert_eq!(result.offsets(), list.offsets());
+ assert_eq!(result.nulls(), list.nulls());
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn non_list_child_returns_dispatch_error() {
+ let batch = RecordBatch::try_from_iter(vec![(
+ "value",
+ Arc::new(Int32Array::from(vec![1])) as arrow::array::ArrayRef,
+ )])
+ .unwrap();
+ let expr = GetArrayStructFields::new(Arc::new(Column::new("value",
0)), 0);
+ assert!(expr
+ .evaluate(&batch)
+ .unwrap_err()
+ .to_string()
+ .contains("Unexpected child type for ListExtract: Int32"));
+ }
+
+ #[test]
+ fn mismatched_output_field_returns_error() {
Review Comment:
This asserts an Arrow invariant rather than any Comet behaviour: it hands
the private helper an
`Int64` field over `Int32` data and checks that `GenericListArray::try_new`
rejects it. Production
cannot construct that input -- `field` comes from
`child_field(batch.schema())`, the same schema the
values were read against -- so the test can never fail for a Comet reason.
It also pins the private
helper's signature, which is what blocks the hoist suggested above.
Suggest dropping it. If the intent is to record why `new` became `try_new`,
a one-line comment at the
`try_new` call carries that better than a test.
##########
native/spark-expr/src/array_funcs/get_array_struct_fields.rs:
##########
@@ -171,3 +181,114 @@ 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:
`get_struct_field.rs:137-211` is the same fix for structs and is the
established pattern in this
crate: three named `#[test]`s, each 6-12 lines, each with a comment saying
what invariant it pins.
This is one 58-line body run 8 times, and the dimensions do not quite pay
for themselves --
`field_nullable` x `element_nullable` yield only 3 distinct outcomes for the
output field, and
`list_nullable` multiplies all four inner cases by two for one extra bit. A
failure also reports a
line inside the loop rather than which of the 8 configurations broke.
The `assert_eq!(values.value(0), 1)` / `assert_eq!(values.is_null(1),
element_nullable)` pair is
re-testing `child_with_parent_nulls`, which this PR does not touch and which
already has its own unit
tests in `native/common/src/struct_nulls.rs`.
<details>
<summary>Named-test version (5 tests, same coverage; passes <code>cargo
test</code> + <code>clippy -D warnings</code>)</summary>
```rust
#[cfg(test)]
mod tests {
use super::*;
use arrow::array::{ArrayRef, Int32Array, ListArray};
use arrow::buffer::{NullBuffer, OffsetBuffer};
use arrow::datatypes::{Field, SchemaRef};
use datafusion::physical_expr::expressions::Column;
/// `l: list<element: struct<a: int>>` over four rows: one struct in
rows 0 and 1, then two
/// empty lists. Row 1's struct is null when `element_nullable`, and the
list itself is null in
/// row 3 when `list_nullable`.
fn list_of_structs<O: OffsetSizeTrait>(
list_nullable: bool,
element_nullable: bool,
field_nullable: bool,
) -> (SchemaRef, RecordBatch) {
let values = Arc::new(StructArray::new(
vec![Arc::new(Field::new("a", DataType::Int32,
field_nullable))].into(),
vec![Arc::new(Int32Array::from(vec![1, 99]))],
element_nullable.then(|| NullBuffer::from(vec![true, false])),
));
let list = GenericListArray::<O>::new(
Arc::new(Field::new(
"element",
values.data_type().clone(),
element_nullable,
)),
OffsetBuffer::from_lengths([1, 1, 0, 0]),
values,
list_nullable.then(|| NullBuffer::from(vec![true, true, true,
false])),
);
let schema = Arc::new(Schema::new(vec![Field::new(
"l",
list.data_type().clone(),
list_nullable,
)]));
let batch =
RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(list) as
ArrayRef]).unwrap();
(schema, batch)
}
/// Evaluates `l.a` over [`list_of_structs`], checking that the array it
produces carries the
/// type `data_type` declared, and returns the element field of that
list.
fn output_element_field<O: OffsetSizeTrait>(
list_nullable: bool,
element_nullable: bool,
field_nullable: bool,
) -> FieldRef {
let (schema, batch) = list_of_structs::<O>(list_nullable,
element_nullable, field_nullable);
let expr = GetArrayStructFields::new(Arc::new(Column::new("l", 0)),
0);
let array = expr
.evaluate(&batch)
.unwrap()
.into_array(batch.num_rows())
.unwrap();
assert_eq!(expr.data_type(&schema).unwrap(), *array.data_type());
match array.data_type() {
DataType::List(field) | DataType::LargeList(field) =>
Arc::clone(field),
data_type => panic!("expected a list, got {data_type:?}"),
}
}
// A required field inside a NULLABLE list element has to be widened:
the element's null mask is
// unioned into the field's values, so an element field still declared
non-nullable would carry
// nulls and `GenericListArray::try_new` would reject it. LargeList goes
through its own
// `evaluate` arm, so both offset widths are checked.
#[test]
fn required_field_of_nullable_element_is_widened() {
assert!(output_element_field::<i32>(true, true,
false).is_nullable());
assert!(output_element_field::<i64>(true, true,
false).is_nullable());
}
// No over-declaring in the other direction: nothing can null the field
when neither the
// element nor the field itself is nullable.
#[test]
fn required_field_of_required_element_stays_required() {
assert!(!output_element_field::<i32>(true, false,
false).is_nullable());
assert!(!output_element_field::<i64>(false, false,
false).is_nullable());
}
// The list column's own nullability decides `nullable()`; the element's
does not leak into it.
#[test]
fn nullable_follows_the_list_column() {
for list_nullable in [true, false] {
let (schema, _) = list_of_structs::<i32>(list_nullable, true,
true);
let expr = GetArrayStructFields::new(Arc::new(Column::new("l",
0)), 0);
assert_eq!(expr.nullable(&schema).unwrap(), list_nullable);
}
}
// The list's shape passes through unchanged and a null element nulls
the extracted field, even
// though the child buffer holds a value there.
#[test]
fn element_nulls_and_list_shape_are_preserved() {
let (_, batch) = list_of_structs::<i32>(true, true, false);
let input =
batch.column(0).as_any().downcast_ref::<ListArray>().unwrap();
let expr = GetArrayStructFields::new(Arc::new(Column::new("l", 0)),
0);
let array = expr
.evaluate(&batch)
.unwrap()
.into_array(batch.num_rows())
.unwrap();
let output = array.as_any().downcast_ref::<ListArray>().unwrap();
assert_eq!(output.offsets(), input.offsets());
assert_eq!(output.nulls(), input.nulls());
let values = output
.values()
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
assert_eq!(values.value(0), 1);
assert!(values.is_null(1), "a field of a null element must be null");
}
#[test]
fn non_list_child_returns_internal_error() {
let batch = RecordBatch::try_from_iter(vec![(
"value",
Arc::new(Int32Array::from(vec![1])) as ArrayRef,
)])
.unwrap();
let expr = GetArrayStructFields::new(Arc::new(Column::new("value",
0)), 0);
assert!(expr
.evaluate(&batch)
.unwrap_err()
.to_string()
.contains("GetArrayStructFields: Int32"));
}
}
```
</details>
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -2214,8 +2218,90 @@ class CometIcebergNativeSuite
}
}
+ test("complex type null residuals are not serialized") {
Review Comment:
This test needs no Spark session, no catalog and no table -- it is a
pure-function assertion on
`icebergExprToProto` -- but it pays `CometTestBase` session startup and an
`assume(icebergAvailable, ...)` to run here.
`spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala`
is the suite
named for this object, and its own doc comment says "Pure-function
assertions, so a lightweight
`AnyFunSuite` (no Spark session) suffices".
`IcebergWriteProtoTranslationSuite` in the same package is
the precedent for an `AnyFunSuite` there that depends on Iceberg being on
the classpath.
`icebergExprToProto` is already public, so the move is a straight copy (it
does add the first
compile-time `org.apache.iceberg` import to that suite, where the existing
one reaches Iceberg
reflectively -- that compiles fine, since this file already imports
`Expressions` under every
profile):
```scala
/** Converts `predicate` over a single column of `dataType` named "value".
*/
private def residualToProto(predicate: Any, dataType: DataType) =
CometIcebergNativeScan.icebergExprToProto(
predicate,
Seq(AttributeReference("value", dataType)()),
Set.empty)
test("null checks on complex columns are dropped from the serialized
residual") {
// iceberg-rust binds accessors only for primitive fields, so a residual
naming a container
// would warn on every task; the post-scan filter enforces the check
instead.
for (dataType <- Seq(
ArrayType(IntegerType),
MapType(StringType, IntegerType),
new StructType().add("value", IntegerType));
predicate <- Seq(Expressions.isNull("value"),
Expressions.notNull("value"))) {
withClue(s"$dataType: $predicate")(residualToProto(predicate,
dataType) shouldBe empty)
}
residualToProto(Expressions.notNull("value"), IntegerType) should not be
empty
}
```
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -2392,9 +2480,11 @@ class CometIcebergNativeSuite
(3, 'Charlie', array(1, 7, 8))
""")
- checkIcebergNativeScanFallback(
- "SELECT * FROM test_cat.db.array_element_filter_test WHERE
array_contains(values, 1) ORDER BY id",
- "Iceberg Java only pushes down NOT NULL, which iceberg-rust rejects")
+ // The element predicate is not pushed to iceberg-rust (Iceberg Java
only pushes NOT
+ // NULL, which iceberg-rust rejects); the residual is skipped and the
post-scan Comet
+ // filter enforces it while the scan stays native
Review Comment:
The trailing two thirds of this restates what `checkIcebergNativeScan`
asserts and what the serde
comment at `CometIcebergNativeScan.scala:632-635` already explains. Only the
first clause is
non-obvious:
```suggestion
// Iceberg Java pushes only NOT NULL here, which iceberg-rust
rejects, so nothing is
// pushed; the post-scan filter enforces the predicate and the scan
stays native.
```
This same comment appears near-identically at `:2522` and `:2598`, with only
the first clause varying. Worth noting the diff already writes the good version
of it two lines long elsewhere (`:2446`, `:2561`: "The scan stays native; the
retained post-scan filter enforces the list null check.").
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -2214,8 +2218,90 @@ class CometIcebergNativeSuite
}
}
+ test("complex type null residuals are not serialized") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ for (dataType <- Seq(
+ ArrayType(IntegerType),
+ MapType(StringType, IntegerType),
+ new StructType().add("value", IntegerType));
+ predicate <- Seq(Expressions.isNull("value"),
Expressions.notNull("value"))) {
+ withClue(s"$dataType: $predicate") {
+ assert(CometIcebergNativeScan
+ .icebergExprToProto(predicate, Seq(AttributeReference("value",
dataType)()), Set.empty)
+ .isEmpty)
+ }
+ }
+ assert(
+ CometIcebergNativeScan
+ .icebergExprToProto(
+ Expressions.notNull("value"),
+ Seq(AttributeReference("value", IntegerType)()),
+ Set.empty)
+ .nonEmpty)
+ }
+
+ test("required field projection preserves null array elements") {
+ 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 tableName = "test_cat.db.required_array_field_test"
+ val catalog = spark.sessionState.catalogManager
+ .catalog("test_cat")
+ .asInstanceOf[SparkCatalog]
+ val element =
+ Types.StructType.of(Types.NestedField.required(4, "a",
Types.IntegerType.get()))
+ val schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "l", Types.ListType.ofOptional(3,
element)))
+ try {
+ catalog.icebergCatalog.createTable(
+ TableIdentifier.of("db", "required_array_field_test"),
+ schema)
+ spark.sql(s"""
+ INSERT INTO $tableName VALUES
+ (1, array(named_struct('a', 1))),
+ (2, NULL),
+ (3, array()),
+ (4, array(NULL)),
+ (5, array(NULL, named_struct('a', 2)))
+ """)
+ assert(
+ catalog.icebergCatalog
+ .loadTable(TableIdentifier.of("db", "required_array_field_test"))
+ .schema()
+ .findField("l.element.a")
+ .isRequired)
+ val query = s"SELECT id, l.a FROM $tableName"
+ val (_, cometPlan) = checkSparkAnswer(query)
+ assertSingleNativeScan(cometPlan)
+ assert(
+ collect(cometPlan) { case project: CometProjectExec => project
}.nonEmpty,
+ s"$cometPlan")
+ checkAnswer(
Review Comment:
`CometTestBase.checkCometAnswer` (`:374`) exists for exactly this; its doc
says it uses "labels that
correctly identify which side is Comet and which is Spark. This avoids the
misleading 'Spark Answer'
label that Spark's built-in `checkAnswer` would apply to the Comet result."
A mismatch here would
print the Comet rows labelled as Spark's. This suite already calls it at
`:5300`, `:5314`, `:5441`.
```suggestion
checkCometAnswer(
```
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -2429,9 +2519,11 @@ class CometIcebergNativeSuite
(3, 'Charlie', array(1, 2, 3))
""")
- checkIcebergNativeScanFallback(
- "SELECT * FROM test_cat.db.array_value_filter_test WHERE values =
array(1, 2, 3) ORDER BY id",
- "Iceberg Java only pushes down NOT NULL, which iceberg-rust rejects")
+ // The whole-array equality is not pushed to iceberg-rust (Iceberg
Java only pushes NOT
+ // NULL, which iceberg-rust rejects); the residual is skipped and the
post-scan Comet
+ // filter enforces it while the scan stays native
Review Comment:
The trailing two thirds of this restates what `checkIcebergNativeScan`
asserts and what the serde
comment at `CometIcebergNativeScan.scala:632-635` already explains. Only the
first clause is
non-obvious:
```suggestion
// Iceberg Java pushes only NOT NULL here, which iceberg-rust
rejects, so nothing is
// pushed; the post-scan filter enforces the predicate and the scan
stays native.
```
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -2503,9 +2595,11 @@ class CometIcebergNativeSuite
(3, 'Charlie', map('age', 30, 'score', 80))
""")
- checkIcebergNativeScanFallback(
- "SELECT * FROM test_cat.db.map_key_filter_test WHERE
properties['age'] = 30 ORDER BY id",
- "Iceberg Java only pushes down NOT NULL, which iceberg-rust rejects")
+ // The map-key predicate is not pushed to iceberg-rust (Iceberg Java
only pushes NOT
+ // NULL, which iceberg-rust rejects); the residual is skipped and the
post-scan Comet
+ // filter enforces it while the scan stays native
Review Comment:
The trailing two thirds of this restates what `checkIcebergNativeScan`
asserts and what the serde
comment at `CometIcebergNativeScan.scala:632-635` already explains. Only the
first clause is
non-obvious:
```suggestion
// Iceberg Java pushes only NOT NULL here, which iceberg-rust
rejects, so nothing is
// pushed; the post-scan filter enforces the predicate and the scan
stays native.
```
##########
docs/source/user-guide/latest/iceberg.md:
##########
@@ -112,6 +112,15 @@ The native Iceberg reader supports the following features:
- Hadoop Distributed File System (HDFS)
- S3-compatible storage (AWS S3, MinIO)
+### Predicate pushdown vs native scanning
+
+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
+at serialization time; 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:
This section qualifies the `- NULL checks (\`IS NULL\`, \`IS NOT NULL\`)`
bullet at `:97` without
touching it, and it sits *inside* the "supported features" block, so a
reader who stops at the
feature list still gets the unqualified claim. There is also an existing
`### Current limitations`
(`:174`) whose last bullet is the same shape of statement ("partition
pruning still works, but
row-level filtering of these transforms falls back").
Two of the sentences are also below the altitude of a user guide: "Their
residuals are omitted at
serialization time" is a Comet internal a reader cannot act on, and "Empty
collections and
collections containing null elements are non-null, matching Spark" is plain
Spark semantics.
Suggest dropping the section and qualifying the bullet in place instead --
edit `:97` to
```markdown
- NULL checks (`IS NULL`, `IS NOT NULL`) on primitive columns
```
and put one paragraph right after the `BETWEEN` bullet that ends that list:
```markdown
A NULL check on a `struct`, `array` or `map` column still runs on the native
scan and returns correct
results, but it is not pushed into iceberg-rust, which binds accessors only
for primitive fields. It
prunes nothing, and neither does a conjunction containing one
([#5883](https://github.com/apache/datafusion-comet/issues/5883)).
```
##########
spark/src/test/resources/sql-tests/expressions/array/get_array_struct_fields.sql:
##########
@@ -21,5 +21,8 @@ CREATE TABLE test_arr_struct(arr array<struct<name: string,
value: int>>) USING
statement
INSERT INTO test_arr_struct VALUES (array(named_struct('name', 'a', 'value',
1), named_struct('name', 'b', 'value', 2))), (array(named_struct('name', 'x',
'value', 10))), (NULL)
+statement
+INSERT INTO test_arr_struct VALUES (array(NULL, named_struct('name', 'b',
'value', 2)))
Review Comment:
Each `statement` is a separate write job and an extra Parquet file; the
table's nullability is fixed
by the `CREATE TABLE` above, so a fourth `VALUES` tuple is equivalent and
matches the file's own
convention (and `expressions/struct/get_struct_field.sql`'s):
```suggestion
INSERT INTO test_arr_struct VALUES (array(named_struct('name', 'a', 'value',
1), named_struct('name', 'b', 'value', 2))), (array(named_struct('name', 'x',
'value', 10))), (NULL), (array(NULL, named_struct('name', 'b', 'value', 2)))
```
Worth flagging what this row does and does not cover: Spark's v1 file
datasource forces the whole read
schema nullable (`DataSource.scala`, `dataSchema = dataSchema.asNullable`),
so a Parquet table can
never present a required nested field and this row cannot reach the new
widening branch in
`child_field`. It does exercise the parent-null union. The widening itself
is covered by the Rust unit
test and the Iceberg projection test, which is worth saying in the PR
description so the coverage
claim is not read as broader than it is.
##########
spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala:
##########
@@ -237,6 +237,69 @@ 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") {
Review Comment:
This is a fixed-schema, hand-written table in the suite whose whole contract
is "run over the
randomised `FuzzDataGenerator` schema" -- every other test here reads
`icebergTableName`, and
`CometFuzzIcebergBase` has no table-creation or temp-warehouse fixture,
which is why this one has to
hand-roll `try`/`finally` cleanup into the shared warehouse. The l/m/s x `IS
NULL`/`IS NOT NULL`
assertion is also the fourth copy: the same diff rewrote
`complex type filter - struct column IS NULL and IS NOT NULL`, `- array
column IS NULL` and
`- map column IS NULL` in `CometIcebergNativeSuite` to assert exactly this.
Two things here are genuinely new -- `getResidualPoolCount == 0`, and the
`explode` /
`explode_outer` cases (which are the PR's actual motivating scenario and
currently the least
discoverable test in the change). Both have a natural home already in this
file.
Also worth noting the cost: `checkSparkAnswer` runs each query twice (once
with Comet off for the
oracle, once with it on), and line 275 then runs it a *third* time against a
literal. The literal is
the same value for all three columns, so that fact is asserted six times
over -- 18 executions where
3 would do.
Suggested split:
**1. Fold the residual-pool check into the existing test above**, which is
the mirror of
`filter pushdown - residual pool reflects whether the column type is
pushable` and needs no new
fixture or extra queries:
```scala
for (name <- complexColumns; predicate <- Seq(col(name).isNull,
col(name).isNotNull)) {
withClue(predicate.toString) {
val (_, cometPlan) = checkSparkAnswer(df.where(predicate))
val scans = collectIcebergNativeScans(cometPlan)
assert(scans.length == 1, s"$cometPlan")
// The mirror of the primitive residual-pool test above:
iceberg-rust binds accessors only
// for primitive fields, so a container null check must never reach
the residual pool. The
// version gate is inline rather than an `assume` so the native-scan
check above still runs
// on older Iceberg, where reading commonData would leak a manifest
stream.
if (!isIcebergVersionLessThan("1.8.0")) {
val common =
OperatorOuterClass.IcebergScanCommon.parseFrom(scans.head.commonData)
assert(
common.getResidualPoolCount == 0,
s"unexpected residual for complex column '$name' ($predicate)")
}
}
}
```
**2. Keep the generators, over the fuzz table's own array column** -- no
hand-built table needed, and
the element type is irrelevant because the generator sits above the scan, so
2 queries cover the
property instead of 4:
```scala
// Spark infers IS NOT NULL on the generator input below an ordinary
generator but not below an
// outer one, so the two forms put different predicates over the same
complex column. The element
// type is irrelevant here because the generator sits above the scan.
test("filter pushdown - generators over a nested fuzz column stay native")
{
val df = spark.table(icebergTableName)
val arrayColumn = df.schema.fields
.collectFirst { case f if f.dataType.isInstanceOf[ArrayType] => f.name
}
.getOrElse(fail("expected an array column in the fuzz schema"))
for (generator <- Seq("explode", "explode_outer")) {
val query = s"SELECT $generator($arrayColumn) FROM $icebergTableName"
withClue(query) {
val (_, cometPlan) = checkSparkAnswer(query)
assert(collectIcebergNativeScans(cometPlan).length == 1,
s"$cometPlan")
}
}
}
```
**3. Move the deterministic table to `CometIcebergNativeSuite`**, where the
fixed-schema Iceberg
tests live and `withTempIcebergDir` / `checkIcebergNativeScan` already
exist. The empty-collection,
null-element and null-struct-field rows are real coverage worth keeping (the
fuzz schema has no map
columns at all -- `generateMap` defaults to `false` in `SchemaGenOptions`
(`FuzzDataGenerator.scala:336`) and `CometFuzzIcebergBase` does not set it),
and one query pins the
semantics for all three columns at once:
<details>
<summary>Moved test</summary>
```scala
test("complex type filter - list, map and struct null checks stay native")
{
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 tableName = "test_cat.db.complex_null_check_test"
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))
""")
// Only row 2 holds null containers: an empty collection, a
collection of nulls and a
// struct whose every field is null are all themselves non-null. One
query pins that for
// the three columns at once, so the loop below only has to check
the scan stays native.
checkCometAnswer(
spark.sql(s"SELECT id, l IS NULL, m IS NULL, s IS NULL FROM
$tableName ORDER BY id"),
Seq(
Row(1, false, false, false),
Row(2, true, true, true),
Row(3, false, false, false),
Row(4, false, false, false),
Row(5, false, false, false)))
for (column <- Seq("l", "m", "s"); predicate <- Seq("IS NULL", "IS
NOT NULL")) {
checkIcebergNativeScan(
s"SELECT id FROM $tableName WHERE $column $predicate ORDER BY
id")
}
spark.sql(s"DROP TABLE $tableName")
}
}
}
```
</details>
With this, the `Row` import added at line 24 is no longer needed in this
file.
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -2214,8 +2218,90 @@ class CometIcebergNativeSuite
}
}
+ test("complex type null residuals are not serialized") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ for (dataType <- Seq(
+ ArrayType(IntegerType),
+ MapType(StringType, IntegerType),
+ new StructType().add("value", IntegerType));
+ predicate <- Seq(Expressions.isNull("value"),
Expressions.notNull("value"))) {
+ withClue(s"$dataType: $predicate") {
+ assert(CometIcebergNativeScan
+ .icebergExprToProto(predicate, Seq(AttributeReference("value",
dataType)()), Set.empty)
+ .isEmpty)
+ }
+ }
+ assert(
+ CometIcebergNativeScan
+ .icebergExprToProto(
+ Expressions.notNull("value"),
+ Seq(AttributeReference("value", IntegerType)()),
+ Set.empty)
+ .nonEmpty)
+ }
+
+ test("required field projection preserves null array elements") {
+ 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 tableName = "test_cat.db.required_array_field_test"
+ val catalog = spark.sessionState.catalogManager
Review Comment:
This drops to the Iceberg Java API with hand-numbered field ids to get a
required field inside a list
element, but Spark DDL expresses that directly. I checked both halves:
- Spark's grammar allows `NOT NULL` on a nested struct field:
`complexColType : errorCapturingIdentifier COLON? dataType
(errorCapturingNot NULL)? commentSpec?`
(`sql/api/.../SqlBaseParser.g4`).
- `SparkTypeToType.struct` maps `!field.nullable()` to
`Types.NestedField.required`, and
`SparkTypeToType.array` maps `containsNull` to `ListType.ofOptional` -- so
`ARRAY<STRUCT<a: INT NOT NULL>>` produces exactly the schema built here by
hand.
The three existing `icebergCatalog.createTable` sites in this file each say
why SQL cannot do the job
(`:2854` -- "uuid is not expressible via Spark SQL CREATE TABLE"); this one
has no such reason.
```scala
val tableName = "test_cat.db.required_array_field_test"
try {
spark.sql(s"""
CREATE TABLE $tableName (id INT, l ARRAY<STRUCT<a: INT NOT
NULL>>) USING iceberg
""")
```
That also removes the four file-level Iceberg imports this PR added. And
once the table comes from
DDL, the schema check becomes worth keeping rather than a restatement of the
line above it -- asserted
on the Spark side, which is what actually reaches the native expression:
```scala
// A required field under a nullable element is the shape the
projection has to widen.
// Only Iceberg preserves it on read; the v1 Parquet source forces
the schema nullable.
val element = spark
.table(tableName)
.schema("l")
.dataType
.asInstanceOf[ArrayType]
.elementType
.asInstanceOf[StructType]
assert(!element("a").nullable, s"expected a required element
field, got $element")
```
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -26,24 +26,28 @@ import java.nio.charset.StandardCharsets.UTF_8
import scala.collection.mutable
import scala.jdk.CollectionConverters._
+import org.apache.iceberg.Schema
+import org.apache.iceberg.catalog.TableIdentifier
import org.apache.iceberg.data.IcebergGenerics
import org.apache.iceberg.expressions.Expressions
-import org.apache.iceberg.spark.Spark3Util
+import org.apache.iceberg.spark.{Spark3Util, SparkCatalog}
+import org.apache.iceberg.types.Types
Review Comment:
These four are redundant: every use site in this file already has its own
method-local import
(`Schema` / `TableIdentifier` / `Types` / `SparkCatalog` at `:1749-1752`,
`:2843-2846`, `:2897-2900`,
plus `TableIdentifier` / `SparkCatalog` at `:5090`, `:5491`, `:5557`,
`:5615`, `:5681`, `:5761`, `:5908`). It compiles because a local
import shadows an identical outer one, but the file now carries both
conventions. If the DDL
suggestion below lands, the new test needs none of them:
```suggestion
import org.apache.iceberg.data.IcebergGenerics
import org.apache.iceberg.expressions.Expressions
import org.apache.iceberg.spark.Spark3Util
```
Two related ones, once the pure-function test moves out and the DDL replaces
the Java API:
- `:38` -- `AttributeReference` has no other use in the file; back to
`import
org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression`.
- `:45` -- `IntegerType` and `MapType` are only used inside the test bodies
at `:3091-3106`, which are
already covered by the method-local `import org.apache.spark.sql.types._`
at `:3084`, so
`{ArrayType, StringType, StructType, TimestampType}` is enough.
- `:50` -- `import org.apache.comet.serde.operator.CometIcebergNativeScan`
becomes unused (the only
other occurrences of that name in the file are inside comment strings).
--
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]
