Flink: Add unit test to write CDC events by SQL. #1978
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -51,6 +51,8 @@ | |
| import org.apache.flink.table.catalog.stats.CatalogTableStatistics; | ||
| import org.apache.flink.table.expressions.Expression; | ||
| import org.apache.flink.table.factories.TableFactory; | ||
| import org.apache.flink.table.types.logical.RowType; | ||
| import org.apache.flink.table.types.utils.TypeConversions; | ||
| import org.apache.flink.util.StringUtils; | ||
| import org.apache.iceberg.CachingCatalog; | ||
| import org.apache.iceberg.DataFile; | ||
|
|
@@ -60,6 +62,7 @@ | |
| import org.apache.iceberg.Schema; | ||
| import org.apache.iceberg.StructLike; | ||
| import org.apache.iceberg.Table; | ||
| import org.apache.iceberg.TableProperties; | ||
| import org.apache.iceberg.Transaction; | ||
| import org.apache.iceberg.UpdateProperties; | ||
| import org.apache.iceberg.catalog.Catalog; | ||
|
|
@@ -75,6 +78,7 @@ | |
| import org.apache.iceberg.relocated.com.google.common.collect.Lists; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Maps; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Sets; | ||
| import org.apache.iceberg.util.PropertyUtil; | ||
|
|
||
| /** | ||
| * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}. | ||
|
|
@@ -359,6 +363,14 @@ public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ig | |
| PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema); | ||
|
|
||
| ImmutableMap.Builder<String, String> properties = ImmutableMap.builder(); | ||
|
|
||
| // Set the equality field columns. | ||
| List<String> equalityFieldColumns = toEqualityColumns(table.getSchema()); | ||
| if (!equalityFieldColumns.isEmpty()) { | ||
| properties.put(TableProperties.EQUALITY_FIELD_COLUMNS, | ||
| org.apache.commons.lang.StringUtils.join(equalityFieldColumns, ",")); | ||
| } | ||
|
|
||
| String location = null; | ||
| for (Map.Entry<String, String> entry : table.getOptions().entrySet()) { | ||
| if ("location".equalsIgnoreCase(entry.getKey())) { | ||
|
|
@@ -447,10 +459,12 @@ private static void validateFlinkTable(CatalogBaseTable table) { | |
| if (!schema.getWatermarkSpecs().isEmpty()) { | ||
| throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet."); | ||
| } | ||
| } | ||
|
|
||
| private static List<String> toEqualityColumns(TableSchema schema) { | ||
|
Contributor
There was a problem hiding this comment. In the format, I think that equality columns should be tracked by ID rather than by name so that renaming columns doesn't break the primary key metadata.
Member
Author
There was a problem hiding this comment. Yes, it should track ID here. I was thinking that it could be a temporary solution before introducing the primary key specification. Considering the renaming issue, it seems not a good idea to track column names in iceberg properties. I will create another PR to introduce the primary key specification based on the discussion from here. |
||
| List<String> equalityColumns = Lists.newArrayList(); | ||
| schema.getPrimaryKey().ifPresent(uniqueConstraint -> equalityColumns.addAll(uniqueConstraint.getColumns())); | ||
|
Contributor
There was a problem hiding this comment. It seems like we should track primary key columns in the table format rather than in properties. If we are going to add primary key columns, then we should add it before v2 so that v2 writers are required to not drop it. |
||
| return equalityColumns; | ||
| } | ||
|
|
||
| private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) { | ||
|
|
@@ -516,7 +530,26 @@ private static void commitChanges(Table table, String setLocation, String setSna | |
| } | ||
|
|
||
| static CatalogTable toCatalogTable(Table table) { | ||
| TableSchema.Builder builder = TableSchema.builder(); | ||
|
|
||
| // Add the table columns. | ||
| RowType rowType = FlinkSchemaUtil.convert(table.schema()); | ||
| for (RowType.RowField field : rowType.getFields()) { | ||
| builder.field(field.getName(), TypeConversions.fromLogicalToDataType(field.getType())); | ||
| } | ||
|
|
||
| // Add the primary keys. | ||
| String concatColumns = PropertyUtil.propertyAsString(table.properties(), | ||
| TableProperties.EQUALITY_FIELD_COLUMNS, | ||
| TableProperties.DEFAULT_EQUALITY_FIELD_COLUMNS); | ||
| String[] columns = org.apache.commons.lang3.StringUtils.split(concatColumns, ","); | ||
|
Contributor
There was a problem hiding this comment. Same here. Could you use |
||
| if (columns != null && columns.length > 0) { | ||
| builder.primaryKey(columns); | ||
| } | ||
|
|
||
| // Build the table schema. | ||
| TableSchema schema = builder.build(); | ||
|
|
||
| List<String> partitionKeys = toPartitionKeys(table.spec(), table.schema()); | ||
|
|
||
| // NOTE: We can not create a IcebergCatalogTable extends CatalogTable, because Flink optimizer may use | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We try to avoid using
StringUtilsand other commons classes because we don't want to leak the dependency. Could you use Guava'sJoinerinstead?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sounds great. I did not realize that there's a
Joinerin Guava. Thanks.