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 d435ec9b05 [spark] adapt data evolution merge into for tables with
existing deletion vectors (#8431)
d435ec9b05 is described below
commit d435ec9b050b84e7da769d148bb8d7218fd994dc
Author: Faiz <[email protected]>
AuthorDate: Thu Jul 2 23:34:34 2026 +0800
[spark] adapt data evolution merge into for tables with existing deletion
vectors (#8431)
If call merge into on some data evolution tables with existing deletion
vectors, there may exist some row id gaps of the merged output. Like:
original table values:
```text
(1, "a"), (2, "b"), (3, "c")
```
After removing the second row, then merge into update with some source
```text
(1, "new_a")
```
The merged result would be
```text
_ROW_ID, old_val, new_val
(0, "a", "new_a")
(2, "c", null)
```
At that case, we should fill the gap of deleted value.
---
.../spark/sql/DataEvolutionDeletionTest.scala | 21 +++
.../spark/sql/DataEvolutionDeletionTest.scala | 21 +++
.../spark/sql/DataEvolutionDeletionTest.scala | 21 +++
.../spark/sql/DataEvolutionDeletionTest.scala | 33 ++++
.../spark/sql/DataEvolutionDeletionTest.scala | 33 ++++
.../spark/sql/DataEvolutionDeletionTest.scala | 33 ++++
.../spark/write/DataEvolutionTableDataWrite.scala | 54 +++++-
.../spark/sql/DataEvolutionDeletionTestBase.scala | 202 +++++++++++++++++++++
.../paimon/spark/sql/DeleteFromTableTestBase.scala | 46 -----
9 files changed, 416 insertions(+), 48 deletions(-)
diff --git
a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
new file mode 100644
index 0000000000..eb40890dae
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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 DataEvolutionDeletionTest extends DataEvolutionDeletionTestBase {}
diff --git
a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
new file mode 100644
index 0000000000..eb40890dae
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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 DataEvolutionDeletionTest extends DataEvolutionDeletionTestBase {}
diff --git
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
new file mode 100644
index 0000000000..eb40890dae
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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 DataEvolutionDeletionTest extends DataEvolutionDeletionTestBase {}
diff --git
a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
new file mode 100644
index 0000000000..43aeb915db
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.spark.SparkConf
+
+class DataEvolutionDeletionTest extends DataEvolutionDeletionTestBase {
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf.set("spark.paimon.write.use-v2-write", "false")
+ }
+}
+
+class V2DataEvolutionDeletionTest extends DataEvolutionDeletionTestBase {
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf.set("spark.paimon.write.use-v2-write", "true")
+ }
+}
diff --git
a/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
new file mode 100644
index 0000000000..43aeb915db
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.spark.SparkConf
+
+class DataEvolutionDeletionTest extends DataEvolutionDeletionTestBase {
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf.set("spark.paimon.write.use-v2-write", "false")
+ }
+}
+
+class V2DataEvolutionDeletionTest extends DataEvolutionDeletionTestBase {
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf.set("spark.paimon.write.use-v2-write", "true")
+ }
+}
diff --git
a/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
new file mode 100644
index 0000000000..43aeb915db
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTest.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.spark.SparkConf
+
+class DataEvolutionDeletionTest extends DataEvolutionDeletionTestBase {
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf.set("spark.paimon.write.use-v2-write", "false")
+ }
+}
+
+class V2DataEvolutionDeletionTest extends DataEvolutionDeletionTestBase {
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf.set("spark.paimon.write.use-v2-write", "true")
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataEvolutionTableDataWrite.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataEvolutionTableDataWrite.scala
index 50f2f691b0..b962f7ea32 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataEvolutionTableDataWrite.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataEvolutionTableDataWrite.scala
@@ -21,6 +21,7 @@ package org.apache.paimon.spark.write
import org.apache.paimon.casting.FallbackMappingRow
import org.apache.paimon.catalog.CatalogContext
import org.apache.paimon.data.{BinaryRow, BlobPlaceholder, GenericRow,
InternalRow}
+import org.apache.paimon.data.serializer.InternalSerializers
import org.apache.paimon.disk.IOManager
import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
@@ -34,6 +35,7 @@ import org.apache.paimon.utils.RecordWriter
import org.apache.paimon.utils.SerializationUtils
import org.apache.spark.sql.Row
+import org.slf4j.LoggerFactory
import java.util.Collections
@@ -58,7 +60,9 @@ case class DataEvolutionTableDataWrite(
private val toPaimonRow = {
SparkRowUtils.toPaimonRow(writeType, -1, catalogContext)
}
+ private lazy val rowSerializer = InternalSerializers.create(writeType)
private val rawBlobFallbackFields =
rawBlobPlaceholderMarkerIndexes.toSeq.sortBy(_._1).toArray
+ private val rawBlobFallbackFieldIndexes = rawBlobFallbackFields.map(_._1)
private val rawBlobFallbackMappings = {
val mappings = Array.fill(writeType.getFieldCount)(-1)
rawBlobFallbackFields.zipWithIndex.foreach {
@@ -148,23 +152,43 @@ case class DataEvolutionTableDataWrite(
recordWriter: RecordWriter[InternalRow],
numRecords: Long) {
- private var numWritten = 0
+ private var numWritten = 0L
+ private var fillerRows = 0L
+ private var fillerRow: InternalRow = _
def matchFirstRowId(firstRowId: Long): Boolean = {
this.firstRowId == firstRowId
}
def write(row: InternalRow, rowId: Long): Unit = {
- assert(rowId == firstRowId + numWritten, "Row ID does not match
expected.")
+ assert(
+ rowId >= firstRowId + numWritten,
+ s"Row ID should be incremental. Expected at least ${firstRowId +
numWritten}, but got $rowId.")
+ assert(
+ rowId < firstRowId + numRecords,
+ s"Row ID $rowId is out of range [$firstRowId, ${firstRowId +
numRecords}).")
+
+ // For tables with existing deletion vectors, there may be some row id
gaps.
+ // We can simply pad any valid values since these rows will never be
exposed.
+ fillGapUntil(rowId, row)
+
numWritten += 1
recordWriter.write(row)
}
def finish(): Seq[CommitMessageImpl] = {
try {
+ fillGapUntil(firstRowId + numRecords)
+
assert(
numRecords == numWritten,
s"Number of written records $numWritten does not match expected
number $numRecords for first row ID $firstRowId.")
+ if (fillerRows > 0) {
+ DataEvolutionTableDataWrite.LOG.warn(
+ s"Data evolution merge wrote $fillerRows filler rows out of
$numRecords rows " +
+ s"for row-id range [$firstRowId, ${firstRowId + numRecords}) to
preserve " +
+ "row-id continuity. Raw blob fields in filler rows are written
as NULL.")
+ }
val result = recordWriter.prepareCommit(false)
val dataFiles = result.newFilesIncrement().newFiles()
val dataFileMetas = assignFirstRowIds(dataFiles.asScala.toSeq)
@@ -184,6 +208,28 @@ case class DataEvolutionTableDataWrite(
}
}
+ private def fillGapUntil(rowId: Long, fillerSourceRow: InternalRow =
null): Unit = {
+ if (fillerRow == null && fillerSourceRow != null) {
+ // Copy the first record this file writer met to minimize the
influences on
+ // file stats, but keep raw blob fields as NULLs so filler rows do not
trigger
+ // blob fallback.
+ val copied = rowSerializer
+ .copyRowData(fillerSourceRow, new
GenericRow(writeType.getFieldCount))
+ .asInstanceOf[GenericRow]
+ rawBlobFallbackFieldIndexes.foreach(copied.setField(_, null))
+ fillerRow = copied
+ }
+
+ assert(
+ fillerRow != null || firstRowId + numWritten == rowId,
+ s"Cannot fill row ID gaps before any real row for first row ID
$firstRowId.")
+ while (firstRowId + numWritten < rowId) {
+ recordWriter.write(fillerRow)
+ numWritten += 1
+ fillerRows += 1
+ }
+ }
+
private def assignFirstRowIds(dataFiles:
Seq[org.apache.paimon.io.DataFileMeta])
: Seq[org.apache.paimon.io.DataFileMeta] = {
val assigned = ListBuffer[org.apache.paimon.io.DataFileMeta]()
@@ -223,3 +269,7 @@ case class DataEvolutionTableDataWrite(
}
}
}
+
+object DataEvolutionTableDataWrite {
+ private val LOG =
LoggerFactory.getLogger(classOf[DataEvolutionTableDataWrite])
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTestBase.scala
new file mode 100644
index 0000000000..5d2ba1d697
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTestBase.scala
@@ -0,0 +1,202 @@
+/*
+ * 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 DataEvolutionDeletionTestBase extends PaimonSparkTestBase {
+
+ test("Data Evolution deletion: delete from table with deletion vectors") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 5)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(5, 10)")
+ sql("ALTER TABLE t ADD COLUMNS (d INT)")
+ sql(
+ "INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c, id +
100 AS d FROM range(10, 13)")
+
+ sql("DELETE FROM t WHERE id IN (1, 4, 6, 11)")
+ checkAnswer(
+ sql("SELECT *, _ROW_ID FROM t ORDER BY id"),
+ Seq(
+ Row(0, 0, 0, null, 0L),
+ Row(2, 2, 2, null, 2L),
+ Row(3, 3, 3, null, 3L),
+ Row(5, 5, 5, null, 5L),
+ Row(7, 7, 7, null, 7L),
+ Row(8, 8, 8, null, 8L),
+ Row(9, 9, 9, null, 9L),
+ Row(10, 10, 10, 110, 10L),
+ Row(12, 12, 12, 112, 12L)
+ )
+ )
+
+ sql("DELETE FROM t WHERE id IN (2, 8)")
+ checkAnswer(
+ sql("SELECT *, _ROW_ID FROM t ORDER BY id"),
+ Seq(
+ Row(0, 0, 0, null, 0L),
+ Row(3, 3, 3, null, 3L),
+ Row(5, 5, 5, null, 5L),
+ Row(7, 7, 7, null, 7L),
+ Row(9, 9, 9, null, 9L),
+ Row(10, 10, 10, 110, 10L),
+ Row(12, 12, 12, 112, 12L))
+ )
+ }
+ }
+
+ test("Data Evolution deletion: global index query skips deleted rows") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (id INT, name STRING, b INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'global-index.search-mode' = 'full',
+ | 'btree-index.records-per-range' = '1000')
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO t VALUES
+ | (1, 'name-1', 10),
+ | (2, 'name-2', 20),
+ | (3, 'name-3', 30),
+ | (4, 'name-4', 40)
+ |""".stripMargin)
+ sql(
+ "CALL sys.create_global_index(table => 'test.t', index_column =>
'name', " +
+ "index_type => 'btree')")
+
+ sql("DELETE FROM t WHERE id IN (2, 4)")
+
+ checkAnswer(
+ sql("SELECT id, name, b FROM t WHERE name IN ('name-2', 'name-4')
ORDER BY id"),
+ Seq.empty[Row])
+ checkAnswer(
+ sql("SELECT id, name, b FROM t WHERE name IN ('name-1', 'name-3')
ORDER BY id"),
+ Seq(Row(1, "name-1", 10), Row(3, "name-3", 30)))
+ }
+ }
+
+ test("Data Evolution deletion: merge update after file-level and partial
deletion") {
+ withTable("s", "t") {
+ sql("CREATE TABLE s (id INT, new_b INT)")
+ sql("INSERT INTO s VALUES (2, 200), (6, 600), (7, 700), (9, 900)")
+
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 5)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(5, 10)")
+ sql("DELETE FROM t WHERE id IN (0, 1, 2, 3, 4, 6, 9)")
+
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.b = s.new_b
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, b, c, _ROW_ID FROM t ORDER BY id"),
+ Seq(Row(5, 5, 5, 5L), Row(7, 700, 7, 7L), Row(8, 8, 8, 8L)))
+ }
+ }
+
+ test("Data Evolution deletion: merge update blob after deletion") {
+ withTable("s", "t") {
+ sql("""
+ |CREATE TABLE t (id INT, b INT, picture BINARY)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'blob-field' = 'picture',
+ | 'blob.target-file-size' = '1 b')
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(1) */ id, b, picture FROM
VALUES
+ | (0, 0, X'00'), (1, 1, X'01'), (2, 2, X'02'), (3, 3, X'03'), (4,
4, X'04')
+ | AS v(id, b, picture)
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(1) */ id, b, picture FROM
VALUES
+ | (5, 5, X'05'), (6, 6, X'06'), (7, 7, X'07'), (8, 8, X'08'), (9,
9, X'09')
+ | AS v(id, b, picture)
+ |""".stripMargin)
+ sql("DELETE FROM t WHERE id IN (0, 1, 2, 3, 4, 6, 9)")
+
+ sql("CREATE TABLE s (id INT, picture BINARY)")
+ sql("INSERT INTO s VALUES (2, X'22'), (6, X'66'), (7, X'4D'), (9,
X'79')")
+
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.picture = s.picture
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, b, picture, _ROW_ID FROM t ORDER BY id"),
+ Seq(
+ Row(5, 5, Array[Byte](5), 5L),
+ Row(7, 7, Array[Byte](77), 7L),
+ Row(8, 8, Array[Byte](8), 8L)))
+ }
+ }
+
+ test("Data Evolution deletion: self merge skips deleted rows") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 5)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(5, 10)")
+ sql("DELETE FROM t WHERE id IN (0, 1, 2, 3, 4, 6, 9)")
+
+ sql("""
+ |MERGE INTO t
+ |USING t AS source
+ |ON t._ROW_ID = source._ROW_ID
+ |WHEN MATCHED AND source.id IN (2, 6, 7, 9) THEN UPDATE SET t.b =
source.b + 100
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, b, c, _ROW_ID FROM t ORDER BY id"),
+ Seq(Row(5, 5, 5, 5L), Row(7, 107, 7, 7L), Row(8, 8, 8, 8L)))
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
index f45a9463b6..2b89afcc68 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
@@ -519,52 +519,6 @@ abstract class DeleteFromTableTestBase extends
PaimonSparkTestBase {
}
}
- test("Paimon delete: data evolution table with deletion vectors") {
- withTable("t") {
- sql("""
- |CREATE TABLE t (id INT, b INT, c INT)
- |TBLPROPERTIES (
- | 'row-tracking.enabled' = 'true',
- | 'data-evolution.enabled' = 'true',
- | 'deletion-vectors.enabled' = 'true')
- |""".stripMargin)
- sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 5)")
- sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(5, 10)")
- sql("ALTER TABLE t ADD COLUMNS (d INT)")
- sql(
- "INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c, id +
100 AS d FROM range(10, 13)")
-
- sql("DELETE FROM t WHERE id IN (1, 4, 6, 11)")
- checkAnswer(
- sql("SELECT *, _ROW_ID FROM t ORDER BY id"),
- Seq(
- Row(0, 0, 0, null, 0L),
- Row(2, 2, 2, null, 2L),
- Row(3, 3, 3, null, 3L),
- Row(5, 5, 5, null, 5L),
- Row(7, 7, 7, null, 7L),
- Row(8, 8, 8, null, 8L),
- Row(9, 9, 9, null, 9L),
- Row(10, 10, 10, 110, 10L),
- Row(12, 12, 12, 112, 12L)
- )
- )
-
- sql("DELETE FROM t WHERE id IN (2, 8)")
- checkAnswer(
- sql("SELECT *, _ROW_ID FROM t ORDER BY id"),
- Seq(
- Row(0, 0, 0, null, 0L),
- Row(3, 3, 3, null, 3L),
- Row(5, 5, 5, null, 5L),
- Row(7, 7, 7, null, 7L),
- Row(9, 9, 9, null, 9L),
- Row(10, 10, 10, 110, 10L),
- Row(12, 12, 12, 112, 12L))
- )
- }
- }
-
test("Paimon delete: delete with range condition") {
withTable("t") {
sql(s"CREATE TABLE t (id INT, v INT)")
