mirror of https://github.com/apache/druid.git
Moved Scan Builder to Druids class and started on Scan Benchmark setup
This commit is contained in:
parent
edee576a7a
commit
10e57d5f9e
|
@ -0,0 +1,371 @@
|
||||||
|
/*
|
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one
|
||||||
|
* or more contributor license agreements. See the NOTICE file
|
||||||
|
* distributed with this work for additional information
|
||||||
|
* regarding copyright ownership. The ASF licenses this file
|
||||||
|
* to you under the Apache License, Version 2.0 (the
|
||||||
|
* "License"); you may not use this file except in compliance
|
||||||
|
* with the License. You may obtain a copy of the License at
|
||||||
|
*
|
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
*
|
||||||
|
* Unless required by applicable law or agreed to in writing,
|
||||||
|
* software distributed under the License is distributed on an
|
||||||
|
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||||
|
* KIND, either express or implied. See the License for the
|
||||||
|
* specific language governing permissions and limitations
|
||||||
|
* under the License.
|
||||||
|
*/
|
||||||
|
|
||||||
|
package org.apache.druid.benchmark.query;
|
||||||
|
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import com.google.common.base.Suppliers;
|
||||||
|
import com.google.common.collect.ImmutableList;
|
||||||
|
import com.google.common.collect.Lists;
|
||||||
|
import com.google.common.io.Files;
|
||||||
|
import org.apache.commons.io.FileUtils;
|
||||||
|
import org.apache.druid.benchmark.datagen.BenchmarkDataGenerator;
|
||||||
|
import org.apache.druid.benchmark.datagen.BenchmarkSchemaInfo;
|
||||||
|
import org.apache.druid.benchmark.datagen.BenchmarkSchemas;
|
||||||
|
import org.apache.druid.data.input.InputRow;
|
||||||
|
import org.apache.druid.hll.HyperLogLogHash;
|
||||||
|
import org.apache.druid.jackson.DefaultObjectMapper;
|
||||||
|
import org.apache.druid.java.util.common.concurrent.Execs;
|
||||||
|
import org.apache.druid.java.util.common.granularity.Granularities;
|
||||||
|
import org.apache.druid.java.util.common.guava.Sequence;
|
||||||
|
import org.apache.druid.java.util.common.logger.Logger;
|
||||||
|
import org.apache.druid.query.DefaultGenericQueryMetricsFactory;
|
||||||
|
import org.apache.druid.query.Druids;
|
||||||
|
import org.apache.druid.query.FinalizeResultsQueryRunner;
|
||||||
|
import org.apache.druid.query.Query;
|
||||||
|
import org.apache.druid.query.QueryPlus;
|
||||||
|
import org.apache.druid.query.QueryRunner;
|
||||||
|
import org.apache.druid.query.QueryRunnerFactory;
|
||||||
|
import org.apache.druid.query.QueryToolChest;
|
||||||
|
import org.apache.druid.query.aggregation.hyperloglog.HyperUniquesSerde;
|
||||||
|
import org.apache.druid.query.extraction.DimExtractionFn;
|
||||||
|
import org.apache.druid.query.extraction.IdentityExtractionFn;
|
||||||
|
import org.apache.druid.query.extraction.LowerExtractionFn;
|
||||||
|
import org.apache.druid.query.extraction.StrlenExtractionFn;
|
||||||
|
import org.apache.druid.query.extraction.SubstringDimExtractionFn;
|
||||||
|
import org.apache.druid.query.extraction.UpperExtractionFn;
|
||||||
|
import org.apache.druid.query.filter.AndDimFilter;
|
||||||
|
import org.apache.druid.query.filter.BoundDimFilter;
|
||||||
|
import org.apache.druid.query.filter.DimFilter;
|
||||||
|
import org.apache.druid.query.filter.InDimFilter;
|
||||||
|
import org.apache.druid.query.filter.SelectorDimFilter;
|
||||||
|
import org.apache.druid.query.scan.ScanQuery;
|
||||||
|
import org.apache.druid.query.scan.ScanQueryConfig;
|
||||||
|
import org.apache.druid.query.scan.ScanQueryEngine;
|
||||||
|
import org.apache.druid.query.scan.ScanQueryQueryToolChest;
|
||||||
|
import org.apache.druid.query.scan.ScanQueryRunnerFactory;
|
||||||
|
import org.apache.druid.query.search.SearchQueryConfig;
|
||||||
|
import org.apache.druid.query.search.SearchQueryQueryToolChest;
|
||||||
|
import org.apache.druid.query.search.SearchQueryRunnerFactory;
|
||||||
|
import org.apache.druid.query.search.SearchStrategySelector;
|
||||||
|
import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
|
||||||
|
import org.apache.druid.query.spec.QuerySegmentSpec;
|
||||||
|
import org.apache.druid.segment.IndexIO;
|
||||||
|
import org.apache.druid.segment.IndexMergerV9;
|
||||||
|
import org.apache.druid.segment.IndexSpec;
|
||||||
|
import org.apache.druid.segment.QueryableIndex;
|
||||||
|
import org.apache.druid.segment.column.ColumnConfig;
|
||||||
|
import org.apache.druid.segment.incremental.IncrementalIndex;
|
||||||
|
import org.apache.druid.segment.serde.ComplexMetrics;
|
||||||
|
import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
|
||||||
|
import org.openjdk.jmh.annotations.Fork;
|
||||||
|
import org.openjdk.jmh.annotations.Measurement;
|
||||||
|
import org.openjdk.jmh.annotations.Param;
|
||||||
|
import org.openjdk.jmh.annotations.Scope;
|
||||||
|
import org.openjdk.jmh.annotations.Setup;
|
||||||
|
import org.openjdk.jmh.annotations.State;
|
||||||
|
import org.openjdk.jmh.annotations.TearDown;
|
||||||
|
import org.openjdk.jmh.annotations.Warmup;
|
||||||
|
|
||||||
|
import java.io.File;
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.Collections;
|
||||||
|
import java.util.HashMap;
|
||||||
|
import java.util.LinkedHashMap;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.ExecutorService;
|
||||||
|
|
||||||
|
@State(Scope.Benchmark)
|
||||||
|
@Fork(value = 1)
|
||||||
|
@Warmup(iterations = 10)
|
||||||
|
@Measurement(iterations = 25)
|
||||||
|
public class ScanBenchmark
|
||||||
|
{
|
||||||
|
@Param({"1"})
|
||||||
|
private int numSegments;
|
||||||
|
|
||||||
|
@Param({"750000"})
|
||||||
|
private int rowsPerSegment;
|
||||||
|
|
||||||
|
@Param({"basic.A"})
|
||||||
|
private String schemaAndQuery;
|
||||||
|
|
||||||
|
@Param({"1000"})
|
||||||
|
private int limit;
|
||||||
|
|
||||||
|
private static final Logger log = new Logger(ScanBenchmark.class);
|
||||||
|
private static final ObjectMapper JSON_MAPPER;
|
||||||
|
private static final IndexMergerV9 INDEX_MERGER_V9;
|
||||||
|
private static final IndexIO INDEX_IO;
|
||||||
|
|
||||||
|
private List<IncrementalIndex> incIndexes;
|
||||||
|
private List<QueryableIndex> qIndexes;
|
||||||
|
|
||||||
|
private QueryRunnerFactory factory;
|
||||||
|
private BenchmarkSchemaInfo schemaInfo;
|
||||||
|
private Druids.ScanQueryBuilder queryBuilder;
|
||||||
|
private ScanQuery query;
|
||||||
|
private File tmpDir;
|
||||||
|
|
||||||
|
private ExecutorService executorService;
|
||||||
|
|
||||||
|
static {
|
||||||
|
JSON_MAPPER = new DefaultObjectMapper();
|
||||||
|
INDEX_IO = new IndexIO(
|
||||||
|
JSON_MAPPER,
|
||||||
|
() -> 0
|
||||||
|
);
|
||||||
|
INDEX_MERGER_V9 = new IndexMergerV9(JSON_MAPPER, INDEX_IO, OffHeapMemorySegmentWriteOutMediumFactory.instance());
|
||||||
|
}
|
||||||
|
|
||||||
|
private static final Map<String, Map<String, Druids.ScanQueryBuilder>> SCHEMA_QUERY_MAP = new LinkedHashMap<>();
|
||||||
|
|
||||||
|
private void setupQueries()
|
||||||
|
{
|
||||||
|
// queries for the basic schema
|
||||||
|
final Map<String, Druids.ScanQueryBuilder> basicQueries = new LinkedHashMap<>();
|
||||||
|
final BenchmarkSchemaInfo basicSchema = BenchmarkSchemas.SCHEMA_MAP.get("basic");
|
||||||
|
|
||||||
|
final List<String> queryTypes = ImmutableList.of("A", "B", "C", "D");
|
||||||
|
for (final String eachType : queryTypes) {
|
||||||
|
basicQueries.put(eachType, makeQuery(eachType, basicSchema));
|
||||||
|
}
|
||||||
|
|
||||||
|
SCHEMA_QUERY_MAP.put("basic", basicQueries);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Druids.ScanQueryBuilder makeQuery(final String name, final BenchmarkSchemaInfo basicSchema)
|
||||||
|
{
|
||||||
|
switch (name) {
|
||||||
|
case "A":
|
||||||
|
return basicA(basicSchema);
|
||||||
|
case "B":
|
||||||
|
return basicB(basicSchema);
|
||||||
|
case "C":
|
||||||
|
return basicC(basicSchema);
|
||||||
|
case "D":
|
||||||
|
return basicD(basicSchema);
|
||||||
|
default:
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Druids.ScanQueryBuilder basicA(final BenchmarkSchemaInfo basicSchema)
|
||||||
|
{
|
||||||
|
final QuerySegmentSpec intervalSpec = new MultipleIntervalSegmentSpec(Collections.singletonList(basicSchema.getDataInterval()));
|
||||||
|
|
||||||
|
return Druids.newScanQueryBuilder()
|
||||||
|
.dataSource("blah")
|
||||||
|
.intervals(intervalSpec)
|
||||||
|
.query("123");
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Druids.ScanQueryBuilder basicB(final BenchmarkSchemaInfo basicSchema)
|
||||||
|
{
|
||||||
|
final QuerySegmentSpec intervalSpec = new MultipleIntervalSegmentSpec(Collections.singletonList(basicSchema.getDataInterval()));
|
||||||
|
|
||||||
|
final List<String> dimUniformFilterVals = new ArrayList<>();
|
||||||
|
int resultNum = (int) (100000 * 0.1);
|
||||||
|
int step = 100000 / resultNum;
|
||||||
|
for (int i = 1; i < 100001 && dimUniformFilterVals.size() < resultNum; i += step) {
|
||||||
|
dimUniformFilterVals.add(String.valueOf(i));
|
||||||
|
}
|
||||||
|
|
||||||
|
List<String> dimHyperUniqueFilterVals = new ArrayList<>();
|
||||||
|
resultNum = (int) (100000 * 0.1);
|
||||||
|
step = 100000 / resultNum;
|
||||||
|
for (int i = 0; i < 100001 && dimHyperUniqueFilterVals.size() < resultNum; i += step) {
|
||||||
|
dimHyperUniqueFilterVals.add(String.valueOf(i));
|
||||||
|
}
|
||||||
|
|
||||||
|
final List<DimFilter> dimFilters = new ArrayList<>();
|
||||||
|
dimFilters.add(new InDimFilter("dimUniform", dimUniformFilterVals, null));
|
||||||
|
dimFilters.add(new InDimFilter("dimHyperUnique", dimHyperUniqueFilterVals, null));
|
||||||
|
|
||||||
|
return Druids.newScanQueryBuilder(); // TODO
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Druids.ScanQueryBuilder basicC(final BenchmarkSchemaInfo basicSchema)
|
||||||
|
{
|
||||||
|
final QuerySegmentSpec intervalSpec = new MultipleIntervalSegmentSpec(Collections.singletonList(basicSchema.getDataInterval()));
|
||||||
|
|
||||||
|
final List<String> dimUniformFilterVals = new ArrayList<>();
|
||||||
|
final int resultNum = (int) (100000 * 0.1);
|
||||||
|
final int step = 100000 / resultNum;
|
||||||
|
for (int i = 1; i < 100001 && dimUniformFilterVals.size() < resultNum; i += step) {
|
||||||
|
dimUniformFilterVals.add(String.valueOf(i));
|
||||||
|
}
|
||||||
|
|
||||||
|
final String dimName = "dimUniform";
|
||||||
|
final List<DimFilter> dimFilters = new ArrayList<>();
|
||||||
|
dimFilters.add(new InDimFilter(dimName, dimUniformFilterVals, IdentityExtractionFn.getInstance()));
|
||||||
|
dimFilters.add(new SelectorDimFilter(dimName, "3", StrlenExtractionFn.instance()));
|
||||||
|
dimFilters.add(new BoundDimFilter(dimName, "100", "10000", true, true, true, new DimExtractionFn()
|
||||||
|
{
|
||||||
|
@Override
|
||||||
|
public byte[] getCacheKey()
|
||||||
|
{
|
||||||
|
return new byte[]{0xF};
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String apply(String value)
|
||||||
|
{
|
||||||
|
return String.valueOf(Long.parseLong(value) + 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public boolean preservesOrdering()
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public ExtractionType getExtractionType()
|
||||||
|
{
|
||||||
|
return ExtractionType.ONE_TO_ONE;
|
||||||
|
}
|
||||||
|
}, null));
|
||||||
|
dimFilters.add(new InDimFilter(dimName, dimUniformFilterVals, new LowerExtractionFn(null)));
|
||||||
|
dimFilters.add(new InDimFilter(dimName, dimUniformFilterVals, new UpperExtractionFn(null)));
|
||||||
|
dimFilters.add(new InDimFilter(dimName, dimUniformFilterVals, new SubstringDimExtractionFn(1, 3)));
|
||||||
|
|
||||||
|
return Druids.newScanQueryBuilder(); // TODO
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Druids.ScanQueryBuilder basicD(final BenchmarkSchemaInfo basicSchema)
|
||||||
|
{
|
||||||
|
final QuerySegmentSpec intervalSpec = new MultipleIntervalSegmentSpec(
|
||||||
|
Collections.singletonList(basicSchema.getDataInterval())
|
||||||
|
);
|
||||||
|
|
||||||
|
final List<String> dimUniformFilterVals = new ArrayList<>();
|
||||||
|
final int resultNum = (int) (100000 * 0.1);
|
||||||
|
final int step = 100000 / resultNum;
|
||||||
|
for (int i = 1; i < 100001 && dimUniformFilterVals.size() < resultNum; i += step) {
|
||||||
|
dimUniformFilterVals.add(String.valueOf(i));
|
||||||
|
}
|
||||||
|
|
||||||
|
final String dimName = "dimUniform";
|
||||||
|
final List<DimFilter> dimFilters = new ArrayList<>();
|
||||||
|
dimFilters.add(new InDimFilter(dimName, dimUniformFilterVals, null));
|
||||||
|
dimFilters.add(new SelectorDimFilter(dimName, "3", null));
|
||||||
|
dimFilters.add(new BoundDimFilter(dimName, "100", "10000", true, true, true, null, null));
|
||||||
|
|
||||||
|
return Druids.newScanQueryBuilder(); // TODO
|
||||||
|
}
|
||||||
|
|
||||||
|
@Setup
|
||||||
|
public void setup() throws IOException
|
||||||
|
{
|
||||||
|
log.info("SETUP CALLED AT " + +System.currentTimeMillis());
|
||||||
|
|
||||||
|
if (ComplexMetrics.getSerdeForType("hyperUnique") == null) {
|
||||||
|
ComplexMetrics.registerSerde("hyperUnique", new HyperUniquesSerde(HyperLogLogHash.getDefault()));
|
||||||
|
}
|
||||||
|
executorService = Execs.multiThreaded(numSegments, "SearchThreadPool");
|
||||||
|
|
||||||
|
setupQueries();
|
||||||
|
|
||||||
|
String[] schemaQuery = schemaAndQuery.split("\\.");
|
||||||
|
String schemaName = schemaQuery[0];
|
||||||
|
String queryName = schemaQuery[1];
|
||||||
|
|
||||||
|
schemaInfo = BenchmarkSchemas.SCHEMA_MAP.get(schemaName);
|
||||||
|
queryBuilder = SCHEMA_QUERY_MAP.get(schemaName).get(queryName);
|
||||||
|
queryBuilder.limit(limit);
|
||||||
|
query = queryBuilder.build();
|
||||||
|
|
||||||
|
incIndexes = new ArrayList<>();
|
||||||
|
for (int i = 0; i < numSegments; i++) {
|
||||||
|
log.info("Generating rows for segment " + i);
|
||||||
|
BenchmarkDataGenerator gen = new BenchmarkDataGenerator(
|
||||||
|
schemaInfo.getColumnSchemas(),
|
||||||
|
System.currentTimeMillis(),
|
||||||
|
schemaInfo.getDataInterval(),
|
||||||
|
rowsPerSegment
|
||||||
|
);
|
||||||
|
|
||||||
|
IncrementalIndex incIndex = makeIncIndex();
|
||||||
|
|
||||||
|
for (int j = 0; j < rowsPerSegment; j++) {
|
||||||
|
InputRow row = gen.nextRow();
|
||||||
|
if (j % 10000 == 0) {
|
||||||
|
log.info(j + " rows generated.");
|
||||||
|
}
|
||||||
|
incIndex.add(row);
|
||||||
|
}
|
||||||
|
incIndexes.add(incIndex);
|
||||||
|
}
|
||||||
|
|
||||||
|
tmpDir = Files.createTempDir();
|
||||||
|
log.info("Using temp dir: " + tmpDir.getAbsolutePath());
|
||||||
|
|
||||||
|
qIndexes = new ArrayList<>();
|
||||||
|
for (int i = 0; i < numSegments; i++) {
|
||||||
|
File indexFile = INDEX_MERGER_V9.persist(
|
||||||
|
incIndexes.get(i),
|
||||||
|
tmpDir,
|
||||||
|
new IndexSpec(),
|
||||||
|
null
|
||||||
|
);
|
||||||
|
|
||||||
|
QueryableIndex qIndex = INDEX_IO.loadIndex(indexFile);
|
||||||
|
qIndexes.add(qIndex);
|
||||||
|
}
|
||||||
|
|
||||||
|
final ScanQueryConfig config = new ScanQueryConfig().setLegacy(false);
|
||||||
|
factory = new ScanQueryRunnerFactory(
|
||||||
|
new ScanQueryQueryToolChest(
|
||||||
|
config,
|
||||||
|
DefaultGenericQueryMetricsFactory.instance()
|
||||||
|
),
|
||||||
|
new ScanQueryEngine()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
@TearDown
|
||||||
|
public void tearDown() throws IOException
|
||||||
|
{
|
||||||
|
FileUtils.deleteDirectory(tmpDir);
|
||||||
|
}
|
||||||
|
|
||||||
|
private IncrementalIndex makeIncIndex()
|
||||||
|
{
|
||||||
|
return new IncrementalIndex.Builder()
|
||||||
|
.setSimpleTestingIndexSchema(schemaInfo.getAggsArray())
|
||||||
|
.setReportParseExceptions(false)
|
||||||
|
.setMaxRowCount(rowsPerSegment)
|
||||||
|
.buildOnheap();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static <T> List<T> runQuery(QueryRunnerFactory factory, QueryRunner runner, Query<T> query)
|
||||||
|
{
|
||||||
|
QueryToolChest toolChest = factory.getToolchest();
|
||||||
|
QueryRunner<T> theRunner = new FinalizeResultsQueryRunner<>(
|
||||||
|
toolChest.mergeResults(toolChest.preMergeQueryDecoration(runner)),
|
||||||
|
toolChest
|
||||||
|
);
|
||||||
|
|
||||||
|
Sequence<T> queryResult = theRunner.run(QueryPlus.wrap(query), new HashMap<>());
|
||||||
|
return queryResult.toList();
|
||||||
|
}
|
||||||
|
}
|
|
@ -315,7 +315,6 @@ public class SelectBenchmark
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@Benchmark
|
@Benchmark
|
||||||
@BenchmarkMode(Mode.AverageTime)
|
@BenchmarkMode(Mode.AverageTime)
|
||||||
@OutputTimeUnit(TimeUnit.MICROSECONDS)
|
@OutputTimeUnit(TimeUnit.MICROSECONDS)
|
||||||
|
|
|
@ -36,6 +36,7 @@ import org.apache.druid.query.filter.InDimFilter;
|
||||||
import org.apache.druid.query.filter.SelectorDimFilter;
|
import org.apache.druid.query.filter.SelectorDimFilter;
|
||||||
import org.apache.druid.query.metadata.metadata.ColumnIncluderator;
|
import org.apache.druid.query.metadata.metadata.ColumnIncluderator;
|
||||||
import org.apache.druid.query.metadata.metadata.SegmentMetadataQuery;
|
import org.apache.druid.query.metadata.metadata.SegmentMetadataQuery;
|
||||||
|
import org.apache.druid.query.scan.ScanQuery;
|
||||||
import org.apache.druid.query.search.ContainsSearchQuerySpec;
|
import org.apache.druid.query.search.ContainsSearchQuerySpec;
|
||||||
import org.apache.druid.query.search.FragmentSearchQuerySpec;
|
import org.apache.druid.query.search.FragmentSearchQuerySpec;
|
||||||
import org.apache.druid.query.search.InsensitiveContainsSearchQuerySpec;
|
import org.apache.druid.query.search.InsensitiveContainsSearchQuerySpec;
|
||||||
|
@ -896,6 +897,162 @@ public class Druids
|
||||||
return new SelectQueryBuilder();
|
return new SelectQueryBuilder();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A Builder for ScanQuery.
|
||||||
|
* <p/>
|
||||||
|
* Required: dataSource(), intervals() must be called before build()
|
||||||
|
* <p/>
|
||||||
|
* Usage example:
|
||||||
|
* <pre><code>
|
||||||
|
* ScanQuery query = new ScanQueryBuilder()
|
||||||
|
* .dataSource("Example")
|
||||||
|
* .interval("2010/2013")
|
||||||
|
* .build();
|
||||||
|
* </code></pre>
|
||||||
|
*
|
||||||
|
* @see ScanQuery
|
||||||
|
*/
|
||||||
|
public static class ScanQueryBuilder
|
||||||
|
{
|
||||||
|
private DataSource dataSource;
|
||||||
|
private QuerySegmentSpec querySegmentSpec;
|
||||||
|
private VirtualColumns virtualColumns;
|
||||||
|
private Map<String, Object> context;
|
||||||
|
private String resultFormat;
|
||||||
|
private int batchSize;
|
||||||
|
private long limit;
|
||||||
|
private DimFilter dimFilter;
|
||||||
|
private List<String> columns;
|
||||||
|
private Boolean legacy;
|
||||||
|
|
||||||
|
public ScanQueryBuilder()
|
||||||
|
{
|
||||||
|
dataSource = null;
|
||||||
|
querySegmentSpec = null;
|
||||||
|
virtualColumns = null;
|
||||||
|
context = null;
|
||||||
|
resultFormat = null;
|
||||||
|
batchSize = 0;
|
||||||
|
limit = 0;
|
||||||
|
dimFilter = null;
|
||||||
|
columns = new ArrayList<>();
|
||||||
|
legacy = null;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQuery build()
|
||||||
|
{
|
||||||
|
return new ScanQuery(
|
||||||
|
dataSource,
|
||||||
|
querySegmentSpec,
|
||||||
|
virtualColumns,
|
||||||
|
resultFormat,
|
||||||
|
batchSize,
|
||||||
|
limit,
|
||||||
|
dimFilter,
|
||||||
|
columns,
|
||||||
|
legacy,
|
||||||
|
context
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
public static ScanQueryBuilder copy(ScanQuery query)
|
||||||
|
{
|
||||||
|
return new ScanQueryBuilder()
|
||||||
|
.dataSource(query.getDataSource())
|
||||||
|
.intervals(query.getQuerySegmentSpec())
|
||||||
|
.virtualColumns(query.getVirtualColumns())
|
||||||
|
.resultFormat(query.getResultFormat())
|
||||||
|
.batchSize(query.getBatchSize())
|
||||||
|
.limit(query.getLimit())
|
||||||
|
.filters(query.getFilter())
|
||||||
|
.columns(query.getColumns())
|
||||||
|
.legacy(query.isLegacy())
|
||||||
|
.context(query.getContext());
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder dataSource(String ds)
|
||||||
|
{
|
||||||
|
dataSource = new TableDataSource(ds);
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder dataSource(DataSource ds)
|
||||||
|
{
|
||||||
|
dataSource = ds;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder intervals(QuerySegmentSpec q)
|
||||||
|
{
|
||||||
|
querySegmentSpec = q;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder virtualColumns(VirtualColumns virtualColumns)
|
||||||
|
{
|
||||||
|
this.virtualColumns = virtualColumns;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder virtualColumns(VirtualColumn... virtualColumns)
|
||||||
|
{
|
||||||
|
return virtualColumns(VirtualColumns.create(Arrays.asList(virtualColumns)));
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder context(Map<String, Object> c)
|
||||||
|
{
|
||||||
|
context = c;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder resultFormat(String r)
|
||||||
|
{
|
||||||
|
resultFormat = r;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder batchSize(int b)
|
||||||
|
{
|
||||||
|
batchSize = b;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder limit(long l)
|
||||||
|
{
|
||||||
|
limit = l;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder filters(DimFilter f)
|
||||||
|
{
|
||||||
|
dimFilter = f;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder columns(List<String> c)
|
||||||
|
{
|
||||||
|
columns = c;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder columns(String... c)
|
||||||
|
{
|
||||||
|
columns = Arrays.asList(c);
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
|
public ScanQueryBuilder legacy(Boolean legacy)
|
||||||
|
{
|
||||||
|
this.legacy = legacy;
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public static ScanQueryBuilder newScanQueryBuilder()
|
||||||
|
{
|
||||||
|
return new ScanQueryBuilder();
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A Builder for DataSourceMetadataQuery.
|
* A Builder for DataSourceMetadataQuery.
|
||||||
* <p/>
|
* <p/>
|
||||||
|
|
|
@ -24,15 +24,12 @@ import com.fasterxml.jackson.annotation.JsonProperty;
|
||||||
import com.google.common.base.Preconditions;
|
import com.google.common.base.Preconditions;
|
||||||
import org.apache.druid.query.BaseQuery;
|
import org.apache.druid.query.BaseQuery;
|
||||||
import org.apache.druid.query.DataSource;
|
import org.apache.druid.query.DataSource;
|
||||||
|
import org.apache.druid.query.Druids;
|
||||||
import org.apache.druid.query.Query;
|
import org.apache.druid.query.Query;
|
||||||
import org.apache.druid.query.TableDataSource;
|
|
||||||
import org.apache.druid.query.filter.DimFilter;
|
import org.apache.druid.query.filter.DimFilter;
|
||||||
import org.apache.druid.query.spec.QuerySegmentSpec;
|
import org.apache.druid.query.spec.QuerySegmentSpec;
|
||||||
import org.apache.druid.segment.VirtualColumn;
|
|
||||||
import org.apache.druid.segment.VirtualColumns;
|
import org.apache.druid.segment.VirtualColumns;
|
||||||
|
|
||||||
import java.util.ArrayList;
|
|
||||||
import java.util.Arrays;
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.Objects;
|
import java.util.Objects;
|
||||||
|
@ -137,30 +134,30 @@ public class ScanQuery extends BaseQuery<ScanResultValue>
|
||||||
|
|
||||||
public ScanQuery withNonNullLegacy(final ScanQueryConfig scanQueryConfig)
|
public ScanQuery withNonNullLegacy(final ScanQueryConfig scanQueryConfig)
|
||||||
{
|
{
|
||||||
return ScanQueryBuilder.copy(this).legacy(legacy != null ? legacy : scanQueryConfig.isLegacy()).build();
|
return Druids.ScanQueryBuilder.copy(this).legacy(legacy != null ? legacy : scanQueryConfig.isLegacy()).build();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Query<ScanResultValue> withQuerySegmentSpec(QuerySegmentSpec querySegmentSpec)
|
public Query<ScanResultValue> withQuerySegmentSpec(QuerySegmentSpec querySegmentSpec)
|
||||||
{
|
{
|
||||||
return ScanQueryBuilder.copy(this).intervals(querySegmentSpec).build();
|
return Druids.ScanQueryBuilder.copy(this).intervals(querySegmentSpec).build();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Query<ScanResultValue> withDataSource(DataSource dataSource)
|
public Query<ScanResultValue> withDataSource(DataSource dataSource)
|
||||||
{
|
{
|
||||||
return ScanQueryBuilder.copy(this).dataSource(dataSource).build();
|
return Druids.ScanQueryBuilder.copy(this).dataSource(dataSource).build();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Query<ScanResultValue> withOverriddenContext(Map<String, Object> contextOverrides)
|
public Query<ScanResultValue> withOverriddenContext(Map<String, Object> contextOverrides)
|
||||||
{
|
{
|
||||||
return ScanQueryBuilder.copy(this).context(computeOverriddenContext(getContext(), contextOverrides)).build();
|
return Druids.ScanQueryBuilder.copy(this).context(computeOverriddenContext(getContext(), contextOverrides)).build();
|
||||||
}
|
}
|
||||||
|
|
||||||
public ScanQuery withDimFilter(DimFilter dimFilter)
|
public ScanQuery withDimFilter(DimFilter dimFilter)
|
||||||
{
|
{
|
||||||
return ScanQueryBuilder.copy(this).filters(dimFilter).build();
|
return Druids.ScanQueryBuilder.copy(this).filters(dimFilter).build();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -206,160 +203,4 @@ public class ScanQuery extends BaseQuery<ScanResultValue>
|
||||||
", legacy=" + legacy +
|
", legacy=" + legacy +
|
||||||
'}';
|
'}';
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* A Builder for ScanQuery.
|
|
||||||
* <p/>
|
|
||||||
* Required: dataSource(), intervals() must be called before build()
|
|
||||||
* <p/>
|
|
||||||
* Usage example:
|
|
||||||
* <pre><code>
|
|
||||||
* ScanQuery query = new ScanQueryBuilder()
|
|
||||||
* .dataSource("Example")
|
|
||||||
* .interval("2010/2013")
|
|
||||||
* .build();
|
|
||||||
* </code></pre>
|
|
||||||
*
|
|
||||||
* @see ScanQuery
|
|
||||||
*/
|
|
||||||
public static class ScanQueryBuilder
|
|
||||||
{
|
|
||||||
private DataSource dataSource;
|
|
||||||
private QuerySegmentSpec querySegmentSpec;
|
|
||||||
private VirtualColumns virtualColumns;
|
|
||||||
private Map<String, Object> context;
|
|
||||||
private String resultFormat;
|
|
||||||
private int batchSize;
|
|
||||||
private long limit;
|
|
||||||
private DimFilter dimFilter;
|
|
||||||
private List<String> columns;
|
|
||||||
private Boolean legacy;
|
|
||||||
|
|
||||||
public ScanQueryBuilder()
|
|
||||||
{
|
|
||||||
dataSource = null;
|
|
||||||
querySegmentSpec = null;
|
|
||||||
virtualColumns = null;
|
|
||||||
context = null;
|
|
||||||
resultFormat = null;
|
|
||||||
batchSize = 0;
|
|
||||||
limit = 0;
|
|
||||||
dimFilter = null;
|
|
||||||
columns = new ArrayList<>();
|
|
||||||
legacy = null;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQuery build()
|
|
||||||
{
|
|
||||||
return new ScanQuery(
|
|
||||||
dataSource,
|
|
||||||
querySegmentSpec,
|
|
||||||
virtualColumns,
|
|
||||||
resultFormat,
|
|
||||||
batchSize,
|
|
||||||
limit,
|
|
||||||
dimFilter,
|
|
||||||
columns,
|
|
||||||
legacy,
|
|
||||||
context
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
public static ScanQueryBuilder copy(ScanQuery query)
|
|
||||||
{
|
|
||||||
return new ScanQueryBuilder()
|
|
||||||
.dataSource(query.getDataSource())
|
|
||||||
.intervals(query.getQuerySegmentSpec())
|
|
||||||
.virtualColumns(query.getVirtualColumns())
|
|
||||||
.resultFormat(query.getResultFormat())
|
|
||||||
.batchSize(query.getBatchSize())
|
|
||||||
.limit(query.getLimit())
|
|
||||||
.filters(query.getFilter())
|
|
||||||
.columns(query.getColumns())
|
|
||||||
.legacy(query.isLegacy())
|
|
||||||
.context(query.getContext());
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder dataSource(String ds)
|
|
||||||
{
|
|
||||||
dataSource = new TableDataSource(ds);
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder dataSource(DataSource ds)
|
|
||||||
{
|
|
||||||
dataSource = ds;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder intervals(QuerySegmentSpec q)
|
|
||||||
{
|
|
||||||
querySegmentSpec = q;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder virtualColumns(VirtualColumns virtualColumns)
|
|
||||||
{
|
|
||||||
this.virtualColumns = virtualColumns;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder virtualColumns(VirtualColumn... virtualColumns)
|
|
||||||
{
|
|
||||||
return virtualColumns(VirtualColumns.create(Arrays.asList(virtualColumns)));
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder context(Map<String, Object> c)
|
|
||||||
{
|
|
||||||
context = c;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder resultFormat(String r)
|
|
||||||
{
|
|
||||||
resultFormat = r;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder batchSize(int b)
|
|
||||||
{
|
|
||||||
batchSize = b;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder limit(long l)
|
|
||||||
{
|
|
||||||
limit = l;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder filters(DimFilter f)
|
|
||||||
{
|
|
||||||
dimFilter = f;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder columns(List<String> c)
|
|
||||||
{
|
|
||||||
columns = c;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder columns(String... c)
|
|
||||||
{
|
|
||||||
columns = Arrays.asList(c);
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
|
|
||||||
public ScanQueryBuilder legacy(Boolean legacy)
|
|
||||||
{
|
|
||||||
this.legacy = legacy;
|
|
||||||
return this;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public static ScanQueryBuilder newScanQueryBuilder()
|
|
||||||
{
|
|
||||||
return new ScanQueryBuilder();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -97,14 +97,14 @@ public class DoubleStorageTest
|
||||||
new ScanQueryEngine()
|
new ScanQueryEngine()
|
||||||
);
|
);
|
||||||
|
|
||||||
private ScanQuery.ScanQueryBuilder newTestQuery()
|
private Druids.ScanQueryBuilder newTestQuery()
|
||||||
{
|
{
|
||||||
return ScanQuery.newScanQueryBuilder()
|
return Druids.newScanQueryBuilder()
|
||||||
.dataSource(new TableDataSource(QueryRunnerTestHelper.dataSource))
|
.dataSource(new TableDataSource(QueryRunnerTestHelper.dataSource))
|
||||||
.columns(Collections.emptyList())
|
.columns(Collections.emptyList())
|
||||||
.intervals(QueryRunnerTestHelper.fullOnIntervalSpec)
|
.intervals(QueryRunnerTestHelper.fullOnIntervalSpec)
|
||||||
.limit(Integer.MAX_VALUE)
|
.limit(Integer.MAX_VALUE)
|
||||||
.legacy(false);
|
.legacy(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -31,6 +31,7 @@ import org.apache.druid.java.util.common.guava.MergeSequence;
|
||||||
import org.apache.druid.java.util.common.guava.Sequence;
|
import org.apache.druid.java.util.common.guava.Sequence;
|
||||||
import org.apache.druid.java.util.common.guava.Sequences;
|
import org.apache.druid.java.util.common.guava.Sequences;
|
||||||
import org.apache.druid.query.DefaultGenericQueryMetricsFactory;
|
import org.apache.druid.query.DefaultGenericQueryMetricsFactory;
|
||||||
|
import org.apache.druid.query.Druids;
|
||||||
import org.apache.druid.query.QueryPlus;
|
import org.apache.druid.query.QueryPlus;
|
||||||
import org.apache.druid.query.QueryRunner;
|
import org.apache.druid.query.QueryRunner;
|
||||||
import org.apache.druid.query.QueryRunnerFactory;
|
import org.apache.druid.query.QueryRunnerFactory;
|
||||||
|
@ -177,9 +178,9 @@ public class MultiSegmentScanQueryTest
|
||||||
this.batchSize = batchSize;
|
this.batchSize = batchSize;
|
||||||
}
|
}
|
||||||
|
|
||||||
private ScanQuery.ScanQueryBuilder newBuilder()
|
private Druids.ScanQueryBuilder newBuilder()
|
||||||
{
|
{
|
||||||
return ScanQuery.newScanQueryBuilder()
|
return Druids.newScanQueryBuilder()
|
||||||
.dataSource(new TableDataSource(QueryRunnerTestHelper.dataSource))
|
.dataSource(new TableDataSource(QueryRunnerTestHelper.dataSource))
|
||||||
.intervals(SelectQueryRunnerTest.I_0112_0114_SPEC)
|
.intervals(SelectQueryRunnerTest.I_0112_0114_SPEC)
|
||||||
.batchSize(batchSize)
|
.batchSize(batchSize)
|
||||||
|
|
|
@ -32,6 +32,7 @@ import org.apache.druid.java.util.common.ISE;
|
||||||
import org.apache.druid.java.util.common.Intervals;
|
import org.apache.druid.java.util.common.Intervals;
|
||||||
import org.apache.druid.java.util.common.StringUtils;
|
import org.apache.druid.java.util.common.StringUtils;
|
||||||
import org.apache.druid.query.DefaultGenericQueryMetricsFactory;
|
import org.apache.druid.query.DefaultGenericQueryMetricsFactory;
|
||||||
|
import org.apache.druid.query.Druids;
|
||||||
import org.apache.druid.query.QueryPlus;
|
import org.apache.druid.query.QueryPlus;
|
||||||
import org.apache.druid.query.QueryRunner;
|
import org.apache.druid.query.QueryRunner;
|
||||||
import org.apache.druid.query.QueryRunnerTestHelper;
|
import org.apache.druid.query.QueryRunnerTestHelper;
|
||||||
|
@ -136,9 +137,9 @@ public class ScanQueryRunnerTest
|
||||||
this.legacy = legacy;
|
this.legacy = legacy;
|
||||||
}
|
}
|
||||||
|
|
||||||
private ScanQuery.ScanQueryBuilder newTestQuery()
|
private Druids.ScanQueryBuilder newTestQuery()
|
||||||
{
|
{
|
||||||
return ScanQuery.newScanQueryBuilder()
|
return Druids.newScanQueryBuilder()
|
||||||
.dataSource(new TableDataSource(QueryRunnerTestHelper.dataSource))
|
.dataSource(new TableDataSource(QueryRunnerTestHelper.dataSource))
|
||||||
.columns(Collections.emptyList())
|
.columns(Collections.emptyList())
|
||||||
.intervals(QueryRunnerTestHelper.fullOnIntervalSpec)
|
.intervals(QueryRunnerTestHelper.fullOnIntervalSpec)
|
||||||
|
|
|
@ -32,6 +32,7 @@ import org.apache.druid.java.util.common.StringUtils;
|
||||||
import org.apache.druid.java.util.common.io.Closer;
|
import org.apache.druid.java.util.common.io.Closer;
|
||||||
import org.apache.druid.java.util.common.logger.Logger;
|
import org.apache.druid.java.util.common.logger.Logger;
|
||||||
import org.apache.druid.math.expr.ExprMacroTable;
|
import org.apache.druid.math.expr.ExprMacroTable;
|
||||||
|
import org.apache.druid.query.Druids;
|
||||||
import org.apache.druid.query.Query;
|
import org.apache.druid.query.Query;
|
||||||
import org.apache.druid.query.QueryContexts;
|
import org.apache.druid.query.QueryContexts;
|
||||||
import org.apache.druid.query.QueryRunnerFactoryConglomerate;
|
import org.apache.druid.query.QueryRunnerFactoryConglomerate;
|
||||||
|
@ -359,9 +360,9 @@ public class BaseCalciteQueryTest extends CalciteTestBase
|
||||||
return new ExpressionPostAggregator(name, expression, null, CalciteTests.createExprMacroTable());
|
return new ExpressionPostAggregator(name, expression, null, CalciteTests.createExprMacroTable());
|
||||||
}
|
}
|
||||||
|
|
||||||
public static ScanQuery.ScanQueryBuilder newScanQueryBuilder()
|
public static Druids.ScanQueryBuilder newScanQueryBuilder()
|
||||||
{
|
{
|
||||||
return new ScanQuery.ScanQueryBuilder().resultFormat(ScanQuery.RESULT_FORMAT_COMPACTED_LIST)
|
return new Druids.ScanQueryBuilder().resultFormat(ScanQuery.RESULT_FORMAT_COMPACTED_LIST)
|
||||||
.legacy(false);
|
.legacy(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue