From 811754a384a47d5da2e8c2e7684cc9acc49d664f Mon Sep 17 00:00:00 2001 From: huzheng Date: Thu, 25 Mar 2021 18:25:47 +0800 Subject: [PATCH 1/5] Flink: Support SQL primary key. --- .../main/java/org/apache/iceberg/Schema.java | 4 +- .../apache/iceberg/flink/FlinkCatalog.java | 6 +- .../apache/iceberg/flink/FlinkSchemaUtil.java | 54 +++- .../iceberg/flink/IcebergTableSink.java | 8 + .../apache/iceberg/flink/SimpleDataUtil.java | 11 +- .../iceberg/flink/TestChangeLogTable.java | 302 ++++++++++++++++++ .../iceberg/flink/TestFlinkCatalogTable.java | 39 +++ .../flink/source/BoundedTableFactory.java | 151 +++++++++ .../flink/source/BoundedTestSource.java | 2 +- .../flink/source/ChangeLogTableTestBase.java | 96 ++++++ .../flink/source/TestBoundedTableFactory.java | 81 +++++ .../org.apache.flink.table.factories.Factory | 16 + 12 files changed, 759 insertions(+), 11 deletions(-) create mode 100644 flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java create mode 100644 flink/src/test/java/org/apache/iceberg/flink/source/BoundedTableFactory.java create mode 100644 flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java create mode 100644 flink/src/test/java/org/apache/iceberg/flink/source/TestBoundedTableFactory.java create mode 100644 flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java index c02e17416235..06a48872edd0 100644 --- a/api/src/main/java/org/apache/iceberg/Schema.java +++ b/api/src/main/java/org/apache/iceberg/Schema.java @@ -88,8 +88,8 @@ public Schema(int schemaId, List columns, Set identifierFi this(schemaId, columns, null, identifierFieldIds); } - private Schema(int schemaId, List columns, Map aliases, - Set identifierFieldIds) { + public Schema(int schemaId, List columns, Map aliases, + Set identifierFieldIds) { this.schemaId = schemaId; this.struct = StructType.of(columns); this.aliasToId = aliases != null ? ImmutableBiMap.copyOf(aliases) : null; diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index de62e67805b5..2060dd0d439c 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -467,10 +467,6 @@ private static void validateFlinkTable(CatalogBaseTable table) { if (!schema.getWatermarkSpecs().isEmpty()) { throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet."); } - - if (schema.getPrimaryKey().isPresent()) { - throw new UnsupportedOperationException("Creating table with primary key is not supported yet."); - } } private static PartitionSpec toPartitionSpec(List partitionKeys, Schema icebergSchema) { @@ -536,7 +532,7 @@ private static void commitChanges(Table table, String setLocation, String setSna } static CatalogTable toCatalogTable(Table table) { - TableSchema schema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema())); + TableSchema schema = FlinkSchemaUtil.toSchema(table.schema()); List partitionKeys = toPartitionKeys(table.spec(), table.schema()); // NOTE: We can not create a IcebergCatalogTable extends CatalogTable, because Flink optimizer may use diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java index fa871e505129..6ad6e43b66d5 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java +++ b/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java @@ -19,12 +19,16 @@ package org.apache.iceberg.flink; +import java.util.List; +import java.util.Set; import org.apache.flink.table.api.TableSchema; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.utils.TypeConversions; import org.apache.iceberg.Schema; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; @@ -61,7 +65,22 @@ public static Schema convert(TableSchema schema) { RowType root = (RowType) schemaType; Type converted = root.accept(new FlinkTypeToType(root)); - return new Schema(converted.asStructType().fields()); + Schema iSchema = new Schema(converted.asStructType().fields()); + return freshIdentifierFieldIds(iSchema, schema); + } + + private static Schema freshIdentifierFieldIds(Schema iSchema, TableSchema schema) { + // Locate the identifier field id list. + Set identifierFieldIds = Sets.newHashSet(); + if (schema.getPrimaryKey().isPresent()) { + for (String column : schema.getPrimaryKey().get().getColumns()) { + Types.NestedField field = iSchema.findField(column); + Preconditions.checkNotNull(field, "Column %s does not found in schema %s", column, iSchema); + identifierFieldIds.add(field.fieldId()); + } + } + + return new Schema(iSchema.schemaId(), iSchema.asStruct().fields(), identifierFieldIds); } /** @@ -83,7 +102,8 @@ public static Schema convert(Schema baseSchema, TableSchema flinkSchema) { // reassign ids to match the base schema Schema schema = TypeUtil.reassignIds(new Schema(struct.fields()), baseSchema); // fix types that can't be represented in Flink (UUID) - return FlinkFixupTypes.fixup(schema, baseSchema); + Schema fixedSchema = FlinkFixupTypes.fixup(schema, baseSchema); + return freshIdentifierFieldIds(fixedSchema, flinkSchema); } /** @@ -121,4 +141,34 @@ public static TableSchema toSchema(RowType rowType) { } return builder.build(); } + + /** + * Convert a {@link Schema} to a {@link TableSchema}. + * + * @param schema iceberg schema to convert. + * @return Flink TableSchema. + */ + public static TableSchema toSchema(Schema schema) { + TableSchema.Builder builder = TableSchema.builder(); + + // Add columns. + for (RowType.RowField field : convert(schema).getFields()) { + builder.field(field.getName(), TypeConversions.fromLogicalToDataType(field.getType())); + } + + // Add primary key. + Set identifierFieldIds = schema.identifierFieldIds(); + if (!identifierFieldIds.isEmpty()) { + List columns = Lists.newArrayListWithExpectedSize(identifierFieldIds.size()); + for (Integer identifierFieldId : identifierFieldIds) { + String columnName = schema.findColumnName(identifierFieldId); + Preconditions.checkNotNull(columnName, "Cannot find field with id %s in schema %s", identifierFieldId, schema); + + columns.add(columnName); + } + builder.primaryKey(columns.toArray(new String[0])); + } + + return builder.build(); + } } diff --git a/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java b/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java index 80f74a91cf38..4cdd8bac7ef7 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java +++ b/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java @@ -19,8 +19,10 @@ package org.apache.iceberg.flink; +import java.util.List; import java.util.Map; import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.api.constraints.UniqueConstraint; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.sink.DataStreamSinkProvider; import org.apache.flink.table.connector.sink.DynamicTableSink; @@ -29,6 +31,7 @@ import org.apache.flink.types.RowKind; import org.apache.flink.util.Preconditions; import org.apache.iceberg.flink.sink.FlinkSink; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; public class IcebergTableSink implements DynamicTableSink, SupportsPartitioning, SupportsOverwrite { private final TableLoader tableLoader; @@ -52,9 +55,14 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) { Preconditions.checkState(!overwrite || context.isBounded(), "Unbounded data stream doesn't support overwrite operation."); + List equalityColumns = tableSchema.getPrimaryKey() + .map(UniqueConstraint::getColumns) + .orElseGet(ImmutableList::of); + return (DataStreamSinkProvider) dataStream -> FlinkSink.forRowData(dataStream) .tableLoader(tableLoader) .tableSchema(tableSchema) + .equalityFieldColumns(equalityColumns) .overwrite(overwrite) .build(); } diff --git a/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java b/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java index d1a75ccde14d..33202fee6084 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java +++ b/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java @@ -209,9 +209,18 @@ public static StructLikeSet expectedRowSet(Table table, Record... records) { } public static StructLikeSet actualRowSet(Table table, String... columns) throws IOException { + table.refresh(); + return actualRowSet(table, table.currentSnapshot().snapshotId(), columns); + } + + public static StructLikeSet actualRowSet(Table table, long snapshotId, String... columns) throws IOException { table.refresh(); StructLikeSet set = StructLikeSet.create(table.schema().asStruct()); - try (CloseableIterable reader = IcebergGenerics.read(table).select(columns).build()) { + try (CloseableIterable reader = IcebergGenerics + .read(table) + .useSnapshot(snapshotId) + .select(columns) + .build()) { reader.forEach(set::add); } return set; diff --git a/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java b/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java new file mode 100644 index 000000000000..55140aa8b121 --- /dev/null +++ b/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java @@ -0,0 +1,302 @@ +/* + * 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.iceberg.flink; + +import java.io.File; +import java.io.IOException; +import java.util.List; +import org.apache.flink.types.Row; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.flink.source.BoundedTableFactory; +import org.apache.iceberg.flink.source.ChangeLogTableTestBase; +import org.apache.iceberg.relocated.com.google.common.base.Joiner; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.util.StructLikeSet; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +@RunWith(Parameterized.class) +public class TestChangeLogTable extends ChangeLogTableTestBase { + private static final Configuration CONF = new Configuration(); + private static final String SOURCE_TABLE = "default_catalog.default_database.source_change_logs"; + + private static final String CATALOG_NAME = "test_catalog"; + private static final String DATABASE_NAME = "test_db"; + private static final String TABLE_NAME = "test_table"; + private static String warehouse; + + private final boolean partitioned; + + @Parameterized.Parameters(name = "PartitionedTable={0}") + public static Iterable parameters() { + return ImmutableList.of( + new Object[] {true}, + new Object[] {false} + ); + } + + public TestChangeLogTable(boolean partitioned) { + this.partitioned = partitioned; + } + + @BeforeClass + public static void createWarehouse() throws IOException { + File warehouseFile = TEMPORARY_FOLDER.newFolder(); + Assert.assertTrue("The warehouse should be deleted", warehouseFile.delete()); + warehouse = String.format("file:%s", warehouseFile); + } + + @Before + public void before() { + sql("CREATE CATALOG %s WITH ('type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')", + CATALOG_NAME, warehouse); + sql("USE CATALOG %s", CATALOG_NAME); + sql("CREATE DATABASE %s", DATABASE_NAME); + sql("USE %s", DATABASE_NAME); + } + + @After + @Override + public void clean() { + sql("DROP TABLE IF EXISTS %s", TABLE_NAME); + sql("DROP DATABASE IF EXISTS %s", DATABASE_NAME); + sql("DROP CATALOG IF EXISTS %s", CATALOG_NAME); + BoundedTableFactory.clearDataSets(); + } + + @Test + public void testSqlChangeLogOnIdKey() throws Exception { + List> inputRowsPerCheckpoint = ImmutableList.of( + ImmutableList.of( + row("+I", 1, "aaa"), + row("-D", 1, "aaa"), + row("+I", 1, "bbb"), + row("+I", 2, "aaa"), + row("-D", 2, "aaa"), + row("+I", 2, "bbb") + ), + ImmutableList.of( + row("-U", 2, "bbb"), + row("+U", 2, "ccc"), + row("-D", 2, "ccc"), + row("+I", 2, "ddd") + ), + ImmutableList.of( + row("-D", 1, "bbb"), + row("+I", 1, "ccc"), + row("-D", 1, "ccc"), + row("+I", 1, "ddd") + ) + ); + + List> expectedRecordsPerCheckpoint = ImmutableList.of( + ImmutableList.of(record(1, "bbb"), record(2, "bbb")), + ImmutableList.of(record(1, "bbb"), record(2, "ddd")), + ImmutableList.of(record(1, "ddd"), record(2, "ddd")) + ); + + testSqlChangeLog(TABLE_NAME, ImmutableList.of("id"), inputRowsPerCheckpoint, + expectedRecordsPerCheckpoint); + } + + @Test + public void testChangeLogOnDataKey() throws Exception { + List> elementsPerCheckpoint = ImmutableList.of( + ImmutableList.of( + row("+I", 1, "aaa"), + row("-D", 1, "aaa"), + row("+I", 2, "bbb"), + row("+I", 1, "bbb"), + row("+I", 2, "aaa") + ), + ImmutableList.of( + row("-U", 2, "aaa"), + row("+U", 1, "ccc"), + row("+I", 1, "aaa") + ), + ImmutableList.of( + row("-D", 1, "bbb"), + row("+I", 2, "aaa"), + row("+I", 2, "ccc") + ) + ); + + List> expectedRecords = ImmutableList.of( + ImmutableList.of(record(1, "bbb"), record(2, "aaa")), + ImmutableList.of(record(1, "aaa"), record(1, "bbb"), record(1, "ccc")), + ImmutableList.of(record(1, "aaa"), record(1, "ccc"), record(2, "aaa"), record(2, "ccc")) + ); + + testSqlChangeLog(TABLE_NAME, ImmutableList.of("data"), elementsPerCheckpoint, expectedRecords); + } + + @Test + public void testChangeLogOnIdDataKey() throws Exception { + List> elementsPerCheckpoint = ImmutableList.of( + ImmutableList.of( + row("+I", 1, "aaa"), + row("-D", 1, "aaa"), + row("+I", 2, "bbb"), + row("+I", 1, "bbb"), + row("+I", 2, "aaa") + ), + ImmutableList.of( + row("-U", 2, "aaa"), + row("+U", 1, "ccc"), + row("+I", 1, "aaa") + ), + ImmutableList.of( + row("-D", 1, "bbb"), + row("+I", 2, "aaa") + ) + ); + + List> expectedRecords = ImmutableList.of( + ImmutableList.of(record(1, "bbb"), record(2, "aaa"), record(2, "bbb")), + ImmutableList.of(record(1, "aaa"), record(1, "bbb"), record(1, "ccc"), record(2, "bbb")), + ImmutableList.of(record(1, "aaa"), record(1, "ccc"), record(2, "aaa"), record(2, "bbb")) + ); + + testSqlChangeLog(TABLE_NAME, ImmutableList.of("data", "id"), elementsPerCheckpoint, expectedRecords); + } + + @Test + public void testPureInsertOnIdKey() throws Exception { + List> elementsPerCheckpoint = ImmutableList.of( + ImmutableList.of( + row("+I", 1, "aaa"), + row("+I", 2, "bbb") + ), + ImmutableList.of( + row("+I", 3, "ccc"), + row("+I", 4, "ddd") + ), + ImmutableList.of( + row("+I", 5, "eee"), + row("+I", 6, "fff") + ) + ); + + List> expectedRecords = ImmutableList.of( + ImmutableList.of( + record(1, "aaa"), + record(2, "bbb") + ), + ImmutableList.of( + record(1, "aaa"), + record(2, "bbb"), + record(3, "ccc"), + record(4, "ddd") + ), + ImmutableList.of( + record(1, "aaa"), + record(2, "bbb"), + record(3, "ccc"), + record(4, "ddd"), + record(5, "eee"), + record(6, "fff") + ) + ); + + testSqlChangeLog(TABLE_NAME, ImmutableList.of("data"), elementsPerCheckpoint, expectedRecords); + } + + private Record record(int id, String data) { + return SimpleDataUtil.createRecord(id, data); + } + + private Table createTable(String tableName, List key, boolean isPartitioned) { + String partitionByCause = isPartitioned ? "PARTITIONED BY (data)" : ""; + sql("CREATE TABLE %s(id INT, data VARCHAR, PRIMARY KEY(%s) NOT ENFORCED) %s", + tableName, Joiner.on(',').join(key), partitionByCause); + + // Upgrade the iceberg table to format v2. + CatalogLoader loader = CatalogLoader.hadoop("my_catalog", CONF, ImmutableMap.of( + CatalogProperties.WAREHOUSE_LOCATION, warehouse + )); + Table table = loader.loadCatalog().loadTable(TableIdentifier.of(DATABASE_NAME, TABLE_NAME)); + TableOperations ops = ((BaseTable) table).operations(); + TableMetadata meta = ops.current(); + ops.commit(meta, meta.upgradeToFormatVersion(2)); + + return table; + } + + private void testSqlChangeLog(String tableName, + List key, + List> inputRowsPerCheckpoint, + List> expectedRecordsPerCheckpoint) throws Exception { + String dataId = BoundedTableFactory.registerDataSet(inputRowsPerCheckpoint); + sql("CREATE TABLE %s(id INT NOT NULL, data STRING NOT NULL)" + + " WITH ('connector'='BoundedSource', 'data-id'='%s')", SOURCE_TABLE, dataId); + + Assert.assertEquals("Should have the expected rows", + listJoin(inputRowsPerCheckpoint), + sql("SELECT * FROM %s", SOURCE_TABLE)); + + Table table = createTable(tableName, key, partitioned); + sql("INSERT INTO %s SELECT * FROM %s", tableName, SOURCE_TABLE); + + table.refresh(); + List snapshots = findValidSnapshots(table); + int expectedSnapshotNum = expectedRecordsPerCheckpoint.size(); + Assert.assertEquals("Should have the expected snapshot number", expectedSnapshotNum, snapshots.size()); + + for (int i = 0; i < expectedSnapshotNum; i++) { + long snapshotId = snapshots.get(i).snapshotId(); + List expectedRecords = expectedRecordsPerCheckpoint.get(i); + Assert.assertEquals("Should have the expected records for the checkpoint#" + i, + expectedRowSet(table, expectedRecords), actualRowSet(table, snapshotId)); + } + } + + private List findValidSnapshots(Table table) { + List validSnapshots = Lists.newArrayList(); + for (Snapshot snapshot : table.snapshots()) { + if (snapshot.allManifests().stream().anyMatch(m -> snapshot.snapshotId() == m.snapshotId())) { + validSnapshots.add(snapshot); + } + } + return validSnapshots; + } + + private static StructLikeSet expectedRowSet(Table table, List records) { + return SimpleDataUtil.expectedRowSet(table, records.toArray(new Record[0])); + } + + private static StructLikeSet actualRowSet(Table table, long snapshotId) throws IOException { + return SimpleDataUtil.actualRowSet(table, snapshotId, "*"); + } +} diff --git a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java index 62efd16160e7..7c691bd4cdad 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java +++ b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java @@ -22,6 +22,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.stream.Collectors; @@ -29,6 +30,7 @@ import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.TableSchema; import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.constraints.UniqueConstraint; import org.apache.flink.table.catalog.CatalogTable; import org.apache.flink.table.catalog.ObjectPath; import org.apache.flink.table.catalog.exceptions.TableNotExistException; @@ -45,9 +47,12 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.types.Types; import org.junit.After; import org.junit.Assert; @@ -121,6 +126,40 @@ public void testCreateTable() throws TableNotExistException { Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions()); } + @Test + public void testCreateTableWithPrimaryKey() throws Exception { + sql("CREATE TABLE tl(id BIGINT, data STRING, key STRING PRIMARY KEY NOT ENFORCED)"); + + Table table = table("tl"); + Assert.assertEquals("Should have the expected row key.", + Sets.newHashSet(table.schema().findField("key").fieldId()), + table.schema().identifierFieldIds()); + + CatalogTable catalogTable = catalogTable("tl"); + Optional uniqueConstraintOptional = catalogTable.getSchema().getPrimaryKey(); + Assert.assertTrue("Should have the expected unique constraint", uniqueConstraintOptional.isPresent()); + Assert.assertEquals("Should have the expected columns", + ImmutableList.of("key"), uniqueConstraintOptional.get().getColumns()); + } + + @Test + public void testCreateTableWithMultiColumnsInPrimaryKey() throws Exception { + sql("CREATE TABLE tl(id BIGINT, data STRING, CONSTRAINT pk_constraint PRIMARY KEY(data, id) NOT ENFORCED)"); + + Table table = table("tl"); + Assert.assertEquals("Should have the expected RowKey", + Sets.newHashSet( + table.schema().findField("id").fieldId(), + table.schema().findField("data").fieldId()), + table.schema().identifierFieldIds()); + + CatalogTable catalogTable = catalogTable("tl"); + Optional uniqueConstraintOptional = catalogTable.getSchema().getPrimaryKey(); + Assert.assertTrue("Should have the expected unique constraint", uniqueConstraintOptional.isPresent()); + Assert.assertEquals("Should have the expected columns", + ImmutableSet.of("data", "id"), ImmutableSet.copyOf(uniqueConstraintOptional.get().getColumns())); + } + @Test public void testCreateTableIfNotExists() { sql("CREATE TABLE tl(id BIGINT)"); diff --git a/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTableFactory.java b/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTableFactory.java new file mode 100644 index 000000000000..3be062aae2b9 --- /dev/null +++ b/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTableFactory.java @@ -0,0 +1,151 @@ +/* + * 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.iceberg.flink.source; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.source.SourceFunction; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.source.DataStreamScanProvider; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.util.DataFormatConverters; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.utils.TableSchemaUtils; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; +import org.apache.iceberg.flink.util.FlinkCompatibilityUtil; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; + +public class BoundedTableFactory implements DynamicTableSourceFactory { + private static final AtomicInteger DATA_SET_ID = new AtomicInteger(0); + private static final Map>> DATA_SETS = new HashMap<>(); + + private static final ConfigOption DATA_ID = ConfigOptions.key("data-id").stringType().noDefaultValue(); + + public static String registerDataSet(List> dataSet) { + String dataSetId = String.valueOf(DATA_SET_ID.incrementAndGet()); + DATA_SETS.put(dataSetId, dataSet); + return dataSetId; + } + + public static void clearDataSets() { + DATA_SETS.clear(); + } + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema()); + + Configuration configuration = Configuration.fromMap(context.getCatalogTable().getOptions()); + String dataId = configuration.getString(DATA_ID); + Preconditions.checkArgument(DATA_SETS.containsKey(dataId), + "data-id %s does not found in registered data set.", dataId); + + return new BoundedTableSource(DATA_SETS.get(dataId), tableSchema); + } + + @Override + public String factoryIdentifier() { + return "BoundedSource"; + } + + @Override + public Set> requiredOptions() { + return ImmutableSet.of(); + } + + @Override + public Set> optionalOptions() { + return ImmutableSet.of(DATA_ID); + } + + private static class BoundedTableSource implements ScanTableSource { + + private final List> elementsPerCheckpoint; + private final TableSchema tableSchema; + + private BoundedTableSource(List> elementsPerCheckpoint, TableSchema tableSchema) { + this.elementsPerCheckpoint = elementsPerCheckpoint; + this.tableSchema = tableSchema; + } + + private BoundedTableSource(BoundedTableSource toCopy) { + this.elementsPerCheckpoint = toCopy.elementsPerCheckpoint; + this.tableSchema = toCopy.tableSchema; + } + + @Override + public ChangelogMode getChangelogMode() { + return ChangelogMode.newBuilder() + .addContainedKind(RowKind.INSERT) + .addContainedKind(RowKind.DELETE) + .addContainedKind(RowKind.UPDATE_BEFORE) + .addContainedKind(RowKind.UPDATE_AFTER) + .build(); + } + + @Override + public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) { + return new DataStreamScanProvider() { + @Override + public DataStream produceDataStream(StreamExecutionEnvironment env) { + SourceFunction source = new BoundedTestSource<>(elementsPerCheckpoint); + + RowType rowType = (RowType) tableSchema.toRowDataType().getLogicalType(); + // Converter to convert the Row to RowData. + DataFormatConverters.RowConverter rowConverter = new DataFormatConverters + .RowConverter(tableSchema.getFieldDataTypes()); + + return env.addSource(source, new RowTypeInfo(tableSchema.getFieldTypes())) + .map(rowConverter::toInternal, FlinkCompatibilityUtil.toTypeInfo(rowType)); + } + + @Override + public boolean isBounded() { + return true; + } + }; + } + + @Override + public DynamicTableSource copy() { + return new BoundedTableSource(this); + } + + @Override + public String asSummaryString() { + return "Bounded test table source"; + } + } +} diff --git a/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTestSource.java b/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTestSource.java index 1ae04ab6d741..13da8d65eef7 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTestSource.java +++ b/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTestSource.java @@ -23,7 +23,7 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; -import org.apache.flink.runtime.state.CheckpointListener; +import org.apache.flink.api.common.state.CheckpointListener; import org.apache.flink.streaming.api.functions.source.SourceFunction; /** diff --git a/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java b/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java new file mode 100644 index 000000000000..d7d27ca5a74e --- /dev/null +++ b/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java @@ -0,0 +1,96 @@ +/* + * 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.iceberg.flink.source; + +import java.util.List; +import java.util.Map; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; +import org.apache.iceberg.flink.FlinkTestBase; +import org.apache.iceberg.flink.MiniClusterResource; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.junit.After; +import org.junit.Rule; +import org.junit.rules.TestName; + +public class ChangeLogTableTestBase extends FlinkTestBase { + private volatile TableEnvironment tEnv = null; + + @Rule + public TestName name = new TestName(); + + @After + public void clean() { + sql("DROP TABLE IF EXISTS %s", name.getMethodName()); + BoundedTableFactory.clearDataSets(); + } + + @Override + protected TableEnvironment getTableEnv() { + if (tEnv == null) { + synchronized (this) { + if (tEnv == null) { + EnvironmentSettings settings = EnvironmentSettings + .newInstance() + .useBlinkPlanner() + .inStreamingMode() + .build(); + + StreamExecutionEnvironment env = StreamExecutionEnvironment + .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG) + .enableCheckpointing(400) + .setMaxParallelism(1) + .setParallelism(1); + + tEnv = StreamTableEnvironment.create(env, settings); + } + } + } + return tEnv; + } + + private static final Map ROW_KIND_MAP = ImmutableMap.of( + "+I", RowKind.INSERT, + "-D", RowKind.DELETE, + "-U", RowKind.UPDATE_BEFORE, + "+U", RowKind.UPDATE_AFTER); + + protected Row row(String rowKind, int id, String data) { + RowKind kind = ROW_KIND_MAP.get(rowKind); + if (kind == null) { + throw new IllegalArgumentException("Unknown row kind: " + rowKind); + } + + return Row.ofKind(kind, id, data); + } + + protected static List listJoin(List> lists) { + List result = Lists.newArrayList(); + for (List list : lists) { + result.addAll(list); + } + return result; + } +} diff --git a/flink/src/test/java/org/apache/iceberg/flink/source/TestBoundedTableFactory.java b/flink/src/test/java/org/apache/iceberg/flink/source/TestBoundedTableFactory.java new file mode 100644 index 000000000000..fc9a8963d9e2 --- /dev/null +++ b/flink/src/test/java/org/apache/iceberg/flink/source/TestBoundedTableFactory.java @@ -0,0 +1,81 @@ +/* + * 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.iceberg.flink.source; + +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; +import org.apache.flink.types.Row; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.Streams; +import org.junit.Assert; +import org.junit.Test; + +public class TestBoundedTableFactory extends ChangeLogTableTestBase { + + @Test + public void testEmptyDataSet() { + String table = name.getMethodName(); + List> emptyDataSet = ImmutableList.of(); + + String dataId = BoundedTableFactory.registerDataSet(emptyDataSet); + sql("CREATE TABLE %s(id INT, data STRING) WITH ('connector'='BoundedSource', 'data-id'='%s')", table, dataId); + + Assert.assertEquals("Should have caught empty change log set.", ImmutableList.of(), + sql("SELECT * FROM %s", table)); + } + + @Test + public void testBoundedTableFactory() { + String table = name.getMethodName(); + List> dataSet = ImmutableList.of( + ImmutableList.of( + row("+I", 1, "aaa"), + row("-D", 1, "aaa"), + row("+I", 1, "bbb"), + row("+I", 2, "aaa"), + row("-D", 2, "aaa"), + row("+I", 2, "bbb") + ), + ImmutableList.of( + row("-U", 2, "bbb"), + row("+U", 2, "ccc"), + row("-D", 2, "ccc"), + row("+I", 2, "ddd") + ), + ImmutableList.of( + row("-D", 1, "bbb"), + row("+I", 1, "ccc"), + row("-D", 1, "ccc"), + row("+I", 1, "ddd") + ) + ); + + String dataId = BoundedTableFactory.registerDataSet(dataSet); + sql("CREATE TABLE %s(id INT, data STRING) WITH ('connector'='BoundedSource', 'data-id'='%s')", table, dataId); + + List rowSet = dataSet.stream().flatMap(Streams::stream).collect(Collectors.toList()); + Assert.assertEquals("Should have the expected change log events.", rowSet, sql("SELECT * FROM %s", table)); + + Assert.assertEquals("Should have the expected change log events", + rowSet.stream().filter(r -> Objects.equals(r.getField(1), "aaa")).collect(Collectors.toList()), + sql("SELECT * FROM %s WHERE data='aaa'", table)); + } +} diff --git a/flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory new file mode 100644 index 000000000000..47a3c94aa991 --- /dev/null +++ b/flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory @@ -0,0 +1,16 @@ +# 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. + +org.apache.iceberg.flink.source.BoundedTableFactory From d9321a5a5e4216f1fbd0f6c144da68fec2a06a51 Mon Sep 17 00:00:00 2001 From: huzheng Date: Thu, 13 May 2021 09:59:06 +0800 Subject: [PATCH 2/5] Addressing comments. --- .../main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java index 6ad6e43b66d5..0827b21786c1 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java +++ b/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java @@ -75,7 +75,8 @@ private static Schema freshIdentifierFieldIds(Schema iSchema, TableSchema schema if (schema.getPrimaryKey().isPresent()) { for (String column : schema.getPrimaryKey().get().getColumns()) { Types.NestedField field = iSchema.findField(column); - Preconditions.checkNotNull(field, "Column %s does not found in schema %s", column, iSchema); + Preconditions.checkNotNull(field, + "Cannot find field ID for the primary key column %s in schema %s", column, iSchema); identifierFieldIds.add(field.fieldId()); } } From c93e157e8a3e62e472bbb9dd5107505157c83111 Mon Sep 17 00:00:00 2001 From: huzheng Date: Thu, 13 May 2021 14:11:45 +0800 Subject: [PATCH 3/5] Address comments. --- .../apache/iceberg/flink/SimpleDataUtil.java | 7 +- .../iceberg/flink/TestChangeLogTable.java | 82 +++++++++---------- .../iceberg/flink/TestFlinkSchemaUtil.java | 35 ++++++++ .../flink/source/ChangeLogTableTestBase.java | 25 +++--- .../flink/source/TestBoundedTableFactory.java | 28 +++---- 5 files changed, 105 insertions(+), 72 deletions(-) diff --git a/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java b/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java index 33202fee6084..5e1372da925d 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java +++ b/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java @@ -209,16 +209,15 @@ public static StructLikeSet expectedRowSet(Table table, Record... records) { } public static StructLikeSet actualRowSet(Table table, String... columns) throws IOException { - table.refresh(); - return actualRowSet(table, table.currentSnapshot().snapshotId(), columns); + return actualRowSet(table, null, columns); } - public static StructLikeSet actualRowSet(Table table, long snapshotId, String... columns) throws IOException { + public static StructLikeSet actualRowSet(Table table, Long snapshotId, String... columns) throws IOException { table.refresh(); StructLikeSet set = StructLikeSet.create(table.schema().asStruct()); try (CloseableIterable reader = IcebergGenerics .read(table) - .useSnapshot(snapshotId) + .useSnapshot(snapshotId == null ? table.currentSnapshot().snapshotId() : snapshotId) .select(columns) .build()) { reader.forEach(set::add); diff --git a/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java b/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java index 55140aa8b121..1f0e8b23989f 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java +++ b/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java @@ -100,24 +100,24 @@ public void clean() { public void testSqlChangeLogOnIdKey() throws Exception { List> inputRowsPerCheckpoint = ImmutableList.of( ImmutableList.of( - row("+I", 1, "aaa"), - row("-D", 1, "aaa"), - row("+I", 1, "bbb"), - row("+I", 2, "aaa"), - row("-D", 2, "aaa"), - row("+I", 2, "bbb") + insertRow(1, "aaa"), + deleteRow(1, "aaa"), + insertRow(1, "bbb"), + insertRow(2, "aaa"), + deleteRow(2, "aaa"), + insertRow(2, "bbb") ), ImmutableList.of( - row("-U", 2, "bbb"), - row("+U", 2, "ccc"), - row("-D", 2, "ccc"), - row("+I", 2, "ddd") + updateBeforeRow(2, "bbb"), + updateAfterRow(2, "ccc"), + deleteRow(2, "ccc"), + insertRow(2, "ddd") ), ImmutableList.of( - row("-D", 1, "bbb"), - row("+I", 1, "ccc"), - row("-D", 1, "ccc"), - row("+I", 1, "ddd") + deleteRow(1, "bbb"), + insertRow(1, "ccc"), + deleteRow(1, "ccc"), + insertRow(1, "ddd") ) ); @@ -135,21 +135,21 @@ public void testSqlChangeLogOnIdKey() throws Exception { public void testChangeLogOnDataKey() throws Exception { List> elementsPerCheckpoint = ImmutableList.of( ImmutableList.of( - row("+I", 1, "aaa"), - row("-D", 1, "aaa"), - row("+I", 2, "bbb"), - row("+I", 1, "bbb"), - row("+I", 2, "aaa") + insertRow(1, "aaa"), + deleteRow(1, "aaa"), + insertRow(2, "bbb"), + insertRow(1, "bbb"), + insertRow(2, "aaa") ), ImmutableList.of( - row("-U", 2, "aaa"), - row("+U", 1, "ccc"), - row("+I", 1, "aaa") + updateBeforeRow(2, "aaa"), + updateAfterRow(1, "ccc"), + insertRow(1, "aaa") ), ImmutableList.of( - row("-D", 1, "bbb"), - row("+I", 2, "aaa"), - row("+I", 2, "ccc") + deleteRow(1, "bbb"), + insertRow(2, "aaa"), + insertRow(2, "ccc") ) ); @@ -166,20 +166,20 @@ public void testChangeLogOnDataKey() throws Exception { public void testChangeLogOnIdDataKey() throws Exception { List> elementsPerCheckpoint = ImmutableList.of( ImmutableList.of( - row("+I", 1, "aaa"), - row("-D", 1, "aaa"), - row("+I", 2, "bbb"), - row("+I", 1, "bbb"), - row("+I", 2, "aaa") + insertRow(1, "aaa"), + deleteRow(1, "aaa"), + insertRow(2, "bbb"), + insertRow(1, "bbb"), + insertRow(2, "aaa") ), ImmutableList.of( - row("-U", 2, "aaa"), - row("+U", 1, "ccc"), - row("+I", 1, "aaa") + updateBeforeRow(2, "aaa"), + updateAfterRow(1, "ccc"), + insertRow(1, "aaa") ), ImmutableList.of( - row("-D", 1, "bbb"), - row("+I", 2, "aaa") + deleteRow(1, "bbb"), + insertRow(2, "aaa") ) ); @@ -196,16 +196,16 @@ public void testChangeLogOnIdDataKey() throws Exception { public void testPureInsertOnIdKey() throws Exception { List> elementsPerCheckpoint = ImmutableList.of( ImmutableList.of( - row("+I", 1, "aaa"), - row("+I", 2, "bbb") + insertRow(1, "aaa"), + insertRow(2, "bbb") ), ImmutableList.of( - row("+I", 3, "ccc"), - row("+I", 4, "ddd") + insertRow(3, "ccc"), + insertRow(4, "ddd") ), ImmutableList.of( - row("+I", 5, "eee"), - row("+I", 6, "fff") + insertRow(5, "eee"), + insertRow(6, "fff") ) ); diff --git a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java index 3272ed65d35d..c87095716ca0 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java +++ b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java @@ -21,6 +21,7 @@ import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.types.logical.BinaryType; import org.apache.flink.table.types.logical.CharType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; @@ -30,7 +31,11 @@ import org.apache.flink.table.types.logical.TimestampType; import org.apache.flink.table.types.logical.VarBinaryType; import org.apache.flink.table.types.logical.VarCharType; +import org.apache.iceberg.AssertHelpers; import org.apache.iceberg.Schema; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.junit.Assert; @@ -270,4 +275,34 @@ private void checkInconsistentType( Types.StructType.of(Types.NestedField.optional(0, "f0", icebergExpectedType)), FlinkSchemaUtil.convert(FlinkSchemaUtil.toSchema(RowType.of(flinkType))).asStruct()); } + + @Test + public void testConvertFlinkSchemaWithPrimaryKeys() { + Schema iSchema = new Schema( + Lists.newArrayList( + Types.NestedField.required(1, "int", Types.IntegerType.get()), + Types.NestedField.required(2, "string", Types.StringType.get()) + ), + Sets.newHashSet(1, 2) + ); + + TableSchema tableSchema = FlinkSchemaUtil.toSchema(iSchema); + Assert.assertTrue(tableSchema.getPrimaryKey().isPresent()); + Assert.assertEquals(ImmutableSet.of("int", "string"), + ImmutableSet.copyOf(tableSchema.getPrimaryKey().get().getColumns())); + } + + @Test + public void testConvertFlinkSchemaWithNestedColumnInPrimaryKeys() { + Schema iSchema = new Schema( + Lists.newArrayList(Types.NestedField.required(1, "struct", + Types.StructType.of(Types.NestedField.required(2, "inner", Types.LongType.get()))) + ), + Sets.newHashSet(1, 2) + ); + AssertHelpers.assertThrows("Does not support the nested columns in flink schema's primary keys", + ValidationException.class, + "Column 'struct.inner' does not exist", + () -> FlinkSchemaUtil.toSchema(iSchema)); + } } diff --git a/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java b/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java index d7d27ca5a74e..5b67813c4ecc 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java +++ b/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java @@ -20,7 +20,6 @@ package org.apache.iceberg.flink.source; import java.util.List; -import java.util.Map; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.TableEnvironment; @@ -29,7 +28,6 @@ import org.apache.flink.types.RowKind; import org.apache.iceberg.flink.FlinkTestBase; import org.apache.iceberg.flink.MiniClusterResource; -import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.junit.After; import org.junit.Rule; @@ -71,19 +69,20 @@ protected TableEnvironment getTableEnv() { return tEnv; } - private static final Map ROW_KIND_MAP = ImmutableMap.of( - "+I", RowKind.INSERT, - "-D", RowKind.DELETE, - "-U", RowKind.UPDATE_BEFORE, - "+U", RowKind.UPDATE_AFTER); + protected static Row insertRow(Object... values) { + return Row.ofKind(RowKind.INSERT, values); + } - protected Row row(String rowKind, int id, String data) { - RowKind kind = ROW_KIND_MAP.get(rowKind); - if (kind == null) { - throw new IllegalArgumentException("Unknown row kind: " + rowKind); - } + protected static Row deleteRow(Object... values) { + return Row.ofKind(RowKind.DELETE, values); + } + + protected static Row updateBeforeRow(Object... values) { + return Row.ofKind(RowKind.UPDATE_BEFORE, values); + } - return Row.ofKind(kind, id, data); + protected static Row updateAfterRow(Object... values) { + return Row.ofKind(RowKind.UPDATE_AFTER, values); } protected static List listJoin(List> lists) { diff --git a/flink/src/test/java/org/apache/iceberg/flink/source/TestBoundedTableFactory.java b/flink/src/test/java/org/apache/iceberg/flink/source/TestBoundedTableFactory.java index fc9a8963d9e2..d163b84c09c6 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/source/TestBoundedTableFactory.java +++ b/flink/src/test/java/org/apache/iceberg/flink/source/TestBoundedTableFactory.java @@ -47,24 +47,24 @@ public void testBoundedTableFactory() { String table = name.getMethodName(); List> dataSet = ImmutableList.of( ImmutableList.of( - row("+I", 1, "aaa"), - row("-D", 1, "aaa"), - row("+I", 1, "bbb"), - row("+I", 2, "aaa"), - row("-D", 2, "aaa"), - row("+I", 2, "bbb") + insertRow(1, "aaa"), + deleteRow(1, "aaa"), + insertRow(1, "bbb"), + insertRow(2, "aaa"), + deleteRow(2, "aaa"), + insertRow(2, "bbb") ), ImmutableList.of( - row("-U", 2, "bbb"), - row("+U", 2, "ccc"), - row("-D", 2, "ccc"), - row("+I", 2, "ddd") + updateBeforeRow(2, "bbb"), + updateAfterRow(2, "ccc"), + deleteRow(2, "ccc"), + insertRow(2, "ddd") ), ImmutableList.of( - row("-D", 1, "bbb"), - row("+I", 1, "ccc"), - row("-D", 1, "ccc"), - row("+I", 1, "ddd") + deleteRow(1, "bbb"), + insertRow(1, "ccc"), + deleteRow(1, "ccc"), + insertRow(1, "ddd") ) ); From fef0bcaa74edd79b09b692faf22e31ca6387eaa6 Mon Sep 17 00:00:00 2001 From: huzheng Date: Thu, 13 May 2021 16:19:39 +0800 Subject: [PATCH 4/5] Add unit test --- .../iceberg/flink/TestFlinkSchemaUtil.java | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java index c87095716ca0..314909bb2cff 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java +++ b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java @@ -276,6 +276,26 @@ private void checkInconsistentType( FlinkSchemaUtil.convert(FlinkSchemaUtil.toSchema(RowType.of(flinkType))).asStruct()); } + @Test + public void testConvertFlinkSchemaBaseOnIcebergSchema() { + Schema baseSchema = new Schema( + Lists.newArrayList( + Types.NestedField.required(101, "int", Types.IntegerType.get()), + Types.NestedField.optional(102, "string", Types.StringType.get()) + ), + Sets.newHashSet(101, 102) + ); + + TableSchema flinkSchema = TableSchema.builder() + .field("int", DataTypes.INT().notNull()) + .field("string", DataTypes.STRING().nullable()) + .primaryKey("int") + .build(); + Schema convertedSchema = FlinkSchemaUtil.convert(baseSchema, flinkSchema); + Assert.assertEquals(baseSchema.asStruct(), convertedSchema.asStruct()); + Assert.assertEquals(ImmutableSet.of(101), convertedSchema.identifierFieldIds()); + } + @Test public void testConvertFlinkSchemaWithPrimaryKeys() { Schema iSchema = new Schema( From 1d92ab8a93f2eda9237443a676c7a649edd13d00 Mon Sep 17 00:00:00 2001 From: huzheng Date: Fri, 14 May 2021 14:55:19 +0800 Subject: [PATCH 5/5] Addressing the comments from Steven. --- .../org/apache/iceberg/flink/TestChangeLogTable.java | 4 ++++ .../org/apache/iceberg/flink/TestFlinkSchemaUtil.java | 8 ++++---- .../apache/iceberg/flink/source/BoundedTestSource.java | 6 ++++++ .../iceberg/flink/source/ChangeLogTableTestBase.java | 10 ++++------ 4 files changed, 18 insertions(+), 10 deletions(-) diff --git a/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java b/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java index 1f0e8b23989f..d44f45ab52fd 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java +++ b/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java @@ -47,6 +47,10 @@ import org.junit.runner.RunWith; import org.junit.runners.Parameterized; +/** + * In this test case, we mainly cover the impact of primary key selection, multiple operations within a single + * transaction, and multiple operations between different txn on the correctness of the data. + */ @RunWith(Parameterized.class) public class TestChangeLogTable extends ChangeLogTableTestBase { private static final Configuration CONF = new Configuration(); diff --git a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java index 314909bb2cff..460869ae8c3f 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java +++ b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java @@ -298,7 +298,7 @@ public void testConvertFlinkSchemaBaseOnIcebergSchema() { @Test public void testConvertFlinkSchemaWithPrimaryKeys() { - Schema iSchema = new Schema( + Schema icebergSchema = new Schema( Lists.newArrayList( Types.NestedField.required(1, "int", Types.IntegerType.get()), Types.NestedField.required(2, "string", Types.StringType.get()) @@ -306,7 +306,7 @@ public void testConvertFlinkSchemaWithPrimaryKeys() { Sets.newHashSet(1, 2) ); - TableSchema tableSchema = FlinkSchemaUtil.toSchema(iSchema); + TableSchema tableSchema = FlinkSchemaUtil.toSchema(icebergSchema); Assert.assertTrue(tableSchema.getPrimaryKey().isPresent()); Assert.assertEquals(ImmutableSet.of("int", "string"), ImmutableSet.copyOf(tableSchema.getPrimaryKey().get().getColumns())); @@ -314,7 +314,7 @@ public void testConvertFlinkSchemaWithPrimaryKeys() { @Test public void testConvertFlinkSchemaWithNestedColumnInPrimaryKeys() { - Schema iSchema = new Schema( + Schema icebergSchema = new Schema( Lists.newArrayList(Types.NestedField.required(1, "struct", Types.StructType.of(Types.NestedField.required(2, "inner", Types.LongType.get()))) ), @@ -323,6 +323,6 @@ public void testConvertFlinkSchemaWithNestedColumnInPrimaryKeys() { AssertHelpers.assertThrows("Does not support the nested columns in flink schema's primary keys", ValidationException.class, "Column 'struct.inner' does not exist", - () -> FlinkSchemaUtil.toSchema(iSchema)); + () -> FlinkSchemaUtil.toSchema(icebergSchema)); } } diff --git a/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTestSource.java b/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTestSource.java index 13da8d65eef7..6f6712dea74e 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTestSource.java +++ b/flink/src/test/java/org/apache/iceberg/flink/source/BoundedTestSource.java @@ -63,6 +63,12 @@ public void run(SourceContext ctx) throws Exception { final int checkpointToAwait; synchronized (ctx.getCheckpointLock()) { + // Let's say checkpointToAwait = numCheckpointsComplete.get() + delta, in fact the value of delta should not + // affect the final table records because we only need to make sure that there will be exactly + // elementsPerCheckpoint.size() checkpoints to emit each records buffer from the original elementsPerCheckpoint. + // Even if the checkpoints that emitted results are not continuous, the correctness of the data should not be + // affected in the end. Setting the delta to be 2 is introducing the variable that produce un-continuous + // checkpoints that emit the records buffer from elementsPerCheckpoints. checkpointToAwait = numCheckpointsComplete.get() + 2; for (T element : elementsPerCheckpoint.get(checkpoint)) { ctx.collect(element); diff --git a/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java b/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java index 5b67813c4ecc..a445e7eb06ce 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java +++ b/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java @@ -20,6 +20,7 @@ package org.apache.iceberg.flink.source; import java.util.List; +import java.util.stream.Collectors; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.TableEnvironment; @@ -28,7 +29,6 @@ import org.apache.flink.types.RowKind; import org.apache.iceberg.flink.FlinkTestBase; import org.apache.iceberg.flink.MiniClusterResource; -import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.junit.After; import org.junit.Rule; import org.junit.rules.TestName; @@ -86,10 +86,8 @@ protected static Row updateAfterRow(Object... values) { } protected static List listJoin(List> lists) { - List result = Lists.newArrayList(); - for (List list : lists) { - result.addAll(list); - } - return result; + return lists.stream() + .flatMap(List::stream) + .collect(Collectors.toList()); } }