[flink] branch master updated: [FLINK-27116][hive] Support read Hive table partitioned by decimal type

[jingzhang]

This is an automated email from the ASF dual-hosted git repository.

jingzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 99ccebf7d90 [FLINK-27116][hive] Support read Hive table partitioned by 
decimal type
99ccebf7d90 is described below

commit 99ccebf7d908c77ccca97ca9d66330474182d36d
Author: luoyuxia <luoyu...@alibaba-inc.com>
AuthorDate: Thu Apr 7 18:13:53 2022 +0800

    [FLINK-27116][hive] Support read Hive table partitioned by decimal type
    
    This closes #19388
---
 .../apache/flink/connectors/hive/util/HivePartitionUtils.java    | 3 +++
 .../java/org/apache/flink/connectors/hive/HiveDialectITCase.java | 9 +++++++++
 2 files changed, 12 insertions(+)

diff --git 
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/util/HivePartitionUtils.java
 
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/util/HivePartitionUtils.java
index fc055b5315f..7bf8bbc83b7 100644
--- 
a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/util/HivePartitionUtils.java
+++ 
b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/util/HivePartitionUtils.java
@@ -42,6 +42,7 @@ import org.apache.hadoop.mapred.JobConf;
 import org.apache.thrift.TException;
 
 import java.io.IOException;
+import java.math.BigDecimal;
 import java.sql.Date;
 import java.sql.Timestamp;
 import java.util.ArrayList;
@@ -120,6 +121,8 @@ public class HivePartitionUtils {
                 return Float.valueOf(valStr);
             case DOUBLE:
                 return Double.valueOf(valStr);
+            case DECIMAL:
+                return new BigDecimal(valStr);
             case DATE:
                 return HiveInspectors.toFlinkObject(
                         HiveInspectors.getObjectInspector(partitionType),
diff --git 
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java
 
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java
index babdd750b9d..491b19104f9 100644
--- 
a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java
+++ 
b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java
@@ -358,6 +358,15 @@ public class HiveDialectITCase {
                 .isEqualTo(
                         "[+I[1, 0, static], +I[1, 1, a], +I[1, 1, static], 
+I[1, 2, b], +I[1, 3, c], +I[2, 0, static],"
                                 + " +I[2, 1, b], +I[2, 1, static], +I[3, 0, 
static], +I[3, 1, c], +I[3, 1, static]]");
+
+        // test table partitioned by decimal type
+        tableEnv.executeSql(
+                "create table dest3 (key int, value string) partitioned by (p1 
decimal(5, 2)) ");
+        tableEnv.executeSql("insert overwrite dest3 partition (p1) select 
1,y,100.45 from src")
+                .await();
+        results = queryResult(tableEnv.sqlQuery("select * from dest3"));
+        assertThat(results.toString())
+                .isEqualTo("[+I[1, a, 100.45], +I[1, b, 100.45], +I[1, c, 
100.45]]");
     }
 
     @Test