diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedMSQPartialDownloadsTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedMSQPartialDownloadsTest.java new file mode 100644 index 000000000000..af3581b6563e --- /dev/null +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedMSQPartialDownloadsTest.java @@ -0,0 +1,204 @@ +/* + * 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.testing.embedded.msq; + +import org.apache.druid.common.utils.IdUtils; +import org.apache.druid.data.input.impl.LocalInputSource; +import org.apache.druid.indexer.TaskState; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.msq.indexing.report.MSQTaskReportPayload; +import org.apache.druid.msq.util.MultiStageQueryContext; +import org.apache.druid.query.DruidProcessingConfigTest; +import org.apache.druid.query.http.SqlTaskStatus; +import org.apache.druid.server.metrics.LatchableEmitter; +import org.apache.druid.server.metrics.StorageMonitor; +import org.apache.druid.sql.calcite.planner.Calcites; +import org.apache.druid.testing.embedded.EmbeddedBroker; +import org.apache.druid.testing.embedded.EmbeddedCoordinator; +import org.apache.druid.testing.embedded.EmbeddedDruidCluster; +import org.apache.druid.testing.embedded.EmbeddedHistorical; +import org.apache.druid.testing.embedded.EmbeddedIndexer; +import org.apache.druid.testing.embedded.EmbeddedOverlord; +import org.apache.druid.testing.embedded.EmbeddedRouter; +import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.util.Collections; +import java.util.Map; +import java.util.Objects; + +/** + * End-to-end wiring test for on-demand partial segment downloads on MSQ worker tasks. Partial downloads default to + * enabled for tasks (the {@code druid.indexer.task.virtualStoragePartialDownloadsEnabled} {@code TaskConfig} default is + * true), so run 1 sets no property or context and verifies the default engages the partial path. Run 2 verifies the + * per-query {@link MultiStageQueryContext#CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS} override can turn it back off. + *

+ * The observable signal is the indexer's {@link StorageMonitor#VSF_READ_COUNT} metric: the partial path performs + * on-demand deep-storage range reads (so the count is positive), while the full-download path does not (the count + * stays zero because {@code virtualStoragePartialDownloadsEnabled=false} makes the worker mount segments as complete + * cache entries and download them in full instead of via range reads). + */ +class EmbeddedMSQPartialDownloadsTest extends EmbeddedClusterTestBase +{ + // A few times the storage monitor's PT1s emission period, so a single missed tick doesn't falsely read as "idle". + private static final long MONITOR_QUIESCE_TIMEOUT_MILLIS = 3_000L; + + private static final String SELECT_SQL = "SELECT channel, COUNT(*) AS c FROM \"%s\" GROUP BY channel"; + + private final EmbeddedBroker broker = new EmbeddedBroker(); + private final EmbeddedIndexer indexer = new EmbeddedIndexer(); + private final EmbeddedOverlord overlord = new EmbeddedOverlord(); + private final EmbeddedHistorical historical = new EmbeddedHistorical(); + private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator(); + private final EmbeddedRouter router = new EmbeddedRouter(); + + private EmbeddedMSQApis msqApis; + + @Override + public EmbeddedDruidCluster createCluster() + { + indexer.setServerMemory(400_000_000) + .addProperty("druid.worker.capacity", "4") + .addProperty("druid.processing.numThreads", "3"); + + broker.setServerMemory(200_000_000); + + coordinator.addProperty("druid.manager.segments.useIncrementalCache", "always"); + overlord.addProperty("druid.manager.segments.useIncrementalCache", "always") + .addProperty("druid.manager.segments.pollDuration", "PT0.1s"); + + return EmbeddedDruidCluster + .withEmbeddedDerbyAndZookeeper() + .useLatchableEmitter() + .useDefaultTimeoutForLatchableEmitter(20) + // Partial downloads require unzipped, V10 segments (range-readable via LocalLoadSpec's directory reader). + .addCommonProperty("druid.storage.zip", "false") + .addCommonProperty("druid.indexer.task.buildV10", "true") + .addCommonProperty("druid.monitoring.emissionPeriod", "PT1s") + .addServer(coordinator) + .addServer(overlord) + .addServer(indexer) + .addServer(historical) + .addServer(broker) + .addServer(router); + } + + @BeforeAll + void loadData() throws IOException + { + msqApis = new EmbeddedMSQApis(cluster, overlord); + dataSource = "wiki-" + IdUtils.getRandomId(); + loadWikiData(); + } + + @Override + protected void refreshDatasourceName() + { + // Keep the datasource stable: it is ingested once in loadData() before all tests. + } + + @Test + void testPartialDownloadsWiredViaRuntimePropertyAndQueryContext() + { + final LatchableEmitter emitter = indexer.latchableEmitter(); + + // Run 1: no context override, so the query uses the indexer's TaskConfig default (partial enabled). The worker + // reads the input segment's columns on demand, which shows up as deep-storage range reads. + emitter.awaitMetricQuiescent(StorageMonitor.VSF_READ_COUNT, MONITOR_QUIESCE_TIMEOUT_MILLIS); + emitter.flush(); + runMsqSelect(Collections.emptyMap()); + // The worker's StorageMonitor only emits while the (ephemeral) task is alive, so the VSF_READ_COUNT events land + // during runMsqSelect. waitForEventAggregate (unlike waitForNextEvent) also considers events already emitted since + // the last flush, so it sees them; it fails the test (timeout) if no on-demand range reads ever happened. + emitter.waitForEventAggregate( + matcher -> matcher.hasMetricName(StorageMonitor.VSF_READ_COUNT), + aggregate -> aggregate.hasSumAtLeast(1L) + ); + + // Run 2: the query context disables partial downloads, so the worker downloads segments in full and performs no + // range reads. The flag is passed as the string "false" (as it can arrive from a client): it is copied into the + // worker task context and must be parsed, not unboxed, when building the toolbox and worker context. + emitter.awaitMetricQuiescent(StorageMonitor.VSF_READ_COUNT, MONITOR_QUIESCE_TIMEOUT_MILLIS); + emitter.flush(); + runMsqSelect(Map.of(MultiStageQueryContext.CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS, "false")); + emitter.awaitMetricQuiescent(StorageMonitor.VSF_READ_COUNT, MONITOR_QUIESCE_TIMEOUT_MILLIS); + Assertions.assertEquals( + 0L, + emitter.getMetricEventLongSum(StorageMonitor.VSF_READ_COUNT), + "with partial downloads disabled via query context, the worker should download in full (no range reads)" + ); + } + + private void runMsqSelect(Map queryContext) + { + final SqlTaskStatus status = msqApis.submitTaskSql(queryContext, SELECT_SQL, dataSource); + cluster.callApi().waitForTaskToSucceed(status.getTaskId(), overlord); + } + + private void loadWikiData() throws IOException + { + final File tmpDir = cluster.getTestFolder().newFolder(); + final File wikiFile = new File(tmpDir, "wiki.gz"); + try (var in = DruidProcessingConfigTest.class.getResourceAsStream("/wikipedia/wikiticker-2015-09-12-sampled.json.gz")) { + Files.copy( + Objects.requireNonNull(in, "wikiticker sample resource not found on the test classpath"), + wikiFile.toPath() + ); + } + + final String sql = StringUtils.format( + "SET waitUntilSegmentsLoad = TRUE;\n" + + "REPLACE INTO \"%s\" OVERWRITE ALL\n" + + "SELECT\n" + + " TIME_PARSE(\"time\") AS __time,\n" + + " channel,\n" + + " countryName,\n" + + " page,\n" + + " \"user\",\n" + + " added,\n" + + " deleted,\n" + + " delta\n" + + "FROM TABLE(\n" + + " EXTERN(\n" + + " %s,\n" + + " '{\"type\":\"json\"}',\n" + + " '[{\"name\":\"channel\",\"type\":\"string\"},{\"name\":\"time\",\"type\":\"string\"},{\"name\":\"page\",\"type\":\"string\"},{\"name\":\"added\",\"type\":\"long\"},{\"name\":\"user\",\"type\":\"string\"},{\"name\":\"delta\",\"type\":\"long\"},{\"name\":\"deleted\",\"type\":\"long\"},{\"name\":\"countryName\",\"type\":\"string\"}]'\n" + + " )\n" + + " )\n" + + "PARTITIONED BY DAY\n" + + "CLUSTERED BY channel", + dataSource, + Calcites.escapeStringLiteral( + broker.bindings() + .jsonMapper() + .writeValueAsString(new LocalInputSource(null, null, Collections.singletonList(wikiFile), null)) + ) + ); + + final MSQTaskReportPayload payload = msqApis.runTaskSqlAndGetReport(sql); + Assertions.assertEquals(TaskState.SUCCESS, payload.getStatus().getStatus()); + Assertions.assertNull(payload.getStatus().getErrorReport()); + } +} diff --git a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactoryTest.java b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactoryTest.java index 7d0c3f20a6ba..160c3eabd9e3 100644 --- a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactoryTest.java +++ b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactoryTest.java @@ -209,7 +209,7 @@ private TestMultipleKubernetesTaskRunnerFactory createFactory( new NoopServiceEmitter(), () -> null, configManager, - new TaskConfig(null, null, false, null, null, null, false, null, false, null, false), + new TaskConfig(null, null, false, null, null, null, false, null, false, null, false, null), new StartupLoggingConfig(), new DruidNode("test-overlord", "localhost", false, 8080, null, true, false), new DruidKubernetesVertxHttpClientFactory(new DruidKubernetesVertxHttpClientConfig(), objectMapper) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java index 402e4be10c52..9897338f9fa8 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java @@ -37,7 +37,6 @@ import org.apache.druid.timeline.DataSegment; import java.io.File; -import java.util.Collections; import java.util.List; /** @@ -47,6 +46,11 @@ public class SegmentCacheManagerFactory { private final IndexIO indexIO; private final ObjectMapper jsonMapper; + /** + * The injected {@code druid.segmentCache} config. Per-task caches are derived from it (see {@link #manufacturate}) + * so operator-tuned virtual-storage settings. + */ + private final SegmentLoaderConfig segmentLoaderConfig; /** * Resolved lazily (only on the {@code virtualStorage=true} path of {@link #manufacturate}) so that injecting this * factory into a component that only builds non-virtual caches (e.g. {@code DruidInputSource}) does not force the @@ -59,11 +63,13 @@ public class SegmentCacheManagerFactory public SegmentCacheManagerFactory( IndexIO indexIO, @Json ObjectMapper mapper, + SegmentLoaderConfig segmentLoaderConfig, @EphemeralStorageLoading Provider loadingThreadPoolProvider ) { this.indexIO = indexIO; this.jsonMapper = mapper; + this.segmentLoaderConfig = segmentLoaderConfig; this.loadingThreadPoolProvider = loadingThreadPoolProvider; } @@ -75,10 +81,12 @@ public SegmentCacheManagerFactory( */ public static SegmentCacheManagerFactory createWithOwnedPool(IndexIO indexIO, ObjectMapper mapper) { + final SegmentLoaderConfig defaults = SegmentLoaderConfig.builder().build(); return new SegmentCacheManagerFactory( indexIO, mapper, - Providers.of(StorageLoadingThreadPool.createFromConfig(new SegmentLoaderConfig().setVirtualStorage(true))) + defaults, + Providers.of(StorageLoadingThreadPool.createFromConfig(defaults.toEphemeralVirtualStorage())) ); } @@ -88,24 +96,41 @@ public static SegmentCacheManagerFactory createWithOwnedPool(IndexIO indexIO, Ob * {@link EphemeralStorageLoading} loading pool (shared across all per-task caches and stopped by the lifecycle) * rather than creating its own. * - * @param storageDir storage location - * @param maxSize size limit, or null for no limit - * @param virtualStorage whether to configure the cache manager in ephemeral virtual storage mode. In this mode, - * loading is triggered by {@link SegmentCacheManager#acquireSegment(DataSegment, AcquireMode)}, - * and segment files are deleted as soon as all holds are closed. + * @param storageDir storage location + * @param maxSize size limit, or null for no limit + * @param virtualStorage whether to configure the cache manager in ephemeral virtual storage mode. In this + * mode, loading is triggered by + * {@link SegmentCacheManager#acquireSegment(DataSegment, AcquireMode)}, and segment + * files are deleted as soon as all holds are closed. + * @param partialDownloadsEnabled when true (and {@code virtualStorage} is true), partial-eligible segments are read + * via on-demand column downloads rather than downloaded in full up front. Has no + * effect when {@code virtualStorage} is false. */ - public SegmentCacheManager manufacturate(File storageDir, Long maxSize, boolean virtualStorage) + public SegmentCacheManager manufacturate( + File storageDir, + Long maxSize, + boolean virtualStorage, + boolean partialDownloadsEnabled + ) { final StorageLocationConfig locationConfig = new StorageLocationConfig( storageDir, maxSize != null ? maxSize : Long.MAX_VALUE, null ); + // For virtual storage, derive an ephemeral virtual-storage cache config from the injected node config: this keeps + // its virtual-storage tuning while dropping the classic on-disk-cache settings that don't apply. A non-virtual + // (eager) per-task cache needs none of that tuning and must NOT inherit the node's transient settings, so it + // starts from fresh defaults instead. Both then get the per-task location and mode flags. + final SegmentLoaderConfig.Builder loaderConfigBuilder = + virtualStorage ? segmentLoaderConfig.toEphemeralVirtualStorage().toBuilder() : SegmentLoaderConfig.builder(); final SegmentLoaderConfig loaderConfig = - new SegmentLoaderConfig() - .setLocations(Collections.singletonList(locationConfig)) - .setVirtualStorage(virtualStorage) - .setVirtualStorageIsEphemeral(virtualStorage); + loaderConfigBuilder + .locations(locationConfig) + .virtualStorage(virtualStorage) + .virtualStorageIsEphemeral(virtualStorage) + .virtualStoragePartialDownloadsEnabled(partialDownloadsEnabled) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); return new SegmentLocalCacheManager( storageLocations, diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java index 6ee32438d49c..655e3399e2af 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java @@ -45,6 +45,7 @@ import org.apache.druid.java.util.emitter.service.ServiceEmitter; import org.apache.druid.java.util.metrics.MonitorScheduler; import org.apache.druid.query.DruidProcessingConfig; +import org.apache.druid.query.QueryContext; import org.apache.druid.query.QueryProcessingPool; import org.apache.druid.query.QueryRunnerFactoryConglomerate; import org.apache.druid.query.policy.PolicyEnforcer; @@ -247,7 +248,16 @@ public TaskToolbox build(TaskConfig config, Task task) .queryProcessingPool(queryProcessingPool) .joinableFactory(joinableFactory) .monitorSchedulerProvider(monitorSchedulerProvider) - .segmentCacheManager(segmentCacheManagerFactory.manufacturate(taskWorkDir, null, true)) + .segmentCacheManager(segmentCacheManagerFactory.manufacturate( + taskWorkDir, + null, + true, + // getBoolean (not the raw getContextValue) so a string "true"/"false" is parsed + QueryContext.of(task.getContext()).getBoolean( + Tasks.VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY, + config.isVirtualStoragePartialDownloadsEnabled() + ) + )) .jsonMapper(jsonMapper) .taskWorkDir(taskWorkDir) .indexIO(indexIO) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java index d043fa7c8254..1b65b9f68435 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java @@ -45,6 +45,7 @@ public class TaskConfig implements TaskDirectory private static final Period DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT = new Period("PT5M"); private static final boolean DEFAULT_STORE_EMPTY_COLUMNS = true; private static final long DEFAULT_TMP_STORAGE_BYTES_PER_TASK = -1; + private static final boolean DEFAULT_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_ENABLED = true; @JsonProperty private final String baseDir; @@ -79,6 +80,9 @@ public class TaskConfig implements TaskDirectory @JsonProperty private final boolean buildV10; + @JsonProperty + private final boolean virtualStoragePartialDownloadsEnabled; + @JsonCreator public TaskConfig( @JsonProperty("baseDir") String baseDir, @@ -91,7 +95,8 @@ public TaskConfig( @JsonProperty("storeEmptyColumns") @Nullable Boolean storeEmptyColumns, @JsonProperty("encapsulatedTask") boolean enableTaskLevelLogPush, @JsonProperty("tmpStorageBytesPerTask") @Nullable Long tmpStorageBytesPerTask, - @JsonProperty("buildV10") boolean buildV10 + @JsonProperty("buildV10") boolean buildV10, + @JsonProperty("virtualStoragePartialDownloadsEnabled") @Nullable Boolean virtualStoragePartialDownloadsEnabled ) { this.baseDir = Configs.valueOrDefault(baseDir, System.getProperty("java.io.tmpdir")); @@ -118,6 +123,10 @@ public TaskConfig( this.storeEmptyColumns = Configs.valueOrDefault(storeEmptyColumns, DEFAULT_STORE_EMPTY_COLUMNS); this.tmpStorageBytesPerTask = Configs.valueOrDefault(tmpStorageBytesPerTask, DEFAULT_TMP_STORAGE_BYTES_PER_TASK); this.buildV10 = buildV10; + this.virtualStoragePartialDownloadsEnabled = Configs.valueOrDefault( + virtualStoragePartialDownloadsEnabled, + DEFAULT_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_ENABLED + ); } private TaskConfig( @@ -131,7 +140,8 @@ private TaskConfig( boolean storeEmptyColumns, boolean encapsulatedTask, long tmpStorageBytesPerTask, - boolean buildV10 + boolean buildV10, + boolean virtualStoragePartialDownloadsEnabled ) { this.baseDir = baseDir; @@ -145,6 +155,7 @@ private TaskConfig( this.encapsulatedTask = encapsulatedTask; this.tmpStorageBytesPerTask = tmpStorageBytesPerTask; this.buildV10 = buildV10; + this.virtualStoragePartialDownloadsEnabled = virtualStoragePartialDownloadsEnabled; } @JsonProperty @@ -243,6 +254,12 @@ public boolean buildV10() return buildV10; } + @JsonProperty + public boolean isVirtualStoragePartialDownloadsEnabled() + { + return virtualStoragePartialDownloadsEnabled; + } + private String defaultDir(@Nullable String configParameter, final String defaultVal) { if (configParameter == null) { @@ -265,7 +282,8 @@ public TaskConfig withBaseTaskDir(File baseTaskDir) storeEmptyColumns, encapsulatedTask, tmpStorageBytesPerTask, - buildV10 + buildV10, + virtualStoragePartialDownloadsEnabled ); } @@ -282,7 +300,8 @@ public TaskConfig withTmpStorageBytesPerTask(long tmpStorageBytesPerTask) storeEmptyColumns, encapsulatedTask, tmpStorageBytesPerTask, - buildV10 + buildV10, + virtualStoragePartialDownloadsEnabled ); } } diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java index 7f64a6cbad0d..5e09198828e7 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java @@ -55,6 +55,7 @@ public class Tasks public static final String SUB_TASK_TIMEOUT_KEY = "subTaskTimeoutMillis"; public static final String FORCE_TIME_CHUNK_LOCK_KEY = "forceTimeChunkLock"; public static final String STORE_EMPTY_COLUMNS_KEY = "storeEmptyColumns"; + public static final String VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY = "virtualStoragePartialDownloadsEnabled"; public static final String USE_SHARED_LOCK = "useSharedLock"; public static final String TASK_LOCK_TYPE = "taskLockType"; public static final String USE_CONCURRENT_LOCKS = "useConcurrentLocks"; diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java index 84fead9916bb..917936c03c56 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java @@ -314,7 +314,7 @@ public DruidInputSource withInterval(Interval interval) protected InputSourceReader fixedFormatReader(InputRowSchema inputRowSchema, @Nullable File temporaryDirectory) { final SegmentCacheManager segmentCacheManager = - segmentCacheManagerFactory.manufacturate(temporaryDirectory, null, false); + segmentCacheManagerFactory.manufacturate(temporaryDirectory, null, false, false); final List> timeline = createTimeline(); final Iterator entityIterator = FluentIterable diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/SegmentCacheManagerFactoryTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/SegmentCacheManagerFactoryTest.java index 876224257943..b24f9e6f0843 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/SegmentCacheManagerFactoryTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/SegmentCacheManagerFactoryTest.java @@ -19,6 +19,7 @@ package org.apache.druid.indexing.common; +import com.fasterxml.jackson.databind.InjectableValues; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.inject.Provider; import com.google.inject.util.Providers; @@ -27,8 +28,10 @@ import org.apache.druid.segment.TestIndex; import org.apache.druid.segment.loading.SegmentCacheManager; import org.apache.druid.segment.loading.SegmentLoaderConfig; +import org.apache.druid.segment.loading.SegmentLocalCacheManager; import org.apache.druid.segment.loading.StorageLoadingThreadPool; import org.apache.druid.server.metrics.NoopServiceEmitter; +import org.apache.druid.utils.RuntimeInfo; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; @@ -60,12 +63,17 @@ void setUp() void testVirtualStorageManagersShareTheInjectedPool() { final StorageLoadingThreadPool shared = - StorageLoadingThreadPool.createFromConfig(new SegmentLoaderConfig().withVirtualStorage(true)); + StorageLoadingThreadPool.createFromConfig(SegmentLoaderConfig.builder().build().toEphemeralVirtualStorage()); try { final SegmentCacheManagerFactory factory = - new SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper, Providers.of(shared)); - final SegmentCacheManager m1 = factory.manufacturate(new File(tempDir, "a"), null, true); - final SegmentCacheManager m2 = factory.manufacturate(new File(tempDir, "b"), null, true); + new SegmentCacheManagerFactory( + TestIndex.INDEX_IO, + jsonMapper, + SegmentLoaderConfig.builder().build(), + Providers.of(shared) + ); + final SegmentCacheManager m1 = factory.manufacturate(new File(tempDir, "a"), null, true, false); + final SegmentCacheManager m2 = factory.manufacturate(new File(tempDir, "b"), null, true, false); Assertions.assertSame(shared, m1.getLoadingThreadPool()); Assertions.assertSame(shared, m2.getLoadingThreadPool()); @@ -84,19 +92,111 @@ void testNonVirtualStorageDoesNotResolveTheLoadingPool() throw new AssertionError("ephemeral loading pool must not be resolved for virtualStorage=false"); }; final SegmentCacheManagerFactory factory = - new SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper, throwingProvider); + new SegmentCacheManagerFactory( + TestIndex.INDEX_IO, + jsonMapper, + SegmentLoaderConfig.builder().build(), + throwingProvider + ); - final SegmentCacheManager m = factory.manufacturate(new File(tempDir, "c"), null, false); + final SegmentCacheManager m = factory.manufacturate(new File(tempDir, "c"), null, false, false); Assertions.assertFalse(m.getLoadingThreadPool().isAvailable()); } + @Test + void testManufacturateDerivesPerTaskConfigFromInjectedConfig() throws Exception + { + // The injected druid.segmentCache config carries operator-tuned virtual-storage settings and (typical of an + // indexing node) is not itself in virtual-storage mode. Deriving per-task caches from it must preserve that tuning. + final ObjectMapper mapper = TestHelper.makeJsonMapper(); + mapper.setInjectableValues(new InjectableValues.Std().addValue(RuntimeInfo.class, new RuntimeInfo())); + final SegmentLoaderConfig injected = mapper.readValue( + "{\"virtualStorageMetadataReservationEstimate\": 99999999," + + " \"virtualStorageCoalesceGapBytes\": 65536," + + " \"virtualStorageMaxFetchRunBytes\": 8388608," + + " \"infoDir\": \"/var/druid/segment-cache/info_dir\"," + + " \"numThreadsToLoadSegmentsIntoPageCacheOnDownload\": 4}", + SegmentLoaderConfig.class + ); + + final StorageLoadingThreadPool shared = + StorageLoadingThreadPool.createFromConfig(SegmentLoaderConfig.builder().build().toEphemeralVirtualStorage()); + try { + final SegmentCacheManagerFactory factory = + new SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper, injected, Providers.of(shared)); + final SegmentLocalCacheManager m = + (SegmentLocalCacheManager) factory.manufacturate(new File(tempDir, "task"), null, true, true); + final SegmentLoaderConfig derived = m.getConfig(); + + // Operator-tuned settings are carried over from the injected config... + Assertions.assertEquals(99999999L, derived.getVirtualStorageMetadataReservationEstimate()); + Assertions.assertEquals(65536L, derived.getVirtualStorageCoalesceGapBytes()); + Assertions.assertEquals(8388608L, derived.getVirtualStorageMaxFetchRunBytes()); + // ...while the per-task location and mode flags are overridden. + Assertions.assertTrue(derived.isVirtualStorage()); + Assertions.assertTrue(derived.isVirtualStorageEphemeral()); + Assertions.assertTrue(derived.isVirtualStoragePartialDownloadsEnabled()); + Assertions.assertEquals(1, derived.getLocations().size()); + Assertions.assertEquals(new File(tempDir, "task"), derived.getLocations().get(0).getPath()); + // ...and settings incompatible with virtual storage are cleared (no shared info dir, no page-cache warming). + Assertions.assertNull(derived.getInfoDir()); + Assertions.assertEquals(0, derived.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload()); + + // The shared injected config is not mutated by the derive. + Assertions.assertFalse(injected.isVirtualStorage()); + Assertions.assertTrue(injected.getLocations().isEmpty()); + } + finally { + shared.stop(); + } + } + + @Test + void testNonVirtualManufacturateUsesFreshDefaultsNotInjectedConfig() throws Exception + { + // A non-virtual (eager) per-task cache must NOT inherit the node's transient settings: a positive + // numThreadsToLoadSegmentsIntoPageCacheOnDownload would leak a per-reader executor (input entities drop segments + // but never shut the manager down), and deleteOnRemove=false would leave downloaded segments behind. It starts + // from fresh defaults instead. + final ObjectMapper mapper = TestHelper.makeJsonMapper(); + mapper.setInjectableValues(new InjectableValues.Std().addValue(RuntimeInfo.class, new RuntimeInfo())); + final SegmentLoaderConfig injected = mapper.readValue( + "{\"numThreadsToLoadSegmentsIntoPageCacheOnDownload\": 4," + + " \"deleteOnRemove\": false," + + " \"lazyLoadOnStart\": true," + + " \"infoDir\": \"/var/druid/segment-cache/info_dir\"}", + SegmentLoaderConfig.class + ); + + // The ephemeral loading pool must not even be resolved for a non-virtual cache. + final Provider throwingProvider = () -> { + throw new AssertionError("ephemeral loading pool must not be resolved for virtualStorage=false"); + }; + final SegmentCacheManagerFactory factory = + new SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper, injected, throwingProvider); + + final SegmentLocalCacheManager m = + (SegmentLocalCacheManager) factory.manufacturate(new File(tempDir, "eager"), null, false, false); + final SegmentLoaderConfig derived = m.getConfig(); + + // Transient settings are the safe defaults, not the injected node's values. + Assertions.assertFalse(derived.isVirtualStorage()); + Assertions.assertEquals(0, derived.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload()); + Assertions.assertTrue(derived.isDeleteOnRemove()); + Assertions.assertNull(derived.getInfoDir()); + Assertions.assertFalse(derived.isLazyLoadOnStart()); + // The per-task location is still applied. + Assertions.assertEquals(1, derived.getLocations().size()); + Assertions.assertEquals(new File(tempDir, "eager"), derived.getLocations().get(0).getPath()); + } + @Test void testCreateWithOwnedPoolBuildsOneAvailablePoolPerFactory() { final SegmentCacheManagerFactory factory = SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, jsonMapper); - final SegmentCacheManager m1 = factory.manufacturate(new File(tempDir, "d"), null, true); - final SegmentCacheManager m2 = factory.manufacturate(new File(tempDir, "e"), null, true); + final SegmentCacheManager m1 = factory.manufacturate(new File(tempDir, "d"), null, true, false); + final SegmentCacheManager m2 = factory.manufacturate(new File(tempDir, "e"), null, true, false); try { Assertions.assertTrue(m1.getLoadingThreadPool().isAvailable()); // createWithOwnedPool builds one pool and shares it across the factory's manufacturate calls. diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java index d83ea2885868..5933f57bdafb 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java @@ -20,6 +20,7 @@ package org.apache.druid.indexing.common; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.ImmutableMap; import org.apache.druid.client.cache.Cache; import org.apache.druid.client.cache.CacheConfig; import org.apache.druid.client.cache.CachePopulatorStats; @@ -109,6 +110,9 @@ public void setUp() throws IOException EasyMock.expect(task.getId()).andReturn("task_id").anyTimes(); EasyMock.expect(task.getDataSource()).andReturn("task_ds").anyTimes(); EasyMock.expect(task.getContextValue(Tasks.STORE_EMPTY_COLUMNS_KEY, true)).andReturn(true).anyTimes(); + EasyMock.expect(task.getContext()) + .andReturn(ImmutableMap.of(Tasks.VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY, "false")) + .anyTimes(); IndexMergerV9 indexMergerV9 = EasyMock.createMock(IndexMergerV9.class); EasyMock.expect(mockIndexMergerV9.create(true)).andReturn(indexMergerV9).anyTimes(); EasyMock.replay(task, mockHandoffNotifierFactory, mockIndexMergerV9); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigBuilder.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigBuilder.java index 69f05e763a0c..ecfbdbbd393d 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigBuilder.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigBuilder.java @@ -37,6 +37,7 @@ public class TaskConfigBuilder private boolean enableTaskLevelLogPush; private Long tmpStorageBytesPerTask; private boolean buildV10; + private Boolean virtualStoragePartialDownloadsEnabled; public TaskConfigBuilder setBaseDir(String baseDir) { @@ -104,6 +105,12 @@ public TaskConfigBuilder setBuildV10(boolean buildV10) return this; } + public TaskConfigBuilder setVirtualStoragePartialDownloadsEnabled(Boolean virtualStoragePartialDownloadsEnabled) + { + this.virtualStoragePartialDownloadsEnabled = virtualStoragePartialDownloadsEnabled; + return this; + } + public TaskConfig build() { return new TaskConfig( @@ -117,7 +124,8 @@ public TaskConfig build() storeEmptyColumns, enableTaskLevelLogPush, tmpStorageBytesPerTask, - buildV10 + buildV10, + virtualStoragePartialDownloadsEnabled ); } } diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigTest.java new file mode 100644 index 000000000000..b1a4d0e3b797 --- /dev/null +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigTest.java @@ -0,0 +1,45 @@ +/* + * 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.indexing.common.config; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.druid.jackson.DefaultObjectMapper; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TaskConfigTest +{ + private final ObjectMapper jsonMapper = new DefaultObjectMapper(); + + @Test + void testVirtualStoragePartialDownloadsEnabledDefaultsToTrue() throws Exception + { + final TaskConfig config = jsonMapper.readValue("{}", TaskConfig.class); + Assertions.assertTrue(config.isVirtualStoragePartialDownloadsEnabled()); + } + + @Test + void testVirtualStoragePartialDownloadsEnabledFromJson() throws Exception + { + final TaskConfig config = + jsonMapper.readValue("{\"virtualStoragePartialDownloadsEnabled\": false}", TaskConfig.class); + Assertions.assertFalse(config.isVirtualStoragePartialDownloadsEnabled()); + } +} diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java index 267cc9068815..65948d2f4057 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java @@ -1268,7 +1268,12 @@ public void testRunWithSpatialDimensions() throws Exception Assert.assertEquals(new NumberedShardSpec(0, 1), segments.get(0).getShardSpec()); final File cacheDir = temporaryFolder.newFolder(); - final SegmentCacheManager segmentCacheManager = segmentCacheManagerFactory.manufacturate(cacheDir, null, false); + final SegmentCacheManager segmentCacheManager = segmentCacheManagerFactory.manufacturate( + cacheDir, + null, + false, + false + ); List rowsFromSegment = new ArrayList<>(); for (DataSegment segment : segments) { @@ -1380,7 +1385,12 @@ public void testRunWithAutoCastDimensions() throws Exception Assert.assertEquals(new NumberedShardSpec(0, 1), segments.get(0).getShardSpec()); final File cacheDir = temporaryFolder.newFolder(); - final SegmentCacheManager segmentCacheManager = segmentCacheManagerFactory.manufacturate(cacheDir, null, false); + final SegmentCacheManager segmentCacheManager = segmentCacheManagerFactory.manufacturate( + cacheDir, + null, + false, + false + ); List rowsFromSegment = new ArrayList<>(); for (DataSegment segment : segments) { @@ -1497,7 +1507,12 @@ public void testRunWithAutoCastDimensionsSortByDimension() throws Exception Assert.assertEquals(new NumberedShardSpec(0, 1), compactSegment.getShardSpec()); final File cacheDir = temporaryFolder.newFolder(); - final SegmentCacheManager segmentCacheManager = segmentCacheManagerFactory.manufacturate(cacheDir, null, false); + final SegmentCacheManager segmentCacheManager = segmentCacheManagerFactory.manufacturate( + cacheDir, + null, + false, + false + ); List rowsFromSegment = new ArrayList<>(); segmentCacheManager.load(compactSegment); @@ -1695,26 +1710,11 @@ protected Builder compactionTaskBuilder(Granularity segmentGranularity1) private TaskToolbox createTaskToolbox(ObjectMapper objectMapper, TaskActionClient taskActionClient) throws IOException { - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return ImmutableList.of(new StorageLocationConfig(localDeepStorage, null, null)); - } - - @Override - public boolean isVirtualStorage() - { - return true; - } - - @Override - public boolean isVirtualStorageEphemeral() - { - return true; - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(new StorageLocationConfig(localDeepStorage, null, null)) + .virtualStorage(true) + .virtualStorageIsEphemeral(true) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); final SegmentCacheManager cacheManager = new SegmentLocalCacheManager( storageLocations, @@ -1754,7 +1754,12 @@ public boolean isVirtualStorageEphemeral() protected List getCSVFormatRowsFromSegments(List segments) throws Exception { final File cacheDir = temporaryFolder.newFolder(); - final SegmentCacheManager segmentCacheManager = segmentCacheManagerFactory.manufacturate(cacheDir, null, false); + final SegmentCacheManager segmentCacheManager = segmentCacheManagerFactory.manufacturate( + cacheDir, + null, + false, + false + ); List rowsFromSegment = new ArrayList<>(); for (DataSegment segment : segments) { diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java index 399cb9a513e0..1aedb848dc59 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java @@ -206,16 +206,9 @@ public void setup() throws IOException { final File cacheDir = temporaryFolder.newFolder(); tmpDir = temporaryFolder.newFolder(); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(cacheDir, null, null) - ); - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(new StorageLocationConfig(cacheDir, null, null)) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); segmentCacheManager = new SegmentLocalCacheManager( storageLocations, diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java index 7673350e8029..a7026e0a627b 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java @@ -227,7 +227,7 @@ public void shutdownTask(Task task) public SegmentCacheManager newSegmentLoader(File storageDir) { - return segmentCacheManagerFactory.manufacturate(storageDir, null, true); + return segmentCacheManagerFactory.manufacturate(storageDir, null, true, false); } public ObjectMapper getObjectMapper() diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java index ef0a730e9325..8154da3536cc 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java @@ -258,7 +258,7 @@ List querySegment(DataSegment dataSegment, List columns private Segment loadSegment(DataSegment dataSegment, File tempSegmentDir) { final SegmentCacheManager cacheManager = SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, getObjectMapper()) - .manufacturate(tempSegmentDir, null, false); + .manufacturate(tempSegmentDir, null, false, false); try { cacheManager.load(dataSegment); return cacheManager.acquireCachedSegment(dataSegment.getId(), AcquireMode.FULL).orElseThrow(); diff --git a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java index 3e1c151e5f21..241241bcae0a 100644 --- a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java +++ b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java @@ -174,6 +174,11 @@ public static IndexerWorkerContext createProductionInstance( { final IndexIO indexIO = injector.getInstance(IndexIO.class); final TaskConfig taskConfig = injector.getInstance(TaskConfig.class); + // Opt into on-demand partial downloads via the MSQ query context (falls back to the node's TaskConfig default). + final boolean partialDownloadsEnabled = MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled( + QueryContext.of(task.getContext()), + taskConfig.isVirtualStoragePartialDownloadsEnabled() + ); final SegmentCacheManager cacheManager = injector.getInstance(SegmentCacheManagerFactory.class) .manufacturate( @@ -181,7 +186,8 @@ public static IndexerWorkerContext createProductionInstance( // Divide tmpStorageBytesPerTask by 3 so the local cache never takes up the majority of space. // In a typical leaf stage run, we may need some disk space for inputs and some for outputs. taskConfig.getTmpStorageBytesPerTask() > 0 ? taskConfig.getTmpStorageBytesPerTask() / 3 : null, - true + true, + partialDownloadsEnabled ); final SegmentManager segmentManager = new SegmentManager(cacheManager); final VirtualStorageManager virtualStorageManager = diff --git a/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java b/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java index 114e70a71328..cf840e7bc039 100644 --- a/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java +++ b/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java @@ -235,6 +235,13 @@ public class MultiStageQueryContext public static final String CTX_INCLUDE_ALL_COUNTERS = "includeAllCounters"; public static final boolean DEFAULT_INCLUDE_ALL_COUNTERS = true; + /** + * Whether worker tasks read input segments via on-demand partial (per-column) downloads instead of downloading each + * segment in full. Aliases the task-context key {@link Tasks#VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY} so it can be set + * on the MSQ query context; when unset it falls back to the node's {@code TaskConfig} default. + */ + public static final String CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS = Tasks.VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY; + /** * Whether workers should send live counter updates to the controller via the message relay. When enabled, workers * periodically send counter snapshots to the controller, allowing the controller to have more up-to-date progress @@ -322,6 +329,14 @@ public static int getMaxConcurrentStagesWithDefault( ); } + public static boolean getVirtualStoragePartialDownloadsEnabled( + final QueryContext queryContext, + final boolean defaultValue + ) + { + return queryContext.getBoolean(CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS, defaultValue); + } + public static boolean isDurableStorageEnabled(final QueryContext queryContext) { return queryContext.getBoolean( diff --git a/multi-stage-query/src/test/java/org/apache/druid/msq/input/RegularLoadableSegmentTest.java b/multi-stage-query/src/test/java/org/apache/druid/msq/input/RegularLoadableSegmentTest.java index 003d8a6275cc..9fa2630f4243 100644 --- a/multi-stage-query/src/test/java/org/apache/druid/msq/input/RegularLoadableSegmentTest.java +++ b/multi-stage-query/src/test/java/org/apache/druid/msq/input/RegularLoadableSegmentTest.java @@ -24,7 +24,6 @@ import com.fasterxml.jackson.annotation.JsonTypeName; import com.fasterxml.jackson.databind.InjectableValues; import com.fasterxml.jackson.databind.ObjectMapper; -import com.google.common.collect.ImmutableList; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; @@ -164,11 +163,12 @@ public void setUp() throws Exception // SegmentManager with virtualStorage for dynamically-loaded data tests cacheDir = tempDir.resolve("cache").toFile(); - final SegmentLoaderConfig virtualLoaderConfig = new SegmentLoaderConfig() - .setLocations(ImmutableList.of(new StorageLocationConfig(cacheDir, 10_000_000_000L, null))) - .setVirtualStorage(true) - .setVirtualStorageIsEphemeral(true) - .setVirtualStoragePartialDownloadsEnabled(true); + final SegmentLoaderConfig virtualLoaderConfig = SegmentLoaderConfig.builder() + .locations(new StorageLocationConfig(cacheDir, 10_000_000_000L, null)) + .virtualStorage(true) + .virtualStorageIsEphemeral(true) + .virtualStoragePartialDownloadsEnabled(true) + .build(); final List virtualLocations = virtualLoaderConfig.toStorageLocations(); segmentManagerDynamic = new SegmentManager( new SegmentLocalCacheManager( @@ -183,8 +183,9 @@ public void setUp() throws Exception // SegmentManager without virtualStorage for pre-loaded data tests preLoadCacheDir = tempDir.resolve("localCache").toFile(); - final SegmentLoaderConfig localLoaderConfig = new SegmentLoaderConfig() - .setLocations(ImmutableList.of(new StorageLocationConfig(preLoadCacheDir, 10_000_000_000L, null))); + final SegmentLoaderConfig localLoaderConfig = SegmentLoaderConfig.builder() + .locations(new StorageLocationConfig(preLoadCacheDir, 10_000_000_000L, null)) + .build(); final List localLocations = localLoaderConfig.toStorageLocations(); segmentManagerPreLoad = new SegmentManager( new SegmentLocalCacheManager( diff --git a/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteMSQTestsHelper.java b/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteMSQTestsHelper.java index 15d74d025c4a..5827a71c50f3 100644 --- a/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteMSQTestsHelper.java +++ b/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteMSQTestsHelper.java @@ -95,7 +95,7 @@ public void configure(Binder binder) public SegmentCacheManager provideSegmentCacheManager(ObjectMapper testMapper, TempDirProducer tempDirProducer) { return SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, testMapper) - .manufacturate(tempDirProducer.newTempFolder("test"), null, true); + .manufacturate(tempDirProducer.newTempFolder("test"), null, true, false); } @Provides diff --git a/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java b/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java index 02cb12d7444a..819b05caa587 100644 --- a/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java +++ b/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java @@ -473,7 +473,8 @@ public void setUp2() indexIO = new IndexIO(objectMapper, ColumnConfig.DEFAULT); segmentCacheManager = - SegmentCacheManagerFactory.createWithOwnedPool(indexIO, objectMapper).manufacturate(newTempFolder("cacheManager"), null, true); + SegmentCacheManagerFactory.createWithOwnedPool(indexIO, objectMapper) + .manufacturate(newTempFolder("cacheManager"), null, true, false); testSegmentManager = new TestSegmentManager(); diff --git a/multi-stage-query/src/test/java/org/apache/druid/msq/util/MultiStageQueryContextTest.java b/multi-stage-query/src/test/java/org/apache/druid/msq/util/MultiStageQueryContextTest.java index 6db3597d6328..44e05c0a81ae 100644 --- a/multi-stage-query/src/test/java/org/apache/druid/msq/util/MultiStageQueryContextTest.java +++ b/multi-stage-query/src/test/java/org/apache/druid/msq/util/MultiStageQueryContextTest.java @@ -470,6 +470,30 @@ public void getMaxThreads_set_returnsCorrectValue() Assert.assertEquals(Integer.valueOf(4), MultiStageQueryContext.getMaxThreads(QueryContext.of(propertyMap))); } + @Test + public void getVirtualStoragePartialDownloadsEnabled_unset_returnsDefault() + { + Assert.assertFalse(MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled(QueryContext.empty(), false)); + Assert.assertTrue(MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled(QueryContext.empty(), true)); + } + + @Test + public void getVirtualStoragePartialDownloadsEnabled_set_overridesDefault() + { + Assert.assertTrue( + MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled( + QueryContext.of(ImmutableMap.of(MultiStageQueryContext.CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS, true)), + false + ) + ); + Assert.assertFalse( + MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled( + QueryContext.of(ImmutableMap.of(MultiStageQueryContext.CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS, false)), + true + ) + ); + } + @Test public void withCommonContext_noTimeout_setsStartTimeOnly() { diff --git a/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java b/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java index 24d4e360c1ee..8feeb2cc25b8 100644 --- a/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java +++ b/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java @@ -146,7 +146,7 @@ public StorageLoadingThreadPool getEphemeralStorageLoadingThreadPool(SegmentLoad { // Force virtual-storage mode: this pool serves per-task caches even when the node itself is not in virtual-storage // mode. Threads are created lazily, so an unused pool on a non-task process is cheap. - return StorageLoadingThreadPool.createFromConfig(config.withVirtualStorage(true)); + return StorageLoadingThreadPool.createFromConfig(config.toEphemeralVirtualStorage()); } @Provides diff --git a/server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java b/server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java index b9c9f0804fed..75d1264ea055 100644 --- a/server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java +++ b/server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java @@ -20,63 +20,74 @@ 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.google.common.collect.Lists; +import org.apache.druid.common.config.Configs; import org.apache.druid.segment.file.PartialSegmentFileMapperV10; import org.apache.druid.utils.RuntimeInfo; +import javax.annotation.Nullable; import java.io.File; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; /** - * + * Configuration for a local segment cache, bound from {@code druid.segmentCache}. */ public class SegmentLoaderConfig { - @JacksonInject - private final RuntimeInfo runtimeInfo = new RuntimeInfo(); - - @JsonProperty - private List locations = Collections.emptyList(); - + private static final boolean DEFAULT_LAZY_LOAD_ON_START = false; + private static final boolean DEFAULT_DELETE_ON_REMOVE = true; + private static final int DEFAULT_DROP_SEGMENT_DELAY_MILLIS = (int) TimeUnit.SECONDS.toMillis(30); + private static final int DEFAULT_ANNOUNCE_INTERVAL_MILLIS = 0; // do not background announce + private static final int DEFAULT_NUM_THREADS_TO_LOAD_INTO_PAGE_CACHE_ON_DOWNLOAD = 0; + private static final int DEFAULT_STATUS_QUEUE_MAX_SIZE = 100; + private static final boolean DEFAULT_VIRTUAL_STORAGE = false; + private static final boolean DEFAULT_VIRTUAL_STORAGE_USE_VIRTUAL_THREADS = true; + private static final boolean DEFAULT_VIRTUAL_STORAGE_IS_EPHEMERAL = false; + private static final long DEFAULT_VIRTUAL_STORAGE_METADATA_RESERVATION_ESTIMATE = 16L * 1024L * 1024L; + private static final boolean DEFAULT_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_ENABLED = false; + + private final RuntimeInfo runtimeInfo; + + @JsonProperty("locations") + private final List locations; @JsonProperty("lazyLoadOnStart") - private boolean lazyLoadOnStart = false; - + private final boolean lazyLoadOnStart; @JsonProperty("deleteOnRemove") - private boolean deleteOnRemove = true; - + private final boolean deleteOnRemove; @JsonProperty("dropSegmentDelayMillis") - private int dropSegmentDelayMillis = (int) TimeUnit.SECONDS.toMillis(30); - + private final int dropSegmentDelayMillis; @JsonProperty("announceIntervalMillis") - private int announceIntervalMillis = 0; // do not background announce - + private final int announceIntervalMillis; @JsonProperty("numLoadingThreads") - private int numLoadingThreads = Math.max(1, runtimeInfo.getAvailableProcessors() / 6); - + private final int numLoadingThreads; + /** + * Nullable so {@link #getNumBootstrapThreads()} can fall back to {@link #numLoadingThreads} when unset. + */ @JsonProperty("numBootstrapThreads") - private Integer numBootstrapThreads = null; - + @Nullable + private final Integer numBootstrapThreads; @JsonProperty("numThreadsToLoadSegmentsIntoPageCacheOnDownload") - private int numThreadsToLoadSegmentsIntoPageCacheOnDownload = 0; - + private final int numThreadsToLoadSegmentsIntoPageCacheOnDownload; + /** + * Nullable so {@link #getNumThreadsToLoadSegmentsIntoPageCacheOnBootstrap()} can fall back to the on-download value. + */ @JsonProperty("numThreadsToLoadSegmentsIntoPageCacheOnBootstrap") - private Integer numThreadsToLoadSegmentsIntoPageCacheOnBootstrap = null; - - @JsonProperty - private File infoDir = null; - - @JsonProperty - private int statusQueueMaxSize = 100; - + @Nullable + private final Integer numThreadsToLoadSegmentsIntoPageCacheOnBootstrap; + @JsonProperty("infoDir") + @Nullable + private final File infoDir; + @JsonProperty("statusQueueMaxSize") + private final int statusQueueMaxSize; @JsonProperty("virtualStorage") - private boolean virtualStorage = false; - + private final boolean virtualStorage; @JsonProperty("virtualStorageLoadThreads") - private int virtualStorageLoadThreads = Math.max(32, 4 * runtimeInfo.getAvailableProcessors()); + private final int virtualStorageLoadThreads; /** * When true (the default), the on-demand load executor uses one virtual thread per task with a {@link @@ -85,15 +96,15 @@ public class SegmentLoaderConfig * particular deep storage SDK or workload. */ @JsonProperty("virtualStorageUseVirtualThreads") - private boolean virtualStorageUseVirtualThreads = true; + private final boolean virtualStorageUseVirtualThreads; /** * When enabled, weakly-held cache entries are evicted immediately upon release of all holds, rather than * waiting for space pressure to trigger eviction. This setting is not intended to be configured directly by - * administrators. Instead, it is expected to be set when appropriate via {@link #setVirtualStorage}. + * administrators. Instead, it is expected to be set when appropriate via {@link Builder#virtualStorageIsEphemeral}. */ @JsonProperty("virtualStorageIsEphemeral") - private boolean virtualStorageIsEphemeral = false; + private final boolean virtualStorageIsEphemeral; /** * Up-front size reservation (in bytes) used when mounting a partial-segment metadata cache entry. The entry @@ -104,7 +115,7 @@ public class SegmentLoaderConfig * need a higher value. */ @JsonProperty("virtualStorageMetadataReservationEstimate") - private long virtualStorageMetadataReservationEstimate = 16L * 1024L * 1024L; + private final long virtualStorageMetadataReservationEstimate; /** * When true, partial-eligible V10 segments are mounted via the partial machinery and @@ -113,7 +124,7 @@ public class SegmentLoaderConfig * {@link AcquireMode#FULL} so the entire segment is downloaded up front (matching pre-partial-download behavior). */ @JsonProperty("virtualStoragePartialDownloadsEnabled") - private boolean virtualStoragePartialDownloadsEnabled = false; + private final boolean virtualStoragePartialDownloadsEnabled; /** * Maximum number of unrequested bytes a partial-download range read will fetch in order to bridge two requested @@ -124,7 +135,7 @@ public class SegmentLoaderConfig * request latency vs streaming throughput. */ @JsonProperty("virtualStorageCoalesceGapBytes") - private long virtualStorageCoalesceGapBytes = PartialSegmentFileMapperV10.DEFAULT_COALESCE_GAP_BYTES; + private final long virtualStorageCoalesceGapBytes; /** * Maximum size of a single range read in a query-driven partial download. Larger fetches split at internal-file @@ -133,36 +144,82 @@ public class SegmentLoaderConfig * containers sequentially and are unaffected. {@code <= 0} disables splitting. Defaults to 64 MiB. */ @JsonProperty("virtualStorageMaxFetchRunBytes") - private long virtualStorageMaxFetchRunBytes = PartialSegmentFileMapperV10.DEFAULT_MAX_FETCH_RUN_BYTES; - - private long combinedMaxSize = 0; - - public SegmentLoaderConfig() - { - } - - private SegmentLoaderConfig(SegmentLoaderConfig other) + private final long virtualStorageMaxFetchRunBytes; + + @JsonCreator + public SegmentLoaderConfig( + @JacksonInject @Nullable RuntimeInfo runtimeInfo, + @JsonProperty("locations") @Nullable List locations, + @JsonProperty("lazyLoadOnStart") @Nullable Boolean lazyLoadOnStart, + @JsonProperty("deleteOnRemove") @Nullable Boolean deleteOnRemove, + @JsonProperty("dropSegmentDelayMillis") @Nullable Integer dropSegmentDelayMillis, + @JsonProperty("announceIntervalMillis") @Nullable Integer announceIntervalMillis, + @JsonProperty("numLoadingThreads") @Nullable Integer numLoadingThreads, + @JsonProperty("numBootstrapThreads") @Nullable Integer numBootstrapThreads, + @JsonProperty("numThreadsToLoadSegmentsIntoPageCacheOnDownload") @Nullable Integer numThreadsToLoadSegmentsIntoPageCacheOnDownload, + @JsonProperty("numThreadsToLoadSegmentsIntoPageCacheOnBootstrap") @Nullable Integer numThreadsToLoadSegmentsIntoPageCacheOnBootstrap, + @JsonProperty("infoDir") @Nullable File infoDir, + @JsonProperty("statusQueueMaxSize") @Nullable Integer statusQueueMaxSize, + @JsonProperty("virtualStorage") @Nullable Boolean virtualStorage, + @JsonProperty("virtualStorageLoadThreads") @Nullable Integer virtualStorageLoadThreads, + @JsonProperty("virtualStorageUseVirtualThreads") @Nullable Boolean virtualStorageUseVirtualThreads, + @JsonProperty("virtualStorageIsEphemeral") @Nullable Boolean virtualStorageIsEphemeral, + @JsonProperty("virtualStorageMetadataReservationEstimate") @Nullable Long virtualStorageMetadataReservationEstimate, + @JsonProperty("virtualStoragePartialDownloadsEnabled") @Nullable Boolean virtualStoragePartialDownloadsEnabled, + @JsonProperty("virtualStorageCoalesceGapBytes") @Nullable Long virtualStorageCoalesceGapBytes, + @JsonProperty("virtualStorageMaxFetchRunBytes") @Nullable Long virtualStorageMaxFetchRunBytes + ) { - this.locations = other.locations; - this.lazyLoadOnStart = other.lazyLoadOnStart; - this.deleteOnRemove = other.deleteOnRemove; - this.dropSegmentDelayMillis = other.dropSegmentDelayMillis; - this.announceIntervalMillis = other.announceIntervalMillis; - this.numLoadingThreads = other.numLoadingThreads; - this.numBootstrapThreads = other.numBootstrapThreads; - this.numThreadsToLoadSegmentsIntoPageCacheOnDownload = other.numThreadsToLoadSegmentsIntoPageCacheOnDownload; - this.numThreadsToLoadSegmentsIntoPageCacheOnBootstrap = other.numThreadsToLoadSegmentsIntoPageCacheOnBootstrap; - this.infoDir = other.infoDir; - this.statusQueueMaxSize = other.statusQueueMaxSize; - this.virtualStorage = other.virtualStorage; - this.virtualStorageLoadThreads = other.virtualStorageLoadThreads; - this.virtualStorageUseVirtualThreads = other.virtualStorageUseVirtualThreads; - this.virtualStorageIsEphemeral = other.virtualStorageIsEphemeral; - this.virtualStorageMetadataReservationEstimate = other.virtualStorageMetadataReservationEstimate; - this.virtualStoragePartialDownloadsEnabled = other.virtualStoragePartialDownloadsEnabled; - this.virtualStorageCoalesceGapBytes = other.virtualStorageCoalesceGapBytes; - this.virtualStorageMaxFetchRunBytes = other.virtualStorageMaxFetchRunBytes; - this.combinedMaxSize = other.combinedMaxSize; + // RuntimeInfo is an injected @LazySingleton (test-overridable); fall back to a fresh instance when it is not + // available, e.g. when deserializing outside a Guice context. Only used to size the thread-count defaults. + final RuntimeInfo resolvedRuntimeInfo = runtimeInfo == null ? new RuntimeInfo() : runtimeInfo; + this.runtimeInfo = resolvedRuntimeInfo; + this.locations = locations == null ? Collections.emptyList() : List.copyOf(locations); + this.lazyLoadOnStart = Configs.valueOrDefault(lazyLoadOnStart, DEFAULT_LAZY_LOAD_ON_START); + this.deleteOnRemove = Configs.valueOrDefault(deleteOnRemove, DEFAULT_DELETE_ON_REMOVE); + this.dropSegmentDelayMillis = Configs.valueOrDefault(dropSegmentDelayMillis, DEFAULT_DROP_SEGMENT_DELAY_MILLIS); + this.announceIntervalMillis = Configs.valueOrDefault(announceIntervalMillis, DEFAULT_ANNOUNCE_INTERVAL_MILLIS); + this.numLoadingThreads = Configs.valueOrDefault( + numLoadingThreads, + Math.max(1, resolvedRuntimeInfo.getAvailableProcessors() / 6) + ); + this.numBootstrapThreads = numBootstrapThreads; + this.numThreadsToLoadSegmentsIntoPageCacheOnDownload = Configs.valueOrDefault( + numThreadsToLoadSegmentsIntoPageCacheOnDownload, + DEFAULT_NUM_THREADS_TO_LOAD_INTO_PAGE_CACHE_ON_DOWNLOAD + ); + this.numThreadsToLoadSegmentsIntoPageCacheOnBootstrap = numThreadsToLoadSegmentsIntoPageCacheOnBootstrap; + this.infoDir = infoDir; + this.statusQueueMaxSize = Configs.valueOrDefault(statusQueueMaxSize, DEFAULT_STATUS_QUEUE_MAX_SIZE); + this.virtualStorage = Configs.valueOrDefault(virtualStorage, DEFAULT_VIRTUAL_STORAGE); + this.virtualStorageLoadThreads = Configs.valueOrDefault( + virtualStorageLoadThreads, + Math.max(32, 4 * resolvedRuntimeInfo.getAvailableProcessors()) + ); + this.virtualStorageUseVirtualThreads = Configs.valueOrDefault( + virtualStorageUseVirtualThreads, + DEFAULT_VIRTUAL_STORAGE_USE_VIRTUAL_THREADS + ); + this.virtualStorageIsEphemeral = Configs.valueOrDefault( + virtualStorageIsEphemeral, + DEFAULT_VIRTUAL_STORAGE_IS_EPHEMERAL + ); + this.virtualStorageMetadataReservationEstimate = Configs.valueOrDefault( + virtualStorageMetadataReservationEstimate, + DEFAULT_VIRTUAL_STORAGE_METADATA_RESERVATION_ESTIMATE + ); + this.virtualStoragePartialDownloadsEnabled = Configs.valueOrDefault( + virtualStoragePartialDownloadsEnabled, + DEFAULT_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_ENABLED + ); + this.virtualStorageCoalesceGapBytes = Configs.valueOrDefault( + virtualStorageCoalesceGapBytes, + PartialSegmentFileMapperV10.DEFAULT_COALESCE_GAP_BYTES + ); + this.virtualStorageMaxFetchRunBytes = Configs.valueOrDefault( + virtualStorageMaxFetchRunBytes, + PartialSegmentFileMapperV10.DEFAULT_MAX_FETCH_RUN_BYTES + ); } public List getLocations() @@ -212,6 +269,7 @@ public int getNumThreadsToLoadSegmentsIntoPageCacheOnBootstrap() numThreadsToLoadSegmentsIntoPageCacheOnBootstrap; } + @Nullable public File getInfoDir() { return infoDir; @@ -224,10 +282,7 @@ public int getStatusQueueMaxSize() public long getCombinedMaxSize() { - if (combinedMaxSize == 0) { - combinedMaxSize = getLocations().stream().mapToLong(StorageLocationConfig::getMaxSize).sum(); - } - return combinedMaxSize; + return getLocations().stream().mapToLong(StorageLocationConfig::getMaxSize).sum(); } public boolean isVirtualStorage() @@ -270,53 +325,29 @@ public long getVirtualStorageMaxFetchRunBytes() return virtualStorageMaxFetchRunBytes; } - public SegmentLoaderConfig setLocations(List locations) - { - this.locations = Lists.newArrayList(locations); - return this; - } - - public SegmentLoaderConfig setVirtualStoragePartialDownloadsEnabled(boolean enabled) - { - this.virtualStoragePartialDownloadsEnabled = enabled; - return this; - } - /** - * Sets {@link #virtualStorage}. + * Returns a copy of this config configured as an ephemeral, on-demand virtual-storage cache: + * {@link #isVirtualStorage()} and {@link #isVirtualStorageEphemeral()} are set, and the settings that only apply to + * a classic on-disk historical cache are dropped. Virtual-storage tuning ({@link #getVirtualStorageLoadThreads()}, + * {@link #isVirtualStorageUseVirtualThreads()}, {@link #getVirtualStorageMetadataReservationEstimate()}, the + * coalescing/fetch-run limits) is preserved, so a per-task cache derived from a node's {@code druid.segmentCache} + * config keeps operator tuning. */ - public SegmentLoaderConfig setVirtualStorage(boolean virtualStorage) + public SegmentLoaderConfig toEphemeralVirtualStorage() { - this.virtualStorage = virtualStorage; - return this; - } - - /** - * Returns a copy of this config with {@link #virtualStorage} set to {@code virtualStorage}. All other settings - * (notably {@link #getVirtualStorageLoadThreads()} and {@link #isVirtualStorageUseVirtualThreads()}) are preserved. - * Used to derive an always-virtual config for the shared ephemeral on-demand loading pool from a node config that - * may not itself run in virtual-storage mode. - */ - public SegmentLoaderConfig withVirtualStorage(boolean virtualStorage) - { - final SegmentLoaderConfig copy = new SegmentLoaderConfig(this); - copy.virtualStorage = virtualStorage; - return copy; - } - - /** - * Sets {@link #virtualStorageIsEphemeral}. - */ - public SegmentLoaderConfig setVirtualStorageIsEphemeral(boolean virtualStorageFabricEphemeral) - { - this.virtualStorageIsEphemeral = virtualStorageFabricEphemeral; - return this; + return toBuilder() + .virtualStorage(true) + .virtualStorageIsEphemeral(true) + .lazyLoadOnStart(false) + .infoDir(null) + .numThreadsToLoadSegmentsIntoPageCacheOnDownload(0) + .numThreadsToLoadSegmentsIntoPageCacheOnBootstrap(null) + .deleteOnRemove(true) + .build(); } /** * Convert a list of {@link StorageLocationConfig} objects to {@link StorageLocation} objects. - *

- * Note: {@link #getLocations} is called instead of variable access because some testcases overrides this method */ public List toStorageLocations() { @@ -338,6 +369,31 @@ public List toStorageLocations() .collect(Collectors.toList()); } + public Builder toBuilder() + { + return new Builder() + .runtimeInfo(runtimeInfo) + .locations(locations) + .lazyLoadOnStart(lazyLoadOnStart) + .deleteOnRemove(deleteOnRemove) + .dropSegmentDelayMillis(dropSegmentDelayMillis) + .announceIntervalMillis(announceIntervalMillis) + .numLoadingThreads(numLoadingThreads) + .numBootstrapThreads(numBootstrapThreads) + .numThreadsToLoadSegmentsIntoPageCacheOnDownload(numThreadsToLoadSegmentsIntoPageCacheOnDownload) + .numThreadsToLoadSegmentsIntoPageCacheOnBootstrap(numThreadsToLoadSegmentsIntoPageCacheOnBootstrap) + .infoDir(infoDir) + .statusQueueMaxSize(statusQueueMaxSize) + .virtualStorage(virtualStorage) + .virtualStorageLoadThreads(virtualStorageLoadThreads) + .virtualStorageUseVirtualThreads(virtualStorageUseVirtualThreads) + .virtualStorageIsEphemeral(virtualStorageIsEphemeral) + .virtualStorageMetadataReservationEstimate(virtualStorageMetadataReservationEstimate) + .virtualStoragePartialDownloadsEnabled(virtualStoragePartialDownloadsEnabled) + .virtualStorageCoalesceGapBytes(virtualStorageCoalesceGapBytes) + .virtualStorageMaxFetchRunBytes(virtualStorageMaxFetchRunBytes); + } + @Override public String toString() { @@ -361,7 +417,212 @@ public String toString() ", virtualStoragePartialDownloadsEnabled=" + virtualStoragePartialDownloadsEnabled + ", virtualStorageCoalesceGapBytes=" + virtualStorageCoalesceGapBytes + ", virtualStorageMaxFetchRunBytes=" + virtualStorageMaxFetchRunBytes + - ", combinedMaxSize=" + combinedMaxSize + '}'; } + + public static Builder builder() + { + return new Builder(); + } + + /** + * Builds a {@link SegmentLoaderConfig}. Any field left unset takes the same default the {@link JsonCreator} + * constructor applies. Obtain one via {@link SegmentLoaderConfig#builder()} (all defaults) or + * {@link SegmentLoaderConfig#toBuilder()} (a copy of an existing config). + */ + public static class Builder + { + @Nullable + private RuntimeInfo runtimeInfo; + @Nullable + private List locations; + @Nullable + private Boolean lazyLoadOnStart; + @Nullable + private Boolean deleteOnRemove; + @Nullable + private Integer dropSegmentDelayMillis; + @Nullable + private Integer announceIntervalMillis; + @Nullable + private Integer numLoadingThreads; + @Nullable + private Integer numBootstrapThreads; + @Nullable + private Integer numThreadsToLoadSegmentsIntoPageCacheOnDownload; + @Nullable + private Integer numThreadsToLoadSegmentsIntoPageCacheOnBootstrap; + @Nullable + private File infoDir; + @Nullable + private Integer statusQueueMaxSize; + @Nullable + private Boolean virtualStorage; + @Nullable + private Integer virtualStorageLoadThreads; + @Nullable + private Boolean virtualStorageUseVirtualThreads; + @Nullable + private Boolean virtualStorageIsEphemeral; + @Nullable + private Long virtualStorageMetadataReservationEstimate; + @Nullable + private Boolean virtualStoragePartialDownloadsEnabled; + @Nullable + private Long virtualStorageCoalesceGapBytes; + @Nullable + private Long virtualStorageMaxFetchRunBytes; + + public Builder runtimeInfo(@Nullable RuntimeInfo runtimeInfo) + { + this.runtimeInfo = runtimeInfo; + return this; + } + + public Builder locations(@Nullable List locations) + { + this.locations = locations; + return this; + } + + public Builder locations(StorageLocationConfig... locations) + { + this.locations = Arrays.asList(locations); + return this; + } + + public Builder lazyLoadOnStart(boolean lazyLoadOnStart) + { + this.lazyLoadOnStart = lazyLoadOnStart; + return this; + } + + public Builder deleteOnRemove(boolean deleteOnRemove) + { + this.deleteOnRemove = deleteOnRemove; + return this; + } + + public Builder dropSegmentDelayMillis(int dropSegmentDelayMillis) + { + this.dropSegmentDelayMillis = dropSegmentDelayMillis; + return this; + } + + public Builder announceIntervalMillis(int announceIntervalMillis) + { + this.announceIntervalMillis = announceIntervalMillis; + return this; + } + + public Builder numLoadingThreads(int numLoadingThreads) + { + this.numLoadingThreads = numLoadingThreads; + return this; + } + + public Builder numBootstrapThreads(@Nullable Integer numBootstrapThreads) + { + this.numBootstrapThreads = numBootstrapThreads; + return this; + } + + public Builder numThreadsToLoadSegmentsIntoPageCacheOnDownload(int numThreads) + { + this.numThreadsToLoadSegmentsIntoPageCacheOnDownload = numThreads; + return this; + } + + public Builder numThreadsToLoadSegmentsIntoPageCacheOnBootstrap(@Nullable Integer numThreads) + { + this.numThreadsToLoadSegmentsIntoPageCacheOnBootstrap = numThreads; + return this; + } + + public Builder infoDir(@Nullable File infoDir) + { + this.infoDir = infoDir; + return this; + } + + public Builder statusQueueMaxSize(int statusQueueMaxSize) + { + this.statusQueueMaxSize = statusQueueMaxSize; + return this; + } + + public Builder virtualStorage(boolean virtualStorage) + { + this.virtualStorage = virtualStorage; + return this; + } + + public Builder virtualStorageLoadThreads(int virtualStorageLoadThreads) + { + this.virtualStorageLoadThreads = virtualStorageLoadThreads; + return this; + } + + public Builder virtualStorageUseVirtualThreads(boolean virtualStorageUseVirtualThreads) + { + this.virtualStorageUseVirtualThreads = virtualStorageUseVirtualThreads; + return this; + } + + public Builder virtualStorageIsEphemeral(boolean virtualStorageIsEphemeral) + { + this.virtualStorageIsEphemeral = virtualStorageIsEphemeral; + return this; + } + + public Builder virtualStorageMetadataReservationEstimate(long virtualStorageMetadataReservationEstimate) + { + this.virtualStorageMetadataReservationEstimate = virtualStorageMetadataReservationEstimate; + return this; + } + + public Builder virtualStoragePartialDownloadsEnabled(boolean virtualStoragePartialDownloadsEnabled) + { + this.virtualStoragePartialDownloadsEnabled = virtualStoragePartialDownloadsEnabled; + return this; + } + + public Builder virtualStorageCoalesceGapBytes(long virtualStorageCoalesceGapBytes) + { + this.virtualStorageCoalesceGapBytes = virtualStorageCoalesceGapBytes; + return this; + } + + public Builder virtualStorageMaxFetchRunBytes(long virtualStorageMaxFetchRunBytes) + { + this.virtualStorageMaxFetchRunBytes = virtualStorageMaxFetchRunBytes; + return this; + } + + public SegmentLoaderConfig build() + { + return new SegmentLoaderConfig( + runtimeInfo, + locations, + lazyLoadOnStart, + deleteOnRemove, + dropSegmentDelayMillis, + announceIntervalMillis, + numLoadingThreads, + numBootstrapThreads, + numThreadsToLoadSegmentsIntoPageCacheOnDownload, + numThreadsToLoadSegmentsIntoPageCacheOnBootstrap, + infoDir, + statusQueueMaxSize, + virtualStorage, + virtualStorageLoadThreads, + virtualStorageUseVirtualThreads, + virtualStorageIsEphemeral, + virtualStorageMetadataReservationEstimate, + virtualStoragePartialDownloadsEnabled, + virtualStorageCoalesceGapBytes, + virtualStorageMaxFetchRunBytes + ); + } + } } diff --git a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java index ec99cd4c1c36..34a8c846d3a2 100644 --- a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java +++ b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java @@ -1595,6 +1595,12 @@ public StorageLoadingThreadPool getLoadingThreadPool() return virtualStorageLoadingThreadPool; } + @VisibleForTesting + public SegmentLoaderConfig getConfig() + { + return config; + } + /** * Checks whether a segment is already cached. This method does not confirm if the segment is actually mounted in * the location, or even that the segment files in some location are valid, just that some files exist in the diff --git a/server/src/test/java/org/apache/druid/client/DruidServerConfigTest.java b/server/src/test/java/org/apache/druid/client/DruidServerConfigTest.java index 185685daab8d..258fec51b9c5 100644 --- a/server/src/test/java/org/apache/druid/client/DruidServerConfigTest.java +++ b/server/src/test/java/org/apache/druid/client/DruidServerConfigTest.java @@ -87,7 +87,7 @@ public void testCombinedSize() locations.add(locationConfig2); DruidServerConfig druidServerConfig = new DruidServerConfig( new RuntimeInfo(), - new SegmentLoaderConfig().setLocations(locations) + SegmentLoaderConfig.builder().locations(locations).build() ); Assert.assertEquals(30000000000L, druidServerConfig.getMaxSize()); } @@ -103,14 +103,7 @@ public long getMaxHeapSizeBytes() return HumanReadableBytes.parse("3GiB"); } }; - SegmentLoaderConfig segmentLoaderConfig = new SegmentLoaderConfig() - { - @Override - public boolean isVirtualStorage() - { - return true; - } - }; + SegmentLoaderConfig segmentLoaderConfig = SegmentLoaderConfig.builder().virtualStorage(true).build(); DruidServerConfig druidServerConfig = new DruidServerConfig(runtimeInfo, segmentLoaderConfig); Assert.assertEquals(HumanReadableBytes.parse("45TiB"), druidServerConfig.getMaxSize()); } @@ -132,7 +125,7 @@ public void testServerMaxSizePrecedence() throws Exception mapper.setInjectableValues(new InjectableValues.Std().addValue(ObjectMapper.class, new DefaultObjectMapper()) .addValue( SegmentLoaderConfig.class, - new SegmentLoaderConfig().setLocations(locations) + SegmentLoaderConfig.builder().locations(locations).build() ) .addValue(RuntimeInfo.class, new RuntimeInfo())); diff --git a/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java b/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java index 06c01e1ad5a4..f45d49a7a2cd 100644 --- a/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java +++ b/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java @@ -174,9 +174,9 @@ public void testEphemeralStorageLoadingThreadPoolIsInjectedAndAvailable() { // The qualified ephemeral loading pool must resolve from the core injector (StorageNodeModule is universal via // CoreInjectorBuilder), so SegmentCacheManagerFactory's @Inject constructor can be satisfied on every process. - // The provider forces virtual storage via config.withVirtualStorage(true) regardless of the node's flag. - Mockito.when(segmentLoaderConfig.withVirtualStorage(true)) - .thenReturn(new SegmentLoaderConfig().setVirtualStorage(true)); + // The provider forces virtual storage via config.toEphemeralVirtualStorage() regardless of the node's flag. + Mockito.when(segmentLoaderConfig.toEphemeralVirtualStorage()) + .thenReturn(SegmentLoaderConfig.builder().virtualStorage(true).build()); final StorageLoadingThreadPool pool = injector().getInstance( Key.get(StorageLoadingThreadPool.class, EphemeralStorageLoading.class) diff --git a/server/src/test/java/org/apache/druid/segment/loading/PermitLimitedSegmentRangeReaderTest.java b/server/src/test/java/org/apache/druid/segment/loading/PermitLimitedSegmentRangeReaderTest.java index 193d8474b1fb..c725ec8db4d0 100644 --- a/server/src/test/java/org/apache/druid/segment/loading/PermitLimitedSegmentRangeReaderTest.java +++ b/server/src/test/java/org/apache/druid/segment/loading/PermitLimitedSegmentRangeReaderTest.java @@ -31,14 +31,7 @@ class PermitLimitedSegmentRangeReaderTest private static StorageLoadingThreadPool oneThreadPool() { return StorageLoadingThreadPool.createFromConfig( - new SegmentLoaderConfig() - { - @Override - public int getVirtualStorageLoadThreads() - { - return 1; - } - }.setVirtualStorage(true) + SegmentLoaderConfig.builder().virtualStorageLoadThreads(1).virtualStorage(true).build() ); } diff --git a/server/src/test/java/org/apache/druid/segment/loading/SegmentLoaderConfigTest.java b/server/src/test/java/org/apache/druid/segment/loading/SegmentLoaderConfigTest.java index a3db293b6ad7..eb520c2169ae 100644 --- a/server/src/test/java/org/apache/druid/segment/loading/SegmentLoaderConfigTest.java +++ b/server/src/test/java/org/apache/druid/segment/loading/SegmentLoaderConfigTest.java @@ -29,18 +29,16 @@ class SegmentLoaderConfigTest { @Test - void testSetVirtualStorage() + void testBuilderVirtualStorageFlags() { - final SegmentLoaderConfig config = new SegmentLoaderConfig(); - - // Verify default values - Assertions.assertFalse(config.isVirtualStorage()); - Assertions.assertFalse(config.isVirtualStorageEphemeral()); - - // Set both to true - config.setVirtualStorage(true).setVirtualStorageIsEphemeral(true); - - // Verify both fields are set + // Defaults are false. + final SegmentLoaderConfig defaults = SegmentLoaderConfig.builder().build(); + Assertions.assertFalse(defaults.isVirtualStorage()); + Assertions.assertFalse(defaults.isVirtualStorageEphemeral()); + + // The builder sets both. + final SegmentLoaderConfig config = + SegmentLoaderConfig.builder().virtualStorage(true).virtualStorageIsEphemeral(true).build(); Assertions.assertTrue(config.isVirtualStorage()); Assertions.assertTrue(config.isVirtualStorageEphemeral()); } @@ -82,15 +80,16 @@ void testVirtualStorageMaxFetchRunBytes() throws Exception } @Test - public void testWithVirtualStorageReturnsCopyAndDoesNotMutateOriginal() + public void testToEphemeralVirtualStorageReturnsCopyAndDoesNotMutateOriginal() { - final SegmentLoaderConfig original = new SegmentLoaderConfig(); + final SegmentLoaderConfig original = SegmentLoaderConfig.builder().build(); Assertions.assertFalse(original.isVirtualStorage()); - final SegmentLoaderConfig copy = original.withVirtualStorage(true); + final SegmentLoaderConfig copy = original.toEphemeralVirtualStorage(); - // The copy has the flag flipped, while other settings are preserved. + // The copy is an ephemeral virtual-storage config, while tuning settings are preserved. Assertions.assertTrue(copy.isVirtualStorage()); + Assertions.assertTrue(copy.isVirtualStorageEphemeral()); Assertions.assertEquals(original.getVirtualStorageLoadThreads(), copy.getVirtualStorageLoadThreads()); Assertions.assertEquals(original.isVirtualStorageUseVirtualThreads(), copy.isVirtualStorageUseVirtualThreads()); @@ -98,4 +97,43 @@ public void testWithVirtualStorageReturnsCopyAndDoesNotMutateOriginal() Assertions.assertNotSame(original, copy); Assertions.assertFalse(original.isVirtualStorage()); } + + @Test + void testEnablingVirtualStorageClearsIncompatibleSettingsButKeepsTuning() throws Exception + { + final ObjectMapper jsonMapper = new DefaultObjectMapper(); + jsonMapper.setInjectableValues(new InjectableValues.Std().addValue(RuntimeInfo.class, new RuntimeInfo())); + final SegmentLoaderConfig node = jsonMapper.readValue( + "{\"lazyLoadOnStart\": true," + + " \"deleteOnRemove\": false," + + " \"infoDir\": \"/var/druid/segment-cache/info_dir\"," + + " \"numThreadsToLoadSegmentsIntoPageCacheOnDownload\": 4," + + " \"numThreadsToLoadSegmentsIntoPageCacheOnBootstrap\": 2," + + " \"virtualStorageMetadataReservationEstimate\": 99999999," + + " \"virtualStorageCoalesceGapBytes\": 65536}", + SegmentLoaderConfig.class + ); + + final SegmentLoaderConfig virtual = node.toEphemeralVirtualStorage(); + + // Classic on-disk-cache settings are reset to virtual-storage-safe values... + Assertions.assertFalse(virtual.isLazyLoadOnStart()); + Assertions.assertTrue(virtual.isDeleteOnRemove()); + Assertions.assertNull(virtual.getInfoDir()); + Assertions.assertEquals(0, virtual.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload()); + Assertions.assertEquals(0, virtual.getNumThreadsToLoadSegmentsIntoPageCacheOnBootstrap()); + // ...while virtual-storage tuning is preserved. + Assertions.assertTrue(virtual.isVirtualStorage()); + Assertions.assertEquals(99999999L, virtual.getVirtualStorageMetadataReservationEstimate()); + Assertions.assertEquals(65536L, virtual.getVirtualStorageCoalesceGapBytes()); + + // The original node config is untouched, and a plain toBuilder copy preserves everything (only the ephemeral + // virtual-storage derivation sanitizes). + Assertions.assertTrue(node.isLazyLoadOnStart()); + Assertions.assertEquals(4, node.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload()); + final SegmentLoaderConfig plainCopy = node.toBuilder().build(); + Assertions.assertTrue(plainCopy.isLazyLoadOnStart()); + Assertions.assertFalse(plainCopy.isDeleteOnRemove()); + Assertions.assertEquals(4, plainCopy.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload()); + } } diff --git a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java index afbcc0bc7507..c3e957fcdf49 100644 --- a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java +++ b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java @@ -129,46 +129,16 @@ public void setUp() throws Exception locations.add(locationConfig); locations.add(locationConfig2); - loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return locations; - } - - @Override - public File getInfoDir() - { - return new File(tempDir, "info"); - } - }; - vsfLoaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return locations; - } - - @Override - public boolean isVirtualStorage() - { - return true; - } - - @Override - public int getVirtualStorageLoadThreads() - { - return Runtime.getRuntime().availableProcessors(); - } - - @Override - public File getInfoDir() - { - return new File(tempDir, "info"); - } - }; + loaderConfig = SegmentLoaderConfig.builder() + .locations(locations) + .infoDir(new File(tempDir, "info")) + .build(); + vsfLoaderConfig = SegmentLoaderConfig.builder() + .locations(locations) + .virtualStorage(true) + .virtualStorageLoadThreads(Runtime.getRuntime().availableProcessors()) + .infoDir(new File(tempDir, "info")) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); location = storageLocations.get(0); location2 = storageLocations.get(1); diff --git a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java index 7a6ed49c37ef..7a823904db63 100644 --- a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java +++ b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java @@ -253,10 +253,11 @@ void setup() throws IOException FileUtils.mkdirp(cacheRoot); final StorageLocationConfig locConfig = new StorageLocationConfig(cacheRoot, 1024L * 1024L * 1024L, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - .setLocations(List.of(locConfig)) - .setVirtualStorage(true) - .setVirtualStoragePartialDownloadsEnabled(true); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(locConfig) + .virtualStorage(true) + .virtualStoragePartialDownloadsEnabled(true) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); manager = new SegmentLocalCacheManager( storageLocations, @@ -350,7 +351,7 @@ void testPartialAcquireRecordsLoadAndCorrectsReservationOvercount() throws ExecutionException, InterruptedException, IOException { final StorageLocation loc = manager.getLocations().get(0); - final long estimate = new SegmentLoaderConfig().getVirtualStorageMetadataReservationEstimate(); + final long estimate = SegmentLoaderConfig.builder().build().getVirtualStorageMetadataReservationEstimate(); // A lazy (PARTIAL) acquire mounts only the metadata entry (downloads the header); bundles stay lazy until a cursor // is built, so exactly one weak reservation + one weak load is expected. @@ -630,10 +631,11 @@ void testCachedFullAcquirePinsBundlesAgainstEviction() final File plainCacheRoot = new File(perTestTempDir, "plain_cache_" + ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE)); FileUtils.mkdirp(plainCacheRoot); final StorageLocationConfig locConfig = new StorageLocationConfig(plainCacheRoot, 1024L * 1024L * 1024L, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - .setLocations(List.of(locConfig)) - .setVirtualStorage(true) - .setVirtualStoragePartialDownloadsEnabled(true); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(locConfig) + .virtualStorage(true) + .virtualStoragePartialDownloadsEnabled(true) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); final SegmentLocalCacheManager plain = new SegmentLocalCacheManager( storageLocations, @@ -697,10 +699,11 @@ void testPartialDownloadsDisabledFallsBackToEager() // eager counterparts: the returned segment is an eager QueryableIndexSegment (NOT PartialQueryableIndexSegment), // and no PartialSegmentMetadataCacheEntry is registered on the location. final StorageLocationConfig locConfig = new StorageLocationConfig(cacheRoot, 1024L * 1024L * 1024L, null); - final SegmentLoaderConfig disabledConfig = new SegmentLoaderConfig() - .setLocations(List.of(locConfig)) - .setVirtualStorage(true) - .setVirtualStoragePartialDownloadsEnabled(false); + final SegmentLoaderConfig disabledConfig = SegmentLoaderConfig.builder() + .locations(locConfig) + .virtualStorage(true) + .virtualStoragePartialDownloadsEnabled(false) + .build(); final List storageLocations = disabledConfig.toStorageLocations(); final SegmentLocalCacheManager disabledManager = new SegmentLocalCacheManager( storageLocations, @@ -807,10 +810,11 @@ void testGetCachedSegmentsDeletesPartialLayoutWhenPartialDisabled() throws IOExc // A manager with partial downloads disabled (operator toggled the flag off) over the same cache dir must reclaim // the now-unusable partial layout at bootstrap rather than reserving it and failing at query time. final StorageLocationConfig locConfig = new StorageLocationConfig(cacheRoot, 1024L * 1024L * 1024L, null); - final SegmentLoaderConfig disabledConfig = new SegmentLoaderConfig() - .setLocations(List.of(locConfig)) - .setVirtualStorage(true) - .setVirtualStoragePartialDownloadsEnabled(false); + final SegmentLoaderConfig disabledConfig = SegmentLoaderConfig.builder() + .locations(locConfig) + .virtualStorage(true) + .virtualStoragePartialDownloadsEnabled(false) + .build(); final List storageLocations = disabledConfig.toStorageLocations(); final SegmentLocalCacheManager disabledManager = new SegmentLocalCacheManager( storageLocations, diff --git a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialDropTest.java b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialDropTest.java index bc4901bcb330..f0f0d227846c 100644 --- a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialDropTest.java +++ b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialDropTest.java @@ -156,7 +156,7 @@ void setup() throws IOException FileUtils.mkdirp(infoDir); final StorageLocationConfig locConfig = new StorageLocationConfig(cacheDir, ESTIMATE * 16, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig().setLocations(List.of(locConfig)); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().locations(locConfig).build(); final List locations = loaderConfig.toStorageLocations(); manager = new SegmentLocalCacheManager( locations, 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 f63563474e4c..1d72d2a2ed45 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 @@ -710,10 +710,11 @@ private SegmentLocalCacheManager makeManagerAtLocations( final List locConfigs = locationRoots.stream() .map(root -> new StorageLocationConfig(root, 1024L * 1024L * 1024L, null)) .toList(); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - .setLocations(locConfigs) - .setVirtualStorage(virtualStorage) - .setVirtualStoragePartialDownloadsEnabled(partialDownloadsEnabled); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(locConfigs) + .virtualStorage(virtualStorage) + .virtualStoragePartialDownloadsEnabled(partialDownloadsEnabled) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); return new SegmentLocalCacheManager( storageLocations, diff --git a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerTest.java b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerTest.java index d6f96739fc20..458da17f2194 100644 --- a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerTest.java +++ b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerTest.java @@ -118,16 +118,10 @@ public void testCanHandleSegmentsWithConfigLocations() { // Only injecting config locations without locations shouldn't really be the case. // It possibly suggests an issue with injection. - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(localSegmentCacheDir, null, null) - ); - } - }; + final SegmentLoaderConfig loaderConfig = + SegmentLoaderConfig.builder() + .locations(new StorageLocationConfig(localSegmentCacheDir, null, null)) + .build(); manager = new SegmentLocalCacheManager( ImmutableList.of(), @@ -146,7 +140,7 @@ public void testCanHandleSegmentsWithLocations() final ImmutableList locations = ImmutableList.of( new StorageLocation(localSegmentCacheDir, 10000000000L, null) ); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig(); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().build(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( locations, loaderConfig, @@ -161,7 +155,7 @@ public void testCanHandleSegmentsWithLocations() @Test public void testCanHandleSegmentsWithEmptyLocationsAndConfigLocations() { - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig(); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().build(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( ImmutableList.of(), loaderConfig, @@ -176,7 +170,7 @@ public void testCanHandleSegmentsWithEmptyLocationsAndConfigLocations() @Test public void testGetCachedSegmentsWhenCanHandleSegmentsIsFalse() { - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig(); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().build(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( null, loaderConfig, @@ -346,24 +340,13 @@ public void testIfSegmentIsLoaded() throws IOException } @Test - public void testLoadSegmentInPageCache() throws IOException, SegmentLoadingException + public void testLoadSegmentInPageCache() throws IOException { - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public int getNumThreadsToLoadSegmentsIntoPageCacheOnDownload() - { - return 1; - } - - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(localSegmentCacheDir, null, null) - ); - } - }; + final SegmentLoaderConfig loaderConfig = + SegmentLoaderConfig.builder() + .numThreadsToLoadSegmentsIntoPageCacheOnDownload(1) + .locations(new StorageLocationConfig(localSegmentCacheDir, null, null)) + .build(); final List locations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( locations, @@ -415,7 +398,7 @@ public void testRetrySuccessAtFirstLocation() throws Exception final StorageLocationConfig locationConfig2 = new StorageLocationConfig(localStorageFolder2, 1000000000L, null); locations.add(locationConfig2); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig().setLocations(locations); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().locations(locations).build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -454,7 +437,7 @@ public void testRetrySuccessAtSecondLocation() throws Exception final StorageLocationConfig locationConfig2 = new StorageLocationConfig(localStorageFolder2, 10000000L, null); locations.add(locationConfig2); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig().setLocations(locations); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().locations(locations).build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -495,7 +478,7 @@ public void testRetryAllFail() throws Exception final StorageLocationConfig locationConfig2 = new StorageLocationConfig(localStorageFolder2, 10000000L, null); locations.add(locationConfig2); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig().setLocations(locations); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().locations(locations).build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -535,7 +518,7 @@ public void testEmptyToFullOrder() throws Exception final StorageLocationConfig locationConfig2 = new StorageLocationConfig(localStorageFolder2, 10L, null); locations.add(locationConfig2); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig().setLocations(locations); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().locations(locations).build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -592,7 +575,7 @@ public void testSegmentDistributionUsingRoundRobinStrategy() throws Exception ); } - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig().setLocations(locationConfigs); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().locations(locationConfigs).build(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( locations, loaderConfig, @@ -670,11 +653,11 @@ public void testSegmentDistributionUsingLeastBytesUsedStrategy() throws Exceptio locations.add(locationConfig2); locations.add(locationConfig3); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig().setLocations(locations); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().locations(locations).build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, - new SegmentLoaderConfig().setLocations(locations), + SegmentLoaderConfig.builder().locations(locations).build(), StorageLoadingThreadPool.createFromConfig(loaderConfig), new LeastBytesUsedStorageLocationSelectorStrategy(storageLocations), TestHelper.getTestIndexIO(jsonMapper, ColumnConfig.DEFAULT), @@ -752,7 +735,7 @@ public void testSegmentDistributionUsingRandomStrategy() throws Exception locationConfigs.add(locationConfig2); locationConfigs.add(locationConfig3); - SegmentLoaderConfig segmentLoaderConfig = new SegmentLoaderConfig().setLocations(locationConfigs); + SegmentLoaderConfig segmentLoaderConfig = SegmentLoaderConfig.builder().locations(locationConfigs).build(); final List locations = segmentLoaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( @@ -835,7 +818,7 @@ public void testGetSegmentFilesWhenDownloadStartMarkerExists() throws Exception public void testGetBootstrapSegment() throws SegmentLoadingException { final StorageLocationConfig locationConfig = new StorageLocationConfig(localSegmentCacheDir, 10000L, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig().setLocations(ImmutableList.of(locationConfig)); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().locations(locationConfig).build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -860,20 +843,10 @@ public void testGetBootstrapSegment() throws SegmentLoadingException public void testGetBootstrapSegmentLazy() throws SegmentLoadingException { final StorageLocationConfig locationConfig = new StorageLocationConfig(localSegmentCacheDir, 10000L, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public boolean isLazyLoadOnStart() - { - return true; - } - - @Override - public List getLocations() - { - return List.of(locationConfig); - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .lazyLoadOnStart(true) + .locations(locationConfig) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -897,31 +870,11 @@ public List getLocations() public void testGetSegmentVirtualStorage() throws Exception { final StorageLocationConfig locationConfig = new StorageLocationConfig(localSegmentCacheDir, 10000L, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return ImmutableList.of(locationConfig); - } - - @Override - public boolean isVirtualStorage() - { - return true; - } - - @Override - public File getInfoDir() - { - try { - return tmpFolder.newFolder(); - } - catch (IOException e) { - throw new RuntimeException(e); - } - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(locationConfig) + .virtualStorage(true) + .infoDir(tmpFolder.newFolder()) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -969,26 +922,11 @@ public File getInfoDir() public void testVirtualStorageRejectsNonPositiveLoadThreads() { final StorageLocationConfig locationConfig = new StorageLocationConfig(localSegmentCacheDir, 10000L, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return ImmutableList.of(locationConfig); - } - - @Override - public boolean isVirtualStorage() - { - return true; - } - - @Override - public int getVirtualStorageLoadThreads() - { - return 0; - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(locationConfig) + .virtualStorage(true) + .virtualStorageLoadThreads(0) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); MatcherAssert.assertThat( Assert.assertThrows( @@ -1014,31 +952,11 @@ public int getVirtualStorageLoadThreads() public void testGetBootstrapSegmentVirtualStorage() throws Exception { final StorageLocationConfig locationConfig = new StorageLocationConfig(localSegmentCacheDir, 10000L, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return ImmutableList.of(locationConfig); - } - - @Override - public boolean isVirtualStorage() - { - return true; - } - - @Override - public File getInfoDir() - { - try { - return tmpFolder.newFolder(); - } - catch (IOException e) { - throw new RuntimeException(e); - } - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(locationConfig) + .virtualStorage(true) + .infoDir(tmpFolder.newFolder()) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -1087,20 +1005,10 @@ public File getInfoDir() public void testGetBootstrapSegmentVirtualStorageSegmentAlreadyCached() throws Exception { final StorageLocationConfig locationConfig = new StorageLocationConfig(localSegmentCacheDir, 10000L, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return ImmutableList.of(locationConfig); - } - - @Override - public boolean isVirtualStorage() - { - return true; - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(locationConfig) + .virtualStorage(true) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -1165,31 +1073,11 @@ public void testGetSegmentAfterDroppedWithNoVirtualStorageEnabled() throws Excep public void testGetSegmentVirtualStorageMountAfterDrop() throws Exception { final StorageLocationConfig locationConfig = new StorageLocationConfig(localSegmentCacheDir, 10L, null); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return ImmutableList.of(locationConfig); - } - - @Override - public boolean isVirtualStorage() - { - return true; - } - - @Override - public File getInfoDir() - { - try { - return tmpFolder.newFolder(); - } - catch (IOException e) { - throw new RuntimeException(e); - } - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(locationConfig) + .virtualStorage(true) + .infoDir(tmpFolder.newFolder()) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -1249,32 +1137,12 @@ public void testGetSegmentVirtualStorageFabricEvictImmediately() throws Exceptio { final StorageLocationConfig locationConfig = new StorageLocationConfig(localSegmentCacheDir, 10000L, null); final File infoDir = tmpFolder.newFolder(); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public List getLocations() - { - return ImmutableList.of(locationConfig); - } - - @Override - public boolean isVirtualStorage() - { - return true; - } - - @Override - public boolean isVirtualStorageEphemeral() - { - return true; - } - - @Override - public File getInfoDir() - { - return infoDir; - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .locations(locationConfig) + .virtualStorage(true) + .virtualStorageIsEphemeral(true) + .infoDir(infoDir) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); final SegmentLocalCacheManager manager = new SegmentLocalCacheManager( storageLocations, @@ -1399,7 +1267,7 @@ private SegmentLocalCacheManager makeDefaultManager(ObjectMapper jsonMapper) final List locationConfigs = new ArrayList<>(); final StorageLocationConfig locationConfig = new StorageLocationConfig(localSegmentCacheDir, 10000000000L, null); locationConfigs.add(locationConfig); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig().setLocations(locationConfigs); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().locations(locationConfigs).build(); final List locations = loaderConfig.toStorageLocations(); return new SegmentLocalCacheManager( locations, diff --git a/server/src/test/java/org/apache/druid/segment/loading/StorageLoadingThreadPoolTest.java b/server/src/test/java/org/apache/druid/segment/loading/StorageLoadingThreadPoolTest.java index 196b1d96437e..3f92a401f34a 100644 --- a/server/src/test/java/org/apache/druid/segment/loading/StorageLoadingThreadPoolTest.java +++ b/server/src/test/java/org/apache/druid/segment/loading/StorageLoadingThreadPoolTest.java @@ -28,46 +28,32 @@ class StorageLoadingThreadPoolTest { private static SegmentLoaderConfig oneVirtualThreadConfig() { - return new SegmentLoaderConfig() - { - @Override - public int getVirtualStorageLoadThreads() - { - return 1; - } - }.setVirtualStorage(true); + return SegmentLoaderConfig.builder().virtualStorageLoadThreads(1).virtualStorage(true).build(); } private static SegmentLoaderConfig fixedThreadConfig() { - return new SegmentLoaderConfig() - { - @Override - public int getVirtualStorageLoadThreads() - { - return 2; - } - - @Override - public boolean isVirtualStorageUseVirtualThreads() - { - return false; - } - }.setVirtualStorage(true); + return SegmentLoaderConfig.builder() + .virtualStorageLoadThreads(2) + .virtualStorageUseVirtualThreads(false) + .virtualStorage(true) + .build(); } @Test void testCreateFromConfigIsUnavailableWhenNotVirtualStorage() { - Assertions.assertFalse(StorageLoadingThreadPool.createFromConfig(new SegmentLoaderConfig()).isAvailable()); + Assertions.assertFalse( + StorageLoadingThreadPool.createFromConfig(SegmentLoaderConfig.builder().build()).isAvailable() + ); } @Test void testCreateFromConfigIsAvailableWhenVirtualStorage() { - // The shared ephemeral pool is built this way: createFromConfig(config.withVirtualStorage(true)). + // The shared ephemeral pool is built this way: createFromConfig(config.toEphemeralVirtualStorage()). final StorageLoadingThreadPool pool = - StorageLoadingThreadPool.createFromConfig(new SegmentLoaderConfig().withVirtualStorage(true)); + StorageLoadingThreadPool.createFromConfig(SegmentLoaderConfig.builder().build().toEphemeralVirtualStorage()); try { Assertions.assertTrue(pool.isAvailable()); Assertions.assertNotNull(pool.getExecutorService()); @@ -80,14 +66,8 @@ void testCreateFromConfigIsAvailableWhenVirtualStorage() @Test void testCreateFromConfigRejectsNonPositiveThreadCountInVirtualStorage() { - final SegmentLoaderConfig config = new SegmentLoaderConfig() - { - @Override - public int getVirtualStorageLoadThreads() - { - return 0; - } - }.setVirtualStorage(true); + final SegmentLoaderConfig config = + SegmentLoaderConfig.builder().virtualStorageLoadThreads(0).virtualStorage(true).build(); Assertions.assertThrows(DruidException.class, () -> StorageLoadingThreadPool.createFromConfig(config)); } diff --git a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java index 93571eba5bad..76344d9565d8 100644 --- a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java +++ b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java @@ -233,14 +233,9 @@ ScanQuery.class, new ScanQueryRunnerFactory( taskIntervalUnlocker ); } else { - SegmentLoaderConfig segmentLoaderConfig = new SegmentLoaderConfig() - { - @Override - public int getDropSegmentDelayMillis() - { - return delayInMilli; - } - }; + SegmentLoaderConfig segmentLoaderConfig = SegmentLoaderConfig.builder() + .dropSegmentDelayMillis(delayInMilli) + .build(); appenderator = Appenderators.createRealtime( segmentLoaderConfig, schema.getDataSource(), diff --git a/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java b/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java index ea0412709a95..22c20731c110 100644 --- a/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java +++ b/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java @@ -70,7 +70,6 @@ import java.io.File; import java.io.IOException; import java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -105,50 +104,21 @@ public void setup() throws IOException { EmittingLogger.registerEmitter(new NoopServiceEmitter()); final File segmentCacheDir = temporaryFolder.newFolder(); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public File getInfoDir() - { - return segmentCacheDir; - } - - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(segmentCacheDir, null, null) - ); - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .infoDir(segmentCacheDir) + .locations(new StorageLocationConfig(segmentCacheDir, null, null)) + .build(); final File vsfRoot = temporaryFolder.newFolder(); final File virtualSegmentCacheDir = new File(vsfRoot, "segmentCache"); FileUtils.mkdirp(virtualSegmentCacheDir); final File vsfInfoDir = new File(vsfRoot, "info"); FileUtils.mkdirp(vsfInfoDir); - final SegmentLoaderConfig virtualLoaderConfig = new SegmentLoaderConfig() - { - @Override - public File getInfoDir() - { - return vsfInfoDir; - } - - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(virtualSegmentCacheDir, null, null) - ); - } - - @Override - public boolean isVirtualStorage() - { - return true; - } - }; + final SegmentLoaderConfig virtualLoaderConfig = SegmentLoaderConfig.builder() + .infoDir(vsfInfoDir) + .locations(new StorageLocationConfig(virtualSegmentCacheDir, null, null)) + .virtualStorage(true) + .build(); final ObjectMapper objectMapper = TestHelper.makeJsonMapper(); objectMapper.registerSubtypes(TestSegmentUtils.TestLoadSpec.class); diff --git a/server/src/test/java/org/apache/druid/server/StatusResourceTest.java b/server/src/test/java/org/apache/druid/server/StatusResourceTest.java index 54dab17a40a8..c677d48ef283 100644 --- a/server/src/test/java/org/apache/druid/server/StatusResourceTest.java +++ b/server/src/test/java/org/apache/druid/server/StatusResourceTest.java @@ -28,6 +28,7 @@ import org.apache.druid.guice.TestDruidModule; import org.apache.druid.initialization.DruidModule; import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.segment.loading.SegmentLoaderConfig; import org.apache.druid.utils.JvmUtils; import org.junit.Assert; import org.junit.Test; @@ -103,7 +104,10 @@ public void testGetReadyReturns503WhenNotReady() private void testHiddenPropertiesWithPropertyFileName(String fileName) throws Exception { Injector injector = new StartupInjectorBuilder() - .add(new PropertiesModule(Collections.singletonList(fileName))) + .add( + new PropertiesModule(Collections.singletonList(fileName)), + binder -> binder.bind(SegmentLoaderConfig.class).toInstance(SegmentLoaderConfig.builder().build()) + ) .build(); Map returnedProperties = injector.getInstance(StatusResource.class).getProperties(); Set lowerCasePropertyNames = returnedProperties.keySet() diff --git a/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperCacheTest.java b/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperCacheTest.java index 7878630d2af7..e5c4063eac69 100644 --- a/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperCacheTest.java +++ b/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperCacheTest.java @@ -46,7 +46,6 @@ import java.io.File; import java.io.IOException; import java.util.ArrayList; -import java.util.Collections; import java.util.List; /** @@ -75,22 +74,10 @@ public void setup() throws IOException { infoDir = temporaryFolder.newFolder(); cacheDir = temporaryFolder.newFolder(); - loaderConfig = new SegmentLoaderConfig() - { - @Override - public File getInfoDir() - { - return infoDir; - } - - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(cacheDir, MAX_SIZE, null) - ); - } - }; + loaderConfig = SegmentLoaderConfig.builder() + .infoDir(infoDir) + .locations(new StorageLocationConfig(cacheDir, MAX_SIZE, null)) + .build(); objectMapper = TestHelper.makeJsonMapper(); objectMapper.registerSubtypes(TestSegmentUtils.TestLoadSpec.class); @@ -116,7 +103,7 @@ public List getLocations() public void testLoadStartStopWithEmptyLocations() throws IOException { final List emptyLocations = ImmutableList.of(); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig(); + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder().build(); segmentManager = new SegmentManager( new SegmentLocalCacheManager( emptyLocations, diff --git a/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperTest.java b/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperTest.java index 594554022c75..75f099883149 100644 --- a/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperTest.java +++ b/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperTest.java @@ -40,7 +40,6 @@ import java.io.File; import java.io.IOException; -import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -65,34 +64,12 @@ public void setUp() throws IOException final File segmentCacheDir = temporaryFolder.newFolder(); segmentAnnouncer = new TestDataSegmentAnnouncer(); - segmentLoaderConfig = new SegmentLoaderConfig() - { - @Override - public File getInfoDir() - { - return segmentCacheDir; - } - - @Override - public int getNumLoadingThreads() - { - return 5; - } - - @Override - public int getAnnounceIntervalMillis() - { - return 50; - } - - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(segmentCacheDir, null, null) - ); - } - }; + segmentLoaderConfig = SegmentLoaderConfig.builder() + .infoDir(segmentCacheDir) + .numLoadingThreads(5) + .announceIntervalMillis(50) + .locations(new StorageLocationConfig(segmentCacheDir, null, null)) + .build(); coordinatorClient = new TestCoordinatorClient(); serviceEmitter = new StubServiceEmitter(); diff --git a/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java b/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java index 4c8fc0c40794..b46150423aa6 100644 --- a/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java +++ b/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java @@ -44,7 +44,6 @@ import java.io.File; import java.io.IOException; import java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -72,40 +71,13 @@ public void setUp() throws IOException scheduledRunnable = new ArrayList<>(); segmentAnnouncer = new TestDataSegmentAnnouncer(); - segmentLoaderConfig = new SegmentLoaderConfig() - { - @Override - public File getInfoDir() - { - return segmentCacheDir; - } - - @Override - public int getNumLoadingThreads() - { - return 5; - } - - @Override - public int getAnnounceIntervalMillis() - { - return 50; - } - - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(segmentCacheDir, null, null) - ); - } - - @Override - public int getDropSegmentDelayMillis() - { - return 0; - } - }; + segmentLoaderConfig = SegmentLoaderConfig.builder() + .infoDir(segmentCacheDir) + .numLoadingThreads(5) + .announceIntervalMillis(50) + .locations(new StorageLocationConfig(segmentCacheDir, null, null)) + .dropSegmentDelayMillis(0) + .build(); scheduledExecutorFactory = (corePoolSize, nameFormat) -> { // Override normal behavior by adding the runnable to a list so that you can make sure @@ -298,40 +270,13 @@ public void testProcessBatchLoadDropLoadSequenceForSameSegment() throws Exceptio Mockito.doNothing().when(segmentManager).dropSegment(ArgumentMatchers.any()); final File storageDir = temporaryFolder.newFolder(); - final SegmentLoaderConfig noAnnouncerSegmentLoaderConfig = new SegmentLoaderConfig() - { - @Override - public File getInfoDir() - { - return storageDir; - } - - @Override - public int getNumLoadingThreads() - { - return 5; - } - - @Override - public int getAnnounceIntervalMillis() - { - return 0; - } - - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(storageDir, null, null) - ); - } - - @Override - public int getDropSegmentDelayMillis() - { - return 0; - } - }; + final SegmentLoaderConfig noAnnouncerSegmentLoaderConfig = SegmentLoaderConfig.builder() + .infoDir(storageDir) + .numLoadingThreads(5) + .announceIntervalMillis(0) + .locations(new StorageLocationConfig(storageDir, null, null)) + .dropSegmentDelayMillis(0) + .build(); final SegmentLoadDropHandler handler = initSegmentLoadDropHandler( noAnnouncerSegmentLoaderConfig, diff --git a/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerBroadcastJoinIndexedTableTest.java b/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerBroadcastJoinIndexedTableTest.java index 5ee311c389b1..a16e24bfc0ee 100644 --- a/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerBroadcastJoinIndexedTableTest.java +++ b/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerBroadcastJoinIndexedTableTest.java @@ -129,22 +129,10 @@ public void setup() throws IOException segmentCacheDir = temporaryFolder.newFolder(); segmentDeepStorageDir = temporaryFolder.newFolder(); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public File getInfoDir() - { - return segmentCacheDir; - } - - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(segmentCacheDir, null, null) - ); - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .infoDir(segmentCacheDir) + .locations(new StorageLocationConfig(segmentCacheDir, null, null)) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); segmentCacheManager = new SegmentLocalCacheManager( diff --git a/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerThreadSafetyTest.java b/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerThreadSafetyTest.java index 75f9c347a9c3..fec53f4c20f6 100644 --- a/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerThreadSafetyTest.java +++ b/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerThreadSafetyTest.java @@ -100,22 +100,10 @@ public void setup() throws IOException segmentCacheDir = temporaryFolder.newFolder(); segmentDeepStorageDir = temporaryFolder.newFolder(); - final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig() - { - @Override - public File getInfoDir() - { - return segmentCacheDir; - } - - @Override - public List getLocations() - { - return Collections.singletonList( - new StorageLocationConfig(segmentCacheDir, null, null) - ); - } - }; + final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder() + .infoDir(segmentCacheDir) + .locations(new StorageLocationConfig(segmentCacheDir, null, null)) + .build(); final List storageLocations = loaderConfig.toStorageLocations(); segmentCacheManager = new SegmentLocalCacheManager( storageLocations, diff --git a/server/src/test/java/org/apache/druid/server/metrics/SegmentStatsMonitorTest.java b/server/src/test/java/org/apache/druid/server/metrics/SegmentStatsMonitorTest.java index 373b1c1dd2d6..5aa4f926da24 100644 --- a/server/src/test/java/org/apache/druid/server/metrics/SegmentStatsMonitorTest.java +++ b/server/src/test/java/org/apache/druid/server/metrics/SegmentStatsMonitorTest.java @@ -52,7 +52,7 @@ public class SegmentStatsMonitorTest private SegmentLoadDropHandler segmentLoadDropMgr; private ServiceEmitter serviceEmitter; private SegmentStatsMonitor monitor; - private final SegmentLoaderConfig segmentLoaderConfig = new SegmentLoaderConfig(); + private final SegmentLoaderConfig segmentLoaderConfig = SegmentLoaderConfig.builder().build(); @Before public void setUp()