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..0827b21786c1 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,23 @@ 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, + "Cannot find field ID for the primary key column %s in schema %s", column, iSchema); + identifierFieldIds.add(field.fieldId()); + } + } + + return new Schema(iSchema.schemaId(), iSchema.asStruct().fields(), identifierFieldIds); } /** @@ -83,7 +103,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 +142,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..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,9 +209,17 @@ public static StructLikeSet expectedRowSet(Table table, Record... records) { } public static StructLikeSet actualRowSet(Table table, String... columns) throws IOException { + return actualRowSet(table, null, 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 == null ? table.currentSnapshot().snapshotId() : 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..d44f45ab52fd --- /dev/null +++ b/flink/src/test/java/org/apache/iceberg/flink/TestChangeLogTable.java @@ -0,0 +1,306 @@ +/* + * 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; + +/** + * 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(); + 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( + insertRow(1, "aaa"), + deleteRow(1, "aaa"), + insertRow(1, "bbb"), + insertRow(2, "aaa"), + deleteRow(2, "aaa"), + insertRow(2, "bbb") + ), + ImmutableList.of( + updateBeforeRow(2, "bbb"), + updateAfterRow(2, "ccc"), + deleteRow(2, "ccc"), + insertRow(2, "ddd") + ), + ImmutableList.of( + deleteRow(1, "bbb"), + insertRow(1, "ccc"), + deleteRow(1, "ccc"), + insertRow(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( + insertRow(1, "aaa"), + deleteRow(1, "aaa"), + insertRow(2, "bbb"), + insertRow(1, "bbb"), + insertRow(2, "aaa") + ), + ImmutableList.of( + updateBeforeRow(2, "aaa"), + updateAfterRow(1, "ccc"), + insertRow(1, "aaa") + ), + ImmutableList.of( + deleteRow(1, "bbb"), + insertRow(2, "aaa"), + insertRow(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( + insertRow(1, "aaa"), + deleteRow(1, "aaa"), + insertRow(2, "bbb"), + insertRow(1, "bbb"), + insertRow(2, "aaa") + ), + ImmutableList.of( + updateBeforeRow(2, "aaa"), + updateAfterRow(1, "ccc"), + insertRow(1, "aaa") + ), + ImmutableList.of( + deleteRow(1, "bbb"), + insertRow(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( + insertRow(1, "aaa"), + insertRow(2, "bbb") + ), + ImmutableList.of( + insertRow(3, "ccc"), + insertRow(4, "ddd") + ), + ImmutableList.of( + insertRow(5, "eee"), + insertRow(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/TestFlinkSchemaUtil.java b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java index 3272ed65d35d..460869ae8c3f 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,54 @@ private void checkInconsistentType( Types.StructType.of(Types.NestedField.optional(0, "f0", icebergExpectedType)), 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 icebergSchema = 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(icebergSchema); + Assert.assertTrue(tableSchema.getPrimaryKey().isPresent()); + Assert.assertEquals(ImmutableSet.of("int", "string"), + ImmutableSet.copyOf(tableSchema.getPrimaryKey().get().getColumns())); + } + + @Test + public void testConvertFlinkSchemaWithNestedColumnInPrimaryKeys() { + Schema icebergSchema = 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(icebergSchema)); + } } 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..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 @@ -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; /** @@ -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 new file mode 100644 index 000000000000..a445e7eb06ce --- /dev/null +++ b/flink/src/test/java/org/apache/iceberg/flink/source/ChangeLogTableTestBase.java @@ -0,0 +1,93 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.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; +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.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; + } + + protected static Row insertRow(Object... values) { + return Row.ofKind(RowKind.INSERT, values); + } + + 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); + } + + protected static Row updateAfterRow(Object... values) { + return Row.ofKind(RowKind.UPDATE_AFTER, values); + } + + protected static List listJoin(List> lists) { + return lists.stream() + .flatMap(List::stream) + .collect(Collectors.toList()); + } +} 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..d163b84c09c6 --- /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( + insertRow(1, "aaa"), + deleteRow(1, "aaa"), + insertRow(1, "bbb"), + insertRow(2, "aaa"), + deleteRow(2, "aaa"), + insertRow(2, "bbb") + ), + ImmutableList.of( + updateBeforeRow(2, "bbb"), + updateAfterRow(2, "ccc"), + deleteRow(2, "ccc"), + insertRow(2, "ddd") + ), + ImmutableList.of( + deleteRow(1, "bbb"), + insertRow(1, "ccc"), + deleteRow(1, "ccc"), + insertRow(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