private static HiveDruidSplit[] distributeScanQuery(String address, ScanQuery query, Path dummyPath)
      throws IOException {
    // If it has a limit, we use it and we do not distribute the query
    final boolean isFetch = query.getScanRowsLimit() < Long.MAX_VALUE;
    if (isFetch) {
      return new HiveDruidSplit[] {new HiveDruidSplit(DruidStorageHandlerUtils.JSON_MAPPER.writeValueAsString(query),
          dummyPath,
          new String[] {address})
      };
    }

    final List<LocatedSegmentDescriptor> segmentDescriptors = fetchLocatedSegmentDescriptors(address, query);

    // Create one input split for each segment
    final int numSplits = segmentDescriptors.size();
    final HiveDruidSplit[] splits = new HiveDruidSplit[segmentDescriptors.size()];
    for (int i = 0; i < numSplits; i++) {
      final LocatedSegmentDescriptor locatedSD = segmentDescriptors.get(i);
      final String[] hosts = new String[locatedSD.getLocations().size() + 1];
      for (int j = 0; j < locatedSD.getLocations().size(); j++) {
        hosts[j] = locatedSD.getLocations().get(j).getHost();
      }
      // Default to broker if all other hosts fail.
      hosts[locatedSD.getLocations().size()] = address;

      // Create partial Select query
      final SegmentDescriptor
          newSD =
          new SegmentDescriptor(locatedSD.getInterval(), locatedSD.getVersion(), locatedSD.getPartitionNumber());
      final Query partialQuery = query.withQuerySegmentSpec(new MultipleSpecificSegmentSpec(Lists.newArrayList(newSD)));
      splits[i] =
          new HiveDruidSplit(DruidStorageHandlerUtils.JSON_MAPPER.writeValueAsString(partialQuery), dummyPath, hosts);
    }
    return splits;
  }
Druid query based input format.

Given a query and the Druid broker address, it will send it, and retrieve
and parse the results.
druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java