You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by cz...@apache.org on 2022/11/08 03:15:36 UTC
[flink-table-store] branch master updated: [FLINK-29840] Fix bug that old record may overwrite new record in Table Store when snapshot committing is slow
This is an automated email from the ASF dual-hosted git repository.
czweng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new f54008c9 [FLINK-29840] Fix bug that old record may overwrite new record in Table Store when snapshot committing is slow
f54008c9 is described below
commit f54008c98f51e419fa4a80d6fe0e62ab97358c54
Author: tsreaper <ts...@gmail.com>
AuthorDate: Tue Nov 8 11:15:30 2022 +0800
[FLINK-29840] Fix bug that old record may overwrite new record in Table Store when snapshot committing is slow
This closes #354.
---
.../store/connector/sink/CommitterOperator.java | 39 ++---
.../table/store/connector/sink/StateUtils.java | 76 +++++++++
.../store/connector/sink/StoreCompactOperator.java | 10 +-
.../table/store/connector/sink/StoreSink.java | 19 ++-
.../store/connector/sink/StoreWriteOperator.java | 45 +++--
.../ChangelogWithKeyFileStoreTableITCase.java | 189 +++++++++++++++++++++
.../connector/sink/CommitterOperatorTest.java | 12 +-
.../source/TestChangelogDataReadWrite.java | 4 +
.../flink/table/store/file/AbstractFileStore.java | 4 +-
.../table/store/file/AppendOnlyFileStore.java | 3 +-
.../apache/flink/table/store/file/FileStore.java | 4 +-
.../flink/table/store/file/KeyValueFileStore.java | 3 +-
.../file/operation/AbstractFileStoreWrite.java | 96 ++++++++---
.../file/operation/AppendOnlyFileStoreWrite.java | 3 +-
.../table/store/file/operation/FileStoreWrite.java | 3 +-
.../file/operation/KeyValueFileStoreWrite.java | 3 +-
.../store/file/operation/MemoryFileStoreWrite.java | 14 +-
.../table/store/table/AbstractFileStoreTable.java | 4 +-
.../store/table/AppendOnlyFileStoreTable.java | 4 +-
.../table/ChangelogValueCountFileStoreTable.java | 4 +-
.../table/ChangelogWithKeyFileStoreTable.java | 4 +-
.../flink/table/store/table/FileStoreTable.java | 6 -
.../flink/table/store/table/SupportsWrite.java | 4 +-
.../flink/table/store/table/sink/TableWrite.java | 2 +-
.../table/store/table/sink/TableWriteImpl.java | 5 +-
.../flink/table/store/file/TestFileStore.java | 10 +-
.../store/file/operation/TestCommitThread.java | 5 +-
.../file/utils/FailingAtomicRenameFileSystem.java | 14 ++
.../store/table/AppendOnlyFileStoreTableTest.java | 16 +-
.../ChangelogValueCountFileStoreTableTest.java | 16 +-
.../table/ChangelogWithKeyFileStoreTableTest.java | 93 ++++++----
.../table/store/table/FileStoreTableTestBase.java | 49 +++---
.../table/store/table/SchemaEvolutionTest.java | 13 +-
.../table/store/table/WritePreemptMemoryTest.java | 8 +-
.../hive/TableStoreHiveStorageHandlerITCase.java | 40 +++--
.../store/mapred/TableStoreRecordReaderTest.java | 26 ++-
.../table/store/spark/SimpleTableTestHelper.java | 9 +-
.../table/store/spark/SimpleTableTestHelper.java | 9 +-
38 files changed, 645 insertions(+), 223 deletions(-)
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/CommitterOperator.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/CommitterOperator.java
index b5df8793..ac55befd 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/CommitterOperator.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/CommitterOperator.java
@@ -30,7 +30,6 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.store.file.manifest.ManifestCommittable;
-import org.apache.flink.util.Preconditions;
import org.apache.flink.util.function.SerializableFunction;
import org.apache.flink.util.function.SerializableSupplier;
@@ -42,7 +41,6 @@ import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
-import java.util.UUID;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -63,6 +61,13 @@ public class CommitterOperator extends AbstractStreamOperator<Committable>
*/
private final boolean streamingCheckpointEnabled;
+ /**
+ * This commitUser is valid only for new jobs. After the job starts, this commitUser will be
+ * recorded into the states of write and commit operators. When the job restarts, commitUser
+ * will be recovered from states and this value is ignored.
+ */
+ private final String initialCommitUser;
+
/** Group the committable by the checkpoint id. */
private final NavigableMap<Long, ManifestCommittable> committablesPerCheckpoint;
@@ -85,10 +90,12 @@ public class CommitterOperator extends AbstractStreamOperator<Committable>
public CommitterOperator(
boolean streamingCheckpointEnabled,
+ String initialCommitUser,
SerializableFunction<String, Committer> committerFactory,
SerializableSupplier<SimpleVersionedSerializer<ManifestCommittable>>
committableSerializer) {
this.streamingCheckpointEnabled = streamingCheckpointEnabled;
+ this.initialCommitUser = initialCommitUser;
this.committableSerializer = committableSerializer;
this.committablesPerCheckpoint = new TreeMap<>();
this.committerFactory = checkNotNull(committerFactory);
@@ -99,32 +106,14 @@ public class CommitterOperator extends AbstractStreamOperator<Committable>
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
- // commit user name state of this job
// each job can only have one user name and this name must be consistent across restarts
- ListState<String> commitUserState =
- context.getOperatorStateStore()
- .getListState(new ListStateDescriptor<>("commit_user_state", String.class));
- List<String> commitUsers = new ArrayList<>();
- commitUserState.get().forEach(commitUsers::add);
- if (context.isRestored()) {
- Preconditions.checkState(
- commitUsers.size() == 1,
- "Expecting 1 commit user name when recovering from checkpoint but found "
- + commitUsers.size()
- + ". This is unexpected.");
- } else {
- Preconditions.checkState(
- commitUsers.isEmpty(),
- "Expecting 0 commit user name for a fresh sink state but found "
- + commitUsers.size()
- + ". This is unexpected.");
- String commitUser = UUID.randomUUID().toString();
- commitUserState.add(commitUser);
- commitUsers.add(commitUser);
- }
// we cannot use job id as commit user name here because user may change job id by creating
// a savepoint, stop the job and then resume from savepoint
- committer = committerFactory.apply(commitUsers.get(0));
+ String commitUser =
+ StateUtils.getSingleValueFromState(
+ context, "commit_user_state", String.class, initialCommitUser);
+ // parallelism of commit operator is always 1, so commitUser will never be null
+ committer = committerFactory.apply(commitUser);
streamingCommitterState =
new SimpleVersionedListState<>(
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StateUtils.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StateUtils.java
new file mode 100644
index 00000000..f08f53d6
--- /dev/null
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StateUtils.java
@@ -0,0 +1,76 @@
+/*
+ * 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.flink.table.store.connector.sink;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Utility class for sink state manipulation. */
+public class StateUtils {
+
+ public static @Nullable <T> T getSingleValueFromState(
+ StateInitializationContext context,
+ String stateName,
+ Class<T> valueClass,
+ T defaultValue)
+ throws Exception {
+ ListState<T> state =
+ context.getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>(stateName, valueClass));
+
+ List<T> values = new ArrayList<>();
+ state.get().forEach(values::add);
+
+ if (context.isRestored()) {
+ // Values may contain 0 element or more than 1 element.
+ //
+ // Let's say a vertex has 3 tasks (A, B and C). If A is finished while B and C are still
+ // running, then states of A will be divided between B and C. That is, if the job
+ // restarts, state of vertex A will be empty, and state of vertex B and C may contain
+ // more than 1 element.
+ if (values.isEmpty()) {
+ return null;
+ }
+
+ // As we're storing the same value for each task, we hereby check if all elements are
+ // equal.
+ for (int i = 1; i < values.size(); i++) {
+ Preconditions.checkState(
+ values.get(i).equals(values.get(i - 1)),
+ "Values in list state are not the same. This is unexpected.");
+ }
+ } else {
+ Preconditions.checkState(
+ values.isEmpty(),
+ "Expecting 0 value for a fresh state but found "
+ + values.size()
+ + ". This is unexpected.");
+ state.add(defaultValue);
+ values.add(defaultValue);
+ }
+
+ return values.get(0);
+ }
+}
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreCompactOperator.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreCompactOperator.java
index e26b0a98..5289ba6f 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreCompactOperator.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreCompactOperator.java
@@ -45,14 +45,18 @@ public class StoreCompactOperator extends PrepareCommitOperator {
private static final Logger LOG = LoggerFactory.getLogger(StoreCompactOperator.class);
private final FileStoreTable table;
+ private final String commitUser;
@Nullable private final Map<String, String> compactPartitionSpec;
private TableScan scan;
private TableWrite write;
public StoreCompactOperator(
- FileStoreTable table, @Nullable Map<String, String> compactPartitionSpec) {
+ FileStoreTable table,
+ String commitUser,
+ @Nullable Map<String, String> compactPartitionSpec) {
this.table = table;
+ this.commitUser = commitUser;
this.compactPartitionSpec = compactPartitionSpec;
}
@@ -67,7 +71,7 @@ public class StoreCompactOperator extends PrepareCommitOperator {
compactPartitionSpec, table.schema().logicalPartitionType()));
}
- write = table.newWrite();
+ write = table.newWrite(commitUser);
}
@Override
@@ -102,7 +106,7 @@ public class StoreCompactOperator extends PrepareCommitOperator {
}
try {
- return write.prepareCommit(true).stream()
+ return write.prepareCommit(true, checkpointId).stream()
.map(c -> new Committable(checkpointId, Committable.Kind.FILE, c))
.collect(Collectors.toList());
} catch (Exception e) {
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreSink.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreSink.java
index c81faeb7..5e9aadeb 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreSink.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreSink.java
@@ -42,6 +42,7 @@ import javax.annotation.Nullable;
import java.io.Serializable;
import java.util.Map;
+import java.util.UUID;
/** Sink of dynamic store. */
public class StoreSink implements Serializable {
@@ -83,11 +84,13 @@ public class StoreSink implements Serializable {
this.logSinkFunction = logSinkFunction;
}
- private OneInputStreamOperator<RowData, Committable> createWriteOperator() {
+ private OneInputStreamOperator<RowData, Committable> createWriteOperator(
+ String initialCommitUser) {
if (compactionTask) {
- return new StoreCompactOperator(table, compactPartitionSpec);
+ return new StoreCompactOperator(table, initialCommitUser, compactPartitionSpec);
}
- return new StoreWriteOperator(table, overwritePartition, logSinkFunction);
+ return new StoreWriteOperator(
+ table, initialCommitUser, overwritePartition, logSinkFunction);
}
private StoreCommitter createCommitter(String user, boolean createEmptyCommit) {
@@ -100,9 +103,16 @@ public class StoreSink implements Serializable {
}
public DataStreamSink<?> sinkTo(DataStream<RowData> input) {
+ // This commitUser is valid only for new jobs.
+ // After the job starts, this commitUser will be recorded into the states of write and
+ // commit operators.
+ // When the job restarts, commitUser will be recovered from states and this value is
+ // ignored.
+ String initialCommitUser = UUID.randomUUID().toString();
+
CommittableTypeInfo typeInfo = new CommittableTypeInfo();
SingleOutputStreamOperator<Committable> written =
- input.transform(WRITER_NAME, typeInfo, createWriteOperator())
+ input.transform(WRITER_NAME, typeInfo, createWriteOperator(initialCommitUser))
.setParallelism(input.getParallelism());
StreamExecutionEnvironment env = input.getExecutionEnvironment();
@@ -121,6 +131,7 @@ public class StoreSink implements Serializable {
typeInfo,
new CommitterOperator(
streamingCheckpointEnabled,
+ initialCommitUser,
// If checkpoint is enabled for streaming job, we have to
// commit new files list even if they're empty.
// Otherwise we can't tell if the commit is successful after
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreWriteOperator.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreWriteOperator.java
index 44027e12..423a4289 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreWriteOperator.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreWriteOperator.java
@@ -52,6 +52,13 @@ public class StoreWriteOperator extends PrepareCommitOperator {
private final FileStoreTable table;
+ /**
+ * This commitUser is valid only for new jobs. After the job starts, this commitUser will be
+ * recorded into the states of write and commit operators. When the job restarts, commitUser
+ * will be recovered from states and this value is ignored.
+ */
+ private final String initialCommitUser;
+
@Nullable private final Map<String, String> overwritePartition;
@Nullable private final LogSinkFunction logSinkFunction;
@@ -61,15 +68,17 @@ public class StoreWriteOperator extends PrepareCommitOperator {
/** We listen to this ourselves because we don't have an {@link InternalTimerService}. */
private long currentWatermark = Long.MIN_VALUE;
- private TableWrite write;
+ @Nullable private TableWrite write;
@Nullable private LogWriteCallback logCallback;
public StoreWriteOperator(
FileStoreTable table,
+ String initialCommitUser,
@Nullable Map<String, String> overwritePartition,
@Nullable LogSinkFunction logSinkFunction) {
this.table = table;
+ this.initialCommitUser = initialCommitUser;
this.overwritePartition = overwritePartition;
this.logSinkFunction = logSinkFunction;
}
@@ -88,18 +97,31 @@ public class StoreWriteOperator extends PrepareCommitOperator {
@Override
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
+
if (logSinkFunction != null) {
StreamingFunctionUtils.restoreFunctionState(context, logSinkFunction);
}
+
+ // each job can only have one user name and this name must be consistent across restarts
+ // we cannot use job id as commit user name here because user may change job id by creating
+ // a savepoint, stop the job and then resume from savepoint
+ String commitUser =
+ StateUtils.getSingleValueFromState(
+ context, "commit_user_state", String.class, initialCommitUser);
+ // see comments of StateUtils.getSingleValueFromState for why commitUser may be null
+ if (commitUser == null) {
+ this.write = null;
+ } else {
+ this.write =
+ table.newWrite(commitUser)
+ .withIOManager(getContainingTask().getEnvironment().getIOManager())
+ .withOverwrite(overwritePartition != null);
+ }
}
@Override
public void open() throws Exception {
super.open();
- this.write =
- table.newWrite()
- .withIOManager(getContainingTask().getEnvironment().getIOManager())
- .withOverwrite(overwritePartition != null);
this.sinkContext = new SimpleContext(getProcessingTimeService());
if (logSinkFunction != null) {
FunctionUtils.openFunction(logSinkFunction, new Configuration());
@@ -187,12 +209,15 @@ public class StoreWriteOperator extends PrepareCommitOperator {
protected List<Committable> prepareCommit(boolean endOfInput, long checkpointId)
throws IOException {
List<Committable> committables = new ArrayList<>();
- try {
- for (FileCommittable committable : write.prepareCommit(endOfInput)) {
- committables.add(new Committable(checkpointId, Committable.Kind.FILE, committable));
+ if (write != null) {
+ try {
+ for (FileCommittable committable : write.prepareCommit(endOfInput, checkpointId)) {
+ committables.add(
+ new Committable(checkpointId, Committable.Kind.FILE, committable));
+ }
+ } catch (Exception e) {
+ throw new IOException(e);
}
- } catch (Exception e) {
- throw new IOException(e);
}
if (logCallback != null) {
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ChangelogWithKeyFileStoreTableITCase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ChangelogWithKeyFileStoreTableITCase.java
new file mode 100644
index 00000000..50205741
--- /dev/null
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ChangelogWithKeyFileStoreTableITCase.java
@@ -0,0 +1,189 @@
+/*
+ * 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.flink.table.store.connector;
+
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.store.file.utils.FailingAtomicRenameFileSystem;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL;
+
+/** Tests for changelog table with primary keys. */
+public class ChangelogWithKeyFileStoreTableITCase extends AbstractTestBase {
+
+ private String path;
+ private String failingName;
+
+ @Before
+ public void before() throws IOException {
+ path = TEMPORARY_FOLDER.newFolder().toPath().toString();
+ // for failure tests
+ failingName = UUID.randomUUID().toString();
+ }
+
+ @Test
+ public void testBatchRandom() throws Exception {
+ TableEnvironment bEnv =
+ TableEnvironmentTestUtils.create(
+ EnvironmentSettings.newInstance().inBatchMode().build());
+ testRandom(bEnv, false);
+ }
+
+ @Test
+ public void testStreamingRandom() throws Exception {
+ TableEnvironment sEnv =
+ TableEnvironmentTestUtils.create(
+ EnvironmentSettings.newInstance().inStreamingMode().build());
+ // set checkpoint interval to a random number to emulate different speed of commit
+ sEnv.getConfig()
+ .getConfiguration()
+ .set(
+ CHECKPOINTING_INTERVAL,
+ Duration.ofMillis(ThreadLocalRandom.current().nextInt(900) + 100));
+ testRandom(sEnv, ThreadLocalRandom.current().nextBoolean());
+ }
+
+ /**
+ * Randomly update {@code numParts} partitions and {@code numKeys} keys for about {@code limit}
+ * records. For the final {@code numParts * numKeys} records, keys are updated to some specific
+ * values for result checking.
+ */
+ private void testRandom(TableEnvironment tEnv, boolean enableFailure) throws Exception {
+ int numParts = 4;
+ int numKeys = 64;
+ int numValues = 1024;
+ int limit = 10000;
+
+ String failingPath = FailingAtomicRenameFileSystem.getFailingPath(failingName, path);
+
+ // no failure when creating catalog and table
+ FailingAtomicRenameFileSystem.reset(failingName, 0, 1);
+ tEnv.executeSql(
+ String.format(
+ "CREATE CATALOG testCatalog WITH ('type'='table-store', 'warehouse'='%s')",
+ failingPath));
+ tEnv.executeSql("USE CATALOG testCatalog");
+ tEnv.executeSql(
+ "CREATE TABLE T("
+ + " pt STRING,"
+ + " k INT,"
+ + " v1 BIGINT,"
+ + " v2 STRING,"
+ + " PRIMARY KEY (pt, k) NOT ENFORCED"
+ + ") PARTITIONED BY (pt) WITH ("
+ + " 'bucket' = '4'"
+ + ")");
+
+ // input data must be strictly ordered
+ tEnv.getConfig()
+ .getConfiguration()
+ .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+ tEnv.executeSql(
+ "CREATE TABLE `default_catalog`.`default_database`.`S` ("
+ + " i INT"
+ + ") WITH ("
+ + " 'connector' = 'datagen',"
+ + " 'fields.i.kind' = 'sequence',"
+ + " 'fields.i.start' = '0',"
+ + " 'fields.i.end' = '"
+ + (limit + numParts * numKeys - 1)
+ + "',"
+ + " 'number-of-rows' = '"
+ + (limit + numParts * numKeys)
+ + "',"
+ + " 'rows-per-second' = '"
+ + (limit / 5 + ThreadLocalRandom.current().nextInt(limit / 10))
+ + "'"
+ + ")")
+ .await();
+
+ // for the last `numParts * numKeys` records, we update every key to a specific value
+ String ptSql =
+ String.format(
+ "IF(i >= %d, CAST((i - %d) / %d AS STRING), CAST(CAST(FLOOR(RAND() * %d) AS INT) AS STRING)) AS pt",
+ limit, limit, numKeys, numParts);
+ String kSql =
+ String.format(
+ "IF(i >= %d, MOD(i - %d, %d), CAST(FLOOR(RAND() * %d) AS INT)) AS k",
+ limit, limit, numKeys, numKeys);
+ String v1Sql =
+ String.format(
+ "IF(i >= %d, i, CAST(FLOOR(RAND() * %d) AS BIGINT)) AS v1",
+ limit, numValues);
+ String v2Sql = "CAST(i AS STRING) || '.str' AS v2";
+ tEnv.executeSql(
+ String.format(
+ "CREATE TEMPORARY VIEW myView AS SELECT %s, %s, %s, %s FROM `default_catalog`.`default_database`.`S`",
+ ptSql, kSql, v1Sql, v2Sql));
+
+ // run test SQL
+ if (enableFailure) {
+ FailingAtomicRenameFileSystem.reset(failingName, 100, 1000);
+ }
+ FailingAtomicRenameFileSystem.retryArtificialException(
+ () ->
+ tEnv.executeSql(
+ "INSERT INTO T /*+ OPTIONS('sink.parallelism' = '2') */ SELECT * FROM myView")
+ .await());
+
+ // check for result
+ TableEnvironment bEnv =
+ TableEnvironmentTestUtils.create(
+ EnvironmentSettings.newInstance().inBatchMode().build());
+ bEnv.executeSql(
+ String.format(
+ "CREATE CATALOG testCatalog WITH ('type'='table-store', 'warehouse'='%s')",
+ path));
+ bEnv.executeSql("USE CATALOG testCatalog");
+ Map<String, String> actual = new HashMap<>();
+ try (CloseableIterator<Row> it = bEnv.executeSql("SELECT * FROM T").collect()) {
+ while (it.hasNext()) {
+ Row row = it.next();
+ String key = row.getField(0) + "|" + row.getField(1);
+ String value = row.getField(2) + "|" + row.getField(3);
+ actual.put(key, value);
+ }
+ }
+
+ Assert.assertEquals(numParts * numKeys, actual.size());
+ for (int i = 0; i < numParts; i++) {
+ for (int j = 0; j < numKeys; j++) {
+ String key = i + "|" + j;
+ int x = limit + i * numKeys + j;
+ String expectedValue = x + "|" + x + ".str";
+ Assert.assertEquals(expectedValue, actual.get(key));
+ }
+ }
+ }
+}
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTest.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTest.java
index e8a2627b..f176132d 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTest.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/CommitterOperatorTest.java
@@ -51,6 +51,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.fail;
@@ -67,10 +68,12 @@ public class CommitterOperatorTest {
@TempDir public java.nio.file.Path tempDir;
private Path tablePath;
+ private String initialCommitUser;
@BeforeEach
public void before() {
tablePath = new Path(tempDir.toString());
+ initialCommitUser = UUID.randomUUID().toString();
}
@Test
@@ -81,12 +84,12 @@ public class CommitterOperatorTest {
createTestHarness(table);
testHarness.open();
- TableWrite write = table.newWrite();
+ TableWrite write = table.newWrite(initialCommitUser);
write.write(GenericRowData.of(1, 10L));
write.write(GenericRowData.of(2, 20L));
long timestamp = 1;
- for (FileCommittable committable : write.prepareCommit(false)) {
+ for (FileCommittable committable : write.prepareCommit(false, 8)) {
testHarness.processElement(
new Committable(8, Committable.Kind.FILE, committable), timestamp++);
}
@@ -129,10 +132,10 @@ public class CommitterOperatorTest {
long cpId = 0;
for (int i = 0; i < 10; i++) {
cpId++;
- TableWrite write = table.newWrite();
+ TableWrite write = table.newWrite(initialCommitUser);
write.write(GenericRowData.of(1, 10L));
write.write(GenericRowData.of(2, 20L));
- for (FileCommittable committable : write.prepareCommit(false)) {
+ for (FileCommittable committable : write.prepareCommit(false, cpId)) {
testHarness.processElement(
new Committable(cpId, Committable.Kind.FILE, committable), 1);
}
@@ -192,6 +195,7 @@ public class CommitterOperatorTest {
CommitterOperator operator =
new CommitterOperator(
true,
+ initialCommitUser,
user -> new StoreCommitter(table.newCommit(user)),
ManifestCommittableSerializer::new);
TypeSerializer<Committable> serializer =
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/TestChangelogDataReadWrite.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/TestChangelogDataReadWrite.java
index ad98eb8c..39ec2c0c 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/TestChangelogDataReadWrite.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/TestChangelogDataReadWrite.java
@@ -53,6 +53,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.function.Function;
@@ -73,6 +74,7 @@ public class TestChangelogDataReadWrite {
private final FileStorePathFactory pathFactory;
private final SnapshotManager snapshotManager;
private final ExecutorService service;
+ private final String commitUser;
public TestChangelogDataReadWrite(String root, ExecutorService service) {
this.avro = FileFormat.fromIdentifier("avro", new Configuration());
@@ -85,6 +87,7 @@ public class TestChangelogDataReadWrite {
CoreOptions.FILE_FORMAT.defaultValue());
this.snapshotManager = new SnapshotManager(new Path(root));
this.service = service;
+ this.commitUser = UUID.randomUUID().toString();
}
public TableRead createReadWithKey() {
@@ -152,6 +155,7 @@ public class TestChangelogDataReadWrite {
new KeyValueFileStoreWrite(
new SchemaManager(tablePath),
0,
+ commitUser,
KEY_TYPE,
VALUE_TYPE,
() -> COMPARATOR,
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AbstractFileStore.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AbstractFileStore.java
index 07300d3d..91e37620 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AbstractFileStore.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AbstractFileStore.java
@@ -95,10 +95,10 @@ public abstract class AbstractFileStore<T> implements FileStore<T> {
}
@Override
- public FileStoreCommitImpl newCommit(String user) {
+ public FileStoreCommitImpl newCommit(String commitUser) {
return new FileStoreCommitImpl(
schemaId,
- user,
+ commitUser,
partitionType,
pathFactory(),
snapshotManager(),
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AppendOnlyFileStore.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AppendOnlyFileStore.java
index d8aa4703..18cc6b36 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AppendOnlyFileStore.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AppendOnlyFileStore.java
@@ -58,10 +58,11 @@ public class AppendOnlyFileStore extends AbstractFileStore<RowData> {
}
@Override
- public AppendOnlyFileStoreWrite newWrite() {
+ public AppendOnlyFileStoreWrite newWrite(String commitUser) {
return new AppendOnlyFileStoreWrite(
newRead(),
schemaId,
+ commitUser,
rowType,
pathFactory(),
snapshotManager(),
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStore.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStore.java
index 1220545b..157d22d2 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStore.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStore.java
@@ -46,9 +46,9 @@ public interface FileStore<T> extends Serializable {
FileStoreRead<T> newRead();
- FileStoreWrite<T> newWrite();
+ FileStoreWrite<T> newWrite(String commitUser);
- FileStoreCommit newCommit(String user);
+ FileStoreCommit newCommit(String commitUser);
FileStoreExpire newExpire();
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueFileStore.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueFileStore.java
index 980ea4fc..7f7602b6 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueFileStore.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueFileStore.java
@@ -78,10 +78,11 @@ public class KeyValueFileStore extends AbstractFileStore<KeyValue> {
}
@Override
- public KeyValueFileStoreWrite newWrite() {
+ public KeyValueFileStoreWrite newWrite(String commitUser) {
return new KeyValueFileStoreWrite(
schemaManager,
schemaId,
+ commitUser,
keyType,
valueType,
keyComparatorSupplier,
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreWrite.java
index 0e9ed821..6de663c4 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreWrite.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.store.file.operation;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.store.file.Snapshot;
import org.apache.flink.table.store.file.io.DataFileMeta;
import org.apache.flink.table.store.file.manifest.ManifestEntry;
import org.apache.flink.table.store.file.utils.RecordWriter;
@@ -28,11 +29,10 @@ import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.store.table.sink.FileCommittable;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
-import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
-
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
@@ -48,17 +48,20 @@ import java.util.concurrent.Executors;
*/
public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
+ private final String commitUser;
private final SnapshotManager snapshotManager;
private final FileStoreScan scan;
@Nullable protected IOManager ioManager;
- protected final Map<BinaryRowData, Map<Integer, RecordWriter<T>>> writers;
+ protected final Map<BinaryRowData, Map<Integer, WriterWithCommit<T>>> writers;
private final ExecutorService compactExecutor;
private boolean overwrite = false;
- protected AbstractFileStoreWrite(SnapshotManager snapshotManager, FileStoreScan scan) {
+ protected AbstractFileStoreWrite(
+ String commitUser, SnapshotManager snapshotManager, FileStoreScan scan) {
+ this.commitUser = commitUser;
this.snapshotManager = snapshotManager;
this.scan = scan;
@@ -76,7 +79,7 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
protected List<DataFileMeta> scanExistingFileMetas(BinaryRowData partition, int bucket) {
Long latestSnapshotId = snapshotManager.latestSnapshotId();
- List<DataFileMeta> existingFileMetas = Lists.newArrayList();
+ List<DataFileMeta> existingFileMetas = new ArrayList<>();
if (latestSnapshotId != null) {
// Concat all the DataFileMeta of existing files into existingFileMetas.
scan.withSnapshot(latestSnapshotId)
@@ -104,21 +107,48 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
}
@Override
- public List<FileCommittable> prepareCommit(boolean endOfInput) throws Exception {
+ public List<FileCommittable> prepareCommit(boolean endOfInput, long commitIdentifier)
+ throws Exception {
+ long latestCommittedIdentifier;
+ if (writers.values().stream()
+ .map(Map::values)
+ .flatMap(Collection::stream)
+ .mapToLong(w -> w.lastModifiedCommitIdentifier)
+ .max()
+ .orElse(Long.MIN_VALUE)
+ == Long.MIN_VALUE) {
+ // Optimization for the first commit.
+ //
+ // If this is the first commit, no writer has previous modified commit, so the value of
+ // `latestCommittedIdentifier` does not matter.
+ //
+ // Without this optimization, we may need to scan through all snapshots only to find
+ // that there is no previous snapshot by this user, which is very inefficient.
+ latestCommittedIdentifier = Long.MIN_VALUE;
+ } else {
+ latestCommittedIdentifier =
+ snapshotManager
+ .latestSnapshotOfUser(commitUser)
+ .map(Snapshot::commitIdentifier)
+ .orElse(Long.MIN_VALUE);
+ }
+
List<FileCommittable> result = new ArrayList<>();
- Iterator<Map.Entry<BinaryRowData, Map<Integer, RecordWriter<T>>>> partIter =
+ Iterator<Map.Entry<BinaryRowData, Map<Integer, WriterWithCommit<T>>>> partIter =
writers.entrySet().iterator();
while (partIter.hasNext()) {
- Map.Entry<BinaryRowData, Map<Integer, RecordWriter<T>>> partEntry = partIter.next();
+ Map.Entry<BinaryRowData, Map<Integer, WriterWithCommit<T>>> partEntry = partIter.next();
BinaryRowData partition = partEntry.getKey();
- Iterator<Map.Entry<Integer, RecordWriter<T>>> bucketIter =
+ Iterator<Map.Entry<Integer, WriterWithCommit<T>>> bucketIter =
partEntry.getValue().entrySet().iterator();
while (bucketIter.hasNext()) {
- Map.Entry<Integer, RecordWriter<T>> entry = bucketIter.next();
+ Map.Entry<Integer, WriterWithCommit<T>> entry = bucketIter.next();
int bucket = entry.getKey();
- RecordWriter<T> writer = entry.getValue();
- RecordWriter.CommitIncrement increment = writer.prepareCommit(endOfInput);
+ WriterWithCommit<T> writerWithCommit = entry.getValue();
+
+ RecordWriter.CommitIncrement increment =
+ writerWithCommit.writer.prepareCommit(endOfInput);
FileCommittable committable =
new FileCommittable(
partition,
@@ -127,12 +157,18 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
increment.compactIncrement());
result.add(committable);
- // clear if no update
- // we need a mechanism to clear writers, otherwise there will be more and more
- // such as yesterday's partition that no longer needs to be written.
if (committable.isEmpty()) {
- writer.close();
- bucketIter.remove();
+ if (writerWithCommit.lastModifiedCommitIdentifier
+ <= latestCommittedIdentifier) {
+ // Clear writer if no update, and if its latest modification has committed.
+ //
+ // We need a mechanism to clear writers, otherwise there will be more and
+ // more such as yesterday's partition that no longer needs to be written.
+ writerWithCommit.writer.close();
+ bucketIter.remove();
+ }
+ } else {
+ writerWithCommit.lastModifiedCommitIdentifier = commitIdentifier;
}
}
@@ -146,9 +182,9 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
@Override
public void close() throws Exception {
- for (Map<Integer, RecordWriter<T>> bucketWriters : writers.values()) {
- for (RecordWriter<T> writer : bucketWriters.values()) {
- writer.close();
+ for (Map<Integer, WriterWithCommit<T>> bucketWriters : writers.values()) {
+ for (WriterWithCommit<T> writerWithCommit : bucketWriters.values()) {
+ writerWithCommit.writer.close();
}
}
writers.clear();
@@ -156,21 +192,21 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
}
private RecordWriter<T> getWriter(BinaryRowData partition, int bucket) {
- Map<Integer, RecordWriter<T>> buckets = writers.get(partition);
+ Map<Integer, WriterWithCommit<T>> buckets = writers.get(partition);
if (buckets == null) {
buckets = new HashMap<>();
writers.put(partition.copy(), buckets);
}
- return buckets.computeIfAbsent(bucket, k -> createWriter(partition.copy(), bucket));
+ return buckets.computeIfAbsent(bucket, k -> createWriter(partition.copy(), bucket)).writer;
}
- private RecordWriter<T> createWriter(BinaryRowData partition, int bucket) {
+ private WriterWithCommit<T> createWriter(BinaryRowData partition, int bucket) {
RecordWriter<T> writer =
overwrite
? createEmptyWriter(partition.copy(), bucket, compactExecutor)
: createWriter(partition.copy(), bucket, compactExecutor);
notifyNewWriter(writer);
- return writer;
+ return new WriterWithCommit<>(writer);
}
protected void notifyNewWriter(RecordWriter<T> writer) {}
@@ -184,4 +220,16 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
@VisibleForTesting
public abstract RecordWriter<T> createEmptyWriter(
BinaryRowData partition, int bucket, ExecutorService compactExecutor);
+
+ /** {@link RecordWriter} with identifier of its last modified commit. */
+ protected static class WriterWithCommit<T> {
+
+ protected final RecordWriter<T> writer;
+ private long lastModifiedCommitIdentifier;
+
+ private WriterWithCommit(RecordWriter<T> writer) {
+ this.writer = writer;
+ this.lastModifiedCommitIdentifier = Long.MIN_VALUE;
+ }
+ }
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AppendOnlyFileStoreWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AppendOnlyFileStoreWrite.java
index 41271012..bb7921c4 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AppendOnlyFileStoreWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AppendOnlyFileStoreWrite.java
@@ -61,12 +61,13 @@ public class AppendOnlyFileStoreWrite extends AbstractFileStoreWrite<RowData> {
public AppendOnlyFileStoreWrite(
AppendOnlyFileStoreRead read,
long schemaId,
+ String commitUser,
RowType rowType,
FileStorePathFactory pathFactory,
SnapshotManager snapshotManager,
FileStoreScan scan,
CoreOptions options) {
- super(snapshotManager, scan);
+ super(commitUser, snapshotManager, scan);
this.read = read;
this.schemaId = schemaId;
this.rowType = rowType;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWrite.java
index b40d4d64..3e93b43e 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWrite.java
@@ -68,10 +68,11 @@ public interface FileStoreWrite<T> {
* Prepare commit in the write.
*
* @param endOfInput if true, the data writing is ended
+ * @param commitIdentifier identifier of the commit being prepared
* @return the file committable list
* @throws Exception the thrown exception
*/
- List<FileCommittable> prepareCommit(boolean endOfInput) throws Exception;
+ List<FileCommittable> prepareCommit(boolean endOfInput, long commitIdentifier) throws Exception;
/**
* Close the writer.
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java
index 1e7b5200..687de200 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java
@@ -62,6 +62,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
public KeyValueFileStoreWrite(
SchemaManager schemaManager,
long schemaId,
+ String commitUser,
RowType keyType,
RowType valueType,
Supplier<Comparator<RowData>> keyComparatorSupplier,
@@ -70,7 +71,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
SnapshotManager snapshotManager,
FileStoreScan scan,
CoreOptions options) {
- super(snapshotManager, scan, options);
+ super(commitUser, snapshotManager, scan, options);
this.readerFactoryBuilder =
KeyValueFileReaderFactory.builder(
schemaManager,
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/MemoryFileStoreWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/MemoryFileStoreWrite.java
index 84c08788..77bd7d7b 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/MemoryFileStoreWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/MemoryFileStoreWrite.java
@@ -40,15 +40,18 @@ public abstract class MemoryFileStoreWrite<T> extends AbstractFileStoreWrite<T>
private final MemoryPoolFactory memoryPoolFactory;
public MemoryFileStoreWrite(
- SnapshotManager snapshotManager, FileStoreScan scan, CoreOptions options) {
- super(snapshotManager, scan);
+ String commitUser,
+ SnapshotManager snapshotManager,
+ FileStoreScan scan,
+ CoreOptions options) {
+ super(commitUser, snapshotManager, scan);
HeapMemorySegmentPool memoryPool =
new HeapMemorySegmentPool(options.writeBufferSize(), options.pageSize());
this.memoryPoolFactory = new MemoryPoolFactory(memoryPool, this::memoryOwners);
}
private Iterator<MemoryOwner> memoryOwners() {
- Iterator<Map<Integer, RecordWriter<T>>> iterator = writers.values().iterator();
+ Iterator<Map<Integer, WriterWithCommit<T>>> iterator = writers.values().iterator();
return Iterators.concat(
new Iterator<Iterator<MemoryOwner>>() {
@Override
@@ -60,7 +63,10 @@ public abstract class MemoryFileStoreWrite<T> extends AbstractFileStoreWrite<T>
public Iterator<MemoryOwner> next() {
return Iterators.transform(
iterator.next().values().iterator(),
- writer -> (MemoryOwner) writer);
+ writerWithCommit ->
+ writerWithCommit == null
+ ? null
+ : (MemoryOwner) (writerWithCommit.writer));
}
});
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
index 36d954d5..c1c59f2a 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
@@ -61,7 +61,7 @@ public abstract class AbstractFileStoreTable implements FileStoreTable {
}
@Override
- public TableCommit newCommit(String user) {
- return new TableCommit(store().newCommit(user), store().newExpire());
+ public TableCommit newCommit(String commitUser) {
+ return new TableCommit(store().newCommit(commitUser), store().newExpire());
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTable.java
index 4d638fb7..665a241a 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTable.java
@@ -104,11 +104,11 @@ public class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
}
@Override
- public TableWrite newWrite() {
+ public TableWrite newWrite(String commitUser) {
SinkRecordConverter recordConverter =
new SinkRecordConverter(store.options().bucket(), tableSchema);
return new TableWriteImpl<>(
- store.newWrite(),
+ store.newWrite(commitUser),
recordConverter,
record -> {
Preconditions.checkState(
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java
index ebde8da4..b96cbaa9 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java
@@ -117,12 +117,12 @@ public class ChangelogValueCountFileStoreTable extends AbstractFileStoreTable {
}
@Override
- public TableWrite newWrite() {
+ public TableWrite newWrite(String commitUser) {
SinkRecordConverter recordConverter =
new SinkRecordConverter(store.options().bucket(), tableSchema);
final KeyValue kv = new KeyValue();
return new TableWriteImpl<>(
- store.newWrite(),
+ store.newWrite(commitUser),
recordConverter,
record -> {
switch (record.row().getRowKind()) {
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
index 1a0a8263..79dff30a 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
@@ -186,7 +186,7 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
}
@Override
- public TableWrite newWrite() {
+ public TableWrite newWrite(String commitUser) {
SinkRecordConverter recordConverter =
new SinkRecordConverter(store.options().bucket(), tableSchema);
final SequenceGenerator sequenceGenerator =
@@ -196,7 +196,7 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
.orElse(null);
final KeyValue kv = new KeyValue();
return new TableWriteImpl<>(
- store.newWrite(),
+ store.newWrite(commitUser),
recordConverter,
record -> {
long sequenceNumber =
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTable.java
index d99019c3..2fd56ae6 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTable.java
@@ -22,8 +22,6 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.file.schema.TableSchema;
import org.apache.flink.table.store.file.utils.SnapshotManager;
-import org.apache.flink.table.store.table.sink.TableCommit;
-import org.apache.flink.table.store.table.sink.TableWrite;
import org.apache.flink.table.store.table.source.DataTableScan;
import org.apache.flink.table.types.logical.RowType;
@@ -60,8 +58,4 @@ public interface FileStoreTable extends Table, SupportsPartition, SupportsWrite
@Override
DataTableScan newScan();
-
- TableWrite newWrite();
-
- TableCommit newCommit(String user);
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/SupportsWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/SupportsWrite.java
index 97dfa5e5..696aa3eb 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/SupportsWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/SupportsWrite.java
@@ -24,7 +24,7 @@ import org.apache.flink.table.store.table.sink.TableWrite;
/** An interface for {@link Table} write support. */
public interface SupportsWrite {
- TableWrite newWrite();
+ TableWrite newWrite(String commitUser);
- TableCommit newCommit(String user);
+ TableCommit newCommit(String commitUser);
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableWrite.java
index 647d74e8..b2e64034 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableWrite.java
@@ -40,7 +40,7 @@ public interface TableWrite {
void compact(BinaryRowData partition, int bucket) throws Exception;
- List<FileCommittable> prepareCommit(boolean endOfInput) throws Exception;
+ List<FileCommittable> prepareCommit(boolean endOfInput, long commitIdentifier) throws Exception;
void close() throws Exception;
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableWriteImpl.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableWriteImpl.java
index 83abf141..03466edc 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableWriteImpl.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableWriteImpl.java
@@ -75,8 +75,9 @@ public class TableWriteImpl<T> implements TableWrite {
}
@Override
- public List<FileCommittable> prepareCommit(boolean endOfInput) throws Exception {
- return write.prepareCommit(endOfInput);
+ public List<FileCommittable> prepareCommit(boolean endOfInput, long commitIdentifier)
+ throws Exception {
+ return write.prepareCommit(endOfInput, commitIdentifier);
}
@Override
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
index c67c81a0..09722d51 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
@@ -82,7 +82,7 @@ public class TestFileStore extends KeyValueFileStore {
private final String root;
private final RowDataSerializer keySerializer;
private final RowDataSerializer valueSerializer;
- private final String user;
+ private final String commitUser;
private long commitIdentifier;
@@ -105,13 +105,13 @@ public class TestFileStore extends KeyValueFileStore {
this.root = root;
this.keySerializer = new RowDataSerializer(keyType);
this.valueSerializer = new RowDataSerializer(valueType);
- this.user = UUID.randomUUID().toString();
+ this.commitUser = UUID.randomUUID().toString();
this.commitIdentifier = 0L;
}
public FileStoreCommitImpl newCommit() {
- return super.newCommit(user);
+ return super.newCommit(commitUser);
}
public FileStoreExpireImpl newExpire(
@@ -176,7 +176,7 @@ public class TestFileStore extends KeyValueFileStore {
Long identifier,
BiConsumer<FileStoreCommit, ManifestCommittable> commitFunction)
throws Exception {
- AbstractFileStoreWrite<KeyValue> write = newWrite();
+ AbstractFileStoreWrite<KeyValue> write = newWrite(commitUser);
Map<BinaryRowData, Map<Integer, RecordWriter<KeyValue>>> writers = new HashMap<>();
for (KeyValue kv : kvs) {
BinaryRowData partition = partitionCalculator.apply(kv);
@@ -206,7 +206,7 @@ public class TestFileStore extends KeyValueFileStore {
.write(kv);
}
- FileStoreCommit commit = newCommit(user);
+ FileStoreCommit commit = newCommit(commitUser);
ManifestCommittable committable =
new ManifestCommittable(identifier == null ? commitIdentifier++ : identifier);
for (Map.Entry<BinaryRowData, Map<Integer, RecordWriter<KeyValue>>> entryWithPartition :
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/TestCommitThread.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/TestCommitThread.java
index 5349fa43..10a183f3 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/TestCommitThread.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/TestCommitThread.java
@@ -92,8 +92,9 @@ public class TestCommitThread extends Thread {
.collect(Collectors.joining("\n")));
}
- this.write = safeStore.newWrite();
- this.commit = testStore.newCommit(UUID.randomUUID().toString()).withCreateEmptyCommit(true);
+ String commitUser = UUID.randomUUID().toString();
+ this.write = safeStore.newWrite(commitUser);
+ this.commit = testStore.newCommit(commitUser).withCreateEmptyCommit(true);
this.commitIdentifier = 0;
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/FailingAtomicRenameFileSystem.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/FailingAtomicRenameFileSystem.java
index b922423f..6b3500ba 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/FailingAtomicRenameFileSystem.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/FailingAtomicRenameFileSystem.java
@@ -30,6 +30,7 @@ import org.apache.flink.core.fs.LocatedFileStatus;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.local.LocalBlockLocation;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.function.RunnableWithException;
import java.io.File;
import java.io.FileNotFoundException;
@@ -238,4 +239,17 @@ public class FailingAtomicRenameFileSystem extends TestAtomicRenameFileSystem {
}
}
}
+
+ public static void retryArtificialException(RunnableWithException runnable) throws Exception {
+ while (true) {
+ try {
+ runnable.run();
+ break;
+ } catch (Throwable t) {
+ if (!ExceptionUtils.findThrowable(t, ArtificialException.class).isPresent()) {
+ throw t;
+ }
+ }
+ }
+ }
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
index 8bbff172..58b78e17 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
@@ -162,9 +162,9 @@ public class AppendOnlyFileStoreTableTest extends FileStoreTableTestBase {
int numOfBucket = Math.max(random.nextInt(8), 1);
FileStoreTable table = createFileStoreTable(numOfBucket);
RowDataSerializer serializer = new RowDataSerializer(table.schema().logicalRowType());
- TableWrite write = table.newWrite();
+ TableWrite write = table.newWrite(commitUser);
- TableCommit commit = table.newCommit("user");
+ TableCommit commit = table.newCommit(commitUser);
List<Map<Integer, List<RowData>>> dataset = new ArrayList<>();
Map<Integer, List<RowData>> dataPerBucket = new HashMap<>(numOfBucket);
int numOfPartition = Math.max(random.nextInt(10), 1);
@@ -189,7 +189,7 @@ public class AppendOnlyFileStoreTableTest extends FileStoreTableTestBase {
dataset.add(new HashMap<>(dataPerBucket));
dataPerBucket.clear();
}
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
int partition = random.nextInt(numOfPartition);
List<Integer> availableBucket = new ArrayList<>(dataset.get(partition).keySet());
@@ -214,23 +214,23 @@ public class AppendOnlyFileStoreTableTest extends FileStoreTableTestBase {
private void writeData() throws Exception {
FileStoreTable table = createFileStoreTable();
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 100L));
write.write(rowData(2, 20, 200L));
write.write(rowData(1, 11, 101L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.write(rowData(1, 12, 102L));
write.write(rowData(2, 21, 201L));
write.write(rowData(2, 22, 202L));
- commit.commit(1, write.prepareCommit(true));
+ commit.commit(1, write.prepareCommit(true, 1));
write.write(rowData(1, 11, 101L));
write.write(rowData(2, 21, 201L));
write.write(rowData(1, 12, 102L));
- commit.commit(2, write.prepareCommit(true));
+ commit.commit(2, write.prepareCommit(true, 2));
write.close();
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
index 630ab6a4..c74abf7e 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
@@ -151,26 +151,26 @@ public class ChangelogValueCountFileStoreTableTest extends FileStoreTableTestBas
private void writeData() throws Exception {
FileStoreTable table = createFileStoreTable();
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 100L));
write.write(rowData(2, 20, 200L));
write.write(rowData(1, 11, 101L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.write(rowData(2, 21, 201L));
write.write(rowData(1, 12, 102L));
write.write(rowData(2, 21, 201L));
write.write(rowData(2, 21, 201L));
- commit.commit(1, write.prepareCommit(true));
+ commit.commit(1, write.prepareCommit(true, 1));
write.write(rowData(1, 11, 101L));
write.write(rowData(2, 22, 202L));
write.write(rowDataWithKind(RowKind.DELETE, 2, 21, 201L));
write.write(rowDataWithKind(RowKind.DELETE, 1, 10, 100L));
write.write(rowDataWithKind(RowKind.DELETE, 2, 21, 201L));
- commit.commit(2, write.prepareCommit(true));
+ commit.commit(2, write.prepareCommit(true, 2));
write.close();
}
@@ -178,13 +178,13 @@ public class ChangelogValueCountFileStoreTableTest extends FileStoreTableTestBas
@Test
public void testChangelogWithoutDataFile() throws Exception {
FileStoreTable table = createFileStoreTable();
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
// no data file should be produced from this commit
write.write(rowData(1, 10, 100L));
write.write(rowDataWithKind(RowKind.DELETE, 1, 10, 100L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.close();
// check that no data file is produced
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
index ef071883..2c9fc7fe 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.table.store.file.predicate.PredicateBuilder;
import org.apache.flink.table.store.file.schema.SchemaManager;
import org.apache.flink.table.store.file.schema.TableSchema;
import org.apache.flink.table.store.file.schema.UpdateSchema;
+import org.apache.flink.table.store.table.sink.FileCommittable;
import org.apache.flink.table.store.table.sink.TableCommit;
import org.apache.flink.table.store.table.sink.TableWrite;
import org.apache.flink.table.store.table.source.DataTableScan;
@@ -54,14 +55,14 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
public void testSequenceNumber() throws Exception {
FileStoreTable table =
createFileStoreTable(conf -> conf.set(CoreOptions.SEQUENCE_FIELD, "b"));
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 200L));
write.write(rowData(1, 10, 100L));
write.write(rowData(1, 11, 101L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.write(rowData(1, 11, 55L));
- commit.commit(1, write.prepareCommit(true));
+ commit.commit(1, write.prepareCommit(true, 1));
write.close();
List<Split> splits = table.newScan().plan().splits();
@@ -172,8 +173,8 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
FileStoreTable table =
createFileStoreTable(
conf -> conf.set(CoreOptions.CHANGELOG_PRODUCER, ChangelogProducer.INPUT));
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 100L));
write.write(rowData(1, 20, 200L));
write.write(rowDataWithKind(RowKind.DELETE, 1, 10, 100L));
@@ -182,7 +183,7 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
write.write(rowDataWithKind(RowKind.UPDATE_AFTER, 1, 20, 201L));
write.write(rowDataWithKind(RowKind.UPDATE_BEFORE, 1, 10, 101L));
write.write(rowDataWithKind(RowKind.UPDATE_AFTER, 1, 10, 102L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.close();
List<Split> splits = table.newScan().withIncremental(true).plan().splits();
@@ -207,8 +208,8 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
conf.set(
CoreOptions.CHANGELOG_PRODUCER,
ChangelogProducer.FULL_COMPACTION));
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 110L));
write.write(rowData(1, 20, 120L));
@@ -217,7 +218,7 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
write.write(rowDataWithKind(RowKind.DELETE, 2, 10, 210L));
write.compact(binaryRow(1), 0);
write.compact(binaryRow(2), 0);
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
List<Split> splits = table.newScan().withIncremental(true).plan().splits();
TableRead read = table.newRead();
@@ -231,13 +232,13 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
write.write(rowData(1, 40, 140L));
write.write(rowData(2, 30, 230L));
write.write(rowData(2, 40, 240L));
- commit.commit(1, write.prepareCommit(true));
+ commit.commit(1, write.prepareCommit(true, 1));
write.write(rowDataWithKind(RowKind.DELETE, 1, 40, 140L));
write.write(rowData(2, 40, 241L));
write.compact(binaryRow(1), 0);
write.compact(binaryRow(2), 0);
- commit.commit(2, write.prepareCommit(true));
+ commit.commit(2, write.prepareCommit(true, 2));
splits = table.newScan().withIncremental(true).plan().splits();
assertThat(getResult(read, splits, binaryRow(1), 0, CHANGELOG_ROW_TO_STRING))
@@ -249,7 +250,7 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
write.write(rowData(1, 20, 121L));
write.write(rowData(1, 30, 131L));
write.write(rowData(2, 30, 231L));
- commit.commit(3, write.prepareCommit(true));
+ commit.commit(3, write.prepareCommit(true, 3));
write.write(rowDataWithKind(RowKind.DELETE, 1, 20, 121L));
write.write(rowData(1, 30, 132L));
@@ -260,7 +261,7 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
write.write(rowData(2, 40, 242L));
write.compact(binaryRow(1), 0);
write.compact(binaryRow(2), 0);
- commit.commit(4, write.prepareCommit(true));
+ commit.commit(4, write.prepareCommit(true, 4));
splits = table.newScan().withIncremental(true).plan().splits();
assertThat(getResult(read, splits, binaryRow(1), 0, CHANGELOG_ROW_TO_STRING))
@@ -355,13 +356,13 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
assertThat(enumerator.call()).isNull();
// write another commit
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowDataWithKind(RowKind.UPDATE_BEFORE, 1, 10, 102L));
write.write(rowDataWithKind(RowKind.UPDATE_AFTER, 1, 10, 103L));
write.write(rowDataWithKind(RowKind.INSERT, 1, 20, 201L));
write.write(rowDataWithKind(RowKind.DELETE, 2, 10, 301L));
- commit.commit(2, write.prepareCommit(true));
+ commit.commit(2, write.prepareCommit(true, 2));
write.close();
assertNextSnapshot.apply(2);
@@ -372,25 +373,25 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
private void writeData() throws Exception {
FileStoreTable table = createFileStoreTable();
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 100L));
write.write(rowData(2, 20, 200L));
write.write(rowData(1, 11, 101L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.write(rowData(1, 10, 1000L));
write.write(rowData(2, 21, 201L));
write.write(rowData(2, 21, 2001L));
- commit.commit(1, write.prepareCommit(true));
+ commit.commit(1, write.prepareCommit(true, 1));
write.write(rowData(1, 11, 1001L));
write.write(rowData(2, 21, 20001L));
write.write(rowData(2, 22, 202L));
write.write(rowDataWithKind(RowKind.DELETE, 1, 11, 1001L));
write.write(rowDataWithKind(RowKind.DELETE, 2, 20, 200L));
- commit.commit(2, write.prepareCommit(true));
+ commit.commit(2, write.prepareCommit(true, 2));
write.close();
}
@@ -400,20 +401,20 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
public void testReadFilter() throws Exception {
FileStoreTable table = createFileStoreTable();
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 100L));
write.write(rowData(1, 20, 200L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.write(rowData(1, 30, 300L));
write.write(rowData(1, 40, 400L));
- commit.commit(1, write.prepareCommit(true));
+ commit.commit(1, write.prepareCommit(true, 1));
write.write(rowData(1, 50, 500L));
write.write(rowData(1, 60, 600L));
- commit.commit(2, write.prepareCommit(true));
+ commit.commit(2, write.prepareCommit(true, 2));
PredicateBuilder builder = new PredicateBuilder(ROW_TYPE);
List<Split> splits = table.newScan().plan().splits();
@@ -446,7 +447,7 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
// update pk 60, 10
write.write(rowData(1, 60, 500L));
write.write(rowData(1, 10, 10L));
- commit.commit(3, write.prepareCommit(true));
+ commit.commit(3, write.prepareCommit(true, 3));
write.close();
@@ -474,13 +475,13 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
CoreOptions.MergeEngine.PARTIAL_UPDATE);
conf.set(CoreOptions.PARTIAL_UPDATE_IGNORE_DELETE, true);
});
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 200L));
write.write(rowDataWithKind(RowKind.DELETE, 1, 10, 100L));
write.write(rowDataWithKind(RowKind.UPDATE_BEFORE, 1, 10, 100L));
write.write(rowDataWithKind(RowKind.DELETE, 1, 20, 400L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.close();
List<Split> splits = table.newScan().plan().splits();
@@ -489,6 +490,36 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
.isEqualTo(Collections.singletonList("1|10|200|binary|varbinary"));
}
+ @Test
+ public void testSlowCommit() throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
+
+ write.write(rowData(1, 10, 100L));
+ write.write(rowData(1, 20, 200L));
+ List<FileCommittable> committables0 = write.prepareCommit(false, 0);
+
+ write.write(rowData(2, 10, 300L));
+ List<FileCommittable> committables1 = write.prepareCommit(false, 1);
+
+ write.write(rowData(1, 20, 201L));
+ List<FileCommittable> committables2 = write.prepareCommit(true, 2);
+
+ commit.commit(0, committables0);
+ commit.commit(1, committables1);
+ commit.commit(2, committables2);
+
+ write.close();
+
+ List<Split> splits = table.newScan().plan().splits();
+ TableRead read = table.newRead();
+ assertThat(getResult(read, splits, binaryRow(1), 0, BATCH_ROW_TO_STRING))
+ .isEqualTo(Arrays.asList("1|10|100|binary|varbinary", "1|20|201|binary|varbinary"));
+ assertThat(getResult(read, splits, binaryRow(2), 0, BATCH_ROW_TO_STRING))
+ .isEqualTo(Collections.singletonList("2|10|300|binary|varbinary"));
+ }
+
@Override
protected FileStoreTable createFileStoreTable(Consumer<Configuration> configure)
throws Exception {
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
index 46cd2645..3aeffe9a 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
@@ -58,6 +58,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
@@ -111,10 +112,12 @@ public abstract class FileStoreTableTestBase {
@TempDir java.nio.file.Path tempDir;
protected Path tablePath;
+ protected String commitUser;
@BeforeEach
public void before() {
tablePath = new Path(TestAtomicRenameFileSystem.SCHEME + "://" + tempDir.toString());
+ commitUser = UUID.randomUUID().toString();
}
@AfterEach
@@ -133,19 +136,19 @@ public abstract class FileStoreTableTestBase {
public void testOverwrite() throws Exception {
FileStoreTable table = createFileStoreTable();
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 100L));
write.write(rowData(2, 20, 200L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.close();
- write = table.newWrite().withOverwrite(true);
- commit = table.newCommit("user");
+ write = table.newWrite(commitUser).withOverwrite(true);
+ commit = table.newCommit(commitUser);
write.write(rowData(2, 21, 201L));
Map<String, String> overwritePartition = new HashMap<>();
overwritePartition.put("pt", "2");
- commit.withOverwritePartition(overwritePartition).commit(1, write.prepareCommit(true));
+ commit.withOverwritePartition(overwritePartition).commit(1, write.prepareCommit(true, 1));
write.close();
List<Split> splits = table.newScan().plan().splits();
@@ -165,13 +168,13 @@ public abstract class FileStoreTableTestBase {
conf.set(BUCKET_KEY, "a");
});
- TableWrite write = table.newWrite();
+ TableWrite write = table.newWrite(commitUser);
write.write(rowData(1, 1, 2L));
write.write(rowData(1, 3, 4L));
write.write(rowData(1, 5, 6L));
write.write(rowData(1, 7, 8L));
write.write(rowData(1, 9, 10L));
- table.newCommit("user").commit(0, write.prepareCommit(true));
+ table.newCommit(commitUser).commit(0, write.prepareCommit(true, 0));
write.close();
List<Split> splits =
@@ -187,20 +190,20 @@ public abstract class FileStoreTableTestBase {
public void testReadFilter() throws Exception {
FileStoreTable table = createFileStoreTable();
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(rowData(1, 10, 100L));
write.write(rowData(1, 20, 200L));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.write(rowData(1, 30, 300L));
write.write(rowData(1, 40, 400L));
- commit.commit(1, write.prepareCommit(true));
+ commit.commit(1, write.prepareCommit(true, 1));
write.write(rowData(1, 50, 500L));
write.write(rowData(1, 60, 600L));
- commit.commit(2, write.prepareCommit(true));
+ commit.commit(2, write.prepareCommit(true, 2));
write.close();
@@ -215,31 +218,31 @@ public abstract class FileStoreTableTestBase {
@Test
public void testPartitionEmptyWriter() throws Exception {
FileStoreTable table = createFileStoreTable();
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
for (int i = 0; i < 4; i++) {
// write lots of records, let compaction be slower
for (int j = 0; j < 1000; j++) {
write.write(rowData(1, 10 * i * j, 100L * i * j));
}
- commit.commit(i, write.prepareCommit(false));
+ commit.commit(i, write.prepareCommit(false, i));
}
write.write(rowData(1, 40, 400L));
- List<FileCommittable> commit4 = write.prepareCommit(false);
+ List<FileCommittable> commit4 = write.prepareCommit(false, 4);
// trigger compaction, but not wait it.
if (commit4.get(0).compactIncrement().compactBefore().isEmpty()) {
// commit4 is not a compaction commit
// do compaction commit5 and compaction commit6
write.write(rowData(2, 20, 200L));
- List<FileCommittable> commit5 = write.prepareCommit(true);
+ List<FileCommittable> commit5 = write.prepareCommit(true, 5);
// wait compaction finish
// commit5 should be a compaction commit
write.write(rowData(1, 60, 600L));
- List<FileCommittable> commit6 = write.prepareCommit(true);
+ List<FileCommittable> commit6 = write.prepareCommit(true, 6);
// if remove writer too fast, will see old files, do another compaction
// then will be conflicts
@@ -250,7 +253,7 @@ public abstract class FileStoreTableTestBase {
// commit4 is a compaction commit
// do compaction commit5
write.write(rowData(2, 20, 200L));
- List<FileCommittable> commit5 = write.prepareCommit(true);
+ List<FileCommittable> commit5 = write.prepareCommit(true, 5);
// wait compaction finish
// commit5 should be a compaction commit
@@ -270,11 +273,11 @@ public abstract class FileStoreTableTestBase {
conf.set(COMPACTION_MAX_FILE_NUM, 5);
});
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
for (int i = 0; i < 10; i++) {
write.write(rowData(1, 1, 100L));
- commit.commit(i, write.prepareCommit(true));
+ commit.commit(i, write.prepareCommit(true, i));
}
write.close();
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTest.java
index 250cfdfa..95b544df 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTest.java
@@ -49,11 +49,11 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.UUID;
import java.util.function.Consumer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.assertj.core.api.Assertions.filter;
/** Test for schema evolution. */
public class SchemaEvolutionTest {
@@ -61,13 +61,14 @@ public class SchemaEvolutionTest {
@TempDir java.nio.file.Path tempDir;
private Path tablePath;
-
private SchemaManager schemaManager;
+ private String commitUser;
@BeforeEach
public void beforeEach() {
tablePath = new Path(tempDir.toUri());
schemaManager = new SchemaManager(tablePath);
+ commitUser = UUID.randomUUID().toString();
}
@Test
@@ -83,20 +84,20 @@ public class SchemaEvolutionTest {
FileStoreTable table = FileStoreTableFactory.create(tablePath);
- TableWrite write = table.newWrite();
+ TableWrite write = table.newWrite(commitUser);
write.write(GenericRowData.of(1, 1L));
write.write(GenericRowData.of(2, 2L));
- table.newCommit("").commit(0, write.prepareCommit(true));
+ table.newCommit(commitUser).commit(0, write.prepareCommit(true, 0));
write.close();
schemaManager.commitChanges(
Collections.singletonList(SchemaChange.addColumn("f3", new BigIntType())));
table = FileStoreTableFactory.create(tablePath);
- write = table.newWrite();
+ write = table.newWrite(commitUser);
write.write(GenericRowData.of(3, 3L, 3L));
write.write(GenericRowData.of(4, 4L, 4L));
- table.newCommit("").commit(1, write.prepareCommit(true));
+ table.newCommit(commitUser).commit(1, write.prepareCommit(true, 1));
write.close();
// read all
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/WritePreemptMemoryTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/WritePreemptMemoryTest.java
index f5c77682..331e4bf4 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/WritePreemptMemoryTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/WritePreemptMemoryTest.java
@@ -26,7 +26,6 @@ import org.apache.flink.table.store.file.WriteMode;
import org.apache.flink.table.store.file.schema.SchemaManager;
import org.apache.flink.table.store.file.schema.TableSchema;
import org.apache.flink.table.store.file.schema.UpdateSchema;
-import org.apache.flink.table.store.table.sink.FileCommittable;
import org.apache.flink.table.store.table.sink.TableCommit;
import org.apache.flink.table.store.table.sink.TableWrite;
import org.apache.flink.table.store.table.source.Split;
@@ -63,8 +62,8 @@ public class WritePreemptMemoryTest extends FileStoreTableTestBase {
private void testWritePreemptMemory(boolean singlePartition) throws Exception {
// write
FileStoreTable table = createFileStoreTable();
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
Random random = new Random();
List<String> expected = new ArrayList<>();
for (int i = 0; i < 10_000; i++) {
@@ -72,8 +71,7 @@ public class WritePreemptMemoryTest extends FileStoreTableTestBase {
write.write(row);
expected.add(BATCH_ROW_TO_STRING.apply(row));
}
- List<FileCommittable> committables = write.prepareCommit(true);
- commit.commit(0, committables);
+ commit.commit(0, write.prepareCommit(true, 0));
write.close();
// read
diff --git a/flink-table-store-hive/flink-table-store-hive-connector/src/test/java/org/apache/flink/table/store/hive/TableStoreHiveStorageHandlerITCase.java b/flink-table-store-hive/flink-table-store-hive-connector/src/test/java/org/apache/flink/table/store/hive/TableStoreHiveStorageHandlerITCase.java
index c4112b36..f114e912 100644
--- a/flink-table-store-hive/flink-table-store-hive-connector/src/test/java/org/apache/flink/table/store/hive/TableStoreHiveStorageHandlerITCase.java
+++ b/flink-table-store-hive/flink-table-store-hive-connector/src/test/java/org/apache/flink/table/store/hive/TableStoreHiveStorageHandlerITCase.java
@@ -76,6 +76,7 @@ public class TableStoreHiveStorageHandlerITCase {
private static String engine;
+ private String commitUser;
private long commitIdentifier;
@BeforeClass
@@ -101,6 +102,7 @@ public class TableStoreHiveStorageHandlerITCase {
hiveShell.execute("CREATE DATABASE IF NOT EXISTS test_db");
hiveShell.execute("USE test_db");
+ commitUser = UUID.randomUUID().toString();
commitIdentifier = 0;
}
@@ -625,15 +627,17 @@ public class TableStoreHiveStorageHandlerITCase {
private String writeData(FileStoreTable table, String path, List<RowData> data)
throws Exception {
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
for (RowData rowData : data) {
write.write(rowData);
if (ThreadLocalRandom.current().nextInt(5) == 0) {
- commit.commit(commitIdentifier++, write.prepareCommit(false));
+ commit.commit(commitIdentifier, write.prepareCommit(false, commitIdentifier));
+ commitIdentifier++;
}
}
- commit.commit(commitIdentifier++, write.prepareCommit(true));
+ commit.commit(commitIdentifier, write.prepareCommit(true, commitIdentifier));
+ commitIdentifier++;
write.close();
String tableName = "test_table_" + (UUID.randomUUID().toString().substring(0, 4));
@@ -673,12 +677,12 @@ public class TableStoreHiveStorageHandlerITCase {
}
}
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
for (GenericRowData rowData : input) {
write.write(rowData);
}
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.close();
hiveShell.execute(
@@ -780,20 +784,20 @@ public class TableStoreHiveStorageHandlerITCase {
// TODO add NaN related tests after FLINK-27627 and FLINK-27628 are fixed
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(GenericRowData.of(1));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.write(GenericRowData.of((Object) null));
- commit.commit(1, write.prepareCommit(true));
+ commit.commit(1, write.prepareCommit(true, 1));
write.write(GenericRowData.of(2));
write.write(GenericRowData.of(3));
write.write(GenericRowData.of((Object) null));
- commit.commit(2, write.prepareCommit(true));
+ commit.commit(2, write.prepareCommit(true, 2));
write.write(GenericRowData.of(4));
write.write(GenericRowData.of(5));
write.write(GenericRowData.of(6));
- commit.commit(3, write.prepareCommit(true));
+ commit.commit(3, write.prepareCommit(true, 3));
write.close();
hiveShell.execute(
@@ -872,23 +876,23 @@ public class TableStoreHiveStorageHandlerITCase {
Collections.emptyList(),
Collections.emptyList());
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(
GenericRowData.of(
375, /* 1971-01-11 */
TimestampData.fromLocalDateTime(
LocalDateTime.of(2022, 5, 17, 17, 29, 20, 100_000_000))));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
write.write(GenericRowData.of(null, null));
- commit.commit(1, write.prepareCommit(true));
+ commit.commit(1, write.prepareCommit(true, 1));
write.write(GenericRowData.of(376 /* 1971-01-12 */, null));
write.write(
GenericRowData.of(
null,
TimestampData.fromLocalDateTime(
LocalDateTime.of(2022, 6, 18, 8, 30, 0, 100_000_000))));
- commit.commit(2, write.prepareCommit(true));
+ commit.commit(2, write.prepareCommit(true, 2));
write.close();
hiveShell.execute(
diff --git a/flink-table-store-hive/flink-table-store-hive-connector/src/test/java/org/apache/flink/table/store/mapred/TableStoreRecordReaderTest.java b/flink-table-store-hive/flink-table-store-hive-connector/src/test/java/org/apache/flink/table/store/mapred/TableStoreRecordReaderTest.java
index 3b1b56bc..76f3a0e5 100644
--- a/flink-table-store-hive/flink-table-store-hive-connector/src/test/java/org/apache/flink/table/store/mapred/TableStoreRecordReaderTest.java
+++ b/flink-table-store-hive/flink-table-store-hive-connector/src/test/java/org/apache/flink/table/store/mapred/TableStoreRecordReaderTest.java
@@ -36,6 +36,7 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -46,6 +47,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
@@ -53,6 +55,12 @@ import static org.assertj.core.api.Assertions.assertThat;
public class TableStoreRecordReaderTest {
@TempDir java.nio.file.Path tempDir;
+ private String commitUser;
+
+ @BeforeEach
+ public void beforeEach() {
+ commitUser = UUID.randomUUID().toString();
+ }
@Test
public void testPk() throws Exception {
@@ -71,14 +79,14 @@ public class TableStoreRecordReaderTest {
Collections.emptyList(),
Collections.singletonList("a"));
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(GenericRowData.of(1L, StringData.fromString("Hi")));
write.write(GenericRowData.of(2L, StringData.fromString("Hello")));
write.write(GenericRowData.of(3L, StringData.fromString("World")));
write.write(GenericRowData.of(1L, StringData.fromString("Hi again")));
write.write(GenericRowData.ofKind(RowKind.DELETE, 2L, StringData.fromString("Hello")));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
TableStoreRecordReader reader = read(table, BinaryRowDataUtil.EMPTY_ROW, 0);
RowDataContainer container = reader.createValue();
@@ -112,15 +120,15 @@ public class TableStoreRecordReaderTest {
Collections.emptyList(),
Collections.emptyList());
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(GenericRowData.of(1, StringData.fromString("Hi")));
write.write(GenericRowData.of(2, StringData.fromString("Hello")));
write.write(GenericRowData.of(3, StringData.fromString("World")));
write.write(GenericRowData.of(1, StringData.fromString("Hi")));
write.write(GenericRowData.ofKind(RowKind.DELETE, 2, StringData.fromString("Hello")));
write.write(GenericRowData.of(1, StringData.fromString("Hi")));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
TableStoreRecordReader reader = read(table, BinaryRowDataUtil.EMPTY_ROW, 0);
RowDataContainer container = reader.createValue();
@@ -155,12 +163,12 @@ public class TableStoreRecordReaderTest {
Collections.emptyList(),
Collections.emptyList());
- TableWrite write = table.newWrite();
- TableCommit commit = table.newCommit("user");
+ TableWrite write = table.newWrite(commitUser);
+ TableCommit commit = table.newCommit(commitUser);
write.write(GenericRowData.of(1, 10L, StringData.fromString("Hi")));
write.write(GenericRowData.of(2, 20L, StringData.fromString("Hello")));
write.write(GenericRowData.of(1, 10L, StringData.fromString("Hi")));
- commit.commit(0, write.prepareCommit(true));
+ commit.commit(0, write.prepareCommit(true, 0));
TableStoreRecordReader reader =
read(table, BinaryRowDataUtil.EMPTY_ROW, 0, Arrays.asList("c", "a"));
diff --git a/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java b/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java
index 1f2e4394..353c16e7 100644
--- a/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java
+++ b/flink-table-store-spark/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java
@@ -59,8 +59,10 @@ public class SimpleTableTestHelper {
Configuration conf = Configuration.fromMap(options);
conf.setString("path", path.toString());
FileStoreTable table = FileStoreTableFactory.create(conf);
- this.writer = table.newWrite();
- this.commit = table.newCommit("user");
+
+ String commitUser = "user";
+ this.writer = table.newWrite(commitUser);
+ this.commit = table.newCommit(commitUser);
this.commitIdentifier = 0;
}
@@ -70,6 +72,7 @@ public class SimpleTableTestHelper {
}
public void commit() throws Exception {
- commit.commit(commitIdentifier++, writer.prepareCommit(true));
+ commit.commit(commitIdentifier, writer.prepareCommit(true, commitIdentifier));
+ commitIdentifier++;
}
}
diff --git a/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java
index 8e7f10ef..de3ac94d 100644
--- a/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java
+++ b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java
@@ -57,8 +57,10 @@ public class SimpleTableTestHelper {
Configuration conf = Configuration.fromMap(options);
conf.setString("path", path.toString());
FileStoreTable table = FileStoreTableFactory.create(conf);
- this.writer = table.newWrite();
- this.commit = table.newCommit("user");
+
+ String commitUser = "user";
+ this.writer = table.newWrite(commitUser);
+ this.commit = table.newCommit(commitUser);
this.commitIdentifier = 0;
}
@@ -68,6 +70,7 @@ public class SimpleTableTestHelper {
}
public void commit() throws Exception {
- commit.commit(commitIdentifier++, writer.prepareCommit(true));
+ commit.commit(commitIdentifier, writer.prepareCommit(true, commitIdentifier));
+ commitIdentifier++;
}
}