You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "kerwin-zk (via GitHub)" <gi...@apache.org> on 2026/06/12 04:24:03 UTC
[PR] [spark] Support V2 DELETE and UPDATE for data evolution tables [paimon]
kerwin-zk opened a new pull request, #8214:
URL: https://github.com/apache/paimon/pull/8214
### Purpose
Support V2 DELETE and UPDATE for data evolution tables
### Tests
CI
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi closed pull request #8214: [spark] Support V2 UPDATE for data evolution tables
URL: https://github.com/apache/paimon/pull/8214
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "kerwin-zk (via GitHub)" <gi...@apache.org>.
kerwin-zk commented on PR #8214:
URL: https://github.com/apache/paimon/pull/8214#issuecomment-4731112185
Rebased onto latest master and resolved the ConflictDetection conflicts. The PR head is now 55f0db499.
Verified locally:
- mvn -nsu -pl paimon-core -am -Dtest=ConflictDetectionTest -DfailIfNoTests=false test
- mvn -nsu -Pspark4 -pl paimon-spark/paimon-spark-4.1 -DwildcardSuites=org.apache.paimon.spark.sql.RowTrackingTest -DfailIfNoTests=false -Drat.skip=true test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 DELETE and UPDATE for data evolution tables [paimon]
Posted by "kerwin-zk (via GitHub)" <gi...@apache.org>.
kerwin-zk commented on PR #8214:
URL: https://github.com/apache/paimon/pull/8214#issuecomment-4688568901
> How to support delete row ids? cc @leaves12138
@JingsongLi Deleted row ids are simply retired — they become holes in the row-id space. Surviving rows always keep their original row ids, and no physical `_ROW_ID` column is ever written, so row ids stay derived from `firstRowId + position`, exactly like the files produced by `DataEvolutionDeleteRewriter` in #8182
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 DELETE and UPDATE for data evolution tables [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8214:
URL: https://github.com/apache/paimon/pull/8214#issuecomment-4688208030
How to support delete row ids? cc @leaves12138
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8214:
URL: https://github.com/apache/paimon/pull/8214#discussion_r3457705161
##########
paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java:
##########
@@ -643,30 +643,81 @@ Optional<RuntimeException> checkRowIdExistence(
.collect(Collectors.toList());
RowRangeIndex existingIndex = RowRangeIndex.create(existingRanges, false);
+ // Row-id ranges removed by this same commit, keyed by partition and bucket. A
+ // copy-on-write update on a data-evolution table deletes whole row-id groups and re-adds
+ // rewritten rows with their original row ids. File rolling may make an added file cover
+ // only a sub-range of a deleted group and not mirror an existing file exactly; it is still
+ // consistent as long as its range is fully covered by ranges deleted in this commit
+ // (concurrent rewrites of those files are caught by the regular deleted-file conflict
+ // checks).
+ Map<Pair<BinaryRow, Integer>, List<Range>> deletedRanges = new HashMap<>();
Review Comment:
Is this needed by UPDATE?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "kerwin-zk (via GitHub)" <gi...@apache.org>.
kerwin-zk commented on code in PR #8214:
URL: https://github.com/apache/paimon/pull/8214#discussion_r3458047583
##########
paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java:
##########
@@ -643,30 +643,81 @@ Optional<RuntimeException> checkRowIdExistence(
.collect(Collectors.toList());
RowRangeIndex existingIndex = RowRangeIndex.create(existingRanges, false);
+ // Row-id ranges removed by this same commit, keyed by partition and bucket. A
+ // copy-on-write update on a data-evolution table deletes whole row-id groups and re-adds
+ // rewritten rows with their original row ids. File rolling may make an added file cover
+ // only a sub-range of a deleted group and not mirror an existing file exactly; it is still
+ // consistent as long as its range is fully covered by ranges deleted in this commit
+ // (concurrent rewrites of those files are caught by the regular deleted-file conflict
+ // checks).
+ Map<Pair<BinaryRow, Integer>, List<Range>> deletedRanges = new HashMap<>();
Review Comment:
@JingsongLi Yes, this is needed by `UPDATE`.
This logic is not for supporting SQL `DELETE`. It exists for the copy-on-write replacement that `UPDATE` performs in one commit: old physical data files are recorded as manifest `DELETE` entries, and rewritten files are added back with the original row ids.
The rewritten `ADD` files may be sub-ranges of a deleted file because of file rolling, so they do not necessarily have the exact same row-id range as any file in the current snapshot.
For example, an existing file covers row ids `[0, 100)`. The `UPDATE` deletes that physical file and may add rewritten files `[0, 40)` and `[40, 100)`. The old `existingIndex.containsExactly(rowRange)` check would require each `ADD` range to exactly match an existing snapshot file and would falsely report `Row ID existence conflict`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "kerwin-zk (via GitHub)" <gi...@apache.org>.
kerwin-zk commented on PR #8214:
URL: https://github.com/apache/paimon/pull/8214#issuecomment-4730810767
Updated this PR to support UPDATE only for data-evolution tables. DELETE now remains unsupported, including metadata-only DELETE, so it can be handled separately with a deletion-vector based approach later.\n\nVerified locally:\n- mvn -nsu -pl paimon-core -am -Dtest=ConflictDetectionTest -DfailIfNoTests=false test\n- mvn -nsu -Pspark4 -pl paimon-spark/paimon-spark-4.1 -DwildcardSuites=org.apache.paimon.spark.sql.RowTrackingTest -DfailIfNoTests=false -Drat.skip=true test\n- mvn -nsu -Pspark4 -pl paimon-spark/paimon-spark-4.0 -DwildcardSuites=org.apache.paimon.spark.sql.RowTrackingTest -DfailIfNoTests=false -Drat.skip=true test\n- mvn -nsu -Pspark3 -pl paimon-spark/paimon-spark-3.5 -DwildcardSuites=org.apache.paimon.spark.sql.RowTrackingTest -DfailIfNoTests=false -Drat.skip=true test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "kerwin-zk (via GitHub)" <gi...@apache.org>.
kerwin-zk commented on code in PR #8214:
URL: https://github.com/apache/paimon/pull/8214#discussion_r3448613581
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataEvolutionDataWriterBase.scala:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.write
+
+import org.apache.paimon.catalog.CatalogContext
+import org.apache.paimon.data.{BinaryRow, InternalRow => PaimonInternalRow}
+import org.apache.paimon.io.{CompactIncrement, DataIncrement}
+import org.apache.paimon.operation.AbstractFileStoreWrite
+import org.apache.paimon.schema.TableSchema
+import org.apache.paimon.spark.SparkInternalRowWrapper
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
+import org.apache.paimon.table.sink.{BatchWriteBuilder, CommitMessage, CommitMessageImpl, RowPartitionKeyExtractor, TableWriteImpl}
+import org.apache.paimon.utils.RecordWriter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+
+import java.util.Collections
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * V2 [[DataWrite]] for copy-on-write UPDATE on data-evolution tables.
+ *
+ * Data-evolution rows derive `_ROW_ID` from the file's `firstRowId` plus the row position, so
+ * rewritten files must hold contiguous row-id runs and carry a `firstRowId` themselves instead of
+ * writing physical row-tracking columns (files with a physical `_ROW_ID` column are not assigned a
+ * `firstRowId` at commit and cannot be grouped by the data-evolution read path). The surviving rows
+ * arrive range-distributed and sorted by `_ROW_ID` (see
+ * [[PaimonWriteRequirement.dataEvolutionCopyOnWrite]]); this writer starts a new file run whenever
+ * the row id is not consecutive or the partition changes, and assigns each produced file the run's
+ * row id offset, mirroring `DataEvolutionTableDataWrite.PerFileWriter` and the commit skip rule in
+ * `RowTrackingCommitUtils` (files with a non-null `firstRowId` keep it). Sequence numbers are not
+ * preserved: the new files keep `minSequenceNumber == 0` and the commit stamps them with the new
+ * snapshot id, the same as any data-evolution rewrite.
+ */
+abstract class PaimonV2DataEvolutionDataWriterBase(
+ writeBuilder: BatchWriteBuilder,
+ tableSchema: TableSchema,
+ writeSchema: org.apache.spark.sql.types.StructType,
+ dataSchema: org.apache.spark.sql.types.StructType,
+ metadataSchema: org.apache.spark.sql.types.StructType,
+ catalogContext: CatalogContext)
+ extends InnerTableV2DataWrite
+ with Logging {
+
+ private val rowIdIndex = metadataSchema.fieldIndex(PaimonMetadataColumn.ROW_ID_COLUMN)
+
+ private val partitionExtractor = new RowPartitionKeyExtractor(tableSchema)
+
+ private val reusableWrapper =
+ new SparkInternalRowWrapper(writeSchema, writeSchema.fields.length, dataSchema, catalogContext)
+
+ private var currentRun: RunWriter = _
+ private val commitMessages = ListBuffer[CommitMessage]()
+
+ protected def writeWithMetadata(metadata: InternalRow, data: InternalRow): Unit = {
+ assert(
+ metadata != null && !metadata.isNullAt(rowIdIndex),
+ "Data-evolution copy-on-write requires a non-null _ROW_ID for every written row.")
+ val rowId = metadata.getLong(rowIdIndex)
+ val paimonRow = reusableWrapper.replace(data)
+ val partition = partitionExtractor.partition(paimonRow)
+ if (currentRun == null || !currentRun.accepts(rowId, partition)) {
+ finishCurrentRun()
+ currentRun = new RunWriter(rowId, partition.copy())
+ }
+ currentRun.write(paimonRow)
+ }
+
+ override def write(record: InternalRow): Unit = {
+ throw new UnsupportedOperationException(
+ "Data-evolution V2 copy-on-write only rewrites existing rows; " +
+ "rows without row-tracking metadata are not supported.")
+ }
+
+ override def commitImpl(): Seq[CommitMessage] = {
+ finishCurrentRun()
+ commitMessages.toSeq
+ }
+
+ override def abort(): Unit = close()
+
+ override def close(): Unit = {
+ if (currentRun != null) {
+ currentRun.closeQuietly()
+ currentRun = null
+ }
+ }
+
+ private def finishCurrentRun(): Unit = {
+ if (currentRun != null) {
+ commitMessages ++= currentRun.finish()
+ currentRun = null
+ }
+ }
+
+ /** Writes one contiguous row-id run; may roll into several files, all stay range-adjacent. */
+ private class RunWriter(startRowId: Long, partition: BinaryRow) {
+
+ private var nextRowId = startRowId
+
+ // The run writer must only append the rewritten rows and never restore the bucket's
+ // existing files. The bucket-unaware append write already enforces this
+ // (AppendFileStoreWrite pins ignorePreviousFiles=true and uses NoopCompactManager), but
+ // declare it explicitly rather than depend on which FileStoreWrite subclass we got.
+ private val recordWriter: RecordWriter[PaimonInternalRow] = writeBuilder
+ .newWrite()
+ .asInstanceOf[TableWriteImpl[PaimonInternalRow]]
+ .withIgnorePreviousFiles(true)
+ .getWrite
+ .asInstanceOf[AbstractFileStoreWrite[PaimonInternalRow]]
+ .createWriter(partition, 0)
Review Comment:
Fixed in 6ebde90ab. The data-evolution V2 run writer now creates and passes an IOManager into TableWriteImpl, then closes it with the data writer. I also passed the existing IOManager through the V1 data-evolution writer path and added a Spark 4.1 regression test for write-buffer-for-append=true with a small spillable buffer.
Verified with:
`mvn -Pspark4 -pl paimon-spark/paimon-spark-4.1 -am -DwildcardSuites=org.apache.paimon.spark.sql.RowTrackingTest -Dtest=none -DfailIfNoTests=false test`
RowTrackingTest: 47 tests passed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8214:
URL: https://github.com/apache/paimon/pull/8214#discussion_r3445282858
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataEvolutionDataWriterBase.scala:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.write
+
+import org.apache.paimon.catalog.CatalogContext
+import org.apache.paimon.data.{BinaryRow, InternalRow => PaimonInternalRow}
+import org.apache.paimon.io.{CompactIncrement, DataIncrement}
+import org.apache.paimon.operation.AbstractFileStoreWrite
+import org.apache.paimon.schema.TableSchema
+import org.apache.paimon.spark.SparkInternalRowWrapper
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
+import org.apache.paimon.table.sink.{BatchWriteBuilder, CommitMessage, CommitMessageImpl, RowPartitionKeyExtractor, TableWriteImpl}
+import org.apache.paimon.utils.RecordWriter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+
+import java.util.Collections
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * V2 [[DataWrite]] for copy-on-write UPDATE on data-evolution tables.
+ *
+ * Data-evolution rows derive `_ROW_ID` from the file's `firstRowId` plus the row position, so
+ * rewritten files must hold contiguous row-id runs and carry a `firstRowId` themselves instead of
+ * writing physical row-tracking columns (files with a physical `_ROW_ID` column are not assigned a
+ * `firstRowId` at commit and cannot be grouped by the data-evolution read path). The surviving rows
+ * arrive range-distributed and sorted by `_ROW_ID` (see
+ * [[PaimonWriteRequirement.dataEvolutionCopyOnWrite]]); this writer starts a new file run whenever
+ * the row id is not consecutive or the partition changes, and assigns each produced file the run's
+ * row id offset, mirroring `DataEvolutionTableDataWrite.PerFileWriter` and the commit skip rule in
+ * `RowTrackingCommitUtils` (files with a non-null `firstRowId` keep it). Sequence numbers are not
+ * preserved: the new files keep `minSequenceNumber == 0` and the commit stamps them with the new
+ * snapshot id, the same as any data-evolution rewrite.
+ */
+abstract class PaimonV2DataEvolutionDataWriterBase(
+ writeBuilder: BatchWriteBuilder,
+ tableSchema: TableSchema,
+ writeSchema: org.apache.spark.sql.types.StructType,
+ dataSchema: org.apache.spark.sql.types.StructType,
+ metadataSchema: org.apache.spark.sql.types.StructType,
+ catalogContext: CatalogContext)
+ extends InnerTableV2DataWrite
+ with Logging {
+
+ private val rowIdIndex = metadataSchema.fieldIndex(PaimonMetadataColumn.ROW_ID_COLUMN)
+
+ private val partitionExtractor = new RowPartitionKeyExtractor(tableSchema)
+
+ private val reusableWrapper =
+ new SparkInternalRowWrapper(writeSchema, writeSchema.fields.length, dataSchema, catalogContext)
+
+ private var currentRun: RunWriter = _
+ private val commitMessages = ListBuffer[CommitMessage]()
+
+ protected def writeWithMetadata(metadata: InternalRow, data: InternalRow): Unit = {
+ assert(
+ metadata != null && !metadata.isNullAt(rowIdIndex),
+ "Data-evolution copy-on-write requires a non-null _ROW_ID for every written row.")
+ val rowId = metadata.getLong(rowIdIndex)
+ val paimonRow = reusableWrapper.replace(data)
+ val partition = partitionExtractor.partition(paimonRow)
+ if (currentRun == null || !currentRun.accepts(rowId, partition)) {
+ finishCurrentRun()
+ currentRun = new RunWriter(rowId, partition.copy())
+ }
+ currentRun.write(paimonRow)
+ }
+
+ override def write(record: InternalRow): Unit = {
+ throw new UnsupportedOperationException(
+ "Data-evolution V2 copy-on-write only rewrites existing rows; " +
+ "rows without row-tracking metadata are not supported.")
+ }
+
+ override def commitImpl(): Seq[CommitMessage] = {
+ finishCurrentRun()
+ commitMessages.toSeq
+ }
+
+ override def abort(): Unit = close()
+
+ override def close(): Unit = {
+ if (currentRun != null) {
+ currentRun.closeQuietly()
+ currentRun = null
+ }
+ }
+
+ private def finishCurrentRun(): Unit = {
+ if (currentRun != null) {
+ commitMessages ++= currentRun.finish()
+ currentRun = null
+ }
+ }
+
+ /** Writes one contiguous row-id run; may roll into several files, all stay range-adjacent. */
+ private class RunWriter(startRowId: Long, partition: BinaryRow) {
+
+ private var nextRowId = startRowId
+
+ // The run writer must only append the rewritten rows and never restore the bucket's
+ // existing files. The bucket-unaware append write already enforces this
+ // (AppendFileStoreWrite pins ignorePreviousFiles=true and uses NoopCompactManager), but
+ // declare it explicitly rather than depend on which FileStoreWrite subclass we got.
+ private val recordWriter: RecordWriter[PaimonInternalRow] = writeBuilder
+ .newWrite()
+ .asInstanceOf[TableWriteImpl[PaimonInternalRow]]
+ .withIgnorePreviousFiles(true)
+ .getWrite
+ .asInstanceOf[AbstractFileStoreWrite[PaimonInternalRow]]
+ .createWriter(partition, 0)
Review Comment:
This path skips the IOManager setup that the normal V2 writer does before creating append writers. If a data-evolution table has `write-buffer-for-append=true` (with the default spillable buffer), `AppendOnlyWriter` builds an `ExternalBuffer` with a null IOManager; once the buffer spills, it will fail at `ioManager.createChannel()`. Please pass an IOManager into this `TableWriteImpl` (and close it with the writer), or otherwise disable the buffered append path here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "kerwin-zk (via GitHub)" <gi...@apache.org>.
kerwin-zk commented on code in PR #8214:
URL: https://github.com/apache/paimon/pull/8214#discussion_r3456621363
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataEvolutionDataWriterBase.scala:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.write
+
+import org.apache.paimon.catalog.CatalogContext
+import org.apache.paimon.data.{BinaryRow, InternalRow => PaimonInternalRow}
+import org.apache.paimon.io.{CompactIncrement, DataIncrement}
+import org.apache.paimon.operation.AbstractFileStoreWrite
+import org.apache.paimon.schema.TableSchema
+import org.apache.paimon.spark.SparkInternalRowWrapper
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
+import org.apache.paimon.table.sink.{BatchWriteBuilder, CommitMessage, CommitMessageImpl, RowPartitionKeyExtractor, TableWriteImpl}
+import org.apache.paimon.utils.RecordWriter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+
+import java.util.Collections
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * V2 [[DataWrite]] for copy-on-write UPDATE on data-evolution tables.
+ *
+ * Data-evolution rows derive `_ROW_ID` from the file's `firstRowId` plus the row position, so
+ * rewritten files must hold contiguous row-id runs and carry a `firstRowId` themselves instead of
+ * writing physical row-tracking columns (files with a physical `_ROW_ID` column are not assigned a
+ * `firstRowId` at commit and cannot be grouped by the data-evolution read path). The surviving rows
+ * arrive range-distributed and sorted by `_ROW_ID` (see
+ * [[PaimonWriteRequirement.dataEvolutionCopyOnWrite]]); this writer starts a new file run whenever
+ * the row id is not consecutive or the partition changes, and assigns each produced file the run's
+ * row id offset, mirroring `DataEvolutionTableDataWrite.PerFileWriter` and the commit skip rule in
+ * `RowTrackingCommitUtils` (files with a non-null `firstRowId` keep it). Sequence numbers are not
+ * preserved: the new files keep `minSequenceNumber == 0` and the commit stamps them with the new
+ * snapshot id, the same as any data-evolution rewrite.
+ */
+abstract class PaimonV2DataEvolutionDataWriterBase(
+ writeBuilder: BatchWriteBuilder,
+ tableSchema: TableSchema,
+ writeSchema: org.apache.spark.sql.types.StructType,
+ dataSchema: org.apache.spark.sql.types.StructType,
+ metadataSchema: org.apache.spark.sql.types.StructType,
+ catalogContext: CatalogContext)
+ extends InnerTableV2DataWrite
+ with Logging {
+
+ private val rowIdIndex = metadataSchema.fieldIndex(PaimonMetadataColumn.ROW_ID_COLUMN)
+
+ private val partitionExtractor = new RowPartitionKeyExtractor(tableSchema)
+
+ private val reusableWrapper =
+ new SparkInternalRowWrapper(writeSchema, writeSchema.fields.length, dataSchema, catalogContext)
+
+ private var currentRun: RunWriter = _
+ private val commitMessages = ListBuffer[CommitMessage]()
+
+ protected def writeWithMetadata(metadata: InternalRow, data: InternalRow): Unit = {
+ assert(
+ metadata != null && !metadata.isNullAt(rowIdIndex),
+ "Data-evolution copy-on-write requires a non-null _ROW_ID for every written row.")
+ val rowId = metadata.getLong(rowIdIndex)
+ val paimonRow = reusableWrapper.replace(data)
+ val partition = partitionExtractor.partition(paimonRow)
+ if (currentRun == null || !currentRun.accepts(rowId, partition)) {
+ finishCurrentRun()
+ currentRun = new RunWriter(rowId, partition.copy())
+ }
+ currentRun.write(paimonRow)
+ }
+
+ override def write(record: InternalRow): Unit = {
+ throw new UnsupportedOperationException(
+ "Data-evolution V2 copy-on-write only rewrites existing rows; " +
+ "rows without row-tracking metadata are not supported.")
+ }
+
+ override def commitImpl(): Seq[CommitMessage] = {
+ finishCurrentRun()
+ commitMessages.toSeq
+ }
+
+ override def abort(): Unit = close()
+
+ override def close(): Unit = {
+ if (currentRun != null) {
+ currentRun.closeQuietly()
+ currentRun = null
+ }
+ }
+
+ private def finishCurrentRun(): Unit = {
+ if (currentRun != null) {
+ commitMessages ++= currentRun.finish()
+ currentRun = null
+ }
+ }
+
+ /** Writes one contiguous row-id run; may roll into several files, all stay range-adjacent. */
+ private class RunWriter(startRowId: Long, partition: BinaryRow) {
+
+ private var nextRowId = startRowId
+
+ // The run writer must only append the rewritten rows and never restore the bucket's
+ // existing files. The bucket-unaware append write already enforces this
+ // (AppendFileStoreWrite pins ignorePreviousFiles=true and uses NoopCompactManager), but
+ // declare it explicitly rather than depend on which FileStoreWrite subclass we got.
+ private val recordWriter: RecordWriter[PaimonInternalRow] = writeBuilder
+ .newWrite()
+ .asInstanceOf[TableWriteImpl[PaimonInternalRow]]
+ .withIgnorePreviousFiles(true)
+ .getWrite
+ .asInstanceOf[AbstractFileStoreWrite[PaimonInternalRow]]
+ .createWriter(partition, 0)
Review Comment:
@JingsongLi done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 DELETE and UPDATE for data evolution tables [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8214:
URL: https://github.com/apache/paimon/pull/8214#issuecomment-4727755076
> > How to support delete row ids? cc @leaves12138
>
> @JingsongLi Deleted row ids are simply retired — they become holes in the row-id space. Surviving rows always keep their original row ids, and no physical `_ROW_ID` column is ever written, so row ids stay derived from `firstRowId + position`, exactly like the files produced by `DataEvolutionDeleteRewriter` in #8182
@kerwin-zk It's too troublesome to maintain the index after deletion. Maybe the correct solution is deletion-vector. Can you just support update in this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "kerwin-zk (via GitHub)" <gi...@apache.org>.
kerwin-zk commented on code in PR #8214:
URL: https://github.com/apache/paimon/pull/8214#discussion_r3448613581
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataEvolutionDataWriterBase.scala:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.write
+
+import org.apache.paimon.catalog.CatalogContext
+import org.apache.paimon.data.{BinaryRow, InternalRow => PaimonInternalRow}
+import org.apache.paimon.io.{CompactIncrement, DataIncrement}
+import org.apache.paimon.operation.AbstractFileStoreWrite
+import org.apache.paimon.schema.TableSchema
+import org.apache.paimon.spark.SparkInternalRowWrapper
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
+import org.apache.paimon.table.sink.{BatchWriteBuilder, CommitMessage, CommitMessageImpl, RowPartitionKeyExtractor, TableWriteImpl}
+import org.apache.paimon.utils.RecordWriter
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+
+import java.util.Collections
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * V2 [[DataWrite]] for copy-on-write UPDATE on data-evolution tables.
+ *
+ * Data-evolution rows derive `_ROW_ID` from the file's `firstRowId` plus the row position, so
+ * rewritten files must hold contiguous row-id runs and carry a `firstRowId` themselves instead of
+ * writing physical row-tracking columns (files with a physical `_ROW_ID` column are not assigned a
+ * `firstRowId` at commit and cannot be grouped by the data-evolution read path). The surviving rows
+ * arrive range-distributed and sorted by `_ROW_ID` (see
+ * [[PaimonWriteRequirement.dataEvolutionCopyOnWrite]]); this writer starts a new file run whenever
+ * the row id is not consecutive or the partition changes, and assigns each produced file the run's
+ * row id offset, mirroring `DataEvolutionTableDataWrite.PerFileWriter` and the commit skip rule in
+ * `RowTrackingCommitUtils` (files with a non-null `firstRowId` keep it). Sequence numbers are not
+ * preserved: the new files keep `minSequenceNumber == 0` and the commit stamps them with the new
+ * snapshot id, the same as any data-evolution rewrite.
+ */
+abstract class PaimonV2DataEvolutionDataWriterBase(
+ writeBuilder: BatchWriteBuilder,
+ tableSchema: TableSchema,
+ writeSchema: org.apache.spark.sql.types.StructType,
+ dataSchema: org.apache.spark.sql.types.StructType,
+ metadataSchema: org.apache.spark.sql.types.StructType,
+ catalogContext: CatalogContext)
+ extends InnerTableV2DataWrite
+ with Logging {
+
+ private val rowIdIndex = metadataSchema.fieldIndex(PaimonMetadataColumn.ROW_ID_COLUMN)
+
+ private val partitionExtractor = new RowPartitionKeyExtractor(tableSchema)
+
+ private val reusableWrapper =
+ new SparkInternalRowWrapper(writeSchema, writeSchema.fields.length, dataSchema, catalogContext)
+
+ private var currentRun: RunWriter = _
+ private val commitMessages = ListBuffer[CommitMessage]()
+
+ protected def writeWithMetadata(metadata: InternalRow, data: InternalRow): Unit = {
+ assert(
+ metadata != null && !metadata.isNullAt(rowIdIndex),
+ "Data-evolution copy-on-write requires a non-null _ROW_ID for every written row.")
+ val rowId = metadata.getLong(rowIdIndex)
+ val paimonRow = reusableWrapper.replace(data)
+ val partition = partitionExtractor.partition(paimonRow)
+ if (currentRun == null || !currentRun.accepts(rowId, partition)) {
+ finishCurrentRun()
+ currentRun = new RunWriter(rowId, partition.copy())
+ }
+ currentRun.write(paimonRow)
+ }
+
+ override def write(record: InternalRow): Unit = {
+ throw new UnsupportedOperationException(
+ "Data-evolution V2 copy-on-write only rewrites existing rows; " +
+ "rows without row-tracking metadata are not supported.")
+ }
+
+ override def commitImpl(): Seq[CommitMessage] = {
+ finishCurrentRun()
+ commitMessages.toSeq
+ }
+
+ override def abort(): Unit = close()
+
+ override def close(): Unit = {
+ if (currentRun != null) {
+ currentRun.closeQuietly()
+ currentRun = null
+ }
+ }
+
+ private def finishCurrentRun(): Unit = {
+ if (currentRun != null) {
+ commitMessages ++= currentRun.finish()
+ currentRun = null
+ }
+ }
+
+ /** Writes one contiguous row-id run; may roll into several files, all stay range-adjacent. */
+ private class RunWriter(startRowId: Long, partition: BinaryRow) {
+
+ private var nextRowId = startRowId
+
+ // The run writer must only append the rewritten rows and never restore the bucket's
+ // existing files. The bucket-unaware append write already enforces this
+ // (AppendFileStoreWrite pins ignorePreviousFiles=true and uses NoopCompactManager), but
+ // declare it explicitly rather than depend on which FileStoreWrite subclass we got.
+ private val recordWriter: RecordWriter[PaimonInternalRow] = writeBuilder
+ .newWrite()
+ .asInstanceOf[TableWriteImpl[PaimonInternalRow]]
+ .withIgnorePreviousFiles(true)
+ .getWrite
+ .asInstanceOf[AbstractFileStoreWrite[PaimonInternalRow]]
+ .createWriter(partition, 0)
Review Comment:
Fixed in 6ebde90ab. The data-evolution V2 run writer now creates and passes an IOManager into TableWriteImpl, then closes it with the data writer. I also passed the existing IOManager through the V1 data-evolution writer path and added a Spark 4.1 regression test for write-buffer-for-append=true with a small spillable buffer.
Verified with:
`mvn -Pspark4 -pl paimon-spark/paimon-spark-4.1 -am -DwildcardSuites=org.apache.paimon.spark.sql.RowTrackingTest -Dtest=none -DfailIfNoTests=false test`
RowTrackingTest: 47 tests passed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support V2 UPDATE for data evolution tables [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8214:
URL: https://github.com/apache/paimon/pull/8214#issuecomment-4778603009
@kerwin-zk We do not intend to support the COPY ON WRITE update / delete method, as it would result in severe index pollution and generate many small files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org