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 06:15:27 UTC
(incubator-paimon) branch release-0.6 updated (73fc2cc79 -> 5534b45bf)
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
from 73fc2cc79 Revert "[hive] Improve hive database is exists yet create/drop database (#2595)"
new 0db173f51 [core] Remove ConfigOptions toList (#2578)
new 454fd9e75 [core] Fix nested-update to nested_update
new e589ddd6a [test] Fix unstable test of parallelism set error. (#2626)
new 7d0e05055 [flink] throws exception if there is a SinkMaterializer before Paimon sink (#2592)
new 69663432e [hotfix] Fix paimon-spark-3.2 pom (#2603)
new 5c04b45a8 [spark] Supports CTAS for different versions (#2591)
new 5534b45bf [spark] Fix call procedure compatibility with spark3.2 (#2707)
The 7 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:
docs/content/concepts/primary-key-table.md | 126 +++++++++-
.../org/apache/paimon/options/ConfigOption.java | 19 +-
.../org/apache/paimon/options/ConfigOptions.java | 54 +---
.../java/org/apache/paimon/options/Options.java | 6 +-
.../org/apache/paimon/options/OptionsUtils.java | 14 +-
.../compact/aggregate/FieldNestedUpdateAgg.java | 91 +++++++
.../configuration/ConfigOptionsDocGenerator.java | 39 +--
.../ConfigOptionsDocGeneratorTest.java | 28 ---
.../org/apache/paimon/flink/sink/FlinkSink.java | 33 +++
.../apache/paimon/flink/PartialUpdateITCase.java | 20 ++
.../apache/paimon/flink/PreAggregationITCase.java | 274 +++++++++++++++++++++
.../flink/PrimaryKeyFileStoreTableITCase.java | 6 +-
.../paimon/flink/sink/CompactorSinkITCase.java | 2 +-
paimon-spark/paimon-spark-3.2/pom.xml | 66 ++++-
.../shim/PaimonCreateTableAsSelectStrategy.scala | 30 +--
.../paimon/spark/sql/CompactProcedureTest.scala | 7 +-
.../org/apache/paimon/spark/sql/DDLTest.scala | 7 +-
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 | 12 +-
.../org/apache/paimon/spark/sql/DDLTest.scala | 7 +-
.../org/apache/paimon/spark/sql/DDLTest.scala | 7 +-
paimon-spark/paimon-spark-common/pom.xml | 14 ++
.../extensions/PaimonSqlExtensionsAstBuilder.scala | 23 +-
...reTest.scala => CompactProcedureTestBase.scala} | 2 +-
.../org/apache/paimon/spark/sql/DDLTest.scala | 59 -----
.../org/apache/paimon/spark/sql/DDLTestBase.scala | 61 +++++
28 files changed, 770 insertions(+), 326 deletions(-)
create mode 100644 paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java
copy paimon-spark/{paimon-spark-3.3 => paimon-spark-3.2}/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala (70%)
copy paimon-common/src/test/resources/codesplit/groups/expected/IfInWhile.txt => paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/CompactProcedureTest.scala (83%)
copy paimon-common/src/test/resources/codesplit/groups/expected/IfInWhile.txt => paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala (91%)
copy paimon-spark/{paimon-spark-3.3 => paimon-spark-3.4}/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala (90%)
copy paimon-common/src/test/resources/codesplit/groups/expected/IfInWhile.txt => paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala (91%)
copy paimon-common/src/test/resources/codesplit/groups/expected/IfInWhile.txt => paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala (91%)
rename paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/{CompactProcedureTest.scala => CompactProcedureTestBase.scala} (99%)
delete mode 100644 paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTest.scala
create mode 100644 paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
(incubator-paimon) 07/07: [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 5534b45bf75fe224a25d600c6970fb36d818c6e7
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) 06/07: [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 5c04b45a8a28ff392faf31b34246a5e107dfae46
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 ++++++
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 +++++++++++++++++++
12 files changed, 353 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.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"))
+ }
+ }
+
+}
(incubator-paimon) 05/07: [hotfix] Fix paimon-spark-3.2 pom (#2603)
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 69663432e60a1dda86e5d254f9515eae734ef2fb
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Tue Jan 2 09:20:04 2024 +0800
[hotfix] Fix paimon-spark-3.2 pom (#2603)
---
paimon-spark/paimon-spark-3.2/pom.xml | 7 -------
1 file changed, 7 deletions(-)
diff --git a/paimon-spark/paimon-spark-3.2/pom.xml b/paimon-spark/paimon-spark-3.2/pom.xml
index 85c0a87fd..3af4ca84b 100644
--- a/paimon-spark/paimon-spark-3.2/pom.xml
+++ b/paimon-spark/paimon-spark-3.2/pom.xml
@@ -153,13 +153,6 @@ under the License.
<version>3.1.0</version>
<scope>test</scope>
</dependency>
-
- <dependency>
- <groupId>org.apache.avro</groupId>
- <artifactId>avro</artifactId>
- <version>${avro.version}</version>
- <scope>test</scope>
- </dependency>
</dependencies>
<build>
(incubator-paimon) 02/07: [core] Fix nested-update to nested_update
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 454fd9e75a76e21dd20b8d0ff8ac3447130860df
Author: Jingsong <ji...@gmail.com>
AuthorDate: Wed Dec 27 19:13:21 2023 +0800
[core] Fix nested-update to nested_update
---
docs/content/concepts/primary-key-table.md | 8 +-
.../compact/aggregate/FieldNestedUpdateAgg.java | 91 +++++++
.../apache/paimon/flink/PreAggregationITCase.java | 274 +++++++++++++++++++++
3 files changed, 369 insertions(+), 4 deletions(-)
diff --git a/docs/content/concepts/primary-key-table.md b/docs/content/concepts/primary-key-table.md
index a79af8a01..541e5fd83 100644
--- a/docs/content/concepts/primary-key-table.md
+++ b/docs/content/concepts/primary-key-table.md
@@ -326,14 +326,14 @@ Current supported aggregate functions and data types are:
The first_not_null_value function selects the first non-null value in a data set.
It supports all data types.
-* `nested-update`:
- The nested-update function collects multiple rows into one array<row> (so-called 'nested table'). It supports ARRAY<ROW> data types.
+* `nested_update`:
+ The nested_update function collects multiple rows into one array<row> (so-called 'nested table'). It supports ARRAY<ROW> data types.
Use `fields.<field-name>.nested-key=pk0,pk1,...` to specify the primary keys of the nested table. If no keys, row will be appended to array<row>.
An example:
- {{< tabs "nested-update-example" >}}
+ {{< tabs "nested_update-example" >}}
{{< tab "Flink" >}}
@@ -363,7 +363,7 @@ Current supported aggregate functions and data types are:
sub_orders ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>
) WITH (
'merge-engine' = 'aggregation',
- 'fields.sub_orders.aggregate-function' = 'nested-update',
+ 'fields.sub_orders.aggregate-function' = 'nested_update',
'fields.sub_orders.nested-key' = 'sub_order_id'
);
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java
new file mode 100644
index 000000000..69e2736ce
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java
@@ -0,0 +1,91 @@
+/*
+ * 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.mergetree.compact.aggregate;
+
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.Projection;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Used to update a field which representing a nested table. The data type of nested table field is
+ * {@code ARRAY<ROW>}.
+ */
+public class FieldNestedUpdateAgg extends FieldAggregator {
+
+ public static final String NAME = "nested_update";
+
+ private final int nestedFields;
+
+ @Nullable private final Projection keyProjection;
+
+ public FieldNestedUpdateAgg(ArrayType dataType, List<String> nestedKey) {
+ super(dataType);
+ RowType nestedType = (RowType) dataType.getElementType();
+ this.nestedFields = nestedType.getFieldCount();
+ this.keyProjection =
+ nestedKey.isEmpty() ? null : CodeGenUtils.newProjection(nestedType, nestedKey);
+ }
+
+ @Override
+ String name() {
+ return NAME;
+ }
+
+ @Override
+ public Object agg(Object accumulator, Object inputField) {
+ if (accumulator == null || inputField == null) {
+ return accumulator == null ? inputField : accumulator;
+ }
+
+ InternalArray acc = (InternalArray) accumulator;
+ InternalArray input = (InternalArray) inputField;
+
+ List<InternalRow> rows = new ArrayList<>();
+ for (int i = 0; i < acc.size(); i++) {
+ rows.add(acc.getRow(i, nestedFields));
+ }
+ for (int i = 0; i < input.size(); i++) {
+ rows.add(input.getRow(i, nestedFields));
+ }
+
+ if (keyProjection != null) {
+ Map<BinaryRow, InternalRow> map = new HashMap<>();
+ for (InternalRow row : rows) {
+ BinaryRow key = keyProjection.apply(row).copy();
+ map.put(key, row);
+ }
+
+ rows = new ArrayList<>(map.values());
+ }
+
+ return new GenericArray(rows.toArray());
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
index 170d5d5fc..63670eb7c 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
@@ -18,18 +18,24 @@
package org.apache.paimon.flink;
+import org.apache.paimon.mergetree.compact.aggregate.FieldNestedUpdateAgg;
import org.apache.paimon.utils.BlockingIterator;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.CloseableIterator;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -969,4 +975,272 @@ public class PreAggregationITCase {
.containsExactlyInAnyOrder(Row.of(1, 3, 1), Row.of(2, 1, 1));
}
}
+
+ /** ITCase for {@link FieldNestedUpdateAgg}. */
+ public static class NestedUpdateAggregationITCase extends CatalogITCaseBase {
+
+ @Override
+ protected List<String> ddl() {
+ String ordersTable =
+ "CREATE TABLE orders (\n"
+ + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
+ + " user_name STRING,\n"
+ + " address STRING\n"
+ + ");";
+
+ String subordersTable =
+ "CREATE TABLE sub_orders (\n"
+ + " order_id INT,\n"
+ + " daily_id INT,\n"
+ + " today STRING,\n"
+ + " product_name STRING,\n"
+ + " price BIGINT,\n"
+ + " PRIMARY KEY (order_id, daily_id, today) NOT ENFORCED\n"
+ + ");";
+
+ String wideTable =
+ "CREATE TABLE order_wide (\n"
+ + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
+ + " user_name STRING,\n"
+ + " address STRING,\n"
+ + " sub_orders ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>\n"
+ + ") WITH (\n"
+ + " 'merge-engine' = 'aggregation',\n"
+ + " 'fields.sub_orders.aggregate-function' = 'nested_update',\n"
+ + " 'fields.sub_orders.nested-key' = 'daily_id,today',\n"
+ + " 'fields.sub_orders.ignore-retract' = 'true',"
+ + " 'fields.user_name.ignore-retract' = 'true',"
+ + " 'fields.address.ignore-retract' = 'true'"
+ + ")";
+
+ String wideAppendTable =
+ "CREATE TABLE order_append_wide (\n"
+ + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
+ + " user_name STRING,\n"
+ + " address STRING,\n"
+ + " sub_orders ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>\n"
+ + ") WITH (\n"
+ + " 'merge-engine' = 'aggregation',\n"
+ + " 'fields.sub_orders.aggregate-function' = 'nested_update',\n"
+ + " 'fields.sub_orders.ignore-retract' = 'true',"
+ + " 'fields.user_name.ignore-retract' = 'true',"
+ + " 'fields.address.ignore-retract' = 'true'"
+ + ")";
+
+ return Arrays.asList(ordersTable, subordersTable, wideTable, wideAppendTable);
+ }
+
+ @Test
+ public void testUseCase() {
+ sql(
+ "INSERT INTO orders VALUES "
+ + "(1, 'Wang', 'HangZhou'),"
+ + "(2, 'Zhao', 'ChengDu'),"
+ + "(3, 'Liu', 'NanJing')");
+
+ sql(
+ "INSERT INTO sub_orders VALUES "
+ + "(1, 1, '12-20', 'Apple', 8000),"
+ + "(1, 2, '12-20', 'Tesla', 400000),"
+ + "(1, 1, '12-21', 'Sangsung', 5000),"
+ + "(2, 1, '12-20', 'Tea', 40),"
+ + "(2, 2, '12-20', 'Pot', 60),"
+ + "(3, 1, '12-25', 'Bat', 15),"
+ + "(3, 1, '12-26', 'Cup', 30)");
+
+ sql(widenSql());
+
+ List<Row> result =
+ sql("SELECT * FROM order_wide").stream()
+ .sorted(Comparator.comparingInt(r -> r.getFieldAs(0)))
+ .collect(Collectors.toList());
+
+ assertThat(
+ checkOneRecord(
+ result.get(0),
+ 1,
+ "Wang",
+ "HangZhou",
+ Row.of(1, "12-20", "Apple", 8000L),
+ Row.of(1, "12-21", "Sangsung", 5000L),
+ Row.of(2, "12-20", "Tesla", 400000L)))
+ .isTrue();
+ assertThat(
+ checkOneRecord(
+ result.get(1),
+ 2,
+ "Zhao",
+ "ChengDu",
+ Row.of(1, "12-20", "Tea", 40L),
+ Row.of(2, "12-20", "Pot", 60L)))
+ .isTrue();
+ assertThat(
+ checkOneRecord(
+ result.get(2),
+ 3,
+ "Liu",
+ "NanJing",
+ Row.of(1, "12-25", "Bat", 15L),
+ Row.of(1, "12-26", "Cup", 30L)))
+ .isTrue();
+
+ // query using UNNEST
+ List<Row> unnested =
+ sql(
+ "SELECT order_id, user_name, address, daily_id, today, product_name, price "
+ + "FROM order_wide, UNNEST(sub_orders) AS so(daily_id, today, product_name, price)");
+
+ assertThat(unnested)
+ .containsExactlyInAnyOrder(
+ Row.of(1, "Wang", "HangZhou", 1, "12-20", "Apple", 8000L),
+ Row.of(1, "Wang", "HangZhou", 2, "12-20", "Tesla", 400000L),
+ Row.of(1, "Wang", "HangZhou", 1, "12-21", "Sangsung", 5000L),
+ Row.of(2, "Zhao", "ChengDu", 1, "12-20", "Tea", 40L),
+ Row.of(2, "Zhao", "ChengDu", 2, "12-20", "Pot", 60L),
+ Row.of(3, "Liu", "NanJing", 1, "12-25", "Bat", 15L),
+ Row.of(3, "Liu", "NanJing", 1, "12-26", "Cup", 30L));
+ }
+
+ @Test
+ public void testUseCaseAppend() {
+ sql(
+ "INSERT INTO orders VALUES "
+ + "(1, 'Wang', 'HangZhou'),"
+ + "(2, 'Zhao', 'ChengDu'),"
+ + "(3, 'Liu', 'NanJing')");
+
+ sql(
+ "INSERT INTO sub_orders VALUES "
+ + "(1, 1, '12-20', 'Apple', 8000),"
+ + "(2, 1, '12-20', 'Tesla', 400000),"
+ + "(3, 1, '12-25', 'Bat', 15),"
+ + "(3, 1, '12-26', 'Cup', 30)");
+
+ sql(widenAppendSql());
+
+ // query using UNNEST
+ List<Row> unnested =
+ sql(
+ "SELECT order_id, user_name, address, daily_id, today, product_name, price "
+ + "FROM order_append_wide, UNNEST(sub_orders) AS so(daily_id, today, product_name, price)");
+
+ assertThat(unnested)
+ .containsExactlyInAnyOrder(
+ Row.of(1, "Wang", "HangZhou", 1, "12-20", "Apple", 8000L),
+ Row.of(2, "Zhao", "ChengDu", 1, "12-20", "Tesla", 400000L),
+ Row.of(3, "Liu", "NanJing", 1, "12-25", "Bat", 15L),
+ Row.of(3, "Liu", "NanJing", 1, "12-26", "Cup", 30L));
+ }
+
+ @Test
+ @Timeout(60)
+ public void testUpdateWithIgnoreRetract() throws Exception {
+ sEnv.getConfig()
+ .set(
+ ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
+ ExecutionConfigOptions.UpsertMaterialize.NONE);
+
+ sql("INSERT INTO orders VALUES (1, 'Wang', 'HangZhou')");
+
+ sql(
+ "INSERT INTO sub_orders VALUES "
+ + "(1, 1, '12-20', 'Apple', 8000),"
+ + "(1, 2, '12-20', 'Tesla', 400000),"
+ + "(1, 1, '12-21', 'Sangsung', 5000)");
+
+ sEnv.executeSql(widenSql());
+
+ boolean checkResult;
+ List<Row> result;
+
+ do {
+ Thread.sleep(500);
+ result = sql("SELECT * FROM order_wide");
+ checkResult =
+ !result.isEmpty()
+ && checkOneRecord(
+ result.get(0),
+ 1,
+ "Wang",
+ "HangZhou",
+ Row.of(1, "12-20", "Apple", 8000L),
+ Row.of(1, "12-21", "Sangsung", 5000L),
+ Row.of(2, "12-20", "Tesla", 400000L));
+ } while (!checkResult);
+
+ sql("INSERT INTO sub_orders VALUES (1, 2, '12-20', 'Benz', 380000)");
+
+ do {
+ Thread.sleep(500);
+ result = sql("SELECT * FROM order_wide");
+ checkResult =
+ !result.isEmpty()
+ && checkOneRecord(
+ result.get(0),
+ 1,
+ "Wang",
+ "HangZhou",
+ Row.of(1, "12-20", "Apple", 8000L),
+ Row.of(1, "12-21", "Sangsung", 5000L),
+ Row.of(2, "12-20", "Benz", 380000L));
+ } while (!checkResult);
+ }
+
+ private String widenSql() {
+ return "INSERT INTO order_wide\n"
+ + "SELECT order_id, user_name, address, "
+ + "CAST (NULL AS ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>) FROM orders\n"
+ + "UNION ALL\n"
+ + "SELECT order_id, CAST (NULL AS STRING), CAST (NULL AS STRING), "
+ + "ARRAY[ROW(daily_id, today, product_name, price)] FROM sub_orders";
+ }
+
+ private String widenAppendSql() {
+ return "INSERT INTO order_append_wide\n"
+ + "SELECT order_id, user_name, address, "
+ + "CAST (NULL AS ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>) FROM orders\n"
+ + "UNION ALL\n"
+ + "SELECT order_id, CAST (NULL AS STRING), CAST (NULL AS STRING), "
+ + "ARRAY[ROW(daily_id, today, product_name, price)] FROM sub_orders";
+ }
+
+ private boolean checkOneRecord(
+ Row record, int orderId, String userName, String address, Row... subOrders) {
+ if ((int) record.getField(0) != orderId) {
+ return false;
+ }
+ if (!record.getFieldAs(1).equals(userName)) {
+ return false;
+ }
+ if (!record.getFieldAs(2).equals(address)) {
+ return false;
+ }
+
+ return checkNestedTable(record.getFieldAs(3), subOrders);
+ }
+
+ private boolean checkNestedTable(Row[] nestedTable, Row... subOrders) {
+ if (nestedTable.length != subOrders.length) {
+ return false;
+ }
+
+ Comparator<Row> comparator =
+ (Comparator)
+ Comparator.comparingInt(r -> ((Row) r).getFieldAs(0))
+ .thenComparing(r -> (String) ((Row) r).getField(1));
+
+ List<Row> sortedActual =
+ Arrays.stream(nestedTable).sorted(comparator).collect(Collectors.toList());
+ List<Row> sortedExpected =
+ Arrays.stream(subOrders).sorted(comparator).collect(Collectors.toList());
+
+ for (int i = 0; i < sortedActual.size(); i++) {
+ if (!sortedActual.get(i).equals(sortedExpected.get(i))) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+ }
}
(incubator-paimon) 03/07: [test] Fix unstable test of parallelism set error. (#2626)
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 e589ddd6ac9f509c77235d4138273b46aaeb98d9
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Wed Jan 3 17:42:24 2024 +0800
[test] Fix unstable test of parallelism set error. (#2626)
---
.../src/test/java/org/apache/paimon/flink/sink/CompactorSinkITCase.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactorSinkITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactorSinkITCase.java
index ffb6f961a..8d5e6c781 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactorSinkITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactorSinkITCase.java
@@ -170,7 +170,7 @@ public class CompactorSinkITCase extends AbstractTestBase {
.withContinuousMode(false)
.withPartitions(getSpecifiedPartitions())
.build();
- Integer sinkParalellism = new Random().nextInt(100);
+ Integer sinkParalellism = new Random().nextInt(100) + 1;
new CompactorSinkBuilder(
table.copy(
new HashMap<String, String>() {
(incubator-paimon) 01/07: [core] Remove ConfigOptions toList (#2578)
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 0db173f519aa4446a727723264c501be147842dc
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Wed Dec 27 16:02:49 2023 +0800
[core] Remove ConfigOptions toList (#2578)
---
docs/content/concepts/primary-key-table.md | 126 +++++++++++++++++++--
.../org/apache/paimon/options/ConfigOption.java | 19 +---
.../org/apache/paimon/options/ConfigOptions.java | 54 +--------
.../java/org/apache/paimon/options/Options.java | 6 +-
.../org/apache/paimon/options/OptionsUtils.java | 14 +--
.../configuration/ConfigOptionsDocGenerator.java | 39 +------
.../ConfigOptionsDocGeneratorTest.java | 28 -----
7 files changed, 133 insertions(+), 153 deletions(-)
diff --git a/docs/content/concepts/primary-key-table.md b/docs/content/concepts/primary-key-table.md
index 537ef47bc..a79af8a01 100644
--- a/docs/content/concepts/primary-key-table.md
+++ b/docs/content/concepts/primary-key-table.md
@@ -278,14 +278,124 @@ Field `price` will be aggregated by the `max` function, and field `sales` will b
Current supported aggregate functions and data types are:
-* `sum`: supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT and DOUBLE.
-* `min`/`max`: support CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP and TIMESTAMP_LTZ.
-* `last_value` / `last_non_null_value`: support all data types.
-* `listagg`: supports STRING data type.
-* `bool_and` / `bool_or`: support BOOLEAN data type.
-* `first_value` / `first_not_null_value`: support all data types.
-
-Only `sum` supports retraction (`UPDATE_BEFORE` and `DELETE`), others aggregate functions do not support retraction.
+* `sum`:
+ The sum function aggregates the values across multiple rows.
+ It supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE data types.
+
+* `product`:
+ The product function can compute product values across multiple lines.
+ It supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE data types.
+
+* `count`:
+ The count function counts the values across multiple rows.
+ It supports INTEGER, BIGINT data types.
+
+* `max`:
+ The max function identifies and retains the maximum value.
+ It supports CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ data types.
+
+* `min`:
+ The min function identifies and retains the minimum value.
+ It supports CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ data types.
+
+* `last_value`:
+ The last_value function replaces the previous value with the most recently imported value.
+ It supports all data types.
+
+* `last_non_null_value`:
+ The last_non_null_value function replaces the previous value with the latest non-null value.
+ It supports all data types.
+
+* `listagg`:
+ The listagg function concatenates multiple string values into a single string.
+ It supports STRING data type.
+
+* `bool_and`:
+ The bool_and function evaluates whether all values in a boolean set are true.
+ It supports BOOLEAN data type.
+
+* `bool_or`:
+ The bool_or function checks if at least one value in a boolean set is true.
+ It supports BOOLEAN data type.
+
+* `first_value`:
+ The first_value function retrieves the first null value from a data set.
+ It supports all data types.
+
+* `first_not_null_value`:
+ The first_not_null_value function selects the first non-null value in a data set.
+ It supports all data types.
+
+* `nested-update`:
+ The nested-update function collects multiple rows into one array<row> (so-called 'nested table'). It supports ARRAY<ROW> data types.
+
+ Use `fields.<field-name>.nested-key=pk0,pk1,...` to specify the primary keys of the nested table. If no keys, row will be appended to array<row>.
+
+ An example:
+
+ {{< tabs "nested-update-example" >}}
+
+ {{< tab "Flink" >}}
+
+ ```sql
+ -- orders table
+ CREATE TABLE orders (
+ order_id BIGINT PRIMARY KEY NOT ENFORCED,
+ user_name STRING,
+ address STRING
+ );
+
+ -- sub orders that have the same order_id
+ -- belongs to the same order
+ CREATE TABLE sub_orders (
+ order_id BIGINT,
+ sub_order_id INT,
+ product_name STRING,
+ price BIGINT,
+ PRIMARY KEY (order_id, sub_order_id) NOT ENFORCED
+ );
+
+ -- wide table
+ CREATE TABLE order_wide (
+ order_id BIGINT PRIMARY KEY NOT ENFORCED,
+ user_name STRING,
+ address STRING,
+ sub_orders ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>
+ ) WITH (
+ 'merge-engine' = 'aggregation',
+ 'fields.sub_orders.aggregate-function' = 'nested-update',
+ 'fields.sub_orders.nested-key' = 'sub_order_id'
+ );
+
+ -- widen
+ INSERT INTO order_wide
+
+ SELECT
+ order_id,
+ user_name,
+ address,
+ CAST (NULL AS ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>)
+ FROM orders
+
+ UNION ALL
+
+ SELECT
+ order_id,
+ CAST (NULL AS STRING),
+ CAST (NULL AS STRING),
+ ARRAY[ROW(sub_order_id, product_name, price)]
+ FROM sub_orders;
+
+ -- query using UNNEST
+ SELECT order_id, user_name, address, sub_order_id, product_name, price
+ FROM order_wide, UNNEST(sub_orders) AS so(sub_order_id, product_name, price)
+ ```
+
+ {{< /tab >}}
+
+ {{< /tabs >}}
+
+Only `sum` and `product` supports retraction (`UPDATE_BEFORE` and `DELETE`), others aggregate functions do not support retraction.
If you allow some functions to ignore retraction messages, you can configure:
`'fields.${field_name}.ignore-retract'='true'`.
diff --git a/paimon-common/src/main/java/org/apache/paimon/options/ConfigOption.java b/paimon-common/src/main/java/org/apache/paimon/options/ConfigOption.java
index d6a0c0ed3..399683aa7 100644
--- a/paimon-common/src/main/java/org/apache/paimon/options/ConfigOption.java
+++ b/paimon-common/src/main/java/org/apache/paimon/options/ConfigOption.java
@@ -70,23 +70,16 @@ public class ConfigOption<T> {
* <ul>
* <li>typeClass == atomic class (e.g. {@code Integer.class}) -> {@code ConfigOption<Integer>}
* <li>typeClass == {@code Map.class} -> {@code ConfigOption<Map<String, String>>}
- * <li>typeClass == atomic class and isList == true for {@code ConfigOption<List<Integer>>}
* </ul>
*/
private final Class<?> clazz;
- private final boolean isList;
-
// ------------------------------------------------------------------------
Class<?> getClazz() {
return clazz;
}
- boolean isList() {
- return isList;
- }
-
/**
* Creates a new config option with fallback keys.
*
@@ -94,8 +87,6 @@ public class ConfigOption<T> {
* @param clazz describes type of the ConfigOption, see description of the clazz field
* @param description Description for that option
* @param defaultValue The default value for this option
- * @param isList tells if the ConfigOption describes a list option, see description of the clazz
- * field
* @param fallbackKeys The list of fallback keys, in the order to be checked
*/
ConfigOption(
@@ -103,14 +94,12 @@ public class ConfigOption<T> {
Class<?> clazz,
Description description,
T defaultValue,
- boolean isList,
FallbackKey... fallbackKeys) {
this.key = checkNotNull(key);
this.description = description;
this.defaultValue = defaultValue;
this.fallbackKeys = fallbackKeys == null || fallbackKeys.length == 0 ? EMPTY : fallbackKeys;
this.clazz = checkNotNull(clazz);
- this.isList = isList;
}
// ------------------------------------------------------------------------
@@ -134,8 +123,7 @@ public class ConfigOption<T> {
// put fallback keys first so that they are prioritized
final FallbackKey[] mergedAlternativeKeys =
Stream.concat(newFallbackKeys, currentAlternativeKeys).toArray(FallbackKey[]::new);
- return new ConfigOption<>(
- key, clazz, description, defaultValue, isList, mergedAlternativeKeys);
+ return new ConfigOption<>(key, clazz, description, defaultValue, mergedAlternativeKeys);
}
/**
@@ -158,8 +146,7 @@ public class ConfigOption<T> {
final FallbackKey[] mergedAlternativeKeys =
Stream.concat(currentAlternativeKeys, newDeprecatedKeys)
.toArray(FallbackKey[]::new);
- return new ConfigOption<>(
- key, clazz, description, defaultValue, isList, mergedAlternativeKeys);
+ return new ConfigOption<>(key, clazz, description, defaultValue, mergedAlternativeKeys);
}
/**
@@ -181,7 +168,7 @@ public class ConfigOption<T> {
* @return A new config option, with given description.
*/
public ConfigOption<T> withDescription(final Description description) {
- return new ConfigOption<>(key, clazz, description, defaultValue, isList, fallbackKeys);
+ return new ConfigOption<>(key, clazz, description, defaultValue, fallbackKeys);
}
// ------------------------------------------------------------------------
diff --git a/paimon-common/src/main/java/org/apache/paimon/options/ConfigOptions.java b/paimon-common/src/main/java/org/apache/paimon/options/ConfigOptions.java
index 72f561335..caef58946 100644
--- a/paimon-common/src/main/java/org/apache/paimon/options/ConfigOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/options/ConfigOptions.java
@@ -22,8 +22,6 @@ import org.apache.paimon.annotation.Public;
import org.apache.paimon.options.description.Description;
import java.time.Duration;
-import java.util.Arrays;
-import java.util.List;
import java.util.Map;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -181,8 +179,7 @@ public class ConfigOptions {
@Deprecated
public <T> ConfigOption<T> defaultValue(T value) {
checkNotNull(value);
- return new ConfigOption<>(
- key, value.getClass(), ConfigOption.EMPTY_DESCRIPTION, value, false);
+ return new ConfigOption<>(key, value.getClass(), ConfigOption.EMPTY_DESCRIPTION, value);
}
/**
@@ -195,8 +192,7 @@ public class ConfigOptions {
*/
@Deprecated
public ConfigOption<String> noDefaultValue() {
- return new ConfigOption<>(
- key, String.class, ConfigOption.EMPTY_DESCRIPTION, null, false);
+ return new ConfigOption<>(key, String.class, ConfigOption.EMPTY_DESCRIPTION, null);
}
}
@@ -214,11 +210,6 @@ public class ConfigOptions {
this.clazz = clazz;
}
- /** Defines that the option's type should be a list of previously defined atomic type. */
- public ListConfigOptionBuilder<T> asList() {
- return new ListConfigOptionBuilder<>(key, clazz);
- }
-
/**
* Creates a ConfigOption with the given default value.
*
@@ -226,7 +217,7 @@ public class ConfigOptions {
* @return The config option with the default value.
*/
public ConfigOption<T> defaultValue(T value) {
- return new ConfigOption<>(key, clazz, ConfigOption.EMPTY_DESCRIPTION, value, false);
+ return new ConfigOption<>(key, clazz, ConfigOption.EMPTY_DESCRIPTION, value);
}
/**
@@ -235,44 +226,7 @@ public class ConfigOptions {
* @return The config option without a default value.
*/
public ConfigOption<T> noDefaultValue() {
- return new ConfigOption<>(
- key, clazz, Description.builder().text("").build(), null, false);
- }
- }
-
- /**
- * Builder for {@link ConfigOption} of list of type {@link E}.
- *
- * @param <E> list element type of the option
- */
- public static class ListConfigOptionBuilder<E> {
- private final String key;
- private final Class<E> clazz;
-
- ListConfigOptionBuilder(String key, Class<E> clazz) {
- this.key = key;
- this.clazz = clazz;
- }
-
- /**
- * Creates a ConfigOption with the given default value.
- *
- * @param values The list of default values for the config option
- * @return The config option with the default value.
- */
- @SafeVarargs
- public final ConfigOption<List<E>> defaultValues(E... values) {
- return new ConfigOption<>(
- key, clazz, ConfigOption.EMPTY_DESCRIPTION, Arrays.asList(values), true);
- }
-
- /**
- * Creates a ConfigOption without a default value.
- *
- * @return The config option without a default value.
- */
- public ConfigOption<List<E>> noDefaultValue() {
- return new ConfigOption<>(key, clazz, ConfigOption.EMPTY_DESCRIPTION, null, true);
+ return new ConfigOption<>(key, clazz, Description.builder().text("").build(), null);
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/options/Options.java b/paimon-common/src/main/java/org/apache/paimon/options/Options.java
index ee8a249df..4510dc8a0 100644
--- a/paimon-common/src/main/java/org/apache/paimon/options/Options.java
+++ b/paimon-common/src/main/java/org/apache/paimon/options/Options.java
@@ -98,11 +98,7 @@ public class Options implements Serializable {
Class<?> clazz = option.getClazz();
try {
- if (option.isList()) {
- return rawValue.map(v -> OptionsUtils.convertToList(v, clazz));
- } else {
- return rawValue.map(v -> OptionsUtils.convertValue(v, clazz));
- }
+ return rawValue.map(v -> OptionsUtils.convertValue(v, clazz));
} catch (Exception e) {
throw new IllegalArgumentException(
String.format(
diff --git a/paimon-common/src/main/java/org/apache/paimon/options/OptionsUtils.java b/paimon-common/src/main/java/org/apache/paimon/options/OptionsUtils.java
index 36d05ef63..47eb45007 100644
--- a/paimon-common/src/main/java/org/apache/paimon/options/OptionsUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/options/OptionsUtils.java
@@ -77,18 +77,6 @@ public class OptionsUtils {
throw new IllegalArgumentException("Unsupported type: " + clazz);
}
- @SuppressWarnings("unchecked")
- static <T> T convertToList(Object rawValue, Class<?> atomicClass) {
- if (rawValue instanceof List) {
- return (T) rawValue;
- } else {
- return (T)
- StructuredOptionsSplitter.splitEscaped(rawValue.toString(), ';').stream()
- .map(s -> convertValue(s, atomicClass))
- .collect(Collectors.toList());
- }
- }
-
@SuppressWarnings("unchecked")
static Map<String, String> convertToProperties(Object o) {
if (o instanceof Map) {
@@ -273,7 +261,7 @@ public class OptionsUtils {
* </pre>
*/
public static boolean canBePrefixMap(ConfigOption<?> configOption) {
- return configOption.getClazz() == Map.class && !configOption.isList();
+ return configOption.getClazz() == Map.class;
}
/** Filter condition for prefix map keys. */
diff --git a/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java b/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
index 209447457..bef414915 100644
--- a/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
+++ b/paimon-docs/src/main/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGenerator.java
@@ -449,49 +449,22 @@ public class ConfigOptionsDocGenerator {
}
}
- private static boolean isList(ConfigOption<?> option) {
- try {
- Method getClazzMethod = ConfigOption.class.getDeclaredMethod("isList");
- getClazzMethod.setAccessible(true);
- boolean isList = (boolean) getClazzMethod.invoke(option);
- getClazzMethod.setAccessible(false);
- return isList;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
@VisibleForTesting
static String typeToHtml(OptionWithMetaInfo optionWithMetaInfo) {
ConfigOption<?> option = optionWithMetaInfo.option;
Class<?> clazz = getClazz(option);
- boolean isList = isList(option);
if (clazz.isEnum()) {
- return enumTypeToHtml(isList);
+ return enumTypeToHtml();
}
- return atomicTypeToHtml(clazz, isList);
+ return atomicTypeToHtml(clazz);
}
- private static String atomicTypeToHtml(Class<?> clazz, boolean isList) {
- String typeName = clazz.getSimpleName();
-
- final String type;
- if (isList) {
- type = String.format("List<%s>", typeName);
- } else {
- type = typeName;
- }
- return escapeCharacters(type);
+ private static String atomicTypeToHtml(Class<?> clazz) {
+ return escapeCharacters(clazz.getSimpleName());
}
- private static String enumTypeToHtml(boolean isList) {
- final String type;
- if (isList) {
- type = "List<Enum>";
- } else {
- type = "Enum";
- }
- return String.format("<p>%s</p>", escapeCharacters(type));
+ private static String enumTypeToHtml() {
+ return String.format("<p>%s</p>", escapeCharacters("Enum"));
}
@VisibleForTesting
diff --git a/paimon-docs/src/test/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGeneratorTest.java b/paimon-docs/src/test/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGeneratorTest.java
index f46cf4c15..96d72c9aa 100644
--- a/paimon-docs/src/test/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGeneratorTest.java
+++ b/paimon-docs/src/test/java/org/apache/paimon/docs/configuration/ConfigOptionsDocGeneratorTest.java
@@ -94,13 +94,6 @@ class ConfigOptionsDocGeneratorTest {
.defaultValue(TestEnum.VALUE_1)
.withDescription("Description");
- public static ConfigOption<List<TestEnum>> enumListOption =
- ConfigOptions.key("option.enum.list")
- .enumType(TestEnum.class)
- .asList()
- .defaultValues(TestEnum.VALUE_1, TestEnum.VALUE_2)
- .withDescription("Description");
-
public static ConfigOption<DescribedTestEnum> describedEnum =
ConfigOptions.key("option.enum.described")
.enumType(DescribedTestEnum.class)
@@ -119,15 +112,6 @@ class ConfigOptionsDocGeneratorTest {
.defaultValue(Collections.singletonMap("key1", "value1"))
.withDescription("Description");
- public static ConfigOption<List<Map<String, String>>> mapListOption =
- ConfigOptions.key("option.map.list")
- .mapType()
- .asList()
- .defaultValues(
- Collections.singletonMap("key1", "value1"),
- Collections.singletonMap("key2", "value2"))
- .withDescription("Description");
-
public static ConfigOption<Duration> durationOption =
ConfigOptions.key("option.duration")
.durationType()
@@ -167,24 +151,12 @@ class ConfigOptionsDocGeneratorTest {
+ " <td>Description<br /><br />Possible values:<ul><li>\"A\": First letter of the alphabet</li><li>\"B\": Second letter of the alphabet</li></ul></td>\n"
+ " </tr>\n"
+ " <tr>\n"
- + " <td><h5>option.enum.list</h5></td>\n"
- + " <td style=\"word-wrap: break-word;\">VALUE_1;<wbr>VALUE_2</td>\n"
- + " <td><p>List<Enum></p></td>\n"
- + " <td>Description<br /><br />Possible values:<ul><li>\"VALUE_1\"</li><li>\"VALUE_2\"</li><li>\"VALUE_3\"</li></ul></td>\n"
- + " </tr>\n"
- + " <tr>\n"
+ " <td><h5>option.map</h5></td>\n"
+ " <td style=\"word-wrap: break-word;\">key1:value1</td>\n"
+ " <td>Map</td>\n"
+ " <td>Description</td>\n"
+ " </tr>\n"
+ " <tr>\n"
- + " <td><h5>option.map.list</h5></td>\n"
- + " <td style=\"word-wrap: break-word;\">key1:value1;<wbr>key2:value2</td>\n"
- + " <td>List<Map></td>\n"
- + " <td>Description</td>\n"
- + " </tr>\n"
- + " <tr>\n"
+ " <td><h5>option.memory</h5></td>\n"
+ " <td style=\"word-wrap: break-word;\">1 kb</td>\n"
+ " <td>MemorySize</td>\n"
(incubator-paimon) 04/07: [flink] throws exception if there is a SinkMaterializer before Paimon sink (#2592)
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 7d0e05055853bbaab955e7e02cb46c4223b8bce0
Author: tsreaper <ts...@gmail.com>
AuthorDate: Fri Dec 29 11:59:53 2023 +0800
[flink] throws exception if there is a SinkMaterializer before Paimon sink (#2592)
---
.../org/apache/paimon/flink/sink/FlinkSink.java | 33 ++++++++++++++++++++++
.../apache/paimon/flink/PartialUpdateITCase.java | 20 +++++++++++++
.../flink/PrimaryKeyFileStoreTableITCase.java | 6 +++-
3 files changed, 58 insertions(+), 1 deletion(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index 7d3cecdbf..f72f66885 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -25,9 +25,11 @@ import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.Preconditions;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.operators.SlotSharingGroup;
+import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.CheckpointingMode;
@@ -39,10 +41,15 @@ import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
import javax.annotation.Nullable;
import java.io.Serializable;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.Queue;
+import java.util.Set;
import java.util.UUID;
import static org.apache.flink.configuration.ClusterOptions.ENABLE_FINE_GRAINED_RESOURCE_MANAGEMENT;
@@ -139,6 +146,8 @@ public abstract class FlinkSink<T> implements Serializable {
}
public DataStreamSink<?> sinkFrom(DataStream<T> input, String initialCommitUser) {
+ assertNoSinkMaterializer(input);
+
// do the actually writing action, no snapshot generated in this stage
SingleOutputStreamOperator<Committable> written =
doWrite(input, initialCommitUser, input.getParallelism());
@@ -147,6 +156,30 @@ public abstract class FlinkSink<T> implements Serializable {
return doCommit(written, initialCommitUser);
}
+ private void assertNoSinkMaterializer(DataStream<T> input) {
+ // traverse the transformation graph with breadth first search
+ Set<Integer> visited = new HashSet<>();
+ Queue<Transformation<?>> queue = new LinkedList<>();
+ queue.add(input.getTransformation());
+ visited.add(input.getTransformation().getId());
+ while (!queue.isEmpty()) {
+ Transformation<?> transformation = queue.poll();
+ Preconditions.checkArgument(
+ !transformation.getName().startsWith("SinkMaterializer"),
+ String.format(
+ "Sink materializer must not be used with Paimon sink. "
+ + "Please set '%s' to '%s' in Flink's config.",
+ ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE.key(),
+ ExecutionConfigOptions.UpsertMaterialize.NONE.name()));
+ for (Transformation<?> prev : transformation.getInputs()) {
+ if (!visited.contains(prev.getId())) {
+ queue.add(prev);
+ visited.add(prev.getId());
+ }
+ }
+ }
+ }
+
public SingleOutputStreamOperator<Committable> doWrite(
DataStream<T> input, String commitUser, Integer parallelism) {
StreamExecutionEnvironment env = input.getExecutionEnvironment();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
index c8e336384..c8172131e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
@@ -37,6 +37,7 @@ import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.fail;
/** ITCase for partial update. */
public class PartialUpdateITCase extends CatalogITCaseBase {
@@ -327,4 +328,23 @@ public class PartialUpdateITCase extends CatalogITCaseBase {
assertThat(sql("SELECT * FROM AGG")).containsExactlyInAnyOrder(Row.of(1, 0, 2));
}
+
+ @Test
+ public void testNoSinkMaterializer() {
+ sEnv.getConfig()
+ .set(
+ ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
+ ExecutionConfigOptions.UpsertMaterialize.FORCE);
+ try (CloseableIterator<Row> ignored =
+ streamSqlIter(
+ "INSERT INTO dwd_orders "
+ + "SELECT OrderID, OrderNumber, PersonID, CAST(NULL AS STRING), CAST(NULL AS STRING), CAST(NULL AS INT) FROM ods_orders "
+ + "UNION ALL "
+ + "SELECT OrderID, CAST(NULL AS INT), dim_persons.PersonID, LastName, FirstName, Age FROM dim_persons JOIN ods_orders ON dim_persons.PersonID = ods_orders.PersonID;")) {
+ fail("Expecting exception");
+ } catch (Exception e) {
+ assertThat(e)
+ .hasMessageContaining("Sink materializer must not be used with Paimon sink.");
+ }
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
index eba3fae70..8b4f452ef 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
@@ -88,6 +88,10 @@ public class PrimaryKeyFileStoreTableITCase extends AbstractTestBase {
sEnv.getConfig()
.getConfiguration()
.set(CHECKPOINTING_INTERVAL, Duration.ofMillis(checkpointIntervalMs));
+ sEnv.getConfig()
+ .set(
+ ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE,
+ ExecutionConfigOptions.UpsertMaterialize.NONE);
return sEnv;
}
@@ -239,7 +243,7 @@ public class PrimaryKeyFileStoreTableITCase extends AbstractTestBase {
// write update data
sEnv.executeSql(
"INSERT INTO `default_catalog`.`default_database`.`S` "
- + "VALUES (1, 'A'), (1, 'B'), (1, 'C'), (1, 'D')")
+ + "VALUES (1, 'D'), (1, 'C'), (1, 'B'), (1, 'A')")
.await();
// read update data