[FLINK-16265][table][csv] CsvTableSourceFactoryBase should compare LogicalTypes instead of TableSchema

[JingsongLi]

What is the purpose of the change

  - name: TableNumber2
    type: source
    $VAR_UPDATE_MODE
    schema:
      - name: IntegerField2
        type: INT
      - name: StringField2
        type: VARCHAR
      - name: TimestampField3
        type: TIMESTAMP
    connector:
      type: filesystem
      path: "$VAR_SOURCE_PATH2"
    format:
      type: csv
      fields:
        - name: IntegerField2
          type: INT
        - name: StringField2
          type: VARCHAR
        - name: TimestampField3
          type: TIMESTAMP
      line-delimiter: "\n"
      comment-prefix: "#"

Table like this will fail in SQL-CLI.
The root cause is we will convert the properties into CatalogTableImpl and then convert into properties again. The schema type properties will use new type systems then which is not equal to the legacy types due to conversion classes.

Brief change log

We should avoid compare TableSchema in CsvTableSourceFactoryBase.createTableSource.

Verifying this change

LocalExecutorITCase

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

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

Documentation

• Does this pull request introduce a new feature? (yes / no)

[flinkbot]

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit b25a551 (Tue Feb 25 13:06:35 UTC 2020)

Warnings:

• No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

• ❓ 1. The [description] looks good.
• ❓ 2. There is [consensus] that the contribution should go into to Flink.
• ❓ 3. Needs [attention] from.
• ❓ 4. The change fits into the overall [architecture].
• ❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
• @flinkbot approve all to approve all aspects
• @flinkbot approve-until architecture to approve everything until architecture
• @flinkbot attention @username1 [@username2 ..] to require somebody's attention
• @flinkbot disapprove architecture to remove an approval you gave earlier

[JingsongLi]

TPC-H E2E Verified in local.

[flinkbot]

CI report:

• ab0c749 Travis: SUCCESS Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[wuchong]

Maybe we can have a small utility in the class to get the LogicalType List from TableSchema? Then we don't need the long method chain.

[JingsongLi]

OK

[wuchong]

LGTM.

[JingsongLi]

Thanks @wuchong for reviewing. Merged.