This is an automated email from the ASF dual-hosted git repository.
FrankChen021 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 3b44cc3857d perf: reuse expression selectors to reduce CPU and memory
usage (#20450)
3b44cc3857d is described below
commit 3b44cc3857d0635df431aaeee55e778364462b59
Author: Frank Chen <[email protected]>
AuthorDate: Wed Oct 14 14:05:50 2026 +0800
perf: reuse expression selectors to reduce CPU and memory usage (#20450)
* perf: cache incremental-index expression plans
* fix: avoid sharing mutable expression state
* fix: return fresh expressions from plan cache
* test: use daemon executor for expression cache concurrency
* docs: explain expression plan cache array
* perf: reuse expression selectors during ingestion
* test: cover expression selector reuse with arrays
* test: clarify expression ingestion expectation
* perf: add expression ingestion benchmark
* fix: reevaluate non-deterministic ingestion expressions
---
.../indexing/ExpressionIngestionBenchmark.java | 136 ++++++++++
.../incremental/OnheapIncrementalIndex.java | 26 +-
.../segment/virtual/ExprEvalSelectorCache.java | 42 +++
.../druid/segment/virtual/ExpressionSelectors.java | 21 +-
.../RowBasedExpressionColumnValueSelector.java | 20 +-
.../incremental/OnheapIncrementalIndexTest.java | 300 ++++++++++++++++++++-
.../segment/virtual/ExpressionSelectorsTest.java | 43 +++
7 files changed, 571 insertions(+), 17 deletions(-)
diff --git
a/benchmarks/src/test/java/org/apache/druid/benchmark/indexing/ExpressionIngestionBenchmark.java
b/benchmarks/src/test/java/org/apache/druid/benchmark/indexing/ExpressionIngestionBenchmark.java
new file mode 100644
index 00000000000..1295e361976
--- /dev/null
+++
b/benchmarks/src/test/java/org/apache/druid/benchmark/indexing/ExpressionIngestionBenchmark.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.benchmark.indexing;
+
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.math.expr.ExpressionProcessing;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.DoubleSumAggregatorFactory;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+import org.apache.druid.segment.incremental.IncrementalIndexSchema;
+import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Measures on-heap incremental ingestion of 100,000 rows into 25,000 rollup
keys using three expression metrics.
+ * Input-row construction is performed once per trial and excluded from the
measured invocation.
+ */
+@State(Scope.Benchmark)
+@Fork(3)
+@Warmup(iterations = 3)
+@Measurement(iterations = 7)
+@BenchmarkMode(Mode.SingleShotTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+public class ExpressionIngestionBenchmark
+{
+ private static final int ROW_COUNT = 100_000;
+ private static final int ROLLUP_KEY_COUNT = 25_000;
+
+ private List<InputRow> rows;
+ private AggregatorFactory[] metrics;
+
+ @Setup(Level.Trial)
+ public void setup()
+ {
+ ExpressionProcessing.initializeForTests();
+ rows = makeRows();
+
+ final ExprMacroTable macroTable = ExprMacroTable.nil();
+ metrics = new AggregatorFactory[]{
+ new DoubleSumAggregatorFactory(
+ "exprDouble1",
+ null,
+ "if(x > 0, x * y + z, y - z)",
+ macroTable
+ ),
+ new DoubleSumAggregatorFactory(
+ "exprDouble2",
+ null,
+ "x + y * z",
+ macroTable
+ ),
+ new LongSumAggregatorFactory(
+ "exprLong",
+ null,
+ "if(x > 0, x + y, z)",
+ macroTable
+ )
+ };
+ }
+
+ @Benchmark
+ public int ingest()
+ {
+ try (
+ final OnheapIncrementalIndex index = (OnheapIncrementalIndex) new
OnheapIncrementalIndex.Builder()
+ .setIndexSchema(
+ new IncrementalIndexSchema.Builder()
+ .withDimensionsSpec(
+ new DimensionsSpec(Collections.singletonList(new
StringDimensionSchema("key")))
+ )
+ .withMetrics(metrics)
+ .withRollup(true)
+ .build()
+ )
+ .setMaxRowCount(ROW_COUNT + 1)
+ .build()
+ ) {
+ for (final InputRow row : rows) {
+ index.add(row);
+ }
+ return index.numRows();
+ }
+ }
+
+ private static List<InputRow> makeRows()
+ {
+ final List<InputRow> rows = new ArrayList<>(ROW_COUNT);
+ for (int rowNumber = 0; rowNumber < ROW_COUNT; rowNumber++) {
+ final Map<String, Object> event = new HashMap<>();
+ event.put("key", "key-" + rowNumber % ROLLUP_KEY_COUNT);
+ event.put("x", (long) (rowNumber % 1_000) - 500L);
+ event.put("y", (long) (rowNumber % 100));
+ event.put("z", (long) (rowNumber % 17));
+ rows.add(new MapBasedInputRow(0L, Collections.singletonList("key"),
event));
+ }
+ return rows;
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java
b/processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java
index f9d6ff965c3..3fcfaeff706 100644
---
a/processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java
+++
b/processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java
@@ -34,6 +34,8 @@ import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.io.Closer;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.java.util.common.parsers.ParseException;
+import org.apache.druid.math.expr.Expr;
+import org.apache.druid.math.expr.ExprEval;
import org.apache.druid.query.aggregation.Aggregator;
import org.apache.druid.query.aggregation.AggregatorAndSize;
import org.apache.druid.query.aggregation.AggregatorFactory;
@@ -52,6 +54,8 @@ import org.apache.druid.segment.column.ValueType;
import org.apache.druid.segment.projections.AggregateProjectionSchema;
import org.apache.druid.segment.projections.Projections;
import org.apache.druid.segment.projections.QueryableProjection;
+import org.apache.druid.segment.virtual.ExprEvalSelectorCache;
+import org.apache.druid.segment.virtual.ExpressionSelectors;
import org.apache.druid.utils.JvmUtils;
import javax.annotation.Nullable;
@@ -62,6 +66,7 @@ import java.util.Collection;
import java.util.Comparator;
import java.util.Deque;
import java.util.HashMap;
+import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -670,18 +675,20 @@ public class OnheapIncrementalIndex extends
IncrementalIndex
}
/**
- * Caches references to selector objects for each column instead of creating
a new object each time in order to save
- * heap space.
+ * Caches references to selector objects instead of creating them for each
row key in order to save heap space and
+ * avoid repeated expression planning and selector construction.
*/
- static class CachingColumnSelectorFactory implements ColumnSelectorFactory
+ static class CachingColumnSelectorFactory implements ColumnSelectorFactory,
ExprEvalSelectorCache
{
private final HashMap<String, ColumnValueSelector<?>> columnSelectorMap;
+ private final IdentityHashMap<Expr, ColumnValueSelector<ExprEval>>
expressionSelectorMap;
private final ColumnSelectorFactory delegate;
public CachingColumnSelectorFactory(ColumnSelectorFactory delegate)
{
this.delegate = delegate;
this.columnSelectorMap = new HashMap<>();
+ this.expressionSelectorMap = new IdentityHashMap<>();
}
@Override
@@ -706,6 +713,19 @@ public class OnheapIncrementalIndex extends
IncrementalIndex
return existing != null ? existing : columnValueSelector;
}
+ @Override
+ public ColumnValueSelector<ExprEval> getOrCreateExprEvalSelector(final
Expr expression)
+ {
+ ColumnValueSelector<ExprEval> selector =
expressionSelectorMap.get(expression);
+ if (selector == null) {
+ // Do not use computeIfAbsent: building the expression selector can
re-enter makeColumnValueSelector on this
+ // factory through expression bindings.
+ selector = ExpressionSelectors.makeExprEvalSelectorUncached(this,
expression);
+ expressionSelectorMap.put(expression, selector);
+ }
+ return selector;
+ }
+
@Nullable
@Override
public ColumnCapabilities getColumnCapabilities(String columnName)
diff --git
a/processing/src/main/java/org/apache/druid/segment/virtual/ExprEvalSelectorCache.java
b/processing/src/main/java/org/apache/druid/segment/virtual/ExprEvalSelectorCache.java
new file mode 100644
index 00000000000..bc228d17d4a
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/virtual/ExprEvalSelectorCache.java
@@ -0,0 +1,42 @@
+/*
+ * 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.segment.virtual;
+
+import org.apache.druid.math.expr.Expr;
+import org.apache.druid.math.expr.ExprEval;
+import org.apache.druid.segment.ColumnValueSelector;
+
+/**
+ * Supplies expression selectors cached for the lifetime of an ingestion
column selector factory.
+ *
+ * Implementations may return one mutable selector shared across callers.
Callers must only use returned selectors from
+ * the thread that ingests rows. Query-time column selector factories, which
may be used from multiple threads, must not
+ * implement this interface.
+ */
+public interface ExprEvalSelectorCache
+{
+ /**
+ * Returns a cached or newly created selector for an expression.
+ *
+ * @param expression expression to evaluate
+ * @return cached or newly created selector
+ */
+ ColumnValueSelector<ExprEval> getOrCreateExprEvalSelector(Expr expression);
+}
diff --git
a/processing/src/main/java/org/apache/druid/segment/virtual/ExpressionSelectors.java
b/processing/src/main/java/org/apache/druid/segment/virtual/ExpressionSelectors.java
index dfb97cf8b5f..66e49c2b707 100644
---
a/processing/src/main/java/org/apache/druid/segment/virtual/ExpressionSelectors.java
+++
b/processing/src/main/java/org/apache/druid/segment/virtual/ExpressionSelectors.java
@@ -133,13 +133,29 @@ public class ExpressionSelectors
Expr expression
)
{
- ExpressionPlan plan = ExpressionPlanner.plan(
+ if (columnSelectorFactory instanceof ExprEvalSelectorCache) {
+ return ((ExprEvalSelectorCache)
columnSelectorFactory).getOrCreateExprEvalSelector(expression);
+ }
+ return makeExprEvalSelectorUncached(columnSelectorFactory, expression);
+ }
+
+ /**
+ * Makes an uncached expression selector. Used by {@link
ExprEvalSelectorCache} implementations to avoid recursively
+ * calling back into the cache.
+ */
+ public static ColumnValueSelector<ExprEval> makeExprEvalSelectorUncached(
+ ColumnSelectorFactory columnSelectorFactory,
+ Expr expression
+ )
+ {
+ final ExpressionPlan plan = ExpressionPlanner.plan(
columnSelectorFactory,
Expr.singleThreaded(expression, columnSelectorFactory)
);
final RowIdSupplier rowIdSupplier =
columnSelectorFactory.getRowIdSupplier();
- if (plan.is(ExpressionPlan.Trait.SINGLE_INPUT_SCALAR)) {
+ // Non-deterministic expressions must be evaluated for each row, so they
cannot use input-value result caches.
+ if (plan.is(ExpressionPlan.Trait.SINGLE_INPUT_SCALAR) &&
!plan.getAnalysis().isNonDeterministic()) {
final String column = plan.getSingleInputName();
final ColumnType inputType = plan.getSingleInputType();
if (inputType.is(ValueType.LONG)) {
@@ -235,7 +251,6 @@ public class ExpressionSelectors
}
}
-
/**
* Returns whether an expression can be applied to unique values of a
particular column (like those in a dictionary)
* rather than being applied to each row individually.
diff --git
a/processing/src/main/java/org/apache/druid/segment/virtual/RowBasedExpressionColumnValueSelector.java
b/processing/src/main/java/org/apache/druid/segment/virtual/RowBasedExpressionColumnValueSelector.java
index 085510aaccf..0ca0394f966 100644
---
a/processing/src/main/java/org/apache/druid/segment/virtual/RowBasedExpressionColumnValueSelector.java
+++
b/processing/src/main/java/org/apache/druid/segment/virtual/RowBasedExpressionColumnValueSelector.java
@@ -20,15 +20,15 @@
package org.apache.druid.segment.virtual;
import com.google.common.collect.Lists;
-import it.unimi.dsi.fastutil.ints.Int2ObjectArrayMap;
-import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
import org.apache.druid.math.expr.Expr;
import org.apache.druid.math.expr.ExprEval;
import org.apache.druid.math.expr.Parser;
import org.apache.druid.segment.RowIdSupplier;
import javax.annotation.Nullable;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
/**
@@ -44,7 +44,7 @@ public class RowBasedExpressionColumnValueSelector extends
BaseExpressionColumnV
private final Expr expression;
private final List<String> unknownColumns;
private final Expr.BindingAnalysis baseBindingAnalysis;
- private final Int2ObjectMap<Expr> transformedCache;
+ private final Map<List<String>, Expr> transformedCache;
public RowBasedExpressionColumnValueSelector(
ExpressionPlan plan,
@@ -60,7 +60,7 @@ public class RowBasedExpressionColumnValueSelector extends
BaseExpressionColumnV
.filter(x ->
!plan.getAnalysis().getArrayBindings().contains(x))
.collect(Collectors.toList());
this.baseBindingAnalysis = plan.getAnalysis();
- this.transformedCache = new Int2ObjectArrayMap<>(unknownColumns.size());
+ this.transformedCache = new HashMap<>();
}
@Override
@@ -78,13 +78,13 @@ public class RowBasedExpressionColumnValueSelector extends
BaseExpressionColumnV
// if there are arrays, we need to transform the expression to one that
applies each value of the array to the
// base expression, we keep a cache of transformed expressions to minimize
extra work
if (!arrayBindings.isEmpty()) {
- final int key = arrayBindings.hashCode();
- if (transformedCache.containsKey(key)) {
- return transformedCache.get(key).eval(bindings);
+ final Expr transformed = transformedCache.get(arrayBindings);
+ if (transformed != null) {
+ return transformed.eval(bindings);
}
- final Expr transformed = Parser.applyUnappliedBindings(expression,
baseBindingAnalysis, arrayBindings);
- transformedCache.put(key, transformed);
- return transformed.eval(bindings);
+ final Expr newTransformed = Parser.applyUnappliedBindings(expression,
baseBindingAnalysis, arrayBindings);
+ transformedCache.put(arrayBindings, newTransformed);
+ return newTransformed.eval(bindings);
}
// no arrays for this row, evaluate base expression
return expression.eval(bindings);
diff --git
a/processing/src/test/java/org/apache/druid/segment/incremental/OnheapIncrementalIndexTest.java
b/processing/src/test/java/org/apache/druid/segment/incremental/OnheapIncrementalIndexTest.java
index d4997bd008c..f626740d847 100644
---
a/processing/src/test/java/org/apache/druid/segment/incremental/OnheapIncrementalIndexTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/incremental/OnheapIncrementalIndexTest.java
@@ -21,7 +21,9 @@ package org.apache.druid.segment.incremental;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
import nl.jqno.equalsverifier.EqualsVerifier;
+import org.apache.druid.data.input.InputRow;
import org.apache.druid.data.input.MapBasedInputRow;
import org.apache.druid.data.input.impl.AggregateProjectionSpec;
import org.apache.druid.data.input.impl.DimensionsSpec;
@@ -31,23 +33,37 @@ import
org.apache.druid.data.input.impl.StringDimensionSchema;
import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.math.expr.Expr;
+import org.apache.druid.math.expr.ExprEval;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.math.expr.Parser;
+import org.apache.druid.query.aggregation.Aggregator;
import org.apache.druid.query.aggregation.AggregatorFactory;
import org.apache.druid.query.aggregation.DoubleSumAggregatorFactory;
import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
import org.apache.druid.query.expression.TestExprMacroTable;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.ColumnValueSelector;
import org.apache.druid.segment.IndexBuilder;
+import org.apache.druid.segment.TestColumnSelectorFactory;
import org.apache.druid.segment.TestHelper;
+import org.apache.druid.segment.TestObjectColumnSelector;
import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.virtual.ExpressionSelectors;
import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
+import org.apache.druid.testing.InitializedNullHandlingTest;
import org.joda.time.DateTime;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
-public class OnheapIncrementalIndexTest
+public class OnheapIncrementalIndexTest extends InitializedNullHandlingTest
{
private static final ObjectMapper MAPPER = TestHelper.makeJsonMapper();
@@ -508,4 +524,286 @@ public class OnheapIncrementalIndexTest
t.getMessage()
);
}
+
+ @Test
+ public void testExpressionSelectorCacheUsesExpressionIdentity()
+ {
+ final OnheapIncrementalIndex.CachingColumnSelectorFactory selectorFactory
= makeCachingColumnSelectorFactory();
+ final Expr expression = Parser.parse("value", TestExprMacroTable.INSTANCE);
+ final Expr structurallyEqualExpression = Parser.parse("value",
TestExprMacroTable.INSTANCE);
+ final Expr differentExpression = Parser.parse("other",
TestExprMacroTable.INSTANCE);
+
+ Assertions.assertNotSame(expression, structurallyEqualExpression);
+
+ final ColumnValueSelector<ExprEval> firstSelector =
selectorFactory.getOrCreateExprEvalSelector(expression);
+ Assertions.assertSame(firstSelector,
selectorFactory.getOrCreateExprEvalSelector(expression));
+
+ final ColumnValueSelector<ExprEval> structurallyEqualSelector =
+
selectorFactory.getOrCreateExprEvalSelector(structurallyEqualExpression);
+ Assertions.assertNotSame(firstSelector, structurallyEqualSelector);
+ Assertions.assertSame(
+ structurallyEqualSelector,
+
selectorFactory.getOrCreateExprEvalSelector(structurallyEqualExpression)
+ );
+
+ final ColumnValueSelector<ExprEval> differentSelector =
+ selectorFactory.getOrCreateExprEvalSelector(differentExpression);
+ Assertions.assertNotSame(structurallyEqualSelector, differentSelector);
+ Assertions.assertSame(differentSelector,
selectorFactory.getOrCreateExprEvalSelector(differentExpression));
+ }
+
+ @Test
+ public void testExpressionSelectorCacheReusesSelector()
+ {
+ final OnheapIncrementalIndex.CachingColumnSelectorFactory selectorFactory
= makeCachingColumnSelectorFactory();
+ final Expr expression = Parser.parse("value + 1",
TestExprMacroTable.INSTANCE);
+
+ final ColumnValueSelector<?> firstSelector =
ExpressionSelectors.makeExprEvalSelector(selectorFactory, expression);
+ final ColumnValueSelector<?> secondSelector =
ExpressionSelectors.makeExprEvalSelector(selectorFactory, expression);
+ Assertions.assertSame(firstSelector, secondSelector);
+ }
+
+ @Test
+ public void testExpressionSelectorCacheReusesConstantSelector()
+ {
+ final OnheapIncrementalIndex.CachingColumnSelectorFactory selectorFactory
= makeCachingColumnSelectorFactory();
+ final Expr constantExpression = Parser.parse("'constant'",
TestExprMacroTable.INSTANCE);
+ final ColumnValueSelector<ExprEval> firstConstantSelector =
+ ExpressionSelectors.makeExprEvalSelector(selectorFactory,
constantExpression);
+ final ColumnValueSelector<ExprEval> secondConstantSelector =
+ ExpressionSelectors.makeExprEvalSelector(selectorFactory,
constantExpression);
+ Assertions.assertSame(firstConstantSelector, secondConstantSelector);
+ }
+
+ @Test
+ public void testExpressionSelectorCacheDoesNotShareAggregatorState()
+ {
+ final OnheapIncrementalIndex.CachingColumnSelectorFactory selectorFactory
= makeCachingColumnSelectorFactory();
+ final AggregatorFactory aggregatorFactory = new LongSumAggregatorFactory(
+ "sum",
+ null,
+ "value + 1",
+ TestExprMacroTable.INSTANCE
+ );
+ final Aggregator firstAggregator =
aggregatorFactory.factorize(selectorFactory);
+ final Aggregator secondAggregator =
aggregatorFactory.factorize(selectorFactory);
+
+ Assertions.assertNotSame(firstAggregator, secondAggregator);
+ firstAggregator.aggregate();
+ Assertions.assertEquals(2L, firstAggregator.getLong());
+ Assertions.assertTrue(secondAggregator.isNull());
+ }
+
+ @Test
+ public void testExpressionIngestionProducesExpectedResults()
+ {
+ final int rowCount = 10_000;
+ final int rollupKeyCount = 100;
+ final List<InputRow> rows = new ArrayList<>(rowCount);
+ long expectedSum = 0L;
+ for (int rowNumber = 0; rowNumber < rowCount; rowNumber++) {
+ final long value = rowNumber % 100;
+ rows.add(
+ new MapBasedInputRow(
+ 0L,
+ Collections.singletonList("key"),
+ ImmutableMap.of(
+ "key", "key-" + rowNumber % rollupKeyCount,
+ "value", value
+ )
+ )
+ );
+ // Each row contributes the result of the ingestion expression "value +
1".
+ expectedSum += value + 1;
+ }
+
+ try (
+ OnheapIncrementalIndex index = (OnheapIncrementalIndex) new
OnheapIncrementalIndex.Builder()
+ .setIndexSchema(
+ new IncrementalIndexSchema.Builder()
+ .withDimensionsSpec(new
DimensionsSpec(Collections.singletonList(new StringDimensionSchema("key"))))
+ .withMetrics(
+ new LongSumAggregatorFactory(
+ "sum",
+ null,
+ "value + 1",
+ TestExprMacroTable.INSTANCE
+ )
+ )
+ .withRollup(true)
+ .build()
+ )
+ .setMaxRowCount(rowCount + 1)
+ .build()
+ ) {
+ for (final InputRow row : rows) {
+ index.add(row);
+ }
+
+ long actualSum = 0L;
+ for (final IncrementalIndexRow row : index.getFacts().keySet()) {
+ actualSum += index.getMetricLongValue(row.getRowIndex(), 0);
+ }
+
+ Assertions.assertEquals(rollupKeyCount, index.numRows());
+ Assertions.assertEquals(expectedSum, actualSum);
+ }
+ }
+
+ @Test
+ public void
testExpressionIngestionWithChangingArrayBindingsAcrossRollupKeys()
+ {
+ final List<InputRow> rows = List.of(
+ new MapBasedInputRow(
+ 0L,
+ Collections.singletonList("key"),
+ ImmutableMap.of("key", "key-0",
+ "Aa", List.of("a1", "a2"),
+ "BB", "b")
+ ),
+ new MapBasedInputRow(
+ 0L,
+ Collections.singletonList("key"),
+ ImmutableMap.of("key", "key-1",
+ "Aa", "a",
+ "BB", List.of("b1", "b2"))
+ ),
+ new MapBasedInputRow(
+ 0L,
+ Collections.singletonList("key"),
+ ImmutableMap.of("key", "key-2",
+ "Aa", List.of("x"),
+ "BB", "yz")
+ ),
+ new MapBasedInputRow(
+ 0L,
+ Collections.singletonList("key"),
+ ImmutableMap.of("key", "key-3",
+ "Aa", "xy",
+ "BB", List.of("z"))
+ )
+ );
+
+ try (
+ OnheapIncrementalIndex index = (OnheapIncrementalIndex) new
OnheapIncrementalIndex.Builder()
+ .setIndexSchema(
+ new IncrementalIndexSchema.Builder()
+ .withDimensionsSpec(new
DimensionsSpec(Collections.singletonList(new StringDimensionSchema("key"))))
+ .withMetrics(
+ new LongSumAggregatorFactory(
+ "sum",
+ null,
+ "strlen(array_to_string(concat(\"Aa\", \"BB\"),
','))",
+ TestExprMacroTable.INSTANCE
+ )
+ )
+ .withRollup(true)
+ .build()
+ )
+ .setMaxRowCount(rows.size() + 1)
+ .build()
+ ) {
+ for (final InputRow row : rows) {
+ index.add(row);
+ }
+
+ long actualSum = 0L;
+ for (final IncrementalIndexRow row : index.getFacts().keySet()) {
+ actualSum += index.getMetricLongValue(row.getRowIndex(), 0);
+ }
+
+ Assertions.assertEquals(rows.size(), index.numRows());
+
+ // Per-row expression results are 7, 7, 3, and 3, for a total of 20.
+ // For example: Row 1: Aa = ["a1", "a2"], BB = "b", concat("Aa", "BB") =
["a1b", "a2b"], array_to_string = "a1b,a2b", length is 7
+ Assertions.assertEquals(20L, actualSum);
+ }
+ }
+
+ @Test
+ public void
testExpressionSelectorCacheReevaluatesNonDeterministicExpressionAcrossRollupKeys()
+ {
+ // MapBasedInputRow exposes its timestamp as __time. Use the same value
for both rollup keys.
+ final long inputTimestamp = 0L;
+ try (
+ OnheapIncrementalIndex index = (OnheapIncrementalIndex) new
OnheapIncrementalIndex.Builder()
+ .setIndexSchema(
+ new IncrementalIndexSchema.Builder()
+ .withDimensionsSpec(new
DimensionsSpec(Collections.singletonList(new StringDimensionSchema("key"))))
+ .withMetrics(
+ new LongSumAggregatorFactory(
+ "expressionResult",
+ null,
+ "now() - __time",
+ ExprMacroTable.nil()
+ )
+ )
+ .withRollup(true)
+ .build()
+ )
+ .setMaxRowCount(3)
+ .build()
+ ) {
+ index.add(
+ new MapBasedInputRow(
+ inputTimestamp,
+ Collections.singletonList("key"),
+ ImmutableMap.of("key", "key-0")
+ )
+ );
+
+ // Since __time is 0, wait until a new now() evaluation must be greater
than the first result.
+ final IncrementalIndexRow firstIndexRow =
index.getFacts().keySet().iterator().next();
+ final long firstExpressionResult =
index.getMetricLongValue(firstIndexRow.getRowIndex(), 0);
+ final long waitDeadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(1);
+ while (System.currentTimeMillis() <= firstExpressionResult) {
+ Assertions.assertTrue(System.nanoTime() < waitDeadline, "Timed out
waiting for the clock to advance");
+ Thread.onSpinWait();
+ }
+
+ index.add(
+ new MapBasedInputRow(
+ inputTimestamp,
+ Collections.singletonList("key"),
+ ImmutableMap.of("key", "key-1")
+ )
+ );
+
+ final List<Long> expressionResults = new ArrayList<>();
+ for (final IncrementalIndexRow row : index.getFacts().keySet()) {
+ expressionResults.add(index.getMetricLongValue(row.getRowIndex(), 0));
+ }
+
+ Assertions.assertEquals(2, expressionResults.size());
+ Assertions.assertTrue(
+ expressionResults.stream().anyMatch(expressionResult ->
expressionResult > firstExpressionResult),
+ "The non-deterministic expression should be reevaluated for the
second rollup key"
+ );
+ }
+ }
+
+ private static OnheapIncrementalIndex.CachingColumnSelectorFactory
makeCachingColumnSelectorFactory()
+ {
+ final TestObjectColumnSelector<Long> valueSelector = new
TestObjectColumnSelector<Long>()
+ {
+ @Override
+ public Class<Long> classOfObject()
+ {
+ return Long.class;
+ }
+
+ @Override
+ public Long getObject()
+ {
+ return 1L;
+ }
+ };
+
+ final ColumnSelectorFactory delegate = new TestColumnSelectorFactory()
+ .addColumnSelector("value", valueSelector)
+ .addCapabilities("value", null)
+ .addColumnSelector("other", valueSelector)
+ .addCapabilities("other", null);
+ return new OnheapIncrementalIndex.CachingColumnSelectorFactory(delegate);
+ }
}
diff --git
a/processing/src/test/java/org/apache/druid/segment/virtual/ExpressionSelectorsTest.java
b/processing/src/test/java/org/apache/druid/segment/virtual/ExpressionSelectorsTest.java
index 05f6c68122b..63e26ceaaff 100644
---
a/processing/src/test/java/org/apache/druid/segment/virtual/ExpressionSelectorsTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/virtual/ExpressionSelectorsTest.java
@@ -303,6 +303,49 @@ public class ExpressionSelectorsTest extends
InitializedNullHandlingTest
}
}
+ @Test
+ public void
test_row_based_selector_distinguishes_colliding_array_binding_sets() throws
IOException
+ {
+ // These distinct binding lists collide when keyed only by their hash
codes.
+ Assertions.assertEquals(List.of("Aa").hashCode(),
List.of("BB").hashCode());
+
+ final DateTime timestamp = DateTimes.nowUtc();
+ final List<InputRow> rows = List.of(
+ new MapBasedInputRow(
+ timestamp,
+ List.of("Aa", "BB"),
+ ImmutableMap.of("Aa", List.of("a1", "a2"), "BB", "b")
+ ),
+ new MapBasedInputRow(
+ timestamp.plusMinutes(1),
+ List.of("Aa", "BB"),
+ ImmutableMap.of("Aa", "a", "BB", List.of("b1", "b2"))
+ )
+ );
+
+ try (final Segment segment = new RowBasedSegment<>(
+ Sequences.simple(rows),
+ RowAdapters.standardRow(),
+ RowSignature.empty()
+ )) {
+ final CursorFactory cursorFactory = segment.as(CursorFactory.class);
+ Assertions.assertNotNull(cursorFactory);
+
+ try (final CursorHolder cursorHolder =
cursorFactory.makeCursorHolder(CursorBuildSpec.FULL_SCAN)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ final ColumnValueSelector<ExprEval> selector =
ExpressionSelectors.makeExprEvalSelector(
+ cursor.getColumnSelectorFactory(),
+ Parser.parse("concat(\"Aa\", \"BB\")", TestExprMacroTable.INSTANCE)
+ );
+
+ Assertions.assertArrayEquals(new Object[]{"a1b", "a2b"},
selector.getObject().asArray());
+
+ cursor.advance();
+ Assertions.assertArrayEquals(new Object[]{"ab1", "ab2"},
selector.getObject().asArray());
+ }
+ }
+ }
+
@Test
public void test_long_bindings()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
