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 209a1ba679 [core][flink] Record row ID mappings for clustering output
(#10681)
209a1ba679 is described below
commit 209a1ba679d7d45dfbf877280a28e23be241ba39
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 12:46:20 2026 +0800
[core][flink] Record row ID mappings for clustering output (#10681)
---
.../dataevolution/ClusteringRowIdMapping.java | 113 ++++++++++
.../dataevolution/ClusteringRowIdMappingTest.java | 139 ++++++++++++
.../cluster/DataEvolutionClusterWriteOperator.java | 240 ++++++++++++++++++++
.../DataEvolutionClusterWriteOperatorTest.java | 250 +++++++++++++++++++++
4 files changed, 742 insertions(+)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringRowIdMapping.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringRowIdMapping.java
new file mode 100644
index 0000000000..09a62c8eba
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringRowIdMapping.java
@@ -0,0 +1,113 @@
+/*
+ * 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.append.dataevolution;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.Closeable;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Old row IDs in the physical row order of a clustering output file. */
+public final class ClusteringRowIdMapping {
+
+ private static final String PREFIX = "cluster-row-id-";
+ private static final String SUFFIX = ".mapping";
+ private static final int MAGIC = 0x43524944;
+ private static final int VERSION = 1;
+
+ private ClusteringRowIdMapping() {}
+
+ public static String fileName(String dataFileName) {
+ return PREFIX + dataFileName + SUFFIX;
+ }
+
+ public static boolean isMappingFile(String fileName) {
+ return fileName.startsWith(PREFIX) && fileName.endsWith(SUFFIX);
+ }
+
+ public static Writer createWriter(FileIO fileIO, Path path) throws
IOException {
+ return new Writer(fileIO, path);
+ }
+
+ public static void read(FileIO fileIO, Path path, long rowCount,
RowIdConsumer consumer)
+ throws IOException {
+ checkArgument(rowCount >= 0, "Invalid clustering row count: %s",
rowCount);
+ try (DataInputStream in =
+ new DataInputStream(new
BufferedInputStream(fileIO.newInputStream(path)))) {
+ checkArgument(in.readInt() == MAGIC, "Invalid clustering row ID
mapping: %s", path);
+ checkArgument(
+ in.readInt() == VERSION, "Unsupported clustering row ID
mapping: %s", path);
+ for (long ordinal = 0; ordinal < rowCount; ordinal++) {
+ long oldRowId = in.readLong();
+ checkArgument(oldRowId >= 0, "Invalid old row ID: %s",
oldRowId);
+ consumer.accept(ordinal, oldRowId);
+ }
+ checkArgument(in.read() == -1, "Unexpected trailing clustering row
IDs: %s", path);
+ }
+ }
+
+ /** Streaming consumer; the ordinal is local to the output data file. */
+ @FunctionalInterface
+ public interface RowIdConsumer {
+ void accept(long ordinal, long oldRowId) throws IOException;
+ }
+
+ /**
+ * Streaming writer, also used when splitting a spilled sort output at
actual file boundaries.
+ */
+ public static final class Writer implements Closeable {
+
+ private final DataOutputStream out;
+
+ private Writer(FileIO fileIO, Path path) throws IOException {
+ out =
+ new DataOutputStream(
+ new
BufferedOutputStream(fileIO.newOutputStream(path, false)));
+ try {
+ out.writeInt(MAGIC);
+ out.writeInt(VERSION);
+ } catch (IOException | RuntimeException | Error failure) {
+ try {
+ out.close();
+ } catch (IOException closeFailure) {
+ failure.addSuppressed(closeFailure);
+ }
+ fileIO.deleteQuietly(path);
+ throw failure;
+ }
+ }
+
+ public void write(long oldRowId) throws IOException {
+ checkArgument(oldRowId >= 0, "Invalid old row ID: %s", oldRowId);
+ out.writeLong(oldRowId);
+ }
+
+ @Override
+ public void close() throws IOException {
+ out.close();
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/ClusteringRowIdMappingTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/ClusteringRowIdMappingTest.java
new file mode 100644
index 0000000000..78ba1a7a36
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/ClusteringRowIdMappingTest.java
@@ -0,0 +1,139 @@
+/*
+ * 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.append.dataevolution;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.DataOutputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests the output-file ordinal mapping and rejects incomplete or corrupt
side files. */
+class ClusteringRowIdMappingTest {
+
+ private static final int MAGIC = 0x43524944;
+ private final FileIO fileIO = LocalFileIO.create();
+
+ @TempDir java.nio.file.Path tempPath;
+
+ @Test
+ void testRoundTripKeepsPhysicalOrder() throws Exception {
+ Path path = path("round-trip");
+ try (ClusteringRowIdMapping.Writer writer =
+ ClusteringRowIdMapping.createWriter(fileIO, path)) {
+ writer.write(10);
+ writer.write(0);
+ writer.write(Long.MAX_VALUE);
+ }
+ List<Long> ordinals = new ArrayList<>();
+ List<Long> oldRowIds = new ArrayList<>();
+ ClusteringRowIdMapping.read(
+ fileIO,
+ path,
+ 3,
+ (ordinal, oldRowId) -> {
+ ordinals.add(ordinal);
+ oldRowIds.add(oldRowId);
+ });
+ assertThat(ordinals).containsExactly(0L, 1L, 2L);
+ assertThat(oldRowIds).containsExactly(10L, 0L, Long.MAX_VALUE);
+ }
+
+ @Test
+ void testRejectsTruncatedAndTrailingRows() throws Exception {
+ Path path = raw("row-count", MAGIC, 1, 10L);
+ assertThatThrownBy(
+ () -> ClusteringRowIdMapping.read(fileIO, path, 2,
(ordinal, oldId) -> {}))
+ .isInstanceOf(EOFException.class);
+ assertThatThrownBy(
+ () -> ClusteringRowIdMapping.read(fileIO, path, 0,
(ordinal, oldId) -> {}))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unexpected trailing clustering row
IDs");
+ }
+
+ @Test
+ void testRejectsInvalidHeaderAndOldRowIds() throws Exception {
+ Path magic = raw("magic", 0, 1);
+ assertThatThrownBy(
+ () -> ClusteringRowIdMapping.read(fileIO, magic, 0,
(ordinal, oldId) -> {}))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid clustering row ID mapping");
+ Path version = raw("version", MAGIC, 2);
+ assertThatThrownBy(
+ () ->
+ ClusteringRowIdMapping.read(
+ fileIO, version, 0, (ordinal, oldId)
-> {}))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unsupported clustering row ID mapping");
+ Path negative = raw("negative", MAGIC, 1, -1);
+ assertThatThrownBy(
+ () ->
+ ClusteringRowIdMapping.read(
+ fileIO, negative, 1, (ordinal, oldId)
-> {}))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid old row ID");
+ assertThatThrownBy(
+ () ->
+ ClusteringRowIdMapping.read(
+ fileIO, negative, -1, (ordinal, oldId)
-> {}))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid clustering row count");
+ }
+
+ @Test
+ void testWriterRejectsNegativeRowIds() throws Exception {
+ Path path = path("invalid-write");
+ try (ClusteringRowIdMapping.Writer writer =
+ ClusteringRowIdMapping.createWriter(fileIO, path)) {
+ assertThatThrownBy(() -> writer.write(-1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid old row ID");
+ writer.write(3);
+ }
+ List<Long> rows = new ArrayList<>();
+ ClusteringRowIdMapping.read(fileIO, path, 1, (ordinal, oldId) ->
rows.add(oldId));
+ assertThat(rows).containsExactly(3L);
+ }
+
+ private Path raw(String name, int magic, int version, long... rows) throws
IOException {
+ Path path = path(name);
+ try (DataOutputStream out = new
DataOutputStream(fileIO.newOutputStream(path, false))) {
+ out.writeInt(magic);
+ out.writeInt(version);
+ for (long row : rows) {
+ out.writeLong(row);
+ }
+ }
+ return path;
+ }
+
+ private Path path(String name) {
+ return new Path(tempPath.resolve(name).toUri());
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/cluster/DataEvolutionClusterWriteOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/cluster/DataEvolutionClusterWriteOperator.java
new file mode 100644
index 0000000000..4199c5d49e
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/cluster/DataEvolutionClusterWriteOperator.java
@@ -0,0 +1,240 @@
+/*
+ * 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.flink.cluster;
+
+import org.apache.paimon.append.dataevolution.ClusteringRowIdMapping;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.flink.memory.FlinkMemorySegmentPool;
+import org.apache.paimon.flink.memory.MemorySegmentAllocator;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+import org.apache.paimon.memory.MemoryPoolFactory;
+import org.apache.paimon.memory.MemorySegmentPool;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.ProjectedRow;
+
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.IntStream;
+
+import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_USE_MANAGED_MEMORY;
+import static
org.apache.paimon.flink.utils.ManagedMemoryUtils.computeManagedMemory;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** Writes sorted rows and records their original row IDs in output-file
order. */
+public class DataEvolutionClusterWriteOperator
+ extends BoundedOneInputOperator<InternalRow, Committable> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FileStoreTable table;
+ private final BinaryRow partition;
+ private final String commitUser;
+
+ private transient IOManager ioManager;
+ private transient BatchTableWrite write;
+ private transient ProjectedRow valueProjection;
+ private transient File rowIdSpool;
+ private transient DataOutputStream rowIds;
+ private transient List<Path> mappingPaths;
+ private transient List<Path> preparedDataPaths;
+ private transient MemorySegmentAllocator memoryAllocator;
+ private transient boolean emitted;
+
+ public DataEvolutionClusterWriteOperator(
+ FileStoreTable table, BinaryRow partition, String commitUser) {
+ this.table = table;
+ this.partition = partition;
+ this.commitUser = commitUser;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ ioManager =
+ IOManager.create(
+ getContainingTask()
+ .getEnvironment()
+ .getIOManager()
+ .getSpillingDirectoriesPaths());
+ rowIdSpool = ioManager.createChannel("cluster-row-ids").getPathFile();
+ rowIds = new DataOutputStream(new BufferedOutputStream(new
FileOutputStream(rowIdSpool)));
+ mappingPaths = new ArrayList<>();
+ preparedDataPaths = new ArrayList<>();
+ valueProjection =
+ ProjectedRow.from(IntStream.range(0,
table.rowType().getFieldCount()).toArray());
+ // Unaware append writers use NoopCompactManager. Their data files,
including files
+ // produced by buffer flushes or spills, retain each partition's input
order.
+ write = table.newWrite(commitUser).withIOManager(ioManager);
+ MemorySegmentPool memoryPool;
+ if (Options.fromMap(table.options()).get(SINK_USE_MANAGED_MEMORY)) {
+ MemoryManager memoryManager =
getContainingTask().getEnvironment().getMemoryManager();
+ memoryAllocator = new MemorySegmentAllocator(this, memoryManager);
+ memoryPool =
+ new FlinkMemorySegmentPool(
+ computeManagedMemory(this),
+ memoryManager.getPageSize(),
+ memoryAllocator);
+ } else {
+ memoryPool =
+ new HeapMemorySegmentPool(
+ table.coreOptions().writeBufferSize(),
table.coreOptions().pageSize());
+ }
+ write.withMemoryPoolFactory(new MemoryPoolFactory(memoryPool));
+ }
+
+ @Override
+ public void processElement(StreamRecord<InternalRow> element) throws
Exception {
+ InternalRow row = element.getValue();
+ int rowIdPosition = table.rowType().getFieldCount();
+ checkArgument(!row.isNullAt(rowIdPosition), "Clustering source row ID
must not be null.");
+ long oldRowId = row.getLong(rowIdPosition);
+ InternalRow value = valueProjection.replaceRow(row);
+ checkArgument(
+ write.getPartition(value).equals(partition),
+ "Clustering writer received a row from another partition.");
+ write.write(value, 0);
+ rowIds.writeLong(oldRowId);
+ }
+
+ @Override
+ public void endInput() throws Exception {
+ rowIds.close();
+ rowIds = null;
+ List<CommitMessage> messages = write.prepareCommit();
+ // prepareCommit transfers these files out of the writer's cleanup
ownership. Keep them
+ // until all mapping files and committables have been produced
successfully.
+ for (CommitMessage message : messages) {
+ DataFilePathFactory preparedPathFactory =
+ table.store()
+ .pathFactory()
+ .createDataFilePathFactory(message.partition(),
message.bucket());
+ for (DataFileMeta file : ((CommitMessageImpl)
message).newFilesIncrement().newFiles()) {
+
preparedDataPaths.addAll(file.collectFiles(preparedPathFactory));
+ }
+ }
+ DataFilePathFactory pathFactory =
+
table.store().pathFactory().createDataFilePathFactory(partition, 0);
+ List<Committable> committables = new ArrayList<>();
+ try (DataInputStream input =
+ new DataInputStream(new BufferedInputStream(new
FileInputStream(rowIdSpool)))) {
+ for (CommitMessage message : messages) {
+ CommitMessageImpl impl = (CommitMessageImpl) message;
+ DataIncrement increment = impl.newFilesIncrement();
+ checkState(
+ message.partition().equals(partition)
+ && message.bucket() == 0
+ && impl.compactIncrement().isEmpty()
+ && increment.deletedFiles().isEmpty()
+ && increment.changelogFiles().isEmpty()
+ && increment.newIndexFiles().isEmpty()
+ && increment.deletedIndexFiles().isEmpty(),
+ "Clustering writer must only append files for its
partition.");
+ List<DataFileMeta> files = new ArrayList<>();
+ for (DataFileMeta file : increment.newFiles()) {
+ checkState(
+ file.firstRowId() == null, "Clustering output must
get new row IDs.");
+ String mappingName =
ClusteringRowIdMapping.fileName(file.fileName());
+ Path mappingPath = pathFactory.toAlignedPath(mappingName,
file);
+ mappingPaths.add(mappingPath);
+ try (ClusteringRowIdMapping.Writer mapping =
+
ClusteringRowIdMapping.createWriter(table.fileIO(), mappingPath)) {
+ for (long i = 0; i < file.rowCount(); i++) {
+ mapping.write(input.readLong());
+ }
+ }
+ List<String> extraFiles = new
ArrayList<>(file.extraFiles());
+ extraFiles.add(mappingName);
+ files.add(file.copy(extraFiles));
+ }
+ committables.add(
+ new Committable(
+ Long.MAX_VALUE,
+ new CommitMessageImpl(
+ message.partition(),
+ message.bucket(),
+ message.totalBuckets(),
+ new DataIncrement(
+ files,
+ increment.deletedFiles(),
+ increment.changelogFiles()),
+ impl.compactIncrement())));
+ }
+ checkState(input.read() == -1, "Clustering output lost rows from
its input.");
+ }
+ for (Committable committable : committables) {
+ output.collect(new StreamRecord<>(committable));
+ }
+ emitted = true;
+ }
+
+ @Override
+ public void close() throws Exception {
+ try {
+ if (rowIds != null) {
+ rowIds.close();
+ }
+ } finally {
+ try {
+ if (write != null) {
+ write.close();
+ }
+ } finally {
+ try {
+ if (memoryAllocator != null) {
+ memoryAllocator.release();
+ }
+ if (!emitted && mappingPaths != null) {
+ mappingPaths.forEach(table.fileIO()::deleteQuietly);
+ }
+ if (!emitted && preparedDataPaths != null) {
+
preparedDataPaths.forEach(table.fileIO()::deleteQuietly);
+ }
+ if (ioManager != null) {
+ ioManager.close();
+ }
+ } finally {
+ super.close();
+ }
+ }
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/cluster/DataEvolutionClusterWriteOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/cluster/DataEvolutionClusterWriteOperatorTest.java
new file mode 100644
index 0000000000..bfaf23bfe7
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/cluster/DataEvolutionClusterWriteOperatorTest.java
@@ -0,0 +1,250 @@
+/*
+ * 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.flink.cluster;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.append.dataevolution.ClusteringRowIdMapping;
+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.flink.sink.Committable;
+import org.apache.paimon.flink.sink.CommittableTypeInfo;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.FileSystemSchemaManager;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.CloseableIterator;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests that old row IDs follow actual output boundaries, including buffered
spills. */
+class DataEvolutionClusterWriteOperatorTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @ParameterizedTest
+ @CsvSource({"false,false", "true,false", "true,true"})
+ void testMappingsFollowWrittenRowsAcrossFileRolls(boolean buffered,
boolean spillable)
+ throws Exception {
+ FileStoreTable table = createTable(buffered, spillable);
+ DataEvolutionClusterWriteOperator operator =
+ new DataEvolutionClusterWriteOperator(table,
BinaryRow.EMPTY_ROW, "cluster-test");
+ List<DataFileMeta> files;
+ int rowCount = 250;
+ char[] payload = new char[2048];
+ java.util.Arrays.fill(payload, 'x');
+ try (OneInputStreamOperatorTestHarness<InternalRow, Committable>
harness =
+ new OneInputStreamOperatorTestHarness<>(operator)) {
+ harness.setup(new CommittableTypeInfo().createSerializer(new
ExecutionConfig()));
+ harness.open();
+ for (int id = 0; id < rowCount; id++) {
+ harness.processElement(
+ new StreamRecord<>(
+ GenericRow.of(
+ id,
+
BinaryString.fromString(String.valueOf(payload)),
+ oldRowId(id, rowCount))));
+ }
+ operator.endInput();
+ files =
+ harness.extractOutputValues().stream()
+ .flatMap(
+ committable ->
+ ((CommitMessageImpl)
committable.commitMessage())
+
.newFilesIncrement().newFiles().stream())
+ .collect(Collectors.toList());
+ }
+
+ assertThat(files).hasSizeGreaterThan(2);
+ assertThat(files).allSatisfy(file ->
assertThat(file.firstRowId()).isNull());
+ assertThat(table.latestSnapshot()).isEmpty();
+ DataFilePathFactory pathFactory =
+
table.store().pathFactory().createDataFilePathFactory(BinaryRow.EMPTY_ROW, 0);
+ List<Integer> writtenIds = new ArrayList<>();
+ for (DataFileMeta file : files) {
+ List<Integer> ids = readIds(table, file);
+ assertThat(ids).hasSize((int) file.rowCount());
+ writtenIds.addAll(ids);
+ String mappingName =
ClusteringRowIdMapping.fileName(file.fileName());
+ assertThat(file.extraFiles()).contains(mappingName);
+ ClusteringRowIdMapping.read(
+ table.fileIO(),
+ pathFactory.toAlignedPath(mappingName, file),
+ file.rowCount(),
+ (ordinal, oldId) ->
+ assertThat(oldId)
+ .isEqualTo(oldRowId(ids.get((int)
ordinal), rowCount)));
+ }
+ assertThat(writtenIds)
+ .containsExactlyElementsOf(
+ IntStream.range(0,
rowCount).boxed().collect(Collectors.toList()));
+ }
+
+ @Test
+ void testEmptyWriterProducesNoMappingsOrCommit() throws Exception {
+ FileStoreTable table = createTable(true, true);
+ DataEvolutionClusterWriteOperator operator =
+ new DataEvolutionClusterWriteOperator(table,
BinaryRow.EMPTY_ROW, "cluster-test");
+ try (OneInputStreamOperatorTestHarness<InternalRow, Committable>
harness =
+ new OneInputStreamOperatorTestHarness<>(operator)) {
+ harness.setup(new CommittableTypeInfo().createSerializer(new
ExecutionConfig()));
+ harness.open();
+ operator.endInput();
+ assertThat(harness.extractOutputValues()).isEmpty();
+ }
+ assertThat(table.latestSnapshot()).isEmpty();
+ }
+
+ @Test
+ void testMappingFailureCleansPreparedDataAndEarlierMappings() throws
Exception {
+ FailingMappingFileIO fileIO = new FailingMappingFileIO();
+ FileStoreTable table = createTable(true, true, fileIO);
+ fileIO.recordOutputs = true;
+ DataEvolutionClusterWriteOperator operator =
+ new DataEvolutionClusterWriteOperator(table,
BinaryRow.EMPTY_ROW, "cluster-test");
+ try (OneInputStreamOperatorTestHarness<InternalRow, Committable>
harness =
+ new OneInputStreamOperatorTestHarness<>(operator)) {
+ harness.setup(new CommittableTypeInfo().createSerializer(new
ExecutionConfig()));
+ harness.open();
+ for (int id = 0; id < 40; id++) {
+ harness.processElement(
+ new StreamRecord<>(
+ GenericRow.of(id,
BinaryString.fromString("payload"), (long) id)));
+ }
+ assertThatThrownBy(operator::endInput)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("injected mapping failure");
+ assertThat(harness.extractOutputValues()).isEmpty();
+ assertThat(fileIO.mappingCount).isEqualTo(2);
+ assertThat(
+ fileIO.outputs.stream()
+ .filter(
+ path ->
+
!ClusteringRowIdMapping.isMappingFile(
+ path.getName())))
+ .isNotEmpty();
+ }
+ for (Path output : fileIO.outputs) {
+ assertThat(fileIO.exists(output)).as("aborted output %s",
output).isFalse();
+ }
+ assertThat(table.latestSnapshot()).isEmpty();
+ }
+
+ private FileStoreTable createTable(boolean buffered, boolean spillable)
throws Exception {
+ return createTable(buffered, spillable, LocalFileIO.create());
+ }
+
+ private FileStoreTable createTable(boolean buffered, boolean spillable,
LocalFileIO fileIO)
+ throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.STRING())
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.WRITE_ONLY.key(), "true")
+ .option(CoreOptions.WRITE_BUFFER_FOR_APPEND.key(),
String.valueOf(buffered))
+ .option(CoreOptions.WRITE_BUFFER_SPILLABLE.key(),
String.valueOf(spillable))
+ .option(CoreOptions.WRITE_BUFFER_SIZE.key(), "16 kb")
+ .option(CoreOptions.PAGE_SIZE.key(), "4 kb")
+ .option(CoreOptions.TARGET_FILE_ROW_NUM.key(), "17")
+ .build();
+ Path path = new Path(tempDir.toString());
+ SchemaManager schemaManager = new FileSystemSchemaManager(fileIO,
path);
+ return FileStoreTableFactory.create(fileIO, path,
schemaManager.createTable(schema));
+ }
+
+ private List<Integer> readIds(FileStoreTable table, DataFileMeta file)
throws Exception {
+ DataSplit split =
+ DataSplit.builder()
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withTotalBuckets(1)
+ .withBucketPath(
+ table.store()
+ .pathFactory()
+ .bucketPath(BinaryRow.EMPTY_ROW, 0)
+ .toString())
+
.withDataFiles(Collections.singletonList(file.assignFirstRowId(0L)))
+ .rawConvertible(false)
+ .build();
+ List<Integer> result = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
table.newRead().createReader(split);
+ CloseableIterator<InternalRow> rows =
reader.toCloseableIterator()) {
+ while (rows.hasNext()) {
+ InternalRow row = rows.next();
+ assertThat(row.getFieldCount()).isEqualTo(2);
+ result.add(row.getInt(0));
+ }
+ }
+ return result;
+ }
+
+ private static long oldRowId(int id, int rowCount) {
+ return 1000L + (rowCount - id - 1) * 3L;
+ }
+
+ private static class FailingMappingFileIO extends LocalFileIO {
+
+ private static final long serialVersionUID = 1L;
+ private final List<Path> outputs = new ArrayList<>();
+ private boolean recordOutputs;
+ private int mappingCount;
+
+ @Override
+ public PositionOutputStream newOutputStream(Path path, boolean
overwrite)
+ throws IOException {
+ if (recordOutputs) {
+ outputs.add(path);
+ if (ClusteringRowIdMapping.isMappingFile(path.getName()) &&
++mappingCount == 2) {
+ throw new IOException("injected mapping failure");
+ }
+ }
+ return super.newOutputStream(path, overwrite);
+ }
+ }
+}
