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 1015cb9871 [spark] Fix table-valued functions failing on Spark 4.0
(#10397)
1015cb9871 is described below
commit 1015cb9871828fd1e7bf751d8db69a87038e080c
Author: YangJie <[email protected]>
AuthorDate: Fri Oct 9 10:02:13 2026 -0400
[spark] Fix table-valued functions failing on Spark 4.0 (#10397)
---
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 10 +-
.../TableValuedFunctionsCompatibilityTest.scala | 124 +++++++++++++++++++++
.../plans/logical/PaimonTableValuedFunctions.scala | 12 +-
.../apache/spark/sql/paimon/shims/SparkShim.scala | 11 +-
.../apache/spark/sql/paimon/shims/Spark3Shim.scala | 10 +-
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 10 +-
6 files changed, 167 insertions(+), 10 deletions(-)
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index c348349fe9..c42c7cda1b 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -42,7 +42,7 @@ import
org.apache.spark.sql.catalyst.plans.logical.MergeRows.Keep
import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution,
Distribution}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.{ArrayData, GeneratedColumn,
IdentityColumn, ResolveDefaultColumns}
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column,
Identifier, StagingTableCatalog, Table, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, CatalogV2Util,
Column, Identifier, StagingTableCatalog, Table, TableCatalog}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.connector.write.BatchWrite
@@ -53,6 +53,7 @@ import
org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, Data
import org.apache.spark.sql.execution.streaming.{FileStreamSink,
MetadataLogFileIndex}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataTypes, StructType, VariantType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.unsafe.types.VariantVal
import java.util.{Map => JMap}
@@ -301,6 +302,13 @@ class Spark4Shim extends SparkShim {
relation.copy(table = table, output = output)
}
+ override def createDataSourceV2Relation(
+ table: Table,
+ catalog: Option[CatalogPlugin],
+ identifier: Option[Identifier],
+ options: CaseInsensitiveStringMap): DataSourceV2Relation =
+ DataSourceV2Relation.create(table, catalog, identifier, options)
+
override def createDataSourceV2ScanRelation(
relation: DataSourceV2ScanRelation,
scan: Scan,
diff --git
a/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsCompatibilityTest.scala
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsCompatibilityTest.scala
new file mode 100644
index 0000000000..368f060021
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsCompatibilityTest.scala
@@ -0,0 +1,124 @@
+/*
+ * 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.SparkTable
+import org.apache.paimon.table.{FullTextSearchTable, HybridSearchTable,
VectorSearchTable}
+
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.plans.logical.Except
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * `TableValuedFunctionsTest` only runs in `paimon-spark-ut`, against the
profile's newest Spark.
+ * These cases run table-valued functions on a Spark 4.0 runtime, which loads
`paimon-spark-common`
+ * classes compiled against a newer Spark.
+ */
+class TableValuedFunctionsCompatibilityTest extends PaimonSparkTestBase {
+
+ private def relationTables(df: DataFrame):
Seq[org.apache.spark.sql.connector.catalog.Table] =
+ df.queryExecution.analyzed.collect { case r: DataSourceV2Relation =>
r.table }
+
+ test("paimon_incremental_query reads the snapshots in range") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT) USING paimon")
+ sql("INSERT INTO t VALUES 1")
+ sql("INSERT INTO t VALUES 2")
+ sql("INSERT INTO t VALUES 3")
+
+ checkAnswer(
+ sql("SELECT * FROM paimon_incremental_query('t', 1, 3) ORDER BY id"),
+ Seq(Row(2), Row(3)))
+ }
+ }
+
+ test("paimon_incremental_query diffs the snapshots when the tags' buckets
differ") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (a INT, b INT) USING paimon
+ |TBLPROPERTIES ('primary-key'='a', 'bucket' = '1')
+ |""".stripMargin)
+
+ val table = loadTable("t")
+ sql("INSERT INTO t VALUES (1, 11), (2, 22)")
+ table.createTag("2024-01-01", 1)
+
+ sql("ALTER TABLE t SET TBLPROPERTIES ('bucket' = '2')")
+ sql("INSERT OVERWRITE t SELECT * FROM t")
+ sql("INSERT INTO t VALUES (3, 33)")
+ table.createTag("2024-01-03", 3)
+
+ val df =
+ sql("SELECT * FROM paimon_incremental_query('t', '2024-01-01',
'2024-01-03') ORDER BY a, b")
+ // The bucket change makes Paimon diff the two snapshots in Spark
instead of in its scan.
+ assert(df.queryExecution.analyzed.collectFirst { case e: Except => e
}.isDefined)
+ checkAnswer(df, Seq(Row(3, 33)))
+ }
+ }
+
+ // The search functions are only analyzed: the relation is built during
analysis.
+
+ test("vector_search resolves to a vector search relation") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, v ARRAY<FLOAT>) USING paimon")
+ val df = sql("SELECT id FROM vector_search('t', 'v', array(1.0f, 0.0f),
1)")
+ relationTables(df) match {
+ case Seq(SparkTable(_: VectorSearchTable)) =>
+ case other => fail(s"Unexpected relations: $other")
+ }
+ }
+ }
+
+ test("hybrid_search resolves to a hybrid search relation") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, v ARRAY<FLOAT>) USING paimon")
+ val df =
+ sql("""
+ |SELECT id FROM hybrid_search(
+ | 't',
+ | array(named_struct('vector_column', 'v', 'query_vector',
array(1.0f, 0.0f))),
+ | array(),
+ | 1)
+ |""".stripMargin)
+ relationTables(df) match {
+ case Seq(SparkTable(_: HybridSearchTable)) =>
+ case other => fail(s"Unexpected relations: $other")
+ }
+ }
+ }
+
+ test("full_text_search resolves to a full-text search relation") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, content STRING) USING paimon")
+ val df =
+ sql("""
+ |SELECT id FROM full_text_search(
+ | 't',
+ | 'content',
+ | '{"match":{"column":"content","terms":"paimon"}}',
+ | 1)
+ |""".stripMargin)
+ relationTables(df) match {
+ case Seq(SparkTable(_: FullTextSearchTable)) =>
+ case other => fail(s"Unexpected relations: $other")
+ }
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
index 4c0123824f..9176ca6720 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
@@ -37,7 +37,7 @@ import org.apache.spark.sql.catalyst.plans.{Inner, JoinType}
import org.apache.spark.sql.catalyst.plans.logical.{Filter, LeafNode,
LogicalPlan, Project, SubqueryAlias, UnaryNode}
import org.apache.spark.sql.catalyst.util.MapData
import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog}
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.paimon.shims.SparkShimLoader
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.unsafe.types.UTF8String
@@ -146,7 +146,7 @@ object PaimonTableValuedFunctions {
case Some(snapshotIdPair: (Long, Long)) =>
sparkIncrementQuery(spark, sparkTable, sparkCatalog, ident,
options, snapshotIdPair)
case _ =>
- DataSourceV2Relation.create(
+ SparkShimLoader.shim.createDataSourceV2Relation(
sparkTable,
Some(sparkCatalog),
Some(ident),
@@ -168,7 +168,7 @@ object PaimonTableValuedFunctions {
}
val vectorSearch = vsq.createVectorSearch(innerTable, argsWithoutTable)
val vectorSearchTable = VectorSearchTable.create(innerTable,
vectorSearch)
- DataSourceV2Relation.create(
+ SparkShimLoader.shim.createDataSourceV2Relation(
st.copy(table = vectorSearchTable),
Some(sparkCatalog),
Some(ident),
@@ -190,7 +190,7 @@ object PaimonTableValuedFunctions {
case st @ SparkTable(innerTable: InnerTable) =>
val hybridSearch = hsq.createHybridSearch(innerTable, argsWithoutTable)
val hybridSearchTable = HybridSearchTable.create(innerTable,
hybridSearch)
- DataSourceV2Relation.create(
+ SparkShimLoader.shim.createDataSourceV2Relation(
st.copy(table = hybridSearchTable),
Some(sparkCatalog),
Some(ident),
@@ -346,7 +346,7 @@ object PaimonTableValuedFunctions {
case st @ SparkTable(innerTable: InnerTable) =>
val fullTextSearch = ftsq.createFullTextSearch(innerTable,
argsWithoutTable)
val fullTextSearchTable = FullTextSearchTable.create(innerTable,
fullTextSearch)
- DataSourceV2Relation.create(
+ SparkShimLoader.shim.createDataSourceV2Relation(
st.copy(table = fullTextSearchTable),
Some(sparkCatalog),
Some(ident),
@@ -393,7 +393,7 @@ object PaimonTableValuedFunctions {
val updatedOptions = filteredOptions + (CoreOptions.SCAN_VERSION.key()
-> snapshotId.toString)
createDataset(
spark,
- DataSourceV2Relation.create(
+ SparkShimLoader.shim.createDataSourceV2Relation(
sparkTable,
Some(sparkCatalog),
Some(ident),
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
index e4a0a315f2..07b7da7c1e 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
@@ -36,13 +36,14 @@ import
org.apache.spark.sql.catalyst.plans.logical.{Assignment, CTERelationRef,
import org.apache.spark.sql.catalyst.plans.physical.Distribution
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.ArrayData
-import org.apache.spark.sql.connector.catalog.{Column, Identifier,
StagingTableCatalog, Table, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Column,
Identifier, StagingTableCatalog, Table, TableCatalog}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.connector.write.BatchWrite
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
DataSourceV2ScanRelation}
import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import java.util.{Map => JMap}
@@ -196,6 +197,14 @@ trait SparkShim {
table: Table,
output: Seq[AttributeReference]): DataSourceV2Relation
+ // Per-version shim: Spark 4.1 added a 5th `timeTravelSpec` parameter with a
default. A 4-arg call
+ // site compiled against 4.1 emits a `create$default$5()` lookup absent on
4.0.
+ def createDataSourceV2Relation(
+ table: Table,
+ catalog: Option[CatalogPlugin],
+ identifier: Option[Identifier],
+ options: CaseInsensitiveStringMap): DataSourceV2Relation
+
/** Creates an internal scan relation whose constructor changed across Spark
minor versions. */
def createDataSourceV2ScanRelation(
relation: DataSourceV2ScanRelation,
diff --git
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
index 9b9b617416..2e292c4d66 100644
---
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
+++
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
@@ -45,7 +45,7 @@ import
org.apache.spark.sql.catalyst.plans.physical.Distribution
// compile time but must also run on 3.2 / 3.3).
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.{ArrayData, GeneratedColumn,
ResolveDefaultColumns}
-import org.apache.spark.sql.connector.catalog.{Column, Identifier,
StagingTableCatalog, Table, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Column,
Identifier, StagingTableCatalog, Table, TableCatalog}
import
org.apache.spark.sql.connector.catalog.CatalogV2Util.structTypeToV2Columns
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.read.Scan
@@ -57,6 +57,7 @@ import
org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, Data
import org.apache.spark.sql.execution.streaming.{FileStreamSink,
MetadataLogFileIndex}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import java.util.{Map => JMap}
@@ -281,6 +282,13 @@ class Spark3Shim extends SparkShim {
relation.copy(table = table, output = output)
}
+ override def createDataSourceV2Relation(
+ table: Table,
+ catalog: Option[CatalogPlugin],
+ identifier: Option[Identifier],
+ options: CaseInsensitiveStringMap): DataSourceV2Relation =
+ DataSourceV2Relation.create(table, catalog, identifier, options)
+
override def createDataSourceV2ScanRelation(
relation: DataSourceV2ScanRelation,
scan: Scan,
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index ae78e6eb79..65bf9e9797 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -42,7 +42,7 @@ import
org.apache.spark.sql.catalyst.plans.logical.MergeRows.{Copy, Insert, Keep
import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution,
Distribution}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.{ArrayData, GeneratedColumn,
IdentityColumn, ResolveDefaultColumns, STUtils}
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column,
Identifier, StagingTableCatalog, Table, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, CatalogV2Util,
Column, Identifier, StagingTableCatalog, Table, TableCatalog}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.connector.write.BatchWrite
@@ -54,6 +54,7 @@ import
org.apache.spark.sql.execution.streaming.runtime.MetadataLogFileIndex
import org.apache.spark.sql.execution.streaming.sinks.FileStreamSink
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataTypes, Geography, GeographyType,
Geometry, GeometryType, StructType, VariantType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.unsafe.types.VariantVal
import java.util.{Map => JMap}
@@ -285,6 +286,13 @@ class Spark4Shim extends SparkShim {
relation.copy(table = table, output = output)
}
+ override def createDataSourceV2Relation(
+ table: Table,
+ catalog: Option[CatalogPlugin],
+ identifier: Option[Identifier],
+ options: CaseInsensitiveStringMap): DataSourceV2Relation =
+ DataSourceV2Relation.create(table, catalog, identifier, options)
+
override def createDataSourceV2ScanRelation(
relation: DataSourceV2ScanRelation,
scan: Scan,
