[jira] [Created] (FLINK-30598) Wrong code generated for WatermarkGenerator because of inconsistent source type info when deserialized from exec plan

["lincoln lee (Jira)" <ji...@apache.org>]

lincoln lee created FLINK-30598:
-----------------------------------

             Summary: Wrong code generated for WatermarkGenerator because of inconsistent source type info when deserialized from exec plan
                 Key: FLINK-30598
                 URL: https://issues.apache.org/jira/browse/FLINK-30598
             Project: Flink
          Issue Type: Bug
          Components: Table SQL / Planner
    Affects Versions: 1.16.0
            Reporter: lincoln lee
            Assignee: lincoln lee
             Fix For: 1.17.0, 1.16.1


When compile from an exist exec plan which contains watermark declaration and it referred the metadata column, the generated code for WatermarkGenerator maybe wrong

because currently `DynamicTableSourceSpec`.getTableSource passes the user defined schema to `SourceAbilitySpec` to perform optimization like projection/watermark pushdown, while optimization path from sql use a fixed reorder form: "PHYSICAL COLUMNS + METADATA COLUMNS", this may cause the problem.

a repro-case:

{code}
@Test
    public void testWatermarkPushDownWithMetadata() throws Exception {
        // to verify FLINK-: the case declares metadata field first, without fix it will get a
        // wrong code generated by WatermarkGeneratorCodeGenerator which reference the incorrect
        // varchar column as the watermark field.
        createTestValuesSourceTable(
                "MyTable",
                JavaScalaConversionUtil.toJava(TestData.data3WithTimestamp()),
                new String[] {
                    "ts timestamp(3) metadata",
                    "a int",
                    "b bigint",
                    "c varchar",
                    "watermark for ts as ts - interval '5' second"
                },
                new HashMap<String, String>() {
                    {
                        put("enable-watermark-push-down", "true");
                        put("readable-metadata", "ts:timestamp(3)");
                    }
                });

        File sinkPath =
                createTestCsvSinkTable(
                        "MySink", "a int", "b bigint", "c varchar", "ts timestamp(3)");

        compileSqlAndExecutePlan("insert into MySink select a, b, c, ts from MyTable where b = 3")
                .await();

        assertResult(
                Arrays.asList(
                        "4,3,Hello world, how are you?," + toLocalDateTime(4000L),
                        "5,3,I am fine.," + toLocalDateTime(5000L),
                        "6,3,Luke Skywalker," + toLocalDateTime(6000L)),
                sinkPath);
    }
{code}

the wrong code snippet(`row.getString(3)` should be a TimestampData):
{code}
public Long currentWatermark(org.apache.flink.table.data.RowData row) throws Exception {
          
          org.apache.flink.table.data.binary.BinaryStringData field$19;
          boolean isNull$19;
          org.apache.flink.table.data.binary.BinaryStringData field$21;
          boolean isNull$22;
          org.apache.flink.table.data.TimestampData result$23;
          boolean isNull$24;
          org.apache.flink.table.data.TimestampData result$25;
          
          isNull$19 = row.isNullAt(3);
          field$19 = org.apache.flink.table.data.binary.BinaryStringData.EMPTY_UTF8;
          if (!isNull$19) {
            field$19 = ((org.apache.flink.table.data.binary.BinaryStringData) row.getString(3));
          }
{code}

 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)