Skip to content

Commit 7cc641a

Browse files
ErikBPFErikBPF
committed
fix: handle complex Iceberg null predicates
1 parent 296a8a4 commit 7cc641a

5 files changed

Lines changed: 69 additions & 67 deletions

File tree

‎docs/source/user-guide/latest/iceberg.md‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -99,10 +99,11 @@ The native Iceberg reader supports the following features:
9999
- `BETWEEN` operations
100100

101101
Native scanning does not imply that every predicate is evaluated inside iceberg-rust.
102-
List/map NULL checks can use native scans; when a residual cannot be bound, the
103-
retained post-scan filter enforces it. Direct struct-column NULL checks retain Spark
104-
scan fallback. Empty collections and collections containing null elements are
105-
non-null, matching Spark.
102+
List, map, and struct NULL checks can use native scans. Their residuals are omitted
103+
before native planning; the retained post-scan filter enforces them. Empty collections
104+
and collections containing null elements are non-null, matching Spark. A conjunction
105+
containing one of these residuals currently loses native row-group pruning for primitive conjuncts;
106+
safe partial pruning is tracked in [#5883](https://github.com/apache/datafusion-comet/issues/5883).
106107

107108
**Partitioning:**
108109

‎native/spark-expr/src/array_funcs/get_array_struct_fields.rs‎

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -159,12 +159,12 @@ fn get_array_struct_fields<O: OffsetSizeTrait>(
159159
// parent's was dropped.
160160
let data = child_with_parent_nulls(values, ordinal)?;
161161

162-
let array = GenericListArray::new(
162+
let array = GenericListArray::try_new(
163163
field,
164164
list_array.offsets().clone(),
165165
data,
166166
list_array.nulls().cloned(),
167-
);
167+
)?;
168168

169169
Ok(ColumnarValue::Array(Arc::new(array)))
170170
}
@@ -247,6 +247,27 @@ mod tests {
247247
}
248248
}
249249

250+
#[test]
251+
fn mismatched_output_field_returns_error() {
252+
let values = Arc::new(StructArray::new(
253+
vec![Arc::new(Field::new("a", DataType::Int32, false))].into(),
254+
vec![Arc::new(Int32Array::from(vec![1]))],
255+
None,
256+
));
257+
let list = GenericListArray::<i32>::new(
258+
Arc::new(Field::new("element", values.data_type().clone(), false)),
259+
OffsetBuffer::from_lengths([1]),
260+
values,
261+
None,
262+
);
263+
assert!(get_array_struct_fields(
264+
&list,
265+
0,
266+
Arc::new(Field::new("element", DataType::Int64, false)),
267+
)
268+
.is_err());
269+
}
270+
250271
#[test]
251272
fn get_array_struct_fields_nullability() {
252273
check_nullability::<i32>();

‎spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala‎

Lines changed: 1 addition & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -830,58 +830,6 @@ case class CometScanRule(session: SparkSession)
830830
true
831831
}
832832

