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 a8575529831 feat: improved partial load rule behavior (#19903)
a8575529831 is described below
commit a85755298317be02b29528b8f5d7a5f7550cb5af
Author: Clint Wylie <[email protected]>
AuthorDate: Mon Aug 10 12:03:33 2026 -0700
feat: improved partial load rule behavior (#19903)
---
.../segment/loading/PartialBaseTableLoadSpec.java | 198 ++++++++++
.../loading/PartialFullSegmentLoadSpec.java | 131 +++++++
.../druid/segment/loading/PartialLoadSpec.java | 15 +
.../loading/PartialBaseTableLoadSpecTest.java | 432 +++++++++++++++++++++
.../loading/PartialFullSegmentLoadSpecTest.java | 277 +++++++++++++
.../apache/druid/guice/PartialLoadSpecModule.java | 10 +-
.../coordinator/rules/CannotMatchBehavior.java | 29 +-
.../server/coordinator/rules/PartialLoadRule.java | 62 ++-
.../rules/ProjectionPartialLoadMatcher.java | 29 +-
...egmentLocalCacheManagerPartialRuleLoadTest.java | 108 ++++++
.../StrategicSegmentAssignerPartialTest.java | 12 +-
.../ExactProjectionPartialLoadMatcherTest.java | 37 +-
.../coordinator/rules/PartialLoadRuleTest.java | 175 +++++++--
.../WildcardProjectionPartialLoadMatcherTest.java | 31 +-
.../druid/server/http/DataSourcesResourceTest.java | 10 +-
15 files changed, 1479 insertions(+), 77 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/segment/loading/PartialBaseTableLoadSpec.java
b/processing/src/main/java/org/apache/druid/segment/loading/PartialBaseTableLoadSpec.java
new file mode 100644
index 00000000000..dbd34cdca24
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/loading/PartialBaseTableLoadSpec.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.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 org.apache.druid.error.DruidException;
+import org.apache.druid.segment.file.SegmentFileBuilder;
+import org.apache.druid.segment.file.SegmentFileMetadata;
+import
org.apache.druid.segment.projections.ClusteredValueGroupsBaseTableSchema;
+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.timeline.DataSegment;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * A {@link PartialLoadSpec} that requests the segment's base table and
nothing else; every row, no projections.
+ * <p>
+ * This is the floor of a partial load: it is what a matcher asks for when it
has no scheme-specific content to
+ * contribute but the segment's rows must still be resident. A projection
matcher whose configured projections aren't
+ * present on a segment resolves to this rather than going opaque, because a
projection is always recomputable from
+ * the base table, so the base table is a correct (if slower) substitute for a
missing one.
+ * <p>
+ * This load spec no scheme-specific field; "the base table" is not a name the
coordinator can resolve. Where the rows
+ * actually live depends on the segment's physical layout, which only {@link
#getSelectedBundleNames} can see, so the
+ * spec is just the base contract and resolution happens on the historical.
+ */
+@JsonTypeName(PartialBaseTableLoadSpec.TYPE)
+public class PartialBaseTableLoadSpec extends PartialLoadSpec
+{
+ public static final String TYPE = "partialBaseTable";
+
+ /**
+ * The fingerprint of every base-table load. Fixed, because the selection
carries no scheme-specific content to
+ * distinguish: it is derived entirely from the segment's layout, so any two
base-table loads of the same segment
+ * resolve identically.
+ */
+ public static final String FINGERPRINT = "v1:partial-base";
+
+ /**
+ * Builds the raw {@link Map} form of a {@link PartialBaseTableLoadSpec}
request. Used by the coordinator-side
+ * matchers, which don't instantiate the typed class because doing so would
require plumbing an
+ * {@link ObjectMapper} through every matcher just to satisfy the
constructor's lazy-delegate supplier.
+ */
+ public static Map<String, Object> wireForm(Map<String, Object> delegate,
String fingerprint)
+ {
+ return Map.of(
+ TYPE_FIELD, TYPE,
+ DELEGATE_FIELD, delegate,
+ FINGERPRINT_FIELD, fingerprint
+ );
+ }
+
+ @JsonCreator
+ public PartialBaseTableLoadSpec(
+ @JsonProperty("delegate") Map<String, Object> delegate,
+ @JsonProperty("fingerprint") String fingerprint,
+ @JacksonInject ObjectMapper jsonMapper
+ )
+ {
+ super(delegate, fingerprint, jsonMapper);
+ }
+
+ /**
+ * Resolves "the base table" to the bundles that physically hold the
segment's rows. Which bundles those are is a
+ * property of the layout, not of the request:
+ * <ul>
+ * <li><b>Clustered</b> — the rows are partitioned across the cluster
groups, and {@code __base} holds only the
+ * parts they share, so the base table is <em>every</em> {@code
__base$<clusteringValueIds>} bundle, plus
+ * {@code __base} itself when the segment carries one.</li>
+ * <li><b>Unclustered</b> — the base table is the single {@code __base}
bundle.</li>
+ * <li><b>Legacy</b> — a V10 segment written before the bundle name was
persisted reports every container under
+ * {@link SegmentFileBuilder#ROOT_BUNDLE_NAME}, which is then the
whole segment. Gated on root being the sole
+ * bundle, matching {@code
PartialSegmentBundleCacheEntry#resolveBundleName}.</li>
+ * </ul>
+ * Every returned name is checked against the bundles the segment actually
carries first. That matters because
+ * {@code resolveBundleName} deliberately passes an unknown name through
unchanged so the acquire fails loudly, so
+ * naming a bundle that isn't there would turn a layout the reader simply
doesn't recognize into a hard load failure.
+ */
+ @Override
+ public List<String> getSelectedBundleNames(DataSegment segment,
SegmentFileMetadata metadata)
+ {
+ final Set<String> present = presentBundleNames(metadata);
+ if (present.size() == 1 &&
present.contains(SegmentFileBuilder.ROOT_BUNDLE_NAME)) {
+ return List.of(SegmentFileBuilder.ROOT_BUNDLE_NAME);
+ }
+
+ final List<String> clusterGroupBundles = clusterGroupBundleNames(metadata);
+ if (clusterGroupBundles == null) {
+ if (!present.contains(Projections.BASE_TABLE_PROJECTION_NAME)) {
+ throw DruidException.defensive(
+ "Cannot resolve base-table bundles for segment[%s]: no [%s] bundle
among %s",
+ segment.getId(),
+ Projections.BASE_TABLE_PROJECTION_NAME,
+ present
+ );
+ }
+ return List.of(Projections.BASE_TABLE_PROJECTION_NAME);
+ }
+
+ final List<String> selected = new ArrayList<>(clusterGroupBundles.size() +
1);
+ // The shared base bundle is optional on a clustered segment: it exists
only once the segment carries shared
+ // column parts. It is also an inferred dependency of every group bundle,
so naming it here is belt and braces.
+ if (present.contains(Projections.BASE_TABLE_PROJECTION_NAME)) {
+ selected.add(Projections.BASE_TABLE_PROJECTION_NAME);
+ }
+ for (String groupBundle : clusterGroupBundles) {
+ if (!present.contains(groupBundle)) {
+ throw DruidException.defensive(
+ "Cannot resolve base-table bundles for segment[%s]: metadata
declares cluster-group bundle[%s] but the"
+ + " segment carries %s",
+ segment.getId(),
+ groupBundle,
+ present
+ );
+ }
+ selected.add(groupBundle);
+ }
+ return selected;
+ }
+
+ /**
+ * Every cluster group's bundle name when the segment's base projection is
clustered, or {@code null} when it isn't.
+ * Unlike {@link PartialClusterGroupLoadSpec}, this reads the group list
straight out of the metadata rather than
+ * indexing into it, so it needs no cross-check against the segment's tuple
count.
+ */
+ private static List<String> clusterGroupBundleNames(SegmentFileMetadata
metadata)
+ {
+ final List<ProjectionMetadata> projections = metadata.getProjections();
+ if (projections == null || projections.isEmpty()) {
+ return null;
+ }
+ if (!(projections.getFirst().getSchema() instanceof
ClusteredValueGroupsBaseTableSchema clusteredSummary)) {
+ return null;
+ }
+ final List<TableClusterGroupSpec> groups =
clusteredSummary.getClusterGroups();
+ final List<String> bundleNames = new ArrayList<>(groups.size());
+ for (TableClusterGroupSpec group : groups) {
+
bundleNames.add(Projections.getClusterGroupBundleName(group.getClusteringValueIds()));
+ }
+ return bundleNames;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PartialBaseTableLoadSpec that = (PartialBaseTableLoadSpec) o;
+ return Objects.equals(getDelegate(), that.getDelegate())
+ && Objects.equals(getFingerprint(), that.getFingerprint());
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(getDelegate(), getFingerprint());
+ }
+
+ @Override
+ public String toString()
+ {
+ return "PartialBaseTableLoadSpec{" +
+ "delegate=" + getDelegate() +
+ ", fingerprint=" + getFingerprint() +
+ '}';
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/segment/loading/PartialFullSegmentLoadSpec.java
b/processing/src/main/java/org/apache/druid/segment/loading/PartialFullSegmentLoadSpec.java
new file mode 100644
index 00000000000..deb2cd667ff
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/loading/PartialFullSegmentLoadSpec.java
@@ -0,0 +1,131 @@
+/*
+ * 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 org.apache.druid.error.DruidException;
+import org.apache.druid.segment.file.SegmentFileMetadata;
+import org.apache.druid.timeline.DataSegment;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * A {@link PartialLoadSpec} that requests every bundle the segment contains.
The ceiling of a partial load,
+ * as {@link PartialBaseTableLoadSpec} is its floor.
+ * <p>
+ * This exists so a rule can ask for the whole segment to be
<em>resident</em>. Dispatching a segment without any
+ * partial-load wrapper does not do that on a virtual-storage historical: the
segment is registered and its bundles
+ * are fetched on demand as queries touch them, which is a different thing
from having them all on disk. Selecting
+ * every bundle through the rule path is what actually pins the whole segment,
and it announces a fingerprint like any
+ * other partial load, so the coordinator can reconcile it.
+ * <p>
+ * Like {@link PartialBaseTableLoadSpec}, this carries no scheme-specific
field — the selection is a property of the
+ * segment's layout, so it is resolved on the historical by {@link
#getSelectedBundleNames}.
+ */
+@JsonTypeName(PartialFullSegmentLoadSpec.TYPE)
+public class PartialFullSegmentLoadSpec extends PartialLoadSpec
+{
+ public static final String TYPE = "partialFullSegment";
+
+ /**
+ * The fingerprint of every full-segment load. Fixed, for the same reason as
+ * {@link PartialBaseTableLoadSpec#FINGERPRINT}: the selection is derived
entirely from the segment's layout, so any
+ * two full-segment loads of the same segment resolve identically.
+ */
+ public static final String FINGERPRINT = "v1:partial-full";
+
+ /**
+ * Builds the raw {@link Map} form of a {@link PartialFullSegmentLoadSpec}
request. Used by the coordinator side,
+ * which doesn't instantiate the typed class because doing so would require
plumbing an {@link ObjectMapper} through
+ * just to satisfy the constructor's lazy-delegate supplier.
+ */
+ public static Map<String, Object> wireForm(Map<String, Object> delegate,
String fingerprint)
+ {
+ return Map.of(
+ TYPE_FIELD, TYPE,
+ DELEGATE_FIELD, delegate,
+ FINGERPRINT_FIELD, fingerprint
+ );
+ }
+
+ @JsonCreator
+ public PartialFullSegmentLoadSpec(
+ @JsonProperty("delegate") Map<String, Object> delegate,
+ @JsonProperty("fingerprint") String fingerprint,
+ @JacksonInject ObjectMapper jsonMapper
+ )
+ {
+ super(delegate, fingerprint, jsonMapper);
+ }
+
+ /**
+ * Every bundle the segment carries, in container order. No layout branching
is needed: whatever the writer put in
+ * the file is what gets selected, so this works the same for clustered,
projection-bearing, plain and legacy
+ * root-only segments alike.
+ */
+ @Override
+ public List<String> getSelectedBundleNames(DataSegment segment,
SegmentFileMetadata metadata)
+ {
+ final Set<String> present = presentBundleNames(metadata);
+ if (present.isEmpty()) {
+ throw DruidException.defensive(
+ "Cannot resolve full-segment bundles for segment[%s]: metadata
declares no containers",
+ segment.getId()
+ );
+ }
+ return List.copyOf(present);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PartialFullSegmentLoadSpec that = (PartialFullSegmentLoadSpec) o;
+ return Objects.equals(getDelegate(), that.getDelegate())
+ && Objects.equals(getFingerprint(), that.getFingerprint());
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(getDelegate(), getFingerprint());
+ }
+
+ @Override
+ public String toString()
+ {
+ return "PartialFullSegmentLoadSpec{" +
+ "delegate=" + getDelegate() +
+ ", fingerprint=" + getFingerprint() +
+ '}';
+ }
+}
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 0e3174647c3..d9fc7de3771 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
@@ -24,14 +24,17 @@ 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.segment.file.SegmentFileContainerMetadata;
import org.apache.druid.segment.file.SegmentFileMetadata;
import org.apache.druid.timeline.DataSegment;
import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* Base for {@link LoadSpec} wrappers that carry partial-load metadata (a
fingerprint identifying the request the
@@ -167,4 +170,16 @@ public abstract class PartialLoadSpec implements LoadSpec
* bootstrap re-discovers the segment).
*/
public abstract List<String> getSelectedBundleNames(DataSegment segment,
SegmentFileMetadata metadata);
+
+ /**
+ * The distinct bundle names carried by {@code metadata}, in container order.
+ */
+ protected static Set<String> presentBundleNames(SegmentFileMetadata metadata)
+ {
+ final Set<String> names = new LinkedHashSet<>();
+ for (SegmentFileContainerMetadata container : metadata.getContainers()) {
+ names.add(container.getBundle());
+ }
+ return names;
+ }
}
diff --git
a/processing/src/test/java/org/apache/druid/segment/loading/PartialBaseTableLoadSpecTest.java
b/processing/src/test/java/org/apache/druid/segment/loading/PartialBaseTableLoadSpecTest.java
new file mode 100644
index 00000000000..9c06a1dbd47
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/segment/loading/PartialBaseTableLoadSpecTest.java
@@ -0,0 +1,432 @@
+/*
+ * 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.SegmentFileBuilder;
+import org.apache.druid.segment.file.SegmentFileContainerMetadata;
+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.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.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class PartialBaseTableLoadSpecTest
+{
+ private static final Map<String, Object> DELEGATE = ImmutableMap.of(
+ "type", "stub",
+ "path", "/var/druid/segments/foo"
+ );
+
+ private static ObjectMapper configuredMapper()
+ {
+ final ObjectMapper m = new DefaultObjectMapper();
+ final SimpleModule module = new SimpleModule();
+ module.registerSubtypes(PartialBaseTableLoadSpec.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
+ {
+ PartialBaseTableLoadSpec spec = new PartialBaseTableLoadSpec(
+ DELEGATE,
+ PartialBaseTableLoadSpec.FINGERPRINT,
+ jsonMapper
+ );
+ String json = jsonMapper.writeValueAsString(spec);
+ LoadSpec reread = jsonMapper.readValue(json, LoadSpec.class);
+ Assertions.assertInstanceOf(PartialBaseTableLoadSpec.class, reread);
+ Assertions.assertEquals(spec, reread);
+ }
+
+ @Test
+ void testWireFormHasPartialBaseTableType() throws Exception
+ {
+ PartialBaseTableLoadSpec spec = new PartialBaseTableLoadSpec(
+ DELEGATE,
+ PartialBaseTableLoadSpec.FINGERPRINT,
+ jsonMapper
+ );
+ Map<String, Object> wireForm = jsonMapper.readValue(
+ jsonMapper.writeValueAsString(spec),
+ new TypeReference<>()
+ {
+ }
+ );
+ Assertions.assertEquals("partialBaseTable", wireForm.get("type"));
+ Assertions.assertEquals(DELEGATE, wireForm.get("delegate"));
+ Assertions.assertEquals(PartialBaseTableLoadSpec.FINGERPRINT,
wireForm.get("fingerprint"));
+ // No scheme-specific field: "the base table" is resolved from layout on
the historical.
+ Assertions.assertEquals(3, wireForm.size());
+ }
+
+ @Test
+ void testFingerprintIsNotTheEmptyLoadSentinel()
+ {
+ // A base-table load puts every row on the historical; an empty load puts
nothing there. Sharing a fingerprint
+ // would make a rule swap between the two invisible to the coordinator,
and the load would never be re-issued.
+ Assertions.assertNotEquals("v1:partial-empty",
PartialBaseTableLoadSpec.FINGERPRINT);
+ }
+
+ @Test
+ void testGetSelectedBundleNamesUnclusteredSegmentSelectsBase()
+ {
+ final SegmentFileMetadata metadata = metadata(
+ List.of(Projections.BASE_TABLE_PROJECTION_NAME, "user_hourly"),
+ List.of(unclusteredBaseProjection(), projection("user_hourly"))
+ );
+ Assertions.assertEquals(
+ List.of(Projections.BASE_TABLE_PROJECTION_NAME),
+ spec().getSelectedBundleNames(anySegment(), metadata)
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesSegmentWithNoProjectionsSelectsBase()
+ {
+ final SegmentFileMetadata metadata =
metadata(List.of(Projections.BASE_TABLE_PROJECTION_NAME), null);
+ Assertions.assertEquals(
+ List.of(Projections.BASE_TABLE_PROJECTION_NAME),
+ spec().getSelectedBundleNames(anySegment(), metadata)
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesClusteredSegmentSelectsEveryGroup()
+ {
+ // On a clustered segment the rows are spread across the group bundles, so
"the base table" is all of them, plus
+ // the shared __base bundle when the segment carries one.
+ final List<TableClusterGroupSpec> groups = List.of(
+ new TableClusterGroupSpec(List.of(0), 10),
+ new TableClusterGroupSpec(List.of(1), 20),
+ new TableClusterGroupSpec(List.of(2), 30)
+ );
+ final List<String> groupBundles = groups.stream()
+ .map(g ->
Projections.getClusterGroupBundleName(g.getClusteringValueIds()))
+ .toList();
+ final List<String> present = new ArrayList<>();
+ present.add(Projections.BASE_TABLE_PROJECTION_NAME);
+ present.addAll(groupBundles);
+ final SegmentFileMetadata metadata = metadata(present,
List.of(clusteredBaseProjection(groups)));
+
+ final List<String> expected = new ArrayList<>();
+ expected.add(Projections.BASE_TABLE_PROJECTION_NAME);
+ expected.addAll(groupBundles);
+ Assertions.assertEquals(expected,
spec().getSelectedBundleNames(anySegment(), metadata));
+ }
+
+ @Test
+ void testGetSelectedBundleNamesClusteredSegmentWithoutSharedBaseBundle()
+ {
+ // __base is optional on a clustered segment: it only exists once there
are shared column parts to put in it.
+ final List<TableClusterGroupSpec> groups = List.of(
+ new TableClusterGroupSpec(List.of(0), 10),
+ new TableClusterGroupSpec(List.of(1), 20)
+ );
+ final List<String> groupBundles = groups.stream()
+ .map(g ->
Projections.getClusterGroupBundleName(g.getClusteringValueIds()))
+ .toList();
+ final SegmentFileMetadata metadata = metadata(groupBundles,
List.of(clusteredBaseProjection(groups)));
+ Assertions.assertEquals(groupBundles,
spec().getSelectedBundleNames(anySegment(), metadata));
+ }
+
+ @Test
+ void testGetSelectedBundleNamesClusteredSegmentExcludesProjectionBundles()
+ {
+ // The base table is the rows, not the precomputation over them.
+ final List<TableClusterGroupSpec> groups = List.of(new
TableClusterGroupSpec(List.of(0), 10));
+ final String groupBundle =
Projections.getClusterGroupBundleName(List.of(0));
+ final SegmentFileMetadata metadata = metadata(
+ List.of(Projections.BASE_TABLE_PROJECTION_NAME, groupBundle,
"user_hourly"),
+ List.of(clusteredBaseProjection(groups), projection("user_hourly"))
+ );
+ Assertions.assertEquals(
+ List.of(Projections.BASE_TABLE_PROJECTION_NAME, groupBundle),
+ spec().getSelectedBundleNames(anySegment(), metadata)
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesLegacyRootOnlySegment()
+ {
+ // A V10 segment written before the bundle name was persisted reports
everything under __root__, which is then the
+ // whole segment. Gated on root being the sole bundle, matching
PartialSegmentBundleCacheEntry#resolveBundleName.
+ final SegmentFileMetadata metadata =
metadata(List.of(SegmentFileBuilder.ROOT_BUNDLE_NAME), null);
+ Assertions.assertEquals(
+ List.of(SegmentFileBuilder.ROOT_BUNDLE_NAME),
+ spec().getSelectedBundleNames(anySegment(), metadata)
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesThrowsWhenNoBaseBundlePresent()
+ {
+ // Named bundles but no __base and no clustering: the reader would fail
loudly on the acquire anyway, so say so
+ // here where the message can name the cause.
+ final SegmentFileMetadata metadata = metadata(List.of("user_hourly"),
List.of(projection("user_hourly")));
+ final DruidException thrown = Assertions.assertThrows(
+ DruidException.class,
+ () -> spec().getSelectedBundleNames(anySegment(), metadata)
+ );
+ Assertions.assertTrue(
+ thrown.getMessage().contains("no [__base] bundle among"),
+ "unexpected message: " + thrown.getMessage()
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesThrowsWhenDeclaredGroupBundleMissing()
+ {
+ // Metadata declares two cluster groups but the segment only carries one
group bundle: writer/reader drift.
+ final List<TableClusterGroupSpec> groups = List.of(
+ new TableClusterGroupSpec(List.of(0), 10),
+ new TableClusterGroupSpec(List.of(1), 20)
+ );
+ final SegmentFileMetadata metadata = metadata(
+ List.of(Projections.getClusterGroupBundleName(List.of(0))),
+ List.of(clusteredBaseProjection(groups))
+ );
+ final DruidException thrown = Assertions.assertThrows(
+ DruidException.class,
+ () -> spec().getSelectedBundleNames(anySegment(), metadata)
+ );
+ Assertions.assertTrue(
+ thrown.getMessage().contains("metadata declares cluster-group bundle"),
+ "unexpected message: " + thrown.getMessage()
+ );
+ }
+
+ @Test
+ void testLoadSegmentDelegatesToInner() throws Exception
+ {
+ 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
+ {
+ StubLoadSpec.RANGE_CALLS.set(0);
+ SegmentRangeReader reader = spec().openRangeReader();
+ Assertions.assertNotNull(reader);
+ Assertions.assertEquals(1, StubLoadSpec.RANGE_CALLS.get());
+ }
+
+ @Test
+ void testOpenRangeReaderReturnsNullWhenInnerDoesNotSupport() throws Exception
+ {
+ PartialBaseTableLoadSpec spec = new PartialBaseTableLoadSpec(
+ ImmutableMap.of("type", "stub", "path", "/", "supportsRange", false),
+ PartialBaseTableLoadSpec.FINGERPRINT,
+ jsonMapper
+ );
+ Assertions.assertNull(spec.openRangeReader());
+ }
+
+ @Test
+ void testRejectsNullDelegate()
+ {
+ Assertions.assertThrows(
+ NullPointerException.class,
+ () -> new PartialBaseTableLoadSpec(null,
PartialBaseTableLoadSpec.FINGERPRINT, jsonMapper)
+ );
+ }
+
+ @Test
+ void testRejectsNullFingerprint()
+ {
+ Assertions.assertThrows(
+ NullPointerException.class,
+ () -> new PartialBaseTableLoadSpec(DELEGATE, null, jsonMapper)
+ );
+ }
+
+ private PartialBaseTableLoadSpec spec()
+ {
+ return new PartialBaseTableLoadSpec(DELEGATE,
PartialBaseTableLoadSpec.FINGERPRINT, jsonMapper);
+ }
+
+ private static final RowSignature CLUSTERING_TENANT = RowSignature.builder()
+
.add("tenant", ColumnType.STRING)
+ .build();
+
+ /**
+ * Metadata carrying one container per entry of {@code bundleNames} — the
base-table spec only reads which bundles
+ * exist, not what is in them — plus the given projection list.
+ */
+ private static SegmentFileMetadata metadata(
+ List<String> bundleNames,
+ @Nullable List<ProjectionMetadata> projections
+ )
+ {
+ final List<SegmentFileContainerMetadata> containers = new
ArrayList<>(bundleNames.size());
+ long offset = 0;
+ for (String bundleName : bundleNames) {
+ containers.add(new SegmentFileContainerMetadata(offset, 100L,
bundleName));
+ offset += 100L;
+ }
+ return new SegmentFileMetadata(containers, Map.of(), null, null, null,
projections, null);
+ }
+
+ private static ProjectionMetadata unclusteredBaseProjection()
+ {
+ return new ProjectionMetadata(
+ 100,
+ new TableProjectionSchema(
+ VirtualColumns.EMPTY,
+ List.of(ColumnHolder.TIME_COLUMN_NAME, "tenant"),
+ null,
+ List.of(OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME))
+ )
+ );
+ }
+
+ private static ProjectionMetadata
clusteredBaseProjection(List<TableClusterGroupSpec> groups)
+ {
+ return new ProjectionMetadata(
+ groups.stream().mapToInt(TableClusterGroupSpec::getNumRows).sum(),
+ 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
+ )
+ );
+ }
+
+ private static ProjectionMetadata projection(String name)
+ {
+ return new ProjectionMetadata(
+ 10,
+ new AggregateProjectionSchema(
+ name,
+ null,
+ null,
+ VirtualColumns.EMPTY,
+ List.of("tenant"),
+ new AggregatorFactory[]{new CountAggregatorFactory("cnt")},
+ List.of(OrderBy.ascending("tenant"))
+ )
+ );
+ }
+
+ private static DataSegment anySegment()
+ {
+ 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/processing/src/test/java/org/apache/druid/segment/loading/PartialFullSegmentLoadSpecTest.java
b/processing/src/test/java/org/apache/druid/segment/loading/PartialFullSegmentLoadSpecTest.java
new file mode 100644
index 00000000000..b67e9e0a7e2
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/segment/loading/PartialFullSegmentLoadSpecTest.java
@@ -0,0 +1,277 @@
+/*
+ * 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.segment.file.SegmentFileBuilder;
+import org.apache.druid.segment.file.SegmentFileContainerMetadata;
+import org.apache.druid.segment.file.SegmentFileMetadata;
+import org.apache.druid.segment.projections.Projections;
+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.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class PartialFullSegmentLoadSpecTest
+{
+ private static final Map<String, Object> DELEGATE = ImmutableMap.of(
+ "type", "stub",
+ "path", "/var/druid/segments/foo"
+ );
+
+ private static ObjectMapper configuredMapper()
+ {
+ final ObjectMapper m = new DefaultObjectMapper();
+ final SimpleModule module = new SimpleModule();
+ module.registerSubtypes(PartialFullSegmentLoadSpec.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
+ {
+ PartialFullSegmentLoadSpec spec = spec();
+ String json = jsonMapper.writeValueAsString(spec);
+ LoadSpec reread = jsonMapper.readValue(json, LoadSpec.class);
+ Assertions.assertInstanceOf(PartialFullSegmentLoadSpec.class, reread);
+ Assertions.assertEquals(spec, reread);
+ }
+
+ @Test
+ void testWireFormHasPartialFullSegmentType() throws Exception
+ {
+ Map<String, Object> wireForm = jsonMapper.readValue(
+ jsonMapper.writeValueAsString(spec()),
+ new TypeReference<>()
+ {
+ }
+ );
+ Assertions.assertEquals("partialFullSegment", wireForm.get("type"));
+ Assertions.assertEquals(DELEGATE, wireForm.get("delegate"));
+ Assertions.assertEquals(PartialFullSegmentLoadSpec.FINGERPRINT,
wireForm.get("fingerprint"));
+ // No scheme-specific field: the selection is the segment's whole layout.
+ Assertions.assertEquals(3, wireForm.size());
+ }
+
+ @Test
+ void testFingerprintIsDistinctFromOtherLayoutDerivedLoads()
+ {
+ // A full-segment load and a base-table load put different amounts on the
historical, and an empty load puts
+ // nothing there. Sharing a fingerprint would hide a rule swap from the
coordinator's reconciliation.
+ Assertions.assertNotEquals(PartialBaseTableLoadSpec.FINGERPRINT,
PartialFullSegmentLoadSpec.FINGERPRINT);
+ Assertions.assertNotEquals("v1:partial-empty",
PartialFullSegmentLoadSpec.FINGERPRINT);
+ }
+
+ @Test
+ void testGetSelectedBundleNamesSelectsEveryBundle()
+ {
+ // Base table, a cluster group and a projection: all of them, no layout
branching.
+ final String groupBundle =
Projections.getClusterGroupBundleName(List.of(0));
+ final List<String> bundles =
List.of(Projections.BASE_TABLE_PROJECTION_NAME, groupBundle, "user_hourly");
+ Assertions.assertEquals(bundles,
spec().getSelectedBundleNames(anySegment(), metadata(bundles)));
+ }
+
+ @Test
+ void testGetSelectedBundleNamesDedupesContainersSharingABundle()
+ {
+ // A bundle spans as many containers as the writer needed; the selection
is by name.
+ final SegmentFileMetadata metadata = metadata(
+ List.of(Projections.BASE_TABLE_PROJECTION_NAME,
Projections.BASE_TABLE_PROJECTION_NAME, "user_hourly")
+ );
+ Assertions.assertEquals(
+ List.of(Projections.BASE_TABLE_PROJECTION_NAME, "user_hourly"),
+ spec().getSelectedBundleNames(anySegment(), metadata)
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesLegacyRootOnlySegment()
+ {
+ final SegmentFileMetadata metadata =
metadata(List.of(SegmentFileBuilder.ROOT_BUNDLE_NAME));
+ Assertions.assertEquals(
+ List.of(SegmentFileBuilder.ROOT_BUNDLE_NAME),
+ spec().getSelectedBundleNames(anySegment(), metadata)
+ );
+ }
+
+ @Test
+ void testGetSelectedBundleNamesThrowsWhenNoContainers()
+ {
+ final SegmentFileMetadata metadata = metadata(List.of());
+ final DruidException thrown = Assertions.assertThrows(
+ DruidException.class,
+ () -> spec().getSelectedBundleNames(anySegment(), metadata)
+ );
+ Assertions.assertTrue(
+ thrown.getMessage().contains("metadata declares no containers"),
+ "unexpected message: " + thrown.getMessage()
+ );
+ }
+
+ @Test
+ void testLoadSegmentDelegatesToInner() throws Exception
+ {
+ 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
+ {
+ StubLoadSpec.RANGE_CALLS.set(0);
+ SegmentRangeReader reader = spec().openRangeReader();
+ Assertions.assertNotNull(reader);
+ Assertions.assertEquals(1, StubLoadSpec.RANGE_CALLS.get());
+ }
+
+ @Test
+ void testOpenRangeReaderReturnsNullWhenInnerDoesNotSupport() throws Exception
+ {
+ PartialFullSegmentLoadSpec spec = new PartialFullSegmentLoadSpec(
+ ImmutableMap.of("type", "stub", "path", "/", "supportsRange", false),
+ PartialFullSegmentLoadSpec.FINGERPRINT,
+ jsonMapper
+ );
+ Assertions.assertNull(spec.openRangeReader());
+ }
+
+ @Test
+ void testRejectsNullDelegate()
+ {
+ Assertions.assertThrows(
+ NullPointerException.class,
+ () -> new PartialFullSegmentLoadSpec(null,
PartialFullSegmentLoadSpec.FINGERPRINT, jsonMapper)
+ );
+ }
+
+ @Test
+ void testRejectsNullFingerprint()
+ {
+ Assertions.assertThrows(
+ NullPointerException.class,
+ () -> new PartialFullSegmentLoadSpec(DELEGATE, null, jsonMapper)
+ );
+ }
+
+ private PartialFullSegmentLoadSpec spec()
+ {
+ return new PartialFullSegmentLoadSpec(DELEGATE,
PartialFullSegmentLoadSpec.FINGERPRINT, jsonMapper);
+ }
+
+ /**
+ * Metadata carrying one container per entry of {@code bundleNames} — this
spec only reads which bundles exist, not
+ * what is in them, and no projection list is needed since it does no layout
branching.
+ */
+ private static SegmentFileMetadata metadata(List<String> bundleNames)
+ {
+ final List<SegmentFileContainerMetadata> containers = new
ArrayList<>(bundleNames.size());
+ long offset = 0;
+ for (String bundleName : bundleNames) {
+ containers.add(new SegmentFileContainerMetadata(offset, 100L,
bundleName));
+ offset += 100L;
+ }
+ return new SegmentFileMetadata(containers, Map.of(), null, null, null,
null, null);
+ }
+
+ private static DataSegment anySegment()
+ {
+ 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 e3b615738fd..d2a004a25c6 100644
--- a/server/src/main/java/org/apache/druid/guice/PartialLoadSpecModule.java
+++ b/server/src/main/java/org/apache/druid/guice/PartialLoadSpecModule.java
@@ -25,15 +25,17 @@ 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.PartialBaseTableLoadSpec;
import org.apache.druid.segment.loading.PartialClusterGroupLoadSpec;
+import org.apache.druid.segment.loading.PartialFullSegmentLoadSpec;
import org.apache.druid.segment.loading.PartialProjectionLoadSpec;
import java.util.List;
/**
- * 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.
+ * Registers every {@link org.apache.druid.segment.loading.PartialLoadSpec}
subtype as a {@link LoadSpec} subtype 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
{
@@ -50,6 +52,8 @@ public class PartialLoadSpecModule implements DruidModule
new SimpleModule().registerSubtypes(
PartialProjectionLoadSpec.class,
PartialClusterGroupLoadSpec.class,
+ PartialBaseTableLoadSpec.class,
+ PartialFullSegmentLoadSpec.class,
CompositePartialLoadSpec.class
)
);
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/rules/CannotMatchBehavior.java
b/server/src/main/java/org/apache/druid/server/coordinator/rules/CannotMatchBehavior.java
index 4da44b7361c..0e99a820f1f 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/rules/CannotMatchBehavior.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/rules/CannotMatchBehavior.java
@@ -25,12 +25,15 @@ import com.fasterxml.jackson.annotation.JsonValue;
import javax.annotation.Nullable;
/**
- * Controls what happens when a {@link PartialLoadRule}'s {@link
PartialLoadMatcher} does not apply to a given segment,
- * for example, a {@link ProjectionPartialLoadMatcher} when faced with a
segment that doesn't have projections.
+ * Controls what happens when a {@link PartialLoadRule}'s {@link
PartialLoadMatcher} does not apply to a given segment.
+ * <p>
+ * Every value other than {@link #FALL_THROUGH} applies the rule; they differ
in how much of the segment the
+ * historical is asked to make resident. {@link #LOAD_ON_DEMAND} asks for none
of it up front, {@link #BASE_LOAD} for
+ * the 'base table' (no projections), and {@link #FULL_LOAD} for everything.
* <p>
* Unknown values deserialize to {@code null} so that an older coordinator
encountering a rule authored on a newer
- * version that introduces a new behavior falls back to the constructor's
default ({@link #FULL_LOAD}) rather than
- * failing to parse the rule.
+ * version that introduces a new behavior falls back to the constructor's
default ({@link #LOAD_ON_DEMAND}) rather
+ * than failing to parse the rule.
*/
public enum CannotMatchBehavior
{
@@ -40,7 +43,23 @@ public enum CannotMatchBehavior
FALL_THROUGH("fallThrough"),
/**
- * The rule applies and the segment is loaded in full on this tier.
+ * The rule applies and the segment is dispatched with no partial-load
wrapper at all. On a virtual-storage
+ * historical that means its bundles are fetched on demand as queries touch
them, rather than being made resident up
+ * front. This is the default, and the cheapest way to say "this rule covers
the segment, but don't pre-load anything
+ * specific for it".
+ */
+ LOAD_ON_DEMAND("loadOnDemand"),
+
+ /**
+ * The rule applies and the segment's base table is made resident: every row
is available, but no projections. The
+ * minimum that keeps the segment fully queryable without waiting on demand
fetches. See
+ * {@link org.apache.druid.segment.loading.PartialBaseTableLoadSpec}.
+ */
+ BASE_LOAD("baseLoad"),
+
+ /**
+ * The rule applies and every bundle the segment carries is made resident.
+ * See {@link org.apache.druid.segment.loading.PartialFullSegmentLoadSpec}.
*/
FULL_LOAD("fullLoad");
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadRule.java
b/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadRule.java
index b16942df6d5..d92a75f2ceb 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadRule.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/rules/PartialLoadRule.java
@@ -21,7 +21,10 @@ package org.apache.druid.server.coordinator.rules;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.druid.common.config.Configs;
+import org.apache.druid.error.DruidException;
import org.apache.druid.error.InvalidInput;
+import org.apache.druid.segment.loading.PartialBaseTableLoadSpec;
+import org.apache.druid.segment.loading.PartialFullSegmentLoadSpec;
import org.apache.druid.server.coordinator.loading.PartialLoadProfile;
import org.apache.druid.timeline.DataSegment;
import org.joda.time.DateTime;
@@ -32,9 +35,9 @@ import java.util.Objects;
/**
* Base class for rules that load only a subset of a segment on a tier. Pairs
a {@link PartialLoadMatcher} (which
- * produces the wrapped load-spec wire form and an accounting fingerprint when
it applies to a segment) with a
- * {@link CannotMatchBehavior} that controls whether the rule falls through or
full-loads when the matcher does not
- * apply.
+ * produces the wrapped load spec and an accounting fingerprint when it
applies to a segment) with a
+ * {@link CannotMatchBehavior} that decides what the rule does when the
matcher does not apply — fall through to the
+ * next rule, or apply anyway and ask for none, some, or all of the segment.
*/
public abstract class PartialLoadRule extends LoadRule
{
@@ -53,7 +56,7 @@ public abstract class PartialLoadRule extends LoadRule
throw InvalidInput.exception("matcher must not be null for a partial
load rule");
}
this.matcher = matcher;
- this.onCannotMatch = Configs.valueOrDefault(onCannotMatch,
CannotMatchBehavior.FULL_LOAD);
+ this.onCannotMatch = Configs.valueOrDefault(onCannotMatch,
CannotMatchBehavior.LOAD_ON_DEMAND);
}
@JsonProperty
@@ -84,7 +87,9 @@ public abstract class PartialLoadRule extends LoadRule
if (result != null) {
return true;
}
- return onCannotMatch == CannotMatchBehavior.FULL_LOAD;
+ // Every behavior other than FALL_THROUGH applies the rule; they differ
only in how much of the segment run()
+ // then asks the historical to make resident.
+ return onCannotMatch != CannotMatchBehavior.FALL_THROUGH;
}
@Override
@@ -99,12 +104,49 @@ public abstract class PartialLoadRule extends LoadRule
PartialLoadProfile.forRequest(result.wrappedLoadSpec(),
result.fingerprint()),
getTieredReplicants()
);
- } else {
- // Matcher does not apply, but the rule still applies because
onCannotMatch == FULL_LOAD (FALL_THROUGH would
- // have caused appliesTo to return false, so run wouldn't be invoked).
Route through the regular full-load
- // handler.
- handler.replicateSegment(segment, getTieredReplicants());
+ return;
}
+ // Matcher does not apply, but the rule still does — FALL_THROUGH would
have made appliesTo return false, so run
+ // wouldn't have been invoked. How much of the segment to make resident is
onCannotMatch's call.
+ switch (onCannotMatch) {
+ case LOAD_ON_DEMAND -> handler.replicateSegment(segment,
getTieredReplicants());
+ case BASE_LOAD -> replicateWholly(
+ segment,
+ handler,
+ PartialBaseTableLoadSpec.wireForm(segment.getLoadSpec(),
PartialBaseTableLoadSpec.FINGERPRINT),
+ PartialBaseTableLoadSpec.FINGERPRINT
+ );
+ case FULL_LOAD -> replicateWholly(
+ segment,
+ handler,
+ PartialFullSegmentLoadSpec.wireForm(segment.getLoadSpec(),
PartialFullSegmentLoadSpec.FINGERPRINT),
+ PartialFullSegmentLoadSpec.FINGERPRINT
+ );
+ default -> throw DruidException.defensive(
+ "Unreachable onCannotMatch[%s] in run() for segment[%s]; appliesTo
should have returned false",
+ onCannotMatch,
+ segment.getId()
+ );
+ }
+ }
+
+ /**
+ * Dispatches a layout-derived partial load — one whose selection is "the
base table" or "everything" rather than
+ * anything the matcher resolved. These still go through the partial-load
handler so the historical actually makes
+ * the bundles resident and announces a fingerprint the coordinator can
reconcile.
+ */
+ private void replicateWholly(
+ DataSegment segment,
+ SegmentActionHandler handler,
+ Map<String, Object> wrappedLoadSpec,
+ String fingerprint
+ )
+ {
+ handler.replicateSegmentPartially(
+ segment,
+ PartialLoadProfile.forRequest(wrappedLoadSpec, fingerprint),
+ getTieredReplicants()
+ );
}
@Override
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/rules/ProjectionPartialLoadMatcher.java
b/server/src/main/java/org/apache/druid/server/coordinator/rules/ProjectionPartialLoadMatcher.java
index a510ed540e3..982c3d73499 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/rules/ProjectionPartialLoadMatcher.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/rules/ProjectionPartialLoadMatcher.java
@@ -22,10 +22,10 @@ package org.apache.druid.server.coordinator.rules;
import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
import com.google.common.io.BaseEncoding;
+import org.apache.druid.segment.loading.PartialBaseTableLoadSpec;
import org.apache.druid.segment.loading.PartialProjectionLoadSpec;
import org.apache.druid.timeline.DataSegment;
-import javax.annotation.Nullable;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -33,12 +33,23 @@ import java.util.Map;
/**
* Base for {@link PartialLoadMatcher} implementations that decide which of a
segment's V10 projections to load.
* Subclasses supply the resolution policy via {@link
#resolveProjectionNames(DataSegment)}; this base handles
- * fingerprint computation and wraps the result into the {@code
partialProjection} load-spec wire form consumed
- * by the historical-side {@link PartialProjectionLoadSpec}.
+ * fingerprint computation and wraps the result into the {@code
partialProjection} load spec consumed by the
+ * historical-side {@link PartialProjectionLoadSpec}.
* <p>
* The fingerprint is a hash of what projections are partially loaded on a
segment by this rule; the data node will
* include this value in the segment announcement so that it can be used as a
lightweight value to compare against
* to handle things like rule change so that we can ensure that the 'right'
partial load is in place from run to run.
+ * <p>
+ * <b>Projection matchers always apply.</b> When none of the configured
projections are present on a segment, the
+ * matcher resolves to a {@link PartialBaseTableLoadSpec} (every row, no
projections) instead of going opaque. A
+ * projection is precomputation that is always recoverable from the base
table, so the base table is a correct
+ * substitute for one the segment doesn't carry, and it is strictly less data
than every bundle on the segment. This
+ * is the ordinary state of affairs mid-rollout, when a new projection is
being reindexed in and only some segments
+ * carry it yet.
+ * <p>
+ * Because {@link #match} therefore never returns {@code null}, a rule whose
only matcher is a projection matcher
+ * never consults its {@link CannotMatchBehavior}, and "segments lacking
projection p belong to a different rule" is
+ * not expressible via {@link CannotMatchBehavior#FALL_THROUGH}. Scope such
rules by interval or period instead.
*/
public abstract class ProjectionPartialLoadMatcher implements
PartialLoadMatcher
{
@@ -47,17 +58,23 @@ public abstract class ProjectionPartialLoadMatcher
implements PartialLoadMatcher
/**
* Returns the sorted, deduped list of projection names from {@link
DataSegment#getProjections()} that this matcher
* selects. Returns an empty list when nothing matches (the segment exposes
no projections, or no configured pattern
- * intersects what the segment has).
+ * intersects what the segment has), which {@link #match} turns into a
base-table load rather than a non-match.
*/
protected abstract List<String> resolveProjectionNames(DataSegment segment);
+ /**
+ * Never returns {@code null}; see the class doc. Either the resolved
projections, or a base-table load when none of
+ * them are present on {@code segment}.
+ */
@Override
- @Nullable
public MatchResult match(DataSegment segment, Map<String, Object>
baseLoadSpec)
{
final List<String> resolved = resolveProjectionNames(segment);
if (resolved.isEmpty()) {
- return null;
+ return new MatchResult(
+ PartialBaseTableLoadSpec.wireForm(baseLoadSpec,
PartialBaseTableLoadSpec.FINGERPRINT),
+ PartialBaseTableLoadSpec.FINGERPRINT
+ );
}
final String fingerprint = computeFingerprint(resolved);
return new MatchResult(PartialProjectionLoadSpec.wireForm(baseLoadSpec,
resolved, fingerprint), fingerprint);
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 87be5378926..a0ba9739b9c 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
@@ -177,6 +177,8 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
jsonMapper.registerSubtypes(new NamedType(LocalLoadSpec.class, "local"));
jsonMapper.registerSubtypes(new NamedType(PartialProjectionLoadSpec.class,
PartialProjectionLoadSpec.TYPE));
jsonMapper.registerSubtypes(new NamedType(CompositePartialLoadSpec.class,
CompositePartialLoadSpec.TYPE));
+ jsonMapper.registerSubtypes(new NamedType(PartialBaseTableLoadSpec.class,
PartialBaseTableLoadSpec.TYPE));
+ jsonMapper.registerSubtypes(new
NamedType(PartialFullSegmentLoadSpec.class, PartialFullSegmentLoadSpec.TYPE));
jsonMapper.registerModule(new SegmentizerModule());
jsonMapper.registerModules(new
LocalDataStorageDruidModule().getJacksonModules());
jsonMapper.setInjectableValues(
@@ -233,6 +235,75 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
Assertions.assertFalse(location.isReserved(aggId), "selected bundle should
NOT be in staticCacheEntries");
}
+ @Test
+ void testLoadBaseTableWrapperHoldsOnlyTheBaseBundle() throws Exception
+ {
+ // What a projection matcher emits when none of its projections are on the
segment. This fixture is unclustered,
+ // so the base table is the single __base bundle and none of the
projections come along for the ride. The
+ // clustered branch (every __base$* group) is covered in
PartialBaseTableLoadSpecTest.
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(baseTableWrapperSegment());
+
+ final PartialSegmentMetadataCacheEntry metadata =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertTrue(metadata.isRuleHeld(), "rule must be applied to the
metadata entry");
+ Assertions.assertEquals(PartialBaseTableLoadSpec.FINGERPRINT,
metadata.getRuleFingerprint());
+ Assertions.assertTrue(
+ metadata.isBundleRuleHeld(Projections.BASE_TABLE_PROJECTION_NAME),
+ "__base should be rule-held by a base-table load"
+ );
+
+ final PartialSegmentFileMapperV10 mapper = metadata.getFileMapper();
+ Assertions.assertNotNull(mapper, "metadata mount should produce a file
mapper");
+ Assertions.assertTrue(
+ mapper.isBundleFullyDownloaded(Projections.BASE_TABLE_PROJECTION_NAME),
+ "__base must be fully downloaded eagerly"
+ );
+
+ for (String projectionBundle : List.of(AGG_BUNDLE, OTHER_AGG_BUNDLE)) {
+ Assertions.assertFalse(
+ location.isWeakReserved(new
PartialSegmentBundleCacheEntryIdentifier(SEGMENT_ID, projectionBundle)),
+ "projection bundle[" + projectionBundle + "] must not be reserved by
a base-table load"
+ );
+ }
+ }
+
+ @Test
+ void testLoadFullSegmentWrapperHoldsEveryBundle() throws Exception
+ {
+ // What CannotMatchBehavior.FULL_LOAD emits. Unlike dispatching with no
wrapper — which on a virtual-storage
+ // historical only makes the segment available on demand — this pins every
bundle up front.
+ manager = makeManager(true, true);
+ final StorageLocation location = manager.getLocations().get(0);
+
+ manager.load(fullSegmentWrapperSegment());
+
+ final PartialSegmentMetadataCacheEntry metadata =
weakReservedMetadata(location, SEGMENT_ID);
+ Assertions.assertTrue(metadata.isRuleHeld(), "rule must be applied to the
metadata entry");
+ Assertions.assertEquals(PartialFullSegmentLoadSpec.FINGERPRINT,
metadata.getRuleFingerprint());
+
+ final PartialSegmentFileMapperV10 mapper = metadata.getFileMapper();
+ Assertions.assertNotNull(mapper, "metadata mount should produce a file
mapper");
+ for (String bundleName : List.of(Projections.BASE_TABLE_PROJECTION_NAME,
AGG_BUNDLE, OTHER_AGG_BUNDLE)) {
+ Assertions.assertTrue(
+ metadata.isBundleRuleHeld(bundleName),
+ "bundle[" + bundleName + "] should be rule-held by a full-segment
load"
+ );
+ Assertions.assertTrue(
+ mapper.isBundleFullyDownloaded(bundleName),
+ "bundle[" + bundleName + "] must be fully downloaded eagerly"
+ );
+ }
+ // Nothing on the segment was left out: not just the three bundles named
above, but every bundle the file carries.
+ for (String bundleName :
PartialSegmentBundleCacheEntry.bundleNames(mapper)) {
+ Assertions.assertTrue(
+ metadata.isBundleRuleHeld(bundleName),
+ "bundle[" + bundleName + "] present on the segment but not rule-held"
+ );
+ }
+ }
+
@Test
void testLoadCompositeWrapperInstallsRuleHoldsOnUnionOfMemberBundles()
throws Exception
{
@@ -900,6 +971,43 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
.build();
}
+ /**
+ * A {@code partialFullSegment} wrapper, matching what {@code
CannotMatchBehavior.FULL_LOAD} emits.
+ */
+ private DataSegment fullSegmentWrapperSegment()
+ {
+ final Map<String, Object> delegate = Map.of(
+ "type", "local",
+ "path", DEEP_STORAGE_DIR.getAbsolutePath()
+ );
+ return DataSegment.builder(SEGMENT_ID)
+ .shardSpec(NoneShardSpec.instance())
+ .loadSpec(
+ PartialFullSegmentLoadSpec.wireForm(delegate,
PartialFullSegmentLoadSpec.FINGERPRINT)
+ )
+ .size(0)
+ .build();
+ }
+
+ /**
+ * A {@code partialBaseTable} wrapper, matching what a projection matcher
emits when none of its configured
+ * projections are present on the segment.
+ */
+ private DataSegment baseTableWrapperSegment()
+ {
+ final Map<String, Object> delegate = Map.of(
+ "type", "local",
+ "path", DEEP_STORAGE_DIR.getAbsolutePath()
+ );
+ return DataSegment.builder(SEGMENT_ID)
+ .shardSpec(NoneShardSpec.instance())
+ .loadSpec(
+ PartialBaseTableLoadSpec.wireForm(delegate,
PartialBaseTableLoadSpec.FINGERPRINT)
+ )
+ .size(0)
+ .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.
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssignerPartialTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssignerPartialTest.java
index 7c9acaac794..0fe129bbba8 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssignerPartialTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssignerPartialTest.java
@@ -38,6 +38,7 @@ import
org.apache.druid.server.coordinator.rules.CannotMatchBehavior;
import
org.apache.druid.server.coordinator.rules.ExactProjectionPartialLoadMatcher;
import org.apache.druid.server.coordinator.rules.ForeverPartialLoadRule;
import org.apache.druid.server.coordinator.rules.PartialLoadRule;
+import
org.apache.druid.server.coordinator.rules.WildcardClusterGroupPartialLoadMatcher;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Stats;
import org.apache.druid.timeline.DataSegment;
@@ -673,10 +674,11 @@ public class StrategicSegmentAssignerPartialTest
}
@Test
- public void testForeverPartialLoadRuleEndToEndFullLoadFallback()
+ public void testForeverPartialLoadRuleEndToEndLoadOnDemandFallback()
{
- // Matcher does not apply (segment has no overlap); FULL_LOAD
onCannotMatch (default) → run() routes through
- // replicateSegment instead, so the peon must not see a profile and the
stat is the regular ASSIGNED.
+ // Matcher does not apply; LOAD_ON_DEMAND onCannotMatch (the default) →
run() routes through replicateSegment, so
+ // the peon must not see a profile and the stat is the regular ASSIGNED. A
cluster-group matcher supplies the
+ // non-match: projection matchers always apply, falling back to a
base-table load.
final ServerHolder server = createServer(TIER1);
final DruidCluster cluster = DruidCluster.builder().addTier(TIER1,
server).build();
@@ -684,8 +686,8 @@ public class StrategicSegmentAssignerPartialTest
final PartialLoadRule rule = new ForeverPartialLoadRule(
ImmutableMap.of(TIER1, 1),
null,
- new ExactProjectionPartialLoadMatcher(List.of("nonexistent")),
- CannotMatchBehavior.FULL_LOAD
+ new
WildcardClusterGroupPartialLoadMatcher(List.of(ImmutableMap.of("tenant",
"acme")), null),
+ CannotMatchBehavior.LOAD_ON_DEMAND
);
final DruidCoordinatorRuntimeParams params = makeRuntimeParams(cluster,
segment);
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/rules/ExactProjectionPartialLoadMatcherTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/rules/ExactProjectionPartialLoadMatcherTest.java
index ade05388eb1..2fccc989b60 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/rules/ExactProjectionPartialLoadMatcherTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/rules/ExactProjectionPartialLoadMatcherTest.java
@@ -26,6 +26,7 @@ import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.segment.loading.PartialBaseTableLoadSpec;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.partition.NumberedShardSpec;
@@ -79,33 +80,47 @@ public class ExactProjectionPartialLoadMatcherTest
}
@Test
- void testMatchReturnsNullWhenNoIntersection()
+ void testMatchFallsBackToBaseTableWhenNoIntersection()
{
+ // None of the configured projections are on the segment. The base table
can always answer what a projection
+ // would have, so the matcher asks for that rather than going opaque.
ExactProjectionPartialLoadMatcher matcher = new
ExactProjectionPartialLoadMatcher(
List.of("x", "y")
);
DataSegment segment = segmentWithProjections(List.of("a", "b"));
- Assertions.assertNull(matcher.match(segment, segment.getLoadSpec()));
+ assertBaseTableLoad(matcher.match(segment, segment.getLoadSpec()),
segment);
}
@Test
- void testMatchReturnsNullForProjectionAgnosticSegment()
+ void testMatchFallsBackToBaseTableForProjectionAgnosticSegment()
{
ExactProjectionPartialLoadMatcher matcher = new
ExactProjectionPartialLoadMatcher(
List.of("a")
);
DataSegment segment = segmentWithProjections(null);
- Assertions.assertNull(matcher.match(segment, segment.getLoadSpec()));
+ assertBaseTableLoad(matcher.match(segment, segment.getLoadSpec()),
segment);
}
@Test
- void testMatchReturnsNullForEmptyProjectionsList()
+ void testMatchFallsBackToBaseTableForEmptyProjectionsList()
{
ExactProjectionPartialLoadMatcher matcher = new
ExactProjectionPartialLoadMatcher(
List.of("a")
);
DataSegment segment = segmentWithProjections(Collections.emptyList());
- Assertions.assertNull(matcher.match(segment, segment.getLoadSpec()));
+ assertBaseTableLoad(matcher.match(segment, segment.getLoadSpec()),
segment);
+ }
+
+ @Test
+ void testBaseTableFallbackFingerprintIsStableAcrossConfigurations()
+ {
+ // Two differently-configured matchers that both find nothing produce the
same load, so they must agree on the
+ // fingerprint or the coordinator would see a rule change that isn't one.
+ DataSegment segment = segmentWithProjections(List.of("a", "b"));
+ Assertions.assertEquals(
+ new ExactProjectionPartialLoadMatcher(List.of("x")).match(segment,
segment.getLoadSpec()),
+ new ExactProjectionPartialLoadMatcher(List.of("y",
"z")).match(segment, segment.getLoadSpec())
+ );
}
@Test
@@ -162,6 +177,16 @@ public class ExactProjectionPartialLoadMatcherTest
EqualsVerifier.forClass(ExactProjectionPartialLoadMatcher.class).usingGetClass().verify();
}
+ private static void assertBaseTableLoad(PartialLoadMatcher.MatchResult
result, DataSegment segment)
+ {
+ Assertions.assertNotNull(result);
+ Assertions.assertEquals(PartialBaseTableLoadSpec.FINGERPRINT,
result.fingerprint());
+ Assertions.assertEquals(
+ PartialBaseTableLoadSpec.wireForm(segment.getLoadSpec(),
PartialBaseTableLoadSpec.FINGERPRINT),
+ result.wrappedLoadSpec()
+ );
+ }
+
private static DataSegment segmentWithProjections(List<String> projections)
{
return BUILDER.projections(projections).build();
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/rules/PartialLoadRuleTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/rules/PartialLoadRuleTest.java
index 35113529c01..66d263b202d 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/rules/PartialLoadRuleTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/rules/PartialLoadRuleTest.java
@@ -27,6 +27,8 @@ import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.segment.loading.PartialBaseTableLoadSpec;
+import org.apache.druid.segment.loading.PartialFullSegmentLoadSpec;
import org.apache.druid.server.coordinator.loading.PartialLoadProfile;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
@@ -87,7 +89,7 @@ public class PartialLoadRuleTest
false,
tier(1),
null,
- exact("nonexistent"),
+ cannotMatch(),
CannotMatchBehavior.FALL_THROUGH
);
DataSegment segment = segmentWithProjections(IN_WINDOW, List.of("a", "b"));
@@ -95,26 +97,45 @@ public class PartialLoadRuleTest
}
@Test
- void testAppliesToMatcherDoesNotApplyFullLoadIsDefault()
+ void testAppliesToMatcherDoesNotApplyDefaultStillApplies()
{
- // FULL_LOAD is the default. Segment has projections but matcher resolves
to nothing,
- // so the rule still applies and the segment gets full-loaded on this tier.
+ // LOAD_ON_DEMAND is the default, and like every behavior other than
FALL_THROUGH it still applies the rule; a
+ // matcher that cannot reason about the segment does not take the segment
out of this tier.
PeriodPartialLoadRule rule = new PeriodPartialLoadRule(
new Period("P30D"),
false,
tier(1),
null,
- exact("nonexistent"),
+ cannotMatch(),
null
);
DataSegment segment = segmentWithProjections(IN_WINDOW, List.of("a", "b"));
Assertions.assertTrue(rule.appliesTo(segment, NOW));
}
+ @Test
+ void testAppliesToProjectionMatcherWithNoMatchingProjection()
+ {
+ // A projection matcher always applies: no matching projection resolves to
a base-table load, not a non-match, so
+ // onCannotMatch never comes into play. Asserted under FALL_THROUGH
because that is where the old behavior
+ // (matcher opaque -> rule skipped) would have been visible.
+ PeriodPartialLoadRule rule = new PeriodPartialLoadRule(
+ new Period("P30D"),
+ false,
+ tier(1),
+ null,
+ exact("nonexistent"),
+ CannotMatchBehavior.FALL_THROUGH
+ );
+ DataSegment segment = segmentWithProjections(IN_WINDOW, List.of("a", "b"));
+ Assertions.assertTrue(rule.appliesTo(segment, NOW));
+ }
+
@Test
void testAppliesToProjectionAgnosticSegmentFallThrough()
{
- // Pre-Druid-32 segment: projections == null. With FALL_THROUGH, rule does
not apply.
+ // Pre-Druid-32 segment: projections == null. The projection matcher still
applies, resolving to a base-table
+ // load, so FALL_THROUGH does not skip the rule.
PeriodPartialLoadRule rule = new PeriodPartialLoadRule(
new Period("P30D"),
false,
@@ -124,23 +145,32 @@ public class PartialLoadRuleTest
CannotMatchBehavior.FALL_THROUGH
);
DataSegment segment = segmentWithProjections(IN_WINDOW, null);
- Assertions.assertFalse(rule.appliesTo(segment, NOW));
+ Assertions.assertTrue(rule.appliesTo(segment, NOW));
}
@Test
- void testAppliesToProjectionAgnosticSegmentFullLoad()
+ void testRunProjectionAgnosticSegmentRoutesToBaseTablePartialLoad()
{
- // Default behavior: pre-Druid-32 segments fall into full-load on this
tier.
+ // Pre-Druid-32 segment: projections == null. The projection matcher
resolves to a base-table load, so run()
+ // dispatches a partial load of the segment's rows rather than falling
back to whole-segment replication.
PeriodPartialLoadRule rule = new PeriodPartialLoadRule(
new Period("P30D"),
false,
- tier(1),
+ tier(2),
null,
exact("a"),
null
);
DataSegment segment = segmentWithProjections(IN_WINDOW, null);
- Assertions.assertTrue(rule.appliesTo(segment, NOW));
+ RecordingHandler handler = new RecordingHandler();
+ rule.run(segment, handler);
+ Assertions.assertEquals(0, handler.replicateCalls);
+ Assertions.assertEquals(1, handler.replicatePartialCalls);
+ Assertions.assertEquals(PartialBaseTableLoadSpec.FINGERPRINT,
handler.lastProfile.fingerprint());
+ Assertions.assertEquals(
+ PartialBaseTableLoadSpec.wireForm(segment.getLoadSpec(),
PartialBaseTableLoadSpec.FINGERPRINT),
+ handler.lastProfile.wrappedLoadSpec()
+ );
}
@Test
@@ -163,26 +193,35 @@ public class PartialLoadRuleTest
@Test
void testCascadeFallThroughToFullLoad()
{
- // Rule 1: partial { "a" } 30 days, explicit FALL_THROUGH when matcher
cannot match
+ // Rule 1: partial cluster-group load over 30 days, explicit FALL_THROUGH
when the matcher cannot match
PartialLoadRule partial = new PeriodPartialLoadRule(
new Period("P30D"),
false,
tier(1),
null,
- exact("a"),
+ cannotMatch(),
CannotMatchBehavior.FALL_THROUGH
);
- // Rule 2: forever full load (no projections required)
+ // Rule 2: forever full load
ForeverLoadRule full = new ForeverLoadRule(tier(1), null);
- // Segment with no projections. Partial rule falls through; cascade lands
on full.
- DataSegment legacy = segmentWithProjections(IN_WINDOW, null);
- Assertions.assertFalse(partial.appliesTo(legacy, NOW));
- Assertions.assertTrue(full.appliesTo(legacy, NOW));
+ // Non-clustered segment: the cluster-group matcher is opaque, so the
partial rule falls through and the cascade
+ // lands on full.
+ DataSegment unclustered = segmentWithProjections(IN_WINDOW, null);
+ Assertions.assertFalse(partial.appliesTo(unclustered, NOW));
+ Assertions.assertTrue(full.appliesTo(unclustered, NOW));
- // Segment with matching projection. Partial rule applies, cascade stops
there.
- DataSegment modern = segmentWithProjections(IN_WINDOW, List.of("a", "b"));
- Assertions.assertTrue(partial.appliesTo(modern, NOW));
+ // A projection rule, by contrast, always applies and so always stops the
cascade.
+ PartialLoadRule projectionPartial = new PeriodPartialLoadRule(
+ new Period("P30D"),
+ false,
+ tier(1),
+ null,
+ exact("a"),
+ CannotMatchBehavior.FALL_THROUGH
+ );
+ Assertions.assertTrue(projectionPartial.appliesTo(unclustered, NOW));
+
Assertions.assertTrue(projectionPartial.appliesTo(segmentWithProjections(IN_WINDOW,
List.of("a", "b")), NOW));
}
@Test
@@ -224,7 +263,7 @@ public class PartialLoadRuleTest
"matcher": {"type": "exactProjection", "names": ["a"]}\
}""";
PeriodPartialLoadRule rule = (PeriodPartialLoadRule)
OBJECT_MAPPER.readValue(json, Rule.class);
- Assertions.assertEquals(CannotMatchBehavior.FULL_LOAD,
rule.getOnCannotMatch());
+ Assertions.assertEquals(CannotMatchBehavior.LOAD_ON_DEMAND,
rule.getOnCannotMatch());
Assertions.assertEquals(PeriodLoadRule.DEFAULT_INCLUDE_FUTURE,
rule.isIncludeFuture());
Assertions.assertEquals(
Map.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS),
@@ -233,11 +272,11 @@ public class PartialLoadRuleTest
}
@Test
- void testUnknownOnCannotMatchValueDeserializesToFullLoadDefault() throws
Exception
+ void testUnknownOnCannotMatchValueDeserializesToDefault() throws Exception
{
// Simulates an older coordinator reading a rule authored by a newer
version that introduced
// a new CannotMatchBehavior value. The rule should parse, with the
unknown value falling
- // back to the constructor's default (FULL_LOAD) rather than failing
deserialization.
+ // back to the constructor's default (LOAD_ON_DEMAND) rather than failing
deserialization.
String json = """
{\
"type": "loadPartialByPeriod",\
@@ -246,7 +285,7 @@ public class PartialLoadRuleTest
"onCannotMatch": "SOME_FUTURE_BEHAVIOR"\
}""";
PeriodPartialLoadRule rule = (PeriodPartialLoadRule)
OBJECT_MAPPER.readValue(json, Rule.class);
- Assertions.assertEquals(CannotMatchBehavior.FULL_LOAD,
rule.getOnCannotMatch());
+ Assertions.assertEquals(CannotMatchBehavior.LOAD_ON_DEMAND,
rule.getOnCannotMatch());
}
@Test
@@ -318,17 +357,17 @@ public class PartialLoadRuleTest
}
@Test
- void testRunWithFullLoadFallbackRoutesToReplicateSegment()
+ void testRunWithLoadOnDemandFallbackRoutesToReplicateSegment()
{
- // Matcher does not apply but the rule's onCannotMatch is FULL_LOAD, so
run() falls back to the regular full-load
- // handler.
+ // Matcher does not apply and onCannotMatch is LOAD_ON_DEMAND, so run()
dispatches the segment with no partial
+ // wrapper at all — on a virtual-storage historical its bundles are then
fetched as queries touch them.
PeriodPartialLoadRule rule = new PeriodPartialLoadRule(
new Period("P30D"),
false,
tier(2),
null,
- exact("nonexistent"),
- CannotMatchBehavior.FULL_LOAD
+ cannotMatch(),
+ CannotMatchBehavior.LOAD_ON_DEMAND
);
DataSegment segment = segmentWithProjections(IN_WINDOW, List.of("a", "b"));
RecordingHandler handler = new RecordingHandler();
@@ -339,6 +378,72 @@ public class PartialLoadRuleTest
Assertions.assertNull(handler.lastProfile);
}
+ @Test
+ void testRunWithBaseLoadFallbackRoutesToBaseTablePartialLoad()
+ {
+ PeriodPartialLoadRule rule = new PeriodPartialLoadRule(
+ new Period("P30D"),
+ false,
+ tier(2),
+ null,
+ cannotMatch(),
+ CannotMatchBehavior.BASE_LOAD
+ );
+ DataSegment segment = segmentWithProjections(IN_WINDOW, List.of("a", "b"));
+ RecordingHandler handler = new RecordingHandler();
+ rule.run(segment, handler);
+ Assertions.assertEquals(0, handler.replicateCalls);
+ Assertions.assertEquals(1, handler.replicatePartialCalls);
+ Assertions.assertEquals(PartialBaseTableLoadSpec.FINGERPRINT,
handler.lastProfile.fingerprint());
+ Assertions.assertEquals(
+ PartialBaseTableLoadSpec.wireForm(segment.getLoadSpec(),
PartialBaseTableLoadSpec.FINGERPRINT),
+ handler.lastProfile.wrappedLoadSpec()
+ );
+ }
+
+ @Test
+ void testRunWithFullLoadFallbackRoutesToFullSegmentPartialLoad()
+ {
+ // FULL_LOAD means every bundle resident, which needs the partial-load
path: dispatching without a wrapper would
+ // only make the segment available on demand.
+ PeriodPartialLoadRule rule = new PeriodPartialLoadRule(
+ new Period("P30D"),
+ false,
+ tier(2),
+ null,
+ cannotMatch(),
+ CannotMatchBehavior.FULL_LOAD
+ );
+ DataSegment segment = segmentWithProjections(IN_WINDOW, List.of("a", "b"));
+ RecordingHandler handler = new RecordingHandler();
+ rule.run(segment, handler);
+ Assertions.assertEquals(0, handler.replicateCalls);
+ Assertions.assertEquals(1, handler.replicatePartialCalls);
+ Assertions.assertEquals(PartialFullSegmentLoadSpec.FINGERPRINT,
handler.lastProfile.fingerprint());
+ Assertions.assertEquals(
+ PartialFullSegmentLoadSpec.wireForm(segment.getLoadSpec(),
PartialFullSegmentLoadSpec.FINGERPRINT),
+ handler.lastProfile.wrappedLoadSpec()
+ );
+ }
+
+ @Test
+ void testBaseLoadAndFullLoadFingerprintsDiffer()
+ {
+ // Otherwise the coordinator could not tell a rule swap between the two
apart, and would never re-issue the load.
+ Assertions.assertNotEquals(PartialBaseTableLoadSpec.FINGERPRINT,
PartialFullSegmentLoadSpec.FINGERPRINT);
+ }
+
+ @Test
+ void testAllCannotMatchBehaviorsRoundTrip() throws Exception
+ {
+ for (CannotMatchBehavior behavior : CannotMatchBehavior.values()) {
+ ForeverPartialLoadRule rule = new ForeverPartialLoadRule(tier(1), null,
exact("a"), behavior);
+ Rule reread =
OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(rule), Rule.class);
+ Assertions.assertEquals(rule, reread, "round trip failed for " +
behavior);
+ Assertions.assertEquals(behavior, ((ForeverPartialLoadRule)
reread).getOnCannotMatch());
+ }
+ }
+
@Test
void testPeriodEquals()
{
@@ -384,6 +489,16 @@ public class PartialLoadRuleTest
return new ExactProjectionPartialLoadMatcher(Arrays.asList(names));
}
+ /**
+ * A matcher that cannot match any segment this test builds, so the rule's
{@link CannotMatchBehavior} decides.
+ * Projection matchers no longer serve this purpose: they always apply,
falling back to a base-table load when none
+ * of their projections are present. A cluster-group matcher is opaque to
the non-clustered segments here.
+ */
+ private static PartialLoadMatcher cannotMatch()
+ {
+ return new WildcardClusterGroupPartialLoadMatcher(List.of(Map.of("tenant",
"acme")), null);
+ }
+
private static Map<String, Integer> tier(int n)
{
return Map.of(DruidServer.DEFAULT_TIER, n);
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardProjectionPartialLoadMatcherTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardProjectionPartialLoadMatcherTest.java
index 0f1a12e85e8..b8d26801641 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardProjectionPartialLoadMatcherTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/rules/WildcardProjectionPartialLoadMatcherTest.java
@@ -26,6 +26,7 @@ import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.segment.loading.PartialBaseTableLoadSpec;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.partition.NumberedShardSpec;
@@ -187,13 +188,13 @@ public class WildcardProjectionPartialLoadMatcherTest
void testMatchReturnsNullWhenNoPatternsHit()
{
// Segment has projections but none match any configured pattern: distinct
code path from the
- // projection-agnostic-segment short-circuit.
+ // projection-agnostic-segment short-circuit. Both resolve to a base-table
load.
WildcardProjectionPartialLoadMatcher matcher = new
WildcardProjectionPartialLoadMatcher(
List.of("user_*"),
null
);
DataSegment segment = segmentWithProjections(List.of("session_daily",
"session_hourly", "other"));
- Assertions.assertNull(matcher.match(segment, segment.getLoadSpec()));
+ assertBaseTableLoad(matcher.match(segment, segment.getLoadSpec()),
segment.getLoadSpec());
}
@Test
@@ -215,14 +216,17 @@ public class WildcardProjectionPartialLoadMatcherTest
}
@Test
- void testReturnsNullForProjectionAgnosticSegment()
+ void testFallsBackToBaseTableForProjectionAgnosticSegment()
{
WildcardProjectionPartialLoadMatcher matcher = new
WildcardProjectionPartialLoadMatcher(
List.of("*"),
null
);
- Assertions.assertNull(matcher.match(segmentWithProjections(null),
BASE_LOAD_SPEC));
-
Assertions.assertNull(matcher.match(segmentWithProjections(Collections.emptyList()),
BASE_LOAD_SPEC));
+ assertBaseTableLoad(matcher.match(segmentWithProjections(null),
BASE_LOAD_SPEC), BASE_LOAD_SPEC);
+ assertBaseTableLoad(
+ matcher.match(segmentWithProjections(Collections.emptyList()),
BASE_LOAD_SPEC),
+ BASE_LOAD_SPEC
+ );
}
@Test
@@ -396,15 +400,26 @@ public class WildcardProjectionPartialLoadMatcherTest
}
@Test
- void testExcludeAllMatchedReturnsNull()
+ void testExcludeAllMatchedFallsBackToBaseTable()
{
- // If excludePatterns consume every match the result is empty; the matcher
reports "does not match".
+ // If excludePatterns consume every match the result is empty, which
resolves to a base-table load — the same
+ // outcome as a segment that never had the projections at all.
WildcardProjectionPartialLoadMatcher matcher = new
WildcardProjectionPartialLoadMatcher(
List.of("user_*"),
List.of("user_*")
);
DataSegment segment = segmentWithProjections(List.of("user_daily",
"user_hourly"));
- Assertions.assertNull(matcher.match(segment, segment.getLoadSpec()));
+ assertBaseTableLoad(matcher.match(segment, segment.getLoadSpec()),
segment.getLoadSpec());
+ }
+
+ private static void assertBaseTableLoad(PartialLoadMatcher.MatchResult
result, Map<String, Object> baseLoadSpec)
+ {
+ Assertions.assertNotNull(result);
+ Assertions.assertEquals(PartialBaseTableLoadSpec.FINGERPRINT,
result.fingerprint());
+ Assertions.assertEquals(
+ PartialBaseTableLoadSpec.wireForm(baseLoadSpec,
PartialBaseTableLoadSpec.FINGERPRINT),
+ result.wrappedLoadSpec()
+ );
}
@Test
diff --git
a/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
b/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
index 4f57d6cb31f..79c07484e45 100644
---
a/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
+++
b/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
@@ -60,6 +60,7 @@ import
org.apache.druid.server.coordinator.rules.IntervalDropRule;
import org.apache.druid.server.coordinator.rules.IntervalLoadRule;
import org.apache.druid.server.coordinator.rules.IntervalPartialLoadRule;
import org.apache.druid.server.coordinator.rules.Rule;
+import
org.apache.druid.server.coordinator.rules.WildcardClusterGroupPartialLoadMatcher;
import org.apache.druid.server.security.Access;
import org.apache.druid.server.security.Action;
import org.apache.druid.server.security.AuthConfig;
@@ -784,7 +785,7 @@ public class DataSourcesResourceTest
null,
null,
new ExactProjectionPartialLoadMatcher(ImmutableList.of("user_daily")),
- CannotMatchBehavior.FULL_LOAD
+ CannotMatchBehavior.LOAD_ON_DEMAND
);
DataSourcesResource dataSourcesResource =
new DataSourcesResource(
@@ -822,15 +823,16 @@ public class DataSourcesResourceTest
@Test
public void testIsHandOffCompleteWithPartialLoadRuleFallThrough()
{
- // A FALL_THROUGH partial rule whose matcher does not resolve on the
segment (the projection it asks for is not
- // present) should not halt the cascade. The next rule (drop) catches the
segment, so the response is true.
+ // A FALL_THROUGH partial rule whose matcher does not resolve on the
segment should not halt the cascade. The next
+ // rule (drop) catches the segment, so the response is true. The matcher
is a cluster-group one because the
+ // segment is not clustered: projection matchers always apply, falling
back to a base-table load.
MetadataRuleManager databaseRuleManager =
EasyMock.createMock(MetadataRuleManager.class);
Interval ruleInterval =
Intervals.of("2013-01-01T00:00:00Z/2013-01-03T00:00:00Z");
Rule partialRule = new IntervalPartialLoadRule(
ruleInterval,
null,
null,
- new ExactProjectionPartialLoadMatcher(ImmutableList.of("user_daily")),
+ new
WildcardClusterGroupPartialLoadMatcher(ImmutableList.of(ImmutableMap.of("tenant",
"acme")), null),
CannotMatchBehavior.FALL_THROUGH
);
Rule dropRule = new IntervalDropRule(ruleInterval);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
