This is an automated email from the ASF dual-hosted git repository.
clintropolis 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 c612a1e82d5 feat: composite partial load matcher and load spec (#19791)
c612a1e82d5 is described below
commit c612a1e82d5ea351a538432292a763accd171ab5
Author: Clint Wylie <[email protected]>
AuthorDate: Thu Jul 30 22:02:10 2026 -0700
feat: composite partial load matcher and load spec (#19791)
---
.../segment/loading/CompositePartialLoadSpec.java | 217 ++++++++
.../loading/PartialClusterGroupLoadSpec.java | 6 +-
.../druid/segment/loading/PartialLoadSpec.java | 41 +-
.../segment/loading/PartialProjectionLoadSpec.java | 6 +-
.../loading/CompositePartialLoadSpecTest.java | 612 +++++++++++++++++++++
.../apache/druid/guice/PartialLoadSpecModule.java | 10 +-
.../rules/CompositePartialLoadMatcher.java | 198 +++++++
.../coordinator/rules/PartialLoadMatcher.java | 3 +-
...egmentLocalCacheManagerPartialRuleLoadTest.java | 94 ++++
.../rules/CompositePartialLoadMatcherTest.java | 457 +++++++++++++++
...WildcardClusterGroupPartialLoadMatcherTest.java | 10 +
11 files changed, 1630 insertions(+), 24 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/segment/loading/CompositePartialLoadSpec.java
b/processing/src/main/java/org/apache/druid/segment/loading/CompositePartialLoadSpec.java
new file mode 100644
index 00000000000..fcc81dd21b6
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/loading/CompositePartialLoadSpec.java
@@ -0,0 +1,217 @@
+/*
+ * 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.loading;
+
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.segment.file.SegmentFileMetadata;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.utils.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A {@link PartialLoadSpec} that combines several sibling partial-load specs,
selecting the union of every member's
+ * bundles. This lets one partial-load rule mix schemes and lets a single
scheme express selections a single
+ * matcher cannot, such as a disjoint union of include/exclude pattern pairs
in wildcard based matchers.
+ * <p>
+ * Members are held as raw {@link Map} load specs and materialized lazily, the
same way the base class handles
+ * {@link #getDelegate()}. Member load specs deliberately omit the {@link
#DELEGATE_FIELD} field: every member of a
+ * composite describes a selection within the <em>same</em> segment, so
repeating the backend load spec once per
+ * member would bloat every load request and announcement for no benefit. This
spec injects its own
+ * {@link #getDelegate()} into each member as it materializes it, which is why
a member carrying its own
+ * {@code delegate} is rejected at construction (it means the producing
matcher failed to strip it, and silently
+ * overwriting it could mask a real mismatch).
+ * <p>
+ * Composites nest: a member that is itself a {@code partialComposite}
receives the injected delegate and injects it
+ * into its own members in turn.
+ */
+@JsonTypeName(CompositePartialLoadSpec.TYPE)
+public class CompositePartialLoadSpec extends PartialLoadSpec
+{
+ public static final String TYPE = "partialComposite";
+
+ /**
+ * Builds the raw {@link Map} form of a {@link CompositePartialLoadSpec}
request. Used by the coordinator-side
+ * matcher, which doesn't instantiate the typed class because doing so would
require plumbing an
+ * {@link ObjectMapper} through every matcher just to satisfy the
constructor's lazy-materialization suppliers.
+ * <p>
+ * Each entry of {@code members} must be the raw {@link Map} form of some
other {@link PartialLoadSpec}
+ * <em>without</em> its {@link #DELEGATE_FIELD} field; see the class doc.
+ */
+ public static Map<String, Object> wireForm(
+ Map<String, Object> delegate,
+ List<Map<String, Object>> members,
+ String fingerprint
+ )
+ {
+ return Map.of(
+ TYPE_FIELD, TYPE,
+ DELEGATE_FIELD, delegate,
+ "members", members,
+ FINGERPRINT_FIELD, fingerprint
+ );
+ }
+
+ private final List<Map<String, Object>> members;
+ private final Supplier<List<PartialLoadSpec>> materializedMembersSupplier;
+
+ @JsonCreator
+ public CompositePartialLoadSpec(
+ @JsonProperty("delegate") Map<String, Object> delegate,
+ @JsonProperty("members") List<Map<String, Object>> members,
+ @JsonProperty("fingerprint") String fingerprint,
+ @JacksonInject ObjectMapper jsonMapper
+ )
+ {
+ super(delegate, fingerprint, jsonMapper);
+ Preconditions.checkArgument(
+ !CollectionUtils.isNullOrEmpty(members),
+ "members must not be null or empty"
+ );
+ final List<Map<String, Object>> copied = new ArrayList<>(members.size());
+ for (int i = 0; i < members.size(); i++) {
+ copied.add(validateMember(members.get(i), i));
+ }
+ this.members = List.copyOf(copied);
+ this.materializedMembersSupplier = Suppliers.memoize(() ->
materializeMembers(jsonMapper));
+ }
+
+ @JsonProperty
+ public List<Map<String, Object>> getMembers()
+ {
+ return members;
+ }
+
+ /**
+ * The union of every member's selected bundles, in member order and then
member-internal order. Duplicates are
+ * dropped: two members of the same scheme can legitimately select
overlapping bundles (e.g. two cluster-group
+ * selections that share a group), and the caller treats the result as a set.
+ * <p>
+ * The base bundle needs no special handling here —
+ * {@code PartialSegmentMetadataCacheEntry#bundlesInMountOrder} expands each
selected bundle's inferred
+ * dependencies, which pins {@code __base} exactly once regardless of how
many members asked for something that
+ * depends on it.
+ * <p>
+ * Returns an empty list only when every member selects nothing (the
"sibling-empty" case propagated through
+ * composition).
+ */
+ @Override
+ public List<String> getSelectedBundleNames(DataSegment segment,
SegmentFileMetadata metadata)
+ {
+ final LinkedHashSet<String> union = new LinkedHashSet<>();
+ for (PartialLoadSpec member : materializedMembersSupplier.get()) {
+ union.addAll(member.getSelectedBundleNames(segment, metadata));
+ }
+ return List.copyOf(union);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ CompositePartialLoadSpec that = (CompositePartialLoadSpec) o;
+ return Objects.equals(getDelegate(), that.getDelegate())
+ && Objects.equals(members, that.members)
+ && Objects.equals(getFingerprint(), that.getFingerprint());
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(getDelegate(), members, getFingerprint());
+ }
+
+ @Override
+ public String toString()
+ {
+ return "CompositePartialLoadSpec{" +
+ "delegate=" + getDelegate() +
+ ", members=" + members +
+ ", fingerprint=" + getFingerprint() +
+ '}';
+ }
+
+ private List<PartialLoadSpec> materializeMembers(ObjectMapper jsonMapper)
+ {
+ final List<PartialLoadSpec> materialized = new ArrayList<>(members.size());
+ for (Map<String, Object> member : members) {
+ // Splice this composite's delegate into the member before materializing
it: members omit it on the wire, but
+ // the PartialLoadSpec constructor requires one.
+ final Map<String, Object> withDelegate = new LinkedHashMap<>(member);
+ withDelegate.put(DELEGATE_FIELD, getDelegate());
+ final LoadSpec memberSpec = jsonMapper.convertValue(withDelegate,
LoadSpec.class);
+ if (!(memberSpec instanceof PartialLoadSpec partialMember)) {
+ throw DruidException.defensive(
+ "Composite partial load spec member of type[%s] materialized to
non-partial type[%s]",
+ member.get(TYPE_FIELD),
+ memberSpec.getClass().getSimpleName()
+ );
+ }
+ materialized.add(partialMember);
+ }
+ return materialized;
+ }
+
+ /**
+ * Validates and defensively copies one member load spec. A member must
carry a partial-load {@link #TYPE_FIELD} and
+ * must not carry a {@link #DELEGATE_FIELD} of its own; see the class doc
for why.
+ */
+ private static Map<String, Object> validateMember(Map<String, Object>
member, int index)
+ {
+ if (member == null || member.isEmpty()) {
+ throw DruidException.defensive("members[%s] must not be null or empty",
index);
+ }
+ if (!hasPartialTypePrefix(member)) {
+ throw DruidException.defensive(
+ "members[%s] must be a partial load spec with a type starting with
[%s], got type[%s]",
+ index,
+ TYPE_PREFIX,
+ member.get(TYPE_FIELD)
+ );
+ }
+ if (member.containsKey(DELEGATE_FIELD)) {
+ throw DruidException.defensive(
+ "members[%s] of type[%s] must not carry its own [%s]; the composite
supplies it",
+ index,
+ member.get(TYPE_FIELD),
+ DELEGATE_FIELD
+ );
+ }
+ return Map.copyOf(member);
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/segment/loading/PartialClusterGroupLoadSpec.java
b/processing/src/main/java/org/apache/druid/segment/loading/PartialClusterGroupLoadSpec.java
index cbc27ad3cf9..2764a100f5c 100644
---
a/processing/src/main/java/org/apache/druid/segment/loading/PartialClusterGroupLoadSpec.java
+++
b/processing/src/main/java/org/apache/druid/segment/loading/PartialClusterGroupLoadSpec.java
@@ -62,10 +62,10 @@ public class PartialClusterGroupLoadSpec extends
PartialLoadSpec
)
{
return Map.of(
- "type", TYPE,
- "delegate", delegate,
+ TYPE_FIELD, TYPE,
+ DELEGATE_FIELD, delegate,
"clusterGroupIndices", clusterGroupIndices,
- "fingerprint", fingerprint
+ FINGERPRINT_FIELD, fingerprint
);
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/loading/PartialLoadSpec.java
b/processing/src/main/java/org/apache/druid/segment/loading/PartialLoadSpec.java
index c9b0e82acdd..0e3174647c3 100644
---
a/processing/src/main/java/org/apache/druid/segment/loading/PartialLoadSpec.java
+++
b/processing/src/main/java/org/apache/druid/segment/loading/PartialLoadSpec.java
@@ -64,32 +64,47 @@ public abstract class PartialLoadSpec implements LoadSpec
*/
public static final String TYPE_PREFIX = "partial";
+ /**
+ * Wire field carrying the Jackson type discriminator. Named here (rather
than only inline) because code that
+ * inspects or rewrites raw {@link Map}-form load specs needs to agree with
the {@code @JsonTypeInfo} property name
+ * on {@link LoadSpec}.
+ */
+ public static final String TYPE_FIELD = "type";
+
+ /**
+ * Wire field carrying the raw inner load spec, provided by {@link
#getDelegate()}.
+ */
+ public static final String DELEGATE_FIELD = "delegate";
+
+ /**
+ * Wire field carrying the partial-load request fingerprint, provided by
{@link #getFingerprint()}.
+ */
+ public static final String FINGERPRINT_FIELD = "fingerprint";
+
/**
* Returns {@code true} if {@code loadSpec} matches the shape of the {@link
PartialLoadSpec} subtype.
- * Convention-based detection (no subtype allowlist): the {@code type} field
must be a {@link String} starting with
- * {@link #TYPE_PREFIX}, the {@code fingerprint} field must be a {@link
String}, and the {@code delegate} field
- * must be a {@link Map}. These properties are enforced by this base class's
{@code @JsonProperty} getters, so any
- * subtype satisfies them automatically.
+ * Convention-based detection (no subtype allowlist): the {@link
#TYPE_FIELD} field must be a {@link String}
+ * starting with {@link #TYPE_PREFIX}, the {@link #FINGERPRINT_FIELD} field
must be a {@link String}, and the
+ * {@link #DELEGATE_FIELD} field must be a {@link Map}. These properties are
enforced by this base class's
+ * {@code @JsonProperty} getters, so any subtype satisfies them
automatically.
*/
public static boolean detectPartialLoadSpec(@Nullable Map<String, Object>
loadSpec)
{
- return loadSpec != null
- && loadSpec.get("type") instanceof String typeString
- && typeString.startsWith(TYPE_PREFIX)
- && loadSpec.get("fingerprint") instanceof String
- && loadSpec.get("delegate") instanceof Map;
+ return hasPartialTypePrefix(loadSpec)
+ && loadSpec.get(FINGERPRINT_FIELD) instanceof String
+ && loadSpec.get(DELEGATE_FIELD) instanceof Map;
}
/**
- * Returns {@code true} if {@code loadSpec}'s {@code type} field claims
partial-load semantics (starts with
+ * Returns {@code true} if {@code loadSpec}'s {@link #TYPE_FIELD} field
claims partial-load semantics (starts with
* {@link #TYPE_PREFIX}), regardless of whether the remaining wire form is
well-formed. Useful when callers want
- * to distinguish "not a partial-load wrapper" from "claims to be partial
but the {@code fingerprint} or
- * {@code delegate} fields are missing or malformed" — the latter typically
indicates a bug worth logging.
+ * to distinguish "not a partial-load wrapper" from "claims to be partial
but the {@link #FINGERPRINT_FIELD} or
+ * {@link #DELEGATE_FIELD} fields are missing or malformed" — the latter
typically indicates a bug worth logging.
*/
public static boolean hasPartialTypePrefix(@Nullable Map<String, Object>
loadSpec)
{
return loadSpec != null
- && loadSpec.get("type") instanceof String typeString
+ && loadSpec.get(TYPE_FIELD) instanceof String typeString
&& typeString.startsWith(TYPE_PREFIX);
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/loading/PartialProjectionLoadSpec.java
b/processing/src/main/java/org/apache/druid/segment/loading/PartialProjectionLoadSpec.java
index c063f33db8e..61bc550102d 100644
---
a/processing/src/main/java/org/apache/druid/segment/loading/PartialProjectionLoadSpec.java
+++
b/processing/src/main/java/org/apache/druid/segment/loading/PartialProjectionLoadSpec.java
@@ -59,10 +59,10 @@ public class PartialProjectionLoadSpec extends
PartialLoadSpec
)
{
return Map.of(
- "type", TYPE,
- "delegate", delegate,
+ TYPE_FIELD, TYPE,
+ DELEGATE_FIELD, delegate,
"projections", projections,
- "fingerprint", fingerprint
+ FINGERPRINT_FIELD, fingerprint
);
}
diff --git
a/processing/src/test/java/org/apache/druid/segment/loading/CompositePartialLoadSpecTest.java
b/processing/src/test/java/org/apache/druid/segment/loading/CompositePartialLoadSpecTest.java
new file mode 100644
index 00000000000..7dbb6b158de
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/segment/loading/CompositePartialLoadSpecTest.java
@@ -0,0 +1,612 @@
+/*
+ * 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.loading;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.InjectableValues;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.file.SegmentFileMetadata;
+import org.apache.druid.segment.projections.AggregateProjectionSchema;
+import
org.apache.druid.segment.projections.ClusteredValueGroupsBaseTableSchema;
+import org.apache.druid.segment.projections.ClusteringDictionaries;
+import org.apache.druid.segment.projections.ProjectionMetadata;
+import org.apache.druid.segment.projections.Projections;
+import org.apache.druid.segment.projections.TableClusterGroupSpec;
+import org.apache.druid.segment.projections.TableProjectionSchema;
+import org.apache.druid.timeline.ClusterGroupTuples;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.partition.NumberedShardSpec;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class CompositePartialLoadSpecTest
+{
+ private static final Map<String, Object> DELEGATE = ImmutableMap.of(
+ "type", "stub",
+ "path", "/var/druid/segments/foo"
+ );
+ private static final String FINGERPRINT = "v1:abcdef0123456789";
+
+ /**
+ * A {@code partialProjection} member load spec, i.e. {@link
PartialProjectionLoadSpec#wireForm} minus its
+ * {@code delegate} — which is what {@code CompositePartialLoadMatcher}
emits and what the composite injects into.
+ */
+ private static Map<String, Object> projectionMember(List<String>
projections, String fingerprint)
+ {
+ return member(PartialProjectionLoadSpec.wireForm(DELEGATE, projections,
fingerprint));
+ }
+
+ private static Map<String, Object> clusterGroupMember(List<Integer> indices,
String fingerprint)
+ {
+ return member(PartialClusterGroupLoadSpec.wireForm(DELEGATE, indices,
fingerprint));
+ }
+
+ private static Map<String, Object> member(Map<String, Object> wireForm)
+ {
+ final Map<String, Object> stripped = new HashMap<>(wireForm);
+ stripped.remove(PartialLoadSpec.DELEGATE_FIELD);
+ return stripped;
+ }
+
+ private static ObjectMapper configuredMapper()
+ {
+ final ObjectMapper m = new DefaultObjectMapper();
+ final SimpleModule module = new SimpleModule();
+ module.registerSubtypes(
+ CompositePartialLoadSpec.class,
+ PartialProjectionLoadSpec.class,
+ PartialClusterGroupLoadSpec.class,
+ StubLoadSpec.class
+ );
+ m.registerModule(module);
+ m.setInjectableValues(new
InjectableValues.Std().addValue(ObjectMapper.class, m));
+ return m;
+ }
+
+ private final ObjectMapper jsonMapper = configuredMapper();
+
+ @Test
+ void testJsonRoundTrip() throws Exception
+ {
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(
+ projectionMember(List.of("user_hourly"), "v1:aaaaaaaaaaaaaaaa"),
+ clusterGroupMember(List.of(0, 2), "v1:bbbbbbbbbbbbbbbb")
+ ),
+ FINGERPRINT,
+ jsonMapper
+ );
+ String json = jsonMapper.writeValueAsString(spec);
+ LoadSpec reread = jsonMapper.readValue(json, LoadSpec.class);
+ Assertions.assertInstanceOf(CompositePartialLoadSpec.class, reread);
+ Assertions.assertEquals(spec, reread);
+ }
+
+ @Test
+ void testWireFormHasPartialCompositeType() throws Exception
+ {
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(
+ projectionMember(List.of("user_hourly"), "v1:aaaaaaaaaaaaaaaa"),
+ clusterGroupMember(List.of(0), "v1:bbbbbbbbbbbbbbbb")
+ ),
+ FINGERPRINT,
+ jsonMapper
+ );
+ Map<String, Object> wireForm = jsonMapper.readValue(
+ jsonMapper.writeValueAsString(spec),
+ new TypeReference<>()
+ {
+ }
+ );
+ Assertions.assertEquals("partialComposite", wireForm.get("type"));
+ Assertions.assertEquals(DELEGATE, wireForm.get("delegate"));
+ Assertions.assertEquals(FINGERPRINT, wireForm.get("fingerprint"));
+ Assertions.assertEquals(
+ List.of(
+ projectionMember(List.of("user_hourly"), "v1:aaaaaaaaaaaaaaaa"),
+ clusterGroupMember(List.of(0), "v1:bbbbbbbbbbbbbbbb")
+ ),
+ wireForm.get("members")
+ );
+ }
+
+ @Test
+ void testMembersOnWireCarryNoDelegate() throws Exception
+ {
+ // The composite carries the backend load spec exactly once, at the top
level.
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(
+ projectionMember(List.of("user_hourly"), "v1:aaaaaaaaaaaaaaaa"),
+ clusterGroupMember(List.of(0), "v1:bbbbbbbbbbbbbbbb")
+ ),
+ FINGERPRINT,
+ jsonMapper
+ );
+ Map<String, Object> wireForm = jsonMapper.readValue(
+ jsonMapper.writeValueAsString(spec),
+ new TypeReference<>()
+ {
+ }
+ );
+ @SuppressWarnings("unchecked")
+ final List<Map<String, Object>> members = (List<Map<String, Object>>)
wireForm.get("members");
+ for (Map<String, Object> m : members) {
+ Assertions.assertFalse(m.containsKey("delegate"), "member should not
carry a delegate: " + m);
+ }
+ }
+
+ @Test
+ void testDelegateIsInjectedIntoMembersOnMaterialization()
+ {
+ // Members omit the delegate on the wire, so the only way a member's own
loadSegment can work is if the composite
+ // spliced its delegate in. Materialization happens lazily inside
getSelectedBundleNames.
+ final SegmentFileMetadata metadata =
projectionMetadata(List.of("user_hourly"));
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(
+ projectionMember(List.of("user_hourly"), "v1:aaaaaaaaaaaaaaaa"),
+ projectionMember(List.of("user_hourly"), "v1:cccccccccccccccc")
+ ),
+ FINGERPRINT,
+ jsonMapper
+ );
+ Assertions.assertEquals(
+ List.of("user_hourly"),
+ spec.getSelectedBundleNames(unclusteredSegment(), metadata)
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesUnionsAcrossSchemes()
+ {
+ final SegmentFileMetadata metadata = clusteredMetadata(
+ List.of(
+ new TableClusterGroupSpec(List.of(0), 10),
+ new TableClusterGroupSpec(List.of(1), 20),
+ new TableClusterGroupSpec(List.of(2), 30)
+ ),
+ List.of("user_hourly")
+ );
+ final DataSegment segment = clusteredSegment(
+ List.of(List.of("acme"), List.of("globex"), List.of("initech"))
+ );
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(
+ projectionMember(List.of("user_hourly"), "v1:aaaaaaaaaaaaaaaa"),
+ clusterGroupMember(List.of(0, 2), "v1:bbbbbbbbbbbbbbbb")
+ ),
+ FINGERPRINT,
+ jsonMapper
+ );
+ Assertions.assertEquals(
+ List.of(
+ "user_hourly",
+ Projections.getClusterGroupBundleName(List.of(0)),
+ Projections.getClusterGroupBundleName(List.of(2))
+ ),
+ spec.getSelectedBundleNames(segment, metadata)
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesDedupesOverlappingMembers()
+ {
+ // Two same-scheme members whose selections overlap: the union drops the
duplicate but keeps first-seen order.
+ final SegmentFileMetadata metadata = clusteredMetadata(
+ List.of(
+ new TableClusterGroupSpec(List.of(0), 10),
+ new TableClusterGroupSpec(List.of(1), 20),
+ new TableClusterGroupSpec(List.of(2), 30)
+ ),
+ null
+ );
+ final DataSegment segment = clusteredSegment(
+ List.of(List.of("acme"), List.of("globex"), List.of("initech"))
+ );
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(
+ clusterGroupMember(List.of(0, 1), "v1:aaaaaaaaaaaaaaaa"),
+ clusterGroupMember(List.of(1, 2), "v1:bbbbbbbbbbbbbbbb")
+ ),
+ FINGERPRINT,
+ jsonMapper
+ );
+ Assertions.assertEquals(
+ List.of(
+ Projections.getClusterGroupBundleName(List.of(0)),
+ Projections.getClusterGroupBundleName(List.of(1)),
+ Projections.getClusterGroupBundleName(List.of(2))
+ ),
+ spec.getSelectedBundleNames(segment, metadata)
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesAllEmptyMembersReturnsEmpty()
+ {
+ // Sibling-empty propagated through composition: every member selected
nothing.
+ final SegmentFileMetadata metadata = clusteredMetadata(
+ List.of(new TableClusterGroupSpec(List.of(0), 1)),
+ null
+ );
+ final DataSegment segment = clusteredSegment(List.of(List.of("acme")));
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(
+ clusterGroupMember(List.of(), "v1:partial-empty"),
+ clusterGroupMember(List.of(), "v1:partial-empty")
+ ),
+ FINGERPRINT,
+ jsonMapper
+ );
+ Assertions.assertEquals(List.of(), spec.getSelectedBundleNames(segment,
metadata));
+ }
+
+ @Test
+ void testNestedCompositeInjectsDelegateRecursively()
+ {
+ final SegmentFileMetadata metadata = clusteredMetadata(
+ List.of(
+ new TableClusterGroupSpec(List.of(0), 10),
+ new TableClusterGroupSpec(List.of(1), 20)
+ ),
+ List.of("user_hourly")
+ );
+ final DataSegment segment = clusteredSegment(List.of(List.of("acme"),
List.of("globex")));
+ final Map<String, Object> nested = member(
+ CompositePartialLoadSpec.wireForm(
+ DELEGATE,
+ List.of(
+ clusterGroupMember(List.of(0), "v1:aaaaaaaaaaaaaaaa"),
+ clusterGroupMember(List.of(1), "v1:bbbbbbbbbbbbbbbb")
+ ),
+ "v1:dddddddddddddddd"
+ )
+ );
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(projectionMember(List.of("user_hourly"),
"v1:cccccccccccccccc"), nested),
+ FINGERPRINT,
+ jsonMapper
+ );
+ Assertions.assertEquals(
+ List.of(
+ "user_hourly",
+ Projections.getClusterGroupBundleName(List.of(0)),
+ Projections.getClusterGroupBundleName(List.of(1))
+ ),
+ spec.getSelectedBundleNames(segment, metadata)
+ );
+ }
+
+ @Test
+ void testMemberDefectPropagates()
+ {
+ // A member's own defensive tripwire is not swallowed by the union.
+ final SegmentFileMetadata metadata =
projectionMetadata(List.of("user_hourly"));
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(
+ projectionMember(List.of("user_hourly"), "v1:aaaaaaaaaaaaaaaa"),
+ projectionMember(List.of("nonexistent"), "v1:bbbbbbbbbbbbbbbb")
+ ),
+ FINGERPRINT,
+ jsonMapper
+ );
+ final DruidException thrown = Assertions.assertThrows(
+ DruidException.class,
+ () -> spec.getSelectedBundleNames(unclusteredSegment(), metadata)
+ );
+ Assertions.assertTrue(
+ thrown.getMessage().contains("does not contain
projection[nonexistent]"),
+ "unexpected message: " + thrown.getMessage()
+ );
+ }
+
+ @Test
+ void testLoadSegmentDelegatesToInner() throws Exception
+ {
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(projectionMember(List.of("user_hourly"),
"v1:aaaaaaaaaaaaaaaa")),
+ FINGERPRINT,
+ jsonMapper
+ );
+ StubLoadSpec.LOAD_CALLS.set(0);
+ LoadSpec.LoadSpecResult result = spec.loadSegment(new File("/tmp/dest"));
+ Assertions.assertEquals(1, StubLoadSpec.LOAD_CALLS.get());
+ Assertions.assertEquals(42L, result.getSize());
+ }
+
+ @Test
+ void testOpenRangeReaderDelegatesToInner() throws Exception
+ {
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(projectionMember(List.of("user_hourly"),
"v1:aaaaaaaaaaaaaaaa")),
+ FINGERPRINT,
+ jsonMapper
+ );
+ StubLoadSpec.RANGE_CALLS.set(0);
+ SegmentRangeReader reader = spec.openRangeReader();
+ Assertions.assertNotNull(reader);
+ Assertions.assertEquals(1, StubLoadSpec.RANGE_CALLS.get());
+ }
+
+ @Test
+ void testOpenRangeReaderReturnsNullWhenInnerDoesNotSupport() throws Exception
+ {
+ CompositePartialLoadSpec spec = new CompositePartialLoadSpec(
+ ImmutableMap.of("type", "stub", "path", "/", "supportsRange", false),
+ List.of(projectionMember(List.of("user_hourly"),
"v1:aaaaaaaaaaaaaaaa")),
+ FINGERPRINT,
+ jsonMapper
+ );
+ Assertions.assertNull(spec.openRangeReader());
+ }
+
+ @Test
+ void testRejectsNullDelegate()
+ {
+ Assertions.assertThrows(
+ NullPointerException.class,
+ () -> new CompositePartialLoadSpec(
+ null,
+ List.of(projectionMember(List.of("a"), "v1:x")),
+ "v1:x",
+ jsonMapper
+ )
+ );
+ }
+
+ @Test
+ void testRejectsNullFingerprint()
+ {
+ Assertions.assertThrows(
+ NullPointerException.class,
+ () -> new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(projectionMember(List.of("a"), "v1:x")),
+ null,
+ jsonMapper
+ )
+ );
+ }
+
+ @Test
+ void testRejectsNullMembers()
+ {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> new CompositePartialLoadSpec(DELEGATE, null, "v1:x", jsonMapper)
+ );
+ }
+
+ @Test
+ void testRejectsEmptyMembers()
+ {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> new CompositePartialLoadSpec(DELEGATE, List.of(), "v1:x",
jsonMapper)
+ );
+ }
+
+ @Test
+ void testRejectsMemberCarryingDelegate()
+ {
+ // Unstripped member: the composite owns the delegate, so silently
overwriting it could mask a real mismatch.
+ final DruidException thrown = Assertions.assertThrows(
+ DruidException.class,
+ () -> new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(PartialProjectionLoadSpec.wireForm(DELEGATE, List.of("a"),
"v1:x")),
+ "v1:x",
+ jsonMapper
+ )
+ );
+ Assertions.assertTrue(
+ thrown.getMessage().contains("must not carry its own [delegate]"),
+ "unexpected message: " + thrown.getMessage()
+ );
+ }
+
+ @Test
+ void testRejectsMemberWithNonPartialType()
+ {
+ final DruidException thrown = Assertions.assertThrows(
+ DruidException.class,
+ () -> new CompositePartialLoadSpec(
+ DELEGATE,
+ List.of(Map.of("type", "stub", "path", "/")),
+ "v1:x",
+ jsonMapper
+ )
+ );
+ Assertions.assertTrue(
+ thrown.getMessage().contains("must be a partial load spec with a type
starting with"),
+ "unexpected message: " + thrown.getMessage()
+ );
+ }
+
+ private static final RowSignature CLUSTERING_TENANT = RowSignature.builder()
+
.add("tenant", ColumnType.STRING)
+ .build();
+
+ private static SegmentFileMetadata clusteredMetadata(
+ List<TableClusterGroupSpec> groups,
+ @Nullable List<String> projections
+ )
+ {
+ final ClusteredValueGroupsBaseTableSchema baseSchema = new
ClusteredValueGroupsBaseTableSchema(
+ VirtualColumns.EMPTY,
+ List.of(ColumnHolder.TIME_COLUMN_NAME, "tenant", "metric"),
+ List.of(OrderBy.ascending("tenant"),
OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME)),
+ CLUSTERING_TENANT,
+ null,
+ new ClusteringDictionaries(List.of("acme", "globex", "initech"), null,
null, null),
+ groups
+ );
+ final int numRows =
groups.stream().mapToInt(TableClusterGroupSpec::getNumRows).sum();
+ final List<ProjectionMetadata> projectionMetadata = new ArrayList<>();
+ projectionMetadata.add(new ProjectionMetadata(numRows, baseSchema));
+ if (projections != null) {
+ for (String name : projections) {
+ projectionMetadata.add(new ProjectionMetadata(numRows,
projectionSchemaNamed(name)));
+ }
+ }
+ return new SegmentFileMetadata(List.of(), Map.of(), null, null, null,
projectionMetadata, null);
+ }
+
+ private static SegmentFileMetadata projectionMetadata(List<String>
projections)
+ {
+ final List<ProjectionMetadata> projectionMetadata = new ArrayList<>();
+ projectionMetadata.add(
+ new ProjectionMetadata(
+ 100,
+ new TableProjectionSchema(
+ VirtualColumns.EMPTY,
+ List.of(ColumnHolder.TIME_COLUMN_NAME, "tenant"),
+ null,
+ List.of(OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME))
+ )
+ )
+ );
+ for (String name : projections) {
+ projectionMetadata.add(new ProjectionMetadata(10,
projectionSchemaNamed(name)));
+ }
+ return new SegmentFileMetadata(List.of(), Map.of(), null, null, null,
projectionMetadata, null);
+ }
+
+ private static AggregateProjectionSchema projectionSchemaNamed(String name)
+ {
+ return new AggregateProjectionSchema(
+ name,
+ null,
+ null,
+ VirtualColumns.EMPTY,
+ List.of("tenant"),
+ new AggregatorFactory[]{new CountAggregatorFactory("cnt")},
+ List.of(OrderBy.ascending("tenant"))
+ );
+ }
+
+ private static DataSegment clusteredSegment(List<List<Object>> tuples)
+ {
+ return DataSegment.builder(
+ SegmentId.of("ds", Intervals.ETERNITY, "v1", new
NumberedShardSpec(0, 1))
+ )
+ .size(0)
+ .clusterGroups(new ClusterGroupTuples(CLUSTERING_TENANT,
tuples))
+ .build();
+ }
+
+ private static DataSegment unclusteredSegment()
+ {
+ return DataSegment.builder(
+ SegmentId.of("ds", Intervals.ETERNITY, "v1", new
NumberedShardSpec(0, 1))
+ )
+ .size(0)
+ .build();
+ }
+
+ /**
+ * Stub LoadSpec used to verify delegation. Uses the same JSON
"type"=="stub" key as the test {@link #DELEGATE}.
+ */
+ @JsonTypeName("stub")
+ public static class StubLoadSpec implements LoadSpec
+ {
+ static final AtomicInteger LOAD_CALLS = new AtomicInteger(0);
+ static final AtomicInteger RANGE_CALLS = new AtomicInteger(0);
+
+ private final String path;
+ private final boolean supportsRange;
+
+ @JsonCreator
+ public StubLoadSpec(
+ @JsonProperty("path") String path,
+ @JsonProperty("supportsRange") @Nullable Boolean supportsRange
+ )
+ {
+ this.path = path;
+ this.supportsRange = supportsRange == null || supportsRange;
+ }
+
+ @JsonProperty
+ public String getPath()
+ {
+ return path;
+ }
+
+ @JsonProperty
+ public boolean isSupportsRange()
+ {
+ return supportsRange;
+ }
+
+ @Override
+ public LoadSpecResult loadSegment(File destDir)
+ {
+ LOAD_CALLS.incrementAndGet();
+ return new LoadSpecResult(42L);
+ }
+
+ @Override
+ @Nullable
+ public SegmentRangeReader openRangeReader()
+ {
+ if (!supportsRange) {
+ return null;
+ }
+ RANGE_CALLS.incrementAndGet();
+ return (filename, offset, length) -> new ByteArrayInputStream(new
byte[0]);
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/guice/PartialLoadSpecModule.java
b/server/src/main/java/org/apache/druid/guice/PartialLoadSpecModule.java
index d60ca44c12f..e3b615738fd 100644
--- a/server/src/main/java/org/apache/druid/guice/PartialLoadSpecModule.java
+++ b/server/src/main/java/org/apache/druid/guice/PartialLoadSpecModule.java
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.databind.Module;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.google.inject.Binder;
import org.apache.druid.initialization.DruidModule;
+import org.apache.druid.segment.loading.CompositePartialLoadSpec;
import org.apache.druid.segment.loading.LoadSpec;
import org.apache.druid.segment.loading.PartialClusterGroupLoadSpec;
import org.apache.druid.segment.loading.PartialProjectionLoadSpec;
@@ -30,9 +31,9 @@ import
org.apache.druid.segment.loading.PartialProjectionLoadSpec;
import java.util.List;
/**
- * Registers {@link PartialProjectionLoadSpec} and {@link
PartialClusterGroupLoadSpec} as {@link LoadSpec} subtypes
- * for serde of partial load rules. This module is added to the always-loaded
core list so they are available
- * alongside any other deep-storage load spec modules.
+ * Registers {@link PartialProjectionLoadSpec}, {@link
PartialClusterGroupLoadSpec} and
+ * {@link CompositePartialLoadSpec} as {@link LoadSpec} subtypes for serde of
partial load rules. This module is added
+ * to the always-loaded core list so they are available alongside any other
deep-storage load spec modules.
*/
public class PartialLoadSpecModule implements DruidModule
{
@@ -48,7 +49,8 @@ public class PartialLoadSpecModule implements DruidModule
return List.of(
new SimpleModule().registerSubtypes(
PartialProjectionLoadSpec.class,
- PartialClusterGroupLoadSpec.class
+ PartialClusterGroupLoadSpec.class,
+ CompositePartialLoadSpec.class
)
);
}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/rules/CompositePartialLoadMatcher.java
b/server/src/main/java/org/apache/druid/server/coordinator/rules/CompositePartialLoadMatcher.java
new file mode 100644
index 00000000000..de571e25c58
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/server/coordinator/rules/CompositePartialLoadMatcher.java
@@ -0,0 +1,198 @@
+/*
+ * 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.server.coordinator.rules;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
+import com.google.common.io.BaseEncoding;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.segment.loading.CompositePartialLoadSpec;
+import org.apache.druid.segment.loading.PartialLoadSpec;
+import org.apache.druid.timeline.DataSegment;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Combines several {@link PartialLoadMatcher}s so that one {@link
PartialLoadRule} can contribute more than one kind
+ * of partial load to a segment. The resolved selection is the <em>union</em>
of what the members select: partial-load
+ * a segment's projections and a subset of its cluster groups together, or
express a single scheme's selection that
+ * its own matcher cannot, such as a disjoint union of include/exclude pattern
pairs over the same clustering columns.
+ * <p>
+ * Composition is a union, not an ordered fallback. Members are not consulted
in priority order and no member can
+ * shadow another; every member that resolves contributes, and the
historical-side {@link CompositePartialLoadSpec}
+ * takes the union of their bundles.
+ * <p>
+ * <b>A member that does not apply vetoes the whole composite.</b> If any
member returns {@code null}, it understands
+ * neither the segment's shape nor how to express a selection for it, e.g. a
cluster-group matcher facing a segment
+ * that isn't clustered, or a matcher type this Druid version doesn't
recognize (see {@link UnknownPartialLoadMatcher})
+ * this matcher returns {@code null} too and the rule's {@link
CannotMatchBehavior} decides for the whole segment.
+ * Skipping such a member instead would silently narrow the load: a composite
whose cluster-group member went opaque
+ * would announce a segment holding only its projections and none of its rows,
and queries against it would quietly
+ * return nothing.
+ */
+public class CompositePartialLoadMatcher implements PartialLoadMatcher
+{
+ public static final String TYPE = "composite";
+
+ static final String FINGERPRINT_VERSION = "v1";
+
+ private final List<PartialLoadMatcher> matchers;
+
+ @JsonCreator
+ public CompositePartialLoadMatcher(@JsonProperty("matchers")
List<PartialLoadMatcher> matchers)
+ {
+ if (matchers == null || matchers.isEmpty()) {
+ throw InvalidInput.exception("matchers must not be null or empty for
composite matcher");
+ }
+ for (int i = 0; i < matchers.size(); i++) {
+ if (matchers.get(i) == null) {
+ throw InvalidInput.exception("matchers[%s] must not be null for
composite matcher", i);
+ }
+ }
+ this.matchers = List.copyOf(matchers);
+ }
+
+ @JsonProperty
+ public List<PartialLoadMatcher> getMatchers()
+ {
+ return matchers;
+ }
+
+ @Override
+ @Nullable
+ public MatchResult match(DataSegment segment, Map<String, Object>
baseLoadSpec)
+ {
+ // Members get the real base load spec, not a stub: a member may
legitimately inspect it, and its delegate is
+ // stripped afterward (see toMember) rather than withheld up front.
+ final List<MatchResult> results = new ArrayList<>(matchers.size());
+ for (PartialLoadMatcher matcher : matchers) {
+ final MatchResult result = matcher.match(segment, baseLoadSpec);
+ if (result == null) {
+ return null;
+ }
+ results.add(result);
+ }
+
+ if (results.size() == 1) {
+ // A single-member composite is exactly its member. Emitting the
member's load spec verbatim keeps a rule that
+ // was wrapped in a composite fingerprint-identical to the same rule
with the bare matcher, so wrapping does not
+ // re-fingerprint (and thus re-apply) every segment the rule covers.
+ return results.getFirst();
+ }
+
+ final List<Map<String, Object>> members = new ArrayList<>(results.size());
+ for (MatchResult result : results) {
+ members.add(toMember(result));
+ }
+ final String fingerprint = computeFingerprint(results);
+ return new MatchResult(
+ CompositePartialLoadSpec.wireForm(baseLoadSpec, members, fingerprint),
+ fingerprint
+ );
+ }
+
+ /**
+ * Converts a member's {@link MatchResult} into the composite's member load
spec by dropping the
+ * {@link PartialLoadSpec#DELEGATE_FIELD} field. Every member of a composite
describes a selection within the same
+ * segment, so the composite carries the backend load spec once at the top
level and re-injects it when it
+ * materializes each member; repeating it per member would bloat every load
request and announcement.
+ */
+ private static Map<String, Object> toMember(MatchResult result)
+ {
+ final Map<String, Object> member = new
LinkedHashMap<>(result.wrappedLoadSpec());
+ member.remove(PartialLoadSpec.DELEGATE_FIELD);
+ return member;
+ }
+
+ /**
+ * Fingerprints the composite over its members' {@code (type, fingerprint)}
pairs. Each member fingerprint already
+ * identifies that member's resolved selection within its own scheme, and
the type disambiguates equal fingerprints
+ * produced by different schemes.
+ * <p>
+ * Pairs are sorted, so reordering a rule's {@code matchers} does not change
the fingerprint — the resolved selection
+ * is a set union and therefore order-independent, and the cascade should
not thrash on equivalent rule rewordings.
+ * <p>
+ * When every member resolved to an empty selection the composite reports
{@link #EMPTY_LOAD_FINGERPRINT}, carrying
+ * the empty-load contract through composition rather than minting a
distinct fingerprint for a load that puts no
+ * scheme-specific content on the historical.
+ * <p>
+ * Note that a composite of two same-scheme members does not fingerprint the
same as a single matcher that resolved
+ * to the same union. The coordinator only compares a segment's fingerprint
against the rule that requested it, so
+ * the difference costs at most one cheap rule re-apply on the historical,
never a re-download.
+ */
+ private static String computeFingerprint(List<MatchResult> results)
+ {
+ boolean allEmpty = true;
+ final List<String> pairs = new ArrayList<>(results.size());
+ for (MatchResult result : results) {
+ allEmpty = allEmpty &&
EMPTY_LOAD_FINGERPRINT.equals(result.fingerprint());
+ pairs.add(memberType(result) + '\0' + result.fingerprint());
+ }
+ if (allEmpty) {
+ return EMPTY_LOAD_FINGERPRINT;
+ }
+ final Hasher hasher = Hashing.sha256().newHasher();
+ for (String pair : pairs.stream().sorted().toList()) {
+ hasher.putUnencodedChars(pair);
+ hasher.putByte((byte) 0);
+ }
+ final String hex =
BaseEncoding.base16().encode(hasher.hash().asBytes()).toLowerCase(Locale.ROOT);
+ // should be good enough without dragging the whole thing around for every
segment
+ return FINGERPRINT_VERSION + ":" + hex.substring(0, 16);
+ }
+
+ private static String memberType(MatchResult result)
+ {
+ return
String.valueOf(result.wrappedLoadSpec().get(PartialLoadSpec.TYPE_FIELD));
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ CompositePartialLoadMatcher that = (CompositePartialLoadMatcher) o;
+ return Objects.equals(matchers, that.matchers);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(matchers);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "CompositePartialLoadMatcher{matchers=" + matchers + "}";
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadMatcher.java
b/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadMatcher.java
index 5bc1690a20e..734ed45fbf6 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadMatcher.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadMatcher.java
@@ -38,7 +38,8 @@ import java.util.Map;
@JsonSubTypes({
@JsonSubTypes.Type(name = ExactProjectionPartialLoadMatcher.TYPE, value =
ExactProjectionPartialLoadMatcher.class),
@JsonSubTypes.Type(name = WildcardProjectionPartialLoadMatcher.TYPE, value
= WildcardProjectionPartialLoadMatcher.class),
- @JsonSubTypes.Type(name = WildcardClusterGroupPartialLoadMatcher.TYPE,
value = WildcardClusterGroupPartialLoadMatcher.class)
+ @JsonSubTypes.Type(name = WildcardClusterGroupPartialLoadMatcher.TYPE,
value = WildcardClusterGroupPartialLoadMatcher.class),
+ @JsonSubTypes.Type(name = CompositePartialLoadMatcher.TYPE, value =
CompositePartialLoadMatcher.class)
})
public interface PartialLoadMatcher
{
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
index 1d72d2a2ed4..e6c6c585451 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
@@ -69,6 +69,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
@@ -175,6 +176,7 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
jsonMapper = TestHelper.makeJsonMapper();
jsonMapper.registerSubtypes(new NamedType(LocalLoadSpec.class, "local"));
jsonMapper.registerSubtypes(new NamedType(PartialProjectionLoadSpec.class,
PartialProjectionLoadSpec.TYPE));
+ jsonMapper.registerSubtypes(new NamedType(CompositePartialLoadSpec.class,
CompositePartialLoadSpec.TYPE));
jsonMapper.registerModule(new SegmentizerModule());
jsonMapper.registerModules(new
LocalDataStorageDruidModule().getJacksonModules());
jsonMapper.setInjectableValues(
@@ -231,6 +233,71 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
Assertions.assertFalse(location.isReserved(aggId), "selected bundle should
NOT be in staticCacheEntries");
}
+ @Test
+ void testLoadCompositeWrapperInstallsRuleHoldsOnUnionOfMemberBundles()
throws Exception
+ {
+ // A partialComposite wrapper must survive the whole historical-side path:
materialize each member with the
+ // composite's delegate spliced in, union their bundles, then rule-hold
and eagerly download every one. Cross-scheme
+ // unions (projections + cluster groups) are covered by
CompositePartialLoadSpecTest; here both members are
+ // projections because this fixture is not clustered.
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+ final String compositeFingerprint = "v1:composite-bundle-test";
+
+ manager.load(compositeWrapperSegment(List.of(AGG_BUNDLE,
OTHER_AGG_BUNDLE), compositeFingerprint));
+
+ final PartialSegmentMetadataCacheEntry metadata =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertTrue(metadata.isRuleHeld(), "rule must be applied to the
metadata entry");
+ Assertions.assertEquals(compositeFingerprint,
metadata.getRuleFingerprint());
+
+ // Every member's bundle is rule-held: the union, not just the first
member's selection.
+ for (String bundleName : List.of(AGG_BUNDLE, OTHER_AGG_BUNDLE)) {
+ Assertions.assertTrue(
+ metadata.isBundleRuleHeld(bundleName),
+ "bundle[" + bundleName + "] should be rule-held by the composite
rule"
+ );
+ }
+ // __base is reserved as a mount-time dependency of both members rather
than by a rule hold of its own.
+ Assertions.assertFalse(
+ metadata.isBundleRuleHeld(Projections.BASE_TABLE_PROJECTION_NAME),
+ "__base is a mount-time dependency, not a rule-selected bundle"
+ );
+ for (String bundleName : List.of(AGG_BUNDLE, OTHER_AGG_BUNDLE,
Projections.BASE_TABLE_PROJECTION_NAME)) {
+ Assertions.assertTrue(
+ location.isWeakReserved(new
PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID, bundleName)),
+ "bundle[" + bundleName + "] should be weak-reserved by the composite
rule"
+ );
+ }
+
+ // Eager downloads completed before load() returned, for both members and
the shared dependency.
+ final PartialSegmentFileMapperV10 mapper = metadata.getFileMapper();
+ Assertions.assertNotNull(mapper, "metadata mount should produce a file
mapper");
+ for (String bundleName : List.of(AGG_BUNDLE, OTHER_AGG_BUNDLE,
Projections.BASE_TABLE_PROJECTION_NAME)) {
+ Assertions.assertTrue(
+ mapper.isBundleFullyDownloaded(bundleName),
+ "bundle[" + bundleName + "] must be fully downloaded eagerly"
+ );
+ }
+ }
+
+ @Test
+ void testLoadCompositeWrapperWithOverlappingMembersHoldsBundleOnce() throws
Exception
+ {
+ // Two members selecting the same projection: getSelectedBundleNames
dedupes, so this must behave exactly like a
+ // single selection rather than double-holding or failing.
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(compositeWrapperSegment(List.of(AGG_BUNDLE, AGG_BUNDLE),
"v1:composite-overlap-test"));
+
+ final PartialSegmentMetadataCacheEntry metadata =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertTrue(metadata.isBundleRuleHeld(AGG_BUNDLE));
+ Assertions.assertFalse(
+ location.isWeakReserved(new
PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID, OTHER_AGG_BUNDLE)),
+ "unselected bundle must not be reserved"
+ );
+ }
+
@Test
void testLoadDoesNotReserveNonSelectedBundles() throws Exception
{
@@ -746,6 +813,33 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
.build();
}
+ /**
+ * A {@code partialComposite} wrapper whose members each select one
projection, matching what
+ * {@code CompositePartialLoadMatcher} emits: the delegate lives once at the
top level and members carry none.
+ */
+ private DataSegment compositeWrapperSegment(List<String>
projectionPerMember, String fingerprint)
+ {
+ final Map<String, Object> delegate = Map.of(
+ "type", "local",
+ "path", DEEP_STORAGE_DIR.getAbsolutePath()
+ );
+ final List<Map<String, Object>> members = projectionPerMember
+ .stream()
+ .map(projection -> {
+ final Map<String, Object> member = new HashMap<>(
+ PartialProjectionLoadSpec.wireForm(delegate,
List.of(projection), fingerprint + ":" + projection)
+ );
+ member.remove(PartialLoadSpec.DELEGATE_FIELD);
+ return member;
+ })
+ .toList();
+ return DataSegment.builder(SEGMENT_ID)
+ .shardSpec(NoneShardSpec.instance())
+ .loadSpec(CompositePartialLoadSpec.wireForm(delegate,
members, fingerprint))
+ .size(0)
+ .build();
+ }
+
/**
* A wrapper whose inner LoadSpec resolves via {@code LocalLoadSpec} against
a directory that holds no V10 file, so
* {@code openRangeReader()} returns {@code null}. Simulates the "backend
doesn't support range reads" case.
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/rules/CompositePartialLoadMatcherTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/rules/CompositePartialLoadMatcherTest.java
new file mode 100644
index 00000000000..ba7f22bf897
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/server/coordinator/rules/CompositePartialLoadMatcherTest.java
@@ -0,0 +1,457 @@
+/*
+ * 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.server.coordinator.rules;
+
+import com.fasterxml.jackson.databind.InjectableValues;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import nl.jqno.equalsverifier.EqualsVerifier;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.DruidExceptionMatcher;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.loading.CompositePartialLoadSpec;
+import org.apache.druid.segment.loading.PartialClusterGroupLoadSpec;
+import org.apache.druid.segment.loading.PartialLoadSpec;
+import org.apache.druid.segment.loading.PartialProjectionLoadSpec;
+import org.apache.druid.timeline.ClusterGroupTuples;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.partition.NumberedShardSpec;
+import org.hamcrest.MatcherAssert;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Tests for {@link CompositePartialLoadMatcher}: the union of member
selections, the veto when a member cannot match,
+ * single-member passthrough, and fingerprint stability.
+ */
+class CompositePartialLoadMatcherTest
+{
+ private static final Map<String, Object> BASE_LOAD_SPEC = Map.of("type",
"local", "path", "/seg");
+
+ private final ObjectMapper mapper = new DefaultObjectMapper();
+
+ @BeforeEach
+ void setUp()
+ {
+ final InjectableValues.Std injectables = new InjectableValues.Std();
+ injectables.addValue(DataSegment.PruneSpecsHolder.class,
DataSegment.PruneSpecsHolder.DEFAULT);
+ mapper.setInjectableValues(injectables);
+ }
+
+ @Test
+ void testConstructorRejectsNullMatchers()
+ {
+ MatcherAssert.assertThat(
+ Assertions.assertThrows(DruidException.class, () -> new
CompositePartialLoadMatcher(null)),
+ DruidExceptionMatcher.invalidInput().expectMessageContains("matchers
must not be null or empty")
+ );
+ }
+
+ @Test
+ void testConstructorRejectsEmptyMatchers()
+ {
+ MatcherAssert.assertThat(
+ Assertions.assertThrows(DruidException.class, () -> new
CompositePartialLoadMatcher(List.of())),
+ DruidExceptionMatcher.invalidInput().expectMessageContains("matchers
must not be null or empty")
+ );
+ }
+
+ @Test
+ void testConstructorRejectsNullMember()
+ {
+ MatcherAssert.assertThat(
+ Assertions.assertThrows(
+ DruidException.class,
+ () -> new
CompositePartialLoadMatcher(Arrays.asList(exactProjection("p"), null))
+ ),
+
DruidExceptionMatcher.invalidInput().expectMessageContains("matchers[1] must
not be null")
+ );
+ }
+
+ @Test
+ void testUnionsAcrossSchemes()
+ {
+ final DataSegment segment =
clusteredSegmentWithProjections(List.of("user_hourly", "user_daily"));
+ final CompositePartialLoadMatcher matcher = new
CompositePartialLoadMatcher(List.of(
+ exactProjection("user_hourly"),
+ globClusterGroup(Map.of("tenant", "acme"))
+ ));
+
+ final PartialLoadMatcher.MatchResult result = matcher.match(segment,
BASE_LOAD_SPEC);
+ Assertions.assertNotNull(result);
+ Assertions.assertEquals(CompositePartialLoadSpec.TYPE,
result.wrappedLoadSpec().get("type"));
+ Assertions.assertEquals(BASE_LOAD_SPEC,
result.wrappedLoadSpec().get("delegate"));
+ Assertions.assertEquals(
+ List.of(
+ Map.of(
+ "type", PartialProjectionLoadSpec.TYPE,
+ "projections", List.of("user_hourly"),
+ "fingerprint", exactProjection("user_hourly").match(segment,
BASE_LOAD_SPEC).fingerprint()
+ ),
+ Map.of(
+ "type", PartialClusterGroupLoadSpec.TYPE,
+ "clusterGroupIndices", List.of(0, 1),
+ "fingerprint", globClusterGroup(Map.of("tenant",
"acme")).match(segment, BASE_LOAD_SPEC).fingerprint()
+ )
+ ),
+ members(result)
+ );
+ }
+
+ @Test
+ void testUnionsTwoSameSchemeMembers()
+ {
+ final DataSegment segment = clusteredSegmentWithProjections(null);
+ final CompositePartialLoadMatcher matcher = new
CompositePartialLoadMatcher(List.of(
+ globClusterGroup(Map.of("tenant", "acme", "region", "us-east-1")),
+ globClusterGroup(Map.of("tenant", "globex", "region", "us-east-1"))
+ ));
+
+ final PartialLoadMatcher.MatchResult result = matcher.match(segment,
BASE_LOAD_SPEC);
+ Assertions.assertNotNull(result);
+ Assertions.assertEquals(
+ List.of(List.of(0), List.of(2)),
+ members(result).stream().map(m ->
m.get("clusterGroupIndices")).toList()
+ );
+ }
+
+ @Test
+ void testUnionsPerBranchIncludeExcludePairs()
+ {
+ // A single globClusterGroup has one excludePatterns list, applied to
every include. That makes an
+ // include/exclude pair unexpressible as a unit whenever an exclude isn't
qualified by its own branch's include:
+ // branch A — every acme region except us-west-2
+ // branch B — every globex region except us-east-1
+ // Collapsing these into one matcher means both excludes apply to both
includes, which over-excludes. Each branch
+ // becomes its own composite member instead.
+ final DataSegment segment = fourGroupSegment();
+ final PartialLoadMatcher branchA = globClusterGroup(
+ List.of(Map.of("tenant", "acme")),
+ List.of(Map.of("region", "us-west-2"))
+ );
+ final PartialLoadMatcher branchB = globClusterGroup(
+ List.of(Map.of("tenant", "globex")),
+ List.of(Map.of("region", "us-east-1"))
+ );
+
+ final PartialLoadMatcher.MatchResult result =
+ new CompositePartialLoadMatcher(List.of(branchA,
branchB)).match(segment, BASE_LOAD_SPEC);
+ Assertions.assertNotNull(result);
+ Assertions.assertEquals(
+ // (acme, us-east-1) from branch A; (globex, us-west-2) from branch B
+ List.of(List.of(0), List.of(3)),
+ members(result).stream().map(m ->
m.get("clusterGroupIndices")).toList()
+ );
+
+ // The collapsed single matcher is not equivalent: pooling the excludes
wipes out every group.
+ final PartialLoadMatcher.MatchResult collapsed = globClusterGroup(
+ List.of(Map.of("tenant", "acme"), Map.of("tenant", "globex")),
+ List.of(Map.of("region", "us-west-2"), Map.of("region", "us-east-1"))
+ ).match(segment, BASE_LOAD_SPEC);
+ Assertions.assertNotNull(collapsed);
+ Assertions.assertEquals(List.of(),
collapsed.wrappedLoadSpec().get("clusterGroupIndices"));
+ }
+
+ @Test
+ void testMembersCarryNoDelegate()
+ {
+ // The composite carries the backend load spec exactly once, at the top
level.
+ final DataSegment segment =
clusteredSegmentWithProjections(List.of("user_hourly"));
+ final CompositePartialLoadMatcher matcher = new
CompositePartialLoadMatcher(List.of(
+ exactProjection("user_hourly"),
+ globClusterGroup(Map.of("tenant", "acme"))
+ ));
+ final PartialLoadMatcher.MatchResult result = matcher.match(segment,
BASE_LOAD_SPEC);
+ Assertions.assertNotNull(result);
+ for (Map<String, Object> member : members(result)) {
+ Assertions.assertFalse(
+ member.containsKey(PartialLoadSpec.DELEGATE_FIELD),
+ "member should not carry a delegate: " + member
+ );
+ }
+ }
+
+ @Test
+ void testNullMemberVetoesWholeComposite()
+ {
+ // The cluster-group member cannot reason about a non-clustered segment.
Skipping it would announce a segment
+ // holding only its projections and none of its rows, so the composite
goes opaque and the rule's
+ // CannotMatchBehavior decides.
+ final DataSegment segment =
unclusteredSegmentWithProjections(List.of("user_hourly"));
+ Assertions.assertNotNull(exactProjection("user_hourly").match(segment,
BASE_LOAD_SPEC));
+ Assertions.assertNull(globClusterGroup(Map.of("tenant",
"acme")).match(segment, BASE_LOAD_SPEC));
+
+ final CompositePartialLoadMatcher matcher = new
CompositePartialLoadMatcher(List.of(
+ exactProjection("user_hourly"),
+ globClusterGroup(Map.of("tenant", "acme"))
+ ));
+ Assertions.assertNull(matcher.match(segment, BASE_LOAD_SPEC));
+ }
+
+ @Test
+ void testUnknownMemberVetoesWholeComposite()
+ {
+ // A matcher type this Druid version doesn't recognize deserializes to
UnknownPartialLoadMatcher, whose match()
+ // returns null. The composite must escalate rather than silently narrow
the load.
+ final DataSegment segment =
clusteredSegmentWithProjections(List.of("user_hourly"));
+ final CompositePartialLoadMatcher matcher = new
CompositePartialLoadMatcher(List.of(
+ exactProjection("user_hourly"),
+ new UnknownPartialLoadMatcher()
+ ));
+ Assertions.assertNull(matcher.match(segment, BASE_LOAD_SPEC));
+ }
+
+ @Test
+ void testSingleMemberPassesThroughVerbatim()
+ {
+ // Wrapping a matcher in a one-element composite must not change the load
spec or the fingerprint, so wrapping an
+ // existing rule doesn't re-fingerprint every segment it covers.
+ final DataSegment segment =
clusteredSegmentWithProjections(List.of("user_hourly"));
+ final PartialLoadMatcher bare = exactProjection("user_hourly");
+ final PartialLoadMatcher.MatchResult bareResult = bare.match(segment,
BASE_LOAD_SPEC);
+ final PartialLoadMatcher.MatchResult wrappedResult =
+ new CompositePartialLoadMatcher(List.of(bare)).match(segment,
BASE_LOAD_SPEC);
+
+ Assertions.assertEquals(bareResult, wrappedResult);
+ Assertions.assertEquals(PartialProjectionLoadSpec.TYPE,
wrappedResult.wrappedLoadSpec().get("type"));
+ }
+
+ @Test
+ void testAllEmptyMembersReportEmptyFingerprint()
+ {
+ // Every member resolved to an empty selection: the empty-load contract
carries through composition.
+ final DataSegment segment = clusteredSegmentWithProjections(null);
+ final CompositePartialLoadMatcher matcher = new
CompositePartialLoadMatcher(List.of(
+ globClusterGroup(Map.of("tenant", "nobody")),
+ globClusterGroup(Map.of("tenant", "nobody-else"))
+ ));
+ final PartialLoadMatcher.MatchResult result = matcher.match(segment,
BASE_LOAD_SPEC);
+ Assertions.assertNotNull(result);
+ Assertions.assertEquals(PartialLoadMatcher.EMPTY_LOAD_FINGERPRINT,
result.fingerprint());
+ Assertions.assertEquals(
+ List.of(List.of(), List.of()),
+ members(result).stream().map(m ->
m.get("clusterGroupIndices")).toList()
+ );
+ }
+
+ @Test
+ void testOneEmptyMemberDoesNotMakeCompositeEmpty()
+ {
+ final DataSegment segment = clusteredSegmentWithProjections(null);
+ final CompositePartialLoadMatcher matcher = new
CompositePartialLoadMatcher(List.of(
+ globClusterGroup(Map.of("tenant", "acme")),
+ globClusterGroup(Map.of("tenant", "nobody"))
+ ));
+ final PartialLoadMatcher.MatchResult result = matcher.match(segment,
BASE_LOAD_SPEC);
+ Assertions.assertNotNull(result);
+ Assertions.assertNotEquals(PartialLoadMatcher.EMPTY_LOAD_FINGERPRINT,
result.fingerprint());
+ }
+
+ @Test
+ void testFingerprintStableAcrossMatcherReordering()
+ {
+ // The resolved selection is a set union, so reordering the matchers must
not thrash the cascade.
+ final DataSegment segment =
clusteredSegmentWithProjections(List.of("user_hourly"));
+ final PartialLoadMatcher projection = exactProjection("user_hourly");
+ final PartialLoadMatcher clusterGroup = globClusterGroup(Map.of("tenant",
"acme"));
+
+ final String forward = new CompositePartialLoadMatcher(List.of(projection,
clusterGroup))
+ .match(segment, BASE_LOAD_SPEC).fingerprint();
+ final String reversed = new
CompositePartialLoadMatcher(List.of(clusterGroup, projection))
+ .match(segment, BASE_LOAD_SPEC).fingerprint();
+ Assertions.assertEquals(forward, reversed);
+ }
+
+ @Test
+ void testFingerprintDiffersOnDifferentMemberContent()
+ {
+ final DataSegment segment =
clusteredSegmentWithProjections(List.of("user_hourly", "user_daily"));
+ final String hourly = new CompositePartialLoadMatcher(List.of(
+ exactProjection("user_hourly"),
+ globClusterGroup(Map.of("tenant", "acme"))
+ )).match(segment, BASE_LOAD_SPEC).fingerprint();
+ final String daily = new CompositePartialLoadMatcher(List.of(
+ exactProjection("user_daily"),
+ globClusterGroup(Map.of("tenant", "acme"))
+ )).match(segment, BASE_LOAD_SPEC).fingerprint();
+ Assertions.assertNotEquals(hourly, daily);
+ }
+
+ @Test
+ void testFingerprintDiffersFromMemberFingerprints()
+ {
+ // Sanity: the composite mints its own fingerprint rather than reusing a
member's, so a rule swap between a
+ // composite and one of its members is detected.
+ final DataSegment segment =
clusteredSegmentWithProjections(List.of("user_hourly"));
+ final PartialLoadMatcher projection = exactProjection("user_hourly");
+ final PartialLoadMatcher clusterGroup = globClusterGroup(Map.of("tenant",
"acme"));
+ final String composite = new
CompositePartialLoadMatcher(List.of(projection, clusterGroup))
+ .match(segment, BASE_LOAD_SPEC).fingerprint();
+ Assertions.assertNotEquals(projection.match(segment,
BASE_LOAD_SPEC).fingerprint(), composite);
+ Assertions.assertNotEquals(clusterGroup.match(segment,
BASE_LOAD_SPEC).fingerprint(), composite);
+ }
+
+ @Test
+ void testNestedComposite()
+ {
+ final DataSegment segment =
clusteredSegmentWithProjections(List.of("user_hourly"));
+ final CompositePartialLoadMatcher matcher = new
CompositePartialLoadMatcher(List.of(
+ exactProjection("user_hourly"),
+ new CompositePartialLoadMatcher(List.of(
+ globClusterGroup(Map.of("tenant", "acme", "region", "us-east-1")),
+ globClusterGroup(Map.of("tenant", "globex", "region", "us-east-1"))
+ ))
+ ));
+ final PartialLoadMatcher.MatchResult result = matcher.match(segment,
BASE_LOAD_SPEC);
+ Assertions.assertNotNull(result);
+ final List<Map<String, Object>> members = members(result);
+ Assertions.assertEquals(PartialProjectionLoadSpec.TYPE,
members.get(0).get("type"));
+ Assertions.assertEquals(CompositePartialLoadSpec.TYPE,
members.get(1).get("type"));
+ // The nested composite's own members are likewise delegate-free.
+ @SuppressWarnings("unchecked")
+ final List<Map<String, Object>> nested = (List<Map<String, Object>>)
members.get(1).get("members");
+ for (Map<String, Object> m : nested) {
+ Assertions.assertFalse(m.containsKey(PartialLoadSpec.DELEGATE_FIELD),
"nested member has a delegate: " + m);
+ }
+ }
+
+ @Test
+ void testJsonRoundTrip() throws Exception
+ {
+ final PartialLoadMatcher matcher = new CompositePartialLoadMatcher(List.of(
+ exactProjection("user_hourly"),
+ globClusterGroup(Map.of("tenant", "acme"))
+ ));
+ final String json = mapper.writeValueAsString(matcher);
+ final PartialLoadMatcher reread = mapper.readValue(json,
PartialLoadMatcher.class);
+ Assertions.assertInstanceOf(CompositePartialLoadMatcher.class, reread);
+ Assertions.assertEquals(matcher, reread);
+ }
+
+ @Test
+ void testJsonRoundTripNested() throws Exception
+ {
+ final PartialLoadMatcher matcher = new CompositePartialLoadMatcher(List.of(
+ exactProjection("user_hourly"),
+ new
CompositePartialLoadMatcher(List.of(globClusterGroup(Map.of("tenant", "acme"))))
+ ));
+ final String json = mapper.writeValueAsString(matcher);
+ Assertions.assertEquals(matcher, mapper.readValue(json,
PartialLoadMatcher.class));
+ }
+
+ @Test
+ void testEquals()
+ {
+
EqualsVerifier.forClass(CompositePartialLoadMatcher.class).usingGetClass().verify();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<Map<String, Object>>
members(PartialLoadMatcher.MatchResult result)
+ {
+ return (List<Map<String, Object>>) result.wrappedLoadSpec().get("members");
+ }
+
+ private static PartialLoadMatcher exactProjection(String name)
+ {
+ return new ExactProjectionPartialLoadMatcher(List.of(name));
+ }
+
+ private static PartialLoadMatcher globClusterGroup(Map<String, String>
pattern)
+ {
+ return new WildcardClusterGroupPartialLoadMatcher(List.of(pattern), null);
+ }
+
+ private static PartialLoadMatcher globClusterGroup(
+ List<Map<String, String>> patterns,
+ List<Map<String, String>> excludePatterns
+ )
+ {
+ return new WildcardClusterGroupPartialLoadMatcher(patterns,
excludePatterns);
+ }
+
+ private static RowSignature tenantRegion()
+ {
+ return RowSignature.builder()
+ .add("tenant", ColumnType.STRING)
+ .add("region", ColumnType.STRING)
+ .build();
+ }
+
+ /** A 3-group fixture: (acme, us-east-1), (acme, us-west-2), (globex,
us-east-1). */
+ private static DataSegment clusteredSegmentWithProjections(@Nullable
List<String> projections)
+ {
+ final DataSegment.Builder builder = baseBuilder()
+ .clusterGroups(new ClusterGroupTuples(
+ tenantRegion(),
+ List.of(
+ List.of("acme", "us-east-1"),
+ List.of("acme", "us-west-2"),
+ List.of("globex", "us-east-1")
+ )
+ ));
+ if (projections != null) {
+ builder.projections(projections);
+ }
+ return builder.build();
+ }
+
+ /**
+ * A 4-group fixture spanning both tenants in both regions — (acme,
us-east-1), (acme, us-west-2),
+ * (globex, us-east-1), (globex, us-west-2) — so a region-only exclude
scoped to one tenant's branch has something
+ * to over-exclude in the other's if the two branches are pooled into a
single matcher.
+ */
+ private static DataSegment fourGroupSegment()
+ {
+ return baseBuilder()
+ .clusterGroups(new ClusterGroupTuples(
+ tenantRegion(),
+ List.of(
+ List.of("acme", "us-east-1"),
+ List.of("acme", "us-west-2"),
+ List.of("globex", "us-east-1"),
+ List.of("globex", "us-west-2")
+ )
+ ))
+ .build();
+ }
+
+ private static DataSegment unclusteredSegmentWithProjections(List<String>
projections)
+ {
+ return baseBuilder().projections(projections).build();
+ }
+
+ private static DataSegment.Builder baseBuilder()
+ {
+ final NumberedShardSpec shardSpec = new NumberedShardSpec(0, 1);
+ return DataSegment.builder(SegmentId.of("ds",
Intervals.of("2026-01-01/2026-01-02"), "v", shardSpec))
+ .shardSpec(shardSpec)
+ .loadSpec(BASE_LOAD_SPEC)
+ .size(0);
+ }
+}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardClusterGroupPartialLoadMatcherTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardClusterGroupPartialLoadMatcherTest.java
index 4c69e98c3ab..331dd6ac488 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardClusterGroupPartialLoadMatcherTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardClusterGroupPartialLoadMatcherTest.java
@@ -21,6 +21,7 @@ package org.apache.druid.server.coordinator.rules;
import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
+import nl.jqno.equalsverifier.EqualsVerifier;
import org.apache.druid.error.DruidException;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.Intervals;
@@ -472,6 +473,15 @@ class WildcardClusterGroupPartialLoadMatcherTest
Assertions.assertEquals(original, back);
}
+ @Test
+ void testEquals()
+ {
+ EqualsVerifier.forClass(WildcardClusterGroupPartialLoadMatcher.class)
+ .withIgnoredFields("compiledPatterns",
"compiledExcludePatterns")
+ .usingGetClass()
+ .verify();
+ }
+
@Test
void testVirtualColumnsOmittedFromJsonWhenEmpty() throws Exception
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
