You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by bi...@apache.org on 2024/01/16 07:21:13 UTC
(incubator-paimon) branch release-0.6 updated (5534b45bf -> ec220f5a6)
This is an automated email from the ASF dual-hosted git repository.
biyan pushed a change to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
discard 5534b45bf [spark] Fix call procedure compatibility with spark3.2 (#2707)
discard 5c04b45a8 [spark] Supports CTAS for different versions (#2591)
new ca245748c [spark] Supports CTAS for different versions (#2591)
new ec220f5a6 [spark] Fix call procedure compatibility with spark3.2 (#2707)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (5534b45bf)
\
N -- N -- N refs/heads/release-0.6 (ec220f5a6)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../src/test/scala/org/apache/spark/paimon/Utils.scala | 0
1 file changed, 0 insertions(+), 0 deletions(-)
copy paimon-spark/{paimon-spark-3.3 => paimon-spark-3.2}/src/test/scala/org/apache/spark/paimon/Utils.scala (100%)
(incubator-paimon) 02/02: [spark] Fix call procedure compatibility with spark3.2 (#2707)
Posted by bi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
biyan pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit ec220f5a6410e286e546193de3699a8de764e61a
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Tue Jan 16 14:06:09 2024 +0800
[spark] Fix call procedure compatibility with spark3.2 (#2707)
---
.../paimon/spark/sql/CompactProcedureTest.scala | 22 +++++++++++++++++++++
.../extensions/PaimonSqlExtensionsAstBuilder.scala | 23 +++++++++++++++++++++-
...reTest.scala => CompactProcedureTestBase.scala} | 2 +-
3 files changed, 45 insertions(+), 2 deletions(-)
diff --git a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/CompactProcedureTest.scala b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/CompactProcedureTest.scala
new file mode 100644
index 000000000..51931ef7e
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/CompactProcedureTest.scala
@@ -0,0 +1,22 @@
+/*
+ * 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.procedure.CompactProcedureTestBase
+
+class CompactProcedureTest extends CompactProcedureTestBase {}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSqlExtensionsAstBuilder.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSqlExtensionsAstBuilder.scala
index 8caab60c2..bb8fd9a33 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSqlExtensionsAstBuilder.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/PaimonSqlExtensionsAstBuilder.scala
@@ -23,13 +23,14 @@ import org.apache.paimon.spark.catalyst.plans.logical.{PaimonCallArgument, Paimo
import org.antlr.v4.runtime._
import org.antlr.v4.runtime.misc.Interval
import org.antlr.v4.runtime.tree.{ParseTree, TerminalNode}
+import org.apache.spark.QueryContext
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.parser.extensions.PaimonParserUtils.withOrigin
import org.apache.spark.sql.catalyst.parser.extensions.PaimonSqlExtensionsParser._
import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin}
+import org.apache.spark.sql.catalyst.trees.SQLQueryContext
import scala.collection.JavaConverters._
@@ -133,3 +134,23 @@ object PaimonParserUtils {
stream.getText(Interval.of(0, stream.size() - 1))
}
}
+
+case class Origin(
+ line: Option[Int] = None,
+ startPosition: Option[Int] = None,
+ startIndex: Option[Int] = None,
+ stopIndex: Option[Int] = None,
+ sqlText: Option[String] = None,
+ objectType: Option[String] = None,
+ objectName: Option[String] = None) {}
+
+object CurrentOrigin {
+ private val value = new ThreadLocal[Origin]() {
+ override def initialValue: Origin = Origin()
+ }
+
+ def get: Origin = value.get()
+ def set(o: Origin): Unit = value.set(o)
+ def reset(): Unit = value.set(Origin())
+}
+/* Apache Spark copy end */
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
similarity index 99%
rename from paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTest.scala
rename to paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
index 788c8dd3e..d37002d25 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
@@ -29,7 +29,7 @@ import org.assertj.core.api.Assertions
import java.util
/** Test sort compact procedure. See [[CompactProcedure]]. */
-class CompactProcedureTest extends PaimonSparkTestBase with StreamTest {
+abstract class CompactProcedureTestBase extends PaimonSparkTestBase with StreamTest {
import testImplicits._
(incubator-paimon) 01/02: [spark] Supports CTAS for different versions (#2591)
Posted by bi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
biyan pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit ca245748cd004bb2486dd4130e4fb31f3b69b168
Author: Yann Byron <bi...@gmail.com>
AuthorDate: Tue Jan 2 16:17:12 2024 +0800
[spark] Supports CTAS for different versions (#2591)
---
paimon-spark/paimon-spark-3.2/pom.xml | 59 ++++++++++++++++++
.../shim/PaimonCreateTableAsSelectStrategy.scala | 57 +++++++++++++++++
.../org/apache/paimon/spark/sql/DDLTest.scala | 20 ++++++
.../test/scala/org/apache/spark/paimon/Utils.scala | 31 ++++++++++
paimon-spark/paimon-spark-3.3/pom.xml | 7 +++
.../org/apache/paimon/spark/sql/DDLTest.scala | 41 +------------
paimon-spark/paimon-spark-3.4/pom.xml | 41 +++++++------
.../shim/PaimonCreateTableAsSelectStrategy.scala | 71 ++++++++++++++++++++++
.../org/apache/paimon/spark/sql/DDLTest.scala | 20 ++++++
.../org/apache/paimon/spark/sql/DDLTest.scala | 20 ++++++
paimon-spark/paimon-spark-common/pom.xml | 14 +++++
.../org/apache/paimon/spark/sql/DDLTest.scala | 59 ------------------
.../org/apache/paimon/spark/sql/DDLTestBase.scala | 61 +++++++++++++++++++
13 files changed, 384 insertions(+), 117 deletions(-)
diff --git a/paimon-spark/paimon-spark-3.2/pom.xml b/paimon-spark/paimon-spark-3.2/pom.xml
index 3af4ca84b..332ce4ae2 100644
--- a/paimon-spark/paimon-spark-3.2/pom.xml
+++ b/paimon-spark/paimon-spark-3.2/pom.xml
@@ -76,9 +76,24 @@ under the License.
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j2-impl</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>org.apache.orc</groupId>
+ <artifactId>orc-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.orc</groupId>
+ <artifactId>orc-mapreduce</artifactId>
+ </exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-spark-common</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
@@ -156,6 +171,26 @@ under the License.
</dependencies>
<build>
+ <pluginManagement>
+ <plugins>
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ <version>${scala-maven-plugin.version}</version>
+ <configuration>
+ <args>
+ <arg>-nobootcp</arg>
+ <arg>-target:jvm-${target.java.version}</arg>
+ </args>
+ <checkMultipleScalaVersions>false</checkMultipleScalaVersions>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
@@ -222,6 +257,30 @@ under the License.
</scala>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest-maven-plugin</artifactId>
+ <version>${scalatest-maven-plugin.version}</version>
+ <configuration>
+ <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
+ <junitxml>.</junitxml>
+ <argLine>-ea -Xmx4g -Xss4m -XX:MaxMetaspaceSize=2g -XX:ReservedCodeCacheSize=${CodeCacheSize} ${extraJavaTestArgs} -Dio.netty.tryReflectionSetAccessible=true</argLine>
+ <filereports>PaimonTestSuite.txt</filereports>
+ <!--
+ the plugin does not support always now, we must be careful w/ singletons.
+ tracking at: https://github.com/scalatest/scalatest-maven-plugin/issues/99
+ -->
+ <forkMode>once</forkMode>
+ </configuration>
+ <executions>
+ <execution>
+ <id>test</id>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
</project>
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
new file mode 100644
index 000000000..299d07f20
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
@@ -0,0 +1,57 @@
+/*
+ * 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.spark.sql.execution.shim
+
+import org.apache.paimon.CoreOptions
+
+import org.apache.spark.sql.{SparkSession, Strategy}
+import org.apache.spark.sql.catalyst.plans.logical.{CreateTableAsSelect, LogicalPlan}
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, StagingTableCatalog}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.datasources.v2.CreateTableAsSelectExec
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import scala.collection.JavaConverters._
+
+case class PaimonCreateTableAsSelectStrategy(spark: SparkSession) extends Strategy {
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case CreateTableAsSelect(catalog, ident, parts, query, props, options, ifNotExists) =>
+ catalog match {
+ case _: StagingTableCatalog =>
+ throw new RuntimeException("Paimon can't extend StagingTableCatalog for now.")
+ case _ =>
+ val coreOptionKeys = CoreOptions.getOptions.asScala.map(_.key()).toSeq
+ val (coreOptions, writeOptions) = options.partition {
+ case (key, _) => coreOptionKeys.contains(key)
+ }
+ val newProps = CatalogV2Util.withDefaultOwnership(props) ++ coreOptions
+ CreateTableAsSelectExec(
+ catalog,
+ ident,
+ parts,
+ query,
+ planLater(query),
+ newProps,
+ new CaseInsensitiveStringMap(writeOptions.asJava),
+ ifNotExists
+ ) :: Nil
+ }
+ case _ => Nil
+ }
+}
diff --git a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
new file mode 100644
index 000000000..8756c5b57
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
@@ -0,0 +1,20 @@
+/*
+ * 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 DDLTest extends DDLTestBase {}
diff --git a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/spark/paimon/Utils.scala b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/spark/paimon/Utils.scala
new file mode 100644
index 000000000..974bbf0c7
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/spark/paimon/Utils.scala
@@ -0,0 +1,31 @@
+/*
+ * 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.spark.paimon
+
+import org.apache.spark.util.{Utils => SparkUtils}
+
+import java.io.File
+
+/**
+ * A wrapper that some Objects or Classes is limited to access beyond [[org.apache.spark]] package.
+ */
+object Utils {
+
+ def createTempDir: File = SparkUtils.createTempDir()
+
+}
diff --git a/paimon-spark/paimon-spark-3.3/pom.xml b/paimon-spark/paimon-spark-3.3/pom.xml
index 75a2d0f91..6191495af 100644
--- a/paimon-spark/paimon-spark-3.3/pom.xml
+++ b/paimon-spark/paimon-spark-3.3/pom.xml
@@ -79,6 +79,13 @@ under the License.
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-spark-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
diff --git a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
index 7187db29d..8756c5b57 100644
--- a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
+++ b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
@@ -17,43 +17,4 @@
*/
package org.apache.paimon.spark.sql
-import org.apache.paimon.spark.PaimonSparkTestBase
-
-import org.junit.jupiter.api.Assertions
-
-class DDLTest extends PaimonSparkTestBase {
-
- import testImplicits._
-
- test("Paimon: Create Table As Select") {
- Seq((1L, "x1", "2023"), (2L, "x2", "2023"))
- .toDF("a", "b", "pt")
- .createOrReplaceTempView("source")
-
- spark.sql("""
- |CREATE TABLE t1 AS SELECT * FROM source
- |""".stripMargin)
- val t1 = loadTable("t1")
- Assertions.assertTrue(t1.primaryKeys().isEmpty)
- Assertions.assertTrue(t1.partitionKeys().isEmpty)
-
- spark.sql(
- """
- |CREATE TABLE t2
- |PARTITIONED BY (pt)
- |TBLPROPERTIES ('bucket' = '5', 'primary-key' = 'a,pt', 'target-file-size' = '128MB')
- |AS SELECT * FROM source
- |""".stripMargin)
- val t2 = loadTable("t2")
- Assertions.assertEquals(2, t2.primaryKeys().size())
- Assertions.assertTrue(t2.primaryKeys().contains("a"))
- Assertions.assertTrue(t2.primaryKeys().contains("pt"))
- Assertions.assertEquals(1, t2.partitionKeys().size())
- Assertions.assertEquals("pt", t2.partitionKeys().get(0))
-
- // check all the core options
- Assertions.assertEquals("5", t2.options().get("bucket"))
- Assertions.assertEquals("128MB", t2.options().get("target-file-size"))
- }
-
-}
+class DDLTest extends DDLTestBase {}
diff --git a/paimon-spark/paimon-spark-3.4/pom.xml b/paimon-spark/paimon-spark-3.4/pom.xml
index c2932c418..9ba83157f 100644
--- a/paimon-spark/paimon-spark-3.4/pom.xml
+++ b/paimon-spark/paimon-spark-3.4/pom.xml
@@ -42,6 +42,22 @@ under the License.
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <version>${scala.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-reflect</artifactId>
+ <version>${scala.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-compiler</artifactId>
+ <version>${scala.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
@@ -63,6 +79,13 @@ under the License.
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-spark-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
@@ -130,24 +153,6 @@ under the License.
</exclusion>
</exclusions>
</dependency>
- <dependency>
- <groupId>org.scala-lang</groupId>
- <artifactId>scala-library</artifactId>
- <version>${scala.version}</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.scala-lang</groupId>
- <artifactId>scala-reflect</artifactId>
- <version>${scala.version}</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.scala-lang</groupId>
- <artifactId>scala-compiler</artifactId>
- <version>${scala.version}</version>
- <scope>test</scope>
- </dependency>
<dependency>
<groupId>org.scalatest</groupId>
<artifactId>scalatest_${scala.binary.version}</artifactId>
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
new file mode 100644
index 000000000..e631bbea6
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
@@ -0,0 +1,71 @@
+/*
+ * 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.spark.sql.execution.shim
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.spark.SparkCatalog
+
+import org.apache.spark.sql.{SparkSession, Strategy}
+import org.apache.spark.sql.catalyst.analysis.ResolvedIdentifier
+import org.apache.spark.sql.catalyst.plans.logical.{CreateTableAsSelect, LogicalPlan, TableSpec}
+import org.apache.spark.sql.connector.catalog.StagingTableCatalog
+import org.apache.spark.sql.execution.{PaimonStrategyHelper, SparkPlan}
+import org.apache.spark.sql.execution.datasources.v2.CreateTableAsSelectExec
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+import scala.collection.JavaConverters._
+
+case class PaimonCreateTableAsSelectStrategy(spark: SparkSession)
+ extends Strategy
+ with PaimonStrategyHelper {
+
+ import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case CreateTableAsSelect(
+ ResolvedIdentifier(catalog: SparkCatalog, ident),
+ parts,
+ query,
+ tableSpec: TableSpec,
+ options,
+ ifNotExists,
+ analyzedQuery) =>
+ assert(analyzedQuery.isDefined)
+ catalog match {
+ case _: StagingTableCatalog =>
+ throw new RuntimeException("Paimon can't extend StagingTableCatalog for now.")
+ case _ =>
+ val coreOptionKeys = CoreOptions.getOptions.asScala.map(_.key()).toSeq
+ val (coreOptions, writeOptions) = options.partition {
+ case (key, _) => coreOptionKeys.contains(key)
+ }
+ val newTableSpec = tableSpec.copy(properties = tableSpec.properties ++ coreOptions)
+ CreateTableAsSelectExec(
+ catalog.asTableCatalog,
+ ident,
+ parts,
+ analyzedQuery.get,
+ planLater(query),
+ qualifyLocInTableSpec(newTableSpec),
+ new CaseInsensitiveStringMap(writeOptions.asJava),
+ ifNotExists
+ ) :: Nil
+ }
+ case _ => Nil
+ }
+}
diff --git a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
new file mode 100644
index 000000000..8756c5b57
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
@@ -0,0 +1,20 @@
+/*
+ * 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 DDLTest extends DDLTestBase {}
diff --git a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
new file mode 100644
index 000000000..8756c5b57
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
@@ -0,0 +1,20 @@
+/*
+ * 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 DDLTest extends DDLTestBase {}
diff --git a/paimon-spark/paimon-spark-common/pom.xml b/paimon-spark/paimon-spark-common/pom.xml
index c7e6b9343..d3f23a71e 100644
--- a/paimon-spark/paimon-spark-common/pom.xml
+++ b/paimon-spark/paimon-spark-common/pom.xml
@@ -235,7 +235,21 @@ under the License.
</plugin>
</plugins>
</pluginManagement>
+
<plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>prepare-test-jar</id>
+ <phase>test-compile</phase>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
deleted file mode 100644
index 7187db29d..000000000
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * 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.junit.jupiter.api.Assertions
-
-class DDLTest extends PaimonSparkTestBase {
-
- import testImplicits._
-
- test("Paimon: Create Table As Select") {
- Seq((1L, "x1", "2023"), (2L, "x2", "2023"))
- .toDF("a", "b", "pt")
- .createOrReplaceTempView("source")
-
- spark.sql("""
- |CREATE TABLE t1 AS SELECT * FROM source
- |""".stripMargin)
- val t1 = loadTable("t1")
- Assertions.assertTrue(t1.primaryKeys().isEmpty)
- Assertions.assertTrue(t1.partitionKeys().isEmpty)
-
- spark.sql(
- """
- |CREATE TABLE t2
- |PARTITIONED BY (pt)
- |TBLPROPERTIES ('bucket' = '5', 'primary-key' = 'a,pt', 'target-file-size' = '128MB')
- |AS SELECT * FROM source
- |""".stripMargin)
- val t2 = loadTable("t2")
- Assertions.assertEquals(2, t2.primaryKeys().size())
- Assertions.assertTrue(t2.primaryKeys().contains("a"))
- Assertions.assertTrue(t2.primaryKeys().contains("pt"))
- Assertions.assertEquals(1, t2.partitionKeys().size())
- Assertions.assertEquals("pt", t2.partitionKeys().get(0))
-
- // check all the core options
- Assertions.assertEquals("5", t2.options().get("bucket"))
- Assertions.assertEquals("128MB", t2.options().get("target-file-size"))
- }
-
-}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
new file mode 100644
index 000000000..8bb3bd5e8
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
@@ -0,0 +1,61 @@
+/*
+ * 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.junit.jupiter.api.Assertions
+
+abstract class DDLTestBase extends PaimonSparkTestBase {
+
+ import testImplicits._
+
+ test("Paimon: Create Table As Select") {
+ withTable("source", "t1", "t2") {
+ Seq((1L, "x1", "2023"), (2L, "x2", "2023"))
+ .toDF("a", "b", "pt")
+ .createOrReplaceTempView("source")
+
+ spark.sql("""
+ |CREATE TABLE t1 AS SELECT * FROM source
+ |""".stripMargin)
+ val t1 = loadTable("t1")
+ Assertions.assertTrue(t1.primaryKeys().isEmpty)
+ Assertions.assertTrue(t1.partitionKeys().isEmpty)
+
+ spark.sql(
+ """
+ |CREATE TABLE t2
+ |PARTITIONED BY (pt)
+ |TBLPROPERTIES ('bucket' = '5', 'primary-key' = 'a,pt', 'target-file-size' = '128MB')
+ |AS SELECT * FROM source
+ |""".stripMargin)
+ val t2 = loadTable("t2")
+ Assertions.assertEquals(2, t2.primaryKeys().size())
+ Assertions.assertTrue(t2.primaryKeys().contains("a"))
+ Assertions.assertTrue(t2.primaryKeys().contains("pt"))
+ Assertions.assertEquals(1, t2.partitionKeys().size())
+ Assertions.assertEquals("pt", t2.partitionKeys().get(0))
+
+ // check all the core options
+ Assertions.assertEquals("5", t2.options().get("bucket"))
+ Assertions.assertEquals("128MB", t2.options().get("target-file-size"))
+ }
+ }
+
+}