You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ju...@apache.org on 2024/08/22 03:33:46 UTC
(paimon) branch master updated: [flink] Partition Mark done should not be affected by compaction (#4027)
This is an automated email from the ASF dual-hosted git repository.
junhao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new cd45b7cfe [flink] Partition Mark done should not be affected by compaction (#4027)
cd45b7cfe is described below
commit cd45b7cfe32005eb1b23aca3512854789c8f5fb7
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu Aug 22 11:33:40 2024 +0800
[flink] Partition Mark done should not be affected by compaction (#4027)
---
.../paimon/manifest/ManifestCommittable.java | 2 +-
.../actions/SuccessFileMarkDoneAction.java | 9 +-
.../apache/paimon/partition/file/SuccessFile.java | 5 +
.../flink/sink/partition/PartitionMarkDone.java | 34 ++--
.../sink/partition/PartitionMarkDoneTrigger.java | 46 ++---
.../sink/partition/PartitionMarkDoneTest.java | 191 +++++++++++++++++++++
6 files changed, 250 insertions(+), 37 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestCommittable.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestCommittable.java
index 38a69d20a..61c4619bd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestCommittable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestCommittable.java
@@ -49,7 +49,7 @@ public class ManifestCommittable {
public ManifestCommittable(
long identifier,
- Long watermark,
+ @Nullable Long watermark,
Map<Integer, Long> logOffsets,
List<CommitMessage> commitMessages) {
this.identifier = identifier;
diff --git a/paimon-core/src/main/java/org/apache/paimon/partition/actions/SuccessFileMarkDoneAction.java b/paimon-core/src/main/java/org/apache/paimon/partition/actions/SuccessFileMarkDoneAction.java
index 7c4ec375a..21620c7f6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/partition/actions/SuccessFileMarkDoneAction.java
+++ b/paimon-core/src/main/java/org/apache/paimon/partition/actions/SuccessFileMarkDoneAction.java
@@ -46,11 +46,10 @@ public class SuccessFileMarkDoneAction implements PartitionMarkDoneAction {
Path successPath = new Path(partitionPath, SUCCESS_FILE_NAME);
long currentTime = System.currentTimeMillis();
- SuccessFile successFile = SuccessFile.safelyFromPath(fileIO, successPath);
- if (successFile == null) {
- successFile = new SuccessFile(currentTime, currentTime);
- } else {
- successFile = successFile.updateModificationTime(currentTime);
+ SuccessFile successFile = new SuccessFile(currentTime, currentTime);
+ if (fileIO.exists(successPath)) {
+ successFile =
+ SuccessFile.fromPath(fileIO, successPath).updateModificationTime(currentTime);
}
fileIO.overwriteFileUtf8(successPath, successFile.toJson());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/partition/file/SuccessFile.java b/paimon-core/src/main/java/org/apache/paimon/partition/file/SuccessFile.java
index 39a202ccc..bd298920f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/partition/file/SuccessFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/partition/file/SuccessFile.java
@@ -86,6 +86,11 @@ public class SuccessFile {
}
}
+ public static SuccessFile fromPath(FileIO fileIO, Path path) throws IOException {
+ String json = fileIO.readFileUtf8(path);
+ return SuccessFile.fromJson(json);
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDone.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDone.java
index ac5895767..39438a101 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDone.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDone.java
@@ -19,12 +19,14 @@
package org.apache.paimon.flink.sink.partition;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.CoreOptions.MergeEngine;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.actions.PartitionMarkDoneAction;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.InternalRowPartitionComputer;
import org.apache.paimon.utils.PartitionPathUtils;
@@ -49,6 +51,7 @@ public class PartitionMarkDone implements Closeable {
private final InternalRowPartitionComputer partitionComputer;
private final PartitionMarkDoneTrigger trigger;
private final List<PartitionMarkDoneAction> actions;
+ private final boolean waitCompaction;
@Nullable
public static PartitionMarkDone create(
@@ -76,7 +79,14 @@ public class PartitionMarkDone implements Closeable {
List<PartitionMarkDoneAction> actions =
PartitionMarkDoneAction.createActions(table, coreOptions);
- return new PartitionMarkDone(partitionComputer, trigger, actions);
+ // if batch read skip level 0 files, we should wait compaction to mark done
+ // otherwise, some data may not be readable, and there might be data delays
+ boolean waitCompaction =
+ !table.primaryKeys().isEmpty()
+ && (coreOptions.deletionVectorsEnabled()
+ || coreOptions.mergeEngine() == MergeEngine.FIRST_ROW);
+
+ return new PartitionMarkDone(partitionComputer, trigger, actions, waitCompaction);
}
private static boolean disablePartitionMarkDone(
@@ -91,30 +101,32 @@ public class PartitionMarkDone implements Closeable {
return true;
}
- List<String> partitionKeys = table.partitionKeys();
- if (partitionKeys.isEmpty()) {
- return true;
- }
-
- return false;
+ return table.partitionKeys().isEmpty();
}
public PartitionMarkDone(
InternalRowPartitionComputer partitionComputer,
PartitionMarkDoneTrigger trigger,
- List<PartitionMarkDoneAction> actions) {
+ List<PartitionMarkDoneAction> actions,
+ boolean waitCompaction) {
this.partitionComputer = partitionComputer;
this.trigger = trigger;
this.actions = actions;
+ this.waitCompaction = waitCompaction;
}
public void notifyCommittable(List<ManifestCommittable> committables) {
Set<BinaryRow> partitions = new HashSet<>();
boolean endInput = false;
for (ManifestCommittable committable : committables) {
- committable.fileCommittables().stream()
- .map(CommitMessage::partition)
- .forEach(partitions::add);
+ for (CommitMessage commitMessage : committable.fileCommittables()) {
+ CommitMessageImpl message = (CommitMessageImpl) commitMessage;
+ if (waitCompaction
+ || !message.indexIncrement().isEmpty()
+ || !message.newFilesIncrement().isEmpty()) {
+ partitions.add(message.partition());
+ }
+ }
if (committable.identifier() == Long.MAX_VALUE) {
endInput = true;
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTrigger.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTrigger.java
index a39b5f18a..24823a789 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTrigger.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTrigger.java
@@ -31,6 +31,8 @@ import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.api.common.typeutils.base.ListSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import javax.annotation.Nullable;
+
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -57,17 +59,19 @@ public class PartitionMarkDoneTrigger {
private final State state;
private final PartitionTimeExtractor timeExtractor;
- private long timeInterval;
- private long idleTime;
- private final boolean partitionMarkDoneWhenEndInput;
+ // can be null when markDoneWhenEndInput is true
+ @Nullable private final Long timeInterval;
+ // can be null when markDoneWhenEndInput is true
+ @Nullable private final Long idleTime;
+ private final boolean markDoneWhenEndInput;
private final Map<String, Long> pendingPartitions;
public PartitionMarkDoneTrigger(
State state,
PartitionTimeExtractor timeExtractor,
- Duration timeInterval,
- Duration idleTime,
- boolean partitionMarkDoneWhenEndInput)
+ @Nullable Duration timeInterval,
+ @Nullable Duration idleTime,
+ boolean markDoneWhenEndInput)
throws Exception {
this(
state,
@@ -75,27 +79,23 @@ public class PartitionMarkDoneTrigger {
timeInterval,
idleTime,
System.currentTimeMillis(),
- partitionMarkDoneWhenEndInput);
+ markDoneWhenEndInput);
}
public PartitionMarkDoneTrigger(
State state,
PartitionTimeExtractor timeExtractor,
- Duration timeInterval,
- Duration idleTime,
+ @Nullable Duration timeInterval,
+ @Nullable Duration idleTime,
long currentTimeMillis,
- boolean partitionMarkDoneWhenEndInput)
+ boolean markDoneWhenEndInput)
throws Exception {
this.pendingPartitions = new HashMap<>();
this.state = state;
this.timeExtractor = timeExtractor;
- if (timeInterval != null) {
- this.timeInterval = timeInterval.toMillis();
- }
- if (idleTime != null) {
- this.idleTime = idleTime.toMillis();
- }
- this.partitionMarkDoneWhenEndInput = partitionMarkDoneWhenEndInput;
+ this.timeInterval = timeInterval == null ? null : timeInterval.toMillis();
+ this.idleTime = idleTime == null ? null : idleTime.toMillis();
+ this.markDoneWhenEndInput = markDoneWhenEndInput;
state.restore().forEach(p -> pendingPartitions.put(p, currentTimeMillis));
}
@@ -103,7 +103,8 @@ public class PartitionMarkDoneTrigger {
notifyPartition(partition, System.currentTimeMillis());
}
- public void notifyPartition(String partition, long currentTimeMillis) {
+ @VisibleForTesting
+ void notifyPartition(String partition, long currentTimeMillis) {
if (!StringUtils.isNullOrWhitespaceOnly(partition)) {
this.pendingPartitions.put(partition, currentTimeMillis);
}
@@ -113,11 +114,16 @@ public class PartitionMarkDoneTrigger {
return donePartitions(endInput, System.currentTimeMillis());
}
- public List<String> donePartitions(boolean endInput, long currentTimeMillis) {
- if (endInput && partitionMarkDoneWhenEndInput) {
+ @VisibleForTesting
+ List<String> donePartitions(boolean endInput, long currentTimeMillis) {
+ if (endInput && markDoneWhenEndInput) {
return new ArrayList<>(pendingPartitions.keySet());
}
+ if (timeInterval == null || idleTime == null) {
+ return Collections.emptyList();
+ }
+
List<String> needDone = new ArrayList<>();
Iterator<Map.Entry<String, Long>> iter = pendingPartitions.entrySet().iterator();
while (iter.hasNext()) {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTest.java
new file mode 100644
index 000000000..6dc14c9d0
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/PartitionMarkDoneTest.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.partition;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileTestUtils;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.io.IndexIncrement;
+import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.api.common.state.BroadcastState;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.OperatorStateStore;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nonnull;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
+import static org.apache.paimon.CoreOptions.PARTITION_MARK_DONE_ACTION;
+import static org.apache.paimon.flink.FlinkConnectorOptions.PARTITION_MARK_DONE_WHEN_END_INPUT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class PartitionMarkDoneTest extends TableTestBase {
+
+ @Test
+ public void testTriggerByCompaction() throws Exception {
+ innerTest(true);
+ }
+
+ @Test
+ public void testNotTriggerByCompaction() throws Exception {
+ innerTest(false);
+ }
+
+ private void innerTest(boolean deletionVectors) throws Exception {
+ Identifier identifier = identifier("T");
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.INT())
+ .column("c", DataTypes.INT())
+ .partitionKeys("a")
+ .primaryKey("a", "b")
+ .option(PARTITION_MARK_DONE_WHEN_END_INPUT.key(), "true")
+ .option(PARTITION_MARK_DONE_ACTION.key(), "success-file")
+ .option(
+ DELETION_VECTORS_ENABLED.key(),
+ Boolean.valueOf(deletionVectors).toString())
+ .build();
+ catalog.createTable(identifier, schema, true);
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+ Path location = catalog.getTableLocation(identifier);
+ Path successFile = new Path(location, "a=0/_SUCCESS");
+ PartitionMarkDone markDone =
+ PartitionMarkDone.create(false, false, new MockOperatorStateStore(), table);
+
+ notifyCommits(markDone, true);
+ assertThat(table.fileIO().exists(successFile)).isEqualTo(deletionVectors);
+
+ if (!deletionVectors) {
+ notifyCommits(markDone, false);
+ assertThat(table.fileIO().exists(successFile)).isEqualTo(true);
+ }
+ }
+
+ private void notifyCommits(PartitionMarkDone markDone, boolean isCompact) {
+ ManifestCommittable committable = new ManifestCommittable(Long.MAX_VALUE);
+ DataFileMeta file = DataFileTestUtils.newFile();
+ CommitMessageImpl compactMessage;
+ if (isCompact) {
+ compactMessage =
+ new CommitMessageImpl(
+ BinaryRow.singleColumn(0),
+ 0,
+ new DataIncrement(emptyList(), emptyList(), emptyList()),
+ new CompactIncrement(singletonList(file), emptyList(), emptyList()),
+ new IndexIncrement(emptyList()));
+ } else {
+ compactMessage =
+ new CommitMessageImpl(
+ BinaryRow.singleColumn(0),
+ 0,
+ new DataIncrement(singletonList(file), emptyList(), emptyList()),
+ new CompactIncrement(emptyList(), emptyList(), emptyList()),
+ new IndexIncrement(emptyList()));
+ }
+ committable.addFileCommittable(compactMessage);
+ markDone.notifyCommittable(singletonList(committable));
+ }
+
+ private static class MockOperatorStateStore implements OperatorStateStore {
+
+ @Override
+ public <K, V> BroadcastState<K, V> getBroadcastState(
+ MapStateDescriptor<K, V> stateDescriptor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <S> ListState<S> getListState(ListStateDescriptor<S> stateDescriptor) {
+ return new MockListState<>();
+ }
+
+ @Override
+ public <S> ListState<S> getUnionListState(ListStateDescriptor<S> stateDescriptor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Set<String> getRegisteredStateNames() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Set<String> getRegisteredBroadcastStateNames() {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ private static class MockListState<T> implements ListState<T> {
+
+ private final List<T> backingList = new ArrayList<>();
+
+ public MockListState() {}
+
+ @Override
+ public void update(List<T> values) {
+ this.backingList.clear();
+ this.addAll(values);
+ }
+
+ @Override
+ public void addAll(List<T> values) {
+ this.backingList.addAll(values);
+ }
+
+ @Override
+ public Iterable<T> get() {
+ return new Iterable<T>() {
+ @Nonnull
+ public Iterator<T> iterator() {
+ return MockListState.this.backingList.iterator();
+ }
+ };
+ }
+
+ @Override
+ public void add(T value) {
+ this.backingList.add(value);
+ }
+
+ @Override
+ public void clear() {
+ this.backingList.clear();
+ }
+ }
+}