This is an automated email from the ASF dual-hosted git repository.
yqm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 9f15763694f Handle 1. empty data in CountAggregator; 2. query's
required physical column is not in grouping, filtering, ordering, or
aggregating (#18307)
9f15763694f is described below
commit 9f15763694fb0ccede2a6f5d84975dac2103d9d1
Author: Cece Mei <[email protected]>
AuthorDate: Wed Jul 23 20:36:58 2025 -0700
Handle 1. empty data in CountAggregator; 2. query's required physical
column is not in grouping, filtering, ordering, or aggregating (#18307)
* add test for null values
* handle default value in NullableNumericAggregatorFactory
* revert style change
* revert style change, and add no data test in group by query
* test change
* use forceNullable rather than defaultValue
* minor change
* some change in test
* AggregateCombiner should care about isNullable
* isNullable should not be user facing in LongSumAggregatorFactory
* unit test failure
* review comments
* fix the unnest test failure
* style fix
* style fix
* javadoc
---
.../query/aggregation/CountAggregatorFactory.java | 7 +-
.../aggregation/LongSumAggregatorFactory.java | 34 ++++++-
.../NullableNumericAggregatorFactory.java | 31 ++++++-
.../druid/segment/AggregateProjectionMetadata.java | 64 ++++++++++---
.../druid/segment/projections/Projections.java | 20 ++++
.../query/groupby/GroupByQueryRunnerTest.java | 72 ++++++++++++---
.../groupby/GroupByTimeseriesQueryRunnerTest.java | 21 +++++
.../query/metadata/SegmentMetadataQueryTest.java | 18 ++--
.../timeseries/TimeseriesQueryRunnerTest.java | 102 +++++++++++++++++++++
.../java/org/apache/druid/segment/TestIndex.java | 23 +++--
10 files changed, 340 insertions(+), 52 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/CountAggregatorFactory.java
b/processing/src/main/java/org/apache/druid/query/aggregation/CountAggregatorFactory.java
index 7089789f90a..e0fa4c643a4 100644
---
a/processing/src/main/java/org/apache/druid/query/aggregation/CountAggregatorFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/CountAggregatorFactory.java
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+import org.apache.druid.math.expr.ExprMacroTable;
import org.apache.druid.segment.ColumnInspector;
import org.apache.druid.segment.ColumnSelectorFactory;
import org.apache.druid.segment.column.ColumnType;
@@ -39,9 +40,7 @@ public class CountAggregatorFactory extends AggregatorFactory
private final String name;
@JsonCreator
- public CountAggregatorFactory(
- @JsonProperty("name") String name
- )
+ public CountAggregatorFactory(@JsonProperty("name") String name)
{
Preconditions.checkNotNull(name, "Must have a valid, non-null aggregator
name");
@@ -93,7 +92,7 @@ public class CountAggregatorFactory extends AggregatorFactory
@Override
public AggregatorFactory getCombiningFactory()
{
- return new LongSumAggregatorFactory(name, name);
+ return new LongSumAggregatorFactory(name, name, null,
ExprMacroTable.nil(), true);
}
@Override
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/LongSumAggregatorFactory.java
b/processing/src/main/java/org/apache/druid/query/aggregation/LongSumAggregatorFactory.java
index 9d1c5fcdd84..8e70fac2536 100644
---
a/processing/src/main/java/org/apache/druid/query/aggregation/LongSumAggregatorFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/LongSumAggregatorFactory.java
@@ -31,11 +31,23 @@ import org.apache.druid.segment.vector.VectorValueSelector;
import javax.annotation.Nullable;
/**
+ * A null-aware aggregator factory. Note that the nullness is handled by
{@link NullableNumericAggregatorFactory}, the
+ * {@link LongSumAggregator}, {@link LongSumBufferAggregator}, and {@link
LongSumVectorAggregator} only aggregates
+ * non-null values, and returns 0 if no data has been aggregated.
+ * <p>
+ * If forceNotNullable is set to true, the aggregator factory will not allow
null values.
*/
public class LongSumAggregatorFactory extends SimpleLongAggregatorFactory
{
private final Supplier<byte[]> cacheKey;
+ private final boolean forceNotNullable;
+
+ public LongSumAggregatorFactory(String name, String fieldName)
+ {
+ this(name, fieldName, null, ExprMacroTable.nil(), false);
+ }
+
@JsonCreator
public LongSumAggregatorFactory(
@JsonProperty("name") String name,
@@ -43,6 +55,17 @@ public class LongSumAggregatorFactory extends
SimpleLongAggregatorFactory
@JsonProperty("expression") @Nullable String expression,
@JacksonInject ExprMacroTable macroTable
)
+ {
+ this(name, fieldName, expression, macroTable, false);
+ }
+
+ public LongSumAggregatorFactory(
+ String name,
+ String fieldName,
+ @Nullable String expression,
+ ExprMacroTable macroTable,
+ boolean forceNotNullable
+ )
{
super(macroTable, name, fieldName, expression);
this.cacheKey = AggregatorUtil.getSimpleAggregatorCacheKeySupplier(
@@ -50,11 +73,13 @@ public class LongSumAggregatorFactory extends
SimpleLongAggregatorFactory
fieldName,
fieldExpression
);
+ this.forceNotNullable = forceNotNullable;
}
- public LongSumAggregatorFactory(String name, String fieldName)
+ @Override
+ public boolean forceNotNullable()
{
- this(name, fieldName, null, ExprMacroTable.nil());
+ return forceNotNullable;
}
@Override
@@ -106,13 +131,13 @@ public class LongSumAggregatorFactory extends
SimpleLongAggregatorFactory
@Override
public AggregatorFactory withName(String newName)
{
- return new LongSumAggregatorFactory(newName, getFieldName(),
getExpression(), macroTable);
+ return new LongSumAggregatorFactory(newName, getFieldName(),
getExpression(), macroTable, forceNotNullable);
}
@Override
public AggregatorFactory getCombiningFactory()
{
- return new LongSumAggregatorFactory(name, name, null, macroTable);
+ return new LongSumAggregatorFactory(name, name, null, macroTable,
forceNotNullable);
}
@Override
@@ -128,6 +153,7 @@ public class LongSumAggregatorFactory extends
SimpleLongAggregatorFactory
"fieldName='" + fieldName + '\'' +
", expression='" + expression + '\'' +
", name='" + name + '\'' +
+ ", forceNotNullable='" + forceNotNullable + '\'' +
'}';
}
}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/NullableNumericAggregatorFactory.java
b/processing/src/main/java/org/apache/druid/query/aggregation/NullableNumericAggregatorFactory.java
index 3dfc2919e44..49dadfd3fea 100644
---
a/processing/src/main/java/org/apache/druid/query/aggregation/NullableNumericAggregatorFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/NullableNumericAggregatorFactory.java
@@ -48,22 +48,36 @@ import org.apache.druid.segment.vector.VectorValueSelector;
public abstract class NullableNumericAggregatorFactory<T extends
BaseNullableColumnValueSelector>
extends AggregatorFactory
{
+ /**
+ * If true, this aggregator will not check for null inputs, instead directly
delegating to the underlying aggregator.
+ * Currently, this is only used by {@link
CountAggregatorFactory#getCombiningFactory}.
+ * This isn't entirely safe, as some selectors might throw an exception when
attempting to retrieve a primitive value from `null`.
+ */
+ public boolean forceNotNullable()
+ {
+ return false;
+ }
+
@Override
public final Aggregator factorize(ColumnSelectorFactory
columnSelectorFactory)
{
T selector = selector(columnSelectorFactory);
- BaseNullableColumnValueSelector nullSelector = makeNullSelector(selector,
columnSelectorFactory);
Aggregator aggregator = factorize(columnSelectorFactory, selector);
- return new NullableNumericAggregator(aggregator, nullSelector);
+ if (this.forceNotNullable()) {
+ return aggregator;
+ }
+ return new NullableNumericAggregator(aggregator,
makeNullSelector(selector, columnSelectorFactory));
}
@Override
public final BufferAggregator factorizeBuffered(ColumnSelectorFactory
columnSelectorFactory)
{
T selector = selector(columnSelectorFactory);
- BaseNullableColumnValueSelector nullSelector = makeNullSelector(selector,
columnSelectorFactory);
BufferAggregator aggregator = factorizeBuffered(columnSelectorFactory,
selector);
- return new NullableNumericBufferAggregator(aggregator, nullSelector);
+ if (this.forceNotNullable()) {
+ return aggregator;
+ }
+ return new NullableNumericBufferAggregator(aggregator,
makeNullSelector(selector, columnSelectorFactory));
}
@Override
@@ -72,6 +86,9 @@ public abstract class NullableNumericAggregatorFactory<T
extends BaseNullableCol
Preconditions.checkState(canVectorize(columnSelectorFactory), "Cannot
vectorize");
VectorValueSelector selector = vectorSelector(columnSelectorFactory);
VectorAggregator aggregator = factorizeVector(columnSelectorFactory,
selector);
+ if (this.forceNotNullable()) {
+ return aggregator;
+ }
return new NullableNumericVectorAggregator(aggregator, selector);
}
@@ -79,12 +96,18 @@ public abstract class NullableNumericAggregatorFactory<T
extends BaseNullableCol
public final AggregateCombiner makeNullableAggregateCombiner()
{
AggregateCombiner<?> combiner = makeAggregateCombiner();
+ if (this.forceNotNullable()) {
+ return combiner;
+ }
return new NullableNumericAggregateCombiner<>(combiner);
}
@Override
public final int getMaxIntermediateSizeWithNulls()
{
+ if (this.forceNotNullable()) {
+ return getMaxIntermediateSize();
+ }
return getMaxIntermediateSize() + Byte.BYTES;
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/AggregateProjectionMetadata.java
b/processing/src/main/java/org/apache/druid/segment/AggregateProjectionMetadata.java
index c561757268a..83b762b629d 100644
---
a/processing/src/main/java/org/apache/druid/segment/AggregateProjectionMetadata.java
+++
b/processing/src/main/java/org/apache/druid/segment/AggregateProjectionMetadata.java
@@ -44,6 +44,7 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
+import java.util.Set;
/**
* Aggregate projection schema and row count information to store in {@link
Metadata} which itself is stored inside a
@@ -349,7 +350,8 @@ public class AggregateProjectionMetadata
if (combining != null) {
matchBuilder.remapColumn(queryAgg.getName(),
projectionAgg.getName())
.addReferencedPhysicalColumn(projectionAgg.getName())
- .addPreAggregatedAggregator(combining);
+ .addPreAggregatedAggregator(combining)
+ .addMatchedQueryColumns(queryAgg.requiredFields());
foundMatch = true;
break;
}
@@ -360,6 +362,40 @@ public class AggregateProjectionMetadata
return null;
}
}
+ // validate physical and virtual columns have all been accounted for
+ final Set<String> matchedQueryColumns =
matchBuilder.getMatchedQueryColumns();
+ if (queryCursorBuildSpec.getPhysicalColumns() != null) {
+ for (String queryColumn : queryCursorBuildSpec.getPhysicalColumns()) {
+ // time is special handled
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(queryColumn)) {
+ continue;
+ }
+ if (!matchedQueryColumns.contains(queryColumn)) {
+ matchBuilder = matchRequiredColumn(
+ matchBuilder,
+ queryColumn,
+ queryCursorBuildSpec.getVirtualColumns(),
+ physicalColumnChecker
+ );
+ if (matchBuilder == null) {
+ return null;
+ }
+ }
+ }
+ for (VirtualColumn vc :
queryCursorBuildSpec.getVirtualColumns().getVirtualColumns()) {
+ if (!matchedQueryColumns.contains(vc.getOutputName())) {
+ matchBuilder = matchRequiredColumn(
+ matchBuilder,
+ vc.getOutputName(),
+ queryCursorBuildSpec.getVirtualColumns(),
+ physicalColumnChecker
+ );
+ if (matchBuilder == null) {
+ return null;
+ }
+ }
+ }
+ }
return matchBuilder.build(queryCursorBuildSpec);
}
@@ -395,10 +431,12 @@ public class AggregateProjectionMetadata
Projections.PhysicalColumnChecker physicalColumnChecker
)
{
- final VirtualColumn buildSpecVirtualColumn =
queryVirtualColumns.getVirtualColumn(column);
- if (buildSpecVirtualColumn != null) {
+ final VirtualColumn queryVirtualColumn =
queryVirtualColumns.getVirtualColumn(column);
+ if (queryVirtualColumn != null) {
+ matchBuilder.addMatchedQueryColumn(column)
+
.addMatchedQueryColumns(queryVirtualColumn.requiredColumns());
// check to see if we have an equivalent virtual column defined in the
projection, if so we can
- final VirtualColumn projectionEquivalent =
virtualColumns.findEquivalent(buildSpecVirtualColumn);
+ final VirtualColumn projectionEquivalent =
virtualColumns.findEquivalent(queryVirtualColumn);
if (projectionEquivalent != null) {
final String remapColumnName;
if (Objects.equals(projectionEquivalent.getOutputName(),
timeColumnName)) {
@@ -406,21 +444,18 @@ public class AggregateProjectionMetadata
} else {
remapColumnName = projectionEquivalent.getOutputName();
}
- if (!buildSpecVirtualColumn.getOutputName().equals(remapColumnName))
{
- matchBuilder.remapColumn(
- buildSpecVirtualColumn.getOutputName(),
- remapColumnName
- );
+ if (!queryVirtualColumn.getOutputName().equals(remapColumnName)) {
+ matchBuilder.remapColumn(queryVirtualColumn.getOutputName(),
remapColumnName);
}
return matchBuilder.addReferencedPhysicalColumn(remapColumnName);
}
- matchBuilder.addReferenceedVirtualColumn(buildSpecVirtualColumn);
- final List<String> requiredInputs =
buildSpecVirtualColumn.requiredColumns();
+ matchBuilder.addReferenceedVirtualColumn(queryVirtualColumn);
+ final List<String> requiredInputs =
queryVirtualColumn.requiredColumns();
if (requiredInputs.size() == 1 &&
ColumnHolder.TIME_COLUMN_NAME.equals(requiredInputs.get(0))) {
// special handle time granularity. in the future this should be
reworked to push this concept into the
// virtual column and underlying expression itself, but this will do
for now
- final Granularity virtualGranularity =
Granularities.fromVirtualColumn(buildSpecVirtualColumn);
+ final Granularity virtualGranularity =
Granularities.fromVirtualColumn(queryVirtualColumn);
if (virtualGranularity != null) {
if (virtualGranularity.isFinerThan(granularity)) {
return null;
@@ -434,7 +469,7 @@ public class AggregateProjectionMetadata
} else {
// anything else with __time requires none granularity
if (Granularities.NONE.equals(granularity)) {
- return matchBuilder;
+ return
matchBuilder.addReferencedPhysicalColumn(ColumnHolder.TIME_COLUMN_NAME);
}
return null;
}
@@ -454,7 +489,8 @@ public class AggregateProjectionMetadata
}
} else {
if (physicalColumnChecker.check(name, column)) {
- return matchBuilder.addReferencedPhysicalColumn(column);
+ return matchBuilder.addMatchedQueryColumn(column)
+ .addReferencedPhysicalColumn(column);
}
return null;
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/projections/Projections.java
b/processing/src/main/java/org/apache/druid/segment/projections/Projections.java
index 2a633aadf9d..db067bde85b 100644
---
a/processing/src/main/java/org/apache/druid/segment/projections/Projections.java
+++
b/processing/src/main/java/org/apache/druid/segment/projections/Projections.java
@@ -46,6 +46,7 @@ import org.apache.druid.segment.vector.VectorValueSelector;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -181,6 +182,7 @@ public class Projections
private final Set<VirtualColumn> referencedVirtualColumns;
private final Map<String, String> remapColumns;
private final List<AggregatorFactory> combiningFactories;
+ private final Set<String> matchedQueryColumns;
public ProjectionMatchBuilder()
{
@@ -188,6 +190,7 @@ public class Projections
this.referencedVirtualColumns = new HashSet<>();
this.remapColumns = new HashMap<>();
this.combiningFactories = new ArrayList<>();
+ this.matchedQueryColumns = new HashSet<>();
}
/**
@@ -235,6 +238,23 @@ public class Projections
return this;
}
+ public ProjectionMatchBuilder addMatchedQueryColumn(String queryColumn)
+ {
+ matchedQueryColumns.add(queryColumn);
+ return this;
+ }
+
+ public ProjectionMatchBuilder addMatchedQueryColumns(Collection<String>
queryColumns)
+ {
+ matchedQueryColumns.addAll(queryColumns);
+ return this;
+ }
+
+ public Set<String> getMatchedQueryColumns()
+ {
+ return matchedQueryColumns;
+ }
+
public ProjectionMatch build(CursorBuildSpec queryCursorBuildSpec)
{
return new ProjectionMatch(
diff --git
a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java
b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java
index 3477d3c7db7..f4e4d1dbd24 100644
---
a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java
+++
b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java
@@ -3399,26 +3399,70 @@ public class GroupByQueryRunnerTest extends
InitializedNullHandlingTest
@Test
public void testGroupByWithCardinality()
{
- GroupByQuery query = makeQueryBuilder()
+
+ GroupByQuery.Builder queryBuilder = makeQueryBuilder()
.setDataSource(QueryRunnerTestHelper.DATA_SOURCE)
.setQuerySegmentSpec(QueryRunnerTestHelper.FIRST_TO_THIRD)
.setAggregatorSpecs(QueryRunnerTestHelper.ROWS_COUNT,
QueryRunnerTestHelper.QUALITY_CARDINALITY)
- .setGranularity(QueryRunnerTestHelper.ALL_GRAN)
- .build();
+ .setGranularity(QueryRunnerTestHelper.ALL_GRAN);
+ GroupByQuery queryNoProjection =
queryBuilder.setContext(Map.of(QueryContexts.NO_PROJECTIONS, "true")).build();
+ GroupByQuery queryWithProjection = queryBuilder.setContext(Map.of(
+ QueryContexts.USE_PROJECTION,
+ "daily_countAndQualityCardinalityAndMaxLongNullable"
+ )).build();
+ // act
+ Iterable<ResultRow> resultsNoProjection =
GroupByQueryRunnerTestHelper.runQuery(factory, runner, queryNoProjection);
+ Iterable<ResultRow> resultsWithProjection =
GroupByQueryRunnerTestHelper.runQuery(
+ factory,
+ runner,
+ queryWithProjection
+ );
- List<ResultRow> expectedResults = Collections.singletonList(
- makeRow(
- query,
- "2011-04-01",
- "rows",
- 26L,
- "cardinality",
- QueryRunnerTestHelper.UNIQUES_9
- )
+ List<ResultRow> expectedResults = Collections.singletonList(makeRow(
+ queryBuilder.build(),
+ "2011-04-01",
+ "rows",
+ 26L,
+ "cardinality",
+ QueryRunnerTestHelper.UNIQUES_9
+ ));
+
+ TestHelper.assertExpectedObjects(expectedResults, resultsNoProjection,
"cardinality");
+ TestHelper.assertExpectedObjects(expectedResults, resultsWithProjection,
"cardinality");
+ }
+
+ @Test
+ public void testGroupByWithCardinalityNoData()
+ {
+
+ GroupByQuery.Builder queryBuilder = makeQueryBuilder()
+ .setDataSource(QueryRunnerTestHelper.DATA_SOURCE)
+ .setInterval("2011-01-21T00:00:00.000Z/P1D")
+ .setAggregatorSpecs(QueryRunnerTestHelper.ROWS_COUNT,
QueryRunnerTestHelper.QUALITY_CARDINALITY)
+ .setGranularity(QueryRunnerTestHelper.ALL_GRAN);
+ GroupByQuery queryNoProjection =
queryBuilder.setContext(Map.of(QueryContexts.NO_PROJECTIONS, "true")).build();
+ GroupByQuery queryWithProjection = queryBuilder.setContext(Map.of(
+ QueryContexts.USE_PROJECTION,
+ "daily_countAndQualityCardinalityAndMaxLongNullable"
+ )).build();
+
+ Iterable<ResultRow> resultsNoProjection =
GroupByQueryRunnerTestHelper.runQuery(factory, runner, queryNoProjection);
+ Iterable<ResultRow> resultsWithProjection =
GroupByQueryRunnerTestHelper.runQuery(
+ factory,
+ runner,
+ queryWithProjection
);
- Iterable<ResultRow> results =
GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
- TestHelper.assertExpectedObjects(expectedResults, results, "cardinality");
+ List<ResultRow> expectedResults = Collections.singletonList(makeRow(
+ queryBuilder.build(),
+ "2011-04-01",
+ "rows",
+ 0L,
+ "cardinality",
+ 0.0d
+ ));
+ TestHelper.assertExpectedObjects(expectedResults, resultsNoProjection, "no
data");
+ TestHelper.assertExpectedObjects(expectedResults, resultsWithProjection,
"no data");
}
@Test
diff --git
a/processing/src/test/java/org/apache/druid/query/groupby/GroupByTimeseriesQueryRunnerTest.java
b/processing/src/test/java/org/apache/druid/query/groupby/GroupByTimeseriesQueryRunnerTest.java
index b3d140d0479..6b2a9cb5120 100644
---
a/processing/src/test/java/org/apache/druid/query/groupby/GroupByTimeseriesQueryRunnerTest.java
+++
b/processing/src/test/java/org/apache/druid/query/groupby/GroupByTimeseriesQueryRunnerTest.java
@@ -280,6 +280,27 @@ public class GroupByTimeseriesQueryRunnerTest extends
TimeseriesQueryRunnerTest
Assert.assertEquals(1870.061F,
result.getValue().getFloatMetric(QueryRunnerTestHelper.FLOAT_MAX_INDEX_METRIC),
0);
}
+ @Override
+ public void testTimeseriesNullableLongMax()
+ {
+ // Skip this test because the timeseries test expects a null value to be
returned for the aggregator even if
+ // there's no data for a period in query, but group by doesn't return data
for periods that have no data.
+ }
+
+ @Override
+ public void testTimeseriesProjections()
+ {
+ // Skip this test because the timeseries test expects a null value to be
returned for the aggregator even if
+ // there's no data for a period in query, but group by doesn't return data
for periods that have no data.
+ }
+
+ @Override
+ public void testTimeseriesProjectionsCounts()
+ {
+ // Skip this test because the timeseries test expects a 0 value to be
returned for the count aggregator even if
+ // there's no data for a period in query, but group by doesn't return data
for periods that have no data.
+ }
+
@Override
public void testEmptyTimeseries()
{
diff --git
a/processing/src/test/java/org/apache/druid/query/metadata/SegmentMetadataQueryTest.java
b/processing/src/test/java/org/apache/druid/query/metadata/SegmentMetadataQueryTest.java
index 5fd5ea98d57..415b2db9572 100644
---
a/processing/src/test/java/org/apache/druid/query/metadata/SegmentMetadataQueryTest.java
+++
b/processing/src/test/java/org/apache/druid/query/metadata/SegmentMetadataQueryTest.java
@@ -103,8 +103,10 @@ public class SegmentMetadataQueryTest extends
InitializedNullHandlingTest
);
private static final ObjectMapper MAPPER = new DefaultObjectMapper();
private static final String DATASOURCE = "testDatasource";
- private static final AggregateProjectionMetadata.Schema PROJECTION_SCHEMA =
TestIndex.PROJECTIONS.get(0).toMetadataSchema();
- private static final int PROJECTION_ROWS = 279;
+ private static final AggregateProjectionMetadata.Schema PROJECTION1_SCHEMA =
TestIndex.PROJECTIONS.get(0).toMetadataSchema();
+ private static final AggregateProjectionMetadata.Schema PROJECTION2_SCHEMA =
TestIndex.PROJECTIONS.get(1).toMetadataSchema();
+ private static final int PROJECTION1_ROWS = 279;
+ private static final int PROJECTION2_ROWS = 93;
@SuppressWarnings("unchecked")
public static QueryRunner makeMMappedQueryRunner(
@@ -226,8 +228,10 @@ public class SegmentMetadataQueryTest extends
InitializedNullHandlingTest
expectedAggregators.put(agg.getName(), agg.getCombiningFactory());
}
final Map<String, AggregateProjectionMetadata> expectedProjections =
ImmutableMap.of(
- PROJECTION_SCHEMA.getName(),
- new AggregateProjectionMetadata(PROJECTION_SCHEMA, PROJECTION_ROWS)
+ PROJECTION1_SCHEMA.getName(),
+ new AggregateProjectionMetadata(PROJECTION1_SCHEMA, PROJECTION1_ROWS),
+ PROJECTION2_SCHEMA.getName(),
+ new AggregateProjectionMetadata(PROJECTION2_SCHEMA, PROJECTION2_ROWS)
);
expectedSegmentAnalysis1 = new SegmentAnalysis(
@@ -705,8 +709,10 @@ public class SegmentMetadataQueryTest extends
InitializedNullHandlingTest
expectedSegmentAnalysis1.getNumRows() +
expectedSegmentAnalysis2.getNumRows(),
expectedAggregators,
ImmutableMap.of(
- PROJECTION_SCHEMA.getName(),
- new AggregateProjectionMetadata(PROJECTION_SCHEMA, PROJECTION_ROWS
* 2)
+ PROJECTION1_SCHEMA.getName(),
+ new AggregateProjectionMetadata(PROJECTION1_SCHEMA,
PROJECTION1_ROWS * 2),
+ PROJECTION2_SCHEMA.getName(),
+ new AggregateProjectionMetadata(PROJECTION2_SCHEMA,
PROJECTION2_ROWS * 2)
),
null,
null,
diff --git
a/processing/src/test/java/org/apache/druid/query/timeseries/TimeseriesQueryRunnerTest.java
b/processing/src/test/java/org/apache/druid/query/timeseries/TimeseriesQueryRunnerTest.java
index d34f3fb84ce..16254aac72b 100644
---
a/processing/src/test/java/org/apache/druid/query/timeseries/TimeseriesQueryRunnerTest.java
+++
b/processing/src/test/java/org/apache/druid/query/timeseries/TimeseriesQueryRunnerTest.java
@@ -52,6 +52,7 @@ import
org.apache.druid.query.aggregation.DoubleSumAggregatorFactory;
import org.apache.druid.query.aggregation.ExpressionLambdaAggregatorFactory;
import org.apache.druid.query.aggregation.FilteredAggregatorFactory;
import org.apache.druid.query.aggregation.FloatSumAggregatorFactory;
+import org.apache.druid.query.aggregation.LongMaxAggregatorFactory;
import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
import
org.apache.druid.query.aggregation.cardinality.CardinalityAggregatorFactory;
import
org.apache.druid.query.aggregation.firstlast.first.DoubleFirstAggregatorFactory;
@@ -87,6 +88,7 @@ import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -331,6 +333,99 @@ public class TimeseriesQueryRunnerTest extends
InitializedNullHandlingTest
Assert.assertEquals(lastResult.toString(), expectedLast,
lastResult.getTimestamp());
}
+ @Test
+ public void testTimeseriesProjections()
+ {
+
+ AggregatorFactory maxQuality = new LongMaxAggregatorFactory("maxQuality",
"qualityLong");
+ Druids.TimeseriesQueryBuilder queryBuilder =
Druids.newTimeseriesQueryBuilder()
+
.dataSource(QueryRunnerTestHelper.DATA_SOURCE)
+
.granularity(Granularities.DAY)
+
.intervals("2011-01-20/2011-01-22")
+
.aggregators(List.of(maxQuality));
+ // construct a query to ignore projection
+ Map<String, Object> contextNoProjection = makeContext();
+ contextNoProjection.put(QueryContexts.NO_PROJECTIONS, "true");
+ TimeseriesQuery queryNoProjection =
queryBuilder.context(contextNoProjection).build();
+ // construct a query to use projection
+ Map<String, Object> contextWithProjection = makeContext();
+ contextWithProjection.put(QueryContexts.USE_PROJECTION,
"daily_market_maxQuality");
+ TimeseriesQuery queryWithProjection =
queryBuilder.context(contextWithProjection).build();
+
+
+ List<Result<TimeseriesResultValue>> resultNoProjection =
runner.run(QueryPlus.wrap(queryNoProjection)).toList();
+ List<Result<TimeseriesResultValue>> resultWithProjection =
runner.run(QueryPlus.wrap(queryWithProjection)).toList();
+
+ List<Result<TimeseriesResultValue>> expectedResults = List.of(
+ new Result<>(DateTimes.of("2011-01-20"),
createTimeseriesResultValue("maxQuality", 1800L)),
+ new Result<>(DateTimes.of("2011-01-21"),
createTimeseriesResultValue("maxQuality", null))
+ );
+ Assert.assertEquals(expectedResults, resultNoProjection);
+ Assert.assertEquals(expectedResults, resultWithProjection);
+ }
+
+ @Test
+ public void testTimeseriesProjectionsCounts()
+ {
+
+ AggregatorFactory countAgg = new CountAggregatorFactory("count");
+ Druids.TimeseriesQueryBuilder queryBuilder =
Druids.newTimeseriesQueryBuilder()
+
.dataSource(QueryRunnerTestHelper.DATA_SOURCE)
+
.granularity(Granularities.DAY)
+
.intervals("2011-01-20/2011-01-22")
+
.aggregators(List.of(countAgg));
+ // construct a query to ignore projection
+ Map<String, Object> contextNoProjection = makeContext();
+ contextNoProjection.put(QueryContexts.NO_PROJECTIONS, "true");
+ TimeseriesQuery queryNoProjection =
queryBuilder.context(contextNoProjection).build();
+ // construct a query to use projection
+ Map<String, Object> contextWithProjection = makeContext();
+ contextWithProjection.put(QueryContexts.USE_PROJECTION,
"daily_countAndQualityCardinalityAndMaxLongNullable");
+ TimeseriesQuery queryWithProjection =
queryBuilder.context(contextWithProjection).build();
+
+
+ List<Result<TimeseriesResultValue>> resultNoProjection =
runner.run(QueryPlus.wrap(queryNoProjection)).toList();
+ List<Result<TimeseriesResultValue>> resultWithProjection =
runner.run(QueryPlus.wrap(queryWithProjection)).toList();
+
+ List<Result<TimeseriesResultValue>> expectedResults = List.of(
+ new Result<>(DateTimes.of("2011-01-20"),
createTimeseriesResultValue("count", 13L)),
+ new Result<>(DateTimes.of("2011-01-21"),
createTimeseriesResultValue("count", 0L))
+ );
+ Assert.assertEquals(expectedResults, resultNoProjection);
+ Assert.assertEquals(expectedResults, resultWithProjection);
+ }
+
+ @Test
+ public void testTimeseriesNullableLongMax()
+ {
+
+ AggregatorFactory longNullableMax = new
LongMaxAggregatorFactory("longNullableMax", "longNumericNull");
+ Druids.TimeseriesQueryBuilder queryBuilder =
Druids.newTimeseriesQueryBuilder()
+
.dataSource(QueryRunnerTestHelper.DATA_SOURCE)
+
.granularity(Granularities.DAY)
+
.intervals("2011-01-20/2011-01-22")
+
.aggregators(List.of(longNullableMax));
+ // construct a query to ignore projection
+ Map<String, Object> contextNoProjection = makeContext();
+ contextNoProjection.put(QueryContexts.NO_PROJECTIONS, "true");
+ TimeseriesQuery queryNoProjection =
queryBuilder.context(contextNoProjection).build();
+ // construct a query to use projection
+ Map<String, Object> contextWithProjection = makeContext();
+ contextWithProjection.put(QueryContexts.USE_PROJECTION,
"daily_countAndQualityCardinalityAndMaxLongNullable");
+ TimeseriesQuery queryWithProjection =
queryBuilder.context(contextWithProjection).build();
+
+
+ List<Result<TimeseriesResultValue>> resultNoProjection =
runner.run(QueryPlus.wrap(queryNoProjection)).toList();
+ List<Result<TimeseriesResultValue>> resultWithProjection =
runner.run(QueryPlus.wrap(queryWithProjection)).toList();
+
+ List<Result<TimeseriesResultValue>> expectedResults = List.of(
+ new Result<>(DateTimes.of("2011-01-20"),
createTimeseriesResultValue("longNullableMax", 80L)),
+ new Result<>(DateTimes.of("2011-01-21"),
createTimeseriesResultValue("longNullableMax", null))
+ );
+ Assert.assertEquals(expectedResults, resultNoProjection);
+ Assert.assertEquals(expectedResults, resultWithProjection);
+ }
+
@Test
public void testFullOnTimeseriesMaxMin()
{
@@ -3201,4 +3296,11 @@ public class TimeseriesQueryRunnerTest extends
InitializedNullHandlingTest
expectedException.expectMessage("Cannot vectorize!");
}
}
+
+ private static TimeseriesResultValue createTimeseriesResultValue(String key,
@Nullable Object val)
+ {
+ Map<String, Object> map = new HashMap<>();
+ map.put(key, val);
+ return new TimeseriesResultValue(map);
+ }
}
diff --git a/processing/src/test/java/org/apache/druid/segment/TestIndex.java
b/processing/src/test/java/org/apache/druid/segment/TestIndex.java
index c538c2b282d..712d5d95002 100644
--- a/processing/src/test/java/org/apache/druid/segment/TestIndex.java
+++ b/processing/src/test/java/org/apache/druid/segment/TestIndex.java
@@ -45,12 +45,14 @@ import
org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.java.util.common.parsers.JSONPathSpec;
import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
import org.apache.druid.query.aggregation.DoubleMaxAggregatorFactory;
import org.apache.druid.query.aggregation.DoubleMinAggregatorFactory;
import org.apache.druid.query.aggregation.DoubleSumAggregatorFactory;
import org.apache.druid.query.aggregation.FloatMaxAggregatorFactory;
import org.apache.druid.query.aggregation.FloatMinAggregatorFactory;
import org.apache.druid.query.aggregation.FloatSumAggregatorFactory;
+import org.apache.druid.query.aggregation.LongMaxAggregatorFactory;
import
org.apache.druid.query.aggregation.hyperloglog.HyperUniquesAggregatorFactory;
import org.apache.druid.query.aggregation.hyperloglog.HyperUniquesSerde;
import org.apache.druid.query.expression.TestExprMacroTable;
@@ -76,6 +78,8 @@ import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
+import static org.apache.druid.query.QueryRunnerTestHelper.QUALITY_CARDINALITY;
+
/**
*
*/
@@ -183,13 +187,20 @@ public class TestIndex
};
public static final ImmutableList<AggregateProjectionSpec> PROJECTIONS =
ImmutableList.of(
new AggregateProjectionSpec(
- "index_projection",
+ "daily_market_maxQuality",
+
VirtualColumns.create(Granularities.toVirtualColumn(Granularities.DAY,
"__gran")),
+ List.of(new LongDimensionSchema("__gran"), new
StringDimensionSchema("market")),
+ new AggregatorFactory[]{new LongMaxAggregatorFactory("maxQuality",
"qualityLong")}
+ ),
+ new AggregateProjectionSpec(
+ "daily_countAndQualityCardinalityAndMaxLongNullable",
VirtualColumns.create(Granularities.toVirtualColumn(Granularities.DAY,
"__gran")),
- Arrays.asList(
- new LongDimensionSchema("__gran"),
- new StringDimensionSchema("market")
- ),
- new AggregatorFactory[]{new DoubleMaxAggregatorFactory("maxQuality",
"qualityLong")}
+ List.of(new LongDimensionSchema("__gran")),
+ new AggregatorFactory[]{
+ new CountAggregatorFactory("count"),
+ QUALITY_CARDINALITY,
+ new LongMaxAggregatorFactory("longNullableMax",
"longNumericNull")
+ }
)
);
public static final IndexSpec INDEX_SPEC = IndexSpec.DEFAULT;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
