From 88efab57f4a777b677eccc69c0432044141fe889 Mon Sep 17 00:00:00 2001 From: saihemanth Date: Thu, 9 Jul 2026 14:23:42 -0700 Subject: [PATCH 1/3] HIVE-29707: Upgrade Apache Druid from 0.17.1 to 26.0.0 --- .../hadoop/hive/druid/DruidKafkaUtils.java | 2 +- .../hive/druid/DruidStorageHandler.java | 7 +++--- .../hive/druid/DruidStorageHandlerUtils.java | 15 ++++++------ .../hive/druid/io/DruidOutputFormat.java | 24 ++++++++++++------- .../druid/io/DruidQueryBasedInputFormat.java | 2 +- .../hive/druid/io/DruidRecordWriter.java | 15 +++++++----- .../hadoop/hive/druid/json/AvroParseSpec.java | 2 +- .../json/KafkaSupervisorTuningConfig.java | 5 ++-- .../SeekableStreamIndexTaskTuningConfig.java | 17 +++++++++++++ .../druid/serde/DruidQueryRecordReader.java | 10 ++++---- .../hive/ql/io/TestDruidRecordWriter.java | 19 +++++++++------ pom.xml | 2 +- 12 files changed, 76 insertions(+), 44 deletions(-) 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/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 b7471070382c..d9a5d094995a 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 From 5b9acacc64d505858eef491d149a69e888763348 Mon Sep 17 00:00:00 2001 From: saihemanth Date: Tue, 14 Jul 2026 11:12:09 -0700 Subject: [PATCH 2/3] Address test failures --- .../TestHiveDruidQueryBasedInputFormat.java | 27 +++++-------------- .../tez/hybridgrace_hashjoin_2.q.out | 4 +-- 2 files changed, 8 insertions(+), 23 deletions(-) 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/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 From 5adc8dca8f80b49734d3f02b1b5d6107566387b6 Mon Sep 17 00:00:00 2001 From: saihemanth Date: Wed, 15 Jul 2026 15:42:28 -0700 Subject: [PATCH 3/3] Fix TestMiniLlapCliDriver OOMs in llap_io_cache.q and script_env_var2.q The Druid 26 upgrade increases the hive-it-qfile test classpath footprint, leaving too little heap in the forked TestMiniLlapCliDriver JVM for Tez AM and mini-cluster components that run in-process alongside LLAP. Cap in-process mini LLAP executor and cache memory in LlapItUtils so Tez AM, mini YARN, and mini HDFS retain heap (fixes script_env_var2.q). Bump the qfile Surefire fork heap to 4 GB for additional headroom. Rewrite llap_io_cache.q to evaluate RPAD at runtime instead of constant- folding a ~16 MB string into the Tez MapWork plan, which OOMs the Tez AM during Kryo deserialization of map.xml. Update the expected q.out for the new INSERT shape and lineage. --- itests/qtest/pom.xml | 2 +- .../org/apache/hadoop/hive/llap/LlapItUtils.java | 15 +++++++++++---- .../test/queries/clientpositive/llap_io_cache.q | 9 +++++++-- .../clientpositive/llap/llap_io_cache.q.out | 14 +++++++++----- 4 files changed, 28 insertions(+), 12 deletions(-) diff --git a/itests/qtest/pom.xml b/itests/qtest/pom.xml index fdf75bd890bc..5ffa790ff45a 100644 --- a/itests/qtest/pom.xml +++ b/itests/qtest/pom.xml @@ -638,7 +638,7 @@ org.apache.maven.plugins maven-surefire-plugin - ${maven.test.jvm.args} + ${maven.test.jvm.args} -Xmx4096m ${test.conf.dir} ${basedir}/${hive.path.to.root}/conf diff --git a/itests/util/src/main/java/org/apache/hadoop/hive/llap/LlapItUtils.java b/itests/util/src/main/java/org/apache/hadoop/hive/llap/LlapItUtils.java index 81dd92bb32a1..4093b0e778c2 100644 --- a/itests/util/src/main/java/org/apache/hadoop/hive/llap/LlapItUtils.java +++ b/itests/util/src/main/java/org/apache/hadoop/hive/llap/LlapItUtils.java @@ -37,6 +37,13 @@ public class LlapItUtils { private static final Logger LOG = LoggerFactory.getLogger(LlapItUtils.class); + /** + * Cap in-process mini LLAP memory so Tez AM / mini YARN / mini HDFS retain heap. + * Without caps, 90% of the test JVM max heap is reserved for LLAP alone. + */ + private static final long MAX_MINI_LLAP_EXECUTOR_MEMORY_BYTES = 768L * 1024 * 1024; + private static final long MAX_MINI_LLAP_CACHE_MEMORY_BYTES = 256L * 1024 * 1024; + public static MiniLlapCluster startAndGetMiniLlapCluster(Configuration conf, MiniZooKeeperCluster miniZkCluster, String confDir) throws @@ -51,10 +58,10 @@ public static MiniLlapCluster startAndGetMiniLlapCluster(Configuration conf, Configuration daemonConf = new LlapDaemonConfiguration(conf); final String clusterName = "llap"; final long maxMemory = LlapDaemon.getTotalHeapSize(); - // 15% for io cache - final long memoryForCache = (long) (0.15f * maxMemory); - // 75% for 4 executors - final long totalExecutorMemory = (long) (0.75f * maxMemory); + // 15% for io cache, 75% for executors, but cap totals for in-process mini cluster. + final long memoryForCache = Math.min((long) (0.15f * maxMemory), MAX_MINI_LLAP_CACHE_MEMORY_BYTES); + final long totalExecutorMemory = + Math.min((long) (0.75f * maxMemory), MAX_MINI_LLAP_EXECUTOR_MEMORY_BYTES); final int numExecutors = HiveConf.getIntVar(conf, HiveConf.ConfVars.LLAP_DAEMON_NUM_EXECUTORS); HiveConf.setIntVar(conf, HiveConf.ConfVars.LLAP_IO_THREADPOOL_SIZE, numExecutors); final boolean asyncIOEnabled = true; diff --git a/ql/src/test/queries/clientpositive/llap_io_cache.q b/ql/src/test/queries/clientpositive/llap_io_cache.q index b5ab5b25bae9..64787d749d60 100644 --- a/ql/src/test/queries/clientpositive/llap_io_cache.q +++ b/ql/src/test/queries/clientpositive/llap_io_cache.q @@ -1,3 +1,4 @@ +--! qt:dataset:src set hive.llap.io.enabled=true; set hive.llap.io.memory.mode=cache; set hive.llap.io.allocator.alloc.max=16Mb; @@ -16,11 +17,15 @@ TBLPROPERTIES ( 'parquet.compression'='UNCOMPRESSED' ); +-- Evaluate RPAD at runtime (not as a folded plan constant) to avoid Tez AM OOM when +-- deserializing the MapWork plan in mini-cluster qtests. INSERT INTO TABLE tbl_parq SELECT 1 AS id, - RPAD('x', 16777177, 'x') AS payload; + RPAD('x', 16777177 + length(key) - length(key), 'x') AS payload +FROM src +LIMIT 1; SELECT LENGTH(payload) FROM tbl_parq; -SELECT SUM(LENGTH(payload)) FROM tbl_parq; \ No newline at end of file +SELECT SUM(LENGTH(payload)) FROM tbl_parq; diff --git a/ql/src/test/results/clientpositive/llap/llap_io_cache.q.out b/ql/src/test/results/clientpositive/llap/llap_io_cache.q.out index 765ae2216310..b613e3e3ec86 100644 --- a/ql/src/test/results/clientpositive/llap/llap_io_cache.q.out +++ b/ql/src/test/results/clientpositive/llap/llap_io_cache.q.out @@ -33,19 +33,23 @@ POSTHOOK: Output: default@tbl_parq PREHOOK: query: INSERT INTO TABLE tbl_parq SELECT 1 AS id, - RPAD('x', 16777177, 'x') AS payload + RPAD('x', 16777177 + length(key) - length(key), 'x') AS payload +FROM src +LIMIT 1 PREHOOK: type: QUERY -PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Input: default@src PREHOOK: Output: default@tbl_parq POSTHOOK: query: INSERT INTO TABLE tbl_parq SELECT 1 AS id, - RPAD('x', 16777177, 'x') AS payload + RPAD('x', 16777177 + length(key) - length(key), 'x') AS payload +FROM src +LIMIT 1 POSTHOOK: type: QUERY -POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Input: default@src POSTHOOK: Output: default@tbl_parq POSTHOOK: Lineage: tbl_parq.id SIMPLE [] -POSTHOOK: Lineage: tbl_parq.payload SIMPLE [] +POSTHOOK: Lineage: tbl_parq.payload EXPRESSION [(src)src.FieldSchema(name:key, type:string, comment:default), ] PREHOOK: query: SELECT LENGTH(payload) FROM tbl_parq PREHOOK: type: QUERY PREHOOK: Input: default@tbl_parq