This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 8e7cee8189 fix: restrict array_filter array_compact fast path to the
lambda variable (#4848)
8e7cee8189 is described below
commit 8e7cee8189b49c724ebf03eac8e17f4815a842c4
Author: Andy Grove <[email protected]>
AuthorDate: Mon Jul 13 14:11:15 2026 -0600
fix: restrict array_filter array_compact fast path to the lambda variable
(#4848)
---
.../main/scala/org/apache/comet/serde/arrays.scala | 12 ++++---
.../apache/comet/shims/Spark4xCometExprShim.scala | 9 +++--
.../sql-tests/expressions/array/array_filter.sql | 40 ++++++++++++++++++++++
3 files changed, 53 insertions(+), 8 deletions(-)
diff --git a/spark/src/main/scala/org/apache/comet/serde/arrays.scala
b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
index 7039d3278b..f92827d3b1 100644
--- a/spark/src/main/scala/org/apache/comet/serde/arrays.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/arrays.scala
@@ -22,7 +22,7 @@ package org.apache.comet.serde
import scala.annotation.tailrec
import scala.jdk.CollectionConverters._
-import org.apache.spark.sql.catalyst.expressions.{And, ArrayAggregate,
ArrayAppend, ArrayContains, ArrayExcept, ArrayExists, ArrayFilter, ArrayForAll,
ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayPosition,
ArrayRemove, ArrayRepeat, ArraySort, ArraysOverlap, ArraysZip, ArrayTransform,
ArrayUnion, Attribute, Cast, CreateArray, ElementAt, EmptyRow, Expression,
Flatten, GetArrayItem, IsNotNull, Literal, Reverse, Sequence, Size, Slice,
SortArray, ZipWith}
+import org.apache.spark.sql.catalyst.expressions.{And, ArrayAggregate,
ArrayAppend, ArrayContains, ArrayExcept, ArrayExists, ArrayFilter, ArrayForAll,
ArrayInsert, ArrayIntersect, ArrayJoin, ArrayMax, ArrayMin, ArrayPosition,
ArrayRemove, ArrayRepeat, ArraySort, ArraysOverlap, ArraysZip, ArrayTransform,
ArrayUnion, Attribute, Cast, CreateArray, ElementAt, EmptyRow, Expression,
Flatten, GetArrayItem, IsNotNull, LambdaFunction, Literal, NamedLambdaVariable,
Reverse, Sequence, Size, Slice, [...]
import org.apache.spark.sql.catalyst.util.GenericArrayData
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -721,10 +721,12 @@ object CometArrayFilter extends
CometExpressionSerde[ArrayFilter] {
expr: ArrayFilter,
inputs: Seq[Attribute],
binding: Boolean): Option[ExprOuterClass.Expr] = {
- expr.function.children.headOption match {
- case Some(_: IsNotNull) =>
- // Fast path: `array_compact` lowers to `filter(arr, x -> x is not
null)`. Use the native
- // array_compact serde to avoid the per-batch JNI cost of the codegen
dispatcher.
+ expr.function match {
+ case LambdaFunction(IsNotNull(v: NamedLambdaVariable), Seq(lambdaVar), _)
+ if v.exprId == lambdaVar.exprId =>
+ // Fast path: Catalyst desugars `array_compact` to `filter(arr, x -> x
IS NOT NULL)`, so
+ // restore the native serde here (avoids per-batch JNI). Guard
requires the IsNotNull
+ // operand to be the lambda variable itself, not a captured column
(#4830).
CometArrayCompact.convert(expr, inputs, binding)
case _ =>
// General lambda: run Spark's own evaluation through the codegen
dispatcher so the result
diff --git
a/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
b/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
index 7efd17f68a..a5e77b4a68 100644
--- a/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
+++ b/spark/src/main/spark-4.x/org/apache/comet/shims/Spark4xCometExprShim.scala
@@ -64,11 +64,14 @@ trait Spark4xCometExprShim extends CometExprShim4x {
case knc: KnownNotContainsNull =>
// On Spark 4.0+, array_compact rewrites to
KnownNotContainsNull(ArrayFilter(IsNotNull)).
- // Strip the wrapper and serialize the inner ArrayFilter as
spark_array_compact.
+ // Strip the wrapper and serialize the inner ArrayFilter as
spark_array_compact. The
+ // guard requires the IsNotNull operand to be the lambda's own
variable (matched by
+ // exprId to handle nested-lambda shadowing), not a captured column
(#4830).
knc.child match {
case filter: ArrayFilter =>
- filter.function.children.headOption match {
- case Some(_: IsNotNull) =>
+ filter.function match {
+ case LambdaFunction(IsNotNull(v: NamedLambdaVariable),
Seq(lambdaVar), _)
+ if v.exprId == lambdaVar.exprId =>
val arrayChild = filter.left
val elementType =
arrayChild.dataType.asInstanceOf[ArrayType].elementType
val arrayExprProto = exprToProtoInternal(arrayChild, inputs,
binding)
diff --git
a/spark/src/test/resources/sql-tests/expressions/array/array_filter.sql
b/spark/src/test/resources/sql-tests/expressions/array/array_filter.sql
index c4511e28e4..2c5381fb2d 100644
--- a/spark/src/test/resources/sql-tests/expressions/array/array_filter.sql
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_filter.sql
@@ -29,3 +29,43 @@ SELECT filter(arr, x -> x >= 0) FROM test_array_filter
query
SELECT filter(arr, (x, i) -> i > 0) FROM test_array_filter
+
+statement
+CREATE TABLE test_array_filter_captured(arr array<int>, c int) USING parquet
+
+statement
+INSERT INTO test_array_filter_captured VALUES
+ (array(1, NULL, 2), 5),
+ (array(3, NULL, 4), NULL),
+ (array(NULL, NULL), 7),
+ (NULL, 1)
+
+-- Genuine array_compact fast path: IsNotNull on the lambda variable drops the
null elements.
+query
+SELECT filter(arr, x -> x IS NOT NULL) FROM test_array_filter_captured
+
+-- Regression for #4830: IsNotNull on captured column `c` is not array_compact
and must not drop
+-- the null elements of `arr`.
+query
+SELECT filter(arr, x -> c IS NOT NULL) FROM test_array_filter_captured
+
+-- IsNotNull on an expression of the lambda var: the operand is not a bare
NamedLambdaVariable,
+-- so the guard must reject it and the codegen dispatcher must run Spark's
semantics.
+query
+SELECT filter(arr, x -> (x + 1) IS NOT NULL) FROM test_array_filter_captured
+
+-- Compound predicate combining IsNotNull with another condition: structural
match must fail so
+-- Spark's own evaluation is used.
+query
+SELECT filter(arr, x -> x IS NOT NULL AND x > 1) FROM
test_array_filter_captured
+
+-- Boundary cases for the fast path: null array input and an all-nulls array.
array_compact must
+-- propagate the row-level null and turn the all-nulls array into an empty
array.
+statement
+CREATE TABLE test_array_filter_boundary(arr array<int>) USING parquet
+
+statement
+INSERT INTO test_array_filter_boundary VALUES (NULL), (array()), (array(NULL,
NULL))
+
+query
+SELECT filter(arr, x -> x IS NOT NULL) FROM test_array_filter_boundary
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
