You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by jc...@apache.org on 2017/02/06 04:01:34 UTC
hive git commit: HIVE-15806: Druid schema inference for Select
queries might produce wrong type for metrics (Jesus Camacho Rodriguez,
reviewed by Ashutosh Chauhan)
Repository: hive
Updated Branches:
refs/heads/master f63dc2d4f -> 8afc9336e
HIVE-15806: Druid schema inference for Select queries might produce wrong type for metrics (Jesus Camacho Rodriguez, reviewed by Ashutosh Chauhan)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/8afc9336
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/8afc9336
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/8afc9336
Branch: refs/heads/master
Commit: 8afc9336ef1b813c0ee26925bf0e827b43f6fc24
Parents: f63dc2d
Author: Jesus Camacho Rodriguez <jc...@apache.org>
Authored: Fri Feb 3 19:19:39 2017 +0000
Committer: Jesus Camacho Rodriguez <jc...@apache.org>
Committed: Mon Feb 6 04:01:31 2017 +0000
----------------------------------------------------------------------
.../hadoop/hive/druid/serde/DruidSerDe.java | 72 ++++++++++++++------
.../hadoop/hive/druid/TestDruidSerDe.java | 2 +-
2 files changed, 52 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/8afc9336/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSerDe.java
----------------------------------------------------------------------
diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSerDe.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSerDe.java
index 65259b0..4235e89 100644
--- a/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSerDe.java
+++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSerDe.java
@@ -211,27 +211,32 @@ public class DruidSerDe extends AbstractSerDe {
Query<?> query;
try {
query = DruidStorageHandlerUtils.JSON_MAPPER.readValue(druidQuery, Query.class);
+
+ switch (query.getType()) {
+ case Query.TIMESERIES:
+ inferSchema((TimeseriesQuery) query, columnNames, columnTypes);
+ break;
+ case Query.TOPN:
+ inferSchema((TopNQuery) query, columnNames, columnTypes);
+ break;
+ case Query.SELECT:
+ String address = HiveConf.getVar(configuration,
+ HiveConf.ConfVars.HIVE_DRUID_BROKER_DEFAULT_ADDRESS);
+ if (org.apache.commons.lang3.StringUtils.isEmpty(address)) {
+ throw new SerDeException("Druid broker address not specified in configuration");
+ }
+ inferSchema((SelectQuery) query, columnNames, columnTypes, address);
+ break;
+ case Query.GROUP_BY:
+ inferSchema((GroupByQuery) query, columnNames, columnTypes);
+ break;
+ default:
+ throw new SerDeException("Not supported Druid query");
+ }
} catch (Exception e) {
throw new SerDeException(e);
}
- switch (query.getType()) {
- case Query.TIMESERIES:
- inferSchema((TimeseriesQuery) query, columnNames, columnTypes);
- break;
- case Query.TOPN:
- inferSchema((TopNQuery) query, columnNames, columnTypes);
- break;
- case Query.SELECT:
- inferSchema((SelectQuery) query, columnNames, columnTypes);
- break;
- case Query.GROUP_BY:
- inferSchema((GroupByQuery) query, columnNames, columnTypes);
- break;
- default:
- throw new SerDeException("Not supported Druid query");
- }
-
columns = new String[columnNames.size()];
types = new PrimitiveTypeInfo[columnNames.size()];
for (int i = 0; i < columnTypes.size(); ++i) {
@@ -303,6 +308,9 @@ public class DruidSerDe extends AbstractSerDe {
columnTypes.add(DruidSerDeUtils.convertDruidToHiveType(af.getTypeName()));
}
// Post-aggregator columns
+ // TODO: Currently Calcite only infers avg for post-aggregate,
+ // but once we recognize other functions, we will need to infer
+ // different types for post-aggregation functions
for (PostAggregator pa : query.getPostAggregatorSpecs()) {
columnNames.add(pa.getName());
columnTypes.add(TypeInfoFactory.floatTypeInfo);
@@ -325,6 +333,9 @@ public class DruidSerDe extends AbstractSerDe {
columnTypes.add(DruidSerDeUtils.convertDruidToHiveType(af.getTypeName()));
}
// Post-aggregator columns
+ // TODO: Currently Calcite only infers avg for post-aggregate,
+ // but once we recognize other functions, we will need to infer
+ // different types for post-aggregation functions
for (PostAggregator pa : query.getPostAggregatorSpecs()) {
columnNames.add(pa.getName());
columnTypes.add(TypeInfoFactory.floatTypeInfo);
@@ -333,8 +344,7 @@ public class DruidSerDe extends AbstractSerDe {
/* Select query */
private void inferSchema(SelectQuery query, List<String> columnNames,
- List<PrimitiveTypeInfo> columnTypes
- ) {
+ List<PrimitiveTypeInfo> columnTypes, String address) throws SerDeException {
// Timestamp column
columnNames.add(DruidTable.DEFAULT_TIMESTAMP_COLUMN);
columnTypes.add(TypeInfoFactory.timestampTypeInfo);
@@ -343,10 +353,27 @@ public class DruidSerDe extends AbstractSerDe {
columnNames.add(ds.getOutputName());
columnTypes.add(TypeInfoFactory.stringTypeInfo);
}
- // Metric columns
+ // The type for metric columns is not explicit in the query, thus in this case
+ // we need to emit a metadata query to know their type
+ SegmentMetadataQueryBuilder builder = new Druids.SegmentMetadataQueryBuilder();
+ builder.dataSource(query.getDataSource());
+ builder.merge(true);
+ builder.analysisTypes();
+ SegmentMetadataQuery metadataQuery = builder.build();
+ // Execute query in Druid
+ SegmentAnalysis schemaInfo;
+ try {
+ schemaInfo = submitMetadataRequest(address, metadataQuery);
+ } catch (IOException e) {
+ throw new SerDeException(e);
+ }
+ if (schemaInfo == null) {
+ throw new SerDeException("Connected to Druid but could not retrieve datasource information");
+ }
for (String metric : query.getMetrics()) {
columnNames.add(metric);
- columnTypes.add(TypeInfoFactory.floatTypeInfo);
+ columnTypes.add(DruidSerDeUtils.convertDruidToHiveType(
+ schemaInfo.getColumns().get(metric).getType()));
}
}
@@ -368,6 +395,9 @@ public class DruidSerDe extends AbstractSerDe {
columnTypes.add(DruidSerDeUtils.convertDruidToHiveType(af.getTypeName()));
}
// Post-aggregator columns
+ // TODO: Currently Calcite only infers avg for post-aggregate,
+ // but once we recognize other functions, we will need to infer
+ // different types for post-aggregation functions
for (PostAggregator pa : query.getPostAggregatorSpecs()) {
columnNames.add(pa.getName());
columnTypes.add(TypeInfoFactory.floatTypeInfo);
http://git-wip-us.apache.org/repos/asf/hive/blob/8afc9336/druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidSerDe.java
----------------------------------------------------------------------
diff --git a/druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidSerDe.java b/druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidSerDe.java
index 8dfa4d7..a67afdb 100644
--- a/druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidSerDe.java
+++ b/druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidSerDe.java
@@ -502,7 +502,7 @@ public class TestDruidSerDe {
IllegalAccessException, IOException, InterruptedException,
NoSuchMethodException, InvocationTargetException {
// Create, initialize, and test the SerDe
- DruidSerDe serDe = new DruidSerDe();
+ QTestDruidSerDe serDe = new QTestDruidSerDe();
Configuration conf = new Configuration();
Properties tbl;
// Timeseries query