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/30 07:02:45 UTC
[flink-table-store] branch master updated: [FLINK-30206] Allow FileStoreScan to read incremental changes from OVERWRITE snapshot in Table Store
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 c7bc1fb9 [FLINK-30206] Allow FileStoreScan to read incremental changes from OVERWRITE snapshot in Table Store
c7bc1fb9 is described below
commit c7bc1fb9d1d656dd93fafe66dbef185b560a33c2
Author: tsreaper <ts...@gmail.com>
AuthorDate: Wed Nov 30 15:02:40 2022 +0800
[FLINK-30206] Allow FileStoreScan to read incremental changes from OVERWRITE snapshot in Table Store
This closes #406.
---
.../source/FileStoreSourceSplitGeneratorTest.java | 4 ++-
.../table/store/file/manifest/ManifestEntry.java | 38 ++++++++++++++++------
.../store/file/manifest/ManifestFileMeta.java | 34 +------------------
.../file/operation/AbstractFileStoreScan.java | 14 ++++----
.../store/file/operation/FileStoreCommitImpl.java | 3 +-
.../table/store/file/operation/FileStoreScan.java | 13 +++++---
.../table/store/table/source/DataTableScan.java | 4 ++-
.../table/ChangelogWithKeyFileStoreTableTest.java | 32 ++++++++++++++++++
8 files changed, 84 insertions(+), 58 deletions(-)
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGeneratorTest.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGeneratorTest.java
index a55a9183..9e30c242 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGeneratorTest.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGeneratorTest.java
@@ -77,7 +77,9 @@ public class FileStoreSourceSplitGeneratorTest {
};
List<DataSplit> scanSplits =
DataTableScan.generateSplits(
- false, Collections::singletonList, plan.groupByPartFiles());
+ false,
+ Collections::singletonList,
+ plan.groupByPartFiles(plan.files(FileKind.ADD)));
DataTableScan.DataFilePlan tableScanPlan = new DataTableScan.DataFilePlan(1L, scanSplits);
List<FileStoreSourceSplit> splits =
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestEntry.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestEntry.java
index 6a01170d..80a2d5c4 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestEntry.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestEntry.java
@@ -30,6 +30,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import static org.apache.flink.table.store.file.utils.SerializationUtils.newBytesType;
@@ -114,33 +115,50 @@ public class ManifestEntry {
return String.format("{%s, %s, %d, %d, %s}", kind, partition, bucket, totalBuckets, file);
}
- public static Collection<ManifestEntry> mergeManifestEntries(List<ManifestEntry> entries) {
+ public static Collection<ManifestEntry> mergeEntries(List<ManifestEntry> entries) {
LinkedHashMap<Identifier, ManifestEntry> map = new LinkedHashMap<>();
+ mergeEntries(entries, map);
+ return map.values();
+ }
+
+ public static void mergeEntries(
+ List<ManifestEntry> entries, Map<Identifier, ManifestEntry> map) {
for (ManifestEntry entry : entries) {
ManifestEntry.Identifier identifier = entry.identifier();
switch (entry.kind()) {
case ADD:
Preconditions.checkState(
!map.containsKey(identifier),
- "Trying to add file %s which is already added. "
- + "Manifest might be corrupted.",
+ "Trying to add file %s which is already added. Manifest might be corrupted.",
identifier);
map.put(identifier, entry);
break;
case DELETE:
- Preconditions.checkState(
- map.containsKey(identifier),
- "Trying to delete file %s which is not previously added. "
- + "Manifest might be corrupted.",
- identifier);
- map.remove(identifier);
+ // each dataFile will only be added once and deleted once,
+ // if we know that it is added before then both add and delete entry can be
+ // removed because there won't be further operations on this file,
+ // otherwise we have to keep the delete entry because the add entry must be
+ // in the previous manifest files
+ if (map.containsKey(identifier)) {
+ map.remove(identifier);
+ } else {
+ map.put(identifier, entry);
+ }
break;
default:
throw new UnsupportedOperationException(
"Unknown value kind " + entry.kind().name());
}
}
- return map.values();
+ }
+
+ public static void assertNoDelete(Collection<ManifestEntry> entries) {
+ for (ManifestEntry entry : entries) {
+ Preconditions.checkState(
+ entry.kind() != FileKind.DELETE,
+ "Trying to delete file %s which is not previously added. Manifest might be corrupted.",
+ entry.file().fileName());
+ }
}
/**
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestFileMeta.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestFileMeta.java
index e80da5f7..b3c74cda 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestFileMeta.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestFileMeta.java
@@ -23,7 +23,6 @@ import org.apache.flink.table.store.file.stats.FieldStatsArraySerializer;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.VarCharType;
-import org.apache.flink.util.Preconditions;
import java.util.ArrayList;
import java.util.LinkedHashMap;
@@ -177,7 +176,7 @@ public class ManifestFileMeta {
Map<ManifestEntry.Identifier, ManifestEntry> map = new LinkedHashMap<>();
for (ManifestFileMeta manifest : candidates) {
- mergeEntries(manifestFile.read(manifest.fileName), map);
+ ManifestEntry.mergeEntries(manifestFile.read(manifest.fileName), map);
}
if (!map.isEmpty()) {
List<ManifestFileMeta> merged = manifestFile.write(new ArrayList<>(map.values()));
@@ -185,35 +184,4 @@ public class ManifestFileMeta {
newMetas.addAll(merged);
}
}
-
- private static void mergeEntries(
- List<ManifestEntry> entries, Map<ManifestEntry.Identifier, ManifestEntry> map) {
- for (ManifestEntry entry : entries) {
- ManifestEntry.Identifier identifier = entry.identifier();
- switch (entry.kind()) {
- case ADD:
- Preconditions.checkState(
- !map.containsKey(identifier),
- "Trying to add file %s which is already added. Manifest might be corrupted.",
- identifier);
- map.put(identifier, entry);
- break;
- case DELETE:
- // each dataFile will only be added once and deleted once,
- // if we know that it is added before then both add and delete entry can be
- // removed because there won't be further operations on this file,
- // otherwise we have to keep the delete entry because the add entry must be
- // in the previous manifest files
- if (map.containsKey(identifier)) {
- map.remove(identifier);
- } else {
- map.put(identifier, entry);
- }
- break;
- default:
- throw new UnsupportedOperationException(
- "Unknown value kind " + entry.kind().name());
- }
- }
- }
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java
index 70d114a0..b98e7211 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/AbstractFileStoreScan.java
@@ -221,7 +221,7 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
}
List<ManifestEntry> files = new ArrayList<>();
- for (ManifestEntry file : ManifestEntry.mergeManifestEntries(entries)) {
+ for (ManifestEntry file : ManifestEntry.mergeEntries(entries)) {
if (checkNumOfBuckets && file.totalBuckets() != numOfBuckets) {
String partInfo =
partitionConverter.getArity() > 0
@@ -290,13 +290,13 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
return manifestList.read(snapshot.changelogManifestList());
}
case NONE:
- if (snapshot.commitKind() == Snapshot.CommitKind.APPEND) {
- return manifestList.read(snapshot.deltaManifestList());
+ if (snapshot.commitKind() == Snapshot.CommitKind.COMPACT) {
+ throw new IllegalStateException(
+ String.format(
+ "Incremental scan does not accept %s snapshot",
+ snapshot.commitKind()));
}
- throw new IllegalStateException(
- String.format(
- "Incremental scan does not accept %s snapshot",
- snapshot.commitKind()));
+ return manifestList.read(snapshot.deltaManifestList());
default:
throw new UnsupportedOperationException(
"Unknown changelog producer " + changelogProducer.name());
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
index 961d82c2..33a3d48a 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
@@ -626,7 +626,8 @@ public class FileStoreCommitImpl implements FileStoreCommit {
Collection<ManifestEntry> mergedEntries;
try {
// merge manifest entries and also check if the files we want to delete are still there
- mergedEntries = ManifestEntry.mergeManifestEntries(allEntries);
+ mergedEntries = ManifestEntry.mergeEntries(allEntries);
+ ManifestEntry.assertNoDelete(mergedEntries);
} catch (Throwable e) {
LOG.warn("File deletion conflicts detected! Give up committing.", e);
throw createConflictException(
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScan.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScan.java
index df3289a0..0ce733b8 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScan.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScan.java
@@ -31,8 +31,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-
-import static org.apache.flink.util.Preconditions.checkArgument;
+import java.util.stream.Collectors;
/** Scan operation which produces a plan. */
public interface FileStoreScan {
@@ -67,12 +66,16 @@ public interface FileStoreScan {
/** Result {@link ManifestEntry} files. */
List<ManifestEntry> files();
+ /** Result {@link ManifestEntry} files with specific file kind. */
+ default List<ManifestEntry> files(FileKind kind) {
+ return files().stream().filter(e -> e.kind() == kind).collect(Collectors.toList());
+ }
+
/** Return a map group by partition and bucket. */
- default Map<BinaryRowData, Map<Integer, List<DataFileMeta>>> groupByPartFiles() {
- List<ManifestEntry> files = files();
+ default Map<BinaryRowData, Map<Integer, List<DataFileMeta>>> groupByPartFiles(
+ List<ManifestEntry> files) {
Map<BinaryRowData, Map<Integer, List<DataFileMeta>>> groupBy = new HashMap<>();
for (ManifestEntry entry : files) {
- checkArgument(entry.kind() == FileKind.ADD);
groupBy.computeIfAbsent(entry.partition(), k -> new HashMap<>())
.computeIfAbsent(entry.bucket(), k -> new ArrayList<>())
.add(entry.file());
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/DataTableScan.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/DataTableScan.java
index e155ec1f..2b8e288b 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/DataTableScan.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/DataTableScan.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.store.CoreOptions;
import org.apache.flink.table.store.file.io.DataFileMeta;
+import org.apache.flink.table.store.file.manifest.FileKind;
import org.apache.flink.table.store.file.operation.FileStoreScan;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.predicate.PredicateBuilder;
@@ -112,7 +113,8 @@ public abstract class DataTableScan implements TableScan {
@Override
public DataFilePlan plan() {
FileStoreScan.Plan plan = scan.plan();
- return new DataFilePlan(plan.snapshotId(), generateSplits(plan.groupByPartFiles()));
+ return new DataFilePlan(
+ plan.snapshotId(), generateSplits(plan.groupByPartFiles(plan.files(FileKind.ADD))));
}
private List<DataSplit> generateSplits(
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 5178181e..7dd1f502 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
@@ -33,6 +33,7 @@ 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.DataSplit;
import org.apache.flink.table.store.table.source.DataTableScan;
import org.apache.flink.table.store.table.source.SnapshotEnumerator;
import org.apache.flink.table.store.table.source.Split;
@@ -47,7 +48,9 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -578,6 +581,35 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
"2|10|300|binary|varbinary|mapKey:mapVal|multiset"));
}
+ @Test
+ public void testIncrementalScanOverwrite() 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));
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ DataTableScan scan = table.newScan().withIncremental(true);
+ List<DataSplit> splits0 = scan.plan().splits;
+ assertThat(splits0).hasSize(1);
+ assertThat(splits0.get(0).files()).hasSize(1);
+
+ write.write(rowData(1, 10, 1000L));
+ write.write(rowData(1, 20, 2000L));
+ Map<String, String> overwritePartition = new HashMap<>();
+ overwritePartition.put("pt", "1");
+ commit.withOverwritePartition(overwritePartition);
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ List<DataSplit> splits1 = scan.plan().splits;
+ assertThat(splits1).hasSize(1);
+ assertThat(splits1.get(0).files()).hasSize(1);
+ assertThat(splits1.get(0).files().get(0).fileName())
+ .isNotEqualTo(splits0.get(0).files().get(0).fileName());
+ }
+
@Override
protected FileStoreTable createFileStoreTable(Consumer<Configuration> configure)
throws Exception {