diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java index fb6ce308fbc7..08538edc010c 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java @@ -191,7 +191,7 @@ static InputRowParser getInputRowParser(Table table, TimestampSpec timestampSpec // Default case JSON if ((parseSpecFormat == null) || "json".equalsIgnoreCase(parseSpecFormat)) { - return new StringInputRowParser(new JSONParseSpec(timestampSpec, dimensionsSpec, null, null), "UTF-8"); + return new StringInputRowParser(new JSONParseSpec(timestampSpec, dimensionsSpec), "UTF-8"); } else if ("csv".equalsIgnoreCase(parseSpecFormat)) { return new StringInputRowParser(new CSVParseSpec(timestampSpec, dimensionsSpec, diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandler.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandler.java index 656fa40c03fa..9f86a06afe83 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandler.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandler.java @@ -47,7 +47,8 @@ import org.apache.druid.metadata.storage.derby.DerbyConnector; import org.apache.druid.metadata.storage.derby.DerbyMetadataStorage; import org.apache.druid.metadata.storage.mysql.MySQLConnector; -import org.apache.druid.metadata.storage.mysql.MySQLConnectorConfig; +import org.apache.druid.metadata.storage.mysql.MySQLConnectorDriverConfig; +import org.apache.druid.metadata.storage.mysql.MySQLConnectorSslConfig; import org.apache.druid.metadata.storage.postgresql.PostgreSQLConnector; import org.apache.druid.metadata.storage.postgresql.PostgreSQLConnectorConfig; import org.apache.druid.metadata.storage.postgresql.PostgreSQLTablesConfig; @@ -319,7 +320,7 @@ private void updateKafkaIngestion(Table table) { + columnNames); } - DimensionsSpec dimensionsSpec = new DimensionsSpec(dimensionsAndAggregates.lhs, null, null); + DimensionsSpec dimensionsSpec = new DimensionsSpec(dimensionsAndAggregates.lhs); String timestampFormat = DruidStorageHandlerUtils .getTableProperty(table, DruidConstants.DRUID_TIMESTAMP_FORMAT); String timestampColumnName = DruidStorageHandlerUtils @@ -884,7 +885,7 @@ private SQLMetadataConnector buildConnector() { connector = new MySQLConnector(storageConnectorConfigSupplier, Suppliers.ofInstance(getDruidMetadataStorageTablesConfig()), - new MySQLConnectorConfig()); + new MySQLConnectorSslConfig(), new MySQLConnectorDriverConfig()); break; case "postgresql": connector = diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandlerUtils.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandlerUtils.java index 7e49b0b7c96e..bf98eedc500a 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandlerUtils.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandlerUtils.java @@ -89,6 +89,7 @@ import org.apache.druid.segment.IndexSpec; import org.apache.druid.segment.VirtualColumn; import org.apache.druid.segment.VirtualColumns; +import org.apache.druid.segment.column.ColumnType; import org.apache.druid.segment.column.ValueType; import org.apache.druid.segment.data.BitmapSerdeFactory; import org.apache.druid.segment.data.ConciseBitmapSerdeFactory; @@ -217,7 +218,7 @@ private DruidStorageHandlerUtils() { /** * Mapper to use to serialize/deserialize Druid objects (SMILE). */ - public static final ObjectMapper SMILE_MAPPER = new DefaultObjectMapper(new SmileFactory()); + public static final ObjectMapper SMILE_MAPPER = new DefaultObjectMapper(new SmileFactory(), "smile"); private static final int DEFAULT_MAX_TRIES = 10; static { @@ -797,7 +798,7 @@ private static ShardSpec getNextPartitionShardSpec(ShardSpec shardSpec) { if (shardSpec instanceof LinearShardSpec) { return new LinearShardSpec(shardSpec.getPartitionNum() + 1); } else if (shardSpec instanceof NumberedShardSpec) { - return new NumberedShardSpec(shardSpec.getPartitionNum(), ((NumberedShardSpec) shardSpec).getPartitions()); + return new NumberedShardSpec(shardSpec.getPartitionNum(), ((NumberedShardSpec) shardSpec).getNumCorePartitions()); } else { // Druid only support appending more partitions to Linear and Numbered ShardSpecs. throw new IllegalStateException(String.format("Cannot expand shard spec [%s]", shardSpec)); @@ -832,12 +833,9 @@ public static IndexSpec getIndexSpec(Configuration jc) { if ("concise".equals(HiveConf.getVar(jc, HiveConf.ConfVars.HIVE_DRUID_BITMAP_FACTORY_TYPE))) { bitmapSerdeFactory = new ConciseBitmapSerdeFactory(); } else { - bitmapSerdeFactory = new RoaringBitmapSerdeFactory(true); + bitmapSerdeFactory = RoaringBitmapSerdeFactory.getInstance(); } - return new IndexSpec(bitmapSerdeFactory, - IndexSpec.DEFAULT_DIMENSION_COMPRESSION, - IndexSpec.DEFAULT_METRIC_COMPRESSION, - IndexSpec.DEFAULT_LONG_ENCODING); + return IndexSpec.builder().withBitmapSerdeFactory(bitmapSerdeFactory).build(); } public static Pair, AggregatorFactory[]> getDimensionsAndAggregates(List columnNames, @@ -1082,7 +1080,8 @@ private static BloomKFilter evaluateBloomFilter(ExprNodeDesc desc, Configuration Set usedColumnNames = virtualColumns.stream().map(col -> col.getOutputName()).collect(Collectors.toSet()); final String name = SqlValidatorUtil.uniquify("vc", usedColumnNames, SqlValidatorUtil.EXPR_SUGGESTER); ExpressionVirtualColumn expressionVirtualColumn = - new ExpressionVirtualColumn(name, virtualColumnExpr, targetType, ExprMacroTable.nil()); + new ExpressionVirtualColumn(name, virtualColumnExpr, ColumnType.fromString(targetType.toString()), + ExprMacroTable.nil()); virtualColumns.add(expressionVirtualColumn); return name; } diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidOutputFormat.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidOutputFormat.java index d90db9cbda92..f8ebc6b46eab 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidOutputFormat.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidOutputFormat.java @@ -121,11 +121,12 @@ public FileSinkOperator.RecordWriter getHiveRecordWriter( .getDimensionsAndAggregates(columnNames, columnTypes); final InputRowParser inputRowParser = new MapInputRowParser(new TimeAndDimsParseSpec( new TimestampSpec(DruidConstants.DEFAULT_TIMESTAMP_COLUMN, "auto", null), - new DimensionsSpec(dimensionsAndAggregates.lhs, Lists - .newArrayList(Constants.DRUID_TIMESTAMP_GRANULARITY_COL_NAME, - Constants.DRUID_SHARD_KEY_COL_NAME - ), null - ) + DimensionsSpec.builder() + .setDimensions(dimensionsAndAggregates.lhs) + .setDimensionExclusions(Lists.newArrayList( + Constants.DRUID_TIMESTAMP_GRANULARITY_COL_NAME, + Constants.DRUID_SHARD_KEY_COL_NAME)) + .build() )); Map @@ -152,9 +153,16 @@ public FileSinkOperator.RecordWriter getHiveRecordWriter( Integer maxRowInMemory = HiveConf.getIntVar(jc, HiveConf.ConfVars.HIVE_DRUID_MAX_ROW_IN_MEMORY); IndexSpec indexSpec = DruidStorageHandlerUtils.getIndexSpec(jc); - RealtimeTuningConfig realtimeTuningConfig = new RealtimeTuningConfig(maxRowInMemory, null, null, null, - new File(basePersistDirectory, dataSource), new CustomVersioningPolicy(version), null, null, null, indexSpec, - null, true, 0, 0, true, null, 0L, null, null); + RealtimeTuningConfig defaults = + RealtimeTuningConfig.makeDefaultTuningConfig(new File(basePersistDirectory, dataSource)); + RealtimeTuningConfig realtimeTuningConfig = new RealtimeTuningConfig(defaults.getAppendableIndexSpec(), + maxRowInMemory, defaults.getMaxBytesInMemory(), defaults.isSkipBytesInMemoryOverheadCheck(), + defaults.getIntermediatePersistPeriod(), defaults.getWindowPeriod(), + new File(basePersistDirectory, dataSource), new CustomVersioningPolicy(version), + defaults.getRejectionPolicyFactory(), defaults.getMaxPendingPersists(), defaults.getShardSpec(), indexSpec, + indexSpec, defaults.getPersistThreadPriority(), defaults.getMergeThreadPriority(), + defaults.isReportParseExceptions(), defaults.getHandoffConditionTimeout(), defaults.getAlertTimeout(), + defaults.getSegmentWriteOutMediumFactory(), defaults.getDedupColumn()); LOG.debug(String.format("running with Data schema [%s] ", dataSchema)); return new DruidRecordWriter(dataSchema, realtimeTuningConfig, diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java index 2a2be067125f..23c553e3dce0 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java @@ -198,7 +198,7 @@ private static List fetchLocatedSegmentDescriptors(Str request = String.format("http://%s/druid/v2/datasources/%s/candidates?intervals=%s", address, - query.getDataSource().getNames().get(0), + query.getDataSource().getTableNames().iterator().next(), URLEncoder.encode(intervals, "UTF-8")); LOG.debug("sending request {} to query for segments", request); final InputStream response; diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidRecordWriter.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidRecordWriter.java index dc16c4e3f791..6eee0ec265f1 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidRecordWriter.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidRecordWriter.java @@ -36,7 +36,9 @@ import org.apache.druid.segment.realtime.appenderator.Appenderators; import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec; import org.apache.druid.segment.realtime.appenderator.SegmentNotWritableException; -import org.apache.druid.segment.realtime.appenderator.SegmentsAndMetadata; +import org.apache.druid.segment.incremental.ParseExceptionHandler; +import org.apache.druid.segment.incremental.SimpleRowIngestionMeters; +import org.apache.druid.segment.realtime.appenderator.SegmentsAndCommitMetadata; import org.apache.druid.segment.realtime.plumber.Committers; import org.apache.druid.timeline.DataSegment; import org.apache.druid.timeline.partition.LinearShardSpec; @@ -103,10 +105,11 @@ public DruidRecordWriter(DataSchema dataSchema, "realtimeTuningConfig is null"); this.dataSchema = Preconditions.checkNotNull(dataSchema, "data schema is null"); - appenderator = Appenderators - .createOffline("hive-offline-appenderator", this.dataSchema, tuningConfig, false, new FireDepartmentMetrics(), - dataSegmentPusher, DruidStorageHandlerUtils.JSON_MAPPER, DruidStorageHandlerUtils.INDEX_IO, - DruidStorageHandlerUtils.INDEX_MERGER_V9); + SimpleRowIngestionMeters rowIngestionMeters = new SimpleRowIngestionMeters(); + appenderator = Appenderators.createOffline("hive-offline-appenderator", this.dataSchema, tuningConfig, + new FireDepartmentMetrics(), dataSegmentPusher, DruidStorageHandlerUtils.JSON_MAPPER, + DruidStorageHandlerUtils.INDEX_IO, DruidStorageHandlerUtils.INDEX_MERGER_V9, rowIngestionMeters, + new ParseExceptionHandler(rowIngestionMeters, tuningConfig.isReportParseExceptions(), 0, 0), false); this.maxPartitionSize = maxPartitionSize; appenderator.startJob(); this.segmentsDescriptorDir = Preconditions.checkNotNull(segmentsDescriptorsDir, "segmentsDescriptorsDir is null"); @@ -170,7 +173,7 @@ private SegmentIdWithShardSpec getSegmentIdentifierAndMaybePush(long truncatedTi private void pushSegments(List segmentsToPush) { try { - SegmentsAndMetadata segmentsAndMetadata = appenderator.push(segmentsToPush, committerSupplier.get(), false).get(); + SegmentsAndCommitMetadata segmentsAndMetadata = appenderator.push(segmentsToPush, committerSupplier.get(), false).get(); final Set pushedSegmentIdentifierHashSet = new HashSet<>(); for (DataSegment pushedSegment : segmentsAndMetadata.getSegments()) { diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/AvroParseSpec.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/AvroParseSpec.java index 48d6cf2d6c8a..3ed817b48fed 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/AvroParseSpec.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/AvroParseSpec.java @@ -41,7 +41,7 @@ public class AvroParseSpec extends ParseSpec { @JsonProperty("dimensionsSpec") DimensionsSpec dimensionsSpec, @JsonProperty("flattenSpec") JSONPathSpec flattenSpec) { super(timestampSpec != null ? timestampSpec : new TimestampSpec(null, null, null), - dimensionsSpec != null ? dimensionsSpec : new DimensionsSpec(null, null, null)); + dimensionsSpec != null ? dimensionsSpec : DimensionsSpec.EMPTY); this.flattenSpec = flattenSpec != null ? flattenSpec : JSONPathSpec.DEFAULT; } diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/KafkaSupervisorTuningConfig.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/KafkaSupervisorTuningConfig.java index 4e171612b39b..c9ede58b25b1 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/KafkaSupervisorTuningConfig.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/KafkaSupervisorTuningConfig.java @@ -22,7 +22,6 @@ import com.fasterxml.jackson.annotation.JsonSubTypes; import com.fasterxml.jackson.annotation.JsonTypeInfo; import org.apache.druid.segment.IndexSpec; -import org.apache.druid.segment.indexing.TuningConfigs; import org.apache.druid.segment.writeout.SegmentWriteOutMediumFactory; import org.joda.time.Duration; import org.joda.time.Period; @@ -136,8 +135,8 @@ public Duration getOffsetFetchPeriod() { @Override public String toString() { return "KafkaSupervisorTuningConfig{" + "maxRowsInMemory=" + getMaxRowsInMemory() + ", maxRowsPerSegment=" - + getMaxRowsPerSegment() + ", maxTotalRows=" + getMaxTotalRows() + ", maxBytesInMemory=" + TuningConfigs - .getMaxBytesInMemoryOrDefault(getMaxBytesInMemory()) + ", intermediatePersistPeriod=" + + getMaxRowsPerSegment() + ", maxTotalRows=" + getMaxTotalRows() + ", maxBytesInMemory=" + + getMaxBytesInMemoryOrDefault() + ", intermediatePersistPeriod=" + getIntermediatePersistPeriod() + ", basePersistDirectory=" + getBasePersistDirectory() + ", maxPendingPersists=" + getMaxPendingPersists() + ", indexSpec=" + getIndexSpec() + ", reportParseExceptions=" + isReportParseExceptions() + ", handoffConditionTimeout=" diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/SeekableStreamIndexTaskTuningConfig.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/SeekableStreamIndexTaskTuningConfig.java index 289f0e8d4308..b858a75dacd2 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/SeekableStreamIndexTaskTuningConfig.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/SeekableStreamIndexTaskTuningConfig.java @@ -19,6 +19,7 @@ package org.apache.hadoop.hive.druid.json; import com.fasterxml.jackson.annotation.JsonProperty; +import org.apache.druid.segment.incremental.AppendableIndexSpec; import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; import org.apache.druid.segment.IndexSpec; import org.apache.druid.segment.indexing.RealtimeTuningConfig; @@ -39,8 +40,10 @@ public abstract class SeekableStreamIndexTaskTuningConfig implements TuningConfi private static final boolean DEFAULT_RESET_OFFSET_AUTOMATICALLY = false; private static final boolean DEFAULT_SKIP_SEQUENCE_NUMBER_AVAILABILITY_CHECK = false; + private final AppendableIndexSpec appendableIndexSpec; private final int maxRowsInMemory; private final long maxBytesInMemory; + private final boolean skipBytesInMemoryOverheadCheck; private final DynamicPartitionsSpec partitionsSpec; private final Period intermediatePersistPeriod; private final File basePersistDirectory; @@ -85,11 +88,13 @@ public SeekableStreamIndexTaskTuningConfig( // Cannot be a static because default basePersistDirectory is unique per-instance final RealtimeTuningConfig defaults = RealtimeTuningConfig.makeDefaultTuningConfig(basePersistDirectory); + this.appendableIndexSpec = defaults.getAppendableIndexSpec(); this.maxRowsInMemory = maxRowsInMemory == null ? defaults.getMaxRowsInMemory() : maxRowsInMemory; this.partitionsSpec = new DynamicPartitionsSpec(maxRowsPerSegment, maxTotalRows); // initializing this to 0, it will be lazily initialized to a value // @see server.src.main.java.org.apache.druid.segment.indexing.TuningConfigs#getMaxBytesInMemoryOrDefault(long) this.maxBytesInMemory = maxBytesInMemory == null ? 0 : maxBytesInMemory; + this.skipBytesInMemoryOverheadCheck = defaults.isSkipBytesInMemoryOverheadCheck(); this.intermediatePersistPeriod = intermediatePersistPeriod == null ? defaults.getIntermediatePersistPeriod() : intermediatePersistPeriod; @@ -131,6 +136,12 @@ public SeekableStreamIndexTaskTuningConfig( : logParseExceptions; } + @Override + @JsonProperty + public AppendableIndexSpec getAppendableIndexSpec() { + return appendableIndexSpec; + } + @Override @JsonProperty public int getMaxRowsInMemory() { @@ -143,6 +154,12 @@ public long getMaxBytesInMemory() { return maxBytesInMemory; } + @Override + @JsonProperty + public boolean isSkipBytesInMemoryOverheadCheck() { + return skipBytesInMemoryOverheadCheck; + } + @Override @JsonProperty public Integer getMaxRowsPerSegment() { diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java index 19379e1724da..a5b23dc05d02 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java @@ -26,10 +26,10 @@ import com.google.common.base.Throwables; import org.apache.druid.java.util.common.IAE; import org.apache.druid.java.util.common.RE; -import org.apache.druid.java.util.common.guava.CloseQuietly; import org.apache.druid.java.util.http.client.HttpClient; import org.apache.druid.java.util.http.client.Request; import org.apache.druid.java.util.http.client.response.InputStreamResponseHandler; +import org.apache.commons.io.IOUtils; import org.apache.druid.query.Query; import org.apache.druid.query.QueryInterruptedException; import org.apache.hadoop.conf.Configuration; @@ -126,7 +126,7 @@ public JsonParserIterator createQueryResultsIterator() { } catch (Exception e) { if (iterator != null) { // We got exception while querying results from this host. - CloseQuietly.close(iterator); + IOUtils.closeQuietly(iterator); } LOG.error("Failure getting results for query[{}] from host[{}] because of [{}]", query, address, e.getMessage()); @@ -200,7 +200,7 @@ public void initialize(InputSplit split, Configuration conf) throws IOException @Override public void close() { if (queryResultsIterator != null) { - CloseQuietly.close(queryResultsIterator); + IOUtils.closeQuietly(queryResultsIterator); } } @@ -248,7 +248,7 @@ public void initialize(InputSplit split, Configuration conf) throws IOException return false; } if (jp.getCurrentToken() == JsonToken.END_ARRAY) { - CloseQuietly.close(jp); + IOUtils.closeQuietly(jp); return false; } @@ -296,7 +296,7 @@ private void init() { } @Override public void close() throws IOException { - CloseQuietly.close(jp); + IOUtils.closeQuietly(jp); } } diff --git a/druid-handler/src/test/org/apache/hadoop/hive/druid/TestHiveDruidQueryBasedInputFormat.java b/druid-handler/src/test/org/apache/hadoop/hive/druid/TestHiveDruidQueryBasedInputFormat.java index 2bcbb14f9fbf..ca114c326652 100644 --- a/druid-handler/src/test/org/apache/hadoop/hive/druid/TestHiveDruidQueryBasedInputFormat.java +++ b/druid-handler/src/test/org/apache/hadoop/hive/druid/TestHiveDruidQueryBasedInputFormat.java @@ -50,12 +50,7 @@ + "\"intervals\":{\"type\":\"LegacySegmentSpec\",\"intervals\":" + "[\"2012-01-01T08:00:00.000Z/2012-01-03T08:00:00.000Z\"]}," + "\"descending\":true," - + "\"virtualColumns\":[]," - + "\"filter\":null," + "\"granularity\":\"DAY\"," - + "\"aggregations\":[]," - + "\"postAggregations\":[]," - + "\"limit\":2147483647," + "\"context\":{\"queryId\":\"\"}}, [localhost:8082]}]"; private static final String @@ -85,20 +80,16 @@ TOPN_QUERY_SPLIT = "[HiveDruidSplit{{\"queryType\":\"topN\"," + "\"dataSource\":{\"type\":\"table\",\"name\":\"sample_data\"}," - + "\"virtualColumns\":[]," + "\"dimension\":{\"type\":\"LegacyDimensionSpec\",\"dimension\":\"sample_dim\"," + "\"outputName\":\"sample_dim\",\"outputType\":\"STRING\"}," + "\"metric\":{\"type\":\"LegacyTopNMetricSpec\",\"metric\":\"count\"}," + "\"threshold\":5," + "\"intervals\":{\"type\":\"LegacySegmentSpec\",\"intervals\":[\"2013-08-31T07:00:00" + ".000Z/2013-09-03T07:00:00.000Z\"]}," - + "\"filter\":null," + "\"granularity\":{\"type\":\"all\"}," - + "\"aggregations\":[{\"type\":\"longSum\",\"name\":\"count\",\"fieldName\":\"count\",\"expression\":null}," - + "{\"type\":\"doubleSum\",\"name\":\"some_metric\",\"fieldName\":\"some_metric\",\"expression\":null}]," - + "\"postAggregations\":[]," - + "\"context\":{\"queryId\":\"\"}," - + "\"descending\":false}, [localhost:8082]}]"; + + "\"aggregations\":[{\"type\":\"longSum\",\"name\":\"count\",\"fieldName\":\"count\"}," + + "{\"type\":\"doubleSum\",\"name\":\"some_metric\",\"fieldName\":\"some_metric\"}]," + + "\"context\":{\"queryId\":\"\"}}, [localhost:8082]}]"; private static final String GROUP_BY_QUERY = @@ -122,24 +113,18 @@ + "\"dataSource\":{\"type\":\"table\",\"name\":\"sample_datasource\"}," + "\"intervals\":{\"type\":\"LegacySegmentSpec\",\"intervals\":[\"2012-01-01T08:00:00" + ".000Z/2012-01-03T08:00:00.000Z\"]}," - + "\"virtualColumns\":[]," - + "\"filter\":null," + "\"granularity\":\"DAY\"," + "\"dimensions\":[{\"type\":\"LegacyDimensionSpec\",\"dimension\":\"country\",\"outputName\":\"country\"," + "\"outputType\":\"STRING\"}," + "{\"type\":\"LegacyDimensionSpec\",\"dimension\":\"device\",\"outputName\":\"device\"," + "\"outputType\":\"STRING\"}]," - + "\"aggregations\":[{\"type\":\"longSum\",\"name\":\"total_usage\",\"fieldName\":\"user_count\"," - + "\"expression\":null}," - + "{\"type\":\"doubleSum\",\"name\":\"data_transfer\",\"fieldName\":\"data_transfer\",\"expression\":null}]," - + "\"postAggregations\":[]," - + "\"having\":null," + + "\"aggregations\":[{\"type\":\"longSum\",\"name\":\"total_usage\",\"fieldName\":\"user_count\"}," + + "{\"type\":\"doubleSum\",\"name\":\"data_transfer\",\"fieldName\":\"data_transfer\"}]," + "\"limitSpec\":{\"type\":\"default\",\"columns\":[{\"dimension\":\"country\",\"direction\":\"ascending\"," + "\"dimensionOrder\":{\"type\":\"lexicographic\"}}," + "{\"dimension\":\"data_transfer\",\"direction\":\"ascending\"," + "\"dimensionOrder\":{\"type\":\"lexicographic\"}}],\"limit\":5000}," - + "\"context\":{\"queryId\":\"\"}," - + "\"descending\":false}, [localhost:8082]}]"; + + "\"context\":{\"queryId\":\"\"}}, [localhost:8082]}]"; @Test public void testTimeZone() throws Exception { diff --git a/druid-handler/src/test/org/apache/hadoop/hive/ql/io/TestDruidRecordWriter.java b/druid-handler/src/test/org/apache/hadoop/hive/ql/io/TestDruidRecordWriter.java index 3fb7bdf6ca19..f2fba695ede1 100644 --- a/druid-handler/src/test/org/apache/hadoop/hive/ql/io/TestDruidRecordWriter.java +++ b/druid-handler/src/test/org/apache/hadoop/hive/ql/io/TestDruidRecordWriter.java @@ -130,7 +130,7 @@ inputRowParser = new MapInputRowParser(new TimeAndDimsParseSpec(new TimestampSpec(DruidConstants.DEFAULT_TIMESTAMP_COLUMN, "auto", - null), new DimensionsSpec(ImmutableList.of(new StringDimensionSchema("host")), null, null))); + null), new DimensionsSpec(ImmutableList.of(new StringDimensionSchema("host"))))); final Map parserMap = objectMapper.convertValue(inputRowParser, new TypeReference>() { @@ -146,12 +146,17 @@ null, objectMapper); - IndexSpec indexSpec = new IndexSpec(new RoaringBitmapSerdeFactory(true), null, null, null); - RealtimeTuningConfig - tuningConfig = - new RealtimeTuningConfig(null, - null, null, null, temporaryFolder.newFolder(), null, null, null, null, indexSpec, null, null, 0, 0, null, - null, 0L, null, null); + IndexSpec indexSpec = IndexSpec.builder().withBitmapSerdeFactory(RoaringBitmapSerdeFactory.getInstance()).build(); + RealtimeTuningConfig defaults = RealtimeTuningConfig.makeDefaultTuningConfig(temporaryFolder.newFolder()); + RealtimeTuningConfig tuningConfig = defaults.withBasePersistDirectory(temporaryFolder.newFolder()) + .withVersioningPolicy(defaults.getVersioningPolicy()); + tuningConfig = new RealtimeTuningConfig(defaults.getAppendableIndexSpec(), defaults.getMaxRowsInMemory(), + defaults.getMaxBytesInMemory(), defaults.isSkipBytesInMemoryOverheadCheck(), + defaults.getIntermediatePersistPeriod(), defaults.getWindowPeriod(), temporaryFolder.newFolder(), + defaults.getVersioningPolicy(), defaults.getRejectionPolicyFactory(), defaults.getMaxPendingPersists(), + defaults.getShardSpec(), indexSpec, indexSpec, defaults.getPersistThreadPriority(), + defaults.getMergeThreadPriority(), defaults.isReportParseExceptions(), defaults.getHandoffConditionTimeout(), + defaults.getAlertTimeout(), defaults.getSegmentWriteOutMediumFactory(), defaults.getDedupColumn()); LocalFileSystem localFileSystem = FileSystem.getLocal(config); DataSegmentPusher dataSegmentPusher = new LocalDataSegmentPusher(new LocalDataSegmentPusherConfig() { @Override public File getStorageDirectory() { diff --git a/pom.xml b/pom.xml index 4481607165db..f679719b5257 100644 --- a/pom.xml +++ b/pom.xml @@ -134,7 +134,7 @@ 10.17.1.0 3.1.0 0.1.2 - 0.17.1 + 26.0.0 2.2.4 1.12.0 22.0 diff --git a/ql/src/test/results/clientpositive/tez/hybridgrace_hashjoin_2.q.out b/ql/src/test/results/clientpositive/tez/hybridgrace_hashjoin_2.q.out index b363de97a495..f6a197306601 100644 --- a/ql/src/test/results/clientpositive/tez/hybridgrace_hashjoin_2.q.out +++ b/ql/src/test/results/clientpositive/tez/hybridgrace_hashjoin_2.q.out @@ -453,11 +453,11 @@ STAGE PLANS: minReductionHashAggr: 0.99 mode: hash outputColumnNames: _col0 - Statistics: Num rows: 1/1 Data size: 8 Basic stats: COMPLETE Column stats: COMPLETE + Statistics: Num rows: 1/2 Data size: 8 Basic stats: COMPLETE Column stats: COMPLETE Reduce Output Operator null sort order: sort order: - Statistics: Num rows: 1/1 Data size: 8 Basic stats: COMPLETE Column stats: COMPLETE + Statistics: Num rows: 1/2 Data size: 8 Basic stats: COMPLETE Column stats: COMPLETE value expressions: _col0 (type: bigint) Execution mode: vectorized Map 5