Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ public CardinalityVectorProcessor makeLongProcessor(ColumnCapabilities capabilit
@Override
public CardinalityVectorProcessor makeObjectProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
{
return NilCardinalityVectorProcessor.INSTANCE;
// Handles string-as-object and complex types.
return new StringObjectCardinalityVectorProcessor(selector);
}
}

This file was deleted.

Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/

package org.apache.druid.query.aggregation.cardinality.vector;

import org.apache.druid.common.config.NullHandling;
import org.apache.druid.hll.HyperLogLogCollector;
import org.apache.druid.query.aggregation.cardinality.types.StringCardinalityAggregatorColumnSelectorStrategy;
import org.apache.druid.segment.vector.VectorObjectSelector;

import javax.annotation.Nullable;
import java.nio.ByteBuffer;
import java.util.List;

public class StringObjectCardinalityVectorProcessor implements CardinalityVectorProcessor
{
private final VectorObjectSelector selector;

public StringObjectCardinalityVectorProcessor(final VectorObjectSelector selector)
{
this.selector = selector;
}

@Override
public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
{
// Save position, limit and restore later instead of allocating a new ByteBuffer object
final int oldPosition = buf.position();
final int oldLimit = buf.limit();

try {
final Object[] vector = selector.getObjectVector();

buf.limit(position + HyperLogLogCollector.getLatestNumBytesForDenseStorage());
buf.position(position);

final HyperLogLogCollector collector = HyperLogLogCollector.makeCollector(buf);

for (int i = startRow; i < endRow; i++) {
addObjectIfString(collector, vector[i]);
}
}
finally {
buf.limit(oldLimit);
buf.position(oldPosition);
}
}

@Override
public void aggregate(ByteBuffer buf, int numRows, int[] positions, @Nullable int[] rows, int positionOffset)
{
// Save position, limit and restore later instead of allocating a new ByteBuffer object
final int oldPosition = buf.position();
final int oldLimit = buf.limit();

try {
final Object[] vector = selector.getObjectVector();

for (int i = 0; i < numRows; i++) {
final Object obj = vector[rows != null ? rows[i] : i];

if (NullHandling.replaceWithDefault() || obj != null) {
final int position = positions[i] + positionOffset;
buf.limit(position + HyperLogLogCollector.getLatestNumBytesForDenseStorage());
buf.position(position);
final HyperLogLogCollector collector = HyperLogLogCollector.makeCollector(buf);
addObjectIfString(collector, obj);
}
}
}
finally {
buf.limit(oldLimit);
buf.position(oldPosition);
}
}

/**
* Adds an Object to a HyperLogLogCollector. If the object is a {@code List<String>} or {@code String} then
* the individual Strings are added to the collector.
*
* If the object is any other type (including null) then behavior depends on null-handling mode:
*
* - In SQL-compatible mode, ignore non-strings and nulls.
* - In replace-with-default mode, treat all non-strings and nulls as empty strings.
*/
private static void addObjectIfString(final HyperLogLogCollector collector, @Nullable final Object obj)
{
if (obj instanceof String) {
StringCardinalityAggregatorColumnSelectorStrategy.addStringToCollector(collector, (String) obj);
} else if (obj instanceof List) {
//noinspection unchecked
for (String s : (List<String>) obj) {
StringCardinalityAggregatorColumnSelectorStrategy.addStringToCollector(collector, s);
}
} else {
StringCardinalityAggregatorColumnSelectorStrategy.addStringToCollector(collector, null);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -41,17 +41,25 @@
public interface VectorColumnProcessorFactory<T>
{
/**
* Called when {@link ColumnCapabilities#getType()} is STRING and the underlying column always has a single value
* Called only if {@link ColumnCapabilities#getType()} is STRING and the underlying column always has a single value
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

woops, I guess I only fixed up docs for VectorColumnSelectorFactory in #10613, should've done these too, my bad

* per row.
*
* Note that for STRING-typed columns where the dictionary does not exist or is not expected to be useful,
* {@link #makeObjectProcessor} may be called instead. To handle all string inputs properly, processors must implement
* all three methods (single-value, multi-value, object).
*/
T makeSingleValueDimensionProcessor(
ColumnCapabilities capabilities,
SingleValueDimensionVectorSelector selector
);

/**
* Called when {@link ColumnCapabilities#getType()} is STRING and the underlying column may have multiple values
* Called only if {@link ColumnCapabilities#getType()} is STRING and the underlying column may have multiple values
* per row.
*
* Note that for STRING-typed columns where the dictionary does not exist or is not expected to be useful,
* {@link #makeObjectProcessor} may be called instead. To handle all string inputs properly, processors must implement
* all three methods (single-value, multi-value, object).
*/
T makeMultiValueDimensionProcessor(
ColumnCapabilities capabilities,
Expand All @@ -74,7 +82,8 @@ T makeMultiValueDimensionProcessor(
T makeLongProcessor(ColumnCapabilities capabilities, VectorValueSelector selector);

/**
* Called when {@link ColumnCapabilities#getType()} is COMPLEX.
* Called when {@link ColumnCapabilities#getType()} is COMPLEX. May also be called for STRING typed columns in
* cases where the dictionary does not exist or is not expected to be useful.
*/
T makeObjectProcessor(@SuppressWarnings("unused") ColumnCapabilities capabilities, VectorObjectSelector selector);
}
Original file line number Diff line number Diff line change
Expand Up @@ -8855,6 +8855,78 @@ public void testGroupByCardinalityAggOnFloat()
TestHelper.assertExpectedObjects(expectedResults, results, "cardinality-agg");
}

@Test
public void testGroupByCardinalityAggOnMultiStringExpression()
{
GroupByQuery query = makeQueryBuilder()
.setDataSource(QueryRunnerTestHelper.DATA_SOURCE)
.setQuerySegmentSpec(QueryRunnerTestHelper.FIRST_TO_THIRD)
.setVirtualColumns(
new ExpressionVirtualColumn("v0", "concat(quality,market)", ValueType.STRING, TestExprMacroTable.INSTANCE)
)
.setAggregatorSpecs(
QueryRunnerTestHelper.ROWS_COUNT,
new CardinalityAggregatorFactory(
"numVals",
ImmutableList.of(DefaultDimensionSpec.of("v0")),
false
)
)
.setGranularity(QueryRunnerTestHelper.ALL_GRAN)
.build();

List<ResultRow> expectedResults = Collections.singletonList(
makeRow(
query,
"2011-04-01",
"rows",
26L,
"numVals",
13.041435202975777d
)
);

Iterable<ResultRow> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
TestHelper.assertExpectedObjects(expectedResults, results, "cardinality-agg");
}

@Test
public void testGroupByCardinalityAggOnHyperUnique()
{
// Cardinality aggregator on complex columns (like hyperUnique) returns 0.

GroupByQuery query = makeQueryBuilder()
.setDataSource(QueryRunnerTestHelper.DATA_SOURCE)
.setQuerySegmentSpec(QueryRunnerTestHelper.FIRST_TO_THIRD)
.setAggregatorSpecs(
QueryRunnerTestHelper.ROWS_COUNT,
new CardinalityAggregatorFactory(
"cardinality",
ImmutableList.of(DefaultDimensionSpec.of("quality_uniques")),
false
),
new HyperUniquesAggregatorFactory("hyperUnique", "quality_uniques", false, false)
)
.setGranularity(QueryRunnerTestHelper.ALL_GRAN)
.build();

List<ResultRow> expectedResults = Collections.singletonList(
makeRow(
query,
"2011-04-01",
"rows",
26L,
"cardinality",
NullHandling.replaceWithDefault() ? 1.0002442201269182 : 0.0d,
"hyperUnique",
9.019833517963864d
)
);

Iterable<ResultRow> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
TestHelper.assertExpectedObjects(expectedResults, results, "cardinality-agg");
}

@Test
public void testGroupByLongColumn()
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,9 +48,12 @@
import org.apache.druid.query.aggregation.ExpressionLambdaAggregatorFactory;
import org.apache.druid.query.aggregation.FilteredAggregatorFactory;
import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
import org.apache.druid.query.aggregation.cardinality.CardinalityAggregatorFactory;
import org.apache.druid.query.aggregation.first.DoubleFirstAggregatorFactory;
import org.apache.druid.query.aggregation.hyperloglog.HyperUniquesAggregatorFactory;
import org.apache.druid.query.aggregation.last.DoubleLastAggregatorFactory;
import org.apache.druid.query.aggregation.post.FieldAccessPostAggregator;
import org.apache.druid.query.dimension.DefaultDimensionSpec;
import org.apache.druid.query.expression.TestExprMacroTable;
import org.apache.druid.query.extraction.MapLookupExtractor;
import org.apache.druid.query.filter.AndDimFilter;
Expand Down Expand Up @@ -3035,6 +3038,84 @@ public void testTimeseriesWithExpressionAggregator()
assertExpectedResults(expectedResults, results);
}

@Test
public void testTimeseriesCardinalityAggOnMultiStringExpression()
{
TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
.dataSource(QueryRunnerTestHelper.DATA_SOURCE)
.intervals(QueryRunnerTestHelper.FIRST_TO_THIRD)
.virtualColumns(
new ExpressionVirtualColumn("v0", "concat(quality,market)", ValueType.STRING, TestExprMacroTable.INSTANCE)
)
.aggregators(
QueryRunnerTestHelper.ROWS_COUNT,
new CardinalityAggregatorFactory(
"numVals",
ImmutableList.of(DefaultDimensionSpec.of("v0")),
false
)
)
.granularity(QueryRunnerTestHelper.ALL_GRAN)
.build();

List<Result<TimeseriesResultValue>> expectedResults = Collections.singletonList(
new Result<>(
DateTimes.of("2011-04-01"),
new TimeseriesResultValue(
ImmutableMap.of(
"rows",
26L,
"numVals",
13.041435202975777d
)
)
)
);

Iterable<Result<TimeseriesResultValue>> results = runner.run(QueryPlus.wrap(query)).toList();
assertExpectedResults(expectedResults, results);
}

@Test
public void testTimeseriesCardinalityAggOnHyperUnique()
{
// Cardinality aggregator on complex columns (like hyperUnique) returns 0.

TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
.dataSource(QueryRunnerTestHelper.DATA_SOURCE)
.intervals(QueryRunnerTestHelper.FIRST_TO_THIRD)
.aggregators(
QueryRunnerTestHelper.ROWS_COUNT,
new CardinalityAggregatorFactory(
"cardinality",
ImmutableList.of(DefaultDimensionSpec.of("quality_uniques")),
false
),
new HyperUniquesAggregatorFactory("hyperUnique", "quality_uniques", false, false)
)
.granularity(QueryRunnerTestHelper.ALL_GRAN)
.build();

List<Result<TimeseriesResultValue>> expectedResults = Collections.singletonList(
new Result<>(
DateTimes.of("2011-04-01"),
new TimeseriesResultValue(
ImmutableMap.of(
"rows",
26L,
"cardinality",
NullHandling.replaceWithDefault() ? 1.0002442201269182 : 0.0d,
"hyperUnique",
9.019833517963864d
)
)
)
);

Iterable<Result<TimeseriesResultValue>> results = runner.run(QueryPlus.wrap(query)).toList();
assertExpectedResults(expectedResults, results);
}

private Map<String, Object> makeContext()
{
return makeContext(ImmutableMap.of());
Expand Down
Loading