833-
// Get filter expressions for complex predicates check
834-
val filterExpressionsOpt = IcebergReflection.getFilterExpressions(scanExec.scan)
835-
836-
// Retain the conservative fallback for direct struct-column null checks. List/map
837-
// null checks can scan natively: the native planner skips residuals it cannot bind,
838-
// while Iceberg's retained post-scan filter enforces row-level correctness.
839-
val complexTypePredicatesSupported = filterExpressionsOpt
840-
.map { filters =>
841-
// Empty filters can't trigger accessor issues
842-
if (filters.isEmpty) {
843-
true
844-
} else {
845-
val readSchema = scanExec.scan.readSchema()
846-
847-
// A struct inside a list/map does not make the container null check a struct check.
848-
val structColumns = readSchema
849-
.filter(field => field.dataType.isInstanceOf[StructType])
850-
.map(_.name)
851-
.toSet
852-
853-
// Detect IS NULL/NOT NULL on struct columns (pattern: is_null(ref(name="col")))
854-
// Nested field filters use different patterns and don't trigger this issue
855-
val hasStructNullCheck = filters.asScala.exists { expr =>
856-
val exprStr = expr.toString
857-
val isNullCheck = exprStr.contains("is_null") || exprStr.contains("not_null")
858-
if (isNullCheck) {
859-
structColumns.exists { colName =>
860-
exprStr.contains(s"""ref(name="$colName")""")
861-
}
862-
} else {
863-
false
864-
}
865-
}
866-
867-
if (hasStructNullCheck) {
868-
fallbackReasons += "IS NULL / IS NOT NULL predicates on struct type columns " +
869-
"require Spark scan fallback"
870-
false
871-
} else {
872-
true
873-
}
874-
}
875-
}
876-
.getOrElse {
877-
// Fall back to Spark if reflection fails - cannot verify safety
878-
val msg =
879-
"Iceberg reflection failure: Could not check for complex type predicates"
880-
logError(msg)
881-
fallbackReasons += msg
882-
false
883-
}
884-
885833
// Check for unsupported transform functions in residual expressions
886834
// iceberg-rust can only handle identity transforms in residuals; all other transforms
887835
// (truncate, bucket, year, month, day, hour) must fall back to Spark
@@ -1021,7 +969,7 @@ case class CometScanRule(session: SparkSession)
1021969
defaultValuesSupported && schemaTypesSupported && encryptionKeyLengthSupported &&
1022970
taskValidation.allParquet && allSupportedFilesystems && allLocationsOpenable &&
1023971
metadataSchemeSupported && partitionTypesSupported && unifiedPartitionTypeSupported &&
1024-
complexTypePredicatesSupported && transformFunctionsSupported &&
972+
transformFunctionsSupported &&
1025973
deleteFileTypesSupported && dppSubqueriesSupported) {
1026974
CometBatchScanExec(
1027975
scanExec.clone().asInstanceOf[BatchScanExec],

‎spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -628,6 +628,13 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
628628
None
629629
} else {
630630
operation match {
631+
// iceberg-rust has accessors only for primitive fields, not containers. Keep
632+
// the post-scan filter without sending residuals that would warn on every task.
633+
case IS_NULL | IS_NOT_NULL | NOT_NULL
634+
if attribute.dataType.isInstanceOf[ArrayType] ||
635+
attribute.dataType.isInstanceOf[MapType] ||
636+
attribute.dataType.isInstanceOf[StructType] =>
637+
None
631638
case IS_NULL => Some(unaryPredicate(columnName, IcebergPredicateOperator.IsNull))
632639
case IS_NOT_NULL | NOT_NULL =>
633640
Some(unaryPredicate(columnName, IcebergPredicateOperator.NotNull))

‎spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala‎

Lines changed: 33 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2214,8 +2214,31 @@ class CometIcebergNativeSuite
22142214
}
22152215
}
22162216

2217+
test("complex type null residuals are not serialized") {
2218+
import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType, StructType}
2219+
import org.apache.comet.serde.operator.CometIcebergNativeScan
2220+
import org.apache.spark.sql.catalyst.expressions.AttributeReference
2221+
for (dataType <- Seq(
2222+
ArrayType(IntegerType),
2223+
MapType(StringType, IntegerType),
2224+
new StructType().add("value", IntegerType));
2225+
predicate <- Seq(Expressions.isNull("value"), Expressions.notNull("value"))) {
2226+
assert(
2227+
CometIcebergNativeScan
2228+
.icebergExprToProto(predicate, Seq(AttributeReference("value", dataType)()), Set.empty)
2229+
.isEmpty)
2230+
}
2231+
assert(
2232+
CometIcebergNativeScan
2233+
.icebergExprToProto(
2234+
Expressions.notNull("value"),
2235+
Seq(AttributeReference("value", IntegerType)()),
2236+
Set.empty)
2237+
.nonEmpty)
2238+
}
2239+
22172240
// Complex type filter tests
2218-
test("complex type filter - struct column IS NULL") {
2241+
test("complex type filter - struct column IS NULL and IS NOT NULL") {
22192242
assume(icebergAvailable, "Iceberg not available in classpath")
22202243

22212244
withTempIcebergDir { warehouseDir =>
@@ -2240,12 +2263,14 @@ class CometIcebergNativeSuite
22402263
VALUES
22412264
(1, 'Alice', struct('NYC', 10001)),
22422265
(2, 'Bob', struct('LA', 90001)),
2243-
(3, 'Charlie', NULL)
2266+
(3, 'Charlie', NULL),
2267+
(4, 'Dana', struct(CAST(NULL AS STRING), CAST(NULL AS INT)))
22442268
""")
22452269

2246-
checkIcebergNativeScanFallback(
2247-
"SELECT * FROM test_cat.db.struct_filter_test WHERE address IS NULL ORDER BY id",
2248-
"iceberg-rust does not support IS NULL on struct type columns")
2270+
checkIcebergNativeScan(
2271+
"SELECT * FROM test_cat.db.struct_filter_test WHERE address IS NULL ORDER BY id")
2272+
checkIcebergNativeScan(
2273+
"SELECT * FROM test_cat.db.struct_filter_test WHERE address IS NOT NULL ORDER BY id")
22492274

22502275
spark.sql("DROP TABLE test_cat.db.struct_filter_test")
22512276
}
@@ -2318,9 +2343,9 @@ class CometIcebergNativeSuite
23182343
(3, 'Charlie', named_struct('city', 'NYC', 'zip', 10001))
23192344
""")
23202345

2321-
checkIcebergNativeScanFallback(
2322-
"SELECT * FROM test_cat.db.struct_value_filter_test WHERE address = named_struct('city', 'NYC', 'zip', 10001) ORDER BY id",
2323-
"Iceberg Java does not push down whole-struct equality filters")
2346+
// Spark retains the whole-struct equality filter above the native scan.
2347+
checkIcebergNativeScan(
2348+
"SELECT * FROM test_cat.db.struct_value_filter_test WHERE address = named_struct('city', 'NYC', 'zip', 10001) ORDER BY id")
23242349

23252350
spark.sql("DROP TABLE test_cat.db.struct_value_filter_test")
23262351
}

0 commit comments

Comments
 (0)