[FLINK-26414][hive] Hive dialect supports macro #19561

luoyuxia · 2022-04-24T11:47:38Z

What is the purpose of the change

Hive dialect supports macro.

Brief change log

Support parse create/drop macro
Convert calling macro to invoking Hive's GenericUDFMacro

Verifying this change

Added test in ...link-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java#testMacro

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? N/A

flinkbot · 2022-04-24T11:53:20Z

CI report:

e7e1338 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

lsyldliu

@luoyuxia Thanks for your contribution, I left some comments.

lsyldliu · 2022-06-06T13:01:01Z

...java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java

+    private Operation convertCreateMacro(HiveParserASTNode ast) throws SemanticException {
+        String macroName = ast.getChild(0).getText();
+        if (FunctionUtils.isQualifiedFunctionName(macroName)) {
+            throw new SemanticException("Temporary macro cannot be created with a qualified name.");


According to the method FunctionUtils.isQualifiedFunctionName, this error message may be not cleared for user? so I think we can improve it.

lsyldliu · 2022-06-06T13:03:05Z

...tor-hive/src/test/java/org/apache/flink/table/planner/delegation/hive/HiveASTParserTest.java

@@ -139,6 +139,13 @@ public void testFunction() throws Exception {
        assertDDLType(HiveASTParser.TOK_SHOWFUNCTIONS, "show functions");
    }

+    @Test
+    public void testMacro() throws Exception {


It will be better if we add a test that create a mcro with qualified name

we may not need for macro with a qualified name will fail in parse phase.
The antrl segment is like follows:
KW_CREATE KW_TEMPORARY KW_MACRO Identifier LPAREN columnNameTypeList? RPAREN expression -> ^(TOK_CREATEMACRO Identifier columnNameTypeList? expression)

lsyldliu · 2022-06-06T13:12:30Z

...-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveFunctionWrapper.java

+     * serialized to string and held on in the HiveFunctionWrapper.
+     */
+    public HiveFunctionWrapper(String className, UDFType serializableInstance) {
+        this.className = className;


we can call it directly this(className);

lsyldliu · 2022-06-06T13:14:06Z

...-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveFunctionWrapper.java

+        } catch (ClassNotFoundException e) {
+            throw new FlinkHiveUDFException(
+                    String.format("Failed to deserialize function %s", className), e);
+        }


"Failed to deserialize function %s."

lsyldliu · 2022-06-06T16:03:29Z

...java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java

+            throw new SemanticException("Temporary macro cannot be created with a qualified name.");
+        }
+
+        List<FieldSchema> arguments = getColumns((HiveParserASTNode) ast.getChild(1), true);


The arguments readability is not good? IMO, it refers the all columns of the table? In other words, we should add an annotation about it.

lsyldliu · 2022-06-06T16:11:03Z

...java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java

+        return actualColumnNames;
+    }
+
+    private void getMacroColumnData(


We can return a Tuple2 here directly? BTW, maybe we can merge the method getMacroColumnData and getBody into one, then return a Tuple3?

Return a Tuple2 is good to me. But I would like separate the getMacroColumnData and getBody,

lsyldliu · 2022-06-06T16:13:27Z

...java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java

+        }
+        Set<String> expectedColumnNames = new LinkedHashSet<>(macroColumnNames);
+        if (!expectedColumnNames.equals(actualColumnNames)) {
+            throw new SemanticException(


String.format("Expected columns [%s], but found [%s].", expectedColumnNames, actualColumnNames) here ?

lsyldliu · 2022-06-06T16:18:02Z

...java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java

+            throw new SemanticException(
+                    "Expected columns " + expectedColumnNames + " but found " + actualColumnNames);
+        }
+        if (expectedColumnNames.size() != macroColumnNames.size()) {


What about expectedColumnNames.size() > macroColumnNames.size() or expectedColumnNames.size() < macroColumnNames.size() case, whether the error message is the same?

expectedColumnNames.size() > macroColumnNames.size() will nerver happen for expectedColumnNames is a set of macroColumnNames.

lsyldliu · 2022-06-06T16:27:29Z

...java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java

+        for (FieldSchema argument : arguments) {
+            TypeInfo columnType = TypeInfoUtils.getTypeInfoFromTypeString(argument.getType());
+            rowResolver.put(
+                    "",


org.apache.commons.lang3.StringUtils.EMPTY?

lsyldliu · 2022-06-06T16:28:11Z

...java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java

+            rowResolver.put(
+                    "",
+                    argument.getName(),
+                    new ColumnInfo(argument.getName(), columnType, "", false));


luoyuxia · 2022-06-08T09:20:41Z

@lsyldliu Thanks for your review. I have adrressed your comments.

luoyuxia · 2022-06-17T13:09:55Z

@flinkbot run azure

wuchong · 2022-07-06T04:14:23Z

...-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveFunctionWrapper.java

+                serializableInstance instanceof Serializable,
+                String.format(
+                        "The UDF %s should be an instance of Serializable.",
+                        serializableInstance.getClass().getName()));


We should check the class name is equal to the class of serializableInstance.

wuchong · 2022-07-06T04:17:18Z

...-connector-hive/src/main/java/org/apache/flink/table/functions/hive/HiveFunctionWrapper.java

+        try {
+            return (UDFType)
+                    SerializationUtilities.deserializeObject(
+                            udfSerializedString, (Class<Serializable>) getUDFClass());


This is problematic in the case the hive connector is dynamically loaded, i.e. in the user classloader instead of the current class loader. We should use the user classloader to load classes. Could you create a JIRA issue for this?

Of course, I have created FLINK-28430 for it.

wuchong · 2022-07-06T04:24:00Z

...java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java

+    private Operation convertCreateMacro(HiveParserASTNode ast) throws SemanticException {
+        String macroName = ast.getChild(0).getText();
+        if (FunctionUtils.isQualifiedFunctionName(macroName)) {
+            throw new SemanticException("The name of the temporary macro can't contain `.`.");


Please also print the wrong function name. For example:

CREATE TEMPORARY MACRO doesn't allow "." character in the macro name, but the name is "%s".

Besides, is there a test for the exception?

I add the test for the exception in HiveDialectITCase#testMacro just now.

wuchong

LGTM

luoyuxia changed the title ~~[FLINK-26366][hive] Hive dialect supports macro~~ [FLINK-26414][hive] Hive dialect supports macro Apr 24, 2022

flinkbot added the component=<none> label Apr 24, 2022

lsyldliu reviewed Jun 6, 2022

View reviewed changes

luoyuxia force-pushed the FLINK-26414-Marco branch from a35efa3 to a24ba17 Compare June 8, 2022 09:20

luoyuxia force-pushed the FLINK-26414-Marco branch from a24ba17 to 9739bf9 Compare June 17, 2022 06:35

wuchong reviewed Jul 6, 2022

View reviewed changes

luoyuxia and others added 3 commits July 7, 2022 09:21

[FLINK-26366][hive] Hive dialect supports macro

99114ec

address comments

d84f69f

address comments

e7e1338

luoyuxia force-pushed the FLINK-26414-Marco branch from 5764a8b to e7e1338 Compare July 7, 2022 01:21

wuchong approved these changes Jul 7, 2022

View reviewed changes

wuchong merged commit db633e8 into apache:master Jul 7, 2022

ericccarlson pushed a commit to ericccarlson/flink that referenced this pull request Jul 11, 2022

[FLINK-26414][hive] Hive dialect supports macro (apache#19561)

fd17298

liujiawinds pushed a commit to liujiawinds/flink that referenced this pull request Jul 22, 2022

[FLINK-26414][hive] Hive dialect supports macro (apache#19561)

4c9a557

huangxiaofeng10047 pushed a commit to huangxiaofeng10047/flink that referenced this pull request Nov 3, 2022

[FLINK-26414][hive] Hive dialect supports macro (apache#19561)

ef9f610

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-26414][hive] Hive dialect supports macro #19561

[FLINK-26414][hive] Hive dialect supports macro #19561

luoyuxia commented Apr 24, 2022 •

edited

flinkbot commented Apr 24, 2022 •

edited

lsyldliu left a comment

lsyldliu Jun 6, 2022

luoyuxia Jun 8, 2022

lsyldliu Jun 6, 2022

luoyuxia Jun 17, 2022

lsyldliu Jun 6, 2022

lsyldliu Jun 6, 2022

lsyldliu Jun 6, 2022

lsyldliu Jun 6, 2022

luoyuxia Jun 8, 2022

lsyldliu Jun 6, 2022

lsyldliu Jun 6, 2022

luoyuxia Jun 8, 2022

lsyldliu Jun 6, 2022

lsyldliu Jun 6, 2022

luoyuxia commented Jun 8, 2022

luoyuxia commented Jun 17, 2022

wuchong Jul 6, 2022

wuchong Jul 6, 2022

luoyuxia Jul 6, 2022

wuchong Jul 6, 2022

luoyuxia Jul 6, 2022

wuchong left a comment

[FLINK-26414][hive] Hive dialect supports macro #19561

[FLINK-26414][hive] Hive dialect supports macro #19561

Conversation

luoyuxia commented Apr 24, 2022 • edited

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

flinkbot commented Apr 24, 2022 • edited

CI report:

lsyldliu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

luoyuxia commented Jun 8, 2022

luoyuxia commented Jun 17, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wuchong left a comment

Choose a reason for hiding this comment

luoyuxia commented Apr 24, 2022 •

edited

flinkbot commented Apr 24, 2022 •

edited