[Improvement]: Support computed columns for mixed-format tables in Flink DDL

[YesOrNo828]

Search before asking

• I have searched in the issues and found no similar issues.

What would you like to be improved?

create table arctic.test.L (id int, proc as PROCTIME())

Caused by: java.lang.UnsupportedOperationException: Creating table with computed columns is not supported yet.
	at com.netease.arctic.flink.catalog.ArcticCatalog.lambda$validateFlinkTable$3(ArcticCatalog.java:328)
	at java.util.ArrayList.forEach(ArrayList.java:1257)
	at com.netease.arctic.flink.catalog.ArcticCatalog.validateFlinkTable(ArcticCatalog.java:326)
	at com.netease.arctic.flink.catalog.ArcticCatalog.createTable(ArcticCatalog.java:271)

How should we improve?

Filter the computed column or watermark definition field.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Subtasks

No response

Code of Conduct

• I agree to follow this project's Code of Conduct

[wuwenchi]

Interested in it.
A PR for this feature has been mentioned to iceberg, but iceberg may be reluctant to support this feature.
(apache/iceberg#4625. Record expressions through table properties.)
Have you finalized your plans now?

[YesOrNo828]

Have you finalized your plans now?

@wuwenchi Yes, the Iceberg doesn't support this feature, but this is a really useful feature. Currently, the Arctic is available to add watermark strategies or computed columns via flink SQL: CREATE TABLE LIKE. But needs another step to create a temporal table to achieve this feature.
BTW, Apache Paimon also supports computed columns and stores expressions in the table properties.

Could you show your proposal here?

[YesOrNo828]

I think when a mixed-format table is created through the Flink catalog, the virtual columns and watermarks could be serialized into the options.
Deserializing to the flink table's schema from the options when get the mixed format table through the Flink Catalog.

[xieyi888]

@YesOrNo828 hi, I'm interested in this issue, Could you assign it to me?

[YesOrNo828]

@YesOrNo828 hi, I'm interested in this issue, Could you assign it to me?

Done, looking forward to your contribution.

[xieyi888]

I think when a mixed-format table is created through the Flink catalog, the virtual columns and watermarks could be serialized into the options. Deserializing to the flink table's schema from the options when get the mixed format table through the Flink Catalog.

Yes, Here is the key point

1. suppoort serialized and deserializing compute columns and watermarks
2. Only convert flink physicalColums into arctic(iceberg) schema during create table, compute columns and watermarks should be serialized into table properties
3. when get table, deserializing compute columns and watermarks ,and combine with flink TableSchema

I had some confusion. I see here is an dimTable in arctic ，should dimTable also refer to the points above？
https://github.com/NetEase/amoro/blob/master/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/table/DynamicTableFactory.java#L167C1-L167C85

[YesOrNo828]

I had some confusion. I see here is an dimTable in arctic ，should dimTable also refer to the points above？

I think this dim table would be removed in the future. This is legacy lookup join implemented in the Amoro and would be deprecated. So I think we couldn't update it.

cc @zhoujinsong WDYT?

[zhoujinsong]

I had some confusion. I see here is an dimTable in arctic ，should dimTable also refer to the points above？

I think this dim table would be removed in the future. This is legacy lookup join implemented in the Amoro and would be deprecated. So I think we couldn't update it.

cc @zhoujinsong WDYT?

I also think so.