support create table like in flink catalog and watermark in windows

[swapna267]

This PR addresses

1. Creation of dynamic Iceberg table in Flink Catalog using the underlying physical Iceberg table using LIKE clause.
2. Iceberg Source to support Source Watermark, so it can be used in Flink WINDOW functions. https://github.com/apache/flink/blob/release-1.18/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsSourceWatermark.java enables Flink to rely on the watermark strategy provided by the ScanTableSource itself.

CREATE TABLE table_wm (
      eventTS AS CAST(t1 AS TIMESTAMP(3)),
      WATERMARK FOR eventTS AS SOURCE_WATERMARK()
) WITH (
  'watermark-column'='t1'
) LIKE iceberg_catalog.db.table;

Reference:
#10219
#9346

[pvary]

@swapna267: Started the test runs so we can see the status of the PR.
Could you please remove the Flink 1.18/1.9 version part of the changes?
It makes the review easier if we concentrate only a single Flink version, and do the backport later when the changes to the main version are merged.

[swapna267]

@pvary reverted 1.18/1.19 changes.

[swapna267]

Background:

1. Creation of dynamic Iceberg table in Flink Catalog using the underlying physical Iceberg table using LIKE clause.

Currently (without the changes in PR), create table in flink catalog works by configuring flink connector as described in,
flink-connector

But that needs user to provide the schema for the table. A way to tackle that is to do create table LIKE using below DDL.

CREATE TABLE table_wm (
      eventTS AS CAST(t1 AS TIMESTAMP(3)),
      WATERMARK FOR eventTS AS SOURCE_WATERMARK()
) WITH (
  'connector'='iceberg',
  'catalog-name'='iceberg_catalog',
  'catalog-database'='testdb',
  'catalog-table'='t'
) LIKE iceberg_catalog.testdb.t;

Options like connector, catalog-name, catalog-database, catalog-table need to be duplicated as Iceberg FlinkCatalog doesn't return any catalog related properties during getTable. This PR addresses the issue by including these properties when getTable is called , which will be used by Flink when creating table in Flink Catalog.

2. Iceberg Source to support Source Watermark
   As raised in The "Emitting watermarks" feature can't be used in flink sql? #10219, Source watermark implemented as part of https://iceberg.apache.org/docs/nightly/flink-queries/#emitting-watermarks cannot be used in Flink window functions.

Flink let's user to push down watermark to source using interface SupportSourceWatermark.java .

Here we are falling back to read options implemented in #9346 , to configure the watermark column on Iceberg Source.

[pvary]

Why do we remove this check?

[swapna267]

Tables can be created using LIKE in

1. Flink Catalog - Not supported currently.
2. Another table in Iceberg catalog itself as detailed in doc

This check basically fails, if we try to create table using LIKE in Iceberg catalog, basically case#2 if we have connector=iceberg in options . For example, DDL like below,

CREATE TABLE  `hive_catalog`.`default`.`sample_like` 
LIKE `hive_catalog`.`default`.`sample`
WITH ('connector'='iceberg')

In order to support Case#1 without user setting any extra Options using WITH clause, we need to add connector in getTable,

iceberg/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

Line 344 in 52bfbdc

catalogAndTableProps.put("connector", FlinkDynamicTableFactory.FACTORY_IDENTIFIER);

This check was added in very old PR,
#2666
#2666 (comment) where Flink SQL didn't support CREATE TABLE A LIKE B , where A and B are in different Catalogs.

So, in this case by removing this check, we are ignoring connector option being passed, so following DDL can create table table_like in Flink catalog backed by iceberg_catalog.db.table. As we know source table is an Iceberg table, adding connector=iceberg would be redundant.

CREATE TABLE table_like (
      eventTS AS CAST(t1 AS TIMESTAMP(3)),
) LIKE iceberg_catalog.db.table;

[pvary]

What happens when the source table is not an Iceberg table?
I'm trying to understand here, where we get the missing information in this case, and wether we have a way to check that we actually get the missing information. If we can create such a check, then we can still throw an exception when we don't get this information from any source

[pvary]

I think we have 2 feature in a single PR:

• CREATE TABLE LIKE
• Watermark support

Could we separate out these features to different PRs?
Could we write tests for both features?

[swapna267]

These features were driven by mainly a use case, where an iceberg table is needed to be used in Flink window functions. This needs an incoming table to have MILLISECOND precision timestamp column and also watermark to be defined on source table.

As iceberg only supports MICROSECOND timestamp columns, we need to have a table with computed columns and we can create these only in Flink Catalog. Iceberg catalog doesn't support creating tables with computed columns.

i am happy to split them into 2 separate PR's .
I have tests for CREATE TABLE LIKE.

As Watermark support is just making Source to implement interface and falling back to #9346 for core logic, i didn't have a test case. I can add a validation on if watermark-column is configured or not , so it can fail fast. And a test case around that.

[pvary]

Please separate out the features to 2 PR

[pvary]

Why is this test removed?

[swapna267]

This is testing the check mentioned in #12116 (comment)

Fail creating a table in Iceberg Catalog if connector=iceberg is specified in the Option. As the check is been deleted, i removed this test case.

[pvary]

I think that this is still a valid check in most cases. Only not valid when the table is created with "CREATE TABLE.. LIKE" and only if the source table is an iceberg table.
Do I miss something?