DruidQuery.java
/*
* 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.calcite.adapter.druid;
import org.apache.calcite.DataContext;
import org.apache.calcite.avatica.ColumnMetaData;
import org.apache.calcite.avatica.util.DateTimeUtils;
import org.apache.calcite.config.CalciteConnectionConfig;
import org.apache.calcite.interpreter.BindableRel;
import org.apache.calcite.interpreter.Bindables;
import org.apache.calcite.interpreter.Compiler;
import org.apache.calcite.interpreter.Node;
import org.apache.calcite.interpreter.Sink;
import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.linq4j.Ord;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptCost;
import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.AbstractRelNode;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelFieldCollation.Direction;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelWriter;
import org.apache.calcite.rel.core.Aggregate;
import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.core.Filter;
import org.apache.calcite.rel.core.Project;
import org.apache.calcite.rel.core.Sort;
import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rel.metadata.RelMdUtil;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.runtime.Hook;
import org.apache.calcite.schema.ScannableTable;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlOperator;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.type.SqlTypeFamily;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.validate.SqlValidatorUtil;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Litmus;
import org.apache.calcite.util.Pair;
import org.apache.calcite.util.Util;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.Interval;
import java.io.IOException;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TimeZone;
import java.util.regex.Pattern;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.util.Objects.requireNonNull;
/**
* Relational expression representing a scan of a Druid data set.
*/
public class DruidQuery extends AbstractRelNode implements BindableRel {
/**
* Provides a standard list of supported Calcite operators that can be converted to
* Druid Expressions. This can be used as is or re-adapted based on underline
* engine operator syntax. Methods can refer to https://druid.apache.org/docs/latest/querying/
*/
public static final List<DruidSqlOperatorConverter> DEFAULT_OPERATORS_LIST =
ImmutableList.<DruidSqlOperatorConverter>builder()
.add(new DirectOperatorConversion(SqlStdOperatorTable.EXP, "exp"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.CONCAT, "concat"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.DIVIDE_INTEGER, "div"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.LIKE, "like"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.LN, "log"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.SQRT, "sqrt"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.LOWER, "lower"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.LOG10, "log10"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.REPLACE, "replace"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.UPPER, "upper"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.POWER, "pow"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.ABS, "abs"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.SIN, "sin"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.COS, "cos"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.TAN, "tan"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.COT, "cot"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.ASIN, "asin"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.ACOS, "acos"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.ATAN, "atan"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.ATAN2, "atan2"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.CASE, "case_searched"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.CHAR_LENGTH, "strlen"))
.add(new DirectOperatorConversion(SqlStdOperatorTable.CHARACTER_LENGTH, "strlen"))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.EQUALS, "=="))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.NOT_EQUALS, "!="))
.add(new NaryOperatorConverter(SqlStdOperatorTable.OR, "||"))
.add(new NaryOperatorConverter(SqlStdOperatorTable.AND, "&&"))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.LESS_THAN, "<"))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.LESS_THAN_OR_EQUAL, "<="))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.GREATER_THAN, ">"))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.GREATER_THAN_OR_EQUAL, ">="))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.PLUS, "+"))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.MINUS, "-"))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.MULTIPLY, "*"))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.DIVIDE, "/"))
.add(new BinaryOperatorConversion(SqlStdOperatorTable.MOD, "%"))
.add(new DruidSqlCastConverter())
.add(new ExtractOperatorConversion())
.add(new UnaryPrefixOperatorConversion(SqlStdOperatorTable.NOT, "!"))
.add(new UnaryPrefixOperatorConversion(SqlStdOperatorTable.UNARY_MINUS, "-"))
.add(new UnarySuffixOperatorConversion(SqlStdOperatorTable.IS_FALSE, "<= 0"))
.add(new UnarySuffixOperatorConversion(SqlStdOperatorTable.IS_NOT_TRUE, "<= 0"))
.add(new UnarySuffixOperatorConversion(SqlStdOperatorTable.IS_TRUE, "> 0"))
.add(new UnarySuffixOperatorConversion(SqlStdOperatorTable.IS_NOT_FALSE, "> 0"))
.add(new UnarySuffixOperatorConversion(SqlStdOperatorTable.IS_NULL, "== null"))
.add(new UnarySuffixOperatorConversion(SqlStdOperatorTable.IS_NOT_NULL, "!= null"))
.add(new FloorOperatorConversion())
.add(new CeilOperatorConversion())
.add(new SubstringOperatorConversion())
.build();
protected @Nullable QuerySpec querySpec;
final RelOptTable table;
final DruidTable druidTable;
final ImmutableList<Interval> intervals;
final ImmutableList<RelNode> rels;
/**
* This operator map provides DruidSqlOperatorConverter instance to convert a Calcite RexNode to
* Druid Expression when possible.
*/
final Map<SqlOperator, DruidSqlOperatorConverter> converterOperatorMap;
private static final Pattern VALID_SIG = Pattern.compile("sf?p?(a?|ah|ah?o)l?");
private static final String EXTRACT_COLUMN_NAME_PREFIX = "extract";
private static final String FLOOR_COLUMN_NAME_PREFIX = "floor";
protected static final String DRUID_QUERY_FETCH = "druid.query.fetch";
private static final int DAYS_IN_TEN_YEARS = 10 * 365;
/**
* Creates a DruidQuery.
*
* @param cluster Cluster
* @param traitSet Traits
* @param table Table
* @param druidTable Druid table
* @param intervals Intervals for the query
* @param rels Internal relational expressions
* @param converterOperatorMap mapping of Calcite Sql Operator to Druid Expression API.
*/
protected DruidQuery(RelOptCluster cluster, RelTraitSet traitSet,
RelOptTable table, DruidTable druidTable,
List<Interval> intervals, List<RelNode> rels,
Map<SqlOperator, DruidSqlOperatorConverter> converterOperatorMap) {
super(cluster, traitSet);
this.table = table;
this.druidTable = druidTable;
this.intervals = ImmutableList.copyOf(intervals);
this.rels = ImmutableList.copyOf(rels);
this.converterOperatorMap =
requireNonNull(converterOperatorMap, "Operator map cannot be null");
assert isValid(Litmus.THROW, null);
}
/** Returns whether a signature represents an sequence of relational operators
* that can be translated into a valid Druid query. */
static boolean isValidSignature(String signature) {
return VALID_SIG.matcher(signature).matches();
}
/** Creates a DruidQuery. */
public static DruidQuery create(RelOptCluster cluster, RelTraitSet traitSet,
RelOptTable table, DruidTable druidTable, List<RelNode> rels) {
final ImmutableMap.Builder<SqlOperator, DruidSqlOperatorConverter> mapBuilder = ImmutableMap
.builder();
for (DruidSqlOperatorConverter converter : DEFAULT_OPERATORS_LIST) {
mapBuilder.put(converter.calciteOperator(), converter);
}
return create(cluster, traitSet, table, druidTable, druidTable.intervals, rels,
mapBuilder.build());
}
/** Creates a DruidQuery. */
public static DruidQuery create(RelOptCluster cluster, RelTraitSet traitSet,
RelOptTable table, DruidTable druidTable, List<RelNode> rels,
Map<SqlOperator, DruidSqlOperatorConverter> converterOperatorMap) {
return create(cluster, traitSet, table, druidTable, druidTable.intervals, rels,
converterOperatorMap);
}
/**
* Creates a DruidQuery.
*/
private static DruidQuery create(RelOptCluster cluster, RelTraitSet traitSet,
RelOptTable table, DruidTable druidTable, List<Interval> intervals,
List<RelNode> rels, Map<SqlOperator, DruidSqlOperatorConverter> converterOperatorMap) {
return new DruidQuery(cluster, traitSet, table, druidTable, intervals, rels,
converterOperatorMap);
}
/** Extends a DruidQuery. */
public static DruidQuery extendQuery(DruidQuery query, RelNode r) {
final ImmutableList.Builder<RelNode> builder = ImmutableList.builder();
return DruidQuery.create(query.getCluster(), r.getTraitSet().replace(query.getConvention()),
query.getTable(), query.druidTable, query.intervals,
builder.addAll(query.rels).add(r).build(), query.getOperatorConversionMap());
}
/** Extends a DruidQuery. */
public static DruidQuery extendQuery(DruidQuery query,
List<Interval> intervals) {
return DruidQuery.create(query.getCluster(), query.getTraitSet(), query.getTable(),
query.druidTable, intervals, query.rels, query.getOperatorConversionMap());
}
/** Check if it is needed to use UTC for DATE and TIMESTAMP types. */
private static boolean needUtcTimeExtract(RexNode rexNode) {
return rexNode.getType().getSqlTypeName() == SqlTypeName.DATE
|| rexNode.getType().getSqlTypeName() == SqlTypeName.TIMESTAMP
|| rexNode.getType().getSqlTypeName()
== SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE;
}
/**
* Converts a {@link RexNode} to a Druid column.
*
* @param rexNode leaf Input Ref to Druid Column
* @param rowType row type
* @param druidQuery Druid query
*
* @return {@link Pair} of Column name and Extraction Function on the top of
* the input ref, or {@code Pair.of(null, null)} when cannot translate to a
* valid Druid column
*/
protected static Pair<String, ExtractionFunction> toDruidColumn(RexNode rexNode,
RelDataType rowType, DruidQuery druidQuery) {
final String columnName;
final ExtractionFunction extractionFunction;
final Granularity granularity;
switch (rexNode.getKind()) {
case INPUT_REF:
columnName = extractColumnName(rexNode, rowType, druidQuery);
if (needUtcTimeExtract(rexNode)) {
extractionFunction =
TimeExtractionFunction.createDefault(DateTimeUtils.UTC_ZONE.getID());
} else {
extractionFunction = null;
}
break;
case EXTRACT:
granularity = DruidDateTimeUtils
.extractGranularity(rexNode, druidQuery.getConnectionConfig().timeZone());
if (granularity == null) {
// unknown Granularity
return Pair.of(null, null);
}
if (!TimeExtractionFunction.isValidTimeExtract(rexNode)) {
return Pair.of(null, null);
}
RexNode extractValueNode = ((RexCall) rexNode).getOperands().get(1);
if (extractValueNode.getType().getSqlTypeName() == SqlTypeName.DATE
|| extractValueNode.getType().getSqlTypeName() == SqlTypeName.TIMESTAMP) {
// Use 'UTC' at the extraction level
extractionFunction =
TimeExtractionFunction.createExtractFromGranularity(
granularity, DateTimeUtils.UTC_ZONE.getID());
columnName = extractColumnName(extractValueNode, rowType, druidQuery);
} else if (extractValueNode.getType().getSqlTypeName()
== SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
// Use local time zone at the extraction level
extractionFunction =
TimeExtractionFunction.createExtractFromGranularity(
granularity, druidQuery.getConnectionConfig().timeZone());
columnName = extractColumnName(extractValueNode, rowType, druidQuery);
} else {
return Pair.of(null, null);
}
break;
case FLOOR:
granularity = DruidDateTimeUtils
.extractGranularity(rexNode, druidQuery.getConnectionConfig().timeZone());
if (granularity == null) {
// unknown Granularity
return Pair.of(null, null);
}
if (!TimeExtractionFunction.isValidTimeFloor(rexNode)) {
return Pair.of(null, null);
}
RexNode floorValueNode = ((RexCall) rexNode).getOperands().get(0);
if (needUtcTimeExtract(floorValueNode)) {
// Use 'UTC' at the extraction level, since all datetime types
// are represented in 'UTC'
extractionFunction =
TimeExtractionFunction.createFloorFromGranularity(
granularity, DateTimeUtils.UTC_ZONE.getID());
columnName = extractColumnName(floorValueNode, rowType, druidQuery);
} else {
return Pair.of(null, null);
}
break;
case CAST:
// CASE we have a cast over InputRef. Check that cast is valid
if (!isValidLeafCast(rexNode)) {
return Pair.of(null, null);
}
RexNode operand0 = ((RexCall) rexNode).getOperands().get(0);
columnName =
extractColumnName(operand0, rowType, druidQuery);
if (needUtcTimeExtract(rexNode)) {
// CASE CAST to TIME/DATE need to make sure that we have valid extraction fn
extractionFunction =
TimeExtractionFunction.translateCastToTimeExtract(rexNode,
TimeZone.getTimeZone(druidQuery.getConnectionConfig().timeZone()));
if (extractionFunction == null) {
// no extraction Function means cast is not valid thus bail out
return Pair.of(null, null);
}
} else {
extractionFunction = null;
}
break;
default:
return Pair.of(null, null);
}
return Pair.of(columnName, extractionFunction);
}
/**
* Returns whether a {@link RexNode} is a valid Druid cast operation.
*
* @param rexNode RexNode
*
* @return whether the operand is an inputRef and it is a valid Druid Cast
* operation
*/
private static boolean isValidLeafCast(RexNode rexNode) {
assert rexNode.isA(SqlKind.CAST);
final RexNode input = ((RexCall) rexNode).getOperands().get(0);
if (!input.isA(SqlKind.INPUT_REF)) {
// it is not a leaf cast don't bother going further.
return false;
}
final SqlTypeName toTypeName = rexNode.getType().getSqlTypeName();
if (toTypeName.getFamily() == SqlTypeFamily.CHARACTER) {
// CAST of input to character type
return true;
}
if (toTypeName.getFamily() == SqlTypeFamily.NUMERIC) {
// CAST of input to numeric type, it is part of a bounded comparison
return true;
}
if (toTypeName.getFamily() == SqlTypeFamily.TIMESTAMP
|| toTypeName.getFamily() == SqlTypeFamily.DATETIME) {
// CAST of literal to timestamp type
return true;
}
if (toTypeName.getFamily().contains(input.getType())) {
// same type it is okay to push it
return true;
}
// Currently other CAST operations cannot be pushed to Druid
return false;
}
/**
* Returns Druid column name or null when it is not possible to translate.
*
* @param rexNode Druid input ref node
* @param rowType Row type
* @param query Druid query
*/
protected static @Nullable String extractColumnName(RexNode rexNode, RelDataType rowType,
DruidQuery query) {
if (rexNode.getKind() == SqlKind.INPUT_REF) {
final RexInputRef ref = (RexInputRef) rexNode;
final String columnName = rowType.getFieldNames().get(ref.getIndex());
if (columnName == null) {
return null;
}
// calcite has this un-direct renaming of timestampFieldName to native druid `__time`
if (query.getDruidTable().timestampFieldName.equals(columnName)) {
return DruidTable.DEFAULT_TIMESTAMP_COLUMN;
}
return columnName;
}
return null;
}
/**
* Equivalent of String.format(Locale.ENGLISH, message, formatArgs).
*/
public static String format(String message, Object... formatArgs) {
return String.format(Locale.ENGLISH, message, formatArgs);
}
/** Returns a string describing the operations inside this query.
*
* <p>For example, "sfpahol" means {@link TableScan} (s)
* followed by {@link Filter} (f)
* followed by {@link Project} (p)
* followed by {@link Aggregate} (a)
* followed by {@link Filter} (h)
* followed by {@link Project} (o)
* followed by {@link Sort} (l).
*
* @see #isValidSignature(String)
*/
String signature() {
final StringBuilder b = new StringBuilder();
boolean flag = false;
for (RelNode rel : rels) {
b.append(rel instanceof TableScan ? 's'
: (rel instanceof Project && flag) ? 'o'
: (rel instanceof Filter && flag) ? 'h'
: rel instanceof Aggregate ? 'a'
: rel instanceof Filter ? 'f'
: rel instanceof Sort ? 'l'
: rel instanceof Project ? 'p'
: '!');
flag = flag || rel instanceof Aggregate;
}
return b.toString();
}
@Override public boolean isValid(Litmus litmus, Context context) {
if (!super.isValid(litmus, context)) {
return false;
}
final String signature = signature();
if (!isValidSignature(signature)) {
return litmus.fail("invalid signature [{}]", signature);
}
if (rels.isEmpty()) {
return litmus.fail("must have at least one rel");
}
for (int i = 0; i < rels.size(); i++) {
final RelNode r = rels.get(i);
if (i == 0) {
if (!(r instanceof TableScan)) {
return litmus.fail("first rel must be TableScan, was ", r);
}
if (r.getTable() != table) {
return litmus.fail("first rel must be based on table table");
}
} else {
final List<RelNode> inputs = r.getInputs();
if (inputs.size() != 1 || inputs.get(0) != rels.get(i - 1)) {
return litmus.fail("each rel must have a single input");
}
if (r instanceof Aggregate) {
final Aggregate aggregate = (Aggregate) r;
if (aggregate.getGroupSets().size() != 1) {
return litmus.fail("no grouping sets");
}
}
if (r instanceof Filter) {
final Filter filter = (Filter) r;
final DruidJsonFilter druidJsonFilter =
DruidJsonFilter.toDruidFilters(filter.getCondition(),
filter.getInput().getRowType(), this,
getCluster().getRexBuilder());
if (druidJsonFilter == null) {
return litmus.fail("invalid filter [{}]", filter.getCondition());
}
}
if (r instanceof Sort) {
final Sort sort = (Sort) r;
if (sort.offset != null && RexLiteral.intValue(sort.offset) != 0) {
return litmus.fail("offset not supported");
}
}
}
}
return true;
}
protected Map<SqlOperator, DruidSqlOperatorConverter> getOperatorConversionMap() {
return converterOperatorMap;
}
@Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
assert inputs.isEmpty();
return this;
}
@Override public RelDataType deriveRowType() {
return getCluster().getTypeFactory().createStructType(
Pair.right(Util.last(rels).getRowType().getFieldList()),
getQuerySpec().fieldNames);
}
public TableScan getTableScan() {
return (TableScan) rels.get(0);
}
public RelNode getTopNode() {
return Util.last(rels);
}
@Override public RelOptTable getTable() {
return table;
}
public DruidTable getDruidTable() {
return druidTable;
}
@Override public RelWriter explainTerms(RelWriter pw) {
for (RelNode rel : rels) {
if (rel instanceof TableScan) {
TableScan tableScan = (TableScan) rel;
pw.item("table", tableScan.getTable().getQualifiedName());
pw.item("intervals", intervals);
} else if (rel instanceof Filter) {
pw.item("filter", ((Filter) rel).getCondition());
} else if (rel instanceof Project) {
if (((Project) rel).getInput() instanceof Aggregate) {
pw.item("post_projects", ((Project) rel).getProjects());
} else {
pw.item("projects", ((Project) rel).getProjects());
}
} else if (rel instanceof Aggregate) {
final Aggregate aggregate = (Aggregate) rel;
pw.item("groups", aggregate.getGroupSet())
.item("aggs", aggregate.getAggCallList());
} else if (rel instanceof Sort) {
final Sort sort = (Sort) rel;
for (Ord<RelFieldCollation> ord
: Ord.zip(sort.collation.getFieldCollations())) {
pw.item("sort" + ord.i, ord.e.getFieldIndex());
}
for (Ord<RelFieldCollation> ord
: Ord.zip(sort.collation.getFieldCollations())) {
if (!pw.expand()) {
pw.item("dir" + ord.i, ord.e.shortString());
} else {
pw.item("dir" + ord.i, ord.e.fullString());
}
}
pw.itemIf("fetch", sort.fetch, sort.fetch != null);
} else {
throw new AssertionError("rel type not supported in Druid query "
+ rel);
}
}
return pw;
}
@Override public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner,
RelMetadataQuery mq) {
final RelOptCost cost =
requireNonNull(Util.last(rels)
.computeSelfCost(planner, mq));
return cost
// Cost increases with the number of fields queried.
// A plan returning 100 or more columns will have 2x the cost of a
// plan returning 2 columns.
// A plan where all extra columns are pruned will be preferred.
.multiplyBy(
RelMdUtil.linear(querySpec.fieldNames.size(), 2, 100, 1d, 2d))
.multiplyBy(getQueryTypeCostMultiplier())
// A Scan leaf filter is better than having filter spec if possible.
.multiplyBy(rels.size() > 1 && rels.get(1) instanceof Filter ? 0.5 : 1.0)
// a plan with sort pushed to druid is better than doing sort outside of druid
.multiplyBy(Util.last(rels) instanceof Sort ? 0.1 : 1.0)
.multiplyBy(getIntervalCostMultiplier());
}
private double getIntervalCostMultiplier() {
long days = 0;
for (Interval interval : intervals) {
days += interval.toDuration().getStandardDays();
}
// Cost increases with the wider interval being queries.
// A plan querying 10 or more years of data will have 10x the cost of a
// plan returning 1 day data.
// A plan where least interval is queries will be preferred.
return RelMdUtil.linear((int) days, 1, DAYS_IN_TEN_YEARS, 0.1d, 1d);
}
private double getQueryTypeCostMultiplier() {
// Cost of Select > GroupBy > Timeseries > TopN
switch (querySpec.queryType) {
case SELECT:
return .1;
case GROUP_BY:
return .08;
case TIMESERIES:
return .06;
case TOP_N:
return .04;
default:
return .2;
}
}
@Override public void register(RelOptPlanner planner) {
for (RelOptRule rule : DruidRules.RULES) {
planner.addRule(rule);
}
for (RelOptRule rule : Bindables.RULES) {
planner.addRule(rule);
}
}
@Override public Class<Object[]> getElementType() {
return Object[].class;
}
@Override public Enumerable<@Nullable Object[]> bind(DataContext dataContext) {
return table.unwrapOrThrow(ScannableTable.class).scan(dataContext);
}
@Override public Node implement(InterpreterImplementor implementor) {
return new DruidQueryNode(implementor.compiler, this);
}
public QuerySpec getQuerySpec() {
if (querySpec == null) {
querySpec =
checkNotNull(deriveQuerySpec(), "null querySpec for %s", this);
}
return querySpec;
}
protected QuerySpec deriveQuerySpec() {
final RelDataType rowType = table.getRowType();
int i = 1;
Filter filterRel = null;
if (i < rels.size() && rels.get(i) instanceof Filter) {
filterRel = (Filter) rels.get(i++);
}
Project project = null;
if (i < rels.size() && rels.get(i) instanceof Project) {
project = (Project) rels.get(i++);
}
ImmutableBitSet groupSet = null;
List<AggregateCall> aggCalls = null;
List<String> aggNames = null;
if (i < rels.size() && rels.get(i) instanceof Aggregate) {
final Aggregate aggregate = (Aggregate) rels.get(i++);
groupSet = aggregate.getGroupSet();
aggCalls = aggregate.getAggCallList();
aggNames =
Util.skip(aggregate.getRowType().getFieldNames(),
groupSet.cardinality());
}
Filter havingFilter = null;
if (i < rels.size() && rels.get(i) instanceof Filter) {
havingFilter = (Filter) rels.get(i++);
}
Project postProject = null;
if (i < rels.size() && rels.get(i) instanceof Project) {
postProject = (Project) rels.get(i++);
}
List<Integer> collationIndexes = null;
List<Direction> collationDirections = null;
ImmutableBitSet.Builder numericCollationBitSetBuilder = ImmutableBitSet.builder();
Integer fetch = null;
if (i < rels.size() && rels.get(i) instanceof Sort) {
final Sort sort = (Sort) rels.get(i++);
collationIndexes = new ArrayList<>();
collationDirections = new ArrayList<>();
for (RelFieldCollation fCol : sort.collation.getFieldCollations()) {
collationIndexes.add(fCol.getFieldIndex());
collationDirections.add(fCol.getDirection());
if (sort.getRowType().getFieldList().get(fCol.getFieldIndex()).getType().getFamily()
== SqlTypeFamily.NUMERIC) {
numericCollationBitSetBuilder.set(fCol.getFieldIndex());
}
}
fetch = sort.fetch != null ? RexLiteral.intValue(sort.fetch) : null;
}
if (i != rels.size()) {
throw new AssertionError("could not implement all rels");
}
return getQuery(rowType, filterRel, project, groupSet, aggCalls, aggNames,
collationIndexes, collationDirections, numericCollationBitSetBuilder.build(), fetch,
postProject, havingFilter);
}
public QueryType getQueryType() {
return getQuerySpec().queryType;
}
public String getQueryString() {
return getQuerySpec().queryString;
}
protected CalciteConnectionConfig getConnectionConfig() {
return getCluster().getPlanner().getContext()
.unwrapOrThrow(CalciteConnectionConfig.class);
}
/**
* Translates Filter rel to Druid Filter Json object if possible.
* Currently Filter rel input has to be Druid Table scan
*
* @param filterRel input filter rel
*
* @return DruidJson Filter or null if cannot translate one of filters
*/
private @Nullable DruidJsonFilter computeFilter(@Nullable Filter filterRel) {
if (filterRel == null) {
return null;
}
final RexNode filter = filterRel.getCondition();
final RelDataType inputRowType = filterRel.getInput().getRowType();
if (filter != null) {
return DruidJsonFilter.toDruidFilters(filter, inputRowType, this,
getCluster().getRexBuilder());
}
return null;
}
/**
* Translates a list of projects to Druid Column names and Virtual Columns if
* any.
*
* <p>We cannot use {@link Pair#zip(Object[], Object[])}, since size may be
* different.
*
* @param projectRel Project
*
* @param druidQuery Druid query
*
* @return Pair of list of Druid Columns and Expression Virtual Columns, or
* null when cannot translate one of the projects
*/
protected static @Nullable Pair<List<String>, List<VirtualColumn>> computeProjectAsScan(
@Nullable Project projectRel, RelDataType inputRowType, DruidQuery druidQuery) {
if (projectRel == null) {
return null;
}
final Set<String> usedFieldNames = new HashSet<>();
final ImmutableList.Builder<VirtualColumn> virtualColumnsBuilder = ImmutableList.builder();
final ImmutableList.Builder<String> projectedColumnsBuilder = ImmutableList.builder();
final List<RexNode> projects = projectRel.getProjects();
for (RexNode project : projects) {
Pair<String, ExtractionFunction> druidColumn =
toDruidColumn(project, inputRowType, druidQuery);
boolean needExtractForOperand = project instanceof RexCall
&& ((RexCall) project).getOperands().stream().anyMatch(DruidQuery::needUtcTimeExtract);
if (druidColumn.left == null || druidColumn.right != null || needExtractForOperand) {
// It is a complex project pushed as expression
final String expression = DruidExpressions
.toDruidExpression(project, inputRowType, druidQuery);
if (expression == null) {
return null;
}
final String virColName =
SqlValidatorUtil.uniquify("vc", usedFieldNames,
SqlValidatorUtil.EXPR_SUGGESTER);
virtualColumnsBuilder.add(VirtualColumn.builder()
.withName(virColName)
.withExpression(expression).withType(
DruidExpressions.EXPRESSION_TYPES.get(project.getType().getSqlTypeName()))
.build());
usedFieldNames.add(virColName);
projectedColumnsBuilder.add(virColName);
} else {
// simple inputRef or extractable function
if (usedFieldNames.contains(druidColumn.left)) {
final String virColName =
SqlValidatorUtil.uniquify("vc", usedFieldNames,
SqlValidatorUtil.EXPR_SUGGESTER);
virtualColumnsBuilder.add(VirtualColumn.builder()
.withName(virColName)
.withExpression(DruidExpressions.fromColumn(druidColumn.left)).withType(
DruidExpressions.EXPRESSION_TYPES.get(project.getType().getSqlTypeName()))
.build());
usedFieldNames.add(virColName);
projectedColumnsBuilder.add(virColName);
} else {
projectedColumnsBuilder.add(druidColumn.left);
usedFieldNames.add(druidColumn.left);
}
}
}
return Pair.of(projectedColumnsBuilder.build(),
virtualColumnsBuilder.build());
}
/**
* Computes the project group set.
*
* @param projectNode Project under the Aggregates if any
* @param groupSet Ids of grouping keys as they are listed in {@code projects} list
* @param inputRowType Input row type under the project
* @param druidQuery Druid query
*
* @return A list of {@link DimensionSpec} containing the group by dimensions,
* and a list of {@link VirtualColumn} containing Druid virtual column
* projections; or null, if translation is not possible.
* Note that the size of lists can be different.
*/
protected static @Nullable Pair<List<DimensionSpec>, List<VirtualColumn>> computeProjectGroupSet(
@Nullable Project projectNode, ImmutableBitSet groupSet,
RelDataType inputRowType, DruidQuery druidQuery) {
final List<DimensionSpec> dimensionSpecList = new ArrayList<>();
final List<VirtualColumn> virtualColumnList = new ArrayList<>();
final Set<String> usedFieldNames = new HashSet<>();
for (int groupKey : groupSet) {
final DimensionSpec dimensionSpec;
final RexNode project;
if (projectNode == null) {
project = RexInputRef.of(groupKey, inputRowType);
} else {
project = projectNode.getProjects().get(groupKey);
}
Pair<String, ExtractionFunction> druidColumn =
toDruidColumn(project, inputRowType, druidQuery);
if (druidColumn.left != null && druidColumn.right == null) {
// SIMPLE INPUT REF
dimensionSpec =
new DefaultDimensionSpec(druidColumn.left, druidColumn.left,
DruidExpressions.EXPRESSION_TYPES.get(project.getType().getSqlTypeName()));
usedFieldNames.add(druidColumn.left);
} else if (druidColumn.left != null && druidColumn.right != null) {
// CASE it is an extraction Dimension
final String columnPrefix;
// TODO Remove it! if else statement is not really needed it
// is here to make tests pass.
if (project.getKind() == SqlKind.EXTRACT) {
columnPrefix =
EXTRACT_COLUMN_NAME_PREFIX + "_" + requireNonNull(DruidDateTimeUtils
.extractGranularity(project, druidQuery.getConnectionConfig().timeZone())
.getType().lowerName);
} else if (project.getKind() == SqlKind.FLOOR) {
columnPrefix =
FLOOR_COLUMN_NAME_PREFIX + "_" + requireNonNull(DruidDateTimeUtils
.extractGranularity(project, druidQuery.getConnectionConfig().timeZone())
.getType().lowerName);
} else {
columnPrefix = "extract";
}
final String uniqueExtractColumnName = SqlValidatorUtil
.uniquify(columnPrefix, usedFieldNames,
SqlValidatorUtil.EXPR_SUGGESTER);
dimensionSpec =
new ExtractionDimensionSpec(druidColumn.left, druidColumn.right,
uniqueExtractColumnName);
usedFieldNames.add(uniqueExtractColumnName);
} else {
// CASE it is Expression
final String expression = DruidExpressions
.toDruidExpression(project, inputRowType, druidQuery);
if (Strings.isNullOrEmpty(expression)) {
return null;
}
final String name = SqlValidatorUtil
.uniquify("vc", usedFieldNames,
SqlValidatorUtil.EXPR_SUGGESTER);
VirtualColumn vc =
new VirtualColumn(name, expression,
DruidExpressions.EXPRESSION_TYPES.get(project.getType().getSqlTypeName()));
virtualColumnList.add(vc);
dimensionSpec =
new DefaultDimensionSpec(name, name,
DruidExpressions.EXPRESSION_TYPES.get(project.getType().getSqlTypeName()));
usedFieldNames.add(name);
}
dimensionSpecList.add(dimensionSpec);
}
return Pair.of(dimensionSpecList, virtualColumnList);
}
/**
* Translates aggregate calls to Druid {@link JsonAggregation}s when
* possible.
*
* @param aggCalls List of AggregateCalls to translate
* @param aggNames List of aggregate names
* @param project Input project under the aggregate calls,
* or null if we have {@link TableScan} immediately under the
* {@link Aggregate}
* @param druidQuery Druid query
*
* @return List of valid Druid {@link JsonAggregation}s, or null if any of the
* aggregates is not supported
*/
protected static @Nullable List<JsonAggregation> computeDruidJsonAgg(List<AggregateCall> aggCalls,
List<String> aggNames, @Nullable Project project, DruidQuery druidQuery) {
final List<JsonAggregation> aggregations = new ArrayList<>();
for (Pair<AggregateCall, String> agg : Pair.zip(aggCalls, aggNames)) {
final String fieldName;
final @Nullable String expression;
final AggregateCall aggCall = agg.left;
final @Nullable RexNode filterNode;
// Type check First
final RelDataType type = aggCall.getType();
final SqlTypeName sqlTypeName = type.getSqlTypeName();
final boolean isNotAcceptedType;
if (SqlTypeFamily.APPROXIMATE_NUMERIC.getTypeNames().contains(sqlTypeName)
|| SqlTypeFamily.INTEGER.getTypeNames().contains(sqlTypeName)) {
isNotAcceptedType = false;
} else if (SqlTypeFamily.EXACT_NUMERIC.getTypeNames().contains(sqlTypeName)
&& (type.getScale() == 0
|| druidQuery.getConnectionConfig().approximateDecimal())) {
// Decimal, If scale is zero or we allow approximating decimal, we can proceed
isNotAcceptedType = false;
} else {
isNotAcceptedType = true;
}
if (isNotAcceptedType) {
return null;
}
// Extract filters
if (project != null && aggCall.hasFilter()) {
filterNode = project.getProjects().get(aggCall.filterArg);
} else {
filterNode = null;
}
if (aggCall.getArgList().isEmpty()) {
fieldName = null;
expression = null;
} else {
int index = Iterables.getOnlyElement(aggCall.getArgList());
if (project == null) {
fieldName = druidQuery.table.getRowType().getFieldNames().get(index);
expression = null;
} else {
final RexNode rexNode = project.getProjects().get(index);
final RelDataType inputRowType = project.getInput().getRowType();
if (rexNode.isA(SqlKind.INPUT_REF)) {
expression = null;
fieldName =
extractColumnName(rexNode, inputRowType, druidQuery);
} else {
expression = DruidExpressions
.toDruidExpression(rexNode, inputRowType, druidQuery);
if (Strings.isNullOrEmpty(expression)) {
return null;
}
fieldName = null;
}
}
// One should be not null and the other should be null.
assert expression == null ^ fieldName == null;
}
final JsonAggregation jsonAggregation =
getJsonAggregation(agg.right, agg.left, filterNode, fieldName,
expression, druidQuery);
if (jsonAggregation == null) {
return null;
}
aggregations.add(jsonAggregation);
}
return aggregations;
}
protected QuerySpec getQuery(RelDataType rowType, Filter filter,
@Nullable Project project, @Nullable ImmutableBitSet groupSet,
@Nullable List<AggregateCall> aggCalls, @Nullable List<String> aggNames,
@Nullable List<Integer> collationIndexes,
@Nullable List<Direction> collationDirections,
ImmutableBitSet numericCollationIndexes, @Nullable Integer fetch,
@Nullable Project postProject, @Nullable Filter havingFilter) {
// Handle filter
final @Nullable DruidJsonFilter jsonFilter = computeFilter(filter);
if (groupSet == null) {
// It is Scan Query since no Grouping
assert aggCalls == null;
assert aggNames == null;
assert collationIndexes == null || collationIndexes.isEmpty();
assert collationDirections == null || collationDirections.isEmpty();
final List<String> scanColumnNames;
final List<VirtualColumn> virtualColumnList = new ArrayList<>();
if (project != null) {
// project some fields only
Pair<List<String>, List<VirtualColumn>> projectResult =
computeProjectAsScan(project, project.getInput().getRowType(), this);
scanColumnNames = projectResult.left;
virtualColumnList.addAll(projectResult.right);
} else {
// Scan all the fields
scanColumnNames = rowType.getFieldNames();
}
final ScanQuery scanQuery =
new ScanQuery(druidTable.dataSource, intervals, jsonFilter,
virtualColumnList, scanColumnNames, fetch);
return new QuerySpec(QueryType.SCAN, scanQuery.toQuery(), scanColumnNames);
}
// At this Stage we have a valid Aggregate thus Query is one of Timeseries, TopN, or GroupBy
// Handling aggregate and sort is more complex, since
// we need to extract the conditions to know whether the query will be executed as a
// Timeseries, TopN, or GroupBy in Druid
requireNonNull(aggCalls, "aggCalls");
requireNonNull(aggNames, "aggNames");
checkArgument(aggCalls.size() == aggNames.size());
final List<JsonExpressionPostAgg> postAggs = new ArrayList<>();
final JsonLimit limit;
final RelDataType aggInputRowType = table.getRowType();
final List<String> aggregateStageFieldNames = new ArrayList<>();
Pair<List<DimensionSpec>, List<VirtualColumn>> projectGroupSet =
computeProjectGroupSet(project, groupSet, aggInputRowType, this);
requireNonNull(projectGroupSet, "projectGroupSet");
final List<DimensionSpec> groupByKeyDims = projectGroupSet.left;
final List<VirtualColumn> virtualColumnList = projectGroupSet.right;
for (DimensionSpec dim : groupByKeyDims) {
aggregateStageFieldNames.add(dim.getOutputName());
}
final List<JsonAggregation> aggregations =
computeDruidJsonAgg(aggCalls, aggNames, project, this);
requireNonNull(aggregations, "aggregations");
for (JsonAggregation jsonAgg : aggregations) {
aggregateStageFieldNames.add(jsonAgg.name);
}
final @Nullable DruidJsonFilter havingJsonFilter;
if (havingFilter != null) {
havingJsonFilter =
DruidJsonFilter.toDruidFilters(havingFilter.getCondition(),
havingFilter.getInput().getRowType(), this,
getCluster().getRexBuilder());
} else {
havingJsonFilter = null;
}
// Then we handle projects after aggregates as Druid Post Aggregates
final List<String> postAggregateStageFieldNames;
if (postProject != null) {
final List<String> postProjectDimListBuilder = new ArrayList<>();
final RelDataType postAggInputRowType = getCluster().getTypeFactory()
.createStructType(Pair.right(postProject.getInput().getRowType().getFieldList()),
aggregateStageFieldNames);
final Set<String> existingAggFieldsNames = new HashSet<>(aggregateStageFieldNames);
// this is an index of existing columns coming out aggregate layer. Will use this index to:
// filter out any project down the road that doesn't change values e.g inputRef/identity cast
Map<String, String> existingProjects = Maps
.uniqueIndex(aggregateStageFieldNames, DruidExpressions::fromColumn);
for (Pair<RexNode, String> pair : postProject.getNamedProjects()) {
final RexNode postProjectRexNode = pair.left;
String expression = DruidExpressions
.toDruidExpression(postProjectRexNode, postAggInputRowType, this);
final String existingFieldName = existingProjects.get(expression);
if (existingFieldName != null) {
// simple input ref or Druid runtime identity cast will skip it, since it is here already
postProjectDimListBuilder.add(existingFieldName);
} else {
final String uniquelyProjectFieldName =
SqlValidatorUtil.uniquify(pair.right,
existingAggFieldsNames, SqlValidatorUtil.EXPR_SUGGESTER);
postAggs.add(new JsonExpressionPostAgg(uniquelyProjectFieldName, expression, null));
postProjectDimListBuilder.add(uniquelyProjectFieldName);
existingAggFieldsNames.add(uniquelyProjectFieldName);
}
}
postAggregateStageFieldNames = postProjectDimListBuilder;
} else {
postAggregateStageFieldNames = null;
}
// final Query output row field names.
final List<String> queryOutputFieldNames = postAggregateStageFieldNames == null
? aggregateStageFieldNames
: postAggregateStageFieldNames;
// handle sort all together
limit =
computeSort(fetch, collationIndexes, collationDirections,
numericCollationIndexes, queryOutputFieldNames);
final String timeSeriesQueryString =
planAsTimeSeries(groupByKeyDims, jsonFilter,
virtualColumnList, aggregations, postAggs, limit, havingJsonFilter);
if (timeSeriesQueryString != null) {
final String timeExtractColumn = groupByKeyDims.isEmpty()
? null
: groupByKeyDims.get(0).getOutputName();
if (timeExtractColumn != null) {
// Case we have transformed the group by time to druid timeseries with Granularity.
// Need to replace the name of the column with druid timestamp field name.
final List<String> timeseriesFieldNames =
Util.transform(queryOutputFieldNames, input -> {
if (timeExtractColumn.equals(input)) {
return "timestamp";
}
return input;
});
return new QuerySpec(QueryType.TIMESERIES, timeSeriesQueryString, timeseriesFieldNames);
}
return new QuerySpec(QueryType.TIMESERIES, timeSeriesQueryString, queryOutputFieldNames);
}
final String topNQuery =
planAsTopN(groupByKeyDims, jsonFilter,
virtualColumnList, aggregations, postAggs, limit, havingJsonFilter);
if (topNQuery != null) {
return new QuerySpec(QueryType.TOP_N, topNQuery, queryOutputFieldNames);
}
final String groupByQuery =
planAsGroupBy(groupByKeyDims, jsonFilter,
virtualColumnList, aggregations, postAggs, limit, havingJsonFilter);
if (groupByQuery == null) {
throw new IllegalStateException("Cannot plan Druid Query");
}
return new QuerySpec(QueryType.GROUP_BY, groupByQuery, queryOutputFieldNames);
}
/**
* Converts a sort specification to a {@link JsonLimit} (never null).
*
* @param fetch limit to fetch
* @param collationIndexes index of fields as listed in query row output
* @param collationDirections direction of sort
* @param numericCollationIndexes flag of to determine sort comparator
* @param queryOutputFieldNames query output fields
*/
private static JsonLimit computeSort(@Nullable Integer fetch,
@Nullable List<Integer> collationIndexes,
@Nullable List<Direction> collationDirections,
ImmutableBitSet numericCollationIndexes,
List<String> queryOutputFieldNames) {
final List<JsonCollation> collations;
if (collationIndexes != null) {
requireNonNull(collationDirections, "collationDirections");
ImmutableList.Builder<JsonCollation> colBuilder = ImmutableList.builder();
for (Pair<Integer, Direction> p : Pair.zip(collationIndexes, collationDirections)) {
final String dimensionOrder = numericCollationIndexes.get(p.left)
? "numeric"
: "lexicographic";
colBuilder.add(
new JsonCollation(queryOutputFieldNames.get(p.left),
p.right == Direction.DESCENDING ? "descending" : "ascending", dimensionOrder));
}
collations = colBuilder.build();
} else {
collations = null;
}
return new JsonLimit("default", fetch, collations);
}
private @Nullable String planAsTimeSeries(List<DimensionSpec> groupByKeyDims,
@Nullable DruidJsonFilter jsonFilter,
List<VirtualColumn> virtualColumnList, List<JsonAggregation> aggregations,
List<JsonExpressionPostAgg> postAggregations, JsonLimit limit,
@Nullable DruidJsonFilter havingFilter) {
if (havingFilter != null) {
return null;
}
if (groupByKeyDims.size() > 1) {
return null;
}
if (limit.limit != null) {
// it has a limit not supported by time series
return null;
}
if (limit.collations != null && limit.collations.size() > 1) {
// it has multiple sort columns
return null;
}
final String sortDirection;
if (limit.collations != null && limit.collations.size() == 1) {
if (groupByKeyDims.isEmpty()
|| !limit.collations.get(0).dimension.equals(groupByKeyDims.get(0).getOutputName())) {
// sort column is not time column
return null;
}
sortDirection = limit.collations.get(0).direction;
} else {
sortDirection = null;
}
final Granularity timeseriesGranularity;
if (groupByKeyDims.size() == 1) {
DimensionSpec dimensionSpec = Iterables.getOnlyElement(groupByKeyDims);
Granularity granularity = ExtractionDimensionSpec.toQueryGranularity(dimensionSpec);
// case we have project expression on the top of the time extract then
// cannot use timeseries
boolean hasExpressionOnTopOfTimeExtract = false;
for (JsonExpressionPostAgg postAgg : postAggregations) {
if (postAgg != null) {
if (postAgg.expression.contains(groupByKeyDims.get(0).getOutputName())) {
hasExpressionOnTopOfTimeExtract = true;
}
}
}
timeseriesGranularity = hasExpressionOnTopOfTimeExtract ? null : granularity;
if (timeseriesGranularity == null) {
// cannot extract granularity bailout
return null;
}
} else {
timeseriesGranularity = Granularities.all();
}
final boolean skipEmptyBuckets = Granularities.all() != timeseriesGranularity;
final StringWriter sw = new StringWriter();
final JsonFactory factory = new JsonFactory();
try {
final JsonGenerator generator = factory.createGenerator(sw);
generator.writeStartObject();
generator.writeStringField("queryType", "timeseries");
generator.writeStringField("dataSource", druidTable.dataSource);
generator.writeBooleanField("descending", sortDirection != null
&& sortDirection.equals("descending"));
writeField(generator, "granularity", timeseriesGranularity);
writeFieldIf(generator, "filter", jsonFilter);
writeField(generator, "aggregations", aggregations);
writeFieldIf(generator, "virtualColumns",
virtualColumnList.isEmpty() ? null : virtualColumnList);
writeFieldIf(generator, "postAggregations",
postAggregations.isEmpty() ? null : postAggregations);
writeField(generator, "intervals", intervals);
generator.writeFieldName("context");
// The following field is necessary to conform with SQL semantics (CALCITE-1589)
generator.writeStartObject();
// Count(*) returns 0 if result set is empty thus need to set skipEmptyBuckets to false
generator.writeBooleanField("skipEmptyBuckets", skipEmptyBuckets);
generator.writeEndObject();
generator.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
return sw.toString();
}
private @Nullable String planAsTopN(List<DimensionSpec> groupByKeyDims,
DruidJsonFilter jsonFilter,
List<VirtualColumn> virtualColumnList, List<JsonAggregation> aggregations,
List<JsonExpressionPostAgg> postAggregations, JsonLimit limit, DruidJsonFilter havingFilter) {
if (havingFilter != null) {
return null;
}
if (!getConnectionConfig().approximateTopN() || groupByKeyDims.size() != 1
|| limit.limit == null || limit.collations == null || limit.collations.size() != 1) {
return null;
}
if (limit.collations.get(0).dimension.equals(groupByKeyDims.get(0).getOutputName())) {
return null;
}
if (limit.collations.get(0).direction.equals("ascending")) {
// Only DESC is allowed
return null;
}
final String topNMetricColumnName = limit.collations.get(0).dimension;
final StringWriter sw = new StringWriter();
final JsonFactory factory = new JsonFactory();
try {
final JsonGenerator generator = factory.createGenerator(sw);
generator.writeStartObject();
generator.writeStringField("queryType", "topN");
generator.writeStringField("dataSource", druidTable.dataSource);
writeField(generator, "granularity", Granularities.all());
writeField(generator, "dimension", groupByKeyDims.get(0));
writeFieldIf(generator, "virtualColumns",
virtualColumnList.isEmpty() ? null : virtualColumnList);
generator.writeStringField("metric", topNMetricColumnName);
writeFieldIf(generator, "filter", jsonFilter);
writeField(generator, "aggregations", aggregations);
writeFieldIf(generator, "postAggregations",
postAggregations.isEmpty() ? null : postAggregations);
writeField(generator, "intervals", intervals);
generator.writeNumberField("threshold", limit.limit);
generator.writeEndObject();
generator.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
return sw.toString();
}
private @Nullable String planAsGroupBy(List<DimensionSpec> groupByKeyDims,
DruidJsonFilter jsonFilter,
List<VirtualColumn> virtualColumnList, List<JsonAggregation> aggregations,
List<JsonExpressionPostAgg> postAggregations, JsonLimit limit, DruidJsonFilter havingFilter) {
final StringWriter sw = new StringWriter();
final JsonFactory factory = new JsonFactory();
try {
final JsonGenerator generator = factory.createGenerator(sw);
generator.writeStartObject();
generator.writeStringField("queryType", "groupBy");
generator.writeStringField("dataSource", druidTable.dataSource);
writeField(generator, "granularity", Granularities.all());
writeField(generator, "dimensions", groupByKeyDims);
writeFieldIf(generator, "virtualColumns",
virtualColumnList.isEmpty() ? null : virtualColumnList);
writeFieldIf(generator, "limitSpec", limit);
writeFieldIf(generator, "filter", jsonFilter);
writeField(generator, "aggregations", aggregations);
writeFieldIf(generator, "postAggregations",
postAggregations.isEmpty() ? null : postAggregations);
writeField(generator, "intervals", intervals);
writeFieldIf(generator, "having",
havingFilter == null ? null : new DruidJsonFilter.JsonDimHavingFilter(havingFilter));
generator.writeEndObject();
generator.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
return sw.toString();
}
/** Druid Scan Query body. */
private static class ScanQuery {
private final String dataSource;
private final List<Interval> intervals;
private final @Nullable DruidJsonFilter jsonFilter;
private final List<VirtualColumn> virtualColumnList;
private final List<String> columns;
private final @Nullable Integer fetchLimit;
ScanQuery(String dataSource, List<Interval> intervals,
@Nullable DruidJsonFilter jsonFilter,
List<VirtualColumn> virtualColumnList,
List<String> columns,
@Nullable Integer fetchLimit) {
this.dataSource = dataSource;
this.intervals = intervals;
this.jsonFilter = jsonFilter;
this.virtualColumnList = virtualColumnList;
this.columns = columns;
this.fetchLimit = fetchLimit;
}
public String toQuery() {
final StringWriter sw = new StringWriter();
try {
final JsonFactory factory = new JsonFactory();
final JsonGenerator generator = factory.createGenerator(sw);
generator.writeStartObject();
generator.writeStringField("queryType", "scan");
generator.writeStringField("dataSource", dataSource);
writeField(generator, "intervals", intervals);
writeFieldIf(generator, "filter", jsonFilter);
writeFieldIf(generator, "virtualColumns",
virtualColumnList.isEmpty() ? null : virtualColumnList);
writeField(generator, "columns", columns);
generator.writeStringField("resultFormat", "compactedList");
if (fetchLimit != null) {
generator.writeNumberField("limit", fetchLimit);
}
generator.writeEndObject();
generator.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
return sw.toString();
}
}
private static @Nullable JsonAggregation getJsonAggregation(
String name, AggregateCall aggCall, @Nullable RexNode filterNode,
@Nullable String fieldName, @Nullable String aggExpression,
DruidQuery druidQuery) {
final boolean fractional;
final RelDataType type = aggCall.getType();
final SqlTypeName sqlTypeName = type.getSqlTypeName();
final JsonAggregation aggregation;
final CalciteConnectionConfig config = druidQuery.getConnectionConfig();
if (SqlTypeFamily.APPROXIMATE_NUMERIC.getTypeNames().contains(sqlTypeName)) {
fractional = true;
} else if (SqlTypeFamily.INTEGER.getTypeNames().contains(sqlTypeName)) {
fractional = false;
} else if (SqlTypeFamily.EXACT_NUMERIC.getTypeNames().contains(sqlTypeName)) {
// Decimal
assert sqlTypeName == SqlTypeName.DECIMAL;
if (type.getScale() == 0) {
fractional = false;
} else {
fractional = true;
}
} else {
// Cannot handle this aggregate function type
return null;
}
// Convert from a complex metric
ComplexMetric complexMetric = druidQuery.druidTable.resolveComplexMetric(fieldName, aggCall);
switch (aggCall.getAggregation().getKind()) {
case COUNT:
if (aggCall.isDistinct()) {
if (aggCall.isApproximate() || config.approximateDistinctCount()) {
if (complexMetric == null) {
aggregation =
new JsonCardinalityAggregation("cardinality", name,
ImmutableList.of(fieldName));
} else {
aggregation =
new JsonAggregation(complexMetric.getMetricType(), name,
complexMetric.getMetricName(), null);
}
break;
} else {
// when approximate results were not told be acceptable.
return null;
}
}
if (aggCall.getArgList().size() == 1 && !aggCall.isDistinct()) {
// case we have count(column) push it as count(*) where column is not null
final DruidJsonFilter matchNulls;
if (fieldName == null) {
matchNulls = new DruidJsonFilter.JsonExpressionFilter(aggExpression + " == null");
} else {
matchNulls = DruidJsonFilter.getSelectorFilter(fieldName, null, null);
}
aggregation =
new JsonFilteredAggregation(DruidJsonFilter.toNotDruidFilter(matchNulls),
new JsonAggregation("count", name, fieldName, aggExpression));
} else if (!aggCall.isDistinct()) {
aggregation = new JsonAggregation("count", name, fieldName, aggExpression);
} else {
aggregation = null;
}
break;
case SUM:
case SUM0:
aggregation =
new JsonAggregation(fractional ? "doubleSum" : "longSum",
name, fieldName, aggExpression);
break;
case MIN:
aggregation =
new JsonAggregation(fractional ? "doubleMin" : "longMin",
name, fieldName, aggExpression);
break;
case MAX:
aggregation =
new JsonAggregation(fractional ? "doubleMax" : "longMax",
name, fieldName, aggExpression);
break;
default:
return null;
}
if (aggregation == null) {
return null;
}
// translate filters
if (filterNode != null) {
DruidJsonFilter druidFilter =
DruidJsonFilter.toDruidFilters(filterNode,
druidQuery.table.getRowType(), druidQuery,
druidQuery.getCluster().getRexBuilder());
if (druidFilter == null) {
// cannot translate filter
return null;
}
return new JsonFilteredAggregation(druidFilter, aggregation);
}
return aggregation;
}
protected static void writeField(JsonGenerator generator, String fieldName,
Object o) throws IOException {
generator.writeFieldName(fieldName);
writeObject(generator, o);
}
protected static void writeFieldIf(JsonGenerator generator, String fieldName,
@Nullable Object o) throws IOException {
if (o != null) {
writeField(generator, fieldName, o);
}
}
protected static void writeArray(JsonGenerator generator, List<?> elements)
throws IOException {
generator.writeStartArray();
for (Object o : elements) {
writeObject(generator, o);
}
generator.writeEndArray();
}
protected static void writeObject(JsonGenerator generator, Object o)
throws IOException {
if (o instanceof String) {
String s = (String) o;
generator.writeString(s);
} else if (o instanceof Interval) {
generator.writeString(o.toString());
} else if (o instanceof Integer) {
Integer i = (Integer) o;
generator.writeNumber(i);
} else if (o instanceof List) {
writeArray(generator, (List<?>) o);
} else if (o instanceof DruidJson) {
((DruidJson) o).write(generator);
} else {
throw new AssertionError("not a json object: " + o);
}
}
/** Generates a JSON string to query metadata about a data source. */
static String metadataQuery(String dataSourceName,
List<Interval> intervals) {
final StringWriter sw = new StringWriter();
final JsonFactory factory = new JsonFactory();
try {
final JsonGenerator generator = factory.createGenerator(sw);
generator.writeStartObject();
generator.writeStringField("queryType", "segmentMetadata");
generator.writeStringField("dataSource", dataSourceName);
generator.writeBooleanField("merge", true);
generator.writeBooleanField("lenientAggregatorMerge", true);
generator.writeArrayFieldStart("analysisTypes");
generator.writeString("aggregators");
generator.writeEndArray();
writeFieldIf(generator, "intervals", intervals);
generator.writeEndObject();
generator.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
return sw.toString();
}
/** Druid query specification. */
public static class QuerySpec {
final QueryType queryType;
final String queryString;
final List<String> fieldNames;
QuerySpec(QueryType queryType, String queryString,
List<String> fieldNames) {
this.queryType = requireNonNull(queryType, "queryType");
this.queryString = requireNonNull(queryString, "queryString");
this.fieldNames = ImmutableList.copyOf(fieldNames);
}
@Override public int hashCode() {
return Objects.hash(queryType, queryString, fieldNames);
}
@Override public boolean equals(Object obj) {
return obj == this
|| obj instanceof QuerySpec
&& queryType == ((QuerySpec) obj).queryType
&& queryString.equals(((QuerySpec) obj).queryString)
&& fieldNames.equals(((QuerySpec) obj).fieldNames);
}
@Override public String toString() {
return "{queryType: " + queryType
+ ", queryString: " + queryString
+ ", fieldNames: " + fieldNames + "}";
}
public String getQueryString(@Nullable String pagingIdentifier,
int offset) {
if (pagingIdentifier == null) {
return queryString;
}
return queryString.replace("\"threshold\":",
"\"pagingIdentifiers\":{\"" + pagingIdentifier + "\":" + offset
+ "},\"threshold\":");
}
}
/** Interpreter node that executes a Druid query and sends the results to a
* {@link Sink}. */
private static class DruidQueryNode implements Node {
private final Sink sink;
private final DruidQuery query;
private final QuerySpec querySpec;
DruidQueryNode(Compiler interpreter, DruidQuery query) {
this.query = query;
this.sink = interpreter.sink(query);
this.querySpec = query.getQuerySpec();
Hook.QUERY_PLAN.run(querySpec);
}
@Override public void run() {
final List<ColumnMetaData.Rep> fieldTypes = new ArrayList<>();
for (RelDataTypeField field : query.getRowType().getFieldList()) {
fieldTypes.add(getPrimitive(field));
}
final DruidConnectionImpl connection =
new DruidConnectionImpl(query.druidTable.schema.url,
query.druidTable.schema.coordinatorUrl);
final boolean limitQuery = containsLimit(querySpec);
final DruidConnectionImpl.Page page = new DruidConnectionImpl.Page();
do {
final String queryString =
querySpec.getQueryString(page.pagingIdentifier, page.offset);
connection.request(querySpec.queryType, queryString, sink,
querySpec.fieldNames, fieldTypes, page);
} while (!limitQuery
&& page.pagingIdentifier != null
&& page.totalRowCount > 0);
}
private static boolean containsLimit(QuerySpec querySpec) {
return querySpec.queryString.contains("\"context\":{\""
+ DRUID_QUERY_FETCH + "\":true");
}
private static ColumnMetaData.Rep getPrimitive(RelDataTypeField field) {
switch (field.getType().getSqlTypeName()) {
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
case TIMESTAMP:
return ColumnMetaData.Rep.JAVA_SQL_TIMESTAMP;
case BIGINT:
return ColumnMetaData.Rep.LONG;
case INTEGER:
return ColumnMetaData.Rep.INTEGER;
case SMALLINT:
return ColumnMetaData.Rep.SHORT;
case TINYINT:
return ColumnMetaData.Rep.BYTE;
case REAL:
return ColumnMetaData.Rep.FLOAT;
case DOUBLE:
case FLOAT:
return ColumnMetaData.Rep.DOUBLE;
default:
return null;
}
}
}
/** Aggregation element of a Druid "groupBy" or "topN" query. */
private static class JsonAggregation implements DruidJson {
final String type;
final String name;
final @Nullable String fieldName;
final @Nullable String expression;
private JsonAggregation(String type, String name, @Nullable String fieldName,
@Nullable String expression) {
this.type = type;
this.name = name;
this.fieldName = fieldName;
this.expression = expression;
}
@Override public void write(JsonGenerator generator) throws IOException {
generator.writeStartObject();
generator.writeStringField("type", type);
generator.writeStringField("name", name);
writeFieldIf(generator, "fieldName", fieldName);
writeFieldIf(generator, "expression", expression);
generator.writeEndObject();
}
}
/**
* Druid Json Expression post aggregate.
*/
private static class JsonExpressionPostAgg extends JsonPostAggregation {
private final String expression;
private final @Nullable String ordering;
private JsonExpressionPostAgg(String name, String expression,
@Nullable String ordering) {
super(name, "expression");
this.expression = expression;
this.ordering = ordering;
}
@Override public void write(JsonGenerator generator) throws IOException {
super.write(generator);
writeFieldIf(generator, "expression", expression);
writeFieldIf(generator, "ordering", ordering);
generator.writeEndObject();
}
}
/** Collation element of a Druid "groupBy" query. */
private static class JsonLimit implements DruidJson {
final String type;
final @Nullable Integer limit;
final @Nullable List<JsonCollation> collations;
private JsonLimit(String type, @Nullable Integer limit,
@Nullable List<JsonCollation> collations) {
this.type = type;
this.limit = limit;
this.collations = collations;
}
@Override public void write(JsonGenerator generator) throws IOException {
generator.writeStartObject();
generator.writeStringField("type", type);
writeFieldIf(generator, "limit", limit);
writeFieldIf(generator, "columns", collations);
generator.writeEndObject();
}
}
/** Collation element of a Druid "groupBy" query. */
private static class JsonCollation implements DruidJson {
final String dimension;
final String direction;
final String dimensionOrder;
private JsonCollation(String dimension, String direction, String dimensionOrder) {
this.dimension = dimension;
this.direction = direction;
this.dimensionOrder = dimensionOrder;
}
@Override public void write(JsonGenerator generator) throws IOException {
generator.writeStartObject();
generator.writeStringField("dimension", dimension);
writeFieldIf(generator, "direction", direction);
writeFieldIf(generator, "dimensionOrder", dimensionOrder);
generator.writeEndObject();
}
}
/** Aggregation element that calls the "cardinality" function. */
private static class JsonCardinalityAggregation extends JsonAggregation {
final List<String> fieldNames;
private JsonCardinalityAggregation(String type, String name,
List<String> fieldNames) {
super(type, name, null, null);
this.fieldNames = fieldNames;
}
@Override public void write(JsonGenerator generator) throws IOException {
generator.writeStartObject();
generator.writeStringField("type", type);
generator.writeStringField("name", name);
writeFieldIf(generator, "fieldNames", fieldNames);
generator.writeEndObject();
}
}
/** Aggregation element that contains a filter. */
private static class JsonFilteredAggregation extends JsonAggregation {
final DruidJsonFilter filter;
final JsonAggregation aggregation;
private JsonFilteredAggregation(DruidJsonFilter filter, JsonAggregation aggregation) {
// Filtered aggregations don't use the "name" and "fieldName" fields directly,
// but rather use the ones defined in their "aggregation" field.
super("filtered", aggregation.name, aggregation.fieldName, null);
this.filter = filter;
this.aggregation = aggregation;
}
@Override public void write(JsonGenerator generator) throws IOException {
generator.writeStartObject();
generator.writeStringField("type", type);
writeField(generator, "filter", filter);
writeField(generator, "aggregator", aggregation);
generator.writeEndObject();
}
}
/** Post-aggregator abstract writer. */
protected abstract static class JsonPostAggregation implements DruidJson {
final String type;
String name;
private JsonPostAggregation(String name, String type) {
this.type = type;
this.name = name;
}
// Expects all subclasses to write the EndObject item
@Override public void write(JsonGenerator generator) throws IOException {
generator.writeStartObject();
generator.writeStringField("type", type);
generator.writeStringField("name", name);
}
public void setName(String name) {
this.name = name;
}
}
/** Returns the index of the timestamp ref, or -1 if not present. */
protected int getTimestampFieldIndex() {
return Iterables.indexOf(this.getRowType().getFieldList(),
input -> druidTable.timestampFieldName.equals(input.getName()));
}
}