You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2022/05/18 07:34:58 UTC
[flink-table-store] branch master updated: [FLINK-27366] Record schema on filesystem path
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 1aac9a1 [FLINK-27366] Record schema on filesystem path
1aac9a1 is described below
commit 1aac9a199d1f4b0e34e8cc2a2b6aca7b0eb72e99
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Wed May 18 15:34:53 2022 +0800
[FLINK-27366] Record schema on filesystem path
This closes #101
---
.../flink/table/store/connector/TableStore.java | 154 +++++++--------
.../table/store/connector/TableStoreFactory.java | 80 ++++++--
.../table/store/connector/FileStoreITCase.java | 70 ++++---
.../store/connector/ReadWriteTableITCase.java | 10 +-
.../store/connector/TableStoreFactoryTest.java | 15 +-
.../store/connector/sink/LogStoreSinkITCase.java | 21 +--
.../connector/source/FileStoreSourceTest.java | 1 +
.../flink/table/store/file/FileStoreImpl.java | 17 +-
.../flink/table/store/file/FileStoreOptions.java | 4 +
.../apache/flink/table/store/file/Snapshot.java | 26 +--
.../store/file/operation/FileStoreCommitImpl.java | 4 +
.../table/store/file/schema/ArrayDataType.java | 60 ++++++
.../table/store/file/schema/AtomicDataType.java | 31 +++
.../flink/table/store/file/schema/DataField.java | 93 +++++++++
.../store/file/schema/DataFieldSerializer.java | 128 +++++++++++++
.../flink/table/store/file/schema/DataType.java | 64 +++++++
.../flink/table/store/file/schema/MapDataType.java | 67 +++++++
.../table/store/file/schema/MultisetDataType.java | 60 ++++++
.../flink/table/store/file/schema/RowDataType.java | 72 +++++++
.../flink/table/store/file/schema/Schema.java | 210 +++++++++++++++++++++
.../table/store/file/schema/SchemaManager.java | 160 ++++++++++++++++
.../table/store/file/schema/SchemaSerializer.java | 107 +++++++++++
.../table/store/file/schema/UpdateSchema.java | 77 ++++++++
.../flink/table/store/file/utils/FileUtils.java | 28 +++
.../table/store/file/utils/JsonSerdeUtil.java | 101 ++++++++++
.../table/store/file/utils/JsonSerializer.java | 32 ++++
.../table/store/file/utils/SnapshotFinder.java | 24 +--
.../flink/table/store/file/TestFileStore.java | 1 +
.../table/store/file/schema/SchemaManagerTest.java | 175 +++++++++++++++++
.../store/file/schema/SchemaSerializationTest.java | 59 ++++++
.../flink/table/store/file/schema/SchemaTest.java | 110 +++++++++++
.../file/utils/FailingAtomicRenameFileSystem.java | 14 ++
.../table/store/mapred/TableStoreInputFormat.java | 8 +-
.../flink/table/store/FileStoreTestHelper.java | 1 +
34 files changed, 1900 insertions(+), 184 deletions(-)
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStore.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStore.java
index 6685bf3..ad0a92c 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStore.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStore.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.connector.source.Source;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DelegatingConfiguration;
import org.apache.flink.connector.base.source.hybrid.HybridSource;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
@@ -48,22 +49,21 @@ import org.apache.flink.table.store.file.FileStoreImpl;
import org.apache.flink.table.store.file.FileStoreOptions;
import org.apache.flink.table.store.file.FileStoreOptions.MergeEngine;
import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.schema.Schema;
+import org.apache.flink.table.store.file.schema.SchemaManager;
import org.apache.flink.table.store.log.LogOptions.LogStartupMode;
import org.apache.flink.table.store.log.LogSinkProvider;
import org.apache.flink.table.store.log.LogSourceProvider;
import org.apache.flink.table.store.utils.TypeUtils;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
-import java.util.stream.Collectors;
import static org.apache.flink.table.store.file.FileStoreOptions.BUCKET;
import static org.apache.flink.table.store.file.FileStoreOptions.CONTINUOUS_DISCOVERY_INTERVAL;
@@ -76,26 +76,32 @@ import static org.apache.flink.table.store.log.LogOptions.SCAN;
@Experimental
public class TableStore {
- private final Configuration options;
-
- /** commit user, default uuid. */
- private String user = UUID.randomUUID().toString();
-
- /** partition keys, default no partition. */
- private int[] partitions = new int[0];
+ private final ObjectIdentifier tableIdentifier;
- /** file store primary keys which exclude partition fields if partitioned, default no key. */
- private int[] primaryKeys = new int[0];
+ private final Configuration options;
- /** log store primary keys which include partition fields if partitioned, default no key. */
- private int[] logPrimaryKeys = new int[0];
+ private final Schema schema;
- private RowType type;
+ private final RowType type;
- private ObjectIdentifier tableIdentifier;
+ /** commit user, default uuid. */
+ private String user = UUID.randomUUID().toString();
- public TableStore(Configuration options) {
+ public TableStore(ObjectIdentifier tableIdentifier, Configuration options) {
+ this.tableIdentifier = tableIdentifier;
this.options = options;
+
+ Path tablePath = new FileStoreOptions(options).path(tableIdentifier);
+ this.schema =
+ new SchemaManager(tablePath)
+ .latest()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Can not find schema in path %s, please create table first.",
+ tablePath)));
+ this.type = schema.logicalRowType();
}
public TableStore withUser(String user) {
@@ -103,35 +109,16 @@ public class TableStore {
return this;
}
- public TableStore withSchema(RowType type) {
- this.type = type;
- return this;
- }
-
- public TableStore withPartitions(int[] partitions) {
- this.partitions = partitions;
- adjustIndexAndValidate();
- return this;
- }
-
- public TableStore withPrimaryKeys(int[] primaryKeys) {
- this.primaryKeys = primaryKeys;
- this.logPrimaryKeys = primaryKeys;
- adjustIndexAndValidate();
- return this;
- }
-
- public TableStore withTableIdentifier(ObjectIdentifier tableIdentifier) {
- this.tableIdentifier = tableIdentifier;
- return this;
+ public RowType type() {
+ return type;
}
public boolean partitioned() {
- return partitions.length > 0;
+ return schema.partitionKeys().size() > 0;
}
public boolean valueCountMode() {
- return primaryKeys.length == 0;
+ return trimmedPrimaryKeys().size() == 0;
}
public List<String> fieldNames() {
@@ -139,14 +126,33 @@ public class TableStore {
}
public List<String> partitionKeys() {
- RowType partitionType = TypeUtils.project(type, partitions);
- return partitionType.getFieldNames();
+ return schema.partitionKeys();
}
@VisibleForTesting
- List<String> primaryKeys() {
- RowType primaryKeyType = TypeUtils.project(type, primaryKeys);
- return primaryKeyType.getFieldNames();
+ List<String> trimmedPrimaryKeys() {
+ return schema.trimmedPrimaryKeys();
+ }
+
+ private int[] toIndex(List<String> fields) {
+ List<String> fieldNames = type.getFieldNames();
+ return fields.stream().mapToInt(fieldNames::indexOf).toArray();
+ }
+
+ private int[] partitionKeysIndex() {
+ return toIndex(schema.partitionKeys());
+ }
+
+ private int[] fullPrimaryKeysIndex() {
+ return toIndex(schema.primaryKeys());
+ }
+
+ private int[] trimmedPrimaryKeysIndex() {
+ return toIndex(trimmedPrimaryKeys());
+ }
+
+ public Schema schema() {
+ return schema;
}
public Configuration options() {
@@ -170,11 +176,13 @@ public class TableStore {
}
private FileStore buildFileStore() {
- RowType partitionType = TypeUtils.project(type, partitions);
+ RowType partitionType = TypeUtils.project(type, partitionKeysIndex());
FileStoreOptions fileStoreOptions = new FileStoreOptions(options);
- if (primaryKeys.length == 0) {
+ int[] trimmedPrimaryKeys = trimmedPrimaryKeysIndex();
+ if (trimmedPrimaryKeys.length == 0) {
return FileStoreImpl.createWithValueCount(
fileStoreOptions.path(tableIdentifier).toString(),
+ schema.id(),
fileStoreOptions,
user,
partitionType,
@@ -182,48 +190,16 @@ public class TableStore {
} else {
return FileStoreImpl.createWithPrimaryKey(
fileStoreOptions.path(tableIdentifier).toString(),
+ schema.id(),
fileStoreOptions,
user,
partitionType,
- TypeUtils.project(type, primaryKeys),
+ TypeUtils.project(type, trimmedPrimaryKeys),
type,
mergeEngine());
}
}
- private void adjustIndexAndValidate() {
- if (logPrimaryKeys.length > 0 && partitions.length > 0) {
- List<Integer> pkList =
- Arrays.stream(logPrimaryKeys).boxed().collect(Collectors.toList());
- List<Integer> partitionList =
- Arrays.stream(partitions).boxed().collect(Collectors.toList());
-
- String pkInfo =
- type == null
- ? pkList.toString()
- : TypeUtils.project(type, logPrimaryKeys).getFieldNames().toString();
- String partitionInfo =
- type == null
- ? partitionList.toString()
- : TypeUtils.project(type, partitions).getFieldNames().toString();
- Preconditions.checkState(
- pkList.containsAll(partitionList),
- String.format(
- "Primary key constraint %s should include all partition fields %s",
- pkInfo, partitionInfo));
- primaryKeys =
- Arrays.stream(logPrimaryKeys)
- .filter(pk -> !partitionList.contains(pk))
- .toArray();
-
- Preconditions.checkState(
- primaryKeys.length > 0,
- String.format(
- "Primary key constraint %s should not be same with partition fields %s, this will result in only one record in a partition",
- pkInfo, partitionInfo));
- }
- }
-
/** Source builder to build a flink {@link Source}. */
public class SourceBuilder {
@@ -284,7 +260,7 @@ public class TableStore {
boolean isContinuous, boolean continuousScanLatest) {
return new FileStoreSource(
buildFileStore(),
- primaryKeys.length == 0,
+ schema.primaryKeys().isEmpty(),
isContinuous,
discoveryIntervalMills(),
continuousScanLatest,
@@ -296,7 +272,7 @@ public class TableStore {
private Source<RowData, ?, ?> buildSource() {
if (isContinuous) {
- if (primaryKeys.length > 0 && mergeEngine() == PARTIAL_UPDATE) {
+ if (schema.primaryKeys().size() > 0 && mergeEngine() == PARTIAL_UPDATE) {
throw new ValidationException(
"Partial update continuous reading is not supported.");
}
@@ -388,7 +364,11 @@ public class TableStore {
BucketStreamPartitioner partitioner =
new BucketStreamPartitioner(
- numBucket, type, partitions, primaryKeys, logPrimaryKeys);
+ numBucket,
+ type,
+ partitionKeysIndex(),
+ trimmedPrimaryKeysIndex(),
+ fullPrimaryKeysIndex());
PartitionTransformation<RowData> partitioned =
new PartitionTransformation<>(input.getTransformation(), partitioner);
if (parallelism != null) {
@@ -399,9 +379,9 @@ public class TableStore {
new StoreSink<>(
tableIdentifier,
fileStore,
- partitions,
- primaryKeys,
- logPrimaryKeys,
+ partitionKeysIndex(),
+ trimmedPrimaryKeysIndex(),
+ fullPrimaryKeysIndex(),
numBucket,
lockFactory,
overwritePartition,
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStoreFactory.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStoreFactory.java
index 78cba33..a72d264 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStoreFactory.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStoreFactory.java
@@ -39,14 +39,19 @@ import org.apache.flink.table.store.connector.source.TableStoreSource;
import org.apache.flink.table.store.connector.utils.TableConfigUtils;
import org.apache.flink.table.store.file.FileStoreOptions;
import org.apache.flink.table.store.file.mergetree.MergeTreeOptions;
+import org.apache.flink.table.store.file.schema.Schema;
+import org.apache.flink.table.store.file.schema.SchemaManager;
+import org.apache.flink.table.store.file.schema.UpdateSchema;
import org.apache.flink.table.store.log.LogOptions.LogChangelogMode;
import org.apache.flink.table.store.log.LogOptions.LogConsistency;
import org.apache.flink.table.store.log.LogOptions.LogStartupMode;
import org.apache.flink.table.store.log.LogStoreTableFactory;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
import java.io.IOException;
import java.io.UncheckedIOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -108,6 +113,19 @@ public class TableStoreFactory
throw new UncheckedIOException(e);
}
+ // update schema
+ Path tablePath =
+ new FileStoreOptions(context.getCatalogTable().getOptions())
+ .path(context.getObjectIdentifier());
+ // TODO pass lock
+ try {
+ new SchemaManager(tablePath).commitNewVersion(newUpdateSchema(context));
+ } catch (IllegalStateException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
createOptionalLogStoreFactory(context)
.ifPresent(
factory ->
@@ -239,26 +257,56 @@ public class TableStoreFactory
Map.Entry::getValue));
}
- @VisibleForTesting
- TableStore buildTableStore(Context context) {
+ private UpdateSchema newUpdateSchema(Context context) {
ResolvedCatalogTable catalogTable = context.getCatalogTable();
ResolvedSchema schema = catalogTable.getResolvedSchema();
RowType rowType = (RowType) schema.toPhysicalRowDataType().getLogicalType();
- List<String> partitionKeys = catalogTable.getPartitionKeys();
- int[] pkIndex = new int[0];
+ List<String> primaryKeys = new ArrayList<>();
if (schema.getPrimaryKey().isPresent()) {
- pkIndex =
- schema.getPrimaryKey().get().getColumns().stream()
- .mapToInt(rowType.getFieldNames()::indexOf)
- .toArray();
+ primaryKeys = schema.getPrimaryKey().get().getColumns();
}
- return new TableStore(Configuration.fromMap(catalogTable.getOptions()))
- .withTableIdentifier(context.getObjectIdentifier())
- .withSchema(rowType)
- .withPrimaryKeys(pkIndex)
- .withPartitions(
- partitionKeys.stream()
- .mapToInt(rowType.getFieldNames()::indexOf)
- .toArray());
+
+ return new UpdateSchema(
+ rowType, catalogTable.getPartitionKeys(), primaryKeys, catalogTable.getOptions());
+ }
+
+ @VisibleForTesting
+ TableStore buildTableStore(Context context) {
+ TableStore store =
+ new TableStore(
+ context.getObjectIdentifier(),
+ Configuration.fromMap(context.getCatalogTable().getOptions()));
+
+ Schema schema = store.schema();
+
+ UpdateSchema updateSchema = newUpdateSchema(context);
+
+ RowType rowType = updateSchema.rowType();
+ List<String> partitionKeys = updateSchema.partitionKeys();
+ List<String> primaryKeys = updateSchema.primaryKeys();
+
+ // compare fields to ignore isNullable for row type
+ Preconditions.checkArgument(
+ schema.logicalRowType().getFields().equals(rowType.getFields()),
+ "Flink schema and store schema are not the same, "
+ + "store schema is %s, Flink schema is %s",
+ schema.logicalRowType(),
+ rowType);
+
+ Preconditions.checkArgument(
+ schema.partitionKeys().equals(partitionKeys),
+ "Flink partitionKeys and store partitionKeys are not the same, "
+ + "store partitionKeys is %s, Flink partitionKeys is %s",
+ schema.partitionKeys(),
+ partitionKeys);
+
+ Preconditions.checkArgument(
+ schema.primaryKeys().equals(primaryKeys),
+ "Flink primaryKeys and store primaryKeys are not the same, "
+ + "store primaryKeys is %s, Flink primaryKeys is %s",
+ schema.primaryKeys(),
+ primaryKeys);
+
+ return store;
}
}
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FileStoreITCase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FileStoreITCase.java
index c6b7d96..601c8ba 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FileStoreITCase.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FileStoreITCase.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.store.connector;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.catalog.ObjectIdentifier;
@@ -33,6 +34,9 @@ import org.apache.flink.table.runtime.typeutils.InternalSerializers;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.store.connector.sink.StoreSink;
import org.apache.flink.table.store.connector.source.FileStoreSource;
+import org.apache.flink.table.store.file.FileStoreOptions;
+import org.apache.flink.table.store.file.schema.SchemaManager;
+import org.apache.flink.table.store.file.schema.UpdateSchema;
import org.apache.flink.table.store.file.utils.BlockingIterator;
import org.apache.flink.table.store.file.utils.FailingAtomicRenameFileSystem;
import org.apache.flink.table.types.logical.IntType;
@@ -59,11 +63,13 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.apache.flink.table.store.file.FileStoreOptions.BUCKET;
import static org.apache.flink.table.store.file.FileStoreOptions.FILE_FORMAT;
import static org.apache.flink.table.store.file.FileStoreOptions.PATH;
+import static org.apache.flink.table.store.file.utils.FailingAtomicRenameFileSystem.retryArtificialException;
import static org.assertj.core.api.Assertions.assertThat;
/** ITCase for {@link FileStoreSource} and {@link StoreSink}. */
@@ -100,12 +106,9 @@ public class FileStoreITCase extends AbstractTestBase {
private final StreamExecutionEnvironment env;
- private final TableStore store;
-
public FileStoreITCase(boolean isBatch) throws IOException {
this.isBatch = isBatch;
this.env = isBatch ? buildBatchEnv() : buildStreamEnv();
- this.store = buildTableStore(isBatch, TEMPORARY_FOLDER);
}
@Parameterized.Parameters(name = "isBatch-{0}")
@@ -119,7 +122,7 @@ public class FileStoreITCase extends AbstractTestBase {
@Test
public void testPartitioned() throws Exception {
- store.withPrimaryKeys(new int[] {1, 2}).withPartitions(new int[] {1});
+ TableStore store = buildTableStore(new int[] {1}, new int[] {1, 2});
// write
store.sinkBuilder().withInput(buildTestSource(env, isBatch)).build();
@@ -138,7 +141,7 @@ public class FileStoreITCase extends AbstractTestBase {
@Test
public void testNonPartitioned() throws Exception {
- store.withPartitions(new int[0]);
+ TableStore store = buildTableStore(new int[0], new int[] {2});
// write
store.sinkBuilder().withInput(buildTestSource(env, isBatch)).build();
@@ -155,7 +158,8 @@ public class FileStoreITCase extends AbstractTestBase {
@Test
public void testOverwrite() throws Exception {
Assume.assumeTrue(isBatch);
- store.withPrimaryKeys(new int[] {1, 2}).withPartitions(new int[] {1});
+
+ TableStore store = buildTableStore(new int[] {1}, new int[] {1, 2});
// write
store.sinkBuilder().withInput(buildTestSource(env, isBatch)).build();
@@ -195,7 +199,7 @@ public class FileStoreITCase extends AbstractTestBase {
@Test
public void testPartitionedNonKey() throws Exception {
- store.withPrimaryKeys(new int[0]).withPartitions(new int[] {1});
+ TableStore store = buildTableStore(new int[] {1}, new int[0]);
// write
store.sinkBuilder().withInput(buildTestSource(env, isBatch)).build();
@@ -216,16 +220,15 @@ public class FileStoreITCase extends AbstractTestBase {
@Test
public void testKeyedProjection() throws Exception {
- testProjection();
+ testProjection(buildTableStore(new int[0], new int[] {2}));
}
@Test
public void testNonKeyedProjection() throws Exception {
- store.withPrimaryKeys(new int[0]);
- testProjection();
+ testProjection(buildTableStore(new int[0], new int[0]));
}
- private void testProjection() throws Exception {
+ private void testProjection(TableStore store) throws Exception {
// write
store.sinkBuilder().withInput(buildTestSource(env, isBatch)).build();
env.execute();
@@ -248,7 +251,7 @@ public class FileStoreITCase extends AbstractTestBase {
// assert
Row[] expected = new Row[] {Row.of("p2", 1), Row.of("p1", 2), Row.of("p2", 5)};
- if (store.primaryKeys().isEmpty()) {
+ if (store.trimmedPrimaryKeys().isEmpty()) {
// in streaming mode, expect origin data X 2 (FiniteTestSource)
Stream<RowData> expectedStream =
isBatch
@@ -265,17 +268,15 @@ public class FileStoreITCase extends AbstractTestBase {
@Test
public void testContinuous() throws Exception {
- store.withPrimaryKeys(new int[] {2});
- innerTestContinuous();
+ innerTestContinuous(buildTableStore(new int[0], new int[] {2}));
}
@Test
public void testContinuousWithoutPK() throws Exception {
- store.withPrimaryKeys(new int[0]);
- innerTestContinuous();
+ innerTestContinuous(buildTableStore(new int[0], new int[0]));
}
- private void innerTestContinuous() throws Exception {
+ private void innerTestContinuous(TableStore store) throws Exception {
Assume.assumeFalse(isBatch);
BlockingIterator<RowData, Row> iterator =
@@ -289,6 +290,7 @@ public class FileStoreITCase extends AbstractTestBase {
Thread.sleep(ThreadLocalRandom.current().nextInt(1000));
sinkAndValidate(
+ store,
Arrays.asList(
srcRow(RowKind.INSERT, 1, "p1", 1), srcRow(RowKind.INSERT, 2, "p2", 2)),
iterator,
@@ -296,6 +298,7 @@ public class FileStoreITCase extends AbstractTestBase {
Row.ofKind(RowKind.INSERT, 2, "p2", 2));
sinkAndValidate(
+ store,
Arrays.asList(
srcRow(RowKind.DELETE, 1, "p1", 1), srcRow(RowKind.INSERT, 3, "p3", 3)),
iterator,
@@ -304,7 +307,10 @@ public class FileStoreITCase extends AbstractTestBase {
}
private void sinkAndValidate(
- List<RowData> src, BlockingIterator<RowData, Row> iterator, Row... expected)
+ TableStore store,
+ List<RowData> src,
+ BlockingIterator<RowData, Row> iterator,
+ Row... expected)
throws Exception {
if (isBatch) {
throw new UnsupportedOperationException();
@@ -316,6 +322,10 @@ public class FileStoreITCase extends AbstractTestBase {
assertThat(iterator.collect(expected.length)).containsExactlyInAnyOrder(expected);
}
+ public TableStore buildTableStore(int[] partitions, int[] primaryKey) throws Exception {
+ return buildTableStore(isBatch, TEMPORARY_FOLDER, partitions, primaryKey);
+ }
+
private static RowData srcRow(RowKind kind, int v, String p, int k) {
return wrap(GenericRowData.ofKind(kind, v, StringData.fromString(p), k));
}
@@ -335,12 +345,24 @@ public class FileStoreITCase extends AbstractTestBase {
return env;
}
- public static TableStore buildTableStore(boolean noFail, TemporaryFolder temporaryFolder)
- throws IOException {
- return new TableStore(buildConfiguration(noFail, temporaryFolder.newFolder()))
- .withSchema(TABLE_TYPE)
- .withPrimaryKeys(new int[] {2})
- .withTableIdentifier(ObjectIdentifier.of("catalog", "db", "t"));
+ public static TableStore buildTableStore(
+ boolean noFail, TemporaryFolder temporaryFolder, int[] partitions, int[] primaryKey)
+ throws Exception {
+ ObjectIdentifier identifier = ObjectIdentifier.of("catalog", "db", "t");
+ Configuration options = buildConfiguration(noFail, temporaryFolder.newFolder());
+ Path tablePath = new FileStoreOptions(options).path(identifier);
+ new SchemaManager(tablePath)
+ .commitNewVersion(
+ new UpdateSchema(
+ TABLE_TYPE,
+ Arrays.stream(partitions)
+ .mapToObj(i -> TABLE_TYPE.getFieldNames().get(i))
+ .collect(Collectors.toList()),
+ Arrays.stream(primaryKey)
+ .mapToObj(i -> TABLE_TYPE.getFieldNames().get(i))
+ .collect(Collectors.toList()),
+ options.toMap()));
+ return retryArtificialException(() -> new TableStore(identifier, options));
}
public static Configuration buildConfiguration(boolean noFail, File folder) {
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ReadWriteTableITCase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ReadWriteTableITCase.java
index 451bc39..d120223 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ReadWriteTableITCase.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ReadWriteTableITCase.java
@@ -46,6 +46,7 @@ import org.junit.Test;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -1196,7 +1197,7 @@ public class ReadWriteTableITCase extends ReadWriteTableTestBase {
}
@Test
- public void testSinkParallelism() {
+ public void testSinkParallelism() throws IOException {
testSinkParallelism(null, env.getParallelism());
testSinkParallelism(23, 23);
}
@@ -1573,13 +1574,17 @@ public class ReadWriteTableITCase extends ReadWriteTableTestBase {
return stream.getParallelism();
}
- private void testSinkParallelism(Integer configParallelism, int expectedParallelism) {
+ private void testSinkParallelism(Integer configParallelism, int expectedParallelism)
+ throws IOException {
// 1. create a mock table sink
Map<String, String> options = new HashMap<>();
options.put(FileStoreOptions.PATH.key(), "/fake/path");
if (configParallelism != null) {
options.put(SINK_PARALLELISM.key(), configParallelism.toString());
}
+ options.put(
+ "path",
+ TEMPORARY_FOLDER.newFolder(UUID.randomUUID().toString()).toURI().toString());
DynamicTableFactory.Context context =
new FactoryUtil.DefaultDynamicTableContext(
@@ -1595,6 +1600,7 @@ public class ReadWriteTableITCase extends ReadWriteTableTestBase {
new Configuration(),
Thread.currentThread().getContextClassLoader(),
false);
+ new TableStoreFactory().onCreateTable(context, false);
DynamicTableSink tableSink = new TableStoreFactory().createDynamicTableSink(context);
assertThat(tableSink).isInstanceOf(TableStoreSink.class);
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/TableStoreFactoryTest.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/TableStoreFactoryTest.java
index b32fb24..021ac98 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/TableStoreFactoryTest.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/TableStoreFactoryTest.java
@@ -45,6 +45,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.Predicate;
import java.util.stream.Stream;
@@ -160,13 +161,18 @@ public class TableStoreFactoryTest {
@ParameterizedTest
@MethodSource("providingResolvedTable")
- public void testBuildTableStore(
+ public void testCreateAndCheckTableStore(
RowType rowType,
List<String> partitions,
List<String> primaryKeys,
TableStoreTestBase.ExpectedResult expectedResult) {
ResolvedCatalogTable catalogTable =
- createResolvedTable(Collections.emptyMap(), rowType, partitions, primaryKeys);
+ createResolvedTable(
+ Collections.singletonMap(
+ "path", sharedTempDir.toAbsolutePath() + "/" + UUID.randomUUID()),
+ rowType,
+ partitions,
+ primaryKeys);
context =
new FactoryUtil.DefaultDynamicTableContext(
TABLE_IDENTIFIER,
@@ -176,6 +182,7 @@ public class TableStoreFactoryTest {
Thread.currentThread().getContextClassLoader(),
false);
if (expectedResult.success) {
+ tableStoreFactory.onCreateTable(context, false);
TableStore tableStore = tableStoreFactory.buildTableStore(context);
assertThat(tableStore.partitioned()).isEqualTo(catalogTable.isPartitioned());
assertThat(tableStore.valueCountMode())
@@ -184,12 +191,12 @@ public class TableStoreFactoryTest {
// check primary key doesn't contain partition
if (tableStore.partitioned() && !tableStore.valueCountMode()) {
assertThat(
- tableStore.primaryKeys().stream()
+ tableStore.trimmedPrimaryKeys().stream()
.noneMatch(pk -> tableStore.partitionKeys().contains(pk)))
.isTrue();
}
} else {
- assertThatThrownBy(() -> tableStoreFactory.buildTableStore(context))
+ assertThatThrownBy(() -> tableStoreFactory.onCreateTable(context, false))
.isInstanceOf(expectedResult.expectedType)
.hasMessageContaining(expectedResult.expectedMessage);
}
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/LogStoreSinkITCase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/LogStoreSinkITCase.java
index 07bd058..d3a602e 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/LogStoreSinkITCase.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/LogStoreSinkITCase.java
@@ -90,21 +90,12 @@ public class LogStoreSinkITCase extends KafkaTableTestBase {
StreamExecutionEnvironment env = isBatch ? buildBatchEnv() : buildStreamEnv();
// in eventual mode, failure will result in duplicate data
- TableStore store = buildTableStore(isBatch || !transaction, TEMPORARY_FOLDER);
- if (partitioned) {
- if (hasPk) {
- store.withPrimaryKeys(new int[] {1, 2});
- } else {
- store.withPrimaryKeys(new int[0]);
- }
- store.withPartitions(new int[] {1});
- } else {
- store.withPartitions(new int[0]);
- }
-
- if (!hasPk) {
- store.withPrimaryKeys(new int[0]);
- }
+ TableStore store =
+ buildTableStore(
+ isBatch || !transaction,
+ TEMPORARY_FOLDER,
+ partitioned ? new int[] {1} : new int[0],
+ hasPk ? new int[] {2} : new int[0]);
// prepare log
DynamicTableFactory.Context context =
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceTest.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceTest.java
index 2cbda14..8ddbf64 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceTest.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceTest.java
@@ -145,6 +145,7 @@ public class FileStoreSourceTest {
hasPk ? new DeduplicateMergeFunction() : new ValueCountMergeFunction();
return new FileStoreImpl(
"/fake/path",
+ 0,
new FileStoreOptions(new Configuration()),
user,
partitionType,
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreImpl.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreImpl.java
index 00065c4..abcb2b1 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreImpl.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreImpl.java
@@ -48,6 +48,7 @@ import java.util.stream.Collectors;
public class FileStoreImpl implements FileStore {
private final String tablePath;
+ private final long schemaId;
private final FileStoreOptions options;
private final String user;
private final RowType partitionType;
@@ -58,6 +59,7 @@ public class FileStoreImpl implements FileStore {
public FileStoreImpl(
String tablePath,
+ long schemaId,
FileStoreOptions options,
String user,
RowType partitionType,
@@ -65,6 +67,7 @@ public class FileStoreImpl implements FileStore {
RowType valueType,
MergeFunction mergeFunction) {
this.tablePath = tablePath;
+ this.schemaId = schemaId;
this.options = options;
this.user = user;
this.partitionType = partitionType;
@@ -126,6 +129,7 @@ public class FileStoreImpl implements FileStore {
@Override
public FileStoreCommitImpl newCommit() {
return new FileStoreCommitImpl(
+ schemaId,
user,
partitionType,
pathFactory(),
@@ -177,6 +181,7 @@ public class FileStoreImpl implements FileStore {
public static FileStoreImpl createWithPrimaryKey(
String tablePath,
+ long schemaId,
FileStoreOptions options,
String user,
RowType partitionType,
@@ -213,11 +218,12 @@ public class FileStoreImpl implements FileStore {
}
return new FileStoreImpl(
- tablePath, options, user, partitionType, keyType, rowType, mergeFunction);
+ tablePath, schemaId, options, user, partitionType, keyType, rowType, mergeFunction);
}
public static FileStoreImpl createWithValueCount(
String tablePath,
+ long schemaId,
FileStoreOptions options,
String user,
RowType partitionType,
@@ -227,6 +233,13 @@ public class FileStoreImpl implements FileStore {
new LogicalType[] {new BigIntType(false)}, new String[] {"_VALUE_COUNT"});
MergeFunction mergeFunction = new ValueCountMergeFunction();
return new FileStoreImpl(
- tablePath, options, user, partitionType, rowType, countType, mergeFunction);
+ tablePath,
+ schemaId,
+ options,
+ user,
+ partitionType,
+ rowType,
+ countType,
+ mergeFunction);
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreOptions.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreOptions.java
index ca5f80c..df59e66 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreOptions.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreOptions.java
@@ -146,6 +146,10 @@ public class FileStoreOptions implements Serializable {
return allOptions;
}
+ public FileStoreOptions(Map<String, String> options) {
+ this(Configuration.fromMap(options));
+ }
+
public FileStoreOptions(Configuration options) {
this.options = options;
// TODO validate all keys
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/Snapshot.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/Snapshot.java
index 65b65e1..72791cc 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/Snapshot.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/Snapshot.java
@@ -22,12 +22,11 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.table.store.file.manifest.ManifestFileMeta;
import org.apache.flink.table.store.file.manifest.ManifestList;
import org.apache.flink.table.store.file.utils.FileUtils;
+import org.apache.flink.table.store.file.utils.JsonSerdeUtil;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.util.ArrayList;
@@ -40,6 +39,7 @@ public class Snapshot {
public static final long FIRST_SNAPSHOT_ID = 1;
private static final String FIELD_ID = "id";
+ private static final String FIELD_SCHEMA_ID = "schemaId";
private static final String FIELD_BASE_MANIFEST_LIST = "baseManifestList";
private static final String FIELD_DELTA_MANIFEST_LIST = "deltaManifestList";
private static final String FIELD_COMMIT_USER = "commitUser";
@@ -51,6 +51,9 @@ public class Snapshot {
@JsonProperty(FIELD_ID)
private final long id;
+ @JsonProperty(FIELD_SCHEMA_ID)
+ private final long schemaId;
+
// a manifest list recording all changes from the previous snapshots
@JsonProperty(FIELD_BASE_MANIFEST_LIST)
private final String baseManifestList;
@@ -79,6 +82,7 @@ public class Snapshot {
@JsonCreator
public Snapshot(
@JsonProperty(FIELD_ID) long id,
+ @JsonProperty(FIELD_SCHEMA_ID) long schemaId,
@JsonProperty(FIELD_BASE_MANIFEST_LIST) String baseManifestList,
@JsonProperty(FIELD_DELTA_MANIFEST_LIST) String deltaManifestList,
@JsonProperty(FIELD_COMMIT_USER) String commitUser,
@@ -87,6 +91,7 @@ public class Snapshot {
@JsonProperty(FIELD_TIME_MILLIS) long timeMillis,
@JsonProperty(FIELD_LOG_OFFSETS) Map<Integer, Long> logOffsets) {
this.id = id;
+ this.schemaId = schemaId;
this.baseManifestList = baseManifestList;
this.deltaManifestList = deltaManifestList;
this.commitUser = commitUser;
@@ -101,6 +106,11 @@ public class Snapshot {
return id;
}
+ @JsonGetter(FIELD_SCHEMA_ID)
+ public long schemaId() {
+ return schemaId;
+ }
+
@JsonGetter(FIELD_BASE_MANIFEST_LIST)
public String baseManifestList() {
return baseManifestList;
@@ -137,11 +147,7 @@ public class Snapshot {
}
public String toJson() {
- try {
- return new ObjectMapper().writeValueAsString(this);
- } catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
+ return JsonSerdeUtil.toJson(this);
}
public List<ManifestFileMeta> readAllManifests(ManifestList manifestList) {
@@ -152,11 +158,7 @@ public class Snapshot {
}
public static Snapshot fromJson(String json) {
- try {
- return new ObjectMapper().readValue(json, Snapshot.class);
- } catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
+ return JsonSerdeUtil.fromJson(json, Snapshot.class);
}
public static Snapshot fromPath(Path path) {
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 b4fa8cf..e8e09fc 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
@@ -73,6 +73,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
private static final Logger LOG = LoggerFactory.getLogger(FileStoreCommitImpl.class);
+ private final long schemaId;
private final String commitUser;
private final RowType partitionType;
private final RowDataToObjectArrayConverter partitionObjectConverter;
@@ -87,6 +88,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
@Nullable private Lock lock;
public FileStoreCommitImpl(
+ long schemaId,
String commitUser,
RowType partitionType,
FileStorePathFactory pathFactory,
@@ -96,6 +98,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
int numBucket,
MemorySize manifestTargetSize,
int manifestMergeMinCount) {
+ this.schemaId = schemaId;
this.commitUser = commitUser;
this.partitionType = partitionType;
this.partitionObjectConverter = new RowDataToObjectArrayConverter(partitionType);
@@ -367,6 +370,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
newSnapshot =
new Snapshot(
newSnapshotId,
+ schemaId,
previousChangesListName,
newChangesListName,
commitUser,
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/ArrayDataType.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/ArrayDataType.java
new file mode 100644
index 0000000..a2d1ee4
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/ArrayDataType.java
@@ -0,0 +1,60 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.types.logical.ArrayType;
+
+import java.util.Objects;
+
+/** A data type for array. */
+public class ArrayDataType extends DataType {
+
+ private static final long serialVersionUID = 1L;
+
+ private final DataType elementType;
+
+ public ArrayDataType(DataType elementType) {
+ super(new ArrayType(elementType.logicalType));
+ this.elementType = elementType;
+ }
+
+ public DataType elementType() {
+ return elementType;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ ArrayDataType that = (ArrayDataType) o;
+ return Objects.equals(elementType, that.elementType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), elementType);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/AtomicDataType.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/AtomicDataType.java
new file mode 100644
index 0000000..6d4db17
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/AtomicDataType.java
@@ -0,0 +1,31 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.types.logical.LogicalType;
+
+/** A data type that does not contain further data types (e.g. {@code INT} or {@code BOOLEAN}). */
+public final class AtomicDataType extends DataType {
+
+ private static final long serialVersionUID = 1L;
+
+ public AtomicDataType(LogicalType logicalType) {
+ super(logicalType);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/DataField.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/DataField.java
new file mode 100644
index 0000000..58eceb7
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/DataField.java
@@ -0,0 +1,93 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.store.file.utils.JsonSerdeUtil;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/** Defines the field of a row type. */
+public final class DataField implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int id;
+
+ private final String name;
+
+ private final DataType type;
+
+ private final @Nullable String description;
+
+ public DataField(int id, String name, DataType dataType) {
+ this(id, name, dataType, null);
+ }
+
+ public DataField(int id, String name, DataType type, @Nullable String description) {
+ this.id = id;
+ this.name = name;
+ this.type = type;
+ this.description = description;
+ }
+
+ public int id() {
+ return id;
+ }
+
+ public String name() {
+ return name;
+ }
+
+ public DataType type() {
+ return type;
+ }
+
+ @Nullable
+ public String description() {
+ return description;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ DataField field = (DataField) o;
+ return Objects.equals(id, field.id)
+ && Objects.equals(name, field.name)
+ && Objects.equals(type, field.type)
+ && Objects.equals(description, field.description);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(id, name, type, description);
+ }
+
+ @Override
+ public String toString() {
+ return JsonSerdeUtil.toJson(this);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/DataFieldSerializer.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/DataFieldSerializer.java
new file mode 100644
index 0000000..d7ab109
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/DataFieldSerializer.java
@@ -0,0 +1,128 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.store.file.utils.JsonSerializer;
+import org.apache.flink.table.types.logical.utils.LogicalTypeParser;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/** A {@link JsonSerializer} for {@link DataField}. */
+public class DataFieldSerializer implements JsonSerializer<DataField> {
+
+ public static final DataFieldSerializer INSTANCE = new DataFieldSerializer();
+
+ @Override
+ public void serialize(DataField dataField, JsonGenerator generator) throws IOException {
+ generator.writeStartObject();
+ generator.writeNumberField("id", dataField.id());
+ generator.writeStringField("name", dataField.name());
+ generator.writeFieldName("type");
+ typeToJson(dataField.type(), generator);
+ if (dataField.description() != null) {
+ generator.writeStringField("description", dataField.description());
+ }
+ generator.writeEndObject();
+ }
+
+ @Override
+ public DataField deserialize(JsonNode node) {
+ int id = node.get("id").asInt();
+ String name = node.get("name").asText();
+ DataType type = typeFromJson(node.get("type"));
+ JsonNode descriptionNode = node.get("description");
+ String description = null;
+ if (descriptionNode != null) {
+ description = descriptionNode.asText();
+ }
+ return new DataField(id, name, type, description);
+ }
+
+ private static void typeToJson(DataType type, JsonGenerator generator) throws IOException {
+ if (type instanceof AtomicDataType) {
+ generator.writeString(type.logicalType.asSerializableString());
+ } else if (type instanceof ArrayDataType) {
+ generator.writeStartObject();
+ generator.writeStringField("type", "array");
+ generator.writeFieldName("element");
+ typeToJson(((ArrayDataType) type).elementType(), generator);
+ generator.writeEndObject();
+ } else if (type instanceof MultisetDataType) {
+ generator.writeStartObject();
+ generator.writeStringField("type", "multiset");
+ generator.writeFieldName("element");
+ typeToJson(((MultisetDataType) type).elementType(), generator);
+ generator.writeEndObject();
+ } else if (type instanceof MapDataType) {
+ generator.writeStartObject();
+ generator.writeStringField("type", "map");
+ generator.writeFieldName("key");
+ typeToJson(((MapDataType) type).keyType(), generator);
+ generator.writeFieldName("value");
+ typeToJson(((MapDataType) type).valueType(), generator);
+ generator.writeEndObject();
+ } else if (type instanceof RowDataType) {
+ generator.writeStartObject();
+ generator.writeStringField("type", "row");
+ generator.writeArrayFieldStart("fields");
+ for (DataField field : ((RowDataType) type).fields()) {
+ INSTANCE.serialize(field, generator);
+ }
+ generator.writeEndArray();
+ generator.writeEndObject();
+ } else {
+ throw new UnsupportedOperationException("Unsupported type: " + type);
+ }
+ }
+
+ private static DataType typeFromJson(JsonNode json) {
+ if (json.isTextual()) {
+ return new AtomicDataType(LogicalTypeParser.parse(json.asText()));
+ } else if (json.isObject()) {
+ String typeString = json.get("type").asText();
+ if ("array".equals(typeString)) {
+ DataType element = typeFromJson(json.get("element"));
+ return new ArrayDataType(element);
+ } else if ("multiset".equals(typeString)) {
+ DataType element = typeFromJson(json.get("element"));
+ return new MultisetDataType(element);
+ } else if ("map".equals(typeString)) {
+ DataType key = typeFromJson(json.get("key"));
+ DataType value = typeFromJson(json.get("value"));
+ return new MapDataType(key, value);
+ } else if ("row".equals(typeString)) {
+ JsonNode fieldArray = json.get("fields");
+ Iterator<JsonNode> iterator = fieldArray.elements();
+ List<DataField> fields = new ArrayList<>(fieldArray.size());
+ while (iterator.hasNext()) {
+ fields.add(INSTANCE.deserialize(iterator.next()));
+ }
+ return new RowDataType(fields);
+ }
+ }
+
+ throw new IllegalArgumentException("Can not parse: " + json);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/DataType.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/DataType.java
new file mode 100644
index 0000000..26cbdbe
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/DataType.java
@@ -0,0 +1,64 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/** Describes the data type in the table store ecosystem. */
+public abstract class DataType implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ protected final LogicalType logicalType;
+
+ DataType(LogicalType logicalType) {
+ this.logicalType =
+ Preconditions.checkNotNull(logicalType, "Logical type must not be null.");
+ }
+
+ /**
+ * Returns the corresponding logical type.
+ *
+ * @return a parameterized instance of {@link LogicalType}
+ */
+ public LogicalType logicalType() {
+ return logicalType;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ DataType dataType = (DataType) o;
+ return Objects.equals(logicalType, dataType.logicalType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(logicalType);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/MapDataType.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/MapDataType.java
new file mode 100644
index 0000000..7f96e19
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/MapDataType.java
@@ -0,0 +1,67 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.types.logical.MapType;
+
+import java.util.Objects;
+
+/** A data type that contains a key and value data type. */
+public class MapDataType extends DataType {
+
+ private static final long serialVersionUID = 1L;
+
+ private final DataType keyType;
+
+ private final DataType valueType;
+
+ public MapDataType(DataType keyType, DataType valueType) {
+ super(new MapType(keyType.logicalType, valueType.logicalType));
+ this.keyType = keyType;
+ this.valueType = valueType;
+ }
+
+ public DataType keyType() {
+ return keyType;
+ }
+
+ public DataType valueType() {
+ return valueType;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ MapDataType that = (MapDataType) o;
+ return Objects.equals(keyType, that.keyType) && Objects.equals(valueType, that.valueType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), keyType, valueType);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/MultisetDataType.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/MultisetDataType.java
new file mode 100644
index 0000000..66a5259
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/MultisetDataType.java
@@ -0,0 +1,60 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.types.logical.ArrayType;
+
+import java.util.Objects;
+
+/** A data type for multiset. */
+public class MultisetDataType extends DataType {
+
+ private static final long serialVersionUID = 1L;
+
+ private final DataType elementType;
+
+ public MultisetDataType(DataType elementType) {
+ super(new ArrayType(elementType.logicalType));
+ this.elementType = elementType;
+ }
+
+ public DataType elementType() {
+ return elementType;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ MultisetDataType that = (MultisetDataType) o;
+ return Objects.equals(elementType, that.elementType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), elementType);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/RowDataType.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/RowDataType.java
new file mode 100644
index 0000000..b17efad
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/RowDataType.java
@@ -0,0 +1,72 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/** A data type that contains field data types. */
+public class RowDataType extends DataType {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List<DataField> fields;
+
+ public RowDataType(List<DataField> fields) {
+ super(toRowType(fields));
+ this.fields = fields;
+ }
+
+ private static RowType toRowType(List<DataField> fields) {
+ List<RowType.RowField> typeFields = new ArrayList<>(fields.size());
+ for (DataField field : fields) {
+ typeFields.add(
+ new RowType.RowField(
+ field.name(), field.type().logicalType, field.description()));
+ }
+ return new RowType(typeFields);
+ }
+
+ public List<DataField> fields() {
+ return fields;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ RowDataType that = (RowDataType) o;
+ return Objects.equals(fields, that.fields);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), fields);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/Schema.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/Schema.java
new file mode 100644
index 0000000..134af4c
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/Schema.java
@@ -0,0 +1,210 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.store.file.utils.JsonSerdeUtil;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/** Schema of table store. */
+public class Schema {
+
+ private final long id;
+
+ private final List<DataField> fields;
+
+ /** Not available from fields, as some fields may have been deleted. */
+ private final int highestFieldId;
+
+ private final List<String> partitionKeys;
+
+ private final List<String> primaryKeys;
+
+ private final Map<String, String> options;
+
+ public Schema(
+ long id,
+ List<DataField> fields,
+ int highestFieldId,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ Map<String, String> options) {
+ this.id = id;
+ this.fields = fields;
+ this.highestFieldId = highestFieldId;
+ this.partitionKeys = partitionKeys;
+ this.primaryKeys = primaryKeys;
+ this.options = Collections.unmodifiableMap(options);
+
+ // try to trim to validate primary keys
+ trimmedPrimaryKeys();
+ }
+
+ public long id() {
+ return id;
+ }
+
+ public List<DataField> fields() {
+ return fields;
+ }
+
+ public int highestFieldId() {
+ return highestFieldId;
+ }
+
+ public List<String> partitionKeys() {
+ return partitionKeys;
+ }
+
+ public List<String> primaryKeys() {
+ return primaryKeys;
+ }
+
+ public List<String> trimmedPrimaryKeys() {
+ if (primaryKeys.size() > 0) {
+ Preconditions.checkState(
+ primaryKeys.containsAll(partitionKeys),
+ String.format(
+ "Primary key constraint %s should include all partition fields %s",
+ primaryKeys, partitionKeys));
+ List<String> adjusted =
+ primaryKeys.stream()
+ .filter(pk -> !partitionKeys.contains(pk))
+ .collect(Collectors.toList());
+
+ Preconditions.checkState(
+ adjusted.size() > 0,
+ String.format(
+ "Primary key constraint %s should not be same with partition fields %s, this will result in only one record in a partition",
+ primaryKeys, partitionKeys));
+
+ return adjusted;
+ }
+
+ return primaryKeys;
+ }
+
+ public Map<String, String> options() {
+ return options;
+ }
+
+ public RowType logicalRowType() {
+ return (RowType) new RowDataType(fields).logicalType;
+ }
+
+ @Override
+ public String toString() {
+ return JsonSerdeUtil.toJson(this);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Schema schema = (Schema) o;
+ return Objects.equals(fields, schema.fields)
+ && Objects.equals(partitionKeys, schema.partitionKeys)
+ && Objects.equals(primaryKeys, schema.primaryKeys)
+ && Objects.equals(options, schema.options);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(fields, partitionKeys, primaryKeys, options);
+ }
+
+ public static List<DataField> newFields(RowType rowType) {
+ return ((RowDataType) toDataType(rowType, new AtomicInteger(-1))).fields();
+ }
+
+ private static DataType toDataType(LogicalType type, AtomicInteger currentHighestFieldId) {
+ if (type instanceof ArrayType) {
+ DataType element =
+ toDataType(((ArrayType) type).getElementType(), currentHighestFieldId);
+ return new ArrayDataType(element);
+ } else if (type instanceof MultisetType) {
+ DataType element =
+ toDataType(((MultisetType) type).getElementType(), currentHighestFieldId);
+ return new MultisetDataType(element);
+ } else if (type instanceof MapType) {
+ DataType key = toDataType(((MapType) type).getKeyType(), currentHighestFieldId);
+ DataType value = toDataType(((MapType) type).getValueType(), currentHighestFieldId);
+ return new MapDataType(key, value);
+ } else if (type instanceof RowType) {
+ List<DataField> fields = new ArrayList<>();
+ for (RowType.RowField field : ((RowType) type).getFields()) {
+ int id = currentHighestFieldId.incrementAndGet();
+ DataType fieldType = toDataType(field.getType(), currentHighestFieldId);
+ fields.add(
+ new DataField(
+ id,
+ field.getName(),
+ fieldType,
+ field.getDescription().orElse(null)));
+ }
+ return new RowDataType(fields);
+ } else {
+ return new AtomicDataType(type);
+ }
+ }
+
+ public static int currentHighestFieldId(List<DataField> fields) {
+ Set<Integer> fieldIds = new HashSet<>();
+ collectFieldIds(new RowDataType(fields), fieldIds);
+ return fieldIds.stream().max(Integer::compareTo).orElse(-1);
+ }
+
+ private static void collectFieldIds(DataType type, Set<Integer> fieldIds) {
+ if (type instanceof ArrayDataType) {
+ collectFieldIds(((ArrayDataType) type).elementType(), fieldIds);
+ } else if (type instanceof MultisetDataType) {
+ collectFieldIds(((MultisetDataType) type).elementType(), fieldIds);
+ } else if (type instanceof MapDataType) {
+ collectFieldIds(((MapDataType) type).keyType(), fieldIds);
+ collectFieldIds(((MapDataType) type).keyType(), fieldIds);
+ } else if (type instanceof RowDataType) {
+ for (DataField field : ((RowDataType) type).fields()) {
+ if (fieldIds.contains(field.id())) {
+ throw new RuntimeException(
+ String.format("Broken schema, field id %s is duplicated.", field.id()));
+ }
+ fieldIds.add(field.id());
+ collectFieldIds(field.type(), fieldIds);
+ }
+ }
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaManager.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaManager.java
new file mode 100644
index 0000000..50610d2
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaManager.java
@@ -0,0 +1,160 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.store.file.operation.Lock;
+import org.apache.flink.table.store.file.utils.FileUtils;
+import org.apache.flink.table.store.file.utils.JsonSerdeUtil;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.store.file.utils.FileUtils.listVersionedFiles;
+
+/** Schema Manager to manage schema versions. */
+public class SchemaManager {
+
+ private static final String SCHEMA_PREFIX = "schema-";
+
+ private final Path tableRoot;
+
+ /** Default no lock. */
+ private Lock lock = Callable::call;
+
+ public SchemaManager(Path tableRoot) {
+ this.tableRoot = tableRoot;
+ }
+
+ public SchemaManager withLock(Lock lock) {
+ this.lock = lock;
+ return this;
+ }
+
+ /** @return latest schema. */
+ public Optional<Schema> latest() {
+ try {
+ return listVersionedFiles(schemaDirectory(), SCHEMA_PREFIX)
+ .reduce(Math::max)
+ .map(this::schema);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ /** List all schema. */
+ public List<Schema> listAll() {
+ try {
+ return listVersionedFiles(schemaDirectory(), SCHEMA_PREFIX)
+ .map(this::schema)
+ .collect(Collectors.toList());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ /** Create a new schema from {@link UpdateSchema}. */
+ public Schema commitNewVersion(UpdateSchema updateSchema) throws Exception {
+ RowType rowType = updateSchema.rowType();
+ List<String> partitionKeys = updateSchema.partitionKeys();
+ List<String> primaryKeys = updateSchema.primaryKeys();
+ Map<String, String> options = updateSchema.options();
+
+ while (true) {
+ long id;
+ int highestFieldId;
+ List<DataField> fields;
+ Optional<Schema> latest = latest();
+ if (latest.isPresent()) {
+ Schema oldSchema = latest.get();
+ Preconditions.checkArgument(
+ oldSchema.primaryKeys().equals(primaryKeys),
+ "Primary key modification is not supported, "
+ + "old primaryKeys is %s, new primaryKeys is %s",
+ oldSchema.primaryKeys(),
+ primaryKeys);
+
+ if (!updateSchema
+ .rowType()
+ .getFields()
+ .equals(oldSchema.logicalRowType().getFields())
+ || !updateSchema.partitionKeys().equals(oldSchema.partitionKeys())) {
+ throw new UnsupportedOperationException(
+ "TODO: support update field types and partition keys. ");
+ }
+
+ fields = oldSchema.fields();
+ id = oldSchema.id() + 1;
+ highestFieldId = oldSchema.highestFieldId();
+ } else {
+ fields = Schema.newFields(rowType);
+ highestFieldId = Schema.currentHighestFieldId(fields);
+ id = 0;
+ }
+
+ Schema schema =
+ new Schema(id, fields, highestFieldId, partitionKeys, primaryKeys, options);
+
+ Path temp = toTmpSchemaPath(id);
+ Path finalFile = toSchemaPath(id);
+
+ boolean success = false;
+ try {
+ FileUtils.writeFileUtf8(temp, schema.toString());
+ success = lock.runWithLock(() -> temp.getFileSystem().rename(temp, finalFile));
+ if (success) {
+ return schema;
+ }
+ } finally {
+ if (!success) {
+ FileUtils.deleteOrWarn(temp);
+ }
+ }
+ }
+ }
+
+ /** Read schema for schema id. */
+ public Schema schema(long id) {
+ try {
+ return JsonSerdeUtil.fromJson(FileUtils.readFileUtf8(toSchemaPath(id)), Schema.class);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private Path schemaDirectory() {
+ return new Path(tableRoot + "/schema");
+ }
+
+ private Path toTmpSchemaPath(long id) {
+ return new Path(tableRoot + "/schema/." + SCHEMA_PREFIX + id + "-" + UUID.randomUUID());
+ }
+
+ private Path toSchemaPath(long id) {
+ return new Path(tableRoot + "/schema/" + SCHEMA_PREFIX + id);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaSerializer.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaSerializer.java
new file mode 100644
index 0000000..d200149
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaSerializer.java
@@ -0,0 +1,107 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.store.file.utils.JsonSerializer;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/** A {@link JsonSerializer} for {@link Schema}. */
+public class SchemaSerializer implements JsonSerializer<Schema> {
+
+ public static final SchemaSerializer INSTANCE = new SchemaSerializer();
+
+ @Override
+ public void serialize(Schema schema, JsonGenerator generator) throws IOException {
+ generator.writeStartObject();
+
+ generator.writeNumberField("id", schema.id());
+
+ generator.writeArrayFieldStart("fields");
+ for (DataField field : schema.fields()) {
+ DataFieldSerializer.INSTANCE.serialize(field, generator);
+ }
+ generator.writeEndArray();
+
+ generator.writeNumberField("highestFieldId", schema.highestFieldId());
+
+ generator.writeArrayFieldStart("partitionKeys");
+ for (String partitionKey : schema.partitionKeys()) {
+ generator.writeString(partitionKey);
+ }
+ generator.writeEndArray();
+
+ generator.writeArrayFieldStart("primaryKeys");
+ for (String primaryKey : schema.primaryKeys()) {
+ generator.writeString(primaryKey);
+ }
+ generator.writeEndArray();
+
+ generator.writeObjectFieldStart("options");
+ for (Map.Entry<String, String> entry : schema.options().entrySet()) {
+ generator.writeStringField(entry.getKey(), entry.getValue());
+ }
+ generator.writeEndObject();
+
+ generator.writeEndObject();
+ }
+
+ @Override
+ public Schema deserialize(JsonNode node) {
+ int id = node.get("id").asInt();
+
+ Iterator<JsonNode> fieldJsons = node.get("fields").elements();
+ List<DataField> fields = new ArrayList<>();
+ while (fieldJsons.hasNext()) {
+ fields.add(DataFieldSerializer.INSTANCE.deserialize(fieldJsons.next()));
+ }
+
+ int highestFieldId = node.get("highestFieldId").asInt();
+
+ Iterator<JsonNode> partitionJsons = node.get("partitionKeys").elements();
+ List<String> partitionKeys = new ArrayList<>();
+ while (partitionJsons.hasNext()) {
+ partitionKeys.add(partitionJsons.next().asText());
+ }
+
+ Iterator<JsonNode> primaryJsons = node.get("primaryKeys").elements();
+ List<String> primaryKeys = new ArrayList<>();
+ while (primaryJsons.hasNext()) {
+ primaryKeys.add(primaryJsons.next().asText());
+ }
+
+ JsonNode optionsJson = node.get("options");
+ Map<String, String> options = new HashMap<>();
+ Iterator<String> optionsKeys = optionsJson.fieldNames();
+ while (optionsKeys.hasNext()) {
+ String key = optionsKeys.next();
+ options.put(key, optionsJson.get(key).asText());
+ }
+
+ return new Schema(id, fields, highestFieldId, partitionKeys, primaryKeys, options);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/UpdateSchema.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/UpdateSchema.java
new file mode 100644
index 0000000..ef0ff7c
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/UpdateSchema.java
@@ -0,0 +1,77 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.List;
+import java.util.Map;
+
+/** A update schema. */
+public class UpdateSchema {
+
+ private final RowType rowType;
+
+ private final List<String> partitionKeys;
+
+ private final List<String> primaryKeys;
+
+ private final Map<String, String> options;
+
+ public UpdateSchema(
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ Map<String, String> options) {
+ this.rowType = rowType;
+ this.partitionKeys = partitionKeys;
+ this.primaryKeys = primaryKeys;
+ this.options = options;
+ }
+
+ public RowType rowType() {
+ return rowType;
+ }
+
+ public List<String> partitionKeys() {
+ return partitionKeys;
+ }
+
+ public List<String> primaryKeys() {
+ return primaryKeys;
+ }
+
+ public Map<String, String> options() {
+ return options;
+ }
+
+ @Override
+ public String toString() {
+ return "UpdateSchema{"
+ + "rowType="
+ + rowType
+ + ", partitionKeys="
+ + partitionKeys
+ + ", primaryKeys="
+ + primaryKeys
+ + ", options="
+ + options
+ + '}';
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FileUtils.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FileUtils.java
index a338cad..c9d6250 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FileUtils.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/FileUtils.java
@@ -42,9 +42,11 @@ import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinWorkerThread;
+import java.util.stream.Stream;
/** Utils for file reading and writing. */
public class FileUtils {
@@ -149,4 +151,30 @@ public class FileUtils {
}
return statuses;
}
+
+ /**
+ * List versioned files for the directory.
+ *
+ * @return version stream
+ */
+ public static Stream<Long> listVersionedFiles(Path dir, String prefix) throws IOException {
+ if (!dir.getFileSystem().exists(dir)) {
+ return Stream.of();
+ }
+
+ FileStatus[] statuses = FileUtils.safelyListFileStatus(dir);
+
+ if (statuses == null) {
+ throw new RuntimeException(
+ String.format(
+ "The return value is null of the listStatus for the '%s' directory.",
+ dir));
+ }
+
+ return Arrays.stream(statuses)
+ .map(FileStatus::getPath)
+ .map(Path::getName)
+ .filter(name -> name.startsWith(prefix))
+ .map(name -> Long.parseLong(name.substring(prefix.length())));
+ }
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/JsonSerdeUtil.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/JsonSerdeUtil.java
new file mode 100644
index 0000000..0d618e6
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/JsonSerdeUtil.java
@@ -0,0 +1,101 @@
+/*
+ * 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.file.utils;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.store.file.schema.DataField;
+import org.apache.flink.table.store.file.schema.DataFieldSerializer;
+import org.apache.flink.table.store.file.schema.Schema;
+import org.apache.flink.table.store.file.schema.SchemaSerializer;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationContext;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.Module;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.module.SimpleModule;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+
+/** A utility class that provide abilities for JSON serialization and deserialization. */
+@Internal
+public class JsonSerdeUtil {
+
+ /**
+ * Object mapper shared instance to serialize and deserialize the plan. Note that creating and
+ * copying of object mappers is expensive and should be avoided.
+ */
+ private static final ObjectMapper OBJECT_MAPPER_INSTANCE;
+
+ static {
+ OBJECT_MAPPER_INSTANCE = new ObjectMapper();
+ OBJECT_MAPPER_INSTANCE.registerModule(createTableStoreJacksonModule());
+ }
+
+ public static <T> T fromJson(String json, Class<T> clazz) {
+ try {
+ return OBJECT_MAPPER_INSTANCE.reader().readValue(json, clazz);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public static <T> String toJson(T t) {
+ try {
+ return OBJECT_MAPPER_INSTANCE.writer().writeValueAsString(t);
+ } catch (JsonProcessingException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private static Module createTableStoreJacksonModule() {
+ SimpleModule module = new SimpleModule("Table store");
+ registerJsonObjects(module, Schema.class, SchemaSerializer.INSTANCE);
+ registerJsonObjects(module, DataField.class, DataFieldSerializer.INSTANCE);
+ return module;
+ }
+
+ private static <T> void registerJsonObjects(
+ SimpleModule module, Class<T> clazz, JsonSerializer<T> serializer) {
+ module.addSerializer(
+ new StdSerializer<T>(clazz) {
+ @Override
+ public void serialize(T t, JsonGenerator generator, SerializerProvider provider)
+ throws IOException {
+ serializer.serialize(t, generator);
+ }
+ });
+ module.addDeserializer(
+ clazz,
+ new StdDeserializer<T>(clazz) {
+ @Override
+ public T deserialize(JsonParser parser, DeserializationContext context)
+ throws IOException {
+ return serializer.deserialize(parser.readValueAsTree());
+ }
+ });
+ }
+
+ private JsonSerdeUtil() {}
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/JsonSerializer.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/JsonSerializer.java
new file mode 100644
index 0000000..bb78e7e
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/JsonSerializer.java
@@ -0,0 +1,32 @@
+/*
+ * 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.file.utils;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.io.IOException;
+
+/** Json serializer for jackson. */
+public interface JsonSerializer<T> {
+
+ void serialize(T t, JsonGenerator generator) throws IOException;
+
+ T deserialize(JsonNode node);
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/SnapshotFinder.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/SnapshotFinder.java
index 11c9026..026d3f2 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/SnapshotFinder.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/SnapshotFinder.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.store.file.utils;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
@@ -30,6 +29,8 @@ import java.io.IOException;
import java.util.UUID;
import java.util.function.BinaryOperator;
+import static org.apache.flink.table.store.file.utils.FileUtils.listVersionedFiles;
+
/** Find latest and earliest snapshot. */
public class SnapshotFinder {
@@ -85,26 +86,7 @@ public class SnapshotFinder {
private static Long findByListFiles(Path snapshotDir, BinaryOperator<Long> reducer)
throws IOException {
- FileStatus[] statuses = FileUtils.safelyListFileStatus(snapshotDir);
-
- if (statuses == null) {
- throw new RuntimeException(
- "The return value is null of the listStatus for the snapshot directory.");
- }
-
- Long result = null;
- for (FileStatus status : statuses) {
- String fileName = status.getPath().getName();
- if (fileName.startsWith(SNAPSHOT_PREFIX)) {
- try {
- long id = Long.parseLong(fileName.substring(SNAPSHOT_PREFIX.length()));
- result = result == null ? id : reducer.apply(result, id);
- } catch (NumberFormatException e) {
- throw new RuntimeException("Invalid snapshot file name found " + fileName, e);
- }
- }
- }
- return result;
+ return listVersionedFiles(snapshotDir, SNAPSHOT_PREFIX).reduce(reducer).orElse(null);
}
public static void commitLatestHint(Path snapshotDir, long snapshotId) throws IOException {
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
index c31fc82..add6c7d 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
@@ -113,6 +113,7 @@ public class TestFileStore extends FileStoreImpl {
MergeFunction mergeFunction) {
super(
options.path(ObjectIdentifier.of("catalog", "database", "table")).toString(),
+ 0L,
options,
UUID.randomUUID().toString(),
partitionType,
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaManagerTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaManagerTest.java
new file mode 100644
index 0000000..f4984a7
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaManagerTest.java
@@ -0,0 +1,175 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.store.file.utils.FailingAtomicRenameFileSystem;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.store.file.utils.FailingAtomicRenameFileSystem.retryArtificialException;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/** Test for {@link SchemaManager}. */
+public class SchemaManagerTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private SchemaManager manager;
+
+ private final List<String> partitionKeys = Collections.singletonList("f0");
+ private final List<String> primaryKeys = Arrays.asList("f0", "f1");
+ private final Map<String, String> options = Collections.singletonMap("key", "value");
+ private final RowType rowType = RowType.of(new IntType(), new BigIntType(), new VarCharType());
+ private final UpdateSchema updateSchema =
+ new UpdateSchema(rowType, partitionKeys, primaryKeys, options);
+
+ @BeforeEach
+ public void beforeEach() throws IOException {
+ // for failure tests
+ FailingAtomicRenameFileSystem.get().reset(100, 5000);
+ String root = FailingAtomicRenameFileSystem.getFailingPath(tempDir.toString());
+ manager = new SchemaManager(new Path(root));
+ }
+
+ @AfterEach
+ public void afterEach() {
+ // assert no temp file
+ File schema = new File(tempDir.toFile(), "schema");
+ assertThat(schema.exists()).isTrue();
+ String[] versions = schema.list();
+ assertThat(versions).isNotNull();
+ for (String version : versions) {
+ assertThat(version.startsWith(".")).isFalse();
+ }
+ }
+
+ @Test
+ public void testCommitNewVersion() throws Exception {
+ Schema schema = retryArtificialException(() -> manager.commitNewVersion(updateSchema));
+
+ Optional<Schema> latest = retryArtificialException(() -> manager.latest());
+
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "f0", new AtomicDataType(new IntType())),
+ new DataField(1, "f1", new AtomicDataType(new BigIntType())),
+ new DataField(2, "f2", new AtomicDataType(new VarCharType())));
+
+ assertThat(latest.isPresent()).isTrue();
+ assertThat(schema).isEqualTo(latest.get());
+ assertThat(schema.fields()).isEqualTo(fields);
+ assertThat(schema.partitionKeys()).isEqualTo(partitionKeys);
+ assertThat(schema.primaryKeys()).isEqualTo(primaryKeys);
+ assertThat(schema.options()).isEqualTo(options);
+ }
+
+ @Test
+ public void testUpdateOptions() throws Exception {
+ retryArtificialException(() -> manager.commitNewVersion(updateSchema));
+ Map<String, String> newOptions = Collections.singletonMap("new_k", "new_v");
+ UpdateSchema updateSchema =
+ new UpdateSchema(rowType, partitionKeys, primaryKeys, newOptions);
+ retryArtificialException(() -> manager.commitNewVersion(updateSchema));
+ Optional<Schema> latest = retryArtificialException(() -> manager.latest());
+ assertThat(latest.isPresent()).isTrue();
+ assertThat(latest.get().options()).isEqualTo(newOptions);
+ }
+
+ @Test
+ public void testUpdateSchema() throws Exception {
+ retryArtificialException(() -> manager.commitNewVersion(updateSchema));
+ RowType newType = RowType.of(new IntType(), new BigIntType());
+ UpdateSchema updateSchema = new UpdateSchema(newType, partitionKeys, primaryKeys, options);
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> retryArtificialException(() -> manager.commitNewVersion(updateSchema)));
+ }
+
+ @Test
+ public void testConcurrentCommit() throws Exception {
+ int threadNumber = ThreadLocalRandom.current().nextInt(3) + 2;
+ List<Thread> threads = new ArrayList<>();
+ for (int i = 0; i < threadNumber; i++) {
+ int id = i;
+ threads.add(
+ new Thread(
+ () -> {
+ // sleep to concurrent commit, exclude the effects of thread startup
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ Map<String, String> options = new HashMap<>();
+ options.put("id", String.valueOf(id));
+ UpdateSchema updateSchema =
+ new UpdateSchema(
+ rowType, partitionKeys, primaryKeys, options);
+ try {
+ retryArtificialException(
+ () -> manager.commitNewVersion(updateSchema));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }));
+ }
+
+ for (Thread thread : threads) {
+ thread.start();
+ }
+
+ for (Thread thread : threads) {
+ thread.join();
+ }
+
+ // assert
+ List<String> ids =
+ retryArtificialException(() -> manager.listAll()).stream()
+ .map(schema -> schema.options().get("id"))
+ .collect(Collectors.toList());
+ assertThat(ids)
+ .containsExactlyInAnyOrder(
+ IntStream.range(0, threadNumber)
+ .mapToObj(String::valueOf)
+ .toArray(String[]::new));
+ }
+}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaSerializationTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaSerializationTest.java
new file mode 100644
index 0000000..bd214ab
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaSerializationTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.store.file.utils.JsonSerdeUtil;
+import org.apache.flink.table.types.logical.IntType;
+
+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 static org.apache.flink.table.store.file.schema.SchemaTest.newRowType;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for serialize {@link Schema}. */
+public class SchemaSerializationTest {
+
+ @Test
+ public void testSchema() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "f0", new AtomicDataType(new IntType())),
+ new DataField(1, "f1", newRowType(2)),
+ new DataField(3, "f2", new ArrayDataType(newRowType(4))),
+ new DataField(5, "f3", new MultisetDataType(newRowType(6))),
+ new DataField(7, "f4", new MapDataType(newRowType(8), newRowType(9))));
+ List<String> partitionKeys = Collections.singletonList("f0");
+ List<String> primaryKeys = Arrays.asList("f0", "f1");
+ Map<String, String> options = new HashMap<>();
+ options.put("option-1", "value-1");
+ options.put("option-2", "value-2");
+
+ Schema schema = new Schema(1, fields, 10, partitionKeys, primaryKeys, options);
+ String serialized = JsonSerdeUtil.toJson(schema);
+
+ Schema deserialized = JsonSerdeUtil.fromJson(serialized, Schema.class);
+ assertThat(deserialized).isEqualTo(schema);
+ }
+}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaTest.java
new file mode 100644
index 0000000..85d4078
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaTest.java
@@ -0,0 +1,110 @@
+/*
+ * 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.file.schema;
+
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.junit.jupiter.api.Assertions;
+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 static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link Schema}. */
+public class SchemaTest {
+
+ @Test
+ public void testInvalidPrimaryKeys() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "f0", new AtomicDataType(new IntType())),
+ new DataField(1, "f1", new AtomicDataType(new IntType())),
+ new DataField(2, "f2", new AtomicDataType(new IntType())));
+ List<String> partitionKeys = Collections.singletonList("f0");
+ List<String> primaryKeys = Collections.singletonList("f1");
+ Map<String, String> options = new HashMap<>();
+
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> new Schema(1, fields, 10, partitionKeys, primaryKeys, options));
+ }
+
+ @Test
+ public void testInvalidFieldIds() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "f0", new AtomicDataType(new IntType())),
+ new DataField(0, "f1", new AtomicDataType(new IntType())));
+ Assertions.assertThrows(RuntimeException.class, () -> Schema.currentHighestFieldId(fields));
+ }
+
+ @Test
+ public void testHighestFieldId() {
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "f0", new AtomicDataType(new IntType())),
+ new DataField(20, "f1", new AtomicDataType(new IntType())));
+ assertThat(Schema.currentHighestFieldId(fields)).isEqualTo(20);
+ }
+
+ @Test
+ public void testTypeToSchema() {
+ RowType type =
+ RowType.of(
+ new LogicalType[] {
+ new IntType(),
+ newLogicalRowType(),
+ new ArrayType(newLogicalRowType()),
+ new MultisetType(newLogicalRowType()),
+ new MapType(newLogicalRowType(), newLogicalRowType())
+ },
+ new String[] {"f0", "f1", "f2", "f3", "f4"});
+
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "f0", new AtomicDataType(new IntType())),
+ new DataField(1, "f1", newRowType(2)),
+ new DataField(3, "f2", new ArrayDataType(newRowType(4))),
+ new DataField(5, "f3", new MultisetDataType(newRowType(6))),
+ new DataField(7, "f4", new MapDataType(newRowType(8), newRowType(9))));
+
+ assertThat(Schema.newFields(type)).isEqualTo(fields);
+ }
+
+ static RowType newLogicalRowType() {
+ return new RowType(
+ Collections.singletonList(new RowType.RowField("nestedField", new IntType())));
+ }
+
+ static RowDataType newRowType(int fieldId) {
+ return new RowDataType(
+ Collections.singletonList(
+ new DataField(fieldId, "nestedField", new AtomicDataType(new IntType()))));
+ }
+}
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 4371b01..8e1b4cc 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
@@ -25,9 +25,11 @@ import org.apache.flink.core.fs.FSDataOutputStreamWrapper;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystemFactory;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.ExceptionUtils;
import java.io.IOException;
import java.net.URI;
+import java.util.concurrent.Callable;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
@@ -158,4 +160,16 @@ public class FailingAtomicRenameFileSystem extends TestAtomicRenameFileSystem {
super.write(b, off, len);
}
}
+
+ public static <T> T retryArtificialException(Callable<T> callable) throws Exception {
+ while (true) {
+ try {
+ return callable.call();
+ } catch (Throwable t) {
+ if (!ExceptionUtils.findThrowable(t, ArtificialException.class).isPresent()) {
+ throw t;
+ }
+ }
+ }
+ }
}
diff --git a/flink-table-store-hive/src/main/java/org/apache/flink/table/store/mapred/TableStoreInputFormat.java b/flink-table-store-hive/src/main/java/org/apache/flink/table/store/mapred/TableStoreInputFormat.java
index b08ba37..f150ee8 100644
--- a/flink-table-store-hive/src/main/java/org/apache/flink/table/store/mapred/TableStoreInputFormat.java
+++ b/flink-table-store-hive/src/main/java/org/apache/flink/table/store/mapred/TableStoreInputFormat.java
@@ -141,6 +141,7 @@ public class TableStoreInputFormat implements InputFormat<Void, RowDataContainer
.collect(Collectors.toList()));
return FileStoreImpl.createWithPrimaryKey(
tableLocation,
+ 0, // TODO
new FileStoreOptions(options),
user,
partitionType,
@@ -149,7 +150,12 @@ public class TableStoreInputFormat implements InputFormat<Void, RowDataContainer
FileStoreOptions.MergeEngine.DEDUPLICATE);
} else {
return FileStoreImpl.createWithValueCount(
- tableLocation, new FileStoreOptions(options), user, partitionType, rowType);
+ tableLocation,
+ 0, // TODO
+ new FileStoreOptions(options),
+ user,
+ partitionType,
+ rowType);
}
}
}
diff --git a/flink-table-store-hive/src/test/java/org/apache/flink/table/store/FileStoreTestHelper.java b/flink-table-store-hive/src/test/java/org/apache/flink/table/store/FileStoreTestHelper.java
index b6963ed..1b1d477 100644
--- a/flink-table-store-hive/src/test/java/org/apache/flink/table/store/FileStoreTestHelper.java
+++ b/flink-table-store-hive/src/test/java/org/apache/flink/table/store/FileStoreTestHelper.java
@@ -71,6 +71,7 @@ public class FileStoreTestHelper {
this.store =
new FileStoreImpl(
options.path(oi).toString(),
+ 0,
options,
UUID.randomUUID().toString(),
partitionType,