This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a3fc1b79ca [spark] Remove upsert-key support (#9093)
a3fc1b79ca is described below
commit a3fc1b79ca8076d05424d61ef57b88ac0e994332
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Aug 7 16:28:06 2026 +0800
[spark] Remove upsert-key support (#9093)
---
docs/docs/spark/sql-upsert.md | 93 ---------------
docs/generated/core_configuration.html | 6 -
docs/redirects.js | 4 -
docs/sidebars.js | 1 -
.../main/java/org/apache/paimon/CoreOptions.java | 15 ---
.../org/apache/paimon/schema/SchemaValidation.java | 8 --
.../apache/paimon/schema/SchemaValidationTest.java | 19 +--
.../paimon/spark/sql/RewriteUpsertTableTest.scala | 21 ----
.../paimon/spark/sql/RewriteUpsertTableTest.scala | 21 ----
.../paimon/spark/sql/RewriteUpsertTableTest.scala | 21 ----
.../paimon/spark/sql/RewriteUpsertTableTest.scala | 21 ----
.../catalyst/analysis/RewriteUpsertTable.scala | 129 --------------------
.../paimon/spark/sql/RewriteUpsertTableTest.scala | 21 ----
.../paimon/spark/sql/RewriteUpsertTableTest.scala | 21 ----
.../catalyst/analysis/RewriteUpsertTable.scala | 130 ---------------------
.../extensions/PaimonSparkSessionExtensions.scala | 4 +-
.../spark/sql/RewriteUpsertTableTestBase.scala | 89 --------------
17 files changed, 4 insertions(+), 620 deletions(-)
diff --git a/docs/docs/spark/sql-upsert.md b/docs/docs/spark/sql-upsert.md
deleted file mode 100644
index 551d7988a6..0000000000
--- a/docs/docs/spark/sql-upsert.md
+++ /dev/null
@@ -1,93 +0,0 @@
----
-title: "SQL Upsert"
-sidebar_position: 10
----
-
-<!--
-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.
--->
-
-# SQL Upsert
-
-For table without primary key, Paimon supports upsert write mode: If the row
with the same upsert key already exists, perform update; otherwise, perform
insert.
-
-## Usage
-
-Specify the following table properties when creating the table
-
-* `upsert-key`: Defines the key columns used for upsert, cannot be used
together with primary key.
-Unlike primary key, the upsert key value can be `null`, and null-equality
matching is supported.
-Multiple columns separated by commas.
-
-* `sequence.field` (optional): When new record share the same upsert key, the
row with the larger `sequence.field` value is kept as the merge result.
-And it will also deduplicate the data being written.
-If `sequence.field` is not set, new record share the same upsert key simply
update the existing one and no deduplication is performed.
-Multiple columns separated by commas.
-
-## Example
-
-Create table:
-
-```sql
-CREATE TABLE t (k1 INT, k2 INT, ts1 INT, ts2 INT, v STRING)
-TBLPROPERTIES ('upsert-key' = 'k1,k2', 'sequence.field' = 'ts1,ts2')
-```
-
-Insert data1:
-
-```sql
-INSERT INTO t values
-(null, null, 2, 1, 'v1'),
-(null, null, 2, 2, 'v4'),
-(1, null, 1, 1, 'v1'),
-(1, 2, 1, 1, 'v1'),
-(1, 2, 2, 1, 'v2')
-```
-
-Query result:
-
-```sql
-SELECT * FROM t ORDER BY k1, k2
-
--- null, null, 2, 2, "v4"
--- 1, null, 1, 1, "v1"
--- 1, 2, 2, 1, "v2"
-```
-
-Insert data2:
-
-```sql
-INSERT INTO t values
-(null, null, 2, 1, 'v5'),
-(null, 1, 1, 1, 'v1'),
-(1, null, 2, 1, 'v2'),
-(1, 1, 1, 1, 'v1'),
-(1, 2, 2, 0, 'v3')
-```
-
-Query result:
-
-```sql
-SELECT * FROM t ORDER BY k1, k2
-
--- null, null, 2, 2, "v4"
--- null, 1, 1, 1, "v1"
--- 1, null, 2, 1, "v2"
--- 1, 1, 1, 1, "v1"
--- 1, 2, 2, 1, "v2"
-```
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index a7c8114dd8..0e399478c8 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1771,12 +1771,6 @@ If the data size allocated for the sorting task is
uneven,which may lead to perf
<td><p>Enum</p></td>
<td>Type of the table.<br /><br />Possible values:<ul><li>"table":
Normal Paimon table.</li><li>"format-table": A file format table refers to a
directory that contains multiple files of the same
format.</li><li>"materialized-table": A materialized table combines normal
Paimon table and materialized SQL.</li><li>"object-table": An object table
combines normal Paimon table and object location.</li><li>"lance-table": A
lance table, see 'https://lancedb.github.io/lance/'.</li><li [...]
</tr>
- <tr>
- <td><h5>upsert-key</h5></td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>String</td>
- <td>Define upsert key to do MERGE INTO when executing INSERT INTO,
cannot be defined with primary key.</td>
- </tr>
<tr>
<td><h5>variant.inferShreddingSchema</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/docs/redirects.js b/docs/redirects.js
index 57077ff895..9d04563c4d 100644
--- a/docs/redirects.js
+++ b/docs/redirects.js
@@ -476,10 +476,6 @@ module.exports = [
"from": "/spark/sql-query.html",
"to": "/spark/sql-query"
},
- {
- "from": "/spark/sql-upsert.html",
- "to": "/spark/sql-upsert"
- },
{
"from": "/spark/sql-write.html",
"to": "/spark/sql-write"
diff --git a/docs/sidebars.js b/docs/sidebars.js
index 56a9c3e211..be7294db7b 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -175,7 +175,6 @@ const sidebars = {
"spark/auxiliary",
"spark/default-value",
"spark/dataframe",
- "spark/sql-upsert",
"spark/structured-streaming",
"spark/procedures"
]
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 3be4d072e1..0ad4beb001 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -595,13 +595,6 @@ public class CoreOptions implements Serializable {
+ " skipped. Set to a larger value to
allow more aggressive"
+ " sort rewriting. The cap only limits
the sorted rewrite portion and full/minor cleanup may still happen beyond it.");
- public static final ConfigOption<String> UPSERT_KEY =
- key("upsert-key")
- .stringType()
- .noDefaultValue()
- .withDescription(
- "Define upsert key to do MERGE INTO when executing
INSERT INTO, cannot be defined with primary key.");
-
public static final ConfigOption<String> PARTITION_DEFAULT_NAME =
key("partition.default-name")
.stringType()
@@ -3152,14 +3145,6 @@ public class CoreOptions implements Serializable {
return options.get(FIELDS_DEFAULT_AGG_FUNC);
}
- public List<String> upsertKey() {
- String upsertKey = options.get(UPSERT_KEY);
- if (StringUtils.isEmpty(upsertKey)) {
- return Collections.emptyList();
- }
- return Arrays.asList(upsertKey.split(","));
- }
-
public static String createCommitUser(Options options) {
String commitUserPrefix = options.get(COMMIT_USER_PREFIX);
return commitUserPrefix == null
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 0e57e364b5..fa28403e82 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -144,14 +144,6 @@ public class SchemaValidation {
validateOnlyContainPrimitiveType(schema.fields(),
schema.primaryKeys(), "primary key");
validateOnlyContainPrimitiveType(schema.fields(),
schema.partitionKeys(), "partition");
- validateOnlyContainPrimitiveType(schema.fields(), options.upsertKey(),
"upsert key");
-
- if (!options.upsertKey().isEmpty() && !schema.primaryKeys().isEmpty())
{
- throw new RuntimeException(
- String.format(
- "Cannot define 'upsert-key' %s with 'primary-key'
%s.",
- options.upsertKey(), schema.primaryKeys()));
- }
validateBucket(schema, options);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index 4d3a71a4f2..e0f141680a 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -1251,7 +1251,7 @@ class SchemaValidationTest {
assertThatThrownBy(
() ->
validateTableSchema(
- vectorTypeSchema(emptyList(),
singletonList("f1"), null)))
+ vectorTypeSchema(emptyList(),
singletonList("f1"))))
.isInstanceOf(UnsupportedOperationException.class)
.hasMessage(
"The type %s in primary key field %s is unsupported",
"VectorType", "f1");
@@ -1259,18 +1259,9 @@ class SchemaValidationTest {
assertThatThrownBy(
() ->
validateTableSchema(
- vectorTypeSchema(singletonList("f1"),
emptyList(), null)))
+ vectorTypeSchema(singletonList("f1"),
emptyList())))
.isInstanceOf(UnsupportedOperationException.class)
.hasMessage("The type %s in partition field %s is
unsupported", "VectorType", "f1");
-
- assertThatThrownBy(
- () ->
- validateTableSchema(
- vectorTypeSchema(
- emptyList(), emptyList(),
singletonList("f1"))))
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessage(
- "The type %s in upsert key field %s is unsupported",
"VectorType", "f1");
}
@Test
@@ -1881,17 +1872,13 @@ class SchemaValidationTest {
new DataField(2, "payload", payloadType));
}
- private TableSchema vectorTypeSchema(
- List<String> partitionKeys, List<String> primaryKeys, List<String>
upsertKeys) {
+ private TableSchema vectorTypeSchema(List<String> partitionKeys,
List<String> primaryKeys) {
List<DataField> fields =
Arrays.asList(
new DataField(0, "f0", DataTypes.INT()),
new DataField(1, "f1", DataTypes.VECTOR(3,
DataTypes.FLOAT())));
Map<String, String> options = new HashMap<>();
options.put(BUCKET.key(), String.valueOf(-1));
- if (upsertKeys != null) {
- options.put(CoreOptions.UPSERT_KEY.key(), String.join(",",
upsertKeys));
- }
return new TableSchema(1, fields, 10, partitionKeys, primaryKeys,
options, "");
}
}
diff --git
a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
deleted file mode 100644
index 412aa3b303..0000000000
---
a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
+++ /dev/null
@@ -1,21 +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
-
-class RewriteUpsertTableTest extends RewriteUpsertTableTestBase {}
diff --git
a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
deleted file mode 100644
index 412aa3b303..0000000000
---
a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
+++ /dev/null
@@ -1,21 +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
-
-class RewriteUpsertTableTest extends RewriteUpsertTableTestBase {}
diff --git
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
deleted file mode 100644
index 412aa3b303..0000000000
---
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
+++ /dev/null
@@ -1,21 +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
-
-class RewriteUpsertTableTest extends RewriteUpsertTableTestBase {}
diff --git
a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
deleted file mode 100644
index 412aa3b303..0000000000
---
a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
+++ /dev/null
@@ -1,21 +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
-
-class RewriteUpsertTableTest extends RewriteUpsertTableTestBase {}
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RewriteUpsertTable.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RewriteUpsertTable.scala
deleted file mode 100644
index fd8776d6db..0000000000
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RewriteUpsertTable.scala
+++ /dev/null
@@ -1,129 +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.catalyst.analysis
-
-import org.apache.paimon.spark.SparkTable
-import org.apache.paimon.table.FileStoreTable
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute,
UnresolvedStar}
-import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
CurrentRow, Descending, EqualNullSafe, EqualTo, LessThanOrEqual, Literal,
RowFrame, RowNumber, SortOrder, SpecifiedWindowFrame, UnboundedPreceding,
WindowExpression, WindowSpecDefinition}
-import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
-import org.apache.spark.sql.paimon.shims.SparkShimLoader
-
-import scala.collection.JavaConverters._
-
-/** Rewrite upsert table to merge into. */
-case class RewriteUpsertTable(spark: SparkSession) extends Rule[LogicalPlan] {
-
- override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsDown {
- case p @ PaimonV2WriteCommand(table) =>
- val (usingUpsert, upsertKey, sequenceField) = usingUpsertTable(table)
- if (!usingUpsert) {
- return p
- }
-
- p match {
- case AppendData(target, source, _, _, _, _) =>
- val deduplicatedSource = if (sequenceField.nonEmpty) {
- deduplicateBySequenceField(source, upsertKey, sequenceField)
- } else {
- source
- }
-
- rewriteToMergeInto(target, deduplicatedSource, upsertKey,
sequenceField)
- case _ => p
- }
- }
-
- private def usingUpsertTable(table: DataSourceV2Relation): (Boolean,
Seq[String], Seq[String]) = {
- table.table match {
- case SparkTable(fileStoreTable: FileStoreTable) =>
- val coreOptions = fileStoreTable.coreOptions()
- val upsertKey = coreOptions.upsertKey().asScala.toSeq
- val sequenceField = coreOptions.sequenceField().asScala.toSeq
- if (fileStoreTable.primaryKeys().isEmpty && upsertKey.nonEmpty) {
- (true, upsertKey, sequenceField)
- } else {
- (false, Seq.empty, Seq.empty)
- }
- case _ => (false, Seq.empty, Seq.empty)
- }
- }
-
- private def deduplicateBySequenceField(
- source: LogicalPlan,
- upsertKey: Seq[String],
- sequenceField: Seq[String]): LogicalPlan = {
- val winSpec = WindowSpecDefinition(
- cols(source.output, upsertKey),
- cols(source.output, sequenceField).map(SortOrder(_, Descending)),
- SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow)
- )
- val rnAlias = Alias(WindowExpression(RowNumber(), winSpec), "__rn__")()
- val withRN = Project(UnresolvedStar(None) :: rnAlias :: Nil, source)
- val filtered = Filter(EqualTo(UnresolvedAttribute("__rn__"), Literal(1)),
withRN)
- Project(source.output, filtered)
- }
-
- private def rewriteToMergeInto(
- target: LogicalPlan,
- source: LogicalPlan,
- upsertKey: Seq[String],
- sequenceField: Seq[String]
- ): MergeIntoTable = {
- val mergeCondition = upsertKey
- .map(k => EqualNullSafe(col(target.output, k), col(source.output, k)))
- .reduce(And)
-
- val updateCondiction = if (sequenceField.nonEmpty) {
- Option.apply(
- sequenceField
- .map(s => LessThanOrEqual(col(target.output, s), col(source.output,
s)))
- .reduce(And))
- } else {
- Option.empty
- }
-
- val assignments: Seq[Assignment] =
- target.output.zip(source.output).map(a => Assignment(a._1, a._2))
-
- val mergeActions = Seq(UpdateAction(updateCondiction, assignments))
- val notMatchedActions = Seq(InsertAction(None, assignments))
-
- SparkShimLoader.shim.createMergeIntoTable(
- target,
- source,
- mergeCondition,
- mergeActions,
- notMatchedActions,
- Seq.empty,
- withSchemaEvolution = false)
- }
-
- private def cols(input: Seq[Attribute], colsNames: Seq[String]):
Seq[Attribute] = {
- colsNames.map(c => col(input, c))
- }
-
- private def col(input: Seq[Attribute], colsName: String): Attribute = {
- input.find(_.name == colsName).get
- }
-}
diff --git
a/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
deleted file mode 100644
index 412aa3b303..0000000000
---
a/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
+++ /dev/null
@@ -1,21 +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
-
-class RewriteUpsertTableTest extends RewriteUpsertTableTestBase {}
diff --git
a/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
deleted file mode 100644
index 412aa3b303..0000000000
---
a/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTest.scala
+++ /dev/null
@@ -1,21 +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
-
-class RewriteUpsertTableTest extends RewriteUpsertTableTestBase {}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RewriteUpsertTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RewriteUpsertTable.scala
deleted file mode 100644
index ba6d0f8efa..0000000000
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RewriteUpsertTable.scala
+++ /dev/null
@@ -1,130 +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.catalyst.analysis
-
-import org.apache.paimon.spark.SparkTable
-import org.apache.paimon.table.FileStoreTable
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute,
UnresolvedStar}
-import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
CurrentRow, Descending, EqualNullSafe, EqualTo, LessThanOrEqual, Literal,
RowFrame, RowNumber, SortOrder, SpecifiedWindowFrame, UnboundedPreceding,
WindowExpression, WindowSpecDefinition}
-import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
-import org.apache.spark.sql.paimon.shims.SparkShimLoader
-
-import scala.collection.JavaConverters._
-
-/** Rewrite upsert table to merge into. */
-case class RewriteUpsertTable(spark: SparkSession) extends Rule[LogicalPlan] {
-
- override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsDown {
- case p @ PaimonV2WriteCommand(table) =>
- val (usingUpsert, upsertKey, sequenceField) = usingUpsertTable(table)
- if (!usingUpsert) {
- return p
- }
-
- p match {
- case AppendData(target, source, _, _, _, _) =>
- val deduplicatedSource = if (sequenceField.nonEmpty) {
- deduplicateBySequenceField(source, upsertKey, sequenceField)
- } else {
- source
- }
-
- rewriteToMergeInto(target, deduplicatedSource, upsertKey,
sequenceField)
- case _ => p
- }
- }
-
- private def usingUpsertTable(table: DataSourceV2Relation): (Boolean,
Seq[String], Seq[String]) = {
- table.table match {
- case SparkTable(fileStoreTable: FileStoreTable) =>
- val coreOptions = fileStoreTable.coreOptions()
- val upsertKey = coreOptions.upsertKey().asScala.toSeq
- val sequenceField = coreOptions.sequenceField().asScala.toSeq
- if (fileStoreTable.primaryKeys().isEmpty && upsertKey.nonEmpty) {
- (true, upsertKey, sequenceField)
- } else {
- (false, Seq.empty, Seq.empty)
- }
- case _ => (false, Seq.empty, Seq.empty)
- }
- }
-
- private def deduplicateBySequenceField(
- source: LogicalPlan,
- upsertKey: Seq[String],
- sequenceField: Seq[String]): LogicalPlan = {
- val winSpec = WindowSpecDefinition(
- cols(source.output, upsertKey),
- cols(source.output, sequenceField).map(SortOrder(_, Descending)),
- SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow)
- )
- val rnAlias = Alias(WindowExpression(RowNumber(), winSpec), "__rn__")()
- val withRN = Project(UnresolvedStar(None) :: rnAlias :: Nil, source)
- val filtered = Filter(EqualTo(UnresolvedAttribute("__rn__"), Literal(1)),
withRN)
- Project(source.output, filtered)
- }
-
- private def rewriteToMergeInto(
- target: LogicalPlan,
- source: LogicalPlan,
- upsertKey: Seq[String],
- sequenceField: Seq[String]
- ): MergeIntoTable = {
- val mergeCondition = upsertKey
- .map(k => EqualNullSafe(col(target.output, k), col(source.output, k)))
- .reduce(And)
-
- val updateCondiction = if (sequenceField.nonEmpty) {
- Option.apply(
- sequenceField
- .map(s => LessThanOrEqual(col(target.output, s), col(source.output,
s)))
- .reduce(And))
- } else {
- Option.empty
- }
-
- val assignments: Seq[Assignment] =
- target.output.zip(source.output).map(a => Assignment(a._1, a._2))
-
- val shim = SparkShimLoader.shim
- val mergeActions = Seq(shim.createUpdateAction(updateCondiction,
assignments))
- val notMatchedActions = Seq(shim.createInsertAction(None, assignments))
-
- SparkShimLoader.shim.createMergeIntoTable(
- target,
- source,
- mergeCondition,
- mergeActions,
- notMatchedActions,
- Seq.empty,
- withSchemaEvolution = false)
- }
-
- private def cols(input: Seq[Attribute], colsNames: Seq[String]):
Seq[Attribute] = {
- colsNames.map(c => col(input, c))
- }
-
- private def col(input: Seq[Attribute], colsName: String): Attribute = {
- input.find(_.name == colsName).get
- }
-}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
index e3701a468a..f856d348a4 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
@@ -18,7 +18,7 @@
package org.apache.paimon.spark.extensions
-import org.apache.paimon.spark.catalyst.analysis.{PaimonAnalysis,
PaimonDeleteTable, PaimonFunctionResolver, PaimonIncompatibleResolutionRules,
PaimonMergeInto, PaimonPostHocResolutionRules, PaimonProcedureResolver,
PaimonUpdateTable, PaimonViewResolver, ReplacePaimonFunctions,
RewriteUpsertTable}
+import org.apache.paimon.spark.catalyst.analysis.{PaimonAnalysis,
PaimonDeleteTable, PaimonFunctionResolver, PaimonIncompatibleResolutionRules,
PaimonMergeInto, PaimonPostHocResolutionRules, PaimonProcedureResolver,
PaimonUpdateTable, PaimonViewResolver, ReplacePaimonFunctions}
import
org.apache.paimon.spark.catalyst.optimizer.{MergePaimonScalarSubqueries,
OptimizeMetadataOnlyDeleteFromPaimonTable, PushDownLateralVectorSearchFilter,
RepartitionLateralVectorSearchInput}
import
org.apache.paimon.spark.catalyst.plans.logical.PaimonTableValuedFunctions
import org.apache.paimon.spark.commands.BucketExpression
@@ -44,8 +44,6 @@ class PaimonSparkSessionExtensions extends
(SparkSessionExtensions => Unit) {
extensions.injectResolutionRule(spark => PaimonFunctionResolver(spark))
extensions.injectResolutionRule(spark =>
SparkShimLoader.shim.createCustomResolution(spark))
extensions.injectResolutionRule(spark =>
PaimonIncompatibleResolutionRules(spark))
- extensions.injectResolutionRule(spark => RewriteUpsertTable(spark))
-
extensions.injectPostHocResolutionRule(spark =>
ReplacePaimonFunctions(spark))
extensions.injectPostHocResolutionRule(spark =>
PaimonPostHocResolutionRules(spark))
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTestBase.scala
deleted file mode 100644
index 7fb2223a98..0000000000
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RewriteUpsertTableTestBase.scala
+++ /dev/null
@@ -1,89 +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.apache.spark.sql.Row
-
-abstract class RewriteUpsertTableTestBase extends PaimonSparkTestBase {
-
- test("Rewrite Upsert Table: cannot define with primary key") {
- assert(intercept[Exception] {
- sql("""
- |CREATE TABLE T (k INT, a INT, b STRING)
- |TBLPROPERTIES ('upsert-key' = 'k', 'primary-key' = 'k')
- |""".stripMargin)
- }.getMessage.contains("Cannot define 'upsert-key' [k] with 'primary-key'
[k]."))
- }
-
- test("Rewrite Upsert Table: rewrite insert without sequence field") {
- sql("""
- |CREATE TABLE T (k INT, a INT, b STRING)
- |TBLPROPERTIES ('upsert-key' = 'k')
- |""".stripMargin)
-
- sql("INSERT INTO T values (1, 1, 'c1'), (null, 2, 'c2'), (3, 3, 'c3')")
- sql("INSERT INTO T values (null, 22, 'c22'), (1, 11, 'c11'), (4, 4, 'c4')")
-
- checkAnswer(
- sql("SELECT * FROM T ORDER BY k"),
- Seq(Row(null, 22, "c22"), Row(1, 11, "c11"), Row(3, 3, "c3"), Row(4, 4,
"c4")))
- }
-
- test("Rewrite Upsert Table: rewrite insert with sequence field") {
- sql("""
- |CREATE TABLE T (k1 INT, k2 INT, ts1 INT, ts2 INT, c STRING)
- |TBLPROPERTIES ('upsert-key' = 'k1,k2', 'sequence.field' = 'ts1,ts2')
- |""".stripMargin)
-
- // test insert deduplicate
- sql("""
- |INSERT INTO T values
- |(null, null, 2, 1, 'v3'),
- |(null, null, 2, 2, 'v4'),
- |(null, null, 1, 1, 'v1'),
- |(null, null, 1, 2, 'v2'),
- |(1, null, 1, 1, 'v1'),
- |(1, 2, 1, 1, 'v1'),
- |(1, 2, 2, 1, 'v2')
- |""".stripMargin)
- checkAnswer(
- sql("SELECT * FROM T ORDER BY k1, k2"),
- Seq(Row(null, null, 2, 2, "v4"), Row(1, null, 1, 1, "v1"), Row(1, 2, 2,
1, "v2")))
-
- // test inset with different sequence field
- sql("""
- |INSERT INTO T values
- |(null, null, 2, 1, 'v44'),
- |(1, null, 2, 1, 'v2'),
- |(null, 1, 1, 1, 'v1'),
- |(1, 2, 2, 2, 'v3'),
- |(1, 1, 1, 1, 'v1')
- |""".stripMargin)
- checkAnswer(
- sql("SELECT * FROM T ORDER BY k1, k2"),
- Seq(
- Row(null, null, 2, 2, "v4"),
- Row(null, 1, 1, 1, "v1"),
- Row(1, null, 2, 1, "v2"),
- Row(1, 1, 1, 1, "v1"),
- Row(1, 2, 2, 2, "v3")))
- }
-}
