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 bf55204d58 [core][spark] Fix row ids and sequence numbers of files
copied by sys.copy (#10391)
bf55204d58 is described below
commit bf55204d582c858a9059bff07bbed19f5da737d4
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Fri Oct 16 13:20:53 2026 +0800
[core][spark] Fix row ids and sequence numbers of files copied by sys.copy
(#10391)
---
.../org/apache/paimon/migrate/CopiedDataFiles.java | 274 +++++++
.../operation/commit/RowTrackingCommitUtils.java | 23 +-
.../apache/paimon/migrate/CopiedDataFilesTest.java | 821 +++++++++++++++++++++
.../paimon/table/DataEvolutionTableTest.java | 2 +
.../paimon/table/RowTrackingCopiedFilesTest.java | 187 +++++
.../paimon/spark/copy/CopyFilesCommitOperator.java | 7 +
.../apache/paimon/spark/copy/CopyFilesUtil.java | 20 +
.../paimon/spark/copy/ListDataFilesOperator.java | 8 +-
.../paimon/spark/copy/CopyFilesUtilTest.java | 28 +
.../spark/procedure/CopyFilesProcedureTest.scala | 288 ++++++++
10 files changed, 1653 insertions(+), 5 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/migrate/CopiedDataFiles.java
b/paimon-core/src/main/java/org/apache/paimon/migrate/CopiedDataFiles.java
new file mode 100644
index 0000000000..b8e53f1dc5
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/migrate/CopiedDataFiles.java
@@ -0,0 +1,274 @@
+/*
+ * 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.migrate;
+
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.PojoDataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+
+/**
+ * Prepares data files copied from another table, as {@code sys.copy} does,
for the commit into the
+ * table they are copied to.
+ *
+ * <p>A copied file arrives with the first row id and the sequence numbers it
had in the source
+ * table. Row ids identify rows within one table, and the target has its own:
committing the source
+ * ones would let the copied rows share row ids with rows of the target, and
on a data-evolution
+ * target the source sequence numbers could hide later updates. The copied
files are therefore
+ * committed like files written for the target: without a first row id and
with sequence numbers
+ * that the commit stamps with its snapshot id. A row-tracking commit then
assigns the row ids,
+ * against the snapshot it commits on, as it does for any writer, so a
concurrent commit to the
+ * target cannot take the same row ids.
+ *
+ * <p>A row-tracking commit gives a blob or vector-store file the row ids of
the normal file before
+ * it, so the files of one row id range are committed in order: the normal
file, then its dedicated
+ * files. A file that updates columns of rows another copied file holds cannot
be committed that
+ * way, and is refused: compacting the source first merges it into one file
per row id range. So is
+ * a blob or vector-store file that the commit would give the row ids of other
rows than it holds.
+ */
+public final class CopiedDataFiles {
+
+ private CopiedDataFiles() {}
+
+ /** Replaces, in place, every copied file by its metadata for {@code
target}. */
+ public static void adaptToTarget(
+ FileStoreTable target, Collection<List<DataFileMeta>>
copiedFilesByBucket) {
+ boolean rowTracking = target.coreOptions().rowTrackingEnabled();
+ boolean vectorStoreFiles =
+ copiedFilesByBucket.stream()
+ .flatMap(List::stream)
+ .anyMatch(file -> kind(file) == 2);
+ for (List<DataFileMeta> files : copiedFilesByBucket) {
+ checkNoColumnUpdates(target, files);
+ if (rowTracking) {
+ checkNoFileStoresRowIds(target, files);
+ // a normal file before the dedicated files of its row id range
+ files.sort(
+ Comparator.comparing(
+ (DataFileMeta file) ->
file.firstRowId(),
+
Comparator.nullsLast(Comparator.naturalOrder()))
+ .thenComparingInt(CopiedDataFiles::kind));
+ checkDedicatedFilesFollowTheirRows(target, files,
vectorStoreFiles);
+ }
+ ListIterator<DataFileMeta> iterator = files.listIterator();
+ while (iterator.hasNext()) {
+ DataFileMeta file = iterator.next();
+ // A table without row tracking keeps no row ids: its own
files have none either.
+ iterator.set(rowTracking ? asNewFile(file) :
file.newFirstRowId(null));
+ }
+ }
+ }
+
+ /**
+ * A file written for the target: no first row id, an append so that the
commit assigns one, and
+ * sequence numbers starting at 0 so that the commit stamps them with its
snapshot id.
+ */
+ private static DataFileMeta asNewFile(DataFileMeta file) {
+ return new PojoDataFileMeta(
+ file.fileName(),
+ file.fileSize(),
+ file.rowCount(),
+ file.minKey(),
+ file.maxKey(),
+ file.keyStats(),
+ file.valueStats(),
+ 0L,
+ 0L,
+ file.schemaId(),
+ file.level(),
+ file.extraFiles(),
+ file.creationTime(),
+ file.deleteRowCount().orElse(null),
+ file.embeddedIndex(),
+ FileSource.APPEND,
+ file.valueStatsCols(),
+ file.externalPath().orElse(null),
+ null,
+ file.writeCols(),
+ null);
+ }
+
+ /**
+ * Refuses a normal file whose row ids another normal file of the copy
holds too: a column
+ * update of a data-evolution table, which only shares the row ids of the
file it updates.
+ */
+ private static void checkNoColumnUpdates(FileStoreTable target,
List<DataFileMeta> files) {
+ List<DataFileMeta> normal = new ArrayList<>();
+ for (DataFileMeta file : files) {
+ if (file.firstRowId() != null && kind(file) == 0) {
+ normal.add(file);
+ }
+ }
+ normal.sort(Comparator.comparingLong(DataFileMeta::nonNullFirstRowId));
+ for (int i = 1; i < normal.size(); i++) {
+ DataFileMeta previous = normal.get(i - 1);
+ DataFileMeta current = normal.get(i);
+ if (current.nonNullFirstRowId() < previous.nonNullFirstRowId() +
previous.rowCount()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot copy data files %s and %s into table
%s: they hold the "
+ + "same rows, one updating columns of
the other. Compact "
+ + "the source table first, which
merges them.",
+ previous.fileName(), current.fileName(),
target.name()));
+ }
+ }
+ }
+
+ /**
+ * A file that stores the row ids of its rows, as a copy-on-write update
writes it, has no first
+ * row id: its row ids are in its data and would collide with the row ids
of the target.
+ */
+ private static void checkNoFileStoresRowIds(FileStoreTable target,
List<DataFileMeta> files) {
+ for (DataFileMeta file : files) {
+ List<String> writeCols = file.writeCols();
+ if (writeCols != null &&
writeCols.contains(SpecialFields.ROW_ID.name())) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot copy data file %s into table %s: its
rows store their row "
+ + "ids, written by a copy-on-write
UPDATE, DELETE or MERGE "
+ + "INTO on the source, so they cannot
be given row ids of "
+ + "the target. Rewrite those rows in
the source first, for "
+ + "example with INSERT OVERWRITE.",
+ file.fileName(), target.name()));
+ }
+ }
+ }
+
+ /**
+ * Refuses a blob or vector-store file that the commit would give the row
ids of other rows than
+ * it holds. The commit gives a normal file the row ids after the normal
file before it, a blob
+ * file those from the start of the normal file before it, or after the
blob file of the same
+ * field following that normal file, and a vector-store file those after
the vector-store file
+ * before it in the whole commit. This replays that on the files of one
bucket, which the commit
+ * keeps together but may order among the other buckets in any way: so
when the copy has
+ * vector-store files, those of every bucket must also hold as many rows
as its normal files. A
+ * source breaks this when a blob or vector column was updated, or added
after rows were
+ * written.
+ */
+ private static void checkDedicatedFilesFollowTheirRows(
+ FileStoreTable target, List<DataFileMeta> files, boolean
vectorStoreFiles) {
+ // the row ids the commit gives, counted from the first one it gives
to the bucket
+ long start = 0;
+ long blobStartDefault = 0;
+ Map<String, Long> blobStarts = new HashMap<>();
+ long vectorStoreStart = 0;
+ List<DataFileMeta> normalFiles = new ArrayList<>();
+ List<Long> normalStarts = new ArrayList<>();
+ List<DataFileMeta> dedicatedFiles = new ArrayList<>();
+ List<Long> dedicatedStarts = new ArrayList<>();
+ for (DataFileMeta file : files) {
+ int kind = kind(file);
+ if (kind == 0) {
+ normalFiles.add(file);
+ normalStarts.add(start);
+ blobStartDefault = start;
+ blobStarts.clear();
+ start += file.rowCount();
+ continue;
+ }
+ long given;
+ if (kind == 1) {
+ String field = file.writeCols().get(0);
+ given = blobStarts.getOrDefault(field, blobStartDefault);
+ blobStarts.put(field, given + file.rowCount());
+ } else {
+ given = vectorStoreStart;
+ vectorStoreStart += file.rowCount();
+ }
+ if (given >= start) {
+ // no normal file holds these row ids yet: the commit refuses
it
+ throw rowsOfOtherFiles(target, file);
+ }
+ dedicatedFiles.add(file);
+ dedicatedStarts.add(given);
+ }
+
+ // the normal and the dedicated files are both in the order of their
source row ids
+ int normal = 0;
+ for (int i = 0; i < dedicatedFiles.size(); i++) {
+ DataFileMeta file = dedicatedFiles.get(i);
+ if (file.firstRowId() == null) {
+ continue;
+ }
+ long row = file.firstRowId();
+ long end = row + file.rowCount();
+ long offset = dedicatedStarts.get(i) - row;
+ while (normal < normalFiles.size() &&
endRowId(normalFiles.get(normal)) <= row) {
+ normal++;
+ }
+ // every row of the file must get the row id the commit gives it
in its normal file
+ for (int n = normal; row < end; n++) {
+ DataFileMeta normalFile = n < normalFiles.size() ?
normalFiles.get(n) : null;
+ if (normalFile == null
+ || normalFile.firstRowId() == null
+ || normalFile.firstRowId() > row
+ || normalStarts.get(n) - normalFile.firstRowId() !=
offset) {
+ throw rowsOfOtherFiles(target, file);
+ }
+ row = endRowId(normalFile);
+ }
+ }
+
+ if (vectorStoreFiles && vectorStoreStart < start) {
+ // rows without a vector-store file move those of the next buckets
onto other rows
+ for (int n = normalFiles.size() - 1; n >= 0; n--) {
+ if (normalStarts.get(n) <= vectorStoreStart) {
+ throw rowsOfOtherFiles(target, normalFiles.get(n));
+ }
+ }
+ }
+ }
+
+ private static long endRowId(DataFileMeta file) {
+ return file.firstRowId() == null ? Long.MIN_VALUE : file.firstRowId()
+ file.rowCount();
+ }
+
+ private static IllegalArgumentException rowsOfOtherFiles(
+ FileStoreTable target, DataFileMeta file) {
+ return new IllegalArgumentException(
+ String.format(
+ "Cannot copy data file %s into table %s: the commit
would give the blob or "
+ + "vector-store files of the source the row
ids of other rows "
+ + "than they hold, because they do not follow
the rows of the "
+ + "normal files, as after updating a blob or
vector column or "
+ + "adding one to rows already written. Rewrite
those rows in "
+ + "the source first, for example with INSERT
OVERWRITE.",
+ file.fileName(), target.name()));
+ }
+
+ /** 0 for a normal file, 1 for a blob file, 2 for a vector-store file. */
+ private static int kind(DataFileMeta file) {
+ if (isBlobFile(file.fileName())) {
+ return 1;
+ }
+ return isVectorStoreFile(file.fileName()) ? 2 : 0;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowTrackingCommitUtils.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowTrackingCommitUtils.java
index f6e0b660ce..8b8fcd3476 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowTrackingCommitUtils.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowTrackingCommitUtils.java
@@ -18,6 +18,7 @@
package org.apache.paimon.operation.commit;
+import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.table.SpecialFields;
@@ -43,10 +44,30 @@ public class RowTrackingCommitUtils {
// assign row id for new files
List<ManifestEntry> rowIdAssigned = new ArrayList<>();
long nextRowIdStart =
- assignRowTrackingMeta(firstRowIdStart, snapshotAssigned,
rowIdAssigned);
+ assignRowTrackingMeta(
+ Math.max(firstRowIdStart, rowIdEnd(deltaFiles)),
+ snapshotAssigned,
+ rowIdAssigned);
return new RowTrackingAssigned(nextRowIdStart, rowIdAssigned);
}
+ /**
+ * The end of the row ids that files added with a first row id they
already have occupy. The
+ * next row id of the table does not cover them when they come from
another table, for example
+ * files copied by {@code sys.copy}: new files must be assigned row ids
after them, and the next
+ * row id of the table must move past them, or later rows get the same row
ids.
+ */
+ private static long rowIdEnd(List<ManifestEntry> deltaFiles) {
+ long end = 0L;
+ for (ManifestEntry entry : deltaFiles) {
+ Long firstRowId = entry.file().firstRowId();
+ if (entry.kind() == FileKind.ADD && firstRowId != null) {
+ end = Math.max(end, firstRowId + entry.file().rowCount());
+ }
+ }
+ return end;
+ }
+
private static void assignSnapshotId(
long snapshotId, List<ManifestEntry> deltaFiles,
List<ManifestEntry> snapshotAssigned) {
for (ManifestEntry entry : deltaFiles) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/migrate/CopiedDataFilesTest.java
b/paimon-core/src/test/java/org/apache/paimon/migrate/CopiedDataFilesTest.java
new file mode 100644
index 0000000000..83c030fc2a
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/migrate/CopiedDataFilesTest.java
@@ -0,0 +1,821 @@
+/*
+ * 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.migrate;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BlobData;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
+
+/**
+ * Tests for {@link CopiedDataFiles}: data files copied from another table and
committed the way
+ * {@code CopyFilesCommitOperator} of {@code sys.copy} does.
+ */
+public class CopiedDataFilesTest extends TableTestBase {
+
+ private static final Identifier SOURCE = new Identifier("default", "src");
+ private static final Identifier TARGET = new Identifier("default", "dst");
+
+ //
---------------------------------------------------------------------------------------------
+ // row ids
+ //
---------------------------------------------------------------------------------------------
+
+ @Test
+ public void testCopyIntoTableWithoutRowsAssignsRowIdsInOrder() throws
Exception {
+ createTable(SOURCE, Options.ROW_TRACKING, false);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ write(table(SOURCE), row(3, "c", "p1"));
+ createTable(TARGET, Options.ROW_TRACKING, false);
+
+ copy(SOURCE, TARGET, null);
+
+ assertThat(rowIds(table(TARGET)))
+ .containsEntry(1, 0L)
+ .containsEntry(2, 1L)
+ .containsEntry(3, 2L);
+
assertThat(table(TARGET).snapshotManager().latestSnapshot().nextRowId()).isEqualTo(3L);
+ write(table(TARGET), row(4, "d", "p1"));
+ assertThat(rowIds(table(TARGET))).containsEntry(4, 3L);
+ assertUniqueRowIds(table(TARGET));
+ }
+
+ @Test
+ public void testCopyIntoOnePartitionFollowsTheRowsOfOtherPartitions()
throws Exception {
+ createTable(SOURCE, Options.ROW_TRACKING, true);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ createTable(TARGET, Options.ROW_TRACKING, true);
+ write(table(TARGET), row(10, "x", "p2"), row(11, "y", "p2"));
+
+ // dynamic partition overwrite replaces p1 only: the rows of p2 keep
row ids 0 and 1
+ copy(SOURCE, TARGET, null);
+
+ FileStoreTable target = table(TARGET);
+ assertThat(values(target)).containsOnlyKeys(1, 2, 10, 11);
+ assertThat(rowIds(target))
+ .containsEntry(10, 0L)
+ .containsEntry(11, 1L)
+ .containsEntry(1, 2L)
+ .containsEntry(2, 3L);
+
assertThat(target.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(4L);
+ assertUniqueRowIds(target);
+ }
+
+ @Test
+ public void testCopyWithPartitionFilterFollowsTheRowsOfOtherPartitions()
throws Exception {
+ createTable(SOURCE, Options.ROW_TRACKING, true);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p2"));
+ createTable(TARGET, Options.ROW_TRACKING, true);
+ write(table(TARGET), row(10, "x", "p2"));
+
+ copy(SOURCE, TARGET, "p1");
+
+ assertThat(values(table(TARGET))).containsOnlyKeys(1, 10);
+ assertThat(rowIds(table(TARGET))).containsEntry(10,
0L).containsEntry(1, 1L);
+ assertUniqueRowIds(table(TARGET));
+ }
+
+ @Test
+ public void testCopyOverEveryRowFollowsTheRowIdsGivenOut() throws
Exception {
+ createTable(SOURCE, Options.ROW_TRACKING, false);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ createTable(TARGET, Options.ROW_TRACKING, false);
+ write(table(TARGET), row(10, "x", "p1"), row(11, "y", "p1"), row(12,
"z", "p1"));
+
+ copy(SOURCE, TARGET, null);
+
+ // the row ids of the overwritten rows are not given out again, as for
any overwrite
+ assertThat(values(table(TARGET))).containsOnlyKeys(1, 2);
+ assertThat(rowIds(table(TARGET))).containsEntry(1,
3L).containsEntry(2, 4L);
+
assertThat(table(TARGET).snapshotManager().latestSnapshot().nextRowId()).isEqualTo(5L);
+ }
+
+ @Test
+ public void testAppendCommittedWhileTheCopyIsPreparedGetsOtherRowIds()
throws Exception {
+ createTable(SOURCE, Options.ROW_TRACKING, true);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ createTable(TARGET, Options.ROW_TRACKING, true);
+ write(table(TARGET), row(10, "x", "p1"), row(11, "y", "p1"));
+
+ // the copy is prepared, then another writer appends to p2, then the
copy commits
+ Map<Pair<BinaryRow, Integer>, List<DataFileMeta>> files =
prepareCopy(SOURCE, TARGET);
+ write(table(TARGET), row(20, "u", "p2"), row(21, "v", "p2"));
+ commitCopy(table(TARGET), files);
+
+ Map<Integer, Long> rowIds = rowIds(table(TARGET));
+ assertThat(rowIds).containsOnlyKeys(1, 2, 20, 21);
+ assertThat(rowIds).containsEntry(20, 2L).containsEntry(21, 3L);
+ assertThat(rowIds).containsEntry(1, 4L).containsEntry(2, 5L);
+ assertUniqueRowIds(table(TARGET));
+ }
+
+ @Test
+ public void testAppendToEmptyTargetWhileTheCopyIsPreparedGetsOtherRowIds()
throws Exception {
+ createTable(SOURCE, Options.ROW_TRACKING, true);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ createTable(TARGET, Options.ROW_TRACKING, true);
+
+ Map<Pair<BinaryRow, Integer>, List<DataFileMeta>> files =
prepareCopy(SOURCE, TARGET);
+ write(table(TARGET), row(20, "u", "p2"), row(21, "v", "p2"));
+ commitCopy(table(TARGET), files);
+
+ assertThat(rowIds(table(TARGET)))
+ .containsEntry(20, 0L)
+ .containsEntry(21, 1L)
+ .containsEntry(1, 2L)
+ .containsEntry(2, 3L);
+ assertUniqueRowIds(table(TARGET));
+ }
+
+ @Test
+ public void testCopyCommitRetriedAfterAConcurrentAppendGetsOtherRowIds()
throws Exception {
+ createTable(SOURCE, Options.ROW_TRACKING, true);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ createTable(TARGET, Options.ROW_TRACKING, true);
+ write(table(TARGET), row(10, "x", "p1"));
+ Map<Pair<BinaryRow, Integer>, List<DataFileMeta>> files =
prepareCopy(SOURCE, TARGET);
+
+ // The copy commit reads the latest snapshot and stops at its first
manifest write. An
+ // append commits meanwhile, so the copy loses the next snapshot id
and retries.
+ CountDownLatch paused = new CountDownLatch(1);
+ CountDownLatch released = new CountDownLatch(1);
+ AtomicReference<Thread> committer = new AtomicReference<>();
+ FileStoreTable pausing =
+ FileStoreTableFactory.create(
+ new PausingFileIO(committer, paused, released),
table(TARGET).location());
+ AtomicReference<Throwable> error = new AtomicReference<>();
+ Thread thread =
+ new Thread(
+ () -> {
+ try {
+ commitCopy(pausing, files);
+ } catch (Throwable t) {
+ error.set(t);
+ }
+ });
+ committer.set(thread);
+ thread.start();
+ assertThat(paused.await(30, TimeUnit.SECONDS)).isTrue();
+ write(table(TARGET), row(20, "u", "p2"), row(21, "v", "p2"));
+ released.countDown();
+ thread.join(TimeUnit.SECONDS.toMillis(60));
+ assertThat(error.get()).isNull();
+
+ Map<Integer, Long> rowIds = rowIds(table(TARGET));
+ assertThat(rowIds).containsOnlyKeys(1, 2, 20, 21);
+ assertThat(rowIds).containsEntry(20, 1L).containsEntry(21, 2L);
+ assertThat(rowIds).containsEntry(1, 3L).containsEntry(2, 4L);
+ assertUniqueRowIds(table(TARGET));
+ }
+
+ @Test
+ public void testColumnUpdatesOfTheSourceAreRefused() throws Exception {
+ createTable(SOURCE, Options.DATA_EVOLUTION, false);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ // a column update of both rows: shares the row id range of the file
above
+ updateColumn(table(SOURCE), 0, "a2", "b2");
+ createTable(TARGET, Options.DATA_EVOLUTION, false);
+
+ assertThatThrownBy(() -> copy(SOURCE, TARGET, null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("they hold the same rows")
+ .hasMessageContaining("Compact the source table first");
+ assertThat(table(TARGET).snapshotManager().latestSnapshot()).isNull();
+ }
+
+ @Test
+ public void testCompactedDataEvolutionSourceIsCopied() throws Exception {
+ createTable(
+ SOURCE,
+ Options.DATA_EVOLUTION,
+ false,
+
Collections.singletonMap(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2"));
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ updateColumn(table(SOURCE), 0, "a2", "b2");
+ compactDataEvolution(table(SOURCE));
+ assertThat(liveFiles(table(SOURCE))).hasSize(1);
+ createTable(TARGET, Options.DATA_EVOLUTION, false);
+ write(table(TARGET), row(10, "x", "p1"));
+
+ copy(SOURCE, TARGET, null);
+
+ FileStoreTable target = table(TARGET);
+ assertThat(values(target)).containsEntry(1, "a2").containsEntry(2,
"b2");
+ assertThat(rowIds(target)).containsEntry(1, 1L).containsEntry(2, 2L);
+ // later column updates of the copied rows win
+ updateColumn(target, 1, "a3", "b3");
+ assertThat(values(table(TARGET))).containsEntry(1,
"a3").containsEntry(2, "b3");
+ }
+
+ @Test
+ public void testBlobFilesKeepTheRowIdsOfTheirRows() throws Exception {
+ createBlobTable(SOURCE);
+ writeBlobs(table(SOURCE), 1, 2);
+ writeBlobs(table(SOURCE), 3);
+ assertThat(liveFiles(table(SOURCE))).anyMatch(file ->
file.fileName().endsWith(".blob"));
+ createBlobTable(TARGET);
+ writeBlobs(table(TARGET), 10);
+
+ copy(SOURCE, TARGET, null);
+
+ // the copy replaces the row of the target; every copied row reads its
own blob, under
+ // row ids after the one the target gave out
+ Map<Integer, Long> rowIds = new TreeMap<>();
+ RowType readType =
+ SpecialFields.rowTypeWithRowTracking(table(TARGET).rowType(),
true, true);
+ for (InternalRow row : read(table(TARGET), readType)) {
+ int id = row.getInt(0);
+ assertThat(row.getBlob(1).toData()).isEqualTo(blob(id));
+ rowIds.put(id, row.getLong(2));
+ }
+ assertThat(rowIds).containsExactly(entry(1, 1L), entry(2, 2L),
entry(3, 3L));
+ }
+
+ @Test
+ public void testDedicatedFilesFollowingTheirRowsAreAccepted() throws
Exception {
+ createTable(TARGET, Options.DATA_EVOLUTION, false);
+ // vector-store files rolled at other rows than the normal files, two
blob files for one
+ // normal file, and a blob column added after the first rows
+ List<DataFileMeta> vectors =
+ files(
+ file("n0.parquet", 0, 100),
+ file("n1.parquet", 100, 100),
+ file("v0.vector.parquet", 0, 150, "e"),
+ file("v1.vector.parquet", 150, 50, "e"));
+ List<DataFileMeta> blobs =
+ files(
+ file("n2.parquet", 200, 2),
+ file("n3.parquet", 202, 4),
+ file("b0.blob", 202, 2, "b"),
+ file("b1.blob", 204, 2, "b"));
+
+ CopiedDataFiles.adaptToTarget(table(TARGET),
Collections.singletonList(vectors));
+ CopiedDataFiles.adaptToTarget(table(TARGET),
Collections.singletonList(blobs));
+
+ assertThat(vectors)
+ .extracting(DataFileMeta::fileName)
+ .containsExactly(
+ "n0.parquet", "v0.vector.parquet", "n1.parquet",
"v1.vector.parquet");
+ assertThat(blobs)
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("n2.parquet", "n3.parquet", "b0.blob",
"b1.blob");
+ }
+
+ @Test
+ public void testDedicatedFilesTheCommitWouldGiveOtherRowsAreRefused()
throws Exception {
+ createTable(TARGET, Options.DATA_EVOLUTION, false);
+ // a vector column added after the first rows: the vector-store file
would get rows 0 and 1
+ assertRefused(
+ "v1.vector.parquet",
+ files(
+ file("n0.parquet", 0, 2),
+ file("n1.parquet", 2, 2),
+ file("v1.vector.parquet", 2, 2, "e")));
+ // a vector column updated
+ assertRefused(
+ "u0.vector.parquet",
+ files(
+ file("n0.parquet", 0, 2),
+ file("v0.vector.parquet", 0, 2, "e"),
+ file("u0.vector.parquet", 0, 1, "e")));
+ // a blob column updated
+ assertRefused(
+ "u0.blob",
+ files(
+ file("n0.parquet", 0, 2),
+ file("b0.blob", 0, 2, "b"),
+ file("u0.blob", 1, 1, "b")));
+ // a bucket without vector-store files would move those of the other
buckets
+ assertRefused(
+ "n0.parquet",
+ files(file("n0.parquet", 0, 2)),
+ files(file("n1.parquet", 2, 2), file("v1.vector.parquet", 2,
2, "e")));
+ }
+
+ @Test
+ public void testFilesThatStoreTheirRowIdsAreRefusedByRowTrackingTargets()
throws Exception {
+ createTable(SOURCE, Options.ROW_TRACKING, false);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ copyOnWriteUpdate(table(SOURCE));
+
assertThat(liveFiles(table(SOURCE))).anyMatch(CopiedDataFilesTest::storesRowIds);
+
+ createTable(TARGET, Options.ROW_TRACKING, false);
+ assertThatThrownBy(() -> copy(SOURCE, TARGET, null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("its rows store their row ids")
+ .hasMessageContaining("INSERT OVERWRITE");
+ assertThat(table(TARGET).snapshotManager().latestSnapshot()).isNull();
+
+ // a table without row tracking does not use them
+ Identifier plain = new Identifier("default", "plain");
+ createTable(plain, Options.PLAIN, false);
+ copy(SOURCE, plain, null);
+
assertThat(liveFiles(table(plain))).anyMatch(CopiedDataFilesTest::storesRowIds);
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // file metadata and sequence numbers
+ //
---------------------------------------------------------------------------------------------
+
+ @Test
+ public void testLaterColumnUpdatesWinOverCopiedRows() throws Exception {
+ createTable(SOURCE, Options.DATA_EVOLUTION, false);
+ for (int i = 0; i < 20; i++) {
+ write(table(SOURCE), row(i, "v" + i, "p1"));
+ }
+ createTable(TARGET, Options.DATA_EVOLUTION, false);
+
+ copy(SOURCE, TARGET, null);
+
+ FileStoreTable target = table(TARGET);
+ long copySnapshot = target.snapshotManager().latestSnapshotId();
+ assertThat(liveFiles(target))
+ .allMatch(
+ file ->
+ file.minSequenceNumber() == copySnapshot
+ && file.maxSequenceNumber() ==
copySnapshot);
+ updateColumn(target, 19, "updated");
+ assertThat(values(table(TARGET))).containsEntry(19,
"updated").containsEntry(18, "v18");
+ }
+
+ @Test
+ public void testCopiedFilesAreCommittedAsFilesWrittenForTheTarget() throws
Exception {
+ createTable(SOURCE, Options.DATA_EVOLUTION, false);
+ write(table(SOURCE), row(1, "a", "p1"));
+ write(table(SOURCE), row(2, "b", "p1"));
+ createTable(TARGET, Options.ROW_TRACKING, false);
+ List<DataFileMeta> source = liveFiles(table(SOURCE));
+ List<DataFileMeta> copied = new ArrayList<>();
+ for (DataFileMeta file : source) {
+ copied.add(file.withWriteColsSequences(new long[] {7L, 7L, 7L}));
+ }
+
+ CopiedDataFiles.adaptToTarget(table(TARGET),
Collections.singletonList(copied));
+
+ assertThat(copied).hasSize(source.size());
+ for (DataFileMeta file : copied) {
+ assertThat(file.firstRowId()).isNull();
+ assertThat(file.fileSource()).contains(FileSource.APPEND);
+ assertThat(file.minSequenceNumber()).isZero();
+ assertThat(file.maxSequenceNumber()).isZero();
+ assertThat(file.writeColsSequences()).isNull();
+ }
+ // in row id order
+
assertThat(copied.get(0).fileName()).isEqualTo(fileWithFirstRowId(source, 0L));
+ }
+
+ @Test
+ public void testTableWithoutRowTrackingDropsCopiedRowIds() throws
Exception {
+ createTable(SOURCE, Options.ROW_TRACKING, false);
+ write(table(SOURCE), row(1, "a", "p1"), row(2, "b", "p1"));
+ createTable(TARGET, Options.PLAIN, false);
+
+ copy(SOURCE, TARGET, null);
+
+ assertThat(values(table(TARGET))).containsOnlyKeys(1, 2);
+ assertThat(liveFiles(table(TARGET))).allMatch(file ->
file.firstRowId() == null);
+ }
+
+ @Test
+ public void testCopyBetweenTablesWithoutRowTrackingChangesNothing() throws
Exception {
+ createTable(SOURCE, Options.PLAIN, false);
+ write(table(SOURCE), row(1, "a", "p1"));
+ createTable(TARGET, Options.PLAIN, false);
+ write(table(TARGET), row(10, "x", "p1"));
+ List<DataFileMeta> before = liveFiles(table(SOURCE));
+
+ List<DataFileMeta> copied = new ArrayList<>(before);
+ CopiedDataFiles.adaptToTarget(table(TARGET),
Collections.singletonList(copied));
+
+ assertThat(copied).isEqualTo(before);
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // helpers
+ //
---------------------------------------------------------------------------------------------
+
+ private static DataFileMeta file(
+ String name, long firstRowId, long rowCount, String... writeCols) {
+ return DataFileMeta.forAppend(
+ name,
+ 1L,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 0L,
+ 0L,
+ 0L,
+ Collections.emptyList(),
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ firstRowId,
+ writeCols.length == 0 ? null : Arrays.asList(writeCols));
+ }
+
+ private static List<DataFileMeta> files(DataFileMeta... files) {
+ return new ArrayList<>(Arrays.asList(files));
+ }
+
+ @SafeVarargs
+ private final void assertRefused(String fileName, List<DataFileMeta>...
buckets)
+ throws Exception {
+ FileStoreTable target = table(TARGET);
+ assertThatThrownBy(() -> CopiedDataFiles.adaptToTarget(target,
Arrays.asList(buckets)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot copy data file " + fileName)
+ .hasMessageContaining("row ids of other rows");
+ }
+
+ private enum Options {
+ PLAIN,
+ ROW_TRACKING,
+ DATA_EVOLUTION
+ }
+
+ private void createTable(Identifier identifier, Options options, boolean
partitioned)
+ throws Exception {
+ createTable(identifier, options, partitioned, Collections.emptyMap());
+ }
+
+ private void createTable(
+ Identifier identifier,
+ Options options,
+ boolean partitioned,
+ Map<String, String> extraOptions)
+ throws Exception {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("v", DataTypes.STRING())
+ .column("pt", DataTypes.STRING())
+ .options(extraOptions);
+ if (partitioned) {
+ builder.partitionKeys("pt");
+ }
+ if (options != Options.PLAIN) {
+ builder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ }
+ if (options == Options.DATA_EVOLUTION) {
+ builder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ }
+ catalog.createTable(identifier, builder.build(), false);
+ }
+
+ private void createBlobTable(Identifier identifier) throws Exception {
+ catalog.createTable(
+ identifier,
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("b", DataTypes.BLOB())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .build(),
+ false);
+ }
+
+ private static byte[] blob(int id) {
+ return ("blob-" + id).getBytes();
+ }
+
+ private static void writeBlobs(FileStoreTable table, int... ids) throws
Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int id : ids) {
+ write.write(GenericRow.of(id, new BlobData(blob(id))));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private FileStoreTable table(Identifier identifier) throws Exception {
+ return (FileStoreTable) catalog.getTable(identifier);
+ }
+
+ private static GenericRow row(int id, String v, String pt) {
+ return GenericRow.of(id, BinaryString.fromString(v),
BinaryString.fromString(pt));
+ }
+
+ private static void write(FileStoreTable table, GenericRow... rows) throws
Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (GenericRow row : rows) {
+ write.write(row);
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ /** A data-evolution update of column {@code v} of the rows from {@code
firstRowId} on. */
+ private static void updateColumn(FileStoreTable table, long firstRowId,
String... values)
+ throws Exception {
+ RowType writeType =
table.rowType().project(Collections.singletonList("v"));
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ for (String value : values) {
+ write.write(GenericRow.of(BinaryString.fromString(value)));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ for (CommitMessage message : messages) {
+ CommitMessageImpl impl = (CommitMessageImpl) message;
+ List<DataFileMeta> files = new
ArrayList<>(impl.newFilesIncrement().newFiles());
+ impl.newFilesIncrement().newFiles().clear();
+ files.forEach(
+ f ->
+ impl.newFilesIncrement()
+ .newFiles()
+ .add(f.assignFirstRowId(firstRowId)));
+ }
+ commit.commit(messages);
+ }
+ }
+
+ private static void compactDataEvolution(FileStoreTable table) throws
Exception {
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(
+ table, false, false,
table.snapshotManager().latestSnapshot());
+ List<CommitMessage> messages = new ArrayList<>();
+ for (DataEvolutionCompactTask task : coordinator.plan()) {
+ messages.add(task.doCompact(table, "compact"));
+ }
+ assertThat(messages).isNotEmpty();
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ /**
+ * Like a copy-on-write UPDATE of the first file of a row-tracking table:
rewrites it with the
+ * row ids of its rows stored in the new file.
+ */
+ private static void copyOnWriteUpdate(FileStoreTable table) throws
Exception {
+ DataFileMeta rewritten = null;
+ BinaryRow partition = null;
+ for (ManifestEntry entry : entries(table)) {
+ if (entry.file().firstRowId() != null && entry.file().firstRowId()
== 0L) {
+ rewritten = entry.file();
+ partition = entry.partition();
+ }
+ }
+ assertThat(rewritten).isNotNull();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+ .withWriteType(
+ SpecialFields.rowTypeWithRowTracking(
+ table.rowType(), false, true));
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(
+ GenericRow.of(
+ 1,
+ BinaryString.fromString("A"),
+ BinaryString.fromString("p1"),
+ 0L,
+ null));
+ write.write(
+ GenericRow.of(
+ 2,
+ BinaryString.fromString("b"),
+ BinaryString.fromString("p1"),
+ 1L,
+ null));
+ CommitMessageImpl message = (CommitMessageImpl)
write.prepareCommit().get(0);
+ commit.commit(
+ Collections.singletonList(
+ new CommitMessageImpl(
+ partition,
+ message.bucket(),
+ message.totalBuckets(),
+ new DataIncrement(
+
message.newFilesIncrement().newFiles(),
+
Collections.singletonList(rewritten),
+ Collections.emptyList()),
+ CompactIncrement.emptyIncrement())));
+ }
+ }
+
+ /** Copies and commits like {@code CopyFilesCommitOperator}, optionally
one partition only. */
+ private void copy(Identifier source, Identifier target, String partition)
throws Exception {
+ commitCopy(table(target), prepareCopy(source, target, partition));
+ }
+
+ private Map<Pair<BinaryRow, Integer>, List<DataFileMeta>> prepareCopy(
+ Identifier source, Identifier target) throws Exception {
+ return prepareCopy(source, target, null);
+ }
+
+ /** Copies the data files physically and adapts their metadata to the
target. */
+ private Map<Pair<BinaryRow, Integer>, List<DataFileMeta>> prepareCopy(
+ Identifier source, Identifier target, String partition) throws
Exception {
+ FileStoreTable from = table(source);
+ FileStoreTable to = table(target);
+ Map<Pair<BinaryRow, Integer>, List<DataFileMeta>> files = new
LinkedHashMap<>();
+ for (ManifestEntry entry : entries(from)) {
+ if (partition != null
+ && (entry.partition().getFieldCount() == 0
+ ||
!partition.equals(entry.partition().getString(0).toString()))) {
+ continue;
+ }
+ Path fromBucket =
+ from.store().pathFactory().bucketPath(entry.partition(),
entry.bucket());
+ Path toBucket =
to.store().pathFactory().bucketPath(entry.partition(), entry.bucket());
+ to.fileIO().mkdirs(toBucket);
+ to.fileIO()
+ .copyFile(
+ new Path(fromBucket, entry.file().fileName()),
+ new Path(toBucket, entry.file().fileName()),
+ false);
+ files.computeIfAbsent(
+ Pair.of(entry.partition().copy(), entry.bucket()),
+ k -> new ArrayList<>())
+ .add(entry.file());
+ }
+ CopiedDataFiles.adaptToTarget(to, files.values());
+ return files;
+ }
+
+ /** The overwrite commit of {@code CopyFilesCommitOperator}. */
+ private static void commitCopy(
+ FileStoreTable target, Map<Pair<BinaryRow, Integer>,
List<DataFileMeta>> files)
+ throws Exception {
+ List<CommitMessage> messages = new ArrayList<>();
+ for (Map.Entry<Pair<BinaryRow, Integer>, List<DataFileMeta>> entry :
files.entrySet()) {
+ messages.add(
+ new CommitMessageImpl(
+ entry.getKey().getLeft(),
+ entry.getKey().getRight(),
+ target.coreOptions().bucket(),
+ new DataIncrement(
+ entry.getValue(),
+ Collections.emptyList(),
+ Collections.emptyList()),
+ CompactIncrement.emptyIncrement()));
+ }
+ try (BatchTableCommit commit =
target.newBatchWriteBuilder().withOverwrite().newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ private static String fileWithFirstRowId(List<DataFileMeta> files, long
firstRowId) {
+ for (DataFileMeta file : files) {
+ if (file.firstRowId() != null && file.firstRowId() == firstRowId) {
+ return file.fileName();
+ }
+ }
+ throw new AssertionError("no file with first row id " + firstRowId);
+ }
+
+ private static List<ManifestEntry> entries(FileStoreTable table) {
+ List<ManifestEntry> entries = new ArrayList<>();
+
table.newSnapshotReader().readFileIterator().forEachRemaining(entries::add);
+ return entries;
+ }
+
+ private static List<DataFileMeta> liveFiles(FileStoreTable table) {
+ List<DataFileMeta> files = new ArrayList<>();
+ entries(table).forEach(entry -> files.add(entry.file()));
+ return files;
+ }
+
+ private static boolean storesRowIds(DataFileMeta file) {
+ return file.writeCols() != null &&
file.writeCols().contains(SpecialFields.ROW_ID.name());
+ }
+
+ private static Map<Integer, String> values(FileStoreTable table) throws
Exception {
+ Map<Integer, String> result = new TreeMap<>();
+ for (InternalRow row : read(table, table.rowType())) {
+ result.put(row.getInt(0), row.getString(1).toString());
+ }
+ return result;
+ }
+
+ private static Map<Integer, Long> rowIds(FileStoreTable table) throws
Exception {
+ Map<Integer, Long> result = new TreeMap<>();
+ RowType readType =
SpecialFields.rowTypeWithRowTracking(table.rowType(), true, true);
+ for (InternalRow row : read(table, readType)) {
+ result.put(row.getInt(0), row.isNullAt(3) ? null : row.getLong(3));
+ }
+ return result;
+ }
+
+ private static void assertUniqueRowIds(FileStoreTable table) throws
Exception {
+ List<Long> ids = new ArrayList<>(rowIds(table).values());
+ assertThat(ids).doesNotContainNull();
+ assertThat(new HashSet<>(ids)).hasSameSizeAs(ids);
+ }
+
+ private static List<InternalRow> read(FileStoreTable table, RowType
readType) throws Exception {
+ ReadBuilder readBuilder =
table.newReadBuilder().withReadType(readType);
+ TableRead read = readBuilder.newRead();
+ InternalRowSerializer serializer = new InternalRowSerializer(readType);
+ List<InternalRow> rows = new ArrayList<>();
+ for (Split split : readBuilder.newScan().plan().splits()) {
+ try (RecordReader<InternalRow> reader = read.createReader(split)) {
+ reader.forEachRemaining(row -> rows.add(serializer.copy(row)));
+ }
+ }
+ return rows;
+ }
+
+ /** Pauses the committer thread at its first manifest write. */
+ private static class PausingFileIO extends LocalFileIO {
+
+ private final AtomicReference<Thread> committer;
+ private final CountDownLatch paused;
+ private final CountDownLatch released;
+ private final AtomicBoolean pausedOnce = new AtomicBoolean();
+
+ private PausingFileIO(
+ AtomicReference<Thread> committer, CountDownLatch paused,
CountDownLatch released) {
+ this.committer = committer;
+ this.paused = paused;
+ this.released = released;
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(Path path, boolean
overwrite)
+ throws IOException {
+ if (Thread.currentThread() == committer.get()
+ && path.getParent().getName().equals("manifest")
+ && pausedOnce.compareAndSet(false, true)) {
+ paused.countDown();
+ try {
+ released.await(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+ return super.newOutputStream(path, overwrite);
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
index aa95263808..38dc6599cb 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
@@ -466,6 +466,8 @@ public class DataEvolutionTableTest extends
DataEvolutionTestBase {
BatchTableCommit commit = builder.newCommit();
List<CommitMessage> commitables = write1.prepareCommit();
+ // a column update of the rows written above
+ setFirstRowId(commitables, 0L);
commit.commit(commitables);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/RowTrackingCopiedFilesTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/RowTrackingCopiedFilesTest.java
new file mode 100644
index 0000000000..3ab5c2b4d2
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/RowTrackingCopiedFilesTest.java
@@ -0,0 +1,187 @@
+/*
+ * 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.table;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for committing files that already carry row ids into a row-tracking
table, as {@code
+ * sys.copy} does.
+ */
+public class RowTrackingCopiedFilesTest extends TableTestBase {
+
+ private static final Identifier SOURCE = new Identifier("default", "src");
+ private static final Identifier TARGET = new Identifier("default", "dst");
+
+ @Test
+ public void testRowsWrittenAfterCopiedFilesGetNewRowIds() throws Exception
{
+ FileStoreTable source = createRowTrackingTable(SOURCE);
+ writeRows(source, row(1, "a"));
+ writeRows(source, row(2, "b"));
+ FileStoreTable target = createRowTrackingTable(TARGET);
+
+ commitCopiedFiles(source, target, Collections.emptyList());
+ target = table(TARGET);
+ assertThat(rowIds(target)).containsEntry(1, 0L).containsEntry(2, 1L);
+ // the next row id moves past the row ids of the copied files
+
assertThat(target.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(2L);
+
+ writeRows(target, row(3, "c"), row(4, "d"));
+ assertThat(rowIds(table(TARGET)))
+ .containsEntry(1, 0L)
+ .containsEntry(2, 1L)
+ .containsEntry(3, 2L)
+ .containsEntry(4, 3L);
+ }
+
+ @Test
+ public void testNewFilesCommittedWithCopiedFilesGetRowIdsAfterThem()
throws Exception {
+ FileStoreTable source = createRowTrackingTable(SOURCE);
+ writeRows(source, row(1, "a"), row(2, "b"));
+ FileStoreTable target = createRowTrackingTable(TARGET);
+
+ // a file written for the target, committed together with the copied
ones
+ List<DataFileMeta> newFiles;
+ try (BatchTableWrite write = target.newBatchWriteBuilder().newWrite())
{
+ write.write(row(3, "c"));
+ newFiles =
+ ((CommitMessageImpl) write.prepareCommit().get(0))
+ .newFilesIncrement()
+ .newFiles();
+ }
+ commitCopiedFiles(source, target, newFiles);
+
+ target = table(TARGET);
+ assertThat(rowIds(target)).containsEntry(1, 0L).containsEntry(2,
1L).containsEntry(3, 2L);
+
assertThat(target.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(3L);
+ }
+
+ /**
+ * Copies the data files of {@code source} into {@code target} and commits
them, with {@code
+ * extraFiles}, as {@code CopyFilesCommitOperator} does: with the row ids
they have.
+ */
+ private void commitCopiedFiles(
+ FileStoreTable source, FileStoreTable target, List<DataFileMeta>
extraFiles)
+ throws Exception {
+ List<DataFileMeta> files = new ArrayList<>();
+ source.newSnapshotReader().readFileIterator().forEachRemaining(e ->
files.add(e.file()));
+ Path sourceBucket =
source.store().pathFactory().bucketPath(BinaryRow.EMPTY_ROW, 0);
+ Path targetBucket =
target.store().pathFactory().bucketPath(BinaryRow.EMPTY_ROW, 0);
+ target.fileIO().mkdirs(targetBucket);
+ for (DataFileMeta file : files) {
+ target.fileIO()
+ .copyFile(
+ new Path(sourceBucket, file.fileName()),
+ new Path(targetBucket, file.fileName()),
+ false);
+ }
+ files.addAll(extraFiles);
+ CommitMessage message =
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ target.coreOptions().bucket(),
+ new DataIncrement(files, Collections.emptyList(),
Collections.emptyList()),
+ CompactIncrement.emptyIncrement());
+ try (BatchTableCommit commit =
target.newBatchWriteBuilder().withOverwrite().newCommit()) {
+ commit.commit(Collections.singletonList(message));
+ }
+ }
+
+ private FileStoreTable createRowTrackingTable(Identifier identifier)
throws Exception {
+ catalog.createTable(
+ identifier,
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("v", DataTypes.STRING())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .build(),
+ false);
+ return table(identifier);
+ }
+
+ private FileStoreTable table(Identifier identifier) throws Exception {
+ return (FileStoreTable) catalog.getTable(identifier);
+ }
+
+ private static GenericRow row(int id, String v) {
+ return GenericRow.of(id, BinaryString.fromString(v));
+ }
+
+ private static void writeRows(FileStoreTable table, GenericRow... rows)
throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (GenericRow row : rows) {
+ write.write(row);
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private static Map<Integer, Long> rowIds(FileStoreTable table) throws
Exception {
+ RowType readType =
SpecialFields.rowTypeWithRowTracking(table.rowType(), true, true);
+ ReadBuilder readBuilder =
table.newReadBuilder().withReadType(readType);
+ TableRead read = readBuilder.newRead();
+ InternalRowSerializer serializer = new InternalRowSerializer(readType);
+ Map<Integer, Long> result = new HashMap<>();
+ for (Split split : readBuilder.newScan().plan().splits()) {
+ try (RecordReader<InternalRow> reader = read.createReader(split)) {
+ reader.forEachRemaining(
+ r -> {
+ InternalRow row = serializer.copy(r);
+ result.put(row.getInt(0), row.isNullAt(2) ? null :
row.getLong(2));
+ });
+ }
+ }
+ return result;
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesCommitOperator.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesCommitOperator.java
index d2284e5e55..e2dd5d65cb 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesCommitOperator.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesCommitOperator.java
@@ -25,6 +25,7 @@ import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexFileMetaSerializer;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.migrate.CopiedDataFiles;
import org.apache.paimon.migrate.FileMetaUtils;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.BatchTableCommit;
@@ -71,6 +72,12 @@ public class CopyFilesCommitOperator extends
CopyFilesOperator {
Map<Tuple2<BinaryRow, Integer>, IndexFileInfo> indexFileMetaMap =
deserializeIndexFileMeta(indexCopyFileInfoRdd);
+ // the copied files keep the row ids and sequence numbers of the
source: adapt them to the
+ // target before committing them
+ List<List<DataFileMeta>> copiedFiles = new ArrayList<>();
+ dataFileMetaMap.values().forEach(info ->
copiedFiles.add(info.dataFileMetas()));
+ CopiedDataFiles.adaptToTarget(targetTable, copiedFiles);
+
// construct commit messages
List<CommitMessage> commitMessages = new ArrayList<>();
for (Map.Entry<Tuple2<BinaryRow, Integer>, DataFileInfo> entry :
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesUtil.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesUtil.java
index d9243cf823..5ec69b7501 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesUtil.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/CopyFilesUtil.java
@@ -24,6 +24,7 @@ import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.io.PojoDataFileMeta;
import org.apache.paimon.schema.TableSchema;
@@ -35,6 +36,9 @@ import java.io.IOException;
import java.util.List;
import java.util.Optional;
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+
/** Utils for copy files. */
public class CopyFilesUtil {
@@ -51,6 +55,22 @@ public class CopyFilesUtil {
}
}
+ /**
+ * A new path for the copy of the data file {@code sourceFileName}. A blob
or vector file keeps
+ * its kind in its name: that is how readers and commits tell it from the
normal file holding
+ * the same rows.
+ */
+ public static Path newDataFilePath(DataFilePathFactory pathFactory, String
sourceFileName) {
+ if (isVectorStoreFile(sourceFileName)) {
+ return
pathFactory.newVectorPath(DataFilePathFactory.formatIdentifier(sourceFileName));
+ }
+ if (isBlobFile(sourceFileName)) {
+ return pathFactory.newPathFromExtension(
+ sourceFileName.substring(sourceFileName.lastIndexOf('.')));
+ }
+ return pathFactory.newPath();
+ }
+
public static DataFileMeta toNewDataFileMeta(
DataFileMeta oldFileMeta, String newFileName, long newSchemaId) {
return toNewDataFileMeta(oldFileMeta, newFileName, newSchemaId, null);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/ListDataFilesOperator.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/ListDataFilesOperator.java
index 366fbc6cc6..cf144438fb 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/ListDataFilesOperator.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/copy/ListDataFilesOperator.java
@@ -109,10 +109,10 @@ public class ListDataFilesOperator extends
CopyFilesOperator {
manifestEntry.partition(),
manifestEntry.bucket())
.toPath(manifestEntry);
Path targetDataFilePath =
- targetFileStorePathFactory
- .createDataFilePathFactory(
- manifestEntry.partition(),
manifestEntry.bucket())
- .newPath();
+ CopyFilesUtil.newDataFilePath(
+ targetFileStorePathFactory.createDataFilePathFactory(
+ manifestEntry.partition(),
manifestEntry.bucket()),
+ manifestEntry.fileName());
DataFileMeta fileMeta = manifestEntry.file();
DataFileMeta targetFileMeta =
CopyFilesUtil.toNewDataFileMeta(
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java
index 08d529725a..67fa687311 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/copy/CopyFilesUtilTest.java
@@ -19,7 +19,9 @@
package org.apache.paimon.spark.copy;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.SimpleStats;
@@ -35,6 +37,32 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link CopyFilesUtil}. */
public class CopyFilesUtilTest {
+ @Test
+ void testNewDataFilePathKeepsTheKindOfTheFile() {
+ DataFilePathFactory pathFactory =
+ new DataFilePathFactory(
+ new Path("/target/bucket-0"),
+ "parquet",
+ "data-",
+ "changelog-",
+ false,
+ null,
+ null);
+
+ assertThat(newName(pathFactory, "data-a-0.parquet"))
+ .endsWith(".parquet")
+ .doesNotContain(".vector.");
+ assertThat(newName(pathFactory,
"data-a-0.vector.lance")).endsWith(".vector.lance");
+ assertThat(newName(pathFactory, "data-a-0.blob")).endsWith(".blob");
+ assertThat(newName(pathFactory, "data-a-0.video")).endsWith(".video");
+ }
+
+ private static String newName(DataFilePathFactory pathFactory, String
sourceFileName) {
+ String name = CopyFilesUtil.newDataFilePath(pathFactory,
sourceFileName).getName();
+ assertThat(name).startsWith("data-").doesNotStartWith("data-a-");
+ return name;
+ }
+
@Test
void testClearWriteColsSequencesWhenChangingSchemaId() {
DataFileMeta source =
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CopyFilesProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CopyFilesProcedureTest.scala
index 6c35af37e8..e9d43b81fd 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CopyFilesProcedureTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CopyFilesProcedureTest.scala
@@ -19,6 +19,7 @@
package org.apache.paimon.spark.procedure
import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.utils.ExceptionUtils
import org.apache.spark.sql.Row
@@ -225,4 +226,291 @@ class CopyFilesProcedureTest extends PaimonSparkTestBase {
}
}
}
+
+ test("Paimon copy files procedure: rows written after the copy get new row
ids") {
+ withTable("src", "dst") {
+ sql("CREATE TABLE src (id INT, v STRING) TBLPROPERTIES
('row-tracking.enabled' = 'true')")
+ sql("INSERT INTO src VALUES (1, 'a')")
+ sql("INSERT INTO src VALUES (2, 'b')")
+
+ checkAnswer(
+ sql("CALL sys.copy(source_table => 'src', target_table => 'dst')"),
+ Row(true) :: Nil)
+ // the copied rows keep their row ids
+ checkAnswer(sql("SELECT id, _ROW_ID FROM dst"), Seq(Row(1, 0), Row(2,
1)))
+
+ // rows written afterwards continue after them instead of reusing them
+ sql("INSERT INTO dst VALUES (3, 'c'), (4, 'd')")
+ checkAnswer(
+ sql("SELECT id, _ROW_ID FROM dst"),
+ Seq(Row(1, 0), Row(2, 1), Row(3, 2), Row(4, 3)))
+ }
+ }
+
+ test("Paimon copy files procedure: copied rows follow the row ids of rows
the copy keeps") {
+ withTable("src", "dst") {
+ sql("""
+ |CREATE TABLE src (id INT, v STRING, pt STRING) PARTITIONED BY (pt)
+ |TBLPROPERTIES ('row-tracking.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO src VALUES (1, 'a', 'p1'), (2, 'b', 'p1')")
+ sql("""
+ |CREATE TABLE dst (id INT, v STRING, pt STRING) PARTITIONED BY (pt)
+ |TBLPROPERTIES ('row-tracking.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO dst VALUES (10, 'x', 'p2'), (11, 'y', 'p2')")
+
+ // only partition p1 is overwritten: the rows of p2 keep row ids 0 and 1
+ checkAnswer(
+ sql(
+ """CALL sys.copy(source_table => 'src', target_table => 'dst', where
=> "pt = 'p1'")"""),
+ Row(true) :: Nil)
+ checkAnswer(
+ sql("SELECT id, v, _ROW_ID FROM dst"),
+ Seq(Row(1, "a", 2), Row(2, "b", 3), Row(10, "x", 0), Row(11, "y", 1)))
+
+ sql("INSERT INTO dst VALUES (3, 'c', 'p1')")
+ checkAnswer(sql("SELECT _ROW_ID FROM dst WHERE id = 3"), Row(4) :: Nil)
+ checkAnswer(sql("SELECT count(DISTINCT _ROW_ID), count(*) FROM dst"),
Row(5, 5) :: Nil)
+ }
+ }
+
+ test("Paimon copy files procedure: updates after copying a data-evolution
table take effect") {
+ withTable("src", "dst") {
+ sql("""
+ |CREATE TABLE src (id INT, v STRING) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'compaction.min.file-num' = '2')
+ |""".stripMargin)
+ // many snapshots: the source files carry sequence numbers above the
snapshots of dst
+ (1 to 10).foreach(i => sql(s"INSERT INTO src VALUES ($i, 'v$i')"))
+ sql("UPDATE src SET v = 'u1' WHERE id = 1")
+
+ // the update is a file holding the same rows as another one, which
cannot be copied
+ val error = intercept[Exception] {
+ sql("CALL sys.copy(source_table => 'src', target_table => 'dst')")
+ }
+ val trace = ExceptionUtils.stringifyException(error)
+ assert(trace.contains("Compact the source table first"), trace)
+ checkAnswer(sql("SELECT count(*) FROM dst"), Row(0) :: Nil)
+
+ sql("CALL sys.compact(table => 'src')")
+ checkAnswer(
+ sql("CALL sys.copy(source_table => 'src', target_table => 'dst')"),
+ Row(true) :: Nil)
+ checkAnswer(sql("SELECT id, v FROM dst"), sql("SELECT id, v FROM src"))
+ checkAnswer(sql("SELECT v FROM dst WHERE id = 1"), Row("u1") :: Nil)
+ checkAnswer(sql("SELECT count(DISTINCT _ROW_ID), count(*) FROM dst"),
Row(10, 10) :: Nil)
+
+ sql("UPDATE dst SET v = 'u2' WHERE id = 2")
+ sql("""
+ |MERGE INTO dst USING (SELECT _ROW_ID AS rid FROM dst WHERE id IN
(1, 10)) s
+ |ON dst._ROW_ID = s.rid
+ |WHEN MATCHED THEN UPDATE SET v = 'm'
+ |""".stripMargin)
+ checkAnswer(
+ sql("SELECT id, v FROM dst WHERE id IN (1, 2, 3, 10) ORDER BY id"),
+ Seq(Row(1, "m"), Row(2, "u2"), Row(3, "v3"), Row(10, "m")))
+ }
+ }
+
+ test("Paimon copy files procedure: files that store their row ids are
refused") {
+ withTable("src", "dst") {
+ sql("CREATE TABLE src (id INT, v STRING) TBLPROPERTIES
('row-tracking.enabled' = 'true')")
+ sql("INSERT INTO src VALUES (1, 'a'), (2, 'b')")
+ // a copy-on-write update stores the row ids of the rewritten rows in
the new file
+ sql("UPDATE src SET v = 'A' WHERE id = 1")
+ assert(
+ sql("SELECT write_cols FROM `src$files`")
+ .collect()
+ .exists(row => !row.isNullAt(0) &&
row.getSeq[String](0).contains("_ROW_ID")))
+
+ val error = intercept[Exception] {
+ sql("CALL sys.copy(source_table => 'src', target_table => 'dst')")
+ }
+ val trace = ExceptionUtils.stringifyException(error)
+ assert(trace.contains("its rows store their row ids"), trace)
+ }
+ }
+
+ test("Paimon copy files procedure: data-evolution table with a vector
field") {
+ withTable("src", "dst") {
+ sql("""
+ |CREATE TABLE src (id INT, embedding ARRAY<FLOAT>) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'file.format' = 'parquet',
+ | 'vector.file.format' = 'parquet',
+ | 'vector-field' = 'embedding',
+ | 'field.embedding.vector-dim' = '2')
+ |""".stripMargin)
+ sql("INSERT INTO src VALUES (1, array(1.0f, 0.0f)), (2, array(2.0f,
0.0f))")
+ val vectorFiles = fileNames("src").count(_.contains(".vector."))
+ assert(vectorFiles > 0, fileNames("src"))
+
+ checkAnswer(
+ sql("CALL sys.copy(source_table => 'src', target_table => 'dst')"),
+ Row(true) :: Nil)
+ // the vector file is copied as a vector file
+ assert(fileNames("dst").count(_.contains(".vector.")) == vectorFiles,
fileNames("dst"))
+ checkAnswer(
+ sql("SELECT id, embedding, _ROW_ID FROM dst"),
+ sql("SELECT id, embedding, _ROW_ID FROM src"))
+
+ sql("INSERT INTO dst VALUES (3, array(3.0f, 0.0f))")
+ checkAnswer(
+ sql("SELECT id, embedding, _ROW_ID FROM dst WHERE id = 3"),
+ Row(3, Seq(3.0f, 0.0f), 2) :: Nil)
+ }
+ }
+
+ test("Paimon copy files procedure: vector field added after the first rows")
{
+ withTable("src", "dst") {
+ sql("""
+ |CREATE TABLE src (id INT) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'file.format' = 'parquet',
+ | 'vector.file.format' = 'parquet')
+ |""".stripMargin)
+ // these rows have no vector file
+ sql("INSERT INTO src VALUES (1), (2)")
+ sql("ALTER TABLE src ADD COLUMN embedding ARRAY<FLOAT> COMMENT
'__VECTOR_FIELD;2'")
+ sql("INSERT INTO src VALUES (3, array(3.0f, 0.0f)), (4, array(4.0f,
0.0f))")
+ assert(fileNames("src").exists(_.contains(".vector.")), fileNames("src"))
+
+ assertCopyRefused("src", "dst")
+ // rewriting the rows gives each of them its vector
+ sql("INSERT OVERWRITE src SELECT * FROM src")
+ checkAnswer(
+ sql("CALL sys.copy(source_table => 'src', target_table => 'dst')"),
+ Row(true) :: Nil)
+ checkAnswer(sql("SELECT id, embedding FROM dst"), sql("SELECT id,
embedding FROM src"))
+ checkAnswer(sql("SELECT count(DISTINCT _ROW_ID) = count(*) FROM dst"),
Row(true) :: Nil)
+ }
+ }
+
+ test("Paimon copy files procedure: vector column updated in the source") {
+ withTable("src", "dst") {
+ sql("""
+ |CREATE TABLE src (id INT, embedding ARRAY<FLOAT>) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'file.format' = 'parquet',
+ | 'vector.file.format' = 'parquet',
+ | 'vector-field' = 'embedding',
+ | 'field.embedding.vector-dim' = '2')
+ |""".stripMargin)
+ sql("INSERT INTO src VALUES (1, array(1.0f, 0.0f)), (2, array(2.0f,
0.0f))")
+ sql("UPDATE src SET embedding = array(9.0f, 0.0f) WHERE id = 1")
+
+ assertCopyRefused("src", "dst")
+ // rewriting the rows gives each of them its vector
+ sql("INSERT OVERWRITE src SELECT * FROM src")
+ checkAnswer(
+ sql("CALL sys.copy(source_table => 'src', target_table => 'dst')"),
+ Row(true) :: Nil)
+ checkAnswer(sql("SELECT id, embedding FROM dst"), sql("SELECT id,
embedding FROM src"))
+ checkAnswer(sql("SELECT count(DISTINCT _ROW_ID) = count(*) FROM dst"),
Row(true) :: Nil)
+ }
+ }
+
+ test("Paimon copy files procedure: data-evolution table with a blob field") {
+ withTable("src", "dst") {
+ sql("""
+ |CREATE TABLE src (id INT, picture BINARY) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'blob-field' = 'picture')
+ |""".stripMargin)
+ sql("INSERT INTO src VALUES (1, CAST('p1' AS BINARY)), (2, CAST('p2' AS
BINARY))")
+ val blobFiles = fileNames("src").count(_.endsWith(".blob"))
+ assert(blobFiles > 0, fileNames("src"))
+
+ checkAnswer(
+ sql("CALL sys.copy(source_table => 'src', target_table => 'dst')"),
+ Row(true) :: Nil)
+ // the blob file is copied as a blob file
+ assert(fileNames("dst").count(_.endsWith(".blob")) == blobFiles,
fileNames("dst"))
+ checkAnswer(
+ sql("SELECT id, CAST(picture AS STRING), _ROW_ID FROM dst"),
+ sql("SELECT id, CAST(picture AS STRING), _ROW_ID FROM src"))
+ checkAnswer(sql("SELECT CAST(picture AS STRING) FROM dst WHERE id = 2"),
Row("p2") :: Nil)
+ }
+ }
+
+ test("Paimon copy files procedure: partitioned data-evolution table with a
vector field") {
+ withTable("src", "dst") {
+ sql("""
+ |CREATE TABLE src (id INT, embedding ARRAY<FLOAT>, pt STRING)
+ |PARTITIONED BY (pt) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'file.format' = 'parquet',
+ | 'vector.file.format' = 'parquet',
+ | 'vector-field' = 'embedding',
+ | 'field.embedding.vector-dim' = '2')
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO src VALUES
+ |(1, array(1.0f, 0.0f), 'a'), (2, array(2.0f, 0.0f), 'b'), (3,
array(3.0f, 0.0f), 'c')
+ |""".stripMargin)
+ sql("INSERT INTO src VALUES (4, array(4.0f, 0.0f), 'a'), (5, array(5.0f,
0.0f), 'c')")
+
+ // the commit may order the partitions in any way
+ checkAnswer(
+ sql("CALL sys.copy(source_table => 'src', target_table => 'dst')"),
+ Row(true) :: Nil)
+ checkAnswer(sql("SELECT * FROM dst"), sql("SELECT * FROM src"))
+ checkAnswer(sql("SELECT count(DISTINCT _ROW_ID) = count(*) FROM dst"),
Row(true) :: Nil)
+
+ // only partition b is overwritten
+ sql("INSERT INTO src VALUES (6, array(6.0f, 0.0f), 'b')")
+ checkAnswer(
+ sql("""CALL sys.copy(source_table => 'src', target_table => 'dst',
where => "pt = 'b'")"""),
+ Row(true) :: Nil)
+ checkAnswer(sql("SELECT * FROM dst"), sql("SELECT * FROM src"))
+ checkAnswer(sql("SELECT count(DISTINCT _ROW_ID) = count(*) FROM dst"),
Row(true) :: Nil)
+ }
+ }
+
+ test("Paimon copy files procedure: partition written before the vector field
was added") {
+ withTable("src", "dst") {
+ sql("""
+ |CREATE TABLE src (id INT, pt STRING) PARTITIONED BY (pt)
TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'file.format' = 'parquet',
+ | 'vector.file.format' = 'parquet')
+ |""".stripMargin)
+ sql("INSERT INTO src VALUES (1, 'a'), (2, 'a')")
+ sql("ALTER TABLE src ADD COLUMN embedding ARRAY<FLOAT> COMMENT
'__VECTOR_FIELD;2'")
+ sql("INSERT INTO src VALUES (3, 'b', array(3.0f, 0.0f)), (4, 'b',
array(4.0f, 0.0f))")
+
+ // the rows of partition a have no vector file, which would move those
of partition b
+ assertCopyRefused("src", "dst")
+
+ // each partition alone has a vector file for all of its rows or for none
+ checkAnswer(
+ sql("""CALL sys.copy(source_table => 'src', target_table => 'dst',
where => "pt = 'b'")"""),
+ Row(true) :: Nil)
+ checkAnswer(
+ sql("""CALL sys.copy(source_table => 'src', target_table => 'dst',
where => "pt = 'a'")"""),
+ Row(true) :: Nil)
+ checkAnswer(sql("SELECT * FROM dst"), sql("SELECT * FROM src"))
+ }
+ }
+
+ private def assertCopyRefused(source: String, target: String): Unit = {
+ val error = intercept[Exception] {
+ sql(s"CALL sys.copy(source_table => '$source', target_table =>
'$target')")
+ }
+ val trace = ExceptionUtils.stringifyException(error)
+ assert(trace.contains("row ids of other rows"), trace)
+ checkAnswer(sql(s"SELECT count(*) FROM $target"), Row(0) :: Nil)
+ }
+
+ private def fileNames(table: String): Seq[String] =
+ sql(s"SELECT file_path FROM
`$table$$files`").collect().map(_.getString(0)).toSeq
}
