Inherit row type of partial views from existing sink

[ryannedolan]

Summary

When creating a partial view, inherit the row type from the sink table, if one already exists.

Details

"Partial views" represent pipelines that materialize part of a table. Typically, this table already exists. In that case, we should validate that the pipeline's row type is compatible with the existing sink table's row type.

At the same time, our CREATE TABLE <sink> DDL should fully-specify the sink schema -- not the schema of the partial view. Otherwise, Flink connectors are likely to complain that the CREATE TABLE DDL does not match what the connector expects.

Testing

Verified that creating a partial view that is incompatible with an existing sink table will fail to validate:

0: Hoptimator> create materialized view ads."AD_CLICKS$part2" as select * from profile.members;
java.lang.AssertionError: Type mismatch: the field sizes are not equal.
AD_CLICKS: RecordType(VARCHAR CAMPAIGN_URN, VARCHAR MEMBER_URN) NOT NULL
pipeline: RecordType(VARCHAR FIRST_NAME, VARCHAR LAST_NAME, VARCHAR MEMBER_URN, VARCHAR COMPANY_URN) NOT NULL

A correct partial view works:

0: Hoptimator> create materialized view ads."AD_CLICKS$part2" as select member_urn, company_urn from profile.members;
No rows affected (0.44 seconds)
0: Hoptimator> select * from ads."AD_CLICKS$part2";
+-------------------+--------------------------+
|    MEMBER_URN     |       COMPANY_URN        |
+-------------------+--------------------------+
| urn:li:member:123 | urn:li:company:linkedin  |
| urn:li:member:456 | urn:li:company:linkedin  |
| urn:li:member:789 | urn:li:company:microsoft |
+-------------------+--------------------------+
3 rows selected (0.554 seconds)