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/06/23 03:58:45 UTC
[flink-table-store] branch master updated: [FLINK-28211] Rename Schema to TableSchema
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 5a34efc [FLINK-28211] Rename Schema to TableSchema
5a34efc is described below
commit 5a34efce8fadf96fc05194b28f62f0680a5afa62
Author: JingsongLi <lz...@aliyun.com>
AuthorDate: Thu Jun 23 11:38:33 2022 +0800
[FLINK-28211] Rename Schema to TableSchema
This closes #172
---
.../store/connector/AbstractTableStoreFactory.java | 16 ++++----
.../connector/sink/BucketStreamPartitioner.java | 10 ++---
.../table/store/connector/RescaleBucketITCase.java | 14 ++++---
.../table/store/connector/sink/StoreSinkTest.java | 8 ++--
.../store/connector/sink/TestFileStoreTable.java | 14 +++----
.../store/file/catalog/FileSystemCatalog.java | 6 +--
.../table/store/file/schema/SchemaManager.java | 43 +++++++++++-----------
.../table/store/file/schema/SchemaSerializer.java | 27 +++++++-------
.../file/schema/{Schema.java => TableSchema.java} | 22 +++++------
.../table/store/file/utils/JsonSerdeUtil.java | 4 +-
.../table/store/table/AbstractFileStoreTable.java | 12 +++---
.../store/table/AppendOnlyFileStoreTable.java | 19 +++++-----
.../table/ChangelogValueCountFileStoreTable.java | 18 ++++-----
.../table/ChangelogWithKeyFileStoreTable.java | 24 ++++++------
.../flink/table/store/table/FileStoreTable.java | 4 +-
.../table/store/table/FileStoreTableFactory.java | 17 +++++----
.../store/table/sink/SinkRecordConverter.java | 12 +++---
.../flink/table/store/table/source/TableScan.java | 15 +++++---
.../table/store/file/schema/SchemaManagerTest.java | 17 +++++----
...Test.java => TableSchemaSerializationTest.java} | 16 ++++----
.../{SchemaTest.java => TableSchemaTest.java} | 13 ++++---
.../store/table/AppendOnlyFileStoreTableTest.java | 7 ++--
.../ChangelogValueCountFileStoreTableTest.java | 6 +--
.../table/ChangelogWithKeyFileStoreTableTest.java | 6 +--
.../apache/flink/table/store/hive/HiveSchema.java | 28 +++++++-------
.../table/store/mapred/TableStoreInputFormat.java | 6 +--
...iveSchemaTest.java => HiveTableSchemaTest.java} | 2 +-
27 files changed, 200 insertions(+), 186 deletions(-)
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/AbstractTableStoreFactory.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/AbstractTableStoreFactory.java
index 4d16cbf..c1e783a 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/AbstractTableStoreFactory.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/AbstractTableStoreFactory.java
@@ -32,7 +32,7 @@ import org.apache.flink.table.store.connector.sink.TableStoreSink;
import org.apache.flink.table.store.connector.source.TableStoreSource;
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.TableSchema;
import org.apache.flink.table.store.file.schema.UpdateSchema;
import org.apache.flink.table.store.log.LogOptions;
import org.apache.flink.table.store.log.LogStoreTableFactory;
@@ -164,7 +164,7 @@ public abstract class AbstractTableStoreFactory
FileStoreTableFactory.create(
Configuration.fromMap(context.getCatalogTable().getOptions()));
- Schema schema = table.schema();
+ TableSchema tableSchema = table.schema();
UpdateSchema updateSchema = UpdateSchema.fromCatalogTable(context.getCatalogTable());
RowType rowType = updateSchema.rowType();
@@ -173,24 +173,24 @@ public abstract class AbstractTableStoreFactory
// compare fields to ignore isNullable for row type
Preconditions.checkArgument(
- schema.logicalRowType().getFields().equals(rowType.getFields()),
+ tableSchema.logicalRowType().getFields().equals(rowType.getFields()),
"Flink schema and store schema are not the same, "
+ "store schema is %s, Flink schema is %s",
- schema.logicalRowType(),
+ tableSchema.logicalRowType(),
rowType);
Preconditions.checkArgument(
- schema.partitionKeys().equals(partitionKeys),
+ tableSchema.partitionKeys().equals(partitionKeys),
"Flink partitionKeys and store partitionKeys are not the same, "
+ "store partitionKeys is %s, Flink partitionKeys is %s",
- schema.partitionKeys(),
+ tableSchema.partitionKeys(),
partitionKeys);
Preconditions.checkArgument(
- schema.primaryKeys().equals(primaryKeys),
+ tableSchema.primaryKeys().equals(primaryKeys),
"Flink primaryKeys and store primaryKeys are not the same, "
+ "store primaryKeys is %s, Flink primaryKeys is %s",
- schema.primaryKeys(),
+ tableSchema.primaryKeys(),
primaryKeys);
return table;
diff --git a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/BucketStreamPartitioner.java b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/BucketStreamPartitioner.java
index 5930514..774d2a1 100644
--- a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/BucketStreamPartitioner.java
+++ b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/sink/BucketStreamPartitioner.java
@@ -23,26 +23,26 @@ import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.store.file.schema.Schema;
+import org.apache.flink.table.store.file.schema.TableSchema;
import org.apache.flink.table.store.table.sink.SinkRecordConverter;
/** A {@link StreamPartitioner} to partition records by bucket. */
public class BucketStreamPartitioner extends StreamPartitioner<RowData> {
private final int numBucket;
- private final Schema schema;
+ private final TableSchema tableSchema;
private transient SinkRecordConverter recordConverter;
- public BucketStreamPartitioner(int numBucket, Schema schema) {
+ public BucketStreamPartitioner(int numBucket, TableSchema tableSchema) {
this.numBucket = numBucket;
- this.schema = schema;
+ this.tableSchema = tableSchema;
}
@Override
public void setup(int numberOfChannels) {
super.setup(numberOfChannels);
- this.recordConverter = new SinkRecordConverter(numBucket, schema);
+ this.recordConverter = new SinkRecordConverter(numBucket, tableSchema);
}
@Override
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/RescaleBucketITCase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/RescaleBucketITCase.java
index 2adfe20..996917c 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/RescaleBucketITCase.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/RescaleBucketITCase.java
@@ -25,8 +25,8 @@ import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.store.file.Snapshot;
-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.TableSchema;
import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.types.Row;
@@ -162,9 +162,10 @@ public class RescaleBucketITCase extends FileStoreTableITCase {
private void assertLatestSchema(
SchemaManager schemaManager, long expectedSchemaId, int expectedBucketNum) {
assertThat(schemaManager.latest()).isPresent();
- Schema schema = schemaManager.latest().get();
- assertThat(schema.id()).isEqualTo(expectedSchemaId);
- assertThat(schema.options()).containsEntry(BUCKET.key(), String.valueOf(expectedBucketNum));
+ TableSchema tableSchema = schemaManager.latest().get();
+ assertThat(tableSchema.id()).isEqualTo(expectedSchemaId);
+ assertThat(tableSchema.options())
+ .containsEntry(BUCKET.key(), String.valueOf(expectedBucketNum));
}
private void assertSnapshotSchema(
@@ -173,8 +174,9 @@ public class RescaleBucketITCase extends FileStoreTableITCase {
long expectedSchemaId,
int expectedBucketNum) {
assertThat(schemaIdFromSnapshot).isEqualTo(expectedSchemaId);
- Schema schema = schemaManager.schema(schemaIdFromSnapshot);
- assertThat(schema.options()).containsEntry(BUCKET.key(), String.valueOf(expectedBucketNum));
+ TableSchema tableSchema = schemaManager.schema(schemaIdFromSnapshot);
+ assertThat(tableSchema.options())
+ .containsEntry(BUCKET.key(), String.valueOf(expectedBucketNum));
}
private void innerTest(String catalogName, String tableName, boolean managedTable) {
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/StoreSinkTest.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/StoreSinkTest.java
index 4efcc09..1ddbad7 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/StoreSinkTest.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/StoreSinkTest.java
@@ -33,8 +33,8 @@ import org.apache.flink.table.store.connector.StatefulPrecommittingSinkWriter;
import org.apache.flink.table.store.connector.sink.TestFileStore.TestRecordWriter;
import org.apache.flink.table.store.file.KeyValue;
import org.apache.flink.table.store.file.manifest.ManifestCommittable;
-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.TableSchema;
import org.apache.flink.table.store.file.schema.UpdateSchema;
import org.apache.flink.table.store.file.writer.RecordWriter;
import org.apache.flink.table.types.logical.IntType;
@@ -89,7 +89,7 @@ public class StoreSinkTest {
@Before
public void before() throws Exception {
- Schema schema =
+ TableSchema tableSchema =
new SchemaManager(new Path(tempFolder.newFolder().toURI().toString()))
.commitNewVersion(
new UpdateSchema(
@@ -105,9 +105,9 @@ public class StoreSinkTest {
new HashMap<>(),
""));
- RowType partitionType = schema.logicalPartitionType();
+ RowType partitionType = tableSchema.logicalPartitionType();
fileStore = new TestFileStore(hasPk, partitionType);
- table = new TestFileStoreTable(fileStore, schema);
+ table = new TestFileStoreTable(fileStore, tableSchema);
}
@Parameterized.Parameters(name = "hasPk-{0}, partitioned-{1}")
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/TestFileStoreTable.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/TestFileStoreTable.java
index 6003d55..fb2727a 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/TestFileStoreTable.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/sink/TestFileStoreTable.java
@@ -21,7 +21,7 @@ package org.apache.flink.table.store.connector.sink;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.store.file.KeyValue;
import org.apache.flink.table.store.file.ValueKind;
-import org.apache.flink.table.store.file.schema.Schema;
+import org.apache.flink.table.store.file.schema.TableSchema;
import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.store.file.writer.RecordWriter;
import org.apache.flink.table.store.table.FileStoreTable;
@@ -39,11 +39,11 @@ import org.apache.flink.types.RowKind;
public class TestFileStoreTable implements FileStoreTable {
private final TestFileStore store;
- private final Schema schema;
+ private final TableSchema tableSchema;
- public TestFileStoreTable(TestFileStore store, Schema schema) {
+ public TestFileStoreTable(TestFileStore store, TableSchema tableSchema) {
this.store = store;
- this.schema = schema;
+ this.tableSchema = tableSchema;
}
@Override
@@ -52,8 +52,8 @@ public class TestFileStoreTable implements FileStoreTable {
}
@Override
- public Schema schema() {
- return schema;
+ public TableSchema schema() {
+ return tableSchema;
}
@Override
@@ -74,7 +74,7 @@ public class TestFileStoreTable implements FileStoreTable {
@Override
public TableWrite newWrite() {
return new AbstractTableWrite<KeyValue>(
- store.newWrite(), new SinkRecordConverter(2, schema)) {
+ store.newWrite(), new SinkRecordConverter(2, tableSchema)) {
@Override
protected void writeSinkRecord(SinkRecord record, RecordWriter<KeyValue> writer)
throws Exception {
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/FileSystemCatalog.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/FileSystemCatalog.java
index 027414b..f31a00d 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/FileSystemCatalog.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/FileSystemCatalog.java
@@ -37,8 +37,8 @@ import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.factories.Factory;
import org.apache.flink.table.factories.FactoryUtil;
-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.TableSchema;
import org.apache.flink.table.store.file.schema.UpdateSchema;
import org.apache.flink.util.function.RunnableWithException;
@@ -167,12 +167,12 @@ public class FileSystemCatalog extends TableStoreCatalog {
public CatalogBaseTable getTable(ObjectPath tablePath)
throws TableNotExistException, CatalogException {
Path path = tablePath(tablePath);
- Schema schema =
+ TableSchema tableSchema =
new SchemaManager(path)
.latest()
.orElseThrow(() -> new TableNotExistException(getName(), tablePath));
- CatalogTable table = schema.toUpdateSchema().toCatalogTable();
+ CatalogTable table = tableSchema.toUpdateSchema().toCatalogTable();
// add path to source and sink
table.getOptions().put(PATH.key(), path.toString());
return table;
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
index 01986ae..7132d9d 100644
--- 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
@@ -50,7 +50,7 @@ public class SchemaManager implements Serializable {
}
/** @return latest schema. */
- public Optional<Schema> latest() {
+ public Optional<TableSchema> latest() {
try {
return listVersionedFiles(schemaDirectory(), SCHEMA_PREFIX)
.reduce(Math::max)
@@ -61,7 +61,7 @@ public class SchemaManager implements Serializable {
}
/** List all schema. */
- public List<Schema> listAll() {
+ public List<TableSchema> listAll() {
return listAllIds().stream().map(this::schema).collect(Collectors.toList());
}
@@ -76,12 +76,12 @@ public class SchemaManager implements Serializable {
}
/** Create a new schema from {@link UpdateSchema}. */
- public Schema commitNewVersion(UpdateSchema updateSchema) throws Exception {
+ public TableSchema commitNewVersion(UpdateSchema updateSchema) throws Exception {
return commitNewVersion(Callable::call, updateSchema);
}
/** Create a new schema from {@link UpdateSchema}. */
- public Schema commitNewVersion(Lock lock, UpdateSchema updateSchema) throws Exception {
+ public TableSchema commitNewVersion(Lock lock, UpdateSchema updateSchema) throws Exception {
RowType rowType = updateSchema.rowType();
List<String> partitionKeys = updateSchema.partitionKeys();
List<String> primaryKeys = updateSchema.primaryKeys();
@@ -91,36 +91,36 @@ public class SchemaManager implements Serializable {
long id;
int highestFieldId;
List<DataField> fields;
- Optional<Schema> latest = latest();
+ Optional<TableSchema> latest = latest();
if (latest.isPresent()) {
- Schema oldSchema = latest.get();
+ TableSchema oldTableSchema = latest.get();
Preconditions.checkArgument(
- oldSchema.primaryKeys().equals(primaryKeys),
+ oldTableSchema.primaryKeys().equals(primaryKeys),
"Primary key modification is not supported, "
+ "old primaryKeys is %s, new primaryKeys is %s",
- oldSchema.primaryKeys(),
+ oldTableSchema.primaryKeys(),
primaryKeys);
if (!updateSchema
.rowType()
.getFields()
- .equals(oldSchema.logicalRowType().getFields())
- || !updateSchema.partitionKeys().equals(oldSchema.partitionKeys())) {
+ .equals(oldTableSchema.logicalRowType().getFields())
+ || !updateSchema.partitionKeys().equals(oldTableSchema.partitionKeys())) {
throw new UnsupportedOperationException(
"TODO: support update field types and partition keys. ");
}
- fields = oldSchema.fields();
- id = oldSchema.id() + 1;
- highestFieldId = oldSchema.highestFieldId();
+ fields = oldTableSchema.fields();
+ id = oldTableSchema.id() + 1;
+ highestFieldId = oldTableSchema.highestFieldId();
} else {
- fields = Schema.newFields(rowType);
- highestFieldId = Schema.currentHighestFieldId(fields);
+ fields = TableSchema.newFields(rowType);
+ highestFieldId = TableSchema.currentHighestFieldId(fields);
id = 0;
}
- Schema schema =
- new Schema(
+ TableSchema tableSchema =
+ new TableSchema(
id,
fields,
highestFieldId,
@@ -140,18 +140,19 @@ public class SchemaManager implements Serializable {
}
return MetaFileWriter.writeFileSafety(
- schemaPath, schema.toString());
+ schemaPath, tableSchema.toString());
});
if (success) {
- return schema;
+ return tableSchema;
}
}
}
/** Read schema for schema id. */
- public Schema schema(long id) {
+ public TableSchema schema(long id) {
try {
- return JsonSerdeUtil.fromJson(FileUtils.readFileUtf8(toSchemaPath(id)), Schema.class);
+ return JsonSerdeUtil.fromJson(
+ FileUtils.readFileUtf8(toSchemaPath(id)), TableSchema.class);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
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
index 089c804..e79a661 100644
--- 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
@@ -31,52 +31,52 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
-/** A {@link JsonSerializer} for {@link Schema}. */
-public class SchemaSerializer implements JsonSerializer<Schema> {
+/** A {@link JsonSerializer} for {@link TableSchema}. */
+public class SchemaSerializer implements JsonSerializer<TableSchema> {
public static final SchemaSerializer INSTANCE = new SchemaSerializer();
@Override
- public void serialize(Schema schema, JsonGenerator generator) throws IOException {
+ public void serialize(TableSchema tableSchema, JsonGenerator generator) throws IOException {
generator.writeStartObject();
- generator.writeNumberField("id", schema.id());
+ generator.writeNumberField("id", tableSchema.id());
generator.writeArrayFieldStart("fields");
- for (DataField field : schema.fields()) {
+ for (DataField field : tableSchema.fields()) {
DataFieldSerializer.INSTANCE.serialize(field, generator);
}
generator.writeEndArray();
- generator.writeNumberField("highestFieldId", schema.highestFieldId());
+ generator.writeNumberField("highestFieldId", tableSchema.highestFieldId());
generator.writeArrayFieldStart("partitionKeys");
- for (String partitionKey : schema.partitionKeys()) {
+ for (String partitionKey : tableSchema.partitionKeys()) {
generator.writeString(partitionKey);
}
generator.writeEndArray();
generator.writeArrayFieldStart("primaryKeys");
- for (String primaryKey : schema.primaryKeys()) {
+ for (String primaryKey : tableSchema.primaryKeys()) {
generator.writeString(primaryKey);
}
generator.writeEndArray();
generator.writeObjectFieldStart("options");
- for (Map.Entry<String, String> entry : schema.options().entrySet()) {
+ for (Map.Entry<String, String> entry : tableSchema.options().entrySet()) {
generator.writeStringField(entry.getKey(), entry.getValue());
}
generator.writeEndObject();
- if (!StringUtils.isNullOrWhitespaceOnly(schema.comment())) {
- generator.writeStringField("comment", schema.comment());
+ if (!StringUtils.isNullOrWhitespaceOnly(tableSchema.comment())) {
+ generator.writeStringField("comment", tableSchema.comment());
}
generator.writeEndObject();
}
@Override
- public Schema deserialize(JsonNode node) {
+ public TableSchema deserialize(JsonNode node) {
int id = node.get("id").asInt();
Iterator<JsonNode> fieldJsons = node.get("fields").elements();
@@ -113,6 +113,7 @@ public class SchemaSerializer implements JsonSerializer<Schema> {
comment = commentNode.asText();
}
- return new Schema(id, fields, highestFieldId, partitionKeys, primaryKeys, options, comment);
+ return new TableSchema(
+ id, fields, highestFieldId, partitionKeys, primaryKeys, options, comment);
}
}
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/TableSchema.java
similarity index 93%
rename from flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/Schema.java
rename to flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/TableSchema.java
index 253227b..89d7ce0 100644
--- 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/TableSchema.java
@@ -37,8 +37,8 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
-/** Schema of table store. */
-public class Schema implements Serializable {
+/** Schema of a table. */
+public class TableSchema implements Serializable {
private static final long serialVersionUID = 1L;
@@ -57,7 +57,7 @@ public class Schema implements Serializable {
private final String comment;
- public Schema(
+ public TableSchema(
long id,
List<DataField> fields,
int highestFieldId,
@@ -160,8 +160,8 @@ public class Schema implements Serializable {
.logicalType;
}
- public Schema copy(Map<String, String> newOptions) {
- return new Schema(
+ public TableSchema copy(Map<String, String> newOptions) {
+ return new TableSchema(
id, fields, highestFieldId, partitionKeys, primaryKeys, newOptions, comment);
}
@@ -178,12 +178,12 @@ public class Schema implements Serializable {
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)
- && Objects.equals(comment, schema.comment);
+ TableSchema tableSchema = (TableSchema) o;
+ return Objects.equals(fields, tableSchema.fields)
+ && Objects.equals(partitionKeys, tableSchema.partitionKeys)
+ && Objects.equals(primaryKeys, tableSchema.primaryKeys)
+ && Objects.equals(options, tableSchema.options)
+ && Objects.equals(comment, tableSchema.comment);
}
@Override
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
index 0d618e6..ee05522 100644
--- 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
@@ -21,8 +21,8 @@ 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.table.store.file.schema.TableSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
@@ -71,7 +71,7 @@ public class JsonSerdeUtil {
private static Module createTableStoreJacksonModule() {
SimpleModule module = new SimpleModule("Table store");
- registerJsonObjects(module, Schema.class, SchemaSerializer.INSTANCE);
+ registerJsonObjects(module, TableSchema.class, SchemaSerializer.INSTANCE);
registerJsonObjects(module, DataField.class, DataFieldSerializer.INSTANCE);
return module;
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
index a993ee2..d04f104 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.store.table;
import org.apache.flink.table.store.file.FileStore;
-import org.apache.flink.table.store.file.schema.Schema;
+import org.apache.flink.table.store.file.schema.TableSchema;
import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.store.table.sink.TableCommit;
import org.apache.flink.table.store.table.sink.TableCompact;
@@ -30,11 +30,11 @@ public abstract class AbstractFileStoreTable implements FileStoreTable {
private static final long serialVersionUID = 1L;
private final String name;
- protected final Schema schema;
+ protected final TableSchema tableSchema;
- public AbstractFileStoreTable(String name, Schema schema) {
+ public AbstractFileStoreTable(String name, TableSchema tableSchema) {
this.name = name;
- this.schema = schema;
+ this.tableSchema = tableSchema;
}
protected abstract FileStore<?> store();
@@ -45,8 +45,8 @@ public abstract class AbstractFileStoreTable implements FileStoreTable {
}
@Override
- public Schema schema() {
- return schema;
+ public TableSchema schema() {
+ return tableSchema;
}
@Override
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTable.java
index a30f1f6..b22fd6e 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTable.java
@@ -27,8 +27,8 @@ import org.apache.flink.table.store.file.data.DataFileMeta;
import org.apache.flink.table.store.file.operation.AppendOnlyFileStoreRead;
import org.apache.flink.table.store.file.operation.AppendOnlyFileStoreScan;
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.file.schema.TableSchema;
import org.apache.flink.table.store.file.utils.FileStorePathFactory;
import org.apache.flink.table.store.file.utils.RecordReader;
import org.apache.flink.table.store.file.writer.RecordWriter;
@@ -53,22 +53,23 @@ public class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
private final AppendOnlyFileStore store;
- AppendOnlyFileStoreTable(String name, SchemaManager schemaManager, Schema schema, String user) {
- super(name, schema);
+ AppendOnlyFileStoreTable(
+ String name, SchemaManager schemaManager, TableSchema tableSchema, String user) {
+ super(name, tableSchema);
this.store =
new AppendOnlyFileStore(
schemaManager,
- schema.id(),
- new FileStoreOptions(schema.options()),
+ tableSchema.id(),
+ new FileStoreOptions(tableSchema.options()),
user,
- schema.logicalPartitionType(),
- schema.logicalRowType());
+ tableSchema.logicalPartitionType(),
+ tableSchema.logicalRowType());
}
@Override
public TableScan newScan() {
AppendOnlyFileStoreScan scan = store.newScan();
- return new TableScan(scan, schema, store.pathFactory()) {
+ return new TableScan(scan, tableSchema, store.pathFactory()) {
@Override
protected SplitGenerator splitGenerator(FileStorePathFactory pathFactory) {
return new AppendOnlySplitGenerator(
@@ -109,7 +110,7 @@ public class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
@Override
public TableWrite newWrite() {
SinkRecordConverter recordConverter =
- new SinkRecordConverter(store.options().bucket(), schema);
+ new SinkRecordConverter(store.options().bucket(), tableSchema);
return new AbstractTableWrite<RowData>(store.newWrite(), recordConverter) {
@Override
protected void writeSinkRecord(SinkRecord record, RecordWriter<RowData> writer)
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java
index b41b5b7..299d4a1 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java
@@ -29,8 +29,8 @@ import org.apache.flink.table.store.file.mergetree.compact.MergeFunction;
import org.apache.flink.table.store.file.mergetree.compact.ValueCountMergeFunction;
import org.apache.flink.table.store.file.operation.KeyValueFileStoreScan;
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.file.schema.TableSchema;
import org.apache.flink.table.store.file.utils.FileStorePathFactory;
import org.apache.flink.table.store.file.utils.RecordReader;
import org.apache.flink.table.store.file.writer.RecordWriter;
@@ -56,8 +56,8 @@ public class ChangelogValueCountFileStoreTable extends AbstractFileStoreTable {
private final KeyValueFileStore store;
ChangelogValueCountFileStoreTable(
- String name, SchemaManager schemaManager, Schema schema, String user) {
- super(name, schema);
+ String name, SchemaManager schemaManager, TableSchema tableSchema, String user) {
+ super(name, tableSchema);
RowType countType =
RowType.of(
new LogicalType[] {new BigIntType(false)}, new String[] {"_VALUE_COUNT"});
@@ -65,11 +65,11 @@ public class ChangelogValueCountFileStoreTable extends AbstractFileStoreTable {
this.store =
new KeyValueFileStore(
schemaManager,
- schema.id(),
- new FileStoreOptions(schema.options()),
+ tableSchema.id(),
+ new FileStoreOptions(tableSchema.options()),
user,
- schema.logicalPartitionType(),
- schema.logicalRowType(),
+ tableSchema.logicalPartitionType(),
+ tableSchema.logicalRowType(),
countType,
mergeFunction);
}
@@ -77,7 +77,7 @@ public class ChangelogValueCountFileStoreTable extends AbstractFileStoreTable {
@Override
public TableScan newScan() {
KeyValueFileStoreScan scan = store.newScan();
- return new TableScan(scan, schema, store.pathFactory()) {
+ return new TableScan(scan, tableSchema, store.pathFactory()) {
@Override
protected SplitGenerator splitGenerator(FileStorePathFactory pathFactory) {
return new MergeTreeSplitGenerator(
@@ -127,7 +127,7 @@ public class ChangelogValueCountFileStoreTable extends AbstractFileStoreTable {
@Override
public TableWrite newWrite() {
SinkRecordConverter recordConverter =
- new SinkRecordConverter(store.options().bucket(), schema);
+ new SinkRecordConverter(store.options().bucket(), tableSchema);
return new AbstractTableWrite<KeyValue>(store.newWrite(), recordConverter) {
@Override
protected void writeSinkRecord(SinkRecord record, RecordWriter<KeyValue> writer)
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
index 8421c81..8b7f95c 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
@@ -31,8 +31,8 @@ import org.apache.flink.table.store.file.mergetree.compact.PartialUpdateMergeFun
import org.apache.flink.table.store.file.operation.KeyValueFileStoreScan;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.predicate.PredicateBuilder;
-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.TableSchema;
import org.apache.flink.table.store.file.utils.FileStorePathFactory;
import org.apache.flink.table.store.file.utils.RecordReader;
import org.apache.flink.table.store.file.writer.RecordWriter;
@@ -62,14 +62,14 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
private final KeyValueFileStore store;
ChangelogWithKeyFileStoreTable(
- String name, SchemaManager schemaManager, Schema schema, String user) {
- super(name, schema);
- RowType rowType = schema.logicalRowType();
+ String name, SchemaManager schemaManager, TableSchema tableSchema, String user) {
+ super(name, tableSchema);
+ RowType rowType = tableSchema.logicalRowType();
// add _KEY_ prefix to avoid conflict with value
RowType keyType =
new RowType(
- schema.logicalTrimmedPrimaryKeysType().getFields().stream()
+ tableSchema.logicalTrimmedPrimaryKeysType().getFields().stream()
.map(
f ->
new RowType.RowField(
@@ -78,7 +78,7 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
f.getDescription().orElse(null)))
.collect(Collectors.toList()));
- Configuration conf = Configuration.fromMap(schema.options());
+ Configuration conf = Configuration.fromMap(tableSchema.options());
FileStoreOptions.MergeEngine mergeEngine = conf.get(FileStoreOptions.MERGE_ENGINE);
MergeFunction mergeFunction;
@@ -101,10 +101,10 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
this.store =
new KeyValueFileStore(
schemaManager,
- schema.id(),
+ tableSchema.id(),
new FileStoreOptions(conf),
user,
- schema.logicalPartitionType(),
+ tableSchema.logicalPartitionType(),
keyType,
rowType,
mergeFunction);
@@ -113,7 +113,7 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
@Override
public TableScan newScan() {
KeyValueFileStoreScan scan = store.newScan();
- return new TableScan(scan, schema, store.pathFactory()) {
+ return new TableScan(scan, tableSchema, store.pathFactory()) {
@Override
protected SplitGenerator splitGenerator(FileStorePathFactory pathFactory) {
return new MergeTreeSplitGenerator(
@@ -133,9 +133,9 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
// file 2 will be ignored, and the final result will be key = a, value = 1 while the
// correct result is an empty set
// TODO support value filter
- List<String> trimmedPrimaryKeys = schema.trimmedPrimaryKeys();
+ List<String> trimmedPrimaryKeys = tableSchema.trimmedPrimaryKeys();
int[] fieldIdxToKeyIdx =
- schema.fields().stream()
+ tableSchema.fields().stream()
.mapToInt(f -> trimmedPrimaryKeys.indexOf(f.name()))
.toArray();
List<Predicate> keyFilters = new ArrayList<>();
@@ -176,7 +176,7 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
@Override
public TableWrite newWrite() {
SinkRecordConverter recordConverter =
- new SinkRecordConverter(store.options().bucket(), schema);
+ new SinkRecordConverter(store.options().bucket(), tableSchema);
return new AbstractTableWrite<KeyValue>(store.newWrite(), recordConverter) {
@Override
protected void writeSinkRecord(SinkRecord record, RecordWriter<KeyValue> writer)
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTable.java
index 82271f2..7b7d13c 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTable.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.store.table;
-import org.apache.flink.table.store.file.schema.Schema;
+import org.apache.flink.table.store.file.schema.TableSchema;
import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.store.table.sink.TableCommit;
import org.apache.flink.table.store.table.sink.TableCompact;
@@ -36,7 +36,7 @@ public interface FileStoreTable extends Serializable {
String name();
- Schema schema();
+ TableSchema schema();
SnapshotManager snapshotManager();
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTableFactory.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTableFactory.java
index 3f2d4c9..418706c 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTableFactory.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/FileStoreTableFactory.java
@@ -22,8 +22,8 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.table.store.file.FileStoreOptions;
import org.apache.flink.table.store.file.WriteMode;
-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.TableSchema;
import java.util.HashMap;
import java.util.Map;
@@ -40,7 +40,7 @@ public class FileStoreTableFactory {
Path tablePath = FileStoreOptions.path(conf);
String name = tablePath.getName();
SchemaManager schemaManager = new SchemaManager(tablePath);
- Schema schema =
+ TableSchema tableSchema =
schemaManager
.latest()
.orElseThrow(
@@ -51,17 +51,18 @@ public class FileStoreTableFactory {
+ ". Please create table first."));
// merge dynamic options into schema.options
- Map<String, String> newOptions = new HashMap<>(schema.options());
+ Map<String, String> newOptions = new HashMap<>(tableSchema.options());
newOptions.putAll(conf.toMap());
- schema = schema.copy(newOptions);
+ tableSchema = tableSchema.copy(newOptions);
if (conf.get(FileStoreOptions.WRITE_MODE) == WriteMode.APPEND_ONLY) {
- return new AppendOnlyFileStoreTable(name, schemaManager, schema, user);
+ return new AppendOnlyFileStoreTable(name, schemaManager, tableSchema, user);
} else {
- if (schema.primaryKeys().isEmpty()) {
- return new ChangelogValueCountFileStoreTable(name, schemaManager, schema, user);
+ if (tableSchema.primaryKeys().isEmpty()) {
+ return new ChangelogValueCountFileStoreTable(
+ name, schemaManager, tableSchema, user);
} else {
- return new ChangelogWithKeyFileStoreTable(name, schemaManager, schema, user);
+ return new ChangelogWithKeyFileStoreTable(name, schemaManager, tableSchema, user);
}
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/SinkRecordConverter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/SinkRecordConverter.java
index 3a2aed4..40b99af 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/SinkRecordConverter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/SinkRecordConverter.java
@@ -22,7 +22,7 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.runtime.generated.Projection;
import org.apache.flink.table.store.codegen.CodeGenUtils;
-import org.apache.flink.table.store.file.schema.Schema;
+import org.apache.flink.table.store.file.schema.TableSchema;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.RowKind;
@@ -44,13 +44,13 @@ public class SinkRecordConverter {
@Nullable private final Projection<RowData, BinaryRowData> logPkProjection;
- public SinkRecordConverter(int numBucket, Schema schema) {
+ public SinkRecordConverter(int numBucket, TableSchema tableSchema) {
this(
numBucket,
- schema.logicalRowType(),
- schema.projection(schema.partitionKeys()),
- schema.projection(schema.trimmedPrimaryKeys()),
- schema.projection(schema.primaryKeys()));
+ tableSchema.logicalRowType(),
+ tableSchema.projection(tableSchema.partitionKeys()),
+ tableSchema.projection(tableSchema.trimmedPrimaryKeys()),
+ tableSchema.projection(tableSchema.primaryKeys()));
}
public SinkRecordConverter(
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableScan.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableScan.java
index e858470..2ca147d 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableScan.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/TableScan.java
@@ -27,7 +27,7 @@ import org.apache.flink.table.store.file.predicate.CompoundPredicate;
import org.apache.flink.table.store.file.predicate.LeafPredicate;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.predicate.PredicateBuilder;
-import org.apache.flink.table.store.file.schema.Schema;
+import org.apache.flink.table.store.file.schema.TableSchema;
import org.apache.flink.table.store.file.utils.FileStorePathFactory;
import javax.annotation.Nullable;
@@ -41,12 +41,13 @@ import java.util.Optional;
public abstract class TableScan {
private final FileStoreScan scan;
- private final Schema schema;
+ private final TableSchema tableSchema;
private final FileStorePathFactory pathFactory;
- protected TableScan(FileStoreScan scan, Schema schema, FileStorePathFactory pathFactory) {
+ protected TableScan(
+ FileStoreScan scan, TableSchema tableSchema, FileStorePathFactory pathFactory) {
this.scan = scan;
- this.schema = schema;
+ this.tableSchema = tableSchema;
this.pathFactory = pathFactory;
}
@@ -63,9 +64,11 @@ public abstract class TableScan {
}
public TableScan withFilter(Predicate predicate) {
- List<String> partitionKeys = schema.partitionKeys();
+ List<String> partitionKeys = tableSchema.partitionKeys();
int[] fieldIdxToPartitionIdx =
- schema.fields().stream().mapToInt(f -> partitionKeys.indexOf(f.name())).toArray();
+ tableSchema.fields().stream()
+ .mapToInt(f -> partitionKeys.indexOf(f.name()))
+ .toArray();
List<Predicate> partitionFilters = new ArrayList<>();
List<Predicate> nonPartitionFilters = new ArrayList<>();
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
index 90084c0..963e5ad 100644
--- 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
@@ -83,9 +83,10 @@ public class SchemaManagerTest {
@Test
public void testCommitNewVersion() throws Exception {
- Schema schema = retryArtificialException(() -> manager.commitNewVersion(updateSchema));
+ TableSchema tableSchema =
+ retryArtificialException(() -> manager.commitNewVersion(updateSchema));
- Optional<Schema> latest = retryArtificialException(() -> manager.latest());
+ Optional<TableSchema> latest = retryArtificialException(() -> manager.latest());
List<DataField> fields =
Arrays.asList(
@@ -94,11 +95,11 @@ public class SchemaManagerTest {
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);
+ assertThat(tableSchema).isEqualTo(latest.get());
+ assertThat(tableSchema.fields()).isEqualTo(fields);
+ assertThat(tableSchema.partitionKeys()).isEqualTo(partitionKeys);
+ assertThat(tableSchema.primaryKeys()).isEqualTo(primaryKeys);
+ assertThat(tableSchema.options()).isEqualTo(options);
}
@Test
@@ -108,7 +109,7 @@ public class SchemaManagerTest {
UpdateSchema updateSchema =
new UpdateSchema(rowType, partitionKeys, primaryKeys, newOptions, "my_comment_2");
retryArtificialException(() -> manager.commitNewVersion(updateSchema));
- Optional<Schema> latest = retryArtificialException(() -> manager.latest());
+ Optional<TableSchema> latest = retryArtificialException(() -> manager.latest());
assertThat(latest.isPresent()).isTrue();
assertThat(latest.get().options()).isEqualTo(newOptions);
}
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/TableSchemaSerializationTest.java
similarity index 79%
rename from flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaSerializationTest.java
rename to flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/TableSchemaSerializationTest.java
index 4aae8ee..9627d08 100644
--- 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/TableSchemaSerializationTest.java
@@ -29,11 +29,11 @@ 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.apache.flink.table.store.file.schema.TableSchemaTest.newRowType;
import static org.assertj.core.api.Assertions.assertThat;
-/** Test for serialize {@link Schema}. */
-public class SchemaSerializationTest {
+/** Test for serialize {@link TableSchema}. */
+public class TableSchemaSerializationTest {
@Test
public void testSchema() {
@@ -50,11 +50,11 @@ public class SchemaSerializationTest {
options.put("option-1", "value-1");
options.put("option-2", "value-2");
- Schema schema =
- new Schema(1, fields, 10, partitionKeys, primaryKeys, options, "my_comment");
- String serialized = JsonSerdeUtil.toJson(schema);
+ TableSchema tableSchema =
+ new TableSchema(1, fields, 10, partitionKeys, primaryKeys, options, "my_comment");
+ String serialized = JsonSerdeUtil.toJson(tableSchema);
- Schema deserialized = JsonSerdeUtil.fromJson(serialized, Schema.class);
- assertThat(deserialized).isEqualTo(schema);
+ TableSchema deserialized = JsonSerdeUtil.fromJson(serialized, TableSchema.class);
+ assertThat(deserialized).isEqualTo(tableSchema);
}
}
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/TableSchemaTest.java
similarity index 90%
rename from flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/SchemaTest.java
rename to flink-table-store-core/src/test/java/org/apache/flink/table/store/file/schema/TableSchemaTest.java
index 3d3bea7..1bd05bd 100644
--- 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/TableSchemaTest.java
@@ -36,8 +36,8 @@ import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
-/** Test for {@link Schema}. */
-public class SchemaTest {
+/** Test for {@link TableSchema}. */
+public class TableSchemaTest {
@Test
public void testInvalidPrimaryKeys() {
@@ -52,7 +52,7 @@ public class SchemaTest {
Assertions.assertThrows(
IllegalStateException.class,
- () -> new Schema(1, fields, 10, partitionKeys, primaryKeys, options, ""));
+ () -> new TableSchema(1, fields, 10, partitionKeys, primaryKeys, options, ""));
}
@Test
@@ -61,7 +61,8 @@ public class SchemaTest {
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));
+ Assertions.assertThrows(
+ RuntimeException.class, () -> TableSchema.currentHighestFieldId(fields));
}
@Test
@@ -70,7 +71,7 @@ public class SchemaTest {
Arrays.asList(
new DataField(0, "f0", new AtomicDataType(new IntType())),
new DataField(20, "f1", new AtomicDataType(new IntType())));
- assertThat(Schema.currentHighestFieldId(fields)).isEqualTo(20);
+ assertThat(TableSchema.currentHighestFieldId(fields)).isEqualTo(20);
}
@Test
@@ -94,7 +95,7 @@ public class SchemaTest {
new DataField(5, "f3", new MultisetDataType(newRowType(6))),
new DataField(7, "f4", new MapDataType(newRowType(8), newRowType(9))));
- assertThat(Schema.newFields(type)).isEqualTo(fields);
+ assertThat(TableSchema.newFields(type)).isEqualTo(fields);
}
static RowType newLogicalRowType() {
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
index f15bf5f..92fc82a 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/AppendOnlyFileStoreTableTest.java
@@ -25,8 +25,8 @@ import org.apache.flink.table.store.file.FileStoreOptions;
import org.apache.flink.table.store.file.WriteMode;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.predicate.PredicateBuilder;
-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.TableSchema;
import org.apache.flink.table.store.file.schema.UpdateSchema;
import org.apache.flink.table.store.table.sink.TableWrite;
import org.apache.flink.table.store.table.source.Split;
@@ -167,7 +167,7 @@ public class AppendOnlyFileStoreTableTest extends FileStoreTableTestBase {
conf.set(FileStoreOptions.FILE_FORMAT, "avro");
conf.set(FileStoreOptions.WRITE_MODE, WriteMode.APPEND_ONLY);
SchemaManager schemaManager = new SchemaManager(tablePath);
- Schema schema =
+ TableSchema tableSchema =
schemaManager.commitNewVersion(
new UpdateSchema(
ROW_TYPE,
@@ -175,6 +175,7 @@ public class AppendOnlyFileStoreTableTest extends FileStoreTableTestBase {
Collections.emptyList(),
conf.toMap(),
""));
- return new AppendOnlyFileStoreTable(tablePath.getName(), schemaManager, schema, "user");
+ return new AppendOnlyFileStoreTable(
+ tablePath.getName(), schemaManager, tableSchema, "user");
}
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
index 2c1509f..6f2cccc 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTableTest.java
@@ -25,8 +25,8 @@ import org.apache.flink.table.store.file.FileStoreOptions;
import org.apache.flink.table.store.file.WriteMode;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.predicate.PredicateBuilder;
-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.TableSchema;
import org.apache.flink.table.store.file.schema.UpdateSchema;
import org.apache.flink.table.store.table.sink.TableWrite;
import org.apache.flink.table.store.table.source.Split;
@@ -170,7 +170,7 @@ public class ChangelogValueCountFileStoreTableTest extends FileStoreTableTestBas
conf.set(FileStoreOptions.FILE_FORMAT, "avro");
conf.set(FileStoreOptions.WRITE_MODE, WriteMode.CHANGE_LOG);
SchemaManager schemaManager = new SchemaManager(tablePath);
- Schema schema =
+ TableSchema tableSchema =
schemaManager.commitNewVersion(
new UpdateSchema(
ROW_TYPE,
@@ -179,6 +179,6 @@ public class ChangelogValueCountFileStoreTableTest extends FileStoreTableTestBas
conf.toMap(),
""));
return new ChangelogValueCountFileStoreTable(
- tablePath.getName(), schemaManager, schema, "user");
+ tablePath.getName(), schemaManager, tableSchema, "user");
}
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
index d248cc4..8cdad46 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTableTest.java
@@ -25,8 +25,8 @@ import org.apache.flink.table.store.file.FileStoreOptions;
import org.apache.flink.table.store.file.WriteMode;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.predicate.PredicateBuilder;
-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.TableSchema;
import org.apache.flink.table.store.file.schema.UpdateSchema;
import org.apache.flink.table.store.table.sink.TableWrite;
import org.apache.flink.table.store.table.source.Split;
@@ -169,7 +169,7 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
conf.set(FileStoreOptions.FILE_FORMAT, "avro");
conf.set(FileStoreOptions.WRITE_MODE, WriteMode.CHANGE_LOG);
SchemaManager schemaManager = new SchemaManager(tablePath);
- Schema schema =
+ TableSchema tableSchema =
schemaManager.commitNewVersion(
new UpdateSchema(
ROW_TYPE,
@@ -178,6 +178,6 @@ public class ChangelogWithKeyFileStoreTableTest extends FileStoreTableTestBase {
conf.toMap(),
""));
return new ChangelogWithKeyFileStoreTable(
- tablePath.getName(), schemaManager, schema, "user");
+ tablePath.getName(), schemaManager, tableSchema, "user");
}
}
diff --git a/flink-table-store-hive/src/main/java/org/apache/flink/table/store/hive/HiveSchema.java b/flink-table-store-hive/src/main/java/org/apache/flink/table/store/hive/HiveSchema.java
index 47b111f..e45a62f 100644
--- a/flink-table-store-hive/src/main/java/org/apache/flink/table/store/hive/HiveSchema.java
+++ b/flink-table-store-hive/src/main/java/org/apache/flink/table/store/hive/HiveSchema.java
@@ -20,8 +20,8 @@ package org.apache.flink.table.store.hive;
import org.apache.flink.core.fs.Path;
import org.apache.flink.table.store.file.schema.DataField;
-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.TableSchema;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
@@ -40,22 +40,24 @@ import java.util.stream.Collectors;
/** Column names, types and comments of a Hive table. */
public class HiveSchema {
- private final Schema schema;
+ private final TableSchema tableSchema;
- private HiveSchema(Schema schema) {
- this.schema = schema;
+ private HiveSchema(TableSchema tableSchema) {
+ this.tableSchema = tableSchema;
}
public List<String> fieldNames() {
- return schema.fieldNames();
+ return tableSchema.fieldNames();
}
public List<LogicalType> fieldTypes() {
- return schema.logicalRowType().getChildren();
+ return tableSchema.logicalRowType().getChildren();
}
public List<String> fieldComments() {
- return schema.fields().stream().map(DataField::description).collect(Collectors.toList());
+ return tableSchema.fields().stream()
+ .map(DataField::description)
+ .collect(Collectors.toList());
}
/** Extract {@link HiveSchema} from Hive serde properties. */
@@ -69,7 +71,7 @@ public class HiveSchema {
+ ". Currently Flink table store only supports external table for Hive "
+ "so location property must be set.");
}
- Schema schema =
+ TableSchema tableSchema =
new SchemaManager(new Path(location))
.latest()
.orElseThrow(
@@ -91,20 +93,20 @@ public class HiveSchema {
List<TypeInfo> typeInfos = TypeInfoUtils.getTypeInfosFromTypeString(columnTypes);
if (names.size() > 0 && typeInfos.size() > 0) {
- checkSchemaMatched(names, typeInfos, schema);
+ checkSchemaMatched(names, typeInfos, tableSchema);
}
}
- return new HiveSchema(schema);
+ return new HiveSchema(tableSchema);
}
private static void checkSchemaMatched(
- List<String> names, List<TypeInfo> typeInfos, Schema schema) {
+ List<String> names, List<TypeInfo> typeInfos, TableSchema tableSchema) {
List<String> ddlNames = new ArrayList<>(names);
List<TypeInfo> ddlTypeInfos = new ArrayList<>(typeInfos);
- List<String> schemaNames = schema.fieldNames();
+ List<String> schemaNames = tableSchema.fieldNames();
List<TypeInfo> schemaTypeInfos =
- schema.logicalRowType().getChildren().stream()
+ tableSchema.logicalRowType().getChildren().stream()
.map(HiveTypeUtils::logicalTypeToTypeInfo)
.collect(Collectors.toList());
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 f1bdb3c..726ca87 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
@@ -24,7 +24,7 @@ import org.apache.flink.table.store.SearchArgumentToPredicateConverter;
import org.apache.flink.table.store.TableStoreJobConf;
import org.apache.flink.table.store.file.FileStoreOptions;
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.TableSchema;
import org.apache.flink.table.store.table.FileStoreTable;
import org.apache.flink.table.store.table.FileStoreTableFactory;
import org.apache.flink.table.store.table.source.TableScan;
@@ -77,7 +77,7 @@ public class TableStoreInputFormat implements InputFormat<Void, RowDataContainer
return FileStoreTableFactory.create(conf, wrapper.getFileStoreUser());
}
- private Optional<Predicate> createPredicate(Schema schema, JobConf jobConf) {
+ private Optional<Predicate> createPredicate(TableSchema tableSchema, JobConf jobConf) {
String hiveFilter = jobConf.get(TableScanDesc.FILTER_EXPR_CONF_STR);
if (hiveFilter == null) {
return Optional.empty();
@@ -88,7 +88,7 @@ public class TableStoreInputFormat implements InputFormat<Void, RowDataContainer
SearchArgument sarg = ConvertAstToSearchArg.create(jobConf, exprNodeDesc);
SearchArgumentToPredicateConverter converter =
new SearchArgumentToPredicateConverter(
- sarg, schema.fieldNames(), schema.logicalRowType().getChildren());
+ sarg, tableSchema.fieldNames(), tableSchema.logicalRowType().getChildren());
return converter.convert();
}
}
diff --git a/flink-table-store-hive/src/test/java/org/apache/flink/table/store/hive/HiveSchemaTest.java b/flink-table-store-hive/src/test/java/org/apache/flink/table/store/hive/HiveTableSchemaTest.java
similarity index 99%
rename from flink-table-store-hive/src/test/java/org/apache/flink/table/store/hive/HiveSchemaTest.java
rename to flink-table-store-hive/src/test/java/org/apache/flink/table/store/hive/HiveTableSchemaTest.java
index 7df77e8..0b1864a 100644
--- a/flink-table-store-hive/src/test/java/org/apache/flink/table/store/hive/HiveSchemaTest.java
+++ b/flink-table-store-hive/src/test/java/org/apache/flink/table/store/hive/HiveTableSchemaTest.java
@@ -37,7 +37,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertThrows;
/** Tests for {@link HiveSchema}. */
-public class HiveSchemaTest {
+public class HiveTableSchemaTest {
private static final RowType ROW_TYPE =
new RowType(