This is an automated email from the ASF dual-hosted git repository.
kfaraz 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 15e50434614 perf: add datasource filter pushdown for sys.segments
table (#19718)
15e50434614 is described below
commit 15e50434614ebdc553cf321cbdaab832078ee1f5
Author: jtuglu1 <[email protected]>
AuthorDate: Wed Jul 22 23:19:25 2026 -0700
perf: add datasource filter pushdown for sys.segments table (#19718)
For large used segment counts (~20M+), the UI and `sys.segments` tables are
extremely slow.
This is the first patch in a series of changes to speed up queries against
the `sys.segments`
(and potentially other sys tables) and, transitively, the Druid console UI.
Changes:
- Push down the datasource filter to `MetadataSegmentView` and
`BrokerSegmentMetadataCache`
- Add `SystemSchemaFilters` utility to extract column values on which a
strict filter may apply
- This will help speed up queries with equality or IN filter on the
datasource column
---
benchmarks/pom.xml | 5 +
.../calcite/schema/SysSegmentsTableBenchmark.java | 286 +++++++++++++++++++++
.../metadata/AbstractSegmentMetadataCache.java | 17 ++
.../sql/calcite/schema/MetadataSegmentView.java | 22 +-
.../druid/sql/calcite/schema/SystemSchema.java | 34 ++-
.../sql/calcite/schema/SystemSchemaFilters.java | 184 +++++++++++++
.../schema/SystemServerPropertiesTable.java | 61 +----
.../calcite/schema/SystemSchemaFiltersTest.java | 274 ++++++++++++++++++++
.../druid/sql/calcite/schema/SystemSchemaTest.java | 127 ++++++++-
9 files changed, 944 insertions(+), 66 deletions(-)
diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml
index ebe7c0d1814..bc921acb159 100644
--- a/benchmarks/pom.xml
+++ b/benchmarks/pom.xml
@@ -124,6 +124,11 @@
<artifactId>calcite-core</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.calcite</groupId>
+ <artifactId>calcite-linq4j</artifactId>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>com.github.wnameless</groupId>
<artifactId>json-flattener</artifactId>
diff --git
a/benchmarks/src/test/java/org/apache/druid/sql/calcite/schema/SysSegmentsTableBenchmark.java
b/benchmarks/src/test/java/org/apache/druid/sql/calcite/schema/SysSegmentsTableBenchmark.java
new file mode 100644
index 00000000000..9eadf5bffee
--- /dev/null
+++
b/benchmarks/src/test/java/org/apache/druid/sql/calcite/schema/SysSegmentsTableBenchmark.java
@@ -0,0 +1,286 @@
+/*
+ * 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.sql.calcite.schema;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.calcite.DataContext;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.linq4j.QueryProvider;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.druid.client.BrokerSegmentWatcherConfig;
+import org.apache.druid.client.InternalQueryConfig;
+import org.apache.druid.client.TimelineServerView;
+import org.apache.druid.client.coordinator.NoopCoordinatorClient;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.CloseableIterators;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.segment.join.JoinableFactory;
+import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig;
+import org.apache.druid.server.QueryLifecycleFactory;
+import org.apache.druid.server.SegmentManager;
+import org.apache.druid.server.metrics.NoopServiceEmitter;
+import org.apache.druid.server.security.AllowAllAuthorizer;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.Authorizer;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.Escalator;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.SegmentStatusInCluster;
+import org.apache.druid.timeline.partition.LinearShardSpec;
+import org.easymock.EasyMock;
+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.Param;
+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 org.openjdk.jmh.infra.Blackhole;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.EnumMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Benchmarks representative {@code sys.segments} queries as they run through
+ * {@link SystemSchema.SegmentsTable#scan}, against a synthetic cluster of
{@code numSegments}
+ * segments spread across {@code numDataSources} datasources.
+ *
+ * <p>The benchmark deliberately has NO in-code "pushdown on/off" switch: it
simply issues each query
+ * through the production scan path. To measure the effect of the datasource
filter push-down, run
+ * this benchmark on the patched code and again on the reverted code (e.g.
{@code git stash} the
+ * SystemSchema / MetadataSegmentView / AbstractSegmentMetadataCache changes,
rebuild, rerun) and
+ * compare. Pre-patch, {@link #SINGLE_DATASOURCE_EQUALS}/{@link
#MULTI_DATASOURCE_IN} return every segment because
+ * {@code scan} ignores its filters; post-patch they return only the matching
datasources. That
+ * difference in materialized work per query IS the optimization being
measured.
+ *
+ * <p>Sample queries:
+ * <ul>
+ * <li>{@code BASE_SCAN} - {@code SELECT * FROM sys.segments} (the
unfiltered console/aggregate baseline)</li>
+ * <li>{@code SINGLE_DATASOURCE_EQUALS} - {@code WHERE datasource =
'datasource_0'} (single-datasource drill-down)</li>
+ * <li>{@code MULTI_DATASOURCE_IN} - {@code WHERE datasource IN
('datasource_0','datasource_1','datasource_2')}</li>
+ * </ul>
+ *
+ * <p>The available-segment cache is left empty so the benchmark isolates the
published-segment path,
+ * which dominates cost on large clusters.
+ */
+@State(Scope.Benchmark)
+@Fork(value = 1, jvmArgsAppend = {"-Xmx12g"})
+@Warmup(iterations = 3, time = 3)
+@Measurement(iterations = 5, time = 3)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+public class SysSegmentsTableBenchmark
+{
+ public enum Query
+ {
+ BASE_SCAN,
+ SINGLE_DATASOURCE_EQUALS,
+ MULTI_DATASOURCE_IN
+ }
+
+ @Param({"1000000", "10000000"})
+ private int numSegments;
+
+ @Param({"1000"})
+ private int numDataSources;
+
+ @Param
+ private Query query;
+
+ private SystemSchema.SegmentsTable segmentsTable;
+ private List<SegmentStatusInCluster> publishedSegments;
+ private DataContext dataContext;
+ private Map<Query, List<RexNode>> filtersByQuery;
+
+ /**
+ * An empty {@link BrokerSegmentMetadataCache}: the benchmark never
announces segments to it, so the
+ * available-segment side of the scan contributes nothing and the published
path is measured in
+ * isolation. Mirrors the null-arg construction proven in
DruidSchemaInternRowSignatureBenchmark.
+ */
+ private static class EmptyBrokerSegmentMetadataCache extends
BrokerSegmentMetadataCache
+ {
+ EmptyBrokerSegmentMetadataCache()
+ {
+ super(
+ EasyMock.mock(QueryLifecycleFactory.class),
+ EasyMock.mock(TimelineServerView.class),
+ BrokerSegmentMetadataCacheConfig.create(),
+ EasyMock.mock(Escalator.class),
+ EasyMock.mock(InternalQueryConfig.class),
+ new NoopServiceEmitter(),
+ new PhysicalDatasourceMetadataFactory(
+ EasyMock.mock(JoinableFactory.class),
+ EasyMock.mock(SegmentManager.class)
+ ),
+ new NoopCoordinatorClient(),
+ CentralizedDatasourceSchemaConfig.create()
+ );
+ }
+ }
+
+ @Setup(Level.Trial)
+ public void setup()
+ {
+ final List<SegmentStatusInCluster> segments = new ArrayList<>(numSegments);
+ for (int i = 0; i < numSegments; i++) {
+ final String dataSource = StringUtils.format("datasource_%d", i %
numDataSources);
+ final int dayOffset = i / numDataSources;
+ final SegmentId segmentId = SegmentId.of(
+ dataSource,
+ Intervals.utc(dayOffset * 86_400_000L, (dayOffset + 1) *
86_400_000L),
+ "1",
+ new LinearShardSpec(0)
+ );
+ final DataSegment segment =
DataSegment.builder(segmentId).size(1000L).build();
+ segments.add(new SegmentStatusInCluster(segment, false, 1, 100L, false));
+ }
+ // sys.segments serves published segments sorted by SegmentId
(datasource-prefixed).
+ segments.sort(Comparator.naturalOrder());
+ publishedSegments = ImmutableList.copyOf(segments);
+
+ // Returns a fresh iterator over the synthetic segments on every fetch
(cache disabled below).
+ final NoopCoordinatorClient coordinatorClient = new NoopCoordinatorClient()
+ {
+ @Override
+ public ListenableFuture<CloseableIterator<SegmentStatusInCluster>>
fetchAllUsedSegmentsWithOvershadowedStatus(
+ Set<String> watchedDataSources,
+ boolean includeRealtimeSegments
+ )
+ {
+ return
Futures.immediateFuture(CloseableIterators.withEmptyBaggage(publishedSegments.iterator()));
+ }
+ };
+
+ final BrokerSegmentMetadataCacheConfig config = new
DefaultObjectMapper().convertValue(
+ ImmutableMap.of("metadataSegmentCacheEnable", false),
+ BrokerSegmentMetadataCacheConfig.class
+ );
+
+ final MetadataSegmentView metadataView = new MetadataSegmentView(
+ coordinatorClient,
+ new BrokerSegmentWatcherConfig(),
+ config,
+ new NoopServiceEmitter()
+ );
+
+ final DruidSchema druidSchema =
+ new DruidSchema(new EmptyBrokerSegmentMetadataCache(), null,
CatalogResolver.NULL_RESOLVER);
+
+ final AuthorizerMapper authorizerMapper = new AuthorizerMapper(null)
+ {
+ @Override
+ public Authorizer getAuthorizer(String name)
+ {
+ return new AllowAllAuthorizer(null);
+ }
+ };
+
+ segmentsTable = new SystemSchema.SegmentsTable(druidSchema, metadataView,
new DefaultObjectMapper(), authorizerMapper);
+
+ filtersByQuery = buildFilters();
+
+ final AuthenticationResult authenticationResult = new
AuthenticationResult("benchmark", "benchmark", null, null);
+ dataContext = new DataContext()
+ {
+ @Override
+ public SchemaPlus getRootSchema()
+ {
+ return null;
+ }
+
+ @Override
+ public JavaTypeFactoryImpl getTypeFactory()
+ {
+ return null;
+ }
+
+ @Override
+ public QueryProvider getQueryProvider()
+ {
+ return null;
+ }
+
+ @Override
+ public Object get(String name)
+ {
+ return PlannerContext.DATA_CTX_AUTHENTICATION_RESULT.equals(name) ?
authenticationResult : null;
+ }
+ };
+ }
+
+ /**
+ * Builds the filter list for each sample query. The datasource input-ref is
given the literal's
+ * type so Calcite does not wrap the literal in a CAST (which would defeat
the push-down extractor).
+ * "datasource" is column index 1 in SEGMENTS_SIGNATURE.
+ */
+ private static Map<Query, List<RexNode>> buildFilters()
+ {
+ final RexBuilder rexBuilder = new RexBuilder(new JavaTypeFactoryImpl());
+ final RexLiteral ds0 = (RexLiteral) rexBuilder.makeLiteral("datasource_0");
+ final RexLiteral ds1 = (RexLiteral) rexBuilder.makeLiteral("datasource_1");
+ final RexLiteral ds2 = (RexLiteral) rexBuilder.makeLiteral("datasource_2");
+ final RexNode dsRef = rexBuilder.makeInputRef(ds0.getType(), 1);
+
+ final Map<Query, List<RexNode>> filters = new EnumMap<>(Query.class);
+ filters.put(Query.BASE_SCAN, ImmutableList.of());
+ filters.put(
+ Query.SINGLE_DATASOURCE_EQUALS,
+ ImmutableList.of(rexBuilder.makeCall(SqlStdOperatorTable.EQUALS,
dsRef, ds0))
+ );
+ filters.put(
+ Query.MULTI_DATASOURCE_IN,
+ ImmutableList.of(rexBuilder.makeIn(dsRef, ImmutableList.of(ds0, ds1,
ds2)))
+ );
+ return filters;
+ }
+
+ @Benchmark
+ public void scan(Blackhole blackhole)
+ {
+ long rows = 0;
+ for (Object[] row : segmentsTable.scan(dataContext,
filtersByQuery.get(query), null)) {
+ blackhole.consume(row);
+ rows++;
+ }
+ blackhole.consume(rows);
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java
b/server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java
index 03d5a44d6ad..d8677fba952 100644
---
a/server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java
+++
b/server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java
@@ -445,6 +445,23 @@ public abstract class AbstractSegmentMetadataCache<T
extends DataSourceInformati
.iterator();
}
+ /**
+ * Like {@link #iterateSegmentMetadata()} but restricted to the given
datasources, so a pushed-down
+ * {@code datasource} predicate from sys.segments scans only the matching
datasources' segment maps
+ * rather than the whole cluster. A {@code null} argument iterates all
datasources.
+ */
+ public Iterator<AvailableSegmentMetadata> iterateSegmentMetadata(@Nullable
Set<String> dataSources)
+ {
+ if (dataSources == null) {
+ return iterateSegmentMetadata();
+ }
+ return FluentIterable.from(dataSources)
+ .transform(segmentMetadataInfo::get)
+ .filter(Objects::nonNull)
+ .transformAndConcat(Map::values)
+ .iterator();
+ }
+
/**
* Get metadata for the specified segment, which includes information like
RowSignature, realtime & numRows.
*
diff --git
a/sql/src/main/java/org/apache/druid/sql/calcite/schema/MetadataSegmentView.java
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/MetadataSegmentView.java
index 34fd6fba779..b292e5e8e04 100644
---
a/sql/src/main/java/org/apache/druid/sql/calcite/schema/MetadataSegmentView.java
+++
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/MetadataSegmentView.java
@@ -23,6 +23,7 @@ import com.google.common.base.Preconditions;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.ImmutableSortedSet;
+import com.google.common.collect.Iterators;
import com.google.common.util.concurrent.Uninterruptibles;
import com.google.inject.Inject;
import org.apache.druid.client.BrokerSegmentWatcherConfig;
@@ -49,7 +50,10 @@ import org.apache.druid.timeline.SegmentStatusInCluster;
import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
import org.joda.time.Duration;
+import javax.annotation.Nullable;
+
import java.util.Iterator;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -185,12 +189,26 @@ public class MetadataSegmentView
*/
Iterator<SegmentStatusInCluster> getSegments()
{
+ return getSegments(null);
+ }
+
+ /**
+ * Returns published (and, with centralized schema, realtime) segment
metadata, optionally
+ * restricted to {@code dataSources}.
+ */
+ Iterator<SegmentStatusInCluster> getSegments(@Nullable Set<String>
dataSources)
+ {
+ final Iterator<SegmentStatusInCluster> base;
if (isCacheEnabled) {
Uninterruptibles.awaitUninterruptibly(cachePopulated);
- return publishedSegments.iterator();
+ base = publishedSegments.iterator();
} else {
- return fetchSegmentMetadataFromCoordinator();
+ // Cache disabled: the Coordinator returns all used segments; filter
client-side to preserve semantics.
+ base = fetchSegmentMetadataFromCoordinator();
}
+ return dataSources == null
+ ? base
+ : Iterators.filter(base, s ->
dataSources.contains(s.getDataSegment().getDataSource()));
}
// Note that coordinator must be up to get segments
diff --git
a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
index 3d3d2a65791..56d6bb593d3 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchema.java
@@ -93,6 +93,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -312,6 +313,8 @@ public class SystemSchema extends AbstractSchema
*/
static class SegmentsTable extends AbstractTable implements
ProjectableFilterableTable
{
+ private static final int DATASOURCE_COLUMN =
SEGMENTS_SIGNATURE.indexOf("datasource");
+
private final DruidSchema druidSchema;
private final ObjectMapper jsonMapper;
private final AuthorizerMapper authorizerMapper;
@@ -352,14 +355,23 @@ public class SystemSchema extends AbstractSchema
// get available segments from druidSchema
final BrokerSegmentMetadataCache availableMetadataCache =
druidSchema.cache();
+ // Best-effort push-down of a `datasource` equality/IN filter so we scan
only the matching
+ // datasources instead of every segment in the cluster. Null => no
usable filter => full scan.
+ // The filters are intentionally left in the list, so Calcite still
applies them and correctness
+ // holds even if this extraction is conservative or over-broad.
+ final Set<String> dataSourceFilter = getDataSourceFilter(filters);
+
// Keep track of which segments we emitted from the publishedSegments
iterator, so we don't emit them again
- // from the availableSegments iterator.
+ // from the availableSegments iterator. When a datasource filter is
pushed down we only emit the matching
+ // datasources' segments, so avoid pre-sizing to the whole-cluster
segment count (a huge, wasted allocation).
final Set<SegmentId> segmentsAlreadySeen =
-
Sets.newHashSetWithExpectedSize(availableMetadataCache.getTotalSegments());
+ dataSourceFilter == null
+ ?
Sets.newHashSetWithExpectedSize(availableMetadataCache.getTotalSegments())
+ : new HashSet<>();
// Get segments from metadata segment cache (if enabled in SQL planner
config), else directly from
// Coordinator. This may include both published and realtime segments.
- final Iterator<SegmentStatusInCluster> metadataStoreSegments =
metadataView.getSegments();
+ final Iterator<SegmentStatusInCluster> metadataStoreSegments =
metadataView.getSegments(dataSourceFilter);
final FluentIterable<Object[]> publishedSegments = FluentIterable
.from(() -> getAuthorizedPublishedSegments(metadataStoreSegments,
root))
.transform(val -> {
@@ -428,7 +440,7 @@ public class SystemSchema extends AbstractSchema
// If druid.centralizedDatasourceSchema.enabled is set on the
Coordinator, all the segments in this loop
// would be covered in the previous iteration since Coordinator would
return realtime segments as well.
final FluentIterable<Object[]> availableSegments = FluentIterable
- .from(() ->
getAuthorizedAvailableSegments(availableMetadataCache.iterateSegmentMetadata(),
root))
+ .from(() ->
getAuthorizedAvailableSegments(availableMetadataCache.iterateSegmentMetadata(dataSourceFilter),
root))
.transform(val -> {
final DataSegment segment = val.getSegment();
if (segmentsAlreadySeen.contains(segment.getId())) {
@@ -517,6 +529,20 @@ public class SystemSchema extends AbstractSchema
return authorizedSegments.iterator();
}
+ /**
+ * Best-effort extraction of an exact-match {@code datasource} constraint
(column
+ * {@link #DATASOURCE_COLUMN}) from the pushed-down filters, so
sys.segments can restrict its scan
+ * to the matching datasources rather than materializing every segment in
the cluster. Delegates to
+ * {@link SystemSchemaFilters}, which handles {@code datasource = 'x'},
{@code datasource IN (...)},
+ * OR-of-equalities, and nested {@code AND}/{@code OR}. Returns {@code
null} when no usable
+ * datasource predicate is present, in which case the previous full-scan
behavior is retained.
+ */
+ @Nullable
+ static Set<String> getDataSourceFilter(List<RexNode> filters)
+ {
+ return SystemSchemaFilters.extractColumnValues(filters,
DATASOURCE_COLUMN);
+ }
+
private static class PartialSegmentData
{
private final long isAvailable;
diff --git
a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchemaFilters.java
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchemaFilters.java
new file mode 100644
index 00000000000..90dd2a7e34c
--- /dev/null
+++
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemSchemaFilters.java
@@ -0,0 +1,184 @@
+/*
+ * 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.sql.calcite.schema;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.util.NlsString;
+import org.apache.calcite.util.Sarg;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Helpers for pushing a simple string-column predicate down from a {@code
ProjectableFilterableTable}
+ * system-table scan (e.g. {@code datasource} for sys.segments, {@code
server}/{@code service_name} for
+ * sys.server_properties) into that table's data source, so the scan
materializes only the matching
+ * rows instead of the whole cluster.
+ *
+ * <p>All extraction is best-effort. The filters are left in the planner's
filter list, so Calcite
+ * still applies them as a post-filter; correctness therefore holds even if
extraction returns a
+ * conservative (over-broad) set - the caller just does a bit of extra work -
or {@code null} (no
+ * constraint), which retains a full scan. The extractor never returns a set
narrower than the true
+ * constraint on the column, so pushing it down can never drop a matching row.
+ */
+final class SystemSchemaFilters
+{
+ private SystemSchemaFilters()
+ {
+ }
+
+ /**
+ * Extracts the finite set of exact string values that column {@code
columnIndex} is constrained to
+ * by the given filters (a top-level list is implicitly ANDed). Handles
{@code col = 'x'},
+ * {@code col IN (...)} (normalized by Calcite to SEARCH), OR-of-equalities,
and arbitrarily nested
+ * {@code AND}/{@code OR} - including a whole {@code WHERE} passed as a
single {@code AND(...)}
+ * RexCall (as Calcite's filter-scan rule may do), where any conjunct that
does not constrain the
+ * column (e.g. {@code is_active = 1}) is simply ignored.
+ *
+ * @return the bounded value set, or {@code null} when the column is not
bounded to a finite set
+ * (no predicate, or a range/{@code LIKE}/{@code !=} predicate)
+ */
+ @Nullable
+ static Set<String> extractColumnValues(List<RexNode> filters, int
columnIndex)
+ {
+ return intersectConjuncts(filters, columnIndex);
+ }
+
+ /**
+ * Multi-column variant of {@link #extractColumnValues(List, int)}: extracts
a bounded value set for
+ * each of {@code columnIndices} independently. A column with no usable
constraint is absent from the
+ * returned map (rather than mapped to {@code null}).
+ */
+ static Map<Integer, Set<String>> extractColumnValues(List<RexNode> filters,
int... columnIndices)
+ {
+ final Map<Integer, Set<String>> result = new HashMap<>();
+ for (final int columnIndex : columnIndices) {
+ final Set<String> values = extractColumnValues(filters, columnIndex);
+ if (values != null) {
+ result.put(columnIndex, values);
+ }
+ }
+ return result;
+ }
+
+ /**
+ * AND semantics: intersect the value sets from the extractable conjuncts,
ignoring conjuncts that
+ * don't constrain the column. Returns {@code null} if no conjunct yields a
value set.
+ */
+ @Nullable
+ private static Set<String> intersectConjuncts(List<RexNode> conjuncts, int
columnIndex)
+ {
+ Set<String> result = null;
+ for (final RexNode conjunct : conjuncts) {
+ final Set<String> values = extractColumnValues(conjunct, columnIndex);
+ if (values != null) {
+ result = (result == null) ? values : Sets.intersection(result,
values).immutableCopy();
+ }
+ }
+ return result;
+ }
+
+ @Nullable
+ private static Set<String> extractColumnValues(RexNode node, int columnIndex)
+ {
+ if (!(node instanceof RexCall)) {
+ return null;
+ }
+ final RexCall call = (RexCall) node;
+ switch (call.getKind()) {
+ case EQUALS:
+ return equalsColumn(call, columnIndex);
+ case AND:
+ return intersectConjuncts(call.getOperands(), columnIndex);
+ case OR:
+ final Set<String> union = new HashSet<>();
+ for (final RexNode operand : call.getOperands()) {
+ final Set<String> values = extractColumnValues(operand, columnIndex);
+ if (values == null) {
+ // An un-extractable disjunct means we cannot bound the value set
for this OR.
+ return null;
+ }
+ union.addAll(values);
+ }
+ return union;
+ case SEARCH:
+ return searchColumn(call, columnIndex);
+ default:
+ return null;
+ }
+ }
+
+ @Nullable
+ private static Set<String> equalsColumn(RexCall call, int columnIndex)
+ {
+ final List<RexNode> ops = call.getOperands();
+ if (ops.size() != 2) {
+ return null;
+ }
+ final RexNode a = ops.get(0);
+ final RexNode b = ops.get(1);
+ final RexLiteral literal;
+ if (isColumnRef(a, columnIndex) && b instanceof RexLiteral) {
+ literal = (RexLiteral) b;
+ } else if (isColumnRef(b, columnIndex) && a instanceof RexLiteral) {
+ literal = (RexLiteral) a;
+ } else {
+ return null;
+ }
+ final String value = RexLiteral.stringValue(literal);
+ return value == null ? null : ImmutableSet.of(value);
+ }
+
+ @Nullable
+ private static Set<String> searchColumn(RexCall call, int columnIndex)
+ {
+ final List<RexNode> ops = call.getOperands();
+ if (ops.size() != 2 || !isColumnRef(ops.get(0), columnIndex) ||
!(ops.get(1) instanceof RexLiteral)) {
+ return null;
+ }
+ final Sarg<?> sarg = ((RexLiteral) ops.get(1)).getValueAs(Sarg.class);
+ // Only exact-value sets (IN / OR-of-=) can bound the scan; ranges (>, <,
LIKE) cannot.
+ if (sarg == null || !sarg.isPoints()) {
+ return null;
+ }
+ final Set<String> values = new HashSet<>();
+ for (final Range<?> range : sarg.rangeSet.asRanges()) {
+ final Object endpoint = range.lowerEndpoint();
+ values.add(endpoint instanceof NlsString ? ((NlsString)
endpoint).getValue() : String.valueOf(endpoint));
+ }
+ return values;
+ }
+
+ private static boolean isColumnRef(RexNode node, int columnIndex)
+ {
+ return node instanceof RexInputRef && ((RexInputRef) node).getIndex() ==
columnIndex;
+ }
+}
diff --git
a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemServerPropertiesTable.java
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemServerPropertiesTable.java
index 4b25c3cefa1..cdadd9d17c2 100644
---
a/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemServerPropertiesTable.java
+++
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/SystemServerPropertiesTable.java
@@ -27,14 +27,10 @@ import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.linq4j.Linq4j;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
-import org.apache.calcite.rex.RexCall;
-import org.apache.calcite.rex.RexInputRef;
-import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.schema.ProjectableFilterableTable;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.impl.AbstractTable;
-import org.apache.calcite.sql.SqlKind;
import org.apache.druid.discovery.DiscoveryDruidNode;
import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
import org.apache.druid.java.util.common.StringUtils;
@@ -60,7 +56,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -139,8 +134,9 @@ public class SystemServerPropertiesTable extends
AbstractTable implements Projec
);
SystemSchema.checkStateReadAccessForServers(authenticationResult,
authorizerMapper);
- // Extract equality filters to skip fetching properties from non-matching
servers.
- final Map<Integer, Set<String>> columnFilters =
extractColumnEqualityFilters(filters, SERVER_INDEX, SERVICE_NAME_INDEX);
+ // Extract server/service_name constraints to skip fetching properties
from non-matching servers.
+ final Map<Integer, Set<String>> columnFilters =
+ SystemSchemaFilters.extractColumnValues(filters, SERVER_INDEX,
SERVICE_NAME_INDEX);
final Set<String> serverFilter = columnFilters.get(SERVER_INDEX);
final Set<String> serviceNameFilter =
columnFilters.get(SERVICE_NAME_INDEX);
@@ -183,57 +179,6 @@ public class SystemServerPropertiesTable extends
AbstractTable implements Projec
return Linq4j.asEnumerable(rows);
}
- /**
- * Extracts simple equality filters ({@code column = 'literal'}) for the
specified columns.
- * Only handles top-level AND equalities; any other predicate (!=, LIKE, OR,
functions) is
- * ignored and left for Calcite to apply as a post-filter.
- *
- * @return map from column index to the set of literal values; absent key
means no filter for that column
- */
- private static Map<Integer, Set<String>> extractColumnEqualityFilters(final
List<RexNode> filters, final int... columnIndices)
- {
- final Map<Integer, Set<String>> result = new HashMap<>();
- for (final RexNode filter : filters) {
- for (final int columnIndex : columnIndices) {
- final String value = extractEqualityOnColumn(filter, columnIndex);
- if (value != null) {
- result.computeIfAbsent(columnIndex, k -> new HashSet<>()).add(value);
- break;
- }
- }
- }
- return result;
- }
-
- /**
- * Returns the string literal value if the node is a simple {@code column =
'literal'} (or reversed) equality
- * on the given column index. Returns null for anything else — Calcite
handles those as post-filters.
- */
- @Nullable
- private static String extractEqualityOnColumn(final RexNode node, final int
columnIndex)
- {
- if (!(node instanceof RexCall)) {
- return null;
- }
- final RexCall call = (RexCall) node;
- if (call.getKind() != SqlKind.EQUALS) {
- return null;
- }
- final RexNode left = call.getOperands().get(0);
- final RexNode right = call.getOperands().get(1);
-
- if (left instanceof RexInputRef && right instanceof RexLiteral) {
- if (((RexInputRef) left).getIndex() == columnIndex) {
- return RexLiteral.stringValue(right);
- }
- } else if (right instanceof RexInputRef && left instanceof RexLiteral) {
- if (((RexInputRef) right).getIndex() == columnIndex) {
- return RexLiteral.stringValue(left);
- }
- }
- return null;
- }
-
private static Object[] projectRow(final Object[] row, @Nullable final int[]
projects)
{
if (projects == null) {
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaFiltersTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaFiltersTest.java
new file mode 100644
index 00000000000..f8344db8ede
--- /dev/null
+++
b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaFiltersTest.java
@@ -0,0 +1,274 @@
+/*
+ * 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.sql.calcite.schema;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Unit tests for the predicate-extraction utilities in {@link
SystemSchemaFilters}. RexNodes are
+ * built by hand (as Calcite's filter-scan rule would hand them to a {@code
ProjectableFilterableTable})
+ * so the extraction logic can be exercised directly without a full planner
run.
+ */
+public class SystemSchemaFiltersTest
+{
+ // Two arbitrary string columns to constrain, plus a numeric-ish column used
for "other column" cases.
+ private static final int COL_A = 1;
+ private static final int COL_B = 2;
+ private static final int COL_OTHER = 4;
+
+ private RexBuilder rexBuilder;
+ private RexLiteral foo;
+ private RexLiteral bar;
+ private RexLiteral baz;
+ private RexNode aRef;
+ private RexNode bRef;
+ private RexNode otherRef;
+
+ @Before
+ public void setUp()
+ {
+ rexBuilder = new RexBuilder(new JavaTypeFactoryImpl());
+ foo = (RexLiteral) rexBuilder.makeLiteral("foo");
+ bar = (RexLiteral) rexBuilder.makeLiteral("bar");
+ baz = (RexLiteral) rexBuilder.makeLiteral("baz");
+ // Match the input-ref type to the literal type so Calcite does not wrap
the literal in a CAST.
+ aRef = rexBuilder.makeInputRef(foo.getType(), COL_A);
+ bRef = rexBuilder.makeInputRef(foo.getType(), COL_B);
+ otherRef = rexBuilder.makeInputRef(foo.getType(), COL_OTHER);
+ }
+
+ @Test
+ public void testEquals()
+ {
+ // col = 'foo'
+ Assert.assertEquals(
+ ImmutableSet.of("foo"),
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(rexBuilder.makeCall(SqlStdOperatorTable.EQUALS,
aRef, foo)),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testEqualsReversedOperands()
+ {
+ // 'foo' = col
+ Assert.assertEquals(
+ ImmutableSet.of("foo"),
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(rexBuilder.makeCall(SqlStdOperatorTable.EQUALS,
foo, aRef)),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testIn()
+ {
+ // col IN ('foo', 'bar') -- Calcite normalizes IN to a SEARCH over a
points Sarg
+ Assert.assertEquals(
+ ImmutableSet.of("foo", "bar"),
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(rexBuilder.makeIn(aRef, ImmutableList.of(foo,
bar))),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testOrOfEqualities()
+ {
+ // col = 'foo' OR col = 'bar'
+ Assert.assertEquals(
+ ImmutableSet.of("foo", "bar"),
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(rexBuilder.makeCall(
+ SqlStdOperatorTable.OR,
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, aRef, foo),
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, aRef, bar)
+ )),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testOrWithUnextractableDisjunctReturnsNull()
+ {
+ // col = 'foo' OR col > 'bar' -- one disjunct cannot bound the set, so the
whole OR is unbounded
+ Assert.assertNull(
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(rexBuilder.makeCall(
+ SqlStdOperatorTable.OR,
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, aRef, foo),
+ rexBuilder.makeCall(SqlStdOperatorTable.GREATER_THAN, aRef,
bar)
+ )),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testTopLevelConjunctsIntersect()
+ {
+ // A top-level filter list is implicitly ANDed: IN ('foo','bar') AND IN
('bar','baz') => {'bar'}
+ Assert.assertEquals(
+ ImmutableSet.of("bar"),
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(
+ rexBuilder.makeIn(aRef, ImmutableList.of(foo, bar)),
+ rexBuilder.makeIn(aRef, ImmutableList.of(bar, baz))
+ ),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testNestedAndConjunctsIntersect()
+ {
+ // A whole WHERE passed as a single AND(...) RexCall: IN ('foo','bar') AND
IN ('bar','baz') => {'bar'}
+ Assert.assertEquals(
+ ImmutableSet.of("bar"),
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(rexBuilder.makeCall(
+ SqlStdOperatorTable.AND,
+ rexBuilder.makeIn(aRef, ImmutableList.of(foo, bar)),
+ rexBuilder.makeIn(aRef, ImmutableList.of(bar, baz))
+ )),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testAndIgnoresNonMatchingConjunct()
+ {
+ // col_a = 'foo' AND col_other = 'foo' => {'foo'} (the non-target conjunct
is ignored)
+ Assert.assertEquals(
+ ImmutableSet.of("foo"),
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(rexBuilder.makeCall(
+ SqlStdOperatorTable.AND,
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, aRef, foo),
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, otherRef, foo)
+ )),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testAndWithNoConstraintOnColumnReturnsNull()
+ {
+ // AND of predicates that never touch the target column => null (full scan
retained)
+ Assert.assertNull(
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(rexBuilder.makeCall(
+ SqlStdOperatorTable.AND,
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, otherRef, foo),
+ rexBuilder.makeCall(SqlStdOperatorTable.GREATER_THAN,
otherRef, bar)
+ )),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testEmptyFilterListReturnsNull()
+ {
+
Assert.assertNull(SystemSchemaFilters.extractColumnValues(ImmutableList.of(),
COL_A));
+ }
+
+ @Test
+ public void testRangePredicateReturnsNull()
+ {
+ // col > 'foo' -- a range cannot bound the value set
+ Assert.assertNull(
+ SystemSchemaFilters.extractColumnValues(
+
ImmutableList.of(rexBuilder.makeCall(SqlStdOperatorTable.GREATER_THAN, aRef,
foo)),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testEqualityOnOtherColumnReturnsNull()
+ {
+ Assert.assertNull(
+ SystemSchemaFilters.extractColumnValues(
+ ImmutableList.of(rexBuilder.makeCall(SqlStdOperatorTable.EQUALS,
otherRef, foo)),
+ COL_A
+ )
+ );
+ }
+
+ @Test
+ public void testNonRexCallNodeReturnsNull()
+ {
+ // A bare input ref (not a RexCall) constrains nothing.
+
Assert.assertNull(SystemSchemaFilters.extractColumnValues(ImmutableList.of(aRef),
COL_A));
+ }
+
+ @Test
+ public void testMultiColumnExtractsEachIndependently()
+ {
+ // col_a = 'foo' AND col_b IN ('bar', 'baz')
+ final RexNode filter = rexBuilder.makeCall(
+ SqlStdOperatorTable.AND,
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, aRef, foo),
+ rexBuilder.makeIn(bRef, ImmutableList.of(bar, baz))
+ );
+ final Map<Integer, Set<String>> result =
+ SystemSchemaFilters.extractColumnValues(ImmutableList.of(filter),
COL_A, COL_B);
+ Assert.assertEquals(
+ ImmutableMap.of(
+ COL_A, ImmutableSet.of("foo"),
+ COL_B, ImmutableSet.of("bar", "baz")
+ ),
+ result
+ );
+ }
+
+ @Test
+ public void testMultiColumnOmitsUnconstrainedColumn()
+ {
+ // Only col_a is constrained; col_b must be absent from the map (not
mapped to null).
+ final RexNode filter = rexBuilder.makeCall(SqlStdOperatorTable.EQUALS,
aRef, foo);
+ final Map<Integer, Set<String>> result =
+ SystemSchemaFilters.extractColumnValues(ImmutableList.of(filter),
COL_A, COL_B);
+ Assert.assertEquals(ImmutableMap.of(COL_A, ImmutableSet.of("foo")),
result);
+ Assert.assertFalse(result.containsKey(COL_B));
+ }
+}
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
index ca7410a0180..d956d7bc1c1 100644
---
a/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
+++
b/sql/src/test/java/org/apache/druid/sql/calcite/schema/SystemSchemaTest.java
@@ -36,6 +36,7 @@ import org.apache.calcite.linq4j.QueryProvider;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.Table;
@@ -607,6 +608,88 @@ public class SystemSchemaTest extends CalciteTestBase
Assert.assertEquals(6, propertiesFields.size());
}
+ @Test
+ public void testSegmentsTableGetDataSourceFilter()
+ {
+ final RexBuilder rexBuilder = new RexBuilder(new JavaTypeFactoryImpl());
+ final RexLiteral foo = (RexLiteral) rexBuilder.makeLiteral("foo");
+ final RexLiteral bar = (RexLiteral) rexBuilder.makeLiteral("bar");
+ final RexLiteral baz = (RexLiteral) rexBuilder.makeLiteral("baz");
+ // Match the input-ref type to the literal type so Calcite does not wrap
the literal in a CAST.
+ // "datasource" is column index 1, "size" is column index 4 in
SEGMENTS_SIGNATURE.
+ final RexNode dsRef = rexBuilder.makeInputRef(foo.getType(), 1);
+ final RexNode sizeRef = rexBuilder.makeInputRef(foo.getType(), 4);
+
+ // datasource = 'foo'
+ Assert.assertEquals(
+ ImmutableSet.of("foo"),
+ SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, dsRef, foo)))
+ );
+ // 'foo' = datasource (reversed operands)
+ Assert.assertEquals(
+ ImmutableSet.of("foo"),
+ SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, foo, dsRef)))
+ );
+ // datasource IN ('foo', 'bar')
+ Assert.assertEquals(
+ ImmutableSet.of("foo", "bar"),
+ SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeIn(dsRef, ImmutableList.of(foo, bar))))
+ );
+ // datasource = 'foo' OR datasource = 'bar'
+ Assert.assertEquals(
+ ImmutableSet.of("foo", "bar"),
+ SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeCall(
+ SqlStdOperatorTable.OR,
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, dsRef, foo),
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, dsRef, bar))))
+ );
+ // ANDed conjuncts intersect: IN ('foo','bar') AND IN ('bar','baz') =>
{'bar'}
+ Assert.assertEquals(
+ ImmutableSet.of("bar"),
+ SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeIn(dsRef, ImmutableList.of(foo, bar)),
+ rexBuilder.makeIn(dsRef, ImmutableList.of(bar, baz))))
+ );
+ // No filters => null (full scan retained)
+ Assert.assertNull(SegmentsTable.getDataSourceFilter(ImmutableList.of()));
+ // Range predicate on datasource cannot bound the scan => null
+ Assert.assertNull(SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeCall(SqlStdOperatorTable.GREATER_THAN, dsRef, foo))));
+ // Equality on a non-datasource column => null
+ Assert.assertNull(SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, sizeRef, foo))));
+
+ // Compound predicate passed as a single AND(...) RexCall (as Calcite's
filter-scan rule may do):
+ // datasource = 'foo' AND <non-datasource predicate> => {foo} (the
non-datasource conjunct is ignored)
+ Assert.assertEquals(
+ ImmutableSet.of("foo"),
+ SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeCall(
+ SqlStdOperatorTable.AND,
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, dsRef, foo),
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, sizeRef,
foo))))
+ );
+ // AND of two datasource constraints intersects: IN ('foo','bar') AND IN
('bar','baz') => {'bar'}
+ Assert.assertEquals(
+ ImmutableSet.of("bar"),
+ SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeCall(
+ SqlStdOperatorTable.AND,
+ rexBuilder.makeIn(dsRef, ImmutableList.of(foo, bar)),
+ rexBuilder.makeIn(dsRef, ImmutableList.of(bar, baz)))))
+ );
+ // AND with no datasource conjunct => null (full scan retained)
+ Assert.assertNull(SegmentsTable.getDataSourceFilter(ImmutableList.of(
+ rexBuilder.makeCall(
+ SqlStdOperatorTable.AND,
+ rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, sizeRef, foo),
+ rexBuilder.makeCall(SqlStdOperatorTable.GREATER_THAN, sizeRef,
foo)))));
+ }
+
@Test
public void testSegmentsTable() throws Exception
{
@@ -619,7 +702,7 @@ public class SystemSchemaTest extends CalciteTestBase
new SegmentStatusInCluster(segment2, false, 0, null, false)
));
-
EasyMock.expect(metadataView.getSegments()).andReturn(publishedSegments.iterator()).once();
+
EasyMock.expect(metadataView.getSegments(EasyMock.anyObject())).andReturn(publishedSegments.iterator()).once();
EasyMock.replay(request, responseHolder, responseHandler, metadataView);
DataContext dataContext = createDataContext(Users.SUPER);
@@ -739,7 +822,7 @@ public class SystemSchemaTest extends CalciteTestBase
new SegmentStatusInCluster(segment2, false, 0, null, false)
));
-
EasyMock.expect(metadataView.getSegments()).andReturn(publishedSegments.iterator()).once();
+
EasyMock.expect(metadataView.getSegments(EasyMock.anyObject())).andReturn(publishedSegments.iterator()).once();
EasyMock.replay(request, responseHolder, responseHandler, metadataView);
DataContext dataContext = createDataContext(Users.SUPER);
@@ -1802,6 +1885,46 @@ public class SystemSchemaTest extends CalciteTestBase
EasyMock.verify(druidNodeDiscoveryProvider, httpClient);
}
+ @Test
+ public void testPropertiesTable_filterPushdownInFilter()
+ {
+ SystemServerPropertiesTable propertiesTable = new
SystemServerPropertiesTable(
+ druidNodeDiscoveryProvider,
+ authMapper,
+ httpClient,
+ MAPPER
+ );
+
+ mockAllNodeRolesWithCoordinator(coordinator, coordinator2);
+
+ // server IN ('localhost:8081', 'nonexistent:9999') — only coordinator
(8081) matches, so exactly
+ // one node is fetched. A single HTTP call proves the IN predicate is
pushed down (pre-refactor the
+ // SEARCH form was not extracted and both nodes would have been fetched).
+ HttpResponse resp = new DefaultHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.OK);
+ StringFullResponseHolder holder = new StringFullResponseHolder(resp,
StandardCharsets.UTF_8);
+ holder.addChunk("{\"druid.key\": \"val\"}");
+ EasyMock.expect(
+ httpClient.go(EasyMock.isA(Request.class),
EasyMock.isA(StringFullResponseHandler.class))
+ ).andReturn(Futures.immediateFuture(holder)).once();
+
+ EasyMock.replay(druidNodeDiscoveryProvider, httpClient);
+
+ final RexBuilder rexBuilder = new RexBuilder(new JavaTypeFactoryImpl());
+ final RelDataType rowType = propertiesTable.getRowType(new
JavaTypeFactoryImpl());
+ final RexNode serverIn = rexBuilder.makeIn(
+
rexBuilder.makeInputRef(rowType.getFieldList().get(SERVER_INDEX).getType(),
SERVER_INDEX),
+ ImmutableList.of(rexBuilder.makeLiteral("localhost:8081"),
rexBuilder.makeLiteral("nonexistent:9999"))
+ );
+
+ final List<Object[]> rows =
+ propertiesTable.scan(createDataContext(Users.SUPER),
ImmutableList.of(serverIn), null).toList();
+
+ Assert.assertEquals(1, rows.size());
+ Assert.assertEquals("localhost:8081", rows.get(0)[0]);
+
+ EasyMock.verify(druidNodeDiscoveryProvider, httpClient);
+ }
+
@Test
public void testPropertiesTable_filterPushdownServiceNameAndNonMatching()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
