This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8b330ca3be [spark] Push down array predicates to multivalue indexes
(#9306)
8b330ca3be is described below
commit 8b330ca3be9ab517f37ee12bdabce2d9082354e2
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Aug 20 08:53:03 2026 +0800
[spark] Push down array predicates to multivalue indexes (#9306)
---
.../sql/ArrayPredicatePushDownVersionTest.scala | 24 +++++
.../sql/ArrayPredicatePushDownVersionTest.scala | 24 +++++
.../paimon/spark/SparkV2FilterConverter.scala | 21 ++++-
.../optimizer/PushDownArrayPredicates.scala | 102 +++++++++++++++++++++
.../extensions/PaimonSparkSessionExtensions.scala | 7 +-
.../spark/util/SparkExpressionConverter.scala | 28 +++++-
.../ArrayPredicatePushDownVersionTestBase.scala | 77 ++++++++++++++++
.../spark/sql/PrimaryKeySortedIndexTest.scala | 70 +++++++++++++-
8 files changed, 346 insertions(+), 7 deletions(-)
diff --git
a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/ArrayPredicatePushDownVersionTest.scala
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/ArrayPredicatePushDownVersionTest.scala
new file mode 100644
index 0000000000..63decd9230
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/ArrayPredicatePushDownVersionTest.scala
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.spark.sql
+
+class ArrayPredicatePushDownVersionTest extends
ArrayPredicatePushDownVersionTestBase {
+
+ override protected def supportsArrayPredicatePushDown: Boolean = false
+}
diff --git
a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/ArrayPredicatePushDownVersionTest.scala
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/ArrayPredicatePushDownVersionTest.scala
new file mode 100644
index 0000000000..ca9049335a
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/ArrayPredicatePushDownVersionTest.scala
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.spark.sql
+
+class ArrayPredicatePushDownVersionTest extends
ArrayPredicatePushDownVersionTestBase {
+
+ override protected def supportsArrayPredicatePushDown: Boolean = true
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkV2FilterConverter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkV2FilterConverter.scala
index edece9340b..168c518cd3 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkV2FilterConverter.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkV2FilterConverter.scala
@@ -19,12 +19,13 @@
package org.apache.paimon.spark
import org.apache.paimon.predicate.{Predicate, PredicateBuilder, Transform}
-import org.apache.paimon.spark.util.SparkExpressionConverter.{toPaimonLiteral,
toPaimonTransform}
-import org.apache.paimon.types.RowType
+import
org.apache.paimon.spark.util.SparkExpressionConverter.{toPaimonArrayLiteral,
toPaimonLiteral, toPaimonTransform}
+import org.apache.paimon.types.{ArrayType, RowType}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.connector.expressions.{Expression, Literal}
import org.apache.spark.sql.connector.expressions.filter.{And, Not, Or,
Predicate => SparkPredicate}
+import org.apache.spark.sql.types.{ArrayType => SparkArrayType}
import scala.collection.JavaConverters._
@@ -161,6 +162,11 @@ case class SparkV2FilterConverter(rowType: RowType)
extends Logging {
case STRING_CONTAINS =>
sparkPredicate match {
+ case ArrayLiteralPredicate(transform, literals) =>
+ builder.arraysOverlap(transform, literals.asJava)
+ case BinaryPredicate(transform, literal)
+ if transform.outputType().isInstanceOf[ArrayType] =>
+ builder.arrayContains(transform, literal)
case BinaryPredicate(transform, literal) =>
builder.contains(transform, literal)
case _ =>
@@ -217,6 +223,17 @@ case class SparkV2FilterConverter(rowType: RowType)
extends Logging {
}
}
}
+
+ private object ArrayLiteralPredicate {
+ def unapply(sparkPredicate: SparkPredicate): Option[(Transform,
Seq[Object])] = {
+ sparkPredicate.children() match {
+ case Array(e: Expression, literal: Literal[_])
+ if literal.dataType().isInstanceOf[SparkArrayType] =>
+ toPaimonTransform(e, rowType).map((_, toPaimonArrayLiteral(literal)))
+ case _ => None
+ }
+ }
+ }
}
object SparkV2FilterConverter extends Logging {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownArrayPredicates.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownArrayPredicates.scala
new file mode 100644
index 0000000000..093d2df736
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownArrayPredicates.scala
@@ -0,0 +1,102 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.spark.catalyst.optimizer
+
+import org.apache.paimon.spark.catalyst.analysis.PaimonRelation
+
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions.{ArrayContains,
ArraysOverlap, BinaryComparison, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
ExprCode}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+
+/** Makes literal array predicates visible to Spark's V2 predicate translator
for Paimon scans. */
+object PushDownArrayPredicates extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.transformUp {
+ case filter @ Filter(condition, child) if
PaimonRelation.isPaimonTable(child) =>
+ filter.copy(condition = condition.transformDown {
+ case ArrayContains(array, literal: Literal)
+ if array.references.nonEmpty &&
+ array.references.subsetOf(child.outputSet) &&
+ literal.value != null =>
+ PaimonArrayContains(array, literal)
+
+ case ArraysOverlap(array, literal: Literal)
+ if array.references.nonEmpty &&
array.references.subsetOf(child.outputSet) =>
+ PaimonArraysOverlap(array, literal)
+
+ case ArraysOverlap(literal: Literal, array)
+ if array.references.nonEmpty &&
array.references.subsetOf(child.outputSet) =>
+ PaimonArraysOverlap(array, literal)
+ })
+ }
+}
+
+/**
+ * Spark's V2 expression builder translates [[BinaryComparison]] generically,
but does not know
+ * [[ArrayContains]]. This comparison-shaped wrapper keeps Spark's original
evaluation semantics for
+ * the residual filter while exposing an array membership marker to the
connector.
+ */
+private[spark] case class PaimonArrayContains(left: Expression, right:
Expression)
+ extends BinaryComparison {
+
+ private def original: ArrayContains = ArrayContains(left, right)
+
+ // Use a V2 predicate name understood by Spark's SQL renderer. The Paimon
converter
+ // distinguishes string and array CONTAINS predicates from the transform
output type.
+ override def symbol: String = "CONTAINS"
+
+ override def nullable: Boolean = original.nullable
+
+ override def checkInputDataTypes(): TypeCheckResult =
original.checkInputDataTypes()
+
+ override protected def nullSafeEval(leftValue: Any, rightValue: Any): Any =
+ original.nullSafeEval(leftValue, rightValue)
+
+ override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode =
+ original.doGenCode(ctx, ev)
+
+ override protected def withNewChildrenInternal(
+ newLeft: Expression,
+ newRight: Expression): PaimonArrayContains = copy(left = newLeft, right
= newRight)
+}
+
+/** Comparison-shaped V2 bridge for Spark's [[ArraysOverlap]] expression. */
+private[spark] case class PaimonArraysOverlap(left: Expression, right:
Expression)
+ extends BinaryComparison {
+
+ private def original: ArraysOverlap = ArraysOverlap(left, right)
+
+ override def symbol: String = "CONTAINS"
+
+ override def nullable: Boolean = original.nullable
+
+ override def checkInputDataTypes(): TypeCheckResult =
original.checkInputDataTypes()
+
+ override protected def nullSafeEval(leftValue: Any, rightValue: Any): Any =
+ original.nullSafeEval(leftValue, rightValue)
+
+ override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode =
+ original.doGenCode(ctx, ev)
+
+ override protected def withNewChildrenInternal(
+ newLeft: Expression,
+ newRight: Expression): PaimonArraysOverlap = copy(left = newLeft, right
= newRight)
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
index f856d348a4..b216454ba4 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
@@ -19,7 +19,7 @@
package org.apache.paimon.spark.extensions
import org.apache.paimon.spark.catalyst.analysis.{PaimonAnalysis,
PaimonDeleteTable, PaimonFunctionResolver, PaimonIncompatibleResolutionRules,
PaimonMergeInto, PaimonPostHocResolutionRules, PaimonProcedureResolver,
PaimonUpdateTable, PaimonViewResolver, ReplacePaimonFunctions}
-import
org.apache.paimon.spark.catalyst.optimizer.{MergePaimonScalarSubqueries,
OptimizeMetadataOnlyDeleteFromPaimonTable, PushDownLateralVectorSearchFilter,
RepartitionLateralVectorSearchInput}
+import
org.apache.paimon.spark.catalyst.optimizer.{MergePaimonScalarSubqueries,
OptimizeMetadataOnlyDeleteFromPaimonTable, PushDownArrayPredicates,
PushDownLateralVectorSearchFilter, RepartitionLateralVectorSearchInput}
import
org.apache.paimon.spark.catalyst.plans.logical.PaimonTableValuedFunctions
import org.apache.paimon.spark.commands.BucketExpression
import org.apache.paimon.spark.execution.{OldCompatibleStrategy,
PaimonStrategy}
@@ -100,6 +100,11 @@ class PaimonSparkSessionExtensions extends
(SparkSessionExtensions => Unit) {
extensions.injectOptimizerRule(spark => ReplacePaimonFunctions(spark))
extensions.injectOptimizerRule(spark =>
OptimizeMetadataOnlyDeleteFromPaimonTable(spark))
extensions.injectOptimizerRule(_ => MergePaimonScalarSubqueries)
+ // Spark 3.2 uses the V1 filter translation path, which cannot translate
the
+ // comparison-shaped array predicate bridge used by this rule.
+ if (org.apache.spark.SPARK_VERSION >= "3.3") {
+ extensions.injectOptimizerRule(_ => PushDownArrayPredicates)
+ }
extensions.injectOptimizerRule(_ => RepartitionLateralVectorSearchInput)
extensions.injectOptimizerRule(_ => PushDownLateralVectorSearchFilter)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala
index a5ff3598fb..00898c98ca 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala
@@ -25,8 +25,9 @@ import
org.apache.paimon.spark.util.shim.TypeUtils.treatPaimonTimestampTypeAsSpa
import org.apache.paimon.types.{DecimalType, RowType}
import org.apache.paimon.types.DataTypeRoot._
-import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.catalyst.util.{ArrayData, DateTimeUtils}
import org.apache.spark.sql.connector.expressions.{Cast, Expression,
GeneralScalarExpression, Literal, NamedReference}
+import org.apache.spark.sql.types.{ArrayType => SparkArrayType, DataType =>
SparkDataType}
import scala.collection.JavaConverters._
@@ -98,8 +99,29 @@ object SparkExpressionConverter {
throw new UnsupportedOperationException(s"Convert value: $literal is
unsupported.")
}
- val dataType = SparkTypeUtils.toPaimonType(literal.dataType())
- val value = literal.value()
+ toPaimonLiteral(literal.value(), literal.dataType())
+ }
+
+ /** Convert a Spark ARRAY [[Literal]] to Paimon element literals. */
+ def toPaimonArrayLiteral(literal: Literal[_]): Seq[Object] = {
+ literal.dataType() match {
+ case SparkArrayType(elementType, _) =>
+ val array = literal.value().asInstanceOf[ArrayData]
+ (0 until array.numElements()).map {
+ i =>
+ if (array.isNullAt(i)) {
+ null
+ } else {
+ toPaimonLiteral(array.get(i, elementType), elementType)
+ }
+ }
+ case _ =>
+ throw new UnsupportedOperationException(s"Convert value: $literal is
unsupported.")
+ }
+ }
+
+ private def toPaimonLiteral(value: Any, sparkDataType: SparkDataType):
Object = {
+ val dataType = SparkTypeUtils.toPaimonType(sparkDataType)
dataType.getTypeRoot match {
case BOOLEAN | BIGINT | DOUBLE | TINYINT | SMALLINT | INTEGER | FLOAT |
DATE =>
value.asInstanceOf[AnyRef]
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ArrayPredicatePushDownVersionTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ArrayPredicatePushDownVersionTestBase.scala
new file mode 100644
index 0000000000..2c6b3e9f13
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/ArrayPredicatePushDownVersionTestBase.scala
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.spark.sql
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.catalyst.optimizer.{PaimonArrayContains,
PaimonArraysOverlap, PushDownArrayPredicates}
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayContains,
ArraysOverlap, Expression}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+
+abstract class ArrayPredicatePushDownVersionTestBase extends
PaimonSparkTestBase {
+
+ protected def supportsArrayPredicatePushDown: Boolean
+
+ test("register array predicate pushdown only for supported Spark versions") {
+ val registered = spark.sessionState.optimizer.batches
+ .flatMap(_.rules)
+ .exists(_.getClass == PushDownArrayPredicates.getClass)
+
+ assert(registered == supportsArrayPredicatePushDown)
+ }
+
+ test("rewrite array predicates only for supported Spark versions") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, tags ARRAY<STRING>)")
+
+ Seq(
+ "SELECT * FROM t WHERE array_contains(tags, 'red')",
+ "SELECT * FROM t WHERE arrays_overlap(tags, array('red', 'blue'))"
+ ).foreach {
+ query =>
+ val optimizedPlan = sql(query).queryExecution.optimizedPlan
+ assert(
+ containsPaimonArrayPredicate(optimizedPlan) ==
supportsArrayPredicatePushDown,
+ optimizedPlan.toString())
+ assert(
+ containsSparkArrayPredicate(optimizedPlan) !=
supportsArrayPredicatePushDown,
+ optimizedPlan.toString())
+ }
+ }
+ }
+
+ private def containsPaimonArrayPredicate(plan: LogicalPlan): Boolean =
+ containsExpression(
+ plan,
+ {
+ case _: PaimonArrayContains | _: PaimonArraysOverlap => true
+ case _ => false
+ })
+
+ private def containsSparkArrayPredicate(plan: LogicalPlan): Boolean =
+ containsExpression(
+ plan,
+ {
+ case _: ArrayContains | _: ArraysOverlap => true
+ case _ => false
+ })
+
+ private def containsExpression(plan: LogicalPlan, predicate: Expression =>
Boolean): Boolean =
+ plan.find(_.expressions.exists(_.find(predicate).isDefined)).isDefined
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
index 4a09bfc18b..823b099582 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
@@ -18,7 +18,9 @@
package org.apache.paimon.spark.sql
+import org.apache.paimon.data.BinaryString
import org.apache.paimon.globalindex.IndexedSplit
+import org.apache.paimon.predicate.{ArrayContains, ArrayContainsAll,
ArraysOverlap, LeafPredicate, PredicateBuilder}
import org.apache.paimon.spark.PaimonSparkTestBase
import org.apache.paimon.table.source.DataSplit
@@ -26,9 +28,75 @@ import org.apache.spark.sql.Row
import scala.collection.JavaConverters._
-/** End-to-end Spark SQL tests for source-backed primary-key BTree and Bitmap
indexes. */
+/** End-to-end Spark SQL tests for source-backed primary-key sorted indexes. */
class PrimaryKeySortedIndexTest extends PaimonSparkTestBase {
+ test("Spark array predicates use multivalue index") {
+ assume(gteqSpark3_3)
+
+ withTable("t") {
+ spark.sql("""
+ |CREATE TABLE t (id INT, tags ARRAY<STRING>, probe STRING)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '1',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'pk-multivalue.index.columns' = 'tags'
+ |)
+ |""".stripMargin)
+ spark.sql("""
+ |INSERT INTO t VALUES
+ | (1, array('red', 'blue'), 'red'),
+ | (2, array('green'), 'red'),
+ | (3, array('red'), 'blue')
+ |""".stripMargin)
+ spark.sql("""
+ |INSERT INTO t VALUES
+ | (4, array('yellow'), 'yellow'),
+ | (5, array(CAST(NULL AS STRING)), 'red'),
+ | (6, CAST(NULL AS ARRAY<STRING>), 'red')
+ |""".stripMargin)
+ spark.sql("CALL sys.compact(table => 't')")
+
+ val sourceIndexes =
loadTable("t").store.newIndexFileHandler.scanEntries.asScala
+ .map(_.indexFile)
+ .filter(meta => meta.globalIndexMeta != null &&
meta.globalIndexMeta.sourceMeta != null)
+ assert(sourceIndexes.map(_.indexType).toSet == Set("multivalue"))
+
+ val predicateBuilder = new PredicateBuilder(loadTable("t").rowType())
+ val red = BinaryString.fromString("red")
+ val blue = BinaryString.fromString("blue")
+ val green = BinaryString.fromString("green")
+
+ val containsQuery = "SELECT id FROM t WHERE array_contains(tags, 'red')"
+ val containsScan = getPaimonScan(containsQuery)
+
assert(containsScan.pushedDataFilters.contains(predicateBuilder.arrayContains(1,
red)))
+ assert(containsScan.inputSplits.exists(_.isInstanceOf[IndexedSplit]))
+ checkAnswer(spark.sql(containsQuery), Seq(Row(1), Row(3)))
+
+ val overlapsQuery =
+ "SELECT id FROM t WHERE arrays_overlap(tags, array('blue', 'green'))"
+ val overlapsScan = getPaimonScan(overlapsQuery)
+ assert(
+ overlapsScan.pushedDataFilters.contains(
+ predicateBuilder.arraysOverlap(1, Seq(blue, green).asJava)))
+ assert(overlapsScan.inputSplits.exists(_.isInstanceOf[IndexedSplit]))
+ checkAnswer(spark.sql(overlapsQuery), Seq(Row(1), Row(2)))
+
+ val dynamicQuery = "SELECT id FROM t WHERE array_contains(tags, probe)"
+ val dynamicScan = getPaimonScan(dynamicQuery)
+ assert(!dynamicScan.pushedDataFilters.exists {
+ case leaf: LeafPredicate =>
+ leaf.function() == ArrayContains.INSTANCE ||
+ leaf.function() == ArraysOverlap.INSTANCE ||
+ leaf.function() == ArrayContainsAll.INSTANCE
+ case _ => false
+ })
+ assert(!dynamicScan.inputSplits.exists(_.isInstanceOf[IndexedSplit]))
+ checkAnswer(spark.sql(dynamicQuery), Seq(Row(1), Row(4)))
+ }
+ }
+
test("postpone bucket builds and applies sorted indexes during compact") {
withTable("t") {
spark.sql("""
