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 184d68bd91 [flink] Commit format table overwrites once (#9433)
184d68bd91 is described below
commit 184d68bd91b4bd40b97b2150b20b2240cb2430b3
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Fri Aug 28 13:48:06 2026 +0800
[flink] Commit format table overwrites once (#9433)
---
.../flink/sink/FlinkFormatTableDataStreamSink.java | 36 +++++-
.../flink/sink/FlinkFormatTableSinkBase.java | 6 +
.../sink/FlinkFormatTableDataStreamSinkTest.java | 143 +++++++++++++++++++--
.../paimon/flink/source/FormatTableITCase.java | 46 +++++++
4 files changed, 214 insertions(+), 17 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
index a223ad5129..029ffe5fac 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
@@ -55,10 +55,16 @@ public class FlinkFormatTableDataStreamSink {
}
public DataStreamSink<?> sinkFrom(DataStream<RowData> dataStream) {
- return dataStream.sinkTo(new FormatTableSink(table, overwrite,
staticPartitions));
+ DataStreamSink<?> sink =
+ dataStream.sinkTo(new FormatTableSink(table, overwrite,
staticPartitions));
+ if (overwrite) {
+ // Parallel overwrite commits could delete files produced by one
another.
+ sink.setParallelism(1);
+ }
+ return sink;
}
- private static class FormatTableSink implements Sink<RowData>,
LineageVertexProvider {
+ static class FormatTableSink implements Sink<RowData>,
LineageVertexProvider {
private final FormatTable table;
private final boolean overwrite;
@@ -93,14 +99,17 @@ public class FlinkFormatTableDataStreamSink {
}
/** Sink writer for format tables using Flink v2 API. */
- private static class FormatTableSinkWriter implements
SinkWriter<RowData> {
+ static class FormatTableSinkWriter implements SinkWriter<RowData> {
+ private final boolean overwrite;
+ private boolean reachedEndOfInput;
private transient BatchWriteBuilder writeBuilder;
private transient FormatTableWrite tableWrite;
private transient BatchTableCommit tableCommit;
public FormatTableSinkWriter(
FormatTable table, boolean overwrite, Map<String, String>
staticPartitions) {
+ this.overwrite = overwrite;
this.writeBuilder = table.newBatchWriteBuilder();
this.tableWrite = (FormatTableWrite) writeBuilder.newWrite();
if (overwrite) {
@@ -121,21 +130,34 @@ public class FlinkFormatTableDataStreamSink {
}
@Override
- public void flush(boolean endOfInput) {}
+ public void flush(boolean endOfInput) {
+ if (endOfInput) {
+ reachedEndOfInput = true;
+ }
+ }
@Override
public void close() throws Exception {
if (tableWrite != null) {
List<CommitMessage> commitMessages = null;
+ boolean shouldCommit = false;
try {
// Prepare commit and commit the data
commitMessages = tableWrite.prepareCommit();
- if (!commitMessages.isEmpty()) {
+ // A normally completed overwrite replaces its target
even with no rows.
+ shouldCommit =
+ !commitMessages.isEmpty() || (overwrite &&
reachedEndOfInput);
+ if (shouldCommit) {
tableCommit.commit(commitMessages);
}
} catch (Exception e) {
- if (commitMessages != null &&
!commitMessages.isEmpty()) {
- tableCommit.abort(commitMessages);
+ if (commitMessages != null && shouldCommit) {
+ try {
+ tableCommit.abort(commitMessages);
+ } catch (Exception abortFailure) {
+ // Report the commit failure, not the cleanup
that followed it.
+ e.addSuppressed(abortFailure);
+ }
}
throw new RuntimeException(e);
} finally {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java
index 7add7f64c1..749cfbbdf8 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableSinkBase.java
@@ -57,6 +57,12 @@ public abstract class FlinkFormatTableSinkBase
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ if (overwrite && !context.isBounded()) {
+ // An overwrite replaces its target once the input ends, which an
unbounded input
+ // never does.
+ throw new UnsupportedOperationException(
+ "Paimon doesn't support streaming INSERT OVERWRITE.");
+ }
return new PaimonDataStreamSinkProvider(
(dataStream) ->
new FlinkFormatTableDataStreamSink(table, overwrite,
staticPartitions)
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
index 9a838bea68..15fb041159 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
@@ -24,25 +24,136 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.format.FormatTableWrite;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.lineage.LineageVertex;
import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-import java.lang.reflect.Constructor;
import java.util.Collections;
-import java.util.Map;
+import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/** Tests for {@link FlinkFormatTableDataStreamSink}. */
class FlinkFormatTableDataStreamSinkTest {
@TempDir java.nio.file.Path temp;
+ @Test
+ void testOverwriteUsesOneSinkWriterWhileAppendKeepsParallelism() {
+ int parallelism = 4;
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(parallelism);
+ RowType rowType = RowType.of(new IntType());
+ DataStream<RowData> input =
+ env.fromCollection(
+ Collections.singletonList((RowData)
GenericRowData.of(1)),
+ InternalTypeInfo.of(toLogicalType(rowType)));
+ FormatTable table = mock(FormatTable.class);
+ when(table.options()).thenReturn(Collections.singletonMap("path",
temp.toUri().toString()));
+ when(table.partitionKeys()).thenReturn(Collections.emptyList());
+ when(table.primaryKeys()).thenReturn(Collections.emptyList());
+ when(table.fullName()).thenReturn("test_db.test_table");
+ when(table.rowType()).thenReturn(rowType);
+
+ DataStreamSink<?> overwriteSink =
+ new FlinkFormatTableDataStreamSink(table, true,
Collections.emptyMap())
+ .sinkFrom(input);
+ DataStreamSink<?> appendSink =
+ new FlinkFormatTableDataStreamSink(table, false,
Collections.emptyMap())
+ .sinkFrom(input);
+
+ assertThat(overwriteSink.getTransformation().getParallelism()).isOne();
+ // Without this the adaptive batch scheduler is free to pick the
parallelism back up.
+
assertThat(overwriteSink.getTransformation().isParallelismConfigured()).isTrue();
+
assertThat(appendSink.getTransformation().getParallelism()).isEqualTo(parallelism);
+ }
+
+ @Test
+ void testEmptyMessagesAreCommittedOnlyForOverwrite() throws Exception {
+ FormatTableWrite overwriteWrite = mock(FormatTableWrite.class);
+ BatchTableCommit overwriteCommit = mock(BatchTableCommit.class);
+
when(overwriteWrite.prepareCommit()).thenReturn(Collections.emptyList());
+
+ SinkWriter<?> overwriteWriter = createWriter(true, overwriteWrite,
overwriteCommit);
+ overwriteWriter.flush(true);
+ overwriteWriter.close();
+
+ verify(overwriteCommit).commit(Collections.emptyList());
+
+ FormatTableWrite appendWrite = mock(FormatTableWrite.class);
+ BatchTableCommit appendCommit = mock(BatchTableCommit.class);
+ when(appendWrite.prepareCommit()).thenReturn(Collections.emptyList());
+
+ SinkWriter<?> appendWriter = createWriter(false, appendWrite,
appendCommit);
+ appendWriter.flush(true);
+ appendWriter.close();
+
+ verify(appendCommit, never()).commit(anyList());
+ }
+
+ @Test
+ void testEmptyOverwriteIsNotCommittedBeforeEndOfInput() throws Exception {
+ FormatTableWrite tableWrite = mock(FormatTableWrite.class);
+ BatchTableCommit tableCommit = mock(BatchTableCommit.class);
+ when(tableWrite.prepareCommit()).thenReturn(Collections.emptyList());
+
+ SinkWriter<?> writer = createWriter(true, tableWrite, tableCommit);
+ // A checkpoint is not the end of the input, and a job that fails
after one must not have
+ // replaced the target.
+ writer.flush(false);
+ writer.close();
+
+ verify(tableCommit, never()).commit(anyList());
+ verify(tableCommit, never()).abort(anyList());
+ }
+
+ @Test
+ void testFailedEmptyOverwriteCommitIsAborted() throws Exception {
+ FormatTableWrite tableWrite = mock(FormatTableWrite.class);
+ BatchTableCommit tableCommit = mock(BatchTableCommit.class);
+ when(tableWrite.prepareCommit()).thenReturn(Collections.emptyList());
+ doThrow(new RuntimeException("commit failed"))
+ .when(tableCommit)
+ .commit(Collections.emptyList());
+ doThrow(new RuntimeException("abort failed"))
+ .when(tableCommit)
+ .abort(Collections.emptyList());
+ SinkWriter<?> writer = createWriter(true, tableWrite, tableCommit);
+ writer.flush(true);
+
+ assertThatThrownBy(writer::close)
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseMessage("commit failed")
+ .rootCause()
+ .satisfies(
+ cause ->
+ assertThat(cause.getSuppressed())
+ .extracting(Throwable::getMessage)
+ .containsExactly("abort failed"));
+ verify(tableCommit).abort(Collections.emptyList());
+ }
+
@Test
void testFormatTableSinkLineageVertex() throws Exception {
FormatTable table =
@@ -57,17 +168,29 @@ class FlinkFormatTableDataStreamSinkTest {
.catalogContext(CatalogContext.create(new Options()))
.build();
- Class<?> sinkClass =
- Class.forName(
-
"org.apache.paimon.flink.sink.FlinkFormatTableDataStreamSink$FormatTableSink");
- Constructor<?> constructor =
- sinkClass.getDeclaredConstructor(FormatTable.class,
boolean.class, Map.class);
- constructor.setAccessible(true);
- Object sink = constructor.newInstance(table, false,
Collections.emptyMap());
+ FlinkFormatTableDataStreamSink.FormatTableSink sink =
+ new FlinkFormatTableDataStreamSink.FormatTableSink(
+ table, false, Collections.emptyMap());
assertThat(sink).isInstanceOf(LineageVertexProvider.class);
- LineageVertex vertex = ((LineageVertexProvider)
sink).getLineageVertex();
+ LineageVertex vertex = sink.getLineageVertex();
assertThat(vertex.datasets()).hasSize(1);
assertThat(vertex.datasets().get(0).name()).isEqualTo("paimon." +
table.fullName());
}
+
+ private SinkWriter<?> createWriter(
+ boolean overwrite, FormatTableWrite tableWrite, BatchTableCommit
tableCommit)
+ throws Exception {
+ FormatTable table = mock(FormatTable.class);
+ BatchWriteBuilder writeBuilder = mock(BatchWriteBuilder.class);
+ when(table.newBatchWriteBuilder()).thenReturn(writeBuilder);
+ when(writeBuilder.newWrite()).thenReturn(tableWrite);
+ when(writeBuilder.newCommit()).thenReturn(tableCommit);
+ if (overwrite) {
+
when(writeBuilder.withOverwrite(Collections.emptyMap())).thenReturn(writeBuilder);
+ }
+
+ return new
FlinkFormatTableDataStreamSink.FormatTableSink.FormatTableSinkWriter(
+ table, overwrite, Collections.emptyMap());
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
index 63da32b3fe..ba8dc66aea 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
@@ -133,6 +133,52 @@ public class FormatTableITCase extends
RESTCatalogITCaseBase {
sql("Drop TABLE %s", tableName);
}
+ @Test
+ public void testInsertOverwriteWithoutRowsReplacesItsTarget() {
+ String tableName = "format_table_empty_overwrite";
+ sql(
+ "CREATE TABLE %s (a INT, b INT) WITH ('file.format'='parquet',
'type'='format-table')",
+ tableName);
+ setDataToken(tableName);
+
+ sql("INSERT INTO %s VALUES (1, 11), (2, 22)", tableName);
+ assertThat(sql("SELECT * FROM %s", tableName))
+ .containsExactlyInAnyOrder(Row.of(1, 11), Row.of(2, 22));
+
+ sql("INSERT OVERWRITE %s SELECT a, b FROM %s WHERE a < 0", tableName,
tableName);
+ assertThat(sql("SELECT * FROM %s", tableName)).isEmpty();
+
+ sql("Drop TABLE %s", tableName);
+ }
+
+ @Test
+ public void testInsertOverwriteWithoutRowsEmptiesOnlyTheNamedPartition() {
+ String tableName = "format_table_empty_overwrite_partitioned";
+ sql(
+ "CREATE TABLE %s (a INT, b INT, c INT) PARTITIONED BY (c) WITH
('file.format'='parquet', 'type'='format-table')",
+ tableName);
+ setDataToken(tableName);
+
+ sql("INSERT INTO %s PARTITION (c = 1) VALUES (1, 11)", tableName);
+ sql("INSERT INTO %s PARTITION (c = 2) VALUES (2, 22)", tableName);
+ assertThat(sql("SELECT a, b, c FROM %s", tableName))
+ .containsExactlyInAnyOrder(Row.of(1, 11, 1), Row.of(2, 22, 2));
+
+ sql(
+ "INSERT OVERWRITE %s PARTITION (c = 1) SELECT a, b FROM %s
WHERE a < 0",
+ tableName, tableName);
+ assertThat(sql("SELECT a, b, c FROM %s", tableName))
+ .containsExactlyInAnyOrder(Row.of(2, 22, 2));
+
+ // Without a PARTITION clause the statement replaces the partitions it
wrote, and it wrote
+ // none, so both engines leave the table alone.
+ sql("INSERT OVERWRITE %s SELECT a, b, c FROM %s WHERE a < 0",
tableName, tableName);
+ assertThat(sql("SELECT a, b, c FROM %s", tableName))
+ .containsExactlyInAnyOrder(Row.of(2, 22, 2));
+
+ sql("Drop TABLE %s", tableName);
+ }
+
@Test
public void testTruncateTable() {
String tableName = "format_table_truncate";
