You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "clintropolis (via GitHub)" <gi...@apache.org> on 2026/06/23 19:15:50 UTC
[PR] feat: partial load for clustered segments (druid)
clintropolis opened a new pull request, #19620:
URL: https://github.com/apache/druid/pull/19620
### Description
Wires up partial load support for clustered segments, which is mostly converting the existing non-partial clustered stuff to be on the partial stuff. I think as a follow-up I'll try to split out the 'clustered' implementations from the non-clustered stuff since it feels like a lot of 'if (clustered ...) do something' blocks and there is a bunch of stuff duplicated, but didn't seem appropriate to do in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: partial load for clustered segments (druid)
Posted by "capistrant (via GitHub)" <gi...@apache.org>.
capistrant commented on code in PR #19620:
URL: https://github.com/apache/druid/pull/19620#discussion_r3462670751
##########
server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java:
##########
@@ -113,10 +118,17 @@ class SegmentLocalCacheManagerPartialAcquireTest
new ListBasedInputRow(ROW_SIGNATURE, TIME.plusMinutes(3), ROW_SIGNATURE.getColumnNames(), Arrays.asList("b", 4L))
);
+ // A second segment that is both clustered and carries an aggregate projection (the apache/druid#19599 combo): no
Review Comment:
```suggestion
// A second segment that is both clustered and carries an aggregate projection: no
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: partial load for clustered segments (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #19620:
URL: https://github.com/apache/druid/pull/19620#discussion_r3462470269
##########
server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java:
##########
@@ -405,6 +478,62 @@
);
}
+ @Test
+ void testPartialAcquireClusteredWithProjectionMountsProjectionBundleWithoutBase()
+ throws ExecutionException, InterruptedException, IOException
+ {
+ final DataSegment clusteredSegment =
+ DataSegment.builder()
Review Comment:
## CodeQL / Deprecated method or constructor invocation
Invoking [DataSegment.builder](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/11316)
##########
processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryClusteredTest.java:
##########
@@ -0,0 +1,466 @@
+/*
+ * 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;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.asyncresource.AsyncResource;
+import org.apache.druid.common.asyncresource.AsyncResources;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.query.dimension.DefaultDimensionSpec;
+import org.apache.druid.query.filter.EqualityFilter;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.ValueType;
+import org.apache.druid.segment.data.CompressionStrategy;
+import org.apache.druid.segment.file.CountingRangeReader;
+import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
+import org.apache.druid.segment.incremental.IncrementalIndexSchema;
+import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Partial (on-demand) download coverage for a clustered base-table V10 segment. Cluster groups are resolved from
+ * metadata only ({@code planClusterGroupQuery}); the async cursor factory downloads one bundle ({@code __base$<ids>})
+ * per surviving group.
+ */
+class PartialQueryableIndexCursorFactoryClusteredTest extends PartialQueryableIndexCursorFactoryTestBase
+{
+ private static final long T0 = DateTimes.of("2025-01-01").getMillis();
+
+ // tenants sort to dictionary ids acme=0, globex=1 → bundles __base$0 (2 rows) and __base$1 (1 row)
+ private static final String ACME_BUNDLE = "__base$0";
+ private static final String GLOBEX_BUNDLE = "__base$1";
+
+ @TempDir
+ static File sharedTempDir;
+
+ private static File segmentDir;
+
+ @BeforeAll
+ static void buildSegment()
+ {
+ final ClusteredValueGroupsBaseTableProjectionSpec clusterSpec =
+ ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .columns(
+ new StringDimensionSchema("tenant"),
+ new StringDimensionSchema("region"),
+ new LongDimensionSchema("__time")
+ )
+ .clusteringColumns("tenant")
+ .build();
+ final IncrementalIndexSchema schema =
+ IncrementalIndexSchema.builder()
+ .withMinTimestamp(T0)
+ .withTimestampSpec(new TimestampSpec("ts", "millis", null))
+ .withQueryGranularity(Granularities.NONE)
+ .withDimensionsSpec(clusterSpec.getDimensionsSpec())
+ .withRollup(false)
+ .withClusterSpec(clusterSpec)
+ .build();
+ final File tmpDir = new File(sharedTempDir, "build_" + ThreadLocalRandom.current().nextInt());
+ segmentDir = IndexBuilder.create()
+ .useV10()
+ .tmpDir(tmpDir)
+ .segmentWriteOutMediumFactory(OffHeapMemorySegmentWriteOutMediumFactory.instance())
+ .schema(schema)
+ .indexSpec(IndexSpec.builder().withMetadataCompression(CompressionStrategy.NONE).build())
+ // ingest out of clustering order; the writer sorts groups by clustering value
+ .rows(List.of(
+ row(T0 + 2, "globex", "eu-west-1"),
+ row(T0, "acme", "us-east-1"),
+ row(T0 + 1, "acme", "us-west-2")
+ ))
+ .buildMMappedIndexFile();
+ }
+
+ @Test
+ void testFilterOnClusteringColumnDownloadsOnlySurvivingGroup() throws IOException
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader, "filter_acme")) {
+ final PartialQueryableIndexCursorFactory factory = factory(opened.index());
+ final CursorBuildSpec spec = CursorBuildSpec.builder()
+ .setFilter(new EqualityFilter("tenant", ColumnType.STRING, "acme", null))
+ .build();
+
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(spec);
+ CursorHolder holder = asyncHolder.release()) {
+ // Only the acme group survives the clustering-column filter, so only its rows are returned.
+ Assertions.assertEquals(
+ List.of(List.of("acme", "us-east-1"), List.of("acme", "us-west-2")),
+ scanTenantRegion(holder)
+ );
+
+ final Set<String> downloaded = opened.mapper().getDownloadedFiles();
+ // The acme group's columns were materialized.
+ Assertions.assertTrue(downloaded.contains(ACME_BUNDLE + "/region"), "got: " + downloaded);
+ Assertions.assertTrue(downloaded.contains(ACME_BUNDLE + "/__time"), "got: " + downloaded);
+ // The globex group bundle was pruned by metadata-only resolution and never touched.
+ Assertions.assertTrue(
+ downloaded.stream().noneMatch(f -> f.startsWith(GLOBEX_BUNDLE + "/")),
+ "globex group must not be downloaded; got: " + downloaded
+ );
+ }
+ }
+ }
+
+ @Test
+ void testFilterMatchingNoGroupDownloadsNothing() throws IOException
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader, "filter_none")) {
+ final PartialQueryableIndexCursorFactory factory = factory(opened.index());
+ final CursorBuildSpec spec = CursorBuildSpec.builder()
+ .setFilter(new EqualityFilter("tenant", ColumnType.STRING, "nobody", null))
+ .build();
+
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(spec);
+ CursorHolder holder = asyncHolder.release()) {
+ Assertions.assertTrue(scanTenantRegion(holder).isEmpty(), "no group should survive");
+ Assertions.assertTrue(
+ opened.mapper().getDownloadedFiles().isEmpty(),
+ "no group bundle should be downloaded when nothing survives; got: " + opened.mapper().getDownloadedFiles()
+ );
+ }
+ }
+ }
+
+ @Test
+ void testFullScanConcatenatesAllGroups() throws IOException
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader, "full_scan")) {
+ final PartialQueryableIndexCursorFactory factory = factory(opened.index());
+
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(CursorBuildSpec.FULL_SCAN);
+ CursorHolder holder = asyncHolder.release()) {
+ // Groups are concatenated in clustering (dictionary id) order: acme then globex, clustering constant injected.
+ Assertions.assertEquals(
+ List.of(
+ List.of("acme", "us-east-1"),
+ List.of("acme", "us-west-2"),
+ List.of("globex", "eu-west-1")
+ ),
+ scanTenantRegion(holder)
+ );
+
+ final Set<String> downloaded = opened.mapper().getDownloadedFiles();
+ Assertions.assertTrue(downloaded.contains(ACME_BUNDLE + "/region"), "got: " + downloaded);
+ Assertions.assertTrue(downloaded.contains(GLOBEX_BUNDLE + "/region"), "got: " + downloaded);
+ }
+ }
+ }
+
+ @Test
+ void testSyncMakeCursorHolderAfterFullDownload() throws IOException
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader, "sync_full")) {
+ final PartialSegmentFileMapperV10 mapper = opened.mapper();
+ final PartialQueryableIndexCursorFactory factory = factory(opened.index());
+
+ // Sync path refuses until everything is resident.
+ Assertions.assertThrows(
+ DruidException.class,
+ () -> factory.makeCursorHolder(CursorBuildSpec.FULL_SCAN)
+ );
+
+ // Eager-acquire equivalent: materialize every internal file, then the sync clustered path works via the delegate.
+ mapper.ensureFilesAvailable(mapper.getSegmentFileMetadata().getFiles().keySet());
+ Assertions.assertTrue(mapper.isFullyDownloaded());
+ try (CursorHolder holder = factory.makeCursorHolder(CursorBuildSpec.FULL_SCAN)) {
+ Assertions.assertEquals(
+ List.of(
+ List.of("acme", "us-east-1"),
+ List.of("acme", "us-west-2"),
+ List.of("globex", "eu-west-1")
+ ),
+ scanTenantRegion(holder)
+ );
+ }
+ }
+ }
+
+ @Test
+ void testAsyncMultiGroupDefersDownloadUntilExecutorRuns() throws Exception
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ final CountDownLatch gate = new CountDownLatch(1);
+ final ExecutorService rawExec = Execs.singleThreaded("partial-clustered-defer-%d");
+ final ListeningExecutorService gatedExec = MoreExecutors.listeningDecorator(rawExec);
+ try (IndexAndMapper opened = openIndex(rangeReader, "async_defer")) {
+ final PartialQueryableIndexCursorFactory factory = new PartialQueryableIndexCursorFactory(
+ opened.index(),
+ QueryableIndexTimeBoundaryInspector.create(opened.index()),
+ noOpAcquirer(gatedExec)
+ );
+ rangeReader.resetCount();
+
+ // Block the download executor so no group download can start yet.
+ @SuppressWarnings("unused")
+ ListenableFuture<?> unused = gatedExec.submit(() -> {
+ try {
+ gate.await();
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
Review Comment:
## CodeQL / Unread local variable
Variable 'ListenableFuture<?> unused' is never read.
[Show more details](https://github.com/apache/druid/security/code-scanning/11317)
##########
processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryClusteredTest.java:
##########
@@ -0,0 +1,466 @@
+/*
+ * 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;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.common.asyncresource.AsyncResource;
+import org.apache.druid.common.asyncresource.AsyncResources;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.query.dimension.DefaultDimensionSpec;
+import org.apache.druid.query.filter.EqualityFilter;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.ValueType;
+import org.apache.druid.segment.data.CompressionStrategy;
+import org.apache.druid.segment.file.CountingRangeReader;
+import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
+import org.apache.druid.segment.incremental.IncrementalIndexSchema;
+import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Partial (on-demand) download coverage for a clustered base-table V10 segment. Cluster groups are resolved from
+ * metadata only ({@code planClusterGroupQuery}); the async cursor factory downloads one bundle ({@code __base$<ids>})
+ * per surviving group.
+ */
+class PartialQueryableIndexCursorFactoryClusteredTest extends PartialQueryableIndexCursorFactoryTestBase
+{
+ private static final long T0 = DateTimes.of("2025-01-01").getMillis();
+
+ // tenants sort to dictionary ids acme=0, globex=1 → bundles __base$0 (2 rows) and __base$1 (1 row)
+ private static final String ACME_BUNDLE = "__base$0";
+ private static final String GLOBEX_BUNDLE = "__base$1";
+
+ @TempDir
+ static File sharedTempDir;
+
+ private static File segmentDir;
+
+ @BeforeAll
+ static void buildSegment()
+ {
+ final ClusteredValueGroupsBaseTableProjectionSpec clusterSpec =
+ ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .columns(
+ new StringDimensionSchema("tenant"),
+ new StringDimensionSchema("region"),
+ new LongDimensionSchema("__time")
+ )
+ .clusteringColumns("tenant")
+ .build();
+ final IncrementalIndexSchema schema =
+ IncrementalIndexSchema.builder()
+ .withMinTimestamp(T0)
+ .withTimestampSpec(new TimestampSpec("ts", "millis", null))
+ .withQueryGranularity(Granularities.NONE)
+ .withDimensionsSpec(clusterSpec.getDimensionsSpec())
+ .withRollup(false)
+ .withClusterSpec(clusterSpec)
+ .build();
+ final File tmpDir = new File(sharedTempDir, "build_" + ThreadLocalRandom.current().nextInt());
+ segmentDir = IndexBuilder.create()
+ .useV10()
+ .tmpDir(tmpDir)
+ .segmentWriteOutMediumFactory(OffHeapMemorySegmentWriteOutMediumFactory.instance())
+ .schema(schema)
+ .indexSpec(IndexSpec.builder().withMetadataCompression(CompressionStrategy.NONE).build())
+ // ingest out of clustering order; the writer sorts groups by clustering value
+ .rows(List.of(
+ row(T0 + 2, "globex", "eu-west-1"),
+ row(T0, "acme", "us-east-1"),
+ row(T0 + 1, "acme", "us-west-2")
+ ))
+ .buildMMappedIndexFile();
+ }
+
+ @Test
+ void testFilterOnClusteringColumnDownloadsOnlySurvivingGroup() throws IOException
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader, "filter_acme")) {
+ final PartialQueryableIndexCursorFactory factory = factory(opened.index());
+ final CursorBuildSpec spec = CursorBuildSpec.builder()
+ .setFilter(new EqualityFilter("tenant", ColumnType.STRING, "acme", null))
+ .build();
+
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(spec);
+ CursorHolder holder = asyncHolder.release()) {
+ // Only the acme group survives the clustering-column filter, so only its rows are returned.
+ Assertions.assertEquals(
+ List.of(List.of("acme", "us-east-1"), List.of("acme", "us-west-2")),
+ scanTenantRegion(holder)
+ );
+
+ final Set<String> downloaded = opened.mapper().getDownloadedFiles();
+ // The acme group's columns were materialized.
+ Assertions.assertTrue(downloaded.contains(ACME_BUNDLE + "/region"), "got: " + downloaded);
+ Assertions.assertTrue(downloaded.contains(ACME_BUNDLE + "/__time"), "got: " + downloaded);
+ // The globex group bundle was pruned by metadata-only resolution and never touched.
+ Assertions.assertTrue(
+ downloaded.stream().noneMatch(f -> f.startsWith(GLOBEX_BUNDLE + "/")),
+ "globex group must not be downloaded; got: " + downloaded
+ );
+ }
+ }
+ }
+
+ @Test
+ void testFilterMatchingNoGroupDownloadsNothing() throws IOException
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader, "filter_none")) {
+ final PartialQueryableIndexCursorFactory factory = factory(opened.index());
+ final CursorBuildSpec spec = CursorBuildSpec.builder()
+ .setFilter(new EqualityFilter("tenant", ColumnType.STRING, "nobody", null))
+ .build();
+
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(spec);
+ CursorHolder holder = asyncHolder.release()) {
+ Assertions.assertTrue(scanTenantRegion(holder).isEmpty(), "no group should survive");
+ Assertions.assertTrue(
+ opened.mapper().getDownloadedFiles().isEmpty(),
+ "no group bundle should be downloaded when nothing survives; got: " + opened.mapper().getDownloadedFiles()
+ );
+ }
+ }
+ }
+
+ @Test
+ void testFullScanConcatenatesAllGroups() throws IOException
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader, "full_scan")) {
+ final PartialQueryableIndexCursorFactory factory = factory(opened.index());
+
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(CursorBuildSpec.FULL_SCAN);
+ CursorHolder holder = asyncHolder.release()) {
+ // Groups are concatenated in clustering (dictionary id) order: acme then globex, clustering constant injected.
+ Assertions.assertEquals(
+ List.of(
+ List.of("acme", "us-east-1"),
+ List.of("acme", "us-west-2"),
+ List.of("globex", "eu-west-1")
+ ),
+ scanTenantRegion(holder)
+ );
+
+ final Set<String> downloaded = opened.mapper().getDownloadedFiles();
+ Assertions.assertTrue(downloaded.contains(ACME_BUNDLE + "/region"), "got: " + downloaded);
+ Assertions.assertTrue(downloaded.contains(GLOBEX_BUNDLE + "/region"), "got: " + downloaded);
+ }
+ }
+ }
+
+ @Test
+ void testSyncMakeCursorHolderAfterFullDownload() throws IOException
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader, "sync_full")) {
+ final PartialSegmentFileMapperV10 mapper = opened.mapper();
+ final PartialQueryableIndexCursorFactory factory = factory(opened.index());
+
+ // Sync path refuses until everything is resident.
+ Assertions.assertThrows(
+ DruidException.class,
+ () -> factory.makeCursorHolder(CursorBuildSpec.FULL_SCAN)
+ );
+
+ // Eager-acquire equivalent: materialize every internal file, then the sync clustered path works via the delegate.
+ mapper.ensureFilesAvailable(mapper.getSegmentFileMetadata().getFiles().keySet());
+ Assertions.assertTrue(mapper.isFullyDownloaded());
+ try (CursorHolder holder = factory.makeCursorHolder(CursorBuildSpec.FULL_SCAN)) {
+ Assertions.assertEquals(
+ List.of(
+ List.of("acme", "us-east-1"),
+ List.of("acme", "us-west-2"),
+ List.of("globex", "eu-west-1")
+ ),
+ scanTenantRegion(holder)
+ );
+ }
+ }
+ }
+
+ @Test
+ void testAsyncMultiGroupDefersDownloadUntilExecutorRuns() throws Exception
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ final CountDownLatch gate = new CountDownLatch(1);
+ final ExecutorService rawExec = Execs.singleThreaded("partial-clustered-defer-%d");
+ final ListeningExecutorService gatedExec = MoreExecutors.listeningDecorator(rawExec);
+ try (IndexAndMapper opened = openIndex(rangeReader, "async_defer")) {
+ final PartialQueryableIndexCursorFactory factory = new PartialQueryableIndexCursorFactory(
+ opened.index(),
+ QueryableIndexTimeBoundaryInspector.create(opened.index()),
+ noOpAcquirer(gatedExec)
+ );
+ rangeReader.resetCount();
+
+ // Block the download executor so no group download can start yet.
+ @SuppressWarnings("unused")
+ ListenableFuture<?> unused = gatedExec.submit(() -> {
+ try {
+ gate.await();
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+
+ // FULL_SCAN survives both groups → both bundles must download before the holder is ready.
+ final AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(CursorBuildSpec.FULL_SCAN);
+ Assertions.assertFalse(asyncHolder.isReady(), "must wait while the download executor is blocked");
+ Assertions.assertEquals(0, rangeReader.getReadCount(), "no group download should have started yet");
+
+ final CountDownLatch ready = new CountDownLatch(1);
+ asyncHolder.addReadyCallback(ready::countDown);
+ gate.countDown();
+ Assertions.assertTrue(ready.await(15, TimeUnit.SECONDS), "ready callback must fire once downloads run");
+ Assertions.assertTrue(asyncHolder.isReady());
+
+ try (CursorHolder holder = asyncHolder.release()) {
+ Assertions.assertEquals(
+ List.of(
+ List.of("acme", "us-east-1"),
+ List.of("acme", "us-west-2"),
+ List.of("globex", "eu-west-1")
+ ),
+ scanTenantRegion(holder)
+ );
+ }
+ }
+ finally {
+ gatedExec.shutdownNow();
+ rawExec.shutdownNow();
+ }
+ }
+
+ @Test
+ void testMultiGroupSuccessCloseReleasesEveryBundleHold() throws IOException
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ final CountingBundleAcquirer acquirer = new CountingBundleAcquirer(directExec());
+ try (IndexAndMapper opened = openIndex(rangeReader, "release_success")) {
+ final PartialQueryableIndexCursorFactory factory = new PartialQueryableIndexCursorFactory(
+ opened.index(),
+ QueryableIndexTimeBoundaryInspector.create(opened.index()),
+ acquirer
+ );
+
+ // FULL_SCAN survives both groups → both bundles acquired during the build; with the direct executor the
+ // downloads run inline so the holder is immediately ready.
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(CursorBuildSpec.FULL_SCAN)) {
+ Assertions.assertTrue(asyncHolder.isReady());
+ Assertions.assertEquals(1, acquirer.acquiredCount(ACME_BUNDLE), "acme group bundle acquired");
+ Assertions.assertEquals(1, acquirer.acquiredCount(GLOBEX_BUNDLE), "globex group bundle acquired");
+
+ final CursorHolder holder = asyncHolder.release();
+ // The produced holder owns every group's hold; nothing is released until it closes.
+ Assertions.assertEquals(0, acquirer.releasedCount(ACME_BUNDLE));
+ Assertions.assertEquals(0, acquirer.releasedCount(GLOBEX_BUNDLE));
+
+ holder.close();
+ // Closing the holder fans out the release across all surviving groups' bundle holds.
+ Assertions.assertEquals(1, acquirer.releasedCount(ACME_BUNDLE), "acme hold released on holder close");
+ Assertions.assertEquals(1, acquirer.releasedCount(GLOBEX_BUNDLE), "globex hold released on holder close");
+ }
+ }
+ }
+
+ @Test
+ void testMultiGroupCancelBeforeReadyReleasesEveryBundleHold() throws Exception
+ {
+ final CountingRangeReader rangeReader = new CountingRangeReader(segmentDir);
+ final CountDownLatch gate = new CountDownLatch(1);
+ final ExecutorService rawExec = Execs.singleThreaded("partial-clustered-cancel-%d");
+ final ListeningExecutorService gatedExec = MoreExecutors.listeningDecorator(rawExec);
+ final CountingBundleAcquirer acquirer = new CountingBundleAcquirer(gatedExec);
+ try (IndexAndMapper opened = openIndex(rangeReader, "release_cancel")) {
+ final PartialQueryableIndexCursorFactory factory = new PartialQueryableIndexCursorFactory(
+ opened.index(),
+ QueryableIndexTimeBoundaryInspector.create(opened.index()),
+ acquirer
+ );
+
+ // Block the download executor so the per-column download bodies stay queued and the holder never becomes ready.
+ @SuppressWarnings("unused")
+ ListenableFuture<?> unused = gatedExec.submit(() -> {
+ try {
+ gate.await();
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
Review Comment:
## CodeQL / Unread local variable
Variable 'ListenableFuture<?> unused' is never read.
[Show more details](https://github.com/apache/druid/security/code-scanning/11318)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: partial load for clustered segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis merged PR #19620:
URL: https://github.com/apache/druid/pull/19620
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org