[FLINK-18529][hive] Query Hive table and filter by timestamp partitio…

…n can fail This closes apache#12856
burkaygur · Jul 14, 2020 · 81f41a6 · 81f41a6
1 parent d677f02
commit 81f41a6
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 14 deletions.
diff --git a/...-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java b/...-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java
@@ -46,6 +46,7 @@
 import org.apache.flink.table.functions.FunctionDefinition;
 import org.apache.flink.table.functions.hive.conversion.HiveInspectors;
 import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
 import org.apache.flink.table.types.logical.LogicalTypeRoot;
 
 import org.apache.hadoop.hive.conf.HiveConf;
@@ -453,22 +454,16 @@ public String visit(ValueLiteralExpression valueLiteral) {
 			if (value == null) {
 				return "null";
 			}
+			LogicalTypeRoot typeRoot = dataType.getLogicalType().getTypeRoot();
+			if (typeRoot.getFamilies().contains(LogicalTypeFamily.DATETIME)) {
+				// hive not support partition filter push down with these types.
+				return null;
+			}
 			value = HiveInspectors.getConversion(HiveInspectors.getObjectInspector(dataType), dataType.getLogicalType(), hiveShim)
 					.toHiveObject(value);
 			String res = value.toString();
-			LogicalTypeRoot typeRoot = dataType.getLogicalType().getTypeRoot();
-			switch (typeRoot) {
-				case CHAR:
-				case VARCHAR:
-					res = "'" + res.replace("'", "''") + "'";
-					break;
-				case DATE:
-				case TIMESTAMP_WITHOUT_TIME_ZONE:
-				case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
-					// hive not support partition filter push down with these types.
-					return null;
-				default:
-					break;
+			if (typeRoot == LogicalTypeRoot.CHAR || typeRoot == LogicalTypeRoot.VARCHAR) {
+				res = "'" + res.replace("'", "''") + "'";
 			}
 			return res;
 		}

diff --git a/...-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java b/...-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
@@ -289,6 +289,14 @@ public void testPartitionFilter() throws Exception {
 			assertTrue(optimizedPlan, optimizedPlan.contains("PartitionPruned: true, PartitionNums: 1"));
 			results = Lists.newArrayList(query.execute().collect());
 			assertEquals("[4]", results.toString());
+
+			query = tableEnv.sqlQuery("select x from db1.part where '' = p2");
+			explain = query.explain().split("==.*==\n");
+			assertFalse(catalog.fallback);
+			optimizedPlan = explain[2];
+			assertTrue(optimizedPlan, optimizedPlan.contains("PartitionPruned: true, PartitionNums: 0"));
+			results = Lists.newArrayList(query.execute().collect());
+			assertEquals("[]", results.toString());
 		} finally {
 			tableEnv.executeSql("drop database db1 cascade");
 		}
@@ -319,7 +327,11 @@ public void testPartitionFilterDateTimestamp() throws Exception {
 			assertTrue(optimizedPlan, optimizedPlan.contains("PartitionPruned: true, PartitionNums: 1"));
 			List<Row> results = Lists.newArrayList(query.execute().collect());
 			assertEquals("[3]", results.toString());
-			System.out.println(results);
+
+			// filter by timestamp partition
+			query = tableEnv.sqlQuery("select x from db1.part where timestamp '2018-08-08 08:08:09' = p2");
+			results = Lists.newArrayList(query.execute().collect());
+			assertEquals("[2]", results.toString());
 		} finally {
 			tableEnv.executeSql("drop database db1 cascade");
 		}