xiangfu0 commented on code in PR #19539:
URL: https://github.com/apache/pinot/pull/19539#discussion_r4067648944
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/partition/PartitionFunction.java:
##########
@@ -63,6 +63,26 @@ default Map<String, String> getFunctionConfig() {
return null;
}
+ /// Returns whether the exposed function configuration is null or empty.
This does not imply that other settings,
+ /// such as the partition id normalizer, have their default values.
+ @JsonIgnore
+ default boolean hasEmptyConfig() {
+ Map<String, String> config = getFunctionConfig();
+ return config == null || config.isEmpty();
+ }
+
+ /// Returns whether partition ids computed by this function can be reused
for the non-null `other` function.
+ /// A true result must guarantee identical results for every input value.
False is conservative, not proof that the
+ /// functions differ. Implementations with additional output-affecting state
must account for it in this method.
+ ///
+ /// The default only permits matching functions with empty exposed
configurations, without comparing config contents.
+ /// Configured implementations may override this method to compare their
effective settings.
+ default boolean canReusePartitionIds(PartitionFunction other) {
+ return hasEmptyConfig() && other.hasEmptyConfig() && getClass() ==
other.getClass()
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): The default now
states its assumptions explicitly. For this change we are intentionally
retaining the agreed Apache/StarTree implementation scope: the audited OSS
functions expose output-affecting configuration, and StarTree Custom overrides
compatibility using its effective settings. Config-map equality would not
protect hidden state when both maps are null, and it reintroduces the
independently deserialized 100 KB comparison cost discussed above. An opt-in
default with built-in overrides is the safer choice if arbitrary plugins must
be supported, but broadens this agreed implementation; leaving that policy
decision open for maintainer/user direction.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/partition/PartitionFunction.java:
##########
@@ -63,6 +63,26 @@ default Map<String, String> getFunctionConfig() {
return null;
}
+ /// Returns whether the exposed function configuration is null or empty.
This does not imply that other settings,
+ /// such as the partition id normalizer, have their default values.
+ @JsonIgnore
+ default boolean hasEmptyConfig() {
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): Kept
hasEmptyConfig public as requested for function implementations, and documented
that it is a reusable helper for canReusePartitionIds overrides. The wording
continues to distinguish an empty exposed map from a default normalizer, and
JsonIgnore remains.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/SinglePartitionColumnSegmentPruner.java:
##########
@@ -94,6 +98,13 @@ public Set<String> prune(BrokerRequest brokerRequest,
Set<String> segments) {
if (filterExpression == null) {
return segments;
}
+ int numSegments = segments.size();
+ if (numSegments == 0) {
+ return segments;
+ }
+ if (numSegments >= MIN_SEGMENTS_FOR_PREPARATION) {
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): Added the
query-level escape hatch enablePartitionPruningCache=false, which uses the
original evaluator while retaining partition pruning. Default/true still honors
the 256-candidate guard, so this is not a force-on switch for small inputs.
Shared predicate, invalid-metadata, refresh, and large-config fixtures now run
at 255/256 and with the cache disabled; the call-count test proves the switch
restores 256 evaluations rather than one. The separate single-evaluator design
discussion remains open above, and the mixed-case benchmark caveat remains in
the description.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/SinglePartitionColumnSegmentPruner.java:
##########
@@ -105,6 +116,30 @@ public Set<String> prune(BrokerRequest brokerRequest,
Set<String> segments) {
return selectedSegments;
}
+ private Set<String> pruneWithPreparedPredicate(Expression filterExpression,
Set<String> segments) {
+ Set<String> selectedSegments = new HashSet<>();
+ PreparedPredicate predicate = null;
+ PartitionFunction cachedFunction = null;
+ for (String segment : segments) {
+ SegmentPartitionInfo partitionInfo = _partitionInfoMap.get(segment);
+ if (partitionInfo == null || partitionInfo ==
SegmentPartitionUtils.INVALID_PARTITION_INFO) {
+ selectedSegments.add(segment);
+ continue;
+ }
+ PartitionFunction function = partitionInfo.getPartitionFunction();
+ if (predicate == null) {
+ predicate = new PreparedPredicate(filterExpression);
+ cachedFunction = function;
+ }
+ // Reuse the filter structure even when the functions cannot share
partition ids.
+ boolean reusePartitionIds =
cachedFunction.canReusePartitionIds(function);
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): Added the
same-instance fast path in PartitionFunction.canReusePartitionIds, so
configured self-comparisons are now true. Distinct configured instances still
conservatively fall back unless an implementation such as Custom opts in. The
single pivot remains intentional: a first-segment outlier can reduce the
optimization's benefit, but does not change results. Re-pivoting or a
per-function map is deferred to preserve the requested minimal query-local
design; that limitation is documented.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/SinglePartitionColumnSegmentPruner.java:
##########
@@ -105,6 +116,30 @@ public Set<String> prune(BrokerRequest brokerRequest,
Set<String> segments) {
return selectedSegments;
}
+ private Set<String> pruneWithPreparedPredicate(Expression filterExpression,
Set<String> segments) {
+ Set<String> selectedSegments = new HashSet<>();
+ PreparedPredicate predicate = null;
+ PartitionFunction cachedFunction = null;
+ for (String segment : segments) {
+ SegmentPartitionInfo partitionInfo = _partitionInfoMap.get(segment);
+ if (partitionInfo == null || partitionInfo ==
SegmentPartitionUtils.INVALID_PARTITION_INFO) {
+ selectedSegments.add(segment);
+ continue;
+ }
+ PartitionFunction function = partitionInfo.getPartitionFunction();
+ if (predicate == null) {
+ predicate = new PreparedPredicate(filterExpression);
+ cachedFunction = function;
+ }
+ // Reuse the filter structure even when the functions cannot share
partition ids.
+ boolean reusePartitionIds =
cachedFunction.canReusePartitionIds(function);
+ if (predicate.matches(partitionInfo.getPartitions(), function,
reusePartitionIds)) {
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): Deferring this
extra optimization to keep the change small. _partitionIds == null is not by
itself proof of constant true: children may be unvisited due to
short-circuiting, and an empty OR returns false despite having no partition
leaves. Establishing constant truth for the whole expression needs more
state/analysis while preserving lazy exceptions. The unrelated-column
regression now exercises both evaluators at the cutoff; the current
conservative loop remains.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/SinglePartitionColumnSegmentPruner.java:
##########
@@ -152,4 +187,82 @@ private boolean isPartitionMatch(Expression
filterExpression, SegmentPartitionIn
return true;
}
}
+
+ /// Lazily prepares only visited expressions and values. Instances belong to
one prune call, never shared by queries.
+ private final class PreparedPredicate {
+ private final Expression _expression;
+ private FilterKind _kind;
+ private List<Expression> _operands;
+ private PreparedPredicate[] _children;
+ private List<Integer> _partitionIds;
+
+ private PreparedPredicate(Expression expression) {
+ _expression = expression;
+ }
+
+ private boolean matches(Set<Integer> partitions, PartitionFunction
partitionFunction, boolean reusePartitionIds) {
+ if (_kind == null) {
+ Function function = _expression.getFunctionCall();
+ _kind = FilterKind.valueOf(function.getOperator());
+ _operands = function.getOperands();
+ if (_kind == FilterKind.AND || _kind == FilterKind.OR) {
+ _children = new PreparedPredicate[_operands.size()];
+ for (int i = 0; i < _children.length; i++) {
+ _children[i] = new PreparedPredicate(_operands.get(i));
+ }
+ } else if (_kind == FilterKind.EQUALS || _kind == FilterKind.IN) {
+ Identifier identifier = _operands.get(0).getIdentifier();
+ if (identifier != null &&
identifier.getName().equals(_partitionColumn)) {
+ _partitionIds = new ArrayList<>(1);
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): Kept incremental
capacity deliberately and added the rationale beside the allocation. The number
of visited literals is not known in advance: an IN with 4,096 values may match
its first value on every segment. Allocating for all operands would add
proportional per-query allocation even in that case. Boxed cached IDs still
avoid re-boxing per segment. A different growth policy should be measured
across both full-scan and early-match workloads rather than assuming full
preallocation is always cheaper.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/SinglePartitionColumnSegmentPruner.java:
##########
@@ -152,4 +187,82 @@ private boolean isPartitionMatch(Expression
filterExpression, SegmentPartitionIn
return true;
}
}
+
+ /// Lazily prepares only visited expressions and values. Instances belong to
one prune call, never shared by queries.
+ private final class PreparedPredicate {
+ private final Expression _expression;
+ private FilterKind _kind;
+ private List<Expression> _operands;
+ private PreparedPredicate[] _children;
+ private List<Integer> _partitionIds;
+
+ private PreparedPredicate(Expression expression) {
+ _expression = expression;
+ }
+
+ private boolean matches(Set<Integer> partitions, PartitionFunction
partitionFunction, boolean reusePartitionIds) {
+ if (_kind == null) {
+ Function function = _expression.getFunctionCall();
+ _kind = FilterKind.valueOf(function.getOperator());
+ _operands = function.getOperands();
+ if (_kind == FilterKind.AND || _kind == FilterKind.OR) {
+ _children = new PreparedPredicate[_operands.size()];
+ for (int i = 0; i < _children.length; i++) {
+ _children[i] = new PreparedPredicate(_operands.get(i));
+ }
+ } else if (_kind == FilterKind.EQUALS || _kind == FilterKind.IN) {
+ Identifier identifier = _operands.get(0).getIdentifier();
+ if (identifier != null &&
identifier.getName().equals(_partitionColumn)) {
+ _partitionIds = new ArrayList<>(1);
+ }
+ }
+ }
+ switch (_kind) {
+ case AND:
+ for (PreparedPredicate child : _children) {
+ if (!child.matches(partitions, partitionFunction,
reusePartitionIds)) {
+ return false;
+ }
+ }
+ return true;
+ case OR:
+ for (PreparedPredicate child : _children) {
+ if (child.matches(partitions, partitionFunction,
reusePartitionIds)) {
+ return true;
+ }
+ }
+ return false;
+ case EQUALS:
+ case IN:
+ if (_partitionIds != null) {
+ int numValues = _kind == FilterKind.EQUALS ? 1 : _operands.size()
- 1;
+ if (!reusePartitionIds) {
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): Thanks for
verifying the visited-prefix invariant. It is preserved, including the
incompatible-function fallback and lazy short-circuit behavior.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/SinglePartitionColumnSegmentPruner.java:
##########
@@ -94,6 +98,13 @@ public Set<String> prune(BrokerRequest brokerRequest,
Set<String> segments) {
if (filterExpression == null) {
return segments;
}
+ int numSegments = segments.size();
+ if (numSegments == 0) {
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): Removed the
explicit empty-candidate branch. An empty input falls through the original
loop, returns a new empty set, and never evaluates the filter. Existing
empty-input tests remain.
##########
pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpruner/SinglePartitionColumnSegmentPrunerTest.java:
##########
@@ -0,0 +1,393 @@
+/**
+ * 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.pinot.broker.routing.segmentpruner;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.annotation.Nullable;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
+import org.apache.pinot.common.request.BrokerRequest;
+import org.apache.pinot.common.request.Expression;
+import org.apache.pinot.common.request.PinotQuery;
+import org.apache.pinot.common.utils.request.RequestUtils;
+import org.apache.pinot.segment.spi.partition.PartitionFunction;
+import org.apache.pinot.segment.spi.partition.PartitionFunctionFactory;
+import org.apache.pinot.segment.spi.partition.PartitionIdNormalizer;
+import org.apache.pinot.segment.spi.partition.metadata.ColumnPartitionMetadata;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.expectThrows;
+
+
+/// Exercises query-local partition ID caching with real metadata
initialization and refresh, without ZooKeeper.
+public class SinglePartitionColumnSegmentPrunerTest {
+ private static final String COLUMN = "memberId";
+ private static final String TABLE = "testTable_OFFLINE";
+
+ @Test
+ public void testHashesOnceAcrossDistinctMetadataInstancesPerQuery() throws
Exception {
+ Map<String, ZNRecord> records = new LinkedHashMap<>();
+ Set<String> expected = new HashSet<>();
+ for (int i = 0; i < 256; i++) {
+ String segment = "segment_" + i;
+ records.put(segment, metadata(segment, "PrunerCounting", 8, Set.of(i %
8), i % 2 == 0 ? null : Map.of()));
+ if (i % 8 == 3) {
+ expected.add(segment);
+ }
+ }
+ SinglePartitionColumnSegmentPruner pruner = pruner(records);
+ CountingPartitionFunction.CALLS.set(0);
+ BrokerRequest request = request(predicate("EQUALS", "3"));
+ assertEquals(pruner.prune(request, records.keySet()), expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 1);
+ assertEquals(pruner.prune(request, records.keySet()), expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 2, "Computed hashes
must not survive a prune call");
+ }
+
+ @Test
+ public void testInterleavedFunctionConfigurationsAndPartitionCounts() throws
Exception {
+ Map<String, ZNRecord> records = new LinkedHashMap<>();
+ records.put("a", metadata("a", "PrunerCounting", 8, Set.of(3), null));
+ records.put("sameFunction", metadata("sameFunction", "PrunerCounting", 8,
Set.of(2), null));
+ records.put("b", metadata("b", "PrunerCounting", 8, Set.of(4),
Map.of("offset", "1")));
+ records.put("c", metadata("c", "PrunerCounting", 8, Set.of(2), null));
+ records.put("d", metadata("d", "PrunerCounting", 8, Set.of(3),
Map.of("offset", "1")));
+ records.put("e", metadata("e", "PrunerCounting", 16, Set.of(11), null));
+ Set<String> expected = new HashSet<>(Set.of("a", "b", "e"));
+ for (int i = 0; i < 250; i++) {
+ String segment = "tail_" + i;
+ records.put(segment, metadata(segment, "PrunerCounting", 8, Set.of(3),
null));
+ expected.add(segment);
+ }
+ SinglePartitionColumnSegmentPruner pruner = pruner(records);
+ CountingPartitionFunction.CALLS.set(0);
+ assertEquals(pruner.prune(request(predicate("EQUALS", "11")),
records.keySet()), expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 4,
+ "Only compatible default functions reuse IDs; configured functions
never compare configuration contents");
+ records.put("b", metadata("b", "PrunerCounting", 16, Set.of(11), null));
+ records.put("d", metadata("d", "PrunerCounting", 8, Set.of(3), null));
+ expected.add("d");
+ CountingPartitionFunction.CALLS.set(0);
+ assertEquals(pruner(records).prune(request(predicate("EQUALS", "11")),
records.keySet()), expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 3,
+ "Different partition counts must not reuse partition IDs");
+ }
+
+ @Test
+ public void testDuplicateInPartitionsAndIncrementalEvaluation() throws
Exception {
+ Map<String, ZNRecord> records = new LinkedHashMap<>();
+ records.put("first", metadata("first", "PrunerCounting", 8, Set.of(1),
null));
+ records.put("configured", metadata("configured", "PrunerCounting", 8,
Set.of(3), Map.of("offset", "1")));
+ records.put("second", metadata("second", "PrunerCounting", 8, Set.of(2),
null));
+ records.put("miss", metadata("miss", "PrunerCounting", 8, Set.of(3, 4),
null));
+ records.put("repeat", metadata("repeat", "PrunerCounting", 8, Set.of(1,
2), null));
+ Set<String> expected = new HashSet<>(Set.of("first", "configured",
"second", "repeat"));
+ for (int i = 0; i < 252; i++) {
+ String segment = "repeat_" + i;
+ records.put(segment, metadata(segment, "PrunerCounting", 8, Set.of(1,
2), null));
+ expected.add(segment);
+ }
+ SinglePartitionColumnSegmentPruner pruner = pruner(records);
+ CountingPartitionFunction.CALLS.set(0);
+ // The configured segment must not consume or extend the prefix cached by
the first segment.
+ assertEquals(pruner.prune(request(predicate("IN", "1", "9", "17", "2")),
records.keySet()),
+ expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 8);
+ // A configured first segment must not seed IDs for later default-config
segments.
+ records.put("first", metadata("first", "PrunerCounting", 8, Set.of(2),
Map.of("offset", "1")));
+ assertEquals(pruner(records).prune(request(predicate("IN", "1", "9", "17",
"2")), records.keySet()), expected);
+ }
+
+ @Test
+ public void testLargeConfigurationsAndUnrelatedPredicates() throws Exception
{
+ String values = "first|" + "x".repeat(100_000);
+ Map<String, String> config = Map.of("columnValues", values,
"columnValuesDelimiter", "|");
+ ZNRecord first = metadata("first", "BoundedColumnValue", 3, Set.of(1),
config);
+ ZNRecord second = metadata("second", "BoundedColumnValue", 3, Set.of(2),
config);
+ SinglePartitionColumnSegmentPruner pruner = pruner(Map.of("first", first,
"second", second));
+ assertEquals(pruner.prune(request(predicate("EQUALS", "first")),
Set.of("first", "second")), Set.of("first"));
+ assertEquals(pruner.prune(request(function("EQUALS",
RequestUtils.getIdentifierExpression("other"),
+ RequestUtils.getLiteralExpression("value"))), Set.of("first",
"second")), Set.of("first", "second"));
+ pruner.refreshSegment("first", metadata("first", "BoundedColumnValue", 3,
Set.of(2), config));
+ assertEquals(pruner.prune(request(predicate("EQUALS", "first")),
Set.of("first", "second")), Set.of());
+ }
+
+ @Test
+ public void testConfigurationHashCollisionsDoNotReusePartitionIds() throws
Exception {
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): Removed the
obsolete configuration-hash-collision test. The remaining configured-function,
mixed-metadata, and expanded large-config fixtures cover the current
compatibility/fallback design without referring to the removed key-hashing
implementation.
##########
pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpruner/SinglePartitionColumnSegmentPrunerTest.java:
##########
@@ -0,0 +1,393 @@
+/**
+ * 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.pinot.broker.routing.segmentpruner;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.annotation.Nullable;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
+import org.apache.pinot.common.request.BrokerRequest;
+import org.apache.pinot.common.request.Expression;
+import org.apache.pinot.common.request.PinotQuery;
+import org.apache.pinot.common.utils.request.RequestUtils;
+import org.apache.pinot.segment.spi.partition.PartitionFunction;
+import org.apache.pinot.segment.spi.partition.PartitionFunctionFactory;
+import org.apache.pinot.segment.spi.partition.PartitionIdNormalizer;
+import org.apache.pinot.segment.spi.partition.metadata.ColumnPartitionMetadata;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.expectThrows;
+
+
+/// Exercises query-local partition ID caching with real metadata
initialization and refresh, without ZooKeeper.
+public class SinglePartitionColumnSegmentPrunerTest {
+ private static final String COLUMN = "memberId";
+ private static final String TABLE = "testTable_OFFLINE";
+
+ @Test
+ public void testHashesOnceAcrossDistinctMetadataInstancesPerQuery() throws
Exception {
+ Map<String, ZNRecord> records = new LinkedHashMap<>();
+ Set<String> expected = new HashSet<>();
+ for (int i = 0; i < 256; i++) {
+ String segment = "segment_" + i;
+ records.put(segment, metadata(segment, "PrunerCounting", 8, Set.of(i %
8), i % 2 == 0 ? null : Map.of()));
+ if (i % 8 == 3) {
+ expected.add(segment);
+ }
+ }
+ SinglePartitionColumnSegmentPruner pruner = pruner(records);
+ CountingPartitionFunction.CALLS.set(0);
+ BrokerRequest request = request(predicate("EQUALS", "3"));
+ assertEquals(pruner.prune(request, records.keySet()), expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 1);
+ assertEquals(pruner.prune(request, records.keySet()), expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 2, "Computed hashes
must not survive a prune call");
+ }
+
+ @Test
+ public void testInterleavedFunctionConfigurationsAndPartitionCounts() throws
Exception {
+ Map<String, ZNRecord> records = new LinkedHashMap<>();
+ records.put("a", metadata("a", "PrunerCounting", 8, Set.of(3), null));
+ records.put("sameFunction", metadata("sameFunction", "PrunerCounting", 8,
Set.of(2), null));
+ records.put("b", metadata("b", "PrunerCounting", 8, Set.of(4),
Map.of("offset", "1")));
+ records.put("c", metadata("c", "PrunerCounting", 8, Set.of(2), null));
+ records.put("d", metadata("d", "PrunerCounting", 8, Set.of(3),
Map.of("offset", "1")));
+ records.put("e", metadata("e", "PrunerCounting", 16, Set.of(11), null));
+ Set<String> expected = new HashSet<>(Set.of("a", "b", "e"));
+ for (int i = 0; i < 250; i++) {
+ String segment = "tail_" + i;
+ records.put(segment, metadata(segment, "PrunerCounting", 8, Set.of(3),
null));
+ expected.add(segment);
+ }
+ SinglePartitionColumnSegmentPruner pruner = pruner(records);
+ CountingPartitionFunction.CALLS.set(0);
+ assertEquals(pruner.prune(request(predicate("EQUALS", "11")),
records.keySet()), expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 4,
+ "Only compatible default functions reuse IDs; configured functions
never compare configuration contents");
+ records.put("b", metadata("b", "PrunerCounting", 16, Set.of(11), null));
+ records.put("d", metadata("d", "PrunerCounting", 8, Set.of(3), null));
+ expected.add("d");
+ CountingPartitionFunction.CALLS.set(0);
+ assertEquals(pruner(records).prune(request(predicate("EQUALS", "11")),
records.keySet()), expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 3,
+ "Different partition counts must not reuse partition IDs");
+ }
+
+ @Test
+ public void testDuplicateInPartitionsAndIncrementalEvaluation() throws
Exception {
+ Map<String, ZNRecord> records = new LinkedHashMap<>();
+ records.put("first", metadata("first", "PrunerCounting", 8, Set.of(1),
null));
+ records.put("configured", metadata("configured", "PrunerCounting", 8,
Set.of(3), Map.of("offset", "1")));
+ records.put("second", metadata("second", "PrunerCounting", 8, Set.of(2),
null));
+ records.put("miss", metadata("miss", "PrunerCounting", 8, Set.of(3, 4),
null));
+ records.put("repeat", metadata("repeat", "PrunerCounting", 8, Set.of(1,
2), null));
+ Set<String> expected = new HashSet<>(Set.of("first", "configured",
"second", "repeat"));
+ for (int i = 0; i < 252; i++) {
+ String segment = "repeat_" + i;
+ records.put(segment, metadata(segment, "PrunerCounting", 8, Set.of(1,
2), null));
+ expected.add(segment);
+ }
+ SinglePartitionColumnSegmentPruner pruner = pruner(records);
+ CountingPartitionFunction.CALLS.set(0);
+ // The configured segment must not consume or extend the prefix cached by
the first segment.
+ assertEquals(pruner.prune(request(predicate("IN", "1", "9", "17", "2")),
records.keySet()),
+ expected);
+ assertEquals(CountingPartitionFunction.CALLS.get(), 8);
+ // A configured first segment must not seed IDs for later default-config
segments.
+ records.put("first", metadata("first", "PrunerCounting", 8, Set.of(2),
Map.of("offset", "1")));
+ assertEquals(pruner(records).prune(request(predicate("IN", "1", "9", "17",
"2")), records.keySet()), expected);
+ }
+
+ @Test
+ public void testLargeConfigurationsAndUnrelatedPredicates() throws Exception
{
+ String values = "first|" + "x".repeat(100_000);
+ Map<String, String> config = Map.of("columnValues", values,
"columnValuesDelimiter", "|");
+ ZNRecord first = metadata("first", "BoundedColumnValue", 3, Set.of(1),
config);
+ ZNRecord second = metadata("second", "BoundedColumnValue", 3, Set.of(2),
config);
+ SinglePartitionColumnSegmentPruner pruner = pruner(Map.of("first", first,
"second", second));
+ assertEquals(pruner.prune(request(predicate("EQUALS", "first")),
Set.of("first", "second")), Set.of("first"));
+ assertEquals(pruner.prune(request(function("EQUALS",
RequestUtils.getIdentifierExpression("other"),
+ RequestUtils.getLiteralExpression("value"))), Set.of("first",
"second")), Set.of("first", "second"));
+ pruner.refreshSegment("first", metadata("first", "BoundedColumnValue", 3,
Set.of(2), config));
+ assertEquals(pruner.prune(request(predicate("EQUALS", "first")),
Set.of("first", "second")), Set.of());
+ }
+
+ @Test
+ public void testConfigurationHashCollisionsDoNotReusePartitionIds() throws
Exception {
+ Map<String, String> firstConfig = Map.of("columnValues", "Aa|BB",
"columnValuesDelimiter", "|");
+ Map<String, String> secondConfig = Map.of("columnValues", "BB|Aa",
"columnValuesDelimiter", "|");
+ assertEquals(firstConfig.hashCode(), secondConfig.hashCode(),
+ "Fixture must exercise a configuration hash collision");
+ ZNRecord first = metadata("first", "BoundedColumnValue", 3, Set.of(1),
firstConfig);
+ ZNRecord second = metadata("second", "BoundedColumnValue", 3, Set.of(2),
secondConfig);
+ Map<String, ZNRecord> records = Map.of("first", first, "second", second);
+ assertEquals(pruner(records).prune(request(predicate("EQUALS", "Aa")),
records.keySet()), records.keySet());
+ }
+
+ @Test
+ public void testMixedFunctionsNormalizersAndFunctionConfig() throws
Exception {
+ Map<String, ZNRecord> moduloRecords = new LinkedHashMap<>();
+ moduloRecords.put("positive", metadata("positive", "Modulo", 8, Set.of(7),
null));
+ moduloRecords.put("abs", metadata("abs", "Modulo", 8, Set.of(1),
Map.of("partitionIdNormalizer", "ABS")));
+ moduloRecords.put("wrongAbs", metadata("wrongAbs", "Modulo", 8, Set.of(7),
Map.of("partitionIdNormalizer", "ABS")));
+ Set<String> expected = new HashSet<>(Set.of("positive", "abs"));
+ for (int i = 0; i < 253; i++) {
+ String segment = "positive_" + i;
+ moduloRecords.put(segment, metadata(segment, "Modulo", 8, Set.of(7),
null));
+ expected.add(segment);
+ }
+ assertEquals(pruner(moduloRecords).prune(request(predicate("EQUALS",
"-1")), moduloRecords.keySet()),
+ expected);
+
+ String value = "80ff0102";
+ Map<String, String> rawConfig = Map.of("useRawBytes", "true");
+ int textPartition =
PartitionFunctionFactory.getPartitionFunction("Murmur", 97,
null).getPartition(value);
+ int rawPartition = PartitionFunctionFactory.getPartitionFunction("Murmur",
97, rawConfig).getPartition(value);
+ assertNotEquals(textPartition, rawPartition, "Fixture must distinguish
raw-byte and string hashing");
+ Map<String, ZNRecord> murmurRecords = new LinkedHashMap<>();
+ murmurRecords.put("text", metadata("text", "Murmur", 97,
Set.of(textPartition), null));
+ murmurRecords.put("raw", metadata("raw", "Murmur", 97,
Set.of(rawPartition), rawConfig));
+ murmurRecords.put("wrongRaw", metadata("wrongRaw", "Murmur", 97,
Set.of(textPartition), rawConfig));
+ expected = new HashSet<>(Set.of("text", "raw"));
+ for (int i = 0; i < 253; i++) {
+ String segment = "text_" + i;
+ murmurRecords.put(segment, metadata(segment, "Murmur", 97,
Set.of(textPartition), null));
+ expected.add(segment);
+ }
+ assertEquals(pruner(murmurRecords).prune(request(predicate("EQUALS",
value)), murmurRecords.keySet()),
+ expected);
+
+ Map<String, ZNRecord> lookupRecords = new LinkedHashMap<>();
+ lookupRecords.put("first", metadata("first", "BoundedColumnValue", 3,
Set.of(1),
+ Map.of("columnValues", "11|12", "columnValuesDelimiter", "|")));
+ lookupRecords.put("second", metadata("second", "BoundedColumnValue", 3,
Set.of(2),
+ Map.of("columnValues", "12|11", "columnValuesDelimiter", "|")));
+ lookupRecords.put("modulo", metadata("modulo", "Modulo", 3, Set.of(2),
null));
+ assertEquals(pruner(lookupRecords).prune(request(predicate("EQUALS",
"11")), lookupRecords.keySet()),
+ lookupRecords.keySet());
+ }
+
+ @DataProvider
+ public Object[][] candidateCounts() {
+ return new Object[][]{{1}, {2}, {256}};
+ }
+
+ @Test(dataProvider = "candidateCounts")
+ public void testAndOrUnsupportedPredicatesAndLazyInValues(int numSegments)
throws Exception {
+ Map<String, ZNRecord> records = new LinkedHashMap<>();
+ for (int i = 0; i < numSegments; i++) {
+ String segment = "segment_" + i;
+ records.put(segment, metadata(segment, "Modulo", 8, Set.of(1), null));
+ }
+ SinglePartitionColumnSegmentPruner pruner = pruner(records);
+ Set<String> segments = records.keySet();
+ Expression invalidValue = predicate("EQUALS", "invalid-number");
+ assertEquals(pruner.prune(request(predicate("IN", "1", "invalid-number")),
segments), segments);
+ expectThrows(NumberFormatException.class,
+ () -> pruner.prune(request(predicate("IN", "2", "invalid-number")),
segments));
+ assertEquals(pruner.prune(request(function("AND", predicate("EQUALS",
"2"), invalidValue)), segments), Set.of());
+ assertEquals(pruner.prune(request(function("OR", predicate("EQUALS", "1"),
invalidValue)), segments), segments);
+ Expression invalidOperator = function("INVALID_OPERATOR");
+ assertEquals(pruner.prune(request(function("OR", predicate("EQUALS", "1"),
invalidOperator)), segments), segments);
+ assertEquals(pruner.prune(request(function("AND", predicate("EQUALS",
"2"), invalidOperator)), segments), Set.of());
+ expectThrows(IllegalArgumentException.class,
+ () -> pruner.prune(request(function("OR", predicate("EQUALS", "2"),
invalidOperator)), segments));
+ expectThrows(IllegalArgumentException.class, () ->
pruner.prune(request(invalidOperator), segments));
+
+ Expression unsupported = predicate("GREATER_THAN", "100");
+ assertEquals(pruner.prune(request(function("AND", predicate("IN", "0",
"1"), unsupported)), segments), segments);
+ assertEquals(pruner.prune(request(function("OR", predicate("EQUALS", "2"),
unsupported)), segments), segments);
+ assertEquals(pruner.prune(request(function("NOT", predicate("EQUALS",
"1"))), segments), segments);
+ assertEquals(pruner.prune(request(function("EQUALS",
RequestUtils.getIdentifierExpression("other"),
+ RequestUtils.getLiteralExpression("invalid-number"))), segments),
segments);
+ Expression transformedColumn = function("LOWER",
RequestUtils.getIdentifierExpression(COLUMN));
+ assertEquals(pruner.prune(request(function("EQUALS", transformedColumn,
+ RequestUtils.getLiteralExpression("invalid-number"))), segments),
segments);
+ BrokerRequest unfilteredRequest = new BrokerRequest();
+ unfilteredRequest.setPinotQuery(new PinotQuery());
+ assertSame(pruner.prune(unfilteredRequest, segments), segments);
+ }
+
+ @Test
+ public void testEmptyCandidatesDoNotEvaluateFilter() throws Exception {
+ SinglePartitionColumnSegmentPruner pruner = pruner(Map.of("one",
metadata("one", "Modulo", 8, Set.of(1), null)));
+ assertEquals(pruner.prune(request(function("INVALID_OPERATOR")),
Set.of()), Set.of());
+ assertEquals(pruner.prune(request(predicate("EQUALS", "invalid-number")),
Set.of()), Set.of());
+ }
+
+ @Test
+ public void testUnknownMetadataIsConservativeAndDoesNotEvaluateFilter()
throws Exception {
Review Comment:
Follow-up in d9135a1518 (38 focused Apache cases passed): Expanded both
fixtures through the shared provider: 255/256 candidates and cache-disabled 256
candidates, as well as the existing small cases. Unknown metadata still
preserves every candidate without decoding invalid operators or values; the
large-config fixture now actually reaches the prepared path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
