[FLINK-21727][hive] Support DDL in HiveParser

[lirui-apache]

What is the purpose of the change

Support DDLs in hive parser

Brief change log

• Add HiveParserDDLSemanticAnalyzer to analyze DDLs.
• Add DDLOperationConverter to convert the parsed DDLs into operations.
• Handle DDLs in HiveParser. DQL and DML are still handled by super class.
• Hive's temp function is more like flink's temp system function. Added an operation to create temp system function with a CatalogFunction instance.
• Misc changes needed to support DDLs.

Verifying this change

Existing and added test cases.

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/Mesos, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? doc

[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 ec1a240 (Sat Aug 28 11:13:34 UTC 2021)

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.

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

[flinkbot]

CI report:

• 0df0c79 UNKNOWN
• ec1a240 Azure: SUCCESS

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]

Thanks @lirui-apache for the great work. I left some comment.

[wuchong]

Why this is delegated to super parser?

[lirui-apache]

Because HiveParser cannot parse queries at the moment.

[wuchong]

Disable like pattern for now?

[lirui-apache]

HiveParserDDLSemanticAnalyzer::analyzeShowFunctions would throw an exception for "SHOW FUNCTIONS LIKE"

[wuchong]

Is it necessary to introduce a new operation CreateTempSystemInlineFunctionOperation? Can we use CreateTempSystemFunctionOperation and pass in the class name? Or can we unify them?

It's quite confusing what's the differences between CreateTempSystemInlineFunctionOperation and CreateTempSystemFunctionOperation.

[wuchong]

Don't we support escape anymore?

[lirui-apache]

We do. This is to verify users no longer have to escape default keyword in hive dialect.

[wuchong]

Could you also add a test with escaption to make sure it also works?

`default`.tbl1

[lirui-apache]

Thanks @wuchong for your review. I have updated the PR to address your comments.

[wuchong]

Thanks for the updating. LGTM.

[wangqinghuan]

I want to use Flink to connect Hive external table Elasticsearch.
Following code doesn't create hive external table correctly after merging this pr into my fork.

        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);
        String name            = "myhive";
        String defaultDatabase = "default";
        String hiveConfDir     = args[0];
        HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
        tableEnv.registerCatalog("myhive", hive);
        tableEnv.useCatalog("myhive");

        tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
        tableEnv.executeSql("CREATE EXTERNAL TABLE IF NOT EXISTS test (\n" +
                "    id      BIGINT,\n" +
                "    name    STRING)\n" +
                "STORED by 'org.elasticsearch.hadoop.hive.EsStorageHandler'\n" +
                "TBLPROPERTIES(\n" +
                "'es.resource' = 'test',\n" +
                "'es.nodes' = '172.16.1.192:9200'\n" +
                ") ");
       tableEnv.executeSql("insert into test values(1,'test')");

Executing this code, Flink create a generic hive table rather than external table for Elasticsearch. Do we support StorageHandler in Flink hive dialect? @lirui-apache

[lirui-apache]

Hey @wangqinghuan , thanks for trying out this feature and reporting the issue. Hive dialect doesn't support hive's storage handler table. That's mainly because hive connector currently doesn't support read/write such tables. So even if you manage to create such tables, you can't really use them in Flink. I think we'd better throw meaningful exceptions for these use cases.