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/09 08:07:42 UTC
[flink-table-store] branch release-0.2 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 release-0.2
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/release-0.2 by this push:
new fe0f333f [FLINK-29840] Fix bug that old record may overwrite new record in Table Store when snapshot committing is slow
fe0f333f is described below
commit fe0f333f5289081ca210540576e81c469ba18dd6
Author: tsreaper <ts...@gmail.com>
AuthorDate: Wed Nov 9 16:07:37 2022 +0800
[FLINK-29840] Fix bug that old record may overwrite new record in Table Store when snapshot committing is slow
This closes #366.
---
.../store/connector/sink/CommitterOperator.java | 39 ++---
.../table/store/connector/sink/StateUtils.java | 76 +++++++++
.../table/store/connector/sink/StoreSink.java | 17 +-
.../store/connector/sink/StoreWriteOperator.java | 39 ++++-
.../ChangelogWithKeyFileStoreTableITCase.java | 189 +++++++++++++++++++++
.../connector/sink/CommitterOperatorTest.java | 12 +-
.../flink/table/store/file/AbstractFileStore.java | 4 +-
.../apache/flink/table/store/file/FileStore.java | 2 +-
.../table/store/table/AbstractFileStoreTable.java | 4 +-
.../store/table/AppendOnlyFileStoreTable.java | 5 +-
.../table/ChangelogValueCountFileStoreTable.java | 5 +-
.../table/ChangelogWithKeyFileStoreTable.java | 5 +-
.../flink/table/store/table/FileStoreTable.java | 4 +-
.../table/store/table/sink/AbstractTableWrite.java | 106 +++++++++---
.../table/store/table/sink/MemoryTableWrite.java | 16 +-
.../flink/table/store/table/sink/TableWrite.java | 2 +-
.../file/utils/FailingAtomicRenameFileSystem.java | 14 ++
.../store/table/AppendOnlyFileStoreTableTest.java | 16 +-
.../ChangelogValueCountFileStoreTableTest.java | 10 +-
.../table/ChangelogWithKeyFileStoreTableTest.java | 71 +++++---
.../table/store/table/FileStoreTableTestBase.java | 43 ++---
.../table/store/table/SchemaEvolutionTest.java | 12 +-
.../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 +-
27 files changed, 603 insertions(+), 180 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/StoreSink.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/StoreSink.java
index c81faeb7..58a62270 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 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 89697b07..b69b8bf1 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,15 +97,28 @@ 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).withOverwrite(overwritePartition != null);
+ }
}
@Override
public void open() throws Exception {
super.open();
- this.write = table.newWrite().withOverwrite(overwritePartition != null);
this.sinkContext = new SimpleContext(getProcessingTimeService());
if (logSinkFunction != null) {
FunctionUtils.openFunction(logSinkFunction, new Configuration());
@@ -184,12 +206,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-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 a8a64c03..d012bf03 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
@@ -94,10 +94,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/FileStore.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStore.java
index 97f390f7..9ab37490 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
@@ -45,7 +45,7 @@ public interface FileStore<T> extends Serializable {
FileStoreWrite<T> newWrite();
- 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/table/AbstractFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
index 61ee3e50..ed650db0 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
@@ -56,8 +56,8 @@ 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());
}
@Override
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 65975cc6..45cbf759 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
@@ -105,10 +105,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 AbstractTableWrite<RowData>(store.newWrite(), recordConverter) {
+ return new AbstractTableWrite<RowData>(
+ commitUser, snapshotManager(), store.newWrite(), recordConverter) {
@Override
protected void writeSinkRecord(SinkRecord record, RecordWriter<RowData> writer)
throws Exception {
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 9dea5237..1e2d01d5 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
@@ -119,10 +119,11 @@ public class ChangelogValueCountFileStoreTable extends AbstractFileStoreTable {
}
@Override
- public TableWrite newWrite() {
+ public TableWrite newWrite(String commitUser) {
SinkRecordConverter recordConverter =
new SinkRecordConverter(store.options().bucket(), tableSchema);
- return new MemoryTableWrite<KeyValue>(store.newWrite(), recordConverter, store.options()) {
+ return new MemoryTableWrite<KeyValue>(
+ commitUser, snapshotManager(), store.newWrite(), recordConverter, store.options()) {
private final KeyValue kv = new KeyValue();
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 c7e3f0d8..16336c04 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
@@ -196,7 +196,7 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
}
@Override
- public TableWrite newWrite() {
+ public TableWrite newWrite(String commitUser) {
SinkRecordConverter recordConverter =
new SinkRecordConverter(store.options().bucket(), tableSchema);
SequenceGenerator sequenceGenerator =
@@ -204,7 +204,8 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
.sequenceField()
.map(field -> new SequenceGenerator(field, schema().logicalRowType()))
.orElse(null);
- return new MemoryTableWrite<KeyValue>(store.newWrite(), recordConverter, store.options()) {
+ return new MemoryTableWrite<KeyValue>(
+ commitUser, snapshotManager(), store.newWrite(), recordConverter, store.options()) {
private final KeyValue kv = new KeyValue();
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 afaa7508..f9113629 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
@@ -45,9 +45,9 @@ public interface FileStoreTable extends Serializable {
TableRead newRead();
- TableWrite newWrite();
+ TableWrite newWrite(String commitUser);
- TableCommit newCommit(String user);
+ TableCommit newCommit(String commitUser);
TableCompact newCompact();
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/AbstractTableWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/AbstractTableWrite.java
index 50eed358..567b4f6d 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/AbstractTableWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/AbstractTableWrite.java
@@ -18,14 +18,16 @@
package org.apache.flink.table.store.table.sink;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.store.file.Snapshot;
import org.apache.flink.table.store.file.operation.FileStoreWrite;
+import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.store.file.writer.RecordWriter;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -40,15 +42,23 @@ import java.util.concurrent.Executors;
*/
public abstract class AbstractTableWrite<T> implements TableWrite {
+ private final String commitUser;
+ private final SnapshotManager snapshotManager;
private final FileStoreWrite<T> write;
private final SinkRecordConverter recordConverter;
- 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 AbstractTableWrite(FileStoreWrite<T> write, SinkRecordConverter recordConverter) {
+ protected AbstractTableWrite(
+ String commitUser,
+ SnapshotManager snapshotManager,
+ FileStoreWrite<T> write,
+ SinkRecordConverter recordConverter) {
+ this.commitUser = commitUser;
+ this.snapshotManager = snapshotManager;
this.write = write;
this.recordConverter = recordConverter;
@@ -78,30 +88,65 @@ public abstract class AbstractTableWrite<T> implements TableWrite {
}
@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();
+ WriterWithCommit<T> writerWithCommit = entry.getValue();
+
FileCommittable committable =
- new FileCommittable(partition, bucket, writer.prepareCommit(endOfInput));
+ new FileCommittable(
+ partition,
+ bucket,
+ writerWithCommit.writer.prepareCommit(endOfInput));
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.increment().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;
}
}
@@ -115,40 +160,47 @@ public abstract class AbstractTableWrite<T> implements TableWrite {
@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();
compactExecutor.shutdownNow();
}
- @VisibleForTesting
- public Map<BinaryRowData, Map<Integer, RecordWriter<T>>> writers() {
- return writers;
- }
-
protected abstract void writeSinkRecord(SinkRecord record, RecordWriter<T> writer)
throws Exception;
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
? write.createEmptyWriter(partition.copy(), bucket, compactExecutor)
: write.createWriter(partition.copy(), bucket, compactExecutor);
notifyNewWriter(writer);
- return writer;
+ return new WriterWithCommit<>(writer);
}
protected void notifyNewWriter(RecordWriter<T> writer) {}
+
+ /** {@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/table/sink/MemoryTableWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/MemoryTableWrite.java
index 1f49b1b7..2744ee77 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/MemoryTableWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/MemoryTableWrite.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.store.file.memory.HeapMemorySegmentPool;
import org.apache.flink.table.store.file.memory.MemoryOwner;
import org.apache.flink.table.store.file.memory.MemoryPoolFactory;
import org.apache.flink.table.store.file.operation.FileStoreWrite;
+import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.store.file.writer.RecordWriter;
import org.apache.flink.shaded.guava30.com.google.common.collect.Iterators;
@@ -38,8 +39,12 @@ public abstract class MemoryTableWrite<T> extends AbstractTableWrite<T> {
private final MemoryPoolFactory memoryPoolFactory;
protected MemoryTableWrite(
- FileStoreWrite<T> write, SinkRecordConverter recordConverter, CoreOptions options) {
- super(write, recordConverter);
+ String commitUser,
+ SnapshotManager snapshotManager,
+ FileStoreWrite<T> write,
+ SinkRecordConverter recordConverter,
+ CoreOptions options) {
+ super(commitUser, snapshotManager, write, recordConverter);
HeapMemorySegmentPool memoryPool =
new HeapMemorySegmentPool(options.writeBufferSize(), options.pageSize());
@@ -47,7 +52,7 @@ public abstract class MemoryTableWrite<T> extends AbstractTableWrite<T> {
}
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
@@ -59,7 +64,10 @@ public abstract class MemoryTableWrite<T> extends AbstractTableWrite<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/sink/TableWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableWrite.java
index 3a5615d8..b331addb 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
@@ -34,7 +34,7 @@ public interface TableWrite {
SinkRecord write(RowData rowData) 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/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 fc804e87..699912d3 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
@@ -161,9 +161,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);
@@ -188,7 +188,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());
@@ -210,23 +210,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 d1cc39ee..83b6392f 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
@@ -149,26 +149,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();
}
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 0950fd40..91e17495 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.SnapshotEnumerator;
@@ -53,14 +54,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;
@@ -171,14 +172,14 @@ 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(rowDataWithKind(RowKind.DELETE, 1, 10, 100L));
write.write(rowData(1, 10, 101L));
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;
@@ -269,13 +270,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);
@@ -286,25 +287,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();
}
@@ -314,20 +315,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;
@@ -341,6 +342,36 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
write.close();
}
+ @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 13defd93..a7844a02 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
@@ -54,6 +54,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;
@@ -104,10 +105,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
@@ -126,19 +129,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;
@@ -158,13 +161,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 =
@@ -180,20 +183,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();
@@ -208,31 +211,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).increment().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
@@ -243,7 +246,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
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 f5a2ed5d..754e6b07 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,6 +49,7 @@ 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;
@@ -61,13 +62,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 +85,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 cab9276c..1ba1e259 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 d9ed166a..f7dd8791 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,22 +876,22 @@ 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))));
- 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))));
- 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 4b43e90e..88484677 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++;
}
}