diff --git a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala index 4aa8fb784089..37bf307b024a 100644 --- a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala +++ b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala @@ -20,7 +20,6 @@ package org.apache.spark.sql.execution.shim import org.apache.paimon.Snapshot import org.apache.paimon.spark.SparkCatalog -import org.apache.paimon.spark.catalog.FormatTableCatalog import org.apache.paimon.spark.write.PaimonWriteOptions import org.apache.spark.sql.{SparkSession, Strategy} @@ -48,18 +47,11 @@ case class PaimonCreateTableAsSelectStrategy(spark: SparkSession) extends Strate splitTableAndWriteOptions(options) val newProps = CatalogV2Util.withDefaultOwnership(props) ++ tableOptions - val isPartitionedFormatTable = { - catalog match { - case formatCatalog: FormatTableCatalog => - formatCatalog.isFormatTable(newProps.get("provider").orNull) && parts.nonEmpty - case _ => false - } - } - - if (isPartitionedFormatTable) { - throw new UnsupportedOperationException( - "Using CTAS with partitioned format table is not supported yet.") - } + catalog.checkPartitionedFormatTableCtas( + ident, + newProps.get("provider").orNull, + parts.nonEmpty, + newProps.asJava) CreateTableAsSelectExec( catalog, diff --git a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala index eb3e0444597e..1f5bcb0209cc 100644 --- a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala +++ b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala @@ -20,7 +20,6 @@ package org.apache.spark.sql.execution.shim import org.apache.paimon.Snapshot import org.apache.paimon.spark.SparkCatalog -import org.apache.paimon.spark.catalog.FormatTableCatalog import org.apache.paimon.spark.write.PaimonWriteOptions import org.apache.spark.sql.{SparkSession, Strategy} @@ -51,18 +50,11 @@ case class PaimonCreateTableAsSelectStrategy(spark: SparkSession) splitTableAndWriteOptions(options) val qualifiedSpec = qualifyTableSpec(tableSpec, tableOptions) - val isPartitionedFormatTable = { - catalog match { - case formatCatalog: FormatTableCatalog => - formatCatalog.isFormatTable(qualifiedSpec.provider.orNull) && parts.nonEmpty - case _ => false - } - } - - if (isPartitionedFormatTable) { - throw new UnsupportedOperationException( - "Using CTAS with partitioned format table is not supported yet.") - } + catalog.checkPartitionedFormatTableCtas( + ident.asIdentifier, + qualifiedSpec.provider.orNull, + parts.nonEmpty, + qualifiedSpec.properties.asJava) CreateTableAsSelectExec( catalog.asTableCatalog, diff --git a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala index 0e0f3037d219..a046d8431415 100644 --- a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala +++ b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala @@ -20,7 +20,6 @@ package org.apache.spark.sql.execution.shim import org.apache.paimon.Snapshot import org.apache.paimon.spark.SparkCatalog -import org.apache.paimon.spark.catalog.FormatTableCatalog import org.apache.paimon.spark.write.PaimonWriteOptions import org.apache.spark.sql.{SparkSession, Strategy} @@ -53,18 +52,11 @@ case class PaimonCreateTableAsSelectStrategy(spark: SparkSession) splitTableAndWriteOptions(options) val qualifiedSpec = qualifyTableSpec(tableSpec, tableOptions) - val isPartitionedFormatTable = { - catalog match { - case formatCatalog: FormatTableCatalog => - formatCatalog.isFormatTable(qualifiedSpec.provider.orNull) && parts.nonEmpty - case _ => false - } - } - - if (isPartitionedFormatTable) { - throw new UnsupportedOperationException( - "Using CTAS with partitioned format table is not supported yet.") - } + catalog.checkPartitionedFormatTableCtas( + ident, + qualifiedSpec.provider.orNull, + parts.nonEmpty, + qualifiedSpec.properties.asJava) CreateTableAsSelectExec( catalog.asTableCatalog, diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java index 16fc1806d1e2..34d9af6f3d74 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java @@ -102,6 +102,7 @@ import static org.apache.paimon.spark.SparkTypeUtils.toPaimonType; import static org.apache.paimon.spark.util.OptionUtils.checkRequiredConfigurations; import static org.apache.paimon.spark.util.OptionUtils.copyWithSQLConf; +import static org.apache.paimon.spark.util.OptionUtils.usePaimonFormatTableImplementation; import static org.apache.paimon.spark.util.OptionUtils.withBranchFromOptions; import static org.apache.paimon.spark.utils.CatalogUtils.checkNamespace; import static org.apache.paimon.spark.utils.CatalogUtils.checkNoDefaultValue; @@ -394,6 +395,24 @@ public boolean dropTable(Identifier ident) { } } + public void checkPartitionedFormatTableCtas( + Identifier ident, + @Nullable String provider, + boolean partitioned, + Map properties) { + if (partitioned + && isFormatTable(provider) + && !usePaimonFormatTableImplementation( + catalogName, + toIdentifier(ident, catalogName), + catalog.options(), + properties)) { + throw new UnsupportedOperationException( + "Using CTAS with a partitioned engine format table is not supported. " + + "Set 'format-table.implementation' to 'paimon'."); + } + } + @Override public StagedTable stageCreate( Identifier ident, diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala index 1649a57eadb6..c357256b7cc0 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala @@ -19,7 +19,7 @@ package org.apache.paimon.spark.util import org.apache.paimon.CoreOptions -import org.apache.paimon.catalog.Identifier +import org.apache.paimon.catalog.{CatalogUtils, Identifier} import org.apache.paimon.options.ConfigOption import org.apache.paimon.spark.{SparkCatalogOptions, SparkConnectorOptions} import org.apache.paimon.table.Table @@ -213,6 +213,18 @@ object OptionUtils extends SQLConfHelper with Logging { } } + def usePaimonFormatTableImplementation( + catalogName: String, + ident: Identifier, + catalogOptions: JMap[String, String], + tableOptions: JMap[String, String]): Boolean = { + val mergedOptions = + new JHashMap[String, String](CatalogUtils.tableDefaultOptions(catalogOptions)) + mergedOptions.putAll(tableOptions) + mergedOptions.putAll(getMergedOptions(catalogName, ident)) + new CoreOptions(mergedOptions).formatTableImplementationIsPaimon + } + def withBranchFromOptions( catalogName: String = null, identifier: Identifier = null, diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala index 1a8e3ffe4b75..ed6193bee1c9 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala @@ -20,7 +20,6 @@ package org.apache.spark.sql.execution.shim import org.apache.paimon.Snapshot import org.apache.paimon.spark.SparkCatalog -import org.apache.paimon.spark.catalog.FormatTableCatalog import org.apache.paimon.spark.write.PaimonWriteOptions import org.apache.spark.sql.SparkSession @@ -30,6 +29,8 @@ import org.apache.spark.sql.execution.{PaimonTableAsSelectHelper, SparkPlan, Spa import org.apache.spark.sql.execution.PaimonTableAsSelectHelper._ import org.apache.spark.sql.execution.datasources.v2.CreateTableAsSelectExec +import scala.collection.JavaConverters._ + case class PaimonCreateTableAsSelectStrategy(spark: SparkSession) extends SparkStrategy with PaimonTableAsSelectHelper { @@ -49,18 +50,11 @@ case class PaimonCreateTableAsSelectStrategy(spark: SparkSession) splitTableAndWriteOptions(options) val qualifiedSpec = qualifyTableSpec(tableSpec, tableOptions) - val isPartitionedFormatTable = { - catalog match { - case formatCatalog: FormatTableCatalog => - formatCatalog.isFormatTable(qualifiedSpec.provider.orNull) && parts.nonEmpty - case _ => false - } - } - - if (isPartitionedFormatTable) { - throw new UnsupportedOperationException( - "Using CTAS with partitioned format table is not supported yet.") - } + catalog.checkPartitionedFormatTableCtas( + ident, + qualifiedSpec.provider.orNull, + parts.nonEmpty, + qualifiedSpec.properties.asJava) CreateTableAsSelectExec( catalog.asTableCatalog, diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/OptionUtilsTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/OptionUtilsTest.scala index c766a281b952..018c49727216 100644 --- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/OptionUtilsTest.scala +++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/OptionUtilsTest.scala @@ -99,6 +99,34 @@ class OptionUtilsTest extends AnyFunSuite { assert(exception.getMessage.contains(METASTORE_PARTITIONED_TABLE.key())) } + test("resolve format table implementation option precedence") { + val ident = Identifier.create("test_db", "format_table") + val catalogOptions = + Map(s"table-default.${FORMAT_TABLE_IMPLEMENTATION.key()}" -> "engine").asJava + + assert( + !OptionUtils.usePaimonFormatTableImplementation( + "test_catalog", + ident, + catalogOptions, + Collections.emptyMap())) + assert( + OptionUtils.usePaimonFormatTableImplementation( + "test_catalog", + ident, + catalogOptions, + Map(FORMAT_TABLE_IMPLEMENTATION.key() -> "paimon").asJava)) + + SQLConf.withExistingConf(engineSQLConf) { + assert( + !OptionUtils.usePaimonFormatTableImplementation( + "test_catalog", + ident, + Collections.emptyMap(), + Map(FORMAT_TABLE_IMPLEMENTATION.key() -> "paimon").asJava)) + } + } + private def engineSQLConf: SQLConf = { val sqlConf = new SQLConf sqlConf.setConfString( diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala index 998c7c591cad..a7e3bdce5e0f 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala @@ -197,18 +197,54 @@ abstract class FormatTableTestBase extends PaimonHiveTestBase with AdaptiveSpark test("Format table: CTAS with partitioned table") { withTable("t1", "t2") { - sql("CREATE TABLE t1 (id INT, p1 INT, p2 INT) USING csv PARTITIONED BY (p1, p2)") - sql("INSERT INTO t1 VALUES (1, 2, 3)") + sql("CREATE TABLE t1 (id INT, p1 INT, p2 INT) USING csv") + sql("INSERT INTO t1 VALUES (1, 2, 3), (2, 2, 4), (3, 5, 6)") - assertThrows[UnsupportedOperationException] { - sql(""" - |CREATE TABLE t2 - |USING csv - |PARTITIONED BY (p1, p2) - |AS SELECT * FROM t1 - |""".stripMargin) + sql(""" + |CREATE TABLE t2 + |USING parquet + |PARTITIONED BY (p1, p2) + |AS SELECT * FROM t1 + |""".stripMargin) + + checkAnswer( + sql("SELECT * FROM t2 ORDER BY id"), + Seq(Row(1, 2, 3), Row(2, 2, 4), Row(3, 5, 6))) + checkAnswer( + sql("SHOW PARTITIONS t2"), + Seq(Row("p1=2/p2=3"), Row("p1=2/p2=4"), Row("p1=5/p2=6"))) + + val filtered = sql("SELECT * FROM t2 WHERE p1 = 2 AND p2 = 4") + checkAnswer(filtered, Seq(Row(2, 2, 4))) + assert(collectFilteredInputSplits(filtered.queryExecution.executedPlan, "t2").size == 1) + } + } + + test("Format table: CTAS with partitioned engine table") { + def checkRejected(tableProperties: String): Unit = { + withTable("t1", "t2") { + sql("CREATE TABLE t1 (id INT, p1 INT, p2 INT) USING csv") + sql("INSERT INTO t1 VALUES (1, 2, 3)") + + val exception = intercept[UnsupportedOperationException] { + sql(s""" + |CREATE TABLE t2 + |USING parquet + |PARTITIONED BY (p1, p2) + |$tableProperties + |AS SELECT * FROM t1 + |""".stripMargin) + } + assert(exception.getMessage.contains("partitioned engine format table")) + assert(!spark.catalog.tableExists("t2")) } } + + checkRejected("TBLPROPERTIES ('format-table.implementation'='engine')") + withSparkSQLConf("spark.paimon.format-table.implementation" -> "engine") { + checkRejected("") + checkRejected("TBLPROPERTIES ('format-table.implementation'='paimon')") + } } test("Format table: create or replace as select supports table type change") {