This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new c9ef52026 [spark] Solve the sql.extensions compatibility with spark
3.1 (#2502)
c9ef52026 is described below
commit c9ef52026c366b90c539d01a3da909ea3900b154
Author: Yann Byron <[email protected]>
AuthorDate: Wed Dec 13 18:55:59 2023 +0800
[spark] Solve the sql.extensions compatibility with spark 3.1 (#2502)
---
.../analysis/PaimonIncompatiblePHRRules.scala} | 22 ++++-------
.../PaimonIncompatibleResolutionRules.scala} | 22 ++++-------
.../spark/catalyst/trees/PaimonLeafLike.scala} | 22 +++--------
.../apache/paimon/spark/leafnode/package.scala} | 27 +++++++------
.../paimon/spark/SparkGenericCatalogTest.java | 4 ++
.../paimon/spark/SparkGenericCatalogTest.java | 4 ++
.../spark/catalyst/analysis/PaimonAnalysis.scala | 17 ---------
...ysis.scala => PaimonIncompatiblePHRRules.scala} | 44 ++--------------------
.../PaimonIncompatibleResolutionRules.scala} | 25 ++++++------
.../catalyst/plans/logical/PaimonCallCommand.scala | 5 ++-
.../plans/logical/PaimonCallStatement.scala | 5 ++-
.../commands/DeleteFromPaimonTableCommand.scala | 6 +--
.../spark/commands/MergeIntoPaimonTable.scala | 4 +-
.../commands/PaimonTruncateTableCommand.scala | 4 +-
.../spark/commands/UpdatePaimonTableCommand.scala | 4 +-
.../paimon/spark/execution/PaimonCallExec.scala | 4 +-
.../extensions/PaimonSparkSessionExtensions.scala | 6 ++-
.../package.scala} | 25 ++++++------
18 files changed, 89 insertions(+), 161 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatiblePHRRules.scala
similarity index 54%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
copy to
paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatiblePHRRules.scala
index 31b3e6777..86b38225f 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
+++
b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatiblePHRRules.scala
@@ -15,23 +15,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark.catalyst.plans.logical
+package org.apache.paimon.spark.catalyst.analysis
-import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.LeafParsedStatement
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
-/** A CALL statement parsed from SQL. */
-case class PaimonCallStatement(name: Seq[String], args:
Seq[PaimonCallArgument])
- extends LeafParsedStatement
+/** These post-hoc resolution rules are incompatible between different
versions of spark. */
+case class PaimonIncompatiblePHRRules(session: SparkSession) extends
Rule[LogicalPlan] {
-/** An argument of a CALL statement. */
-sealed trait PaimonCallArgument {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan
- def expr: Expression
}
-
-/** An argument identified by name. */
-case class PaimonNamedArgument(name: String, expr: Expression) extends
PaimonCallArgument
-
-/** An argument identified by position. */
-case class PaimonPositionalArgument(expr: Expression) extends
PaimonCallArgument
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatibleResolutionRules.scala
similarity index 54%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
copy to
paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatibleResolutionRules.scala
index 31b3e6777..11e511a3a 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
+++
b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatibleResolutionRules.scala
@@ -15,23 +15,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark.catalyst.plans.logical
+package org.apache.paimon.spark.catalyst.analysis
-import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.LeafParsedStatement
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
-/** A CALL statement parsed from SQL. */
-case class PaimonCallStatement(name: Seq[String], args:
Seq[PaimonCallArgument])
- extends LeafParsedStatement
+/** These resolution rules are incompatible between different versions of
spark. */
+case class PaimonIncompatibleResolutionRules(session: SparkSession) extends
Rule[LogicalPlan] {
-/** An argument of a CALL statement. */
-sealed trait PaimonCallArgument {
+ override def apply(plan: LogicalPlan): LogicalPlan = plan
- def expr: Expression
}
-
-/** An argument identified by name. */
-case class PaimonNamedArgument(name: String, expr: Expression) extends
PaimonCallArgument
-
-/** An argument identified by position. */
-case class PaimonPositionalArgument(expr: Expression) extends
PaimonCallArgument
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/catalyst/trees/PaimonLeafLike.scala
similarity index 53%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
copy to
paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/catalyst/trees/PaimonLeafLike.scala
index 31b3e6777..6e5d56988 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
+++
b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/catalyst/trees/PaimonLeafLike.scala
@@ -15,23 +15,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark.catalyst.plans.logical
+package org.apache.paimon.spark.catalyst.trees
-import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.LeafParsedStatement
+import org.apache.spark.sql.catalyst.trees.TreeNode
-/** A CALL statement parsed from SQL. */
-case class PaimonCallStatement(name: Seq[String], args:
Seq[PaimonCallArgument])
- extends LeafParsedStatement
+trait PaimonLeafLike[T <: TreeNode[T]] {
+ self: TreeNode[T] =>
+ final override def children: Seq[T] = Nil
-/** An argument of a CALL statement. */
-sealed trait PaimonCallArgument {
-
- def expr: Expression
+ final override def mapChildren(f: T => T): T = this.asInstanceOf[T]
}
-
-/** An argument identified by name. */
-case class PaimonNamedArgument(name: String, expr: Expression) extends
PaimonCallArgument
-
-/** An argument identified by position. */
-case class PaimonPositionalArgument(expr: Expression) extends
PaimonCallArgument
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/leafnode/package.scala
similarity index 52%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
copy to
paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/leafnode/package.scala
index 31b3e6777..6b987b5f8 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
+++
b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/leafnode/package.scala
@@ -15,23 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark.catalyst.plans.logical
+package org.apache.paimon.spark
-import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.LeafParsedStatement
+import org.apache.paimon.spark.catalyst.trees.PaimonLeafLike
-/** A CALL statement parsed from SQL. */
-case class PaimonCallStatement(name: Seq[String], args:
Seq[PaimonCallArgument])
- extends LeafParsedStatement
+import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan,
ParsedStatement}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.datasources.v2.V2CommandExec
-/** An argument of a CALL statement. */
-sealed trait PaimonCallArgument {
+package object leafnode {
+ trait PaimonLeafParsedStatement extends ParsedStatement with
PaimonLeafLike[LogicalPlan]
- def expr: Expression
-}
+ trait PaimonLeafRunnableCommand extends RunnableCommand with
PaimonLeafLike[LogicalPlan]
+
+ trait PaimonLeafCommand extends Command with PaimonLeafLike[LogicalPlan]
-/** An argument identified by name. */
-case class PaimonNamedArgument(name: String, expr: Expression) extends
PaimonCallArgument
+ trait PaimonLeafV2CommandExec extends V2CommandExec with
PaimonLeafLike[SparkPlan]
-/** An argument identified by position. */
-case class PaimonPositionalArgument(expr: Expression) extends
PaimonCallArgument
+}
diff --git
a/paimon-spark/paimon-spark-3.1/src/test/java/org/apache/paimon/spark/SparkGenericCatalogTest.java
b/paimon-spark/paimon-spark-3.1/src/test/java/org/apache/paimon/spark/SparkGenericCatalogTest.java
index f632998c5..c04837178 100644
---
a/paimon-spark/paimon-spark-3.1/src/test/java/org/apache/paimon/spark/SparkGenericCatalogTest.java
+++
b/paimon-spark/paimon-spark-3.1/src/test/java/org/apache/paimon/spark/SparkGenericCatalogTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.sink.BatchTableCommit;
@@ -52,6 +53,9 @@ public class SparkGenericCatalogTest {
spark =
SparkSession.builder()
.config("spark.sql.warehouse.dir",
warehousePath.toString())
+ .config(
+ "spark.sql.extensions",
+ PaimonSparkSessionExtensions.class.getName())
.master("local[2]")
.getOrCreate();
spark.conf().set("spark.sql.catalog.spark_catalog",
SparkGenericCatalog.class.getName());
diff --git
a/paimon-spark/paimon-spark-3.2/src/test/java/org/apache/paimon/spark/SparkGenericCatalogTest.java
b/paimon-spark/paimon-spark-3.2/src/test/java/org/apache/paimon/spark/SparkGenericCatalogTest.java
index f632998c5..c04837178 100644
---
a/paimon-spark/paimon-spark-3.2/src/test/java/org/apache/paimon/spark/SparkGenericCatalogTest.java
+++
b/paimon-spark/paimon-spark-3.2/src/test/java/org/apache/paimon/spark/SparkGenericCatalogTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.sink.BatchTableCommit;
@@ -52,6 +53,9 @@ public class SparkGenericCatalogTest {
spark =
SparkSession.builder()
.config("spark.sql.warehouse.dir",
warehousePath.toString())
+ .config(
+ "spark.sql.extensions",
+ PaimonSparkSessionExtensions.class.getName())
.master("local[2]")
.getOrCreate();
spark.conf().set("spark.sql.catalog.spark_catalog",
SparkGenericCatalog.class.getName());
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
index a1cd1a8e0..472e80694 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
@@ -33,9 +33,6 @@ class PaimonAnalysis(session: SparkSession) extends
Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsDown {
- case func: PaimonTableValueFunction if func.args.forall(_.resolved) =>
- PaimonTableValuedFunctions.resolvePaimonTableValuedFunction(session,
func)
-
case o @ PaimonDynamicPartitionOverwrite(r, d) if o.resolved =>
PaimonDynamicPartitionOverwriteCommand(r, d, o.query, o.writeOptions,
o.isByName)
@@ -52,20 +49,6 @@ case class PaimonPostHocResolutionRules(session:
SparkSession) extends Rule[Logi
case t @ TruncateTable(PaimonRelation(table)) if t.resolved =>
PaimonTruncateTableCommand(table, Map.empty)
- case t @ TruncatePartition(PaimonRelation(table),
ResolvedPartitionSpec(names, ident, _))
- if t.resolved =>
- assert(names.length == ident.numFields, "Names and values of partition
don't match")
- val resolver = session.sessionState.conf.resolver
- val schema = table.schema()
- val partitionSpec = names.zipWithIndex.map {
- case (name, index) =>
- val field = schema.find(f => resolver(f.name, name)).getOrElse {
- throw new RuntimeException(s"$name is not a valid partition
column in $schema.")
- }
- (name -> ident.get(index, field.dataType).toString)
- }.toMap
- PaimonTruncateTableCommand(table, partitionSpec)
-
case _ => plan
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatiblePHRRules.scala
similarity index 52%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
copy to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatiblePHRRules.scala
index a1cd1a8e0..43ea28084 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatiblePHRRules.scala
@@ -17,41 +17,18 @@
*/
package org.apache.paimon.spark.catalyst.analysis
-import org.apache.paimon.spark.SparkTable
-import
org.apache.paimon.spark.catalyst.plans.logical.{PaimonTableValuedFunctions,
PaimonTableValueFunction}
-import
org.apache.paimon.spark.commands.{PaimonDynamicPartitionOverwriteCommand,
PaimonTruncateTableCommand}
-import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.spark.commands.PaimonTruncateTableCommand
-import PaimonRelation.isPaimonTable
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.ResolvedPartitionSpec
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan,
MergeIntoTable, OverwritePartitionsDynamic, TruncatePartition, TruncateTable}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan,
TruncatePartition}
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
-class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
-
- override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsDown {
-
- case func: PaimonTableValueFunction if func.args.forall(_.resolved) =>
- PaimonTableValuedFunctions.resolvePaimonTableValuedFunction(session,
func)
-
- case o @ PaimonDynamicPartitionOverwrite(r, d) if o.resolved =>
- PaimonDynamicPartitionOverwriteCommand(r, d, o.query, o.writeOptions,
o.isByName)
-
- case merge: MergeIntoTable if isPaimonTable(merge.targetTable) &&
merge.childrenResolved =>
- PaimonMergeIntoResolver(merge, session)
- }
-
-}
-
-case class PaimonPostHocResolutionRules(session: SparkSession) extends
Rule[LogicalPlan] {
+/** These post-hoc resolution rules are incompatible between different
versions of spark. */
+case class PaimonIncompatiblePHRRules(session: SparkSession) extends
Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = {
plan match {
- case t @ TruncateTable(PaimonRelation(table)) if t.resolved =>
- PaimonTruncateTableCommand(table, Map.empty)
-
case t @ TruncatePartition(PaimonRelation(table),
ResolvedPartitionSpec(names, ident, _))
if t.resolved =>
assert(names.length == ident.numFields, "Names and values of partition
don't match")
@@ -69,18 +46,5 @@ case class PaimonPostHocResolutionRules(session:
SparkSession) extends Rule[Logi
case _ => plan
}
}
-}
-object PaimonDynamicPartitionOverwrite {
- def unapply(o: OverwritePartitionsDynamic): Option[(DataSourceV2Relation,
FileStoreTable)] = {
- if (o.query.resolved) {
- o.table match {
- case r: DataSourceV2Relation if r.table.isInstanceOf[SparkTable] =>
- Some((r,
r.table.asInstanceOf[SparkTable].getTable.asInstanceOf[FileStoreTable]))
- case _ => None
- }
- } else {
- None
- }
- }
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatibleResolutionRules.scala
similarity index 52%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
copy to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatibleResolutionRules.scala
index 0f1dd4eef..0edabfc15 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonIncompatibleResolutionRules.scala
@@ -15,23 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark.execution
+package org.apache.paimon.spark.catalyst.analysis
-import org.apache.paimon.spark.procedure.Procedure
+import
org.apache.paimon.spark.catalyst.plans.logical.{PaimonTableValuedFunctions,
PaimonTableValueFunction}
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.util.truncatedString
-import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
-case class PaimonCallExec(output: Seq[Attribute], procedure: Procedure, input:
InternalRow)
- extends LeafV2CommandExec {
+/** These resolution rules are incompatible between different versions of
spark. */
+case class PaimonIncompatibleResolutionRules(session: SparkSession) extends
Rule[LogicalPlan] {
- override protected def run(): Seq[InternalRow] = {
- procedure.call(input)
- }
+ override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsDown {
+
+ case func: PaimonTableValueFunction if func.args.forall(_.resolved) =>
+ PaimonTableValuedFunctions.resolvePaimonTableValuedFunction(session,
func)
- override def simpleString(maxFields: Int): String = {
- s"CallExec${truncatedString(output, "[", ", ", "]", maxFields)}
${procedure.description}"
}
+
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallCommand.scala
index 6b1ce53f6..91ea90dcd 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallCommand.scala
@@ -17,14 +17,15 @@
*/
package org.apache.paimon.spark.catalyst.plans.logical
+import org.apache.paimon.spark.leafnode.PaimonLeafCommand
import org.apache.paimon.spark.procedure.Procedure
import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, Expression}
-import org.apache.spark.sql.catalyst.plans.logical.LeafCommand
import org.apache.spark.sql.catalyst.util.truncatedString
/** A CALL command that resolves stored procedure from SQL. */
-case class PaimonCallCommand(procedure: Procedure, args: Seq[Expression])
extends LeafCommand {
+case class PaimonCallCommand(procedure: Procedure, args: Seq[Expression])
+ extends PaimonLeafCommand {
override lazy val output: Seq[Attribute] =
procedure.outputType.map(
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
index 31b3e6777..49de3fca1 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonCallStatement.scala
@@ -17,12 +17,13 @@
*/
package org.apache.paimon.spark.catalyst.plans.logical
+import org.apache.paimon.spark.leafnode.PaimonLeafParsedStatement
+
import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.LeafParsedStatement
/** A CALL statement parsed from SQL. */
case class PaimonCallStatement(name: Seq[String], args:
Seq[PaimonCallArgument])
- extends LeafParsedStatement
+ extends PaimonLeafParsedStatement
/** An argument of a CALL statement. */
sealed trait PaimonCallArgument {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
index fa1a46333..5504f319b 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
@@ -20,6 +20,7 @@ package org.apache.paimon.spark.commands
import org.apache.paimon.options.Options
import org.apache.paimon.predicate.OnlyPartitionKeyEqualVisitor
import org.apache.paimon.spark.{InsertInto, SparkTable}
+import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
import org.apache.paimon.spark.schema.SparkSystemColumns.ROW_KIND_COL
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.sink.BatchWriteBuilder
@@ -28,8 +29,7 @@ import org.apache.paimon.types.RowKind
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.Utils.createDataset
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
-import org.apache.spark.sql.catalyst.plans.logical.{DeleteFromTable, Filter,
LogicalPlan, SupportsSubquery}
-import org.apache.spark.sql.execution.command.LeafRunnableCommand
+import org.apache.spark.sql.catalyst.plans.logical.{DeleteFromTable, Filter}
import org.apache.spark.sql.functions.lit
import java.util.{Collections, UUID}
@@ -37,7 +37,7 @@ import java.util.{Collections, UUID}
import scala.util.control.NonFatal
case class DeleteFromPaimonTableCommand(v2Table: SparkTable, delete:
DeleteFromTable)
- extends LeafRunnableCommand
+ extends PaimonLeafRunnableCommand
with PaimonCommand {
override def table: FileStoreTable =
v2Table.getTable.asInstanceOf[FileStoreTable]
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
index 95caadaa1..4d3f43152 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
@@ -20,6 +20,7 @@ package org.apache.paimon.spark.commands
import org.apache.paimon.options.Options
import org.apache.paimon.spark.{InsertInto, SparkTable}
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
+import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
import org.apache.paimon.spark.schema.SparkSystemColumns
import org.apache.paimon.spark.util.EncoderUtils
import org.apache.paimon.table.FileStoreTable
@@ -33,7 +34,6 @@ import org.apache.spark.sql.catalyst.expressions.{Alias,
Attribute, BasePredicat
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
import org.apache.spark.sql.catalyst.expressions.codegen.GeneratePredicate
import org.apache.spark.sql.catalyst.plans.logical.{DeleteAction, Filter,
InsertAction, LogicalPlan, MergeAction, UpdateAction}
-import org.apache.spark.sql.execution.command.LeafRunnableCommand
import org.apache.spark.sql.functions.{col, lit, monotonically_increasing_id,
sum}
import org.apache.spark.sql.types.{ByteType, StructField, StructType}
@@ -45,7 +45,7 @@ case class MergeIntoPaimonTable(
mergeCondition: Expression,
matchedActions: Seq[MergeAction],
notMatchedActions: Seq[MergeAction])
- extends LeafRunnableCommand
+ extends PaimonLeafRunnableCommand
with WithFileStoreTable
with ExpressionHelper
with PredicateHelper {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonTruncateTableCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonTruncateTableCommand.scala
index 132d2da65..051a8fe6f 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonTruncateTableCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonTruncateTableCommand.scala
@@ -18,19 +18,19 @@
package org.apache.paimon.spark.commands
import org.apache.paimon.spark.SparkTable
+import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.sink.BatchWriteBuilder
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
-import org.apache.spark.sql.execution.command.LeafRunnableCommand
import java.util.{Collections, UUID}
import scala.collection.JavaConverters._
case class PaimonTruncateTableCommand(v2Table: SparkTable, partitionSpec:
TablePartitionSpec)
- extends LeafRunnableCommand
+ extends PaimonLeafRunnableCommand
with PaimonCommand {
override def table: FileStoreTable =
v2Table.getTable.asInstanceOf[FileStoreTable]
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
index 1b7de80be..f14c61f64 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
@@ -20,6 +20,7 @@ package org.apache.paimon.spark.commands
import org.apache.paimon.options.Options
import org.apache.paimon.spark.{InsertInto, SparkTable}
import org.apache.paimon.spark.catalyst.analysis.{AssignmentAlignmentHelper,
PaimonRelation}
+import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
import org.apache.paimon.spark.schema.SparkSystemColumns.ROW_KIND_COL
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.types.RowKind
@@ -29,11 +30,10 @@ import org.apache.spark.sql.Utils.createDataset
import org.apache.spark.sql.catalyst.expressions.Alias
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
import org.apache.spark.sql.catalyst.plans.logical.{Filter, Project,
UpdateTable}
-import org.apache.spark.sql.execution.command.LeafRunnableCommand
import org.apache.spark.sql.functions.lit
case class UpdatePaimonTableCommand(u: UpdateTable)
- extends LeafRunnableCommand
+ extends PaimonLeafRunnableCommand
with AssignmentAlignmentHelper {
override def run(sparkSession: SparkSession): Seq[Row] = {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
index 0f1dd4eef..63e8ca593 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
@@ -17,15 +17,15 @@
*/
package org.apache.paimon.spark.execution
+import org.apache.paimon.spark.leafnode.PaimonLeafV2CommandExec
import org.apache.paimon.spark.procedure.Procedure
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.util.truncatedString
-import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec
case class PaimonCallExec(output: Seq[Attribute], procedure: Procedure, input:
InternalRow)
- extends LeafV2CommandExec {
+ extends PaimonLeafV2CommandExec {
override protected def run(): Seq[InternalRow] = {
procedure.call(input)
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 c4ec03ce5..1760ba358 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
@@ -17,7 +17,7 @@
*/
package org.apache.paimon.spark.extensions
-import org.apache.paimon.spark.catalyst.analysis.{PaimonAnalysis,
PaimonDeleteTable, PaimonMergeInto, PaimonPostHocResolutionRules,
PaimonProcedureResolver, PaimonUpdateTable}
+import org.apache.paimon.spark.catalyst.analysis.{PaimonAnalysis,
PaimonDeleteTable, PaimonIncompatiblePHRRules,
PaimonIncompatibleResolutionRules, PaimonMergeInto,
PaimonPostHocResolutionRules, PaimonProcedureResolver, PaimonUpdateTable}
import
org.apache.paimon.spark.catalyst.plans.logical.PaimonTableValuedFunctions
import org.apache.paimon.spark.execution.PaimonStrategy
@@ -32,10 +32,12 @@ class PaimonSparkSessionExtensions extends
(SparkSessionExtensions => Unit) {
extensions.injectParser { case (_, parser) => new
PaimonSparkSqlExtensionsParser(parser) }
// analyzer extensions
- extensions.injectResolutionRule(sparkSession => new
PaimonAnalysis(sparkSession))
+ extensions.injectResolutionRule(spark => new PaimonAnalysis(spark))
extensions.injectResolutionRule(spark => PaimonProcedureResolver(spark))
+ extensions.injectResolutionRule(spark =>
PaimonIncompatibleResolutionRules(spark))
extensions.injectPostHocResolutionRule(spark =>
PaimonPostHocResolutionRules(spark))
+ extensions.injectPostHocResolutionRule(spark =>
PaimonIncompatiblePHRRules(spark))
extensions.injectPostHocResolutionRule(_ => PaimonUpdateTable)
extensions.injectPostHocResolutionRule(_ => PaimonDeleteTable)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/leafnode/package.scala
similarity index 58%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
copy to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/leafnode/package.scala
index 0f1dd4eef..d0e2e5682 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonCallExec.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/leafnode/package.scala
@@ -15,23 +15,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.paimon.spark.execution
+package org.apache.paimon.spark
-import org.apache.paimon.spark.procedure.Procedure
-
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.util.truncatedString
+import org.apache.spark.sql.catalyst.plans.logical.{LeafCommand,
LeafParsedStatement}
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec
-case class PaimonCallExec(output: Seq[Attribute], procedure: Procedure, input:
InternalRow)
- extends LeafV2CommandExec {
+package object leafnode {
+
+ trait PaimonLeafParsedStatement extends LeafParsedStatement
+
+ trait PaimonLeafRunnableCommand extends LeafRunnableCommand
+
+ trait PaimonLeafCommand extends LeafCommand
- override protected def run(): Seq[InternalRow] = {
- procedure.call(input)
- }
+ trait PaimonLeafV2CommandExec extends LeafV2CommandExec
- override def simpleString(maxFields: Int): String = {
- s"CallExec${truncatedString(output, "[", ", ", "]", maxFields)}
${procedure.description}"
- }
}
