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 000000000000..dbd34cdca24e --- /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. + *

+ * 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. + *

+ * 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 wireForm(Map delegate, String fingerprint) + { + return Map.of( + TYPE_FIELD, TYPE, + DELEGATE_FIELD, delegate, + FINGERPRINT_FIELD, fingerprint + ); + } + + @JsonCreator + public PartialBaseTableLoadSpec( + @JsonProperty("delegate") Map 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: + *

+ * 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 getSelectedBundleNames(DataSegment segment, SegmentFileMetadata metadata) + { + final Set present = presentBundleNames(metadata); + if (present.size() == 1 && present.contains(SegmentFileBuilder.ROOT_BUNDLE_NAME)) { + return List.of(SegmentFileBuilder.ROOT_BUNDLE_NAME); + } + + final List 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 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 clusterGroupBundleNames(SegmentFileMetadata metadata) + { + final List projections = metadata.getProjections(); + if (projections == null || projections.isEmpty()) { + return null; + } + if (!(projections.getFirst().getSchema() instanceof ClusteredValueGroupsBaseTableSchema clusteredSummary)) { + return null; + } + final List groups = clusteredSummary.getClusterGroups(); + final List 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 000000000000..deb2cd667ffd --- /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. + *

+ * This exists so a rule can ask for the whole segment to be resident. 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. + *

+ * 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 wireForm(Map delegate, String fingerprint) + { + return Map.of( + TYPE_FIELD, TYPE, + DELEGATE_FIELD, delegate, + FINGERPRINT_FIELD, fingerprint + ); + } + + @JsonCreator + public PartialFullSegmentLoadSpec( + @JsonProperty("delegate") Map 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 getSelectedBundleNames(DataSegment segment, SegmentFileMetadata metadata) + { + final Set 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 0e3174647c36..d9fc7de37719 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.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 SegmentRangeReader openRangeReader() throws IOException * bootstrap re-discovers the segment). */ public abstract List getSelectedBundleNames(DataSegment segment, SegmentFileMetadata metadata); + + /** + * The distinct bundle names carried by {@code metadata}, in container order. + */ + protected static Set presentBundleNames(SegmentFileMetadata metadata) + { + final Set 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 000000000000..9c06a1dbd47b --- /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 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 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 groups = List.of( + new TableClusterGroupSpec(List.of(0), 10), + new TableClusterGroupSpec(List.of(1), 20), + new TableClusterGroupSpec(List.of(2), 30) + ); + final List groupBundles = groups.stream() + .map(g -> Projections.getClusterGroupBundleName(g.getClusteringValueIds())) + .toList(); + final List present = new ArrayList<>(); + present.add(Projections.BASE_TABLE_PROJECTION_NAME); + present.addAll(groupBundles); + final SegmentFileMetadata metadata = metadata(present, List.of(clusteredBaseProjection(groups))); + + final List 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 groups = List.of( + new TableClusterGroupSpec(List.of(0), 10), + new TableClusterGroupSpec(List.of(1), 20) + ); + final List 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 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 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 bundleNames, + @Nullable List projections + ) + { + final List 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 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 000000000000..b67e9e0a7e20 --- /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 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 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 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 bundleNames) + { + final List 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 e3b615738fdc..d2a004a25c63 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 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 List getJacksonModules() 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 4da44b7361c2..0e99a820f1f7 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 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. + *

+ * 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. *

* 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 b16942df6d51..d92a75f2ceb4 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 @@ 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 @@ /** * 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 @@ protected PartialLoadRule( 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 boolean appliesTo(DataSegment segment, DateTime referenceTimestamp) 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 void run(DataSegment segment, SegmentActionHandler handler) 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 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 a510ed540e3b..982c3d734995 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 @@ 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 @@ /** * 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}. *

* 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. + *

+ * Projection matchers always apply. 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. + *

+ * 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 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 baseLoadSpec) { final List 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 87be5378926d..a0ba9739b9c9 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 @@ void setup() throws IOException 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 @@ void testLoadInstallsRuleHoldsOnMetadataAndSelectedBundle() throws Exception 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 @@ private DataSegment partialWrapperSegment(List selectedProjections, Stri .build(); } + /** + * A {@code partialFullSegment} wrapper, matching what {@code CannotMatchBehavior.FULL_LOAD} emits. + */ + private DataSegment fullSegmentWrapperSegment() + { + final Map 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 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 c6290b1f7768..4c9a1917b941 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.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; @@ -671,10 +672,11 @@ public void testForeverPartialLoadRuleEndToEndThreadsProfile() } @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(); @@ -682,8 +684,8 @@ public void testForeverPartialLoadRuleEndToEndFullLoadFallback() 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 ade05388eb1a..2fccc989b60b 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.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 @@ void testMatchProducesResultWhenIntersectionNonEmpty() } @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 @@ void testEquals() 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 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 35113529c010..66d263b202df 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.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 @@ void testAppliesToMatcherDoesNotApplyFallThroughReturnsFalse() false, tier(1), null, - exact("nonexistent"), + cannotMatch(), CannotMatchBehavior.FALL_THROUGH ); DataSegment segment = segmentWithProjections(IN_WINDOW, List.of("a", "b")); @@ -95,26 +97,45 @@ void testAppliesToMatcherDoesNotApplyFallThroughReturnsFalse() } @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 @@ void testAppliesToProjectionAgnosticSegmentFallThrough() 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 @@ void testIntervalOverloadIgnoresMatcher() @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 @@ void testPeriodSerdeDefaults() throws Exception "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 @@ void testPeriodSerdeDefaults() throws Exception } @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 @@ void testUnknownOnCannotMatchValueDeserializesToFullLoadDefault() throws Excepti "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 @@ void testRunWithMatchRoutesToReplicateSegmentPartially() } @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 @@ void testRunWithFullLoadFallbackRoutesToReplicateSegment() 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 @@ private static PartialLoadMatcher exact(String... names) 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 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 0f1a12e85e8a..b8d268016419 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.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 @@ void testMatchIsCaseSensitive() 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 @@ void testMultiplePatternsUnioned() } @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 @@ void testExcludeNotMatchedIsNoop() } @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 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 4f57d6cb31ff..79c07484e458 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.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 void testIsHandOffCompleteForcesMetadataRefreshOnSnapshotMiss() null, null, new ExactProjectionPartialLoadMatcher(ImmutableList.of("user_daily")), - CannotMatchBehavior.FULL_LOAD + CannotMatchBehavior.LOAD_ON_DEMAND ); DataSourcesResource dataSourcesResource = new DataSourcesResource( @@ -822,15 +823,16 @@ public void testIsHandOffCompleteForcesMetadataRefreshOnSnapshotMiss() @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);