IMPALA-13859: Add decimal to Kudu's supported primary key types

Add DECIMAL to KuduUtil.isSupportedKeyType because it is a supported
type for primary key by Kudu.

Testing:
 * add fe tests
 * add e2e tests

Change-Id: I0e8685fe89e4e4e511ab1d815c51ca5a021b4081
Reviewed-on: http://gerrit.cloudera.org:8080/22674
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
Reviewed-by: Zoltan Borok-Nagy <boroknagyz@cloudera.com>

fe/src/main/java/org/apache/impala/util/KuduUtil.java

@@ -390,7 +390,7 @@ public class KuduUtil {
   }
 
   public static boolean isSupportedKeyType(org.apache.impala.catalog.Type type) {
-    return type.isIntegerType() || type.isStringType() || type.isTimestamp()
+    return type.isFixedPointType() || type.isStringType() || type.isTimestamp()
         || type.isDate();
   }
 

fe/src/test/java/org/apache/impala/analysis/AnalyzeKuduDDLTest.java

@@ -247,6 +247,10 @@ public class AnalyzeKuduDDLTest extends FrontendTestBase {
         "(partition value = 1, partition value = 'abc', partition 3 <= values) " +
         "stored as kudu", "Range partition value 'abc' (type: STRING) is not type " +
         "compatible with partitioning column 'a' (type: INT).", isExternalPurgeTbl);
+    AnalysisError("create table tab (a int primary key) partition by range (a) " +
+        "(partition value = 1.2, partition value = 2) stored as kudu",
+        "Range partition value 1.2 (type: DECIMAL(2,1)) is not type compatible with " +
+        "partitioning column 'a' (type: INT).", isExternalPurgeTbl);
     AnalysisError("create table tab (a tinyint primary key) partition by range (a) " +
         "(partition value = 128) stored as kudu", "Range partition value 128 " +
         "(type: SMALLINT) is not type compatible with partitioning column 'a' " +
@@ -263,10 +267,17 @@ public class AnalyzeKuduDDLTest extends FrontendTestBase {
         "(partition value = 9223372036854775808) stored as kudu", "Range partition " +
         "value 9223372036854775808 (type: DECIMAL(19,0)) is not type compatible with " +
         "partitioning column 'a' (type: BIGINT)", isExternalPurgeTbl);
+    AnalysisError("create table tab (a decimal(18,2) primary key)" +
+        "partition by range (a) (partition value = 0.1234) stored as kudu",
+        "Range partition value 0.1234 (type: DECIMAL(4,4)) is not type compatible with " +
+        "partitioning column 'a' (type: DECIMAL(18,2))", isExternalPurgeTbl);
     // Test implicit casting/folding of partition values.
     AnalyzesOk("create table tab (a int primary key) partition by range (a) " +
         "(partition value = false, partition value = true) stored as kudu",
         isExternalPurgeTbl);
+    AnalyzesOk("create table tab (a decimal(18,2) primary key) partition by range (a) " +
+        "(partition value = 0, partition 0 < values <= 0.1) stored as kudu",
+        isExternalPurgeTbl);
     // Non-key column used in PARTITION BY
     AnalysisError("create table tab (a int, b string, c bigint, primary key (a)) " +
         "partition by range (b) (partition value = 'abc') stored as kudu",
@@ -278,6 +289,11 @@ public class AnalyzeKuduDDLTest extends FrontendTestBase {
         "range (a) (partition value = 1.2, partition value = 2) stored as kudu",
         "Range partition value 1.2 (type: DECIMAL(2,1)) is not type compatible with " +
         "partitioning column 'a' (type: INT).", isExternalPurgeTbl);
+    // Decimal range partition values
+    AnalyzesOk("create table tab (a decimal(8,2), b int, c int, primary key (a, b))" +
+        "partition by hash (a, b) partitions 8, " +
+        "range (a) (partition 0 < values < 1.23, partition 1.23 <= values)" +
+        "stored as kudu", isExternalPurgeTbl);
     // Non-existing column used in PARTITION BY
     AnalysisError("create table tab (a int, b int, primary key (a, b)) " +
         "partition by range(unknown_column) (partition value = 'abc') stored as kudu",

testdata/workloads/functional-query/queries/QueryTest/kudu_create.test

@@ -665,3 +665,45 @@ select id, int_col, auto_incrementing_id from non_unique_key_create_tbl15 order
 ---- TYPES
 INT,INT,BIGINT
 ====
+---- QUERY
+# Create Kudu table with decimal primary key and range partitions.
+create table decimal_partitioned_tbl
+  (c1 decimal(10,4) primary key, c2 decimal(15,5), c3 decimal(1,1))
+partition by range (c1) (partition values <= 0.2000, partition 0.2000 < values <= 0.7575,
+  partition 0.7575 < values)
+stored as kudu
+---- RESULTS
+'Table has been created.'
+====
+---- QUERY
+show range partitions decimal_partitioned_tbl
+---- RESULTS
+'VALUES < 0.2001'
+'0.2001 <= VALUES < 0.7576'
+'VALUES >= 0.7576'
+---- TYPES
+STRING
+====
+---- QUERY
+insert into decimal_partitioned_tbl
+  select c1, c2, c3 from functional.decimal_tiny order by c1 asc limit 10
+---- RUNTIME_PROFILE
+NumModifiedRows: 10
+NumRowErrors: 0
+====
+---- QUERY
+select c1, c2, c3 from decimal_partitioned_tbl order by c1 asc
+---- RESULTS
+0.0000,100.00000,0.0
+0.1111,101.22222,0.1
+0.2222,102.44444,0.2
+0.3333,103.66666,0.3
+0.4444,104.88888,0.4
+0.5555,106.11110,0.5
+0.6666,107.33332,0.6
+0.7777,108.55554,0.7
+0.8888,109.77776,0.8
+0.9999,110.99998,0.9
+---- TYPES
+DECIMAL,DECIMAL,DECIMAL
+====