[FLINK-20539][table] Type mismatch when using computed ROW column

[snuyanzin]

When using cast, Flink uses ExtendedSqlRowTypeNameSpec to derive the row type, and we should change the StructKind from FULLY_QUALIFIED in Calcite to PEEK_FIELDS_NO_EXPAND in Flink.

Also if extra cast for nested field happened because of mismatched type with enclosing row it checks whether previous cast should be dropped.

This will resolve the issue with nested column casts
for instance

CREATE TABLE testCastOfTestToSameTypeWithNullableNestedType (
      `field1` ROW<`data` ROW<`nested` ROW<`trId` STRING>>NOT NULL>,
      `field2` AS CAST(`field1` AS ROW<`data` ROW<`nested` ROW<`trId` STRING>>NOT NULL>)) 
WITH ('connector' = 'datagen')

and then

EXPLAIN SELECT `field2`, COALESCE(TRY_CAST(`field1`.`data`.`nested`.`trId` AS STRING)) AS transactionId 
FROM testCastOfTestToSameTypeWithNullableNestedType

gives

Calc(select=[field1 AS field2, COALESCE(CAST(field1.data AS RecordType:peek_no_expand(RecordType:peek_no_expand(VARCHAR(2147483647) trId) nested)).nested.trId) AS transactionId])
+- TableSourceScan(table=[[default_catalog, default_database, testCastOfTestToSameTypeWithNullableNestedType]], fields=[field1])

while in fact it should be simplified to

Calc(select=[field1 AS field2, COALESCE(field1.data.nested.trId) AS transactionId])
+- TableSourceScan(table=[[default_catalog, default_database, testCastOfTestToSameTypeWithNullableNestedType]], fields=[field1])

I noticed that part of my changes are very similar to the PR #24994 (about PEEK_FIELDS_NO_EXPAND) done by @xuyangzhong,
however cherry-picking leads to multiple conflicts, so I just added him as a co-author.

Brief change log

Return a Row type with PEEK_FIELDS_NO_EXPAND as the target type in CAST
Drop extra casts for rexFieldAccess

Verifying this change

A number of tests in CalcTest and CalcITTest

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): ( no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): ( no)
• The serializers: ( no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? ( no)
• If yes, how is the feature documented? (not applicable)

[flinkbot]

CI report:

• bffc43a Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[snuyanzin]

Now align with what is mentioned in comment
Seems it is a very old bug also mentioned at https://issues.apache.org/jira/browse/FLINK-13604

[davidradl]

A basic question - I am curious why we are doing cast processing when the reported problem was for a computed row - are they related?

[snuyanzin]

yes, if you look at the comment for makeFieldAccess in this class, for instance

flink/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkRexBuilder.java

Lines 34 to 41 in ed0ae9d

	/**
	* Compared to the original method we adjust the nullability of the nested column based on the
	* nullability of the enclosing type.
	*
	* <p>If the fields type is NOT NULL, but the enclosing ROW is nullable we still can produce
	* nulls.
	*/
	@Override

[snuyanzin]

so we adjust nullability, however with such approach for heavily nested columns it might happen that several casts might appear after that. The idea here is to keep only zero or one if needed and not more since anyway one cast is enough for nullability adjustment