Hive 29707

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,

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 =

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<List<DimensionSchema>, AggregatorFactory[]> getDimensionsAndAggregates(List<String> columnNames,
@@ -1082,7 +1080,8 @@ private static BloomKFilter evaluateBloomFilter(ExprNodeDesc desc, Configuration
     Set<String> 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;
   }

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<String, Object>
@@ -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,

druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java

@@ -198,7 +198,7 @@ private static List<LocatedSegmentDescriptor> 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;

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<SegmentIdWithShardSpec> segmentsToPush) {
     try {
-      SegmentsAndMetadata segmentsAndMetadata = appenderator.push(segmentsToPush, committerSupplier.get(), false).get();
+      SegmentsAndCommitMetadata segmentsAndMetadata = appenderator.push(segmentsToPush, committerSupplier.get(), false).get();
       final Set<String> pushedSegmentIdentifierHashSet = new HashSet<>();
 
       for (DataSegment pushedSegment : segmentsAndMetadata.getSegments()) {

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;
   }

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="

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() {

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<R> 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);
     }
   }