somandal commented on code in PR #16033: URL: https://github.com/apache/pinot/pull/16033#discussion_r3867005617
########## pinot-controller/src/main/java/org/apache/pinot/controller/util/PageCacheWarmupControllerExecutor.java: ########## @@ -0,0 +1,271 @@ +/** + * 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.pinot.controller.util; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.google.common.collect.BiMap; +import java.io.File; +import java.io.InputStream; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import javax.annotation.Nullable; +import org.apache.hc.core5.http.ClassicHttpRequest; +import org.apache.hc.core5.http.HttpHeaders; +import org.apache.hc.core5.http.HttpStatus; +import org.apache.hc.core5.http.HttpVersion; +import org.apache.hc.core5.http.io.support.ClassicRequestBuilder; +import org.apache.pinot.common.metrics.ControllerMeter; +import org.apache.pinot.common.metrics.ControllerMetrics; +import org.apache.pinot.common.utils.SimpleHttpResponse; +import org.apache.pinot.common.utils.URIUtils; +import org.apache.pinot.common.utils.http.HttpClient; +import org.apache.pinot.controller.ControllerConf; +import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; +import org.apache.pinot.spi.config.table.PageCacheWarmupConfig; +import org.apache.pinot.spi.config.table.PageCacheWarmupRequest; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.filesystem.PinotFS; +import org.apache.pinot.spi.filesystem.PinotFSFactory; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.builder.TableNameBuilder; +import org.apache.pinot.spi.utils.retry.RetryPolicies; +import org.apache.pinot.spi.utils.retry.RetryPolicy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/// Executes a "page‑cache warm‑up" after segments are refreshed. +/// +/// <p>The executor reads the warm‑up query file from +/// <code>{controllerConf.pageCacheWarmupDataDir}/{tableNameWithType}/queries</code>, +/// prepends {@code SET isSecondaryWorkload=true;} to intend the server to run the query on +/// its secondary workload queue, wraps the list in a +/// {@link PageCacheWarmupRequest}, +/// and POSTs the request to every server that hosts the table.</p> +/// +/// <p>The call is retried with an exponential back‑off (3 attempts, starting at +/// 3 seconds) and the overall fan‑out is bounded by the controller‑level +/// {@code controller.page.cache.warmup.duration.ms} config. +/// </p> +/// +/// <h2>Sequence</h2> +/// <ol> +/// <li>Validate that the table has warm‑up enabled.</li> +/// <li>Load the most recently modified warm‑up file (or throws if none exist).</li> +/// <li>Add the <em>secondary workload</em> hint to each query.</li> +/// <li>Look up server admin endpoints via the Helix resource manager.</li> +/// <li>Send parallel warm‑up requests and wait until all complete or the timeout elapses.</li> +/// </ol> +/// +public class PageCacheWarmupControllerExecutor { + private static final Logger LOGGER = LoggerFactory.getLogger(PageCacheWarmupControllerExecutor.class); + + private static final RetryPolicy DEFAULT_RETRY_POLICY = RetryPolicies.exponentialBackoffRetryPolicy(3, 3000L, 2.0f); + + private final PinotHelixResourceManager _pinotHelixResourceManager; + private final ControllerMetrics _controllerMetrics; + private final String _pageCacheWarmupQueriesDataDir; + private final long _maxPageCacheWarmupDurationMs; + private final ExecutorService _warmupRequestExecutor; + + /// Creates an executor bound to the given Controller services. The warmup query directory and the + /// overall warmup duration are read from the {@link ControllerConf} (defaults are used when it is + /// {@code null}). + public PageCacheWarmupControllerExecutor(PinotHelixResourceManager pinotHelixResourceManager, + @Nullable ControllerConf controllerConf) { + _pinotHelixResourceManager = pinotHelixResourceManager; + _controllerMetrics = ControllerMetrics.get(); + _pageCacheWarmupQueriesDataDir = + controllerConf != null ? controllerConf.getPageCacheWarmupQueriesDataDir() : null; + _maxPageCacheWarmupDurationMs = + controllerConf != null ? controllerConf.getPageCacheWarmupDurationMs() + : ControllerConf.DEFAULT_PAGE_CACHE_WARMUP_DURATION_MS; + _warmupRequestExecutor = Executors.newCachedThreadPool(runnable -> { + Thread thread = new Thread(runnable, "page-cache-warmup-request"); + thread.setDaemon(true); + return thread; + }); + } + + /// Orchestrates page‑cache warm‑up for the specified table. + /// + /// <p>The method spawns an asynchronous task that: + /// <ul> + /// <li>Loads the most recently modified warm‑up file (or throws if none exist).</li> + /// <li>Builds a {@link PageCacheWarmupRequest} that + /// optionally restricts the warm‑up to the supplied segment list.</li> + /// <li>Sends the request to every server in parallel with retry semantics.</li> + /// </ul> + /// The calling thread blocks only until the task finishes or the + /// warm‑up timeout defined in the table config expires.</p> + /// + /// @param tableNameWithType fully‑qualified table name, e.g. {@code myTable_OFFLINE} + /// @param segmentsTo list of segment names to touch; {@code null} or empty means all segments + public void triggerPageCacheWarmup(String tableNameWithType, List<String> segmentsTo) { + try { + TableType tableType = TableNameBuilder.getTableTypeFromTableName(tableNameWithType); + if (tableType != TableType.OFFLINE) { + return; + } + String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType); + TableConfig tableConfig = _pinotHelixResourceManager.getOfflineTableConfig(rawTableName); + if (tableConfig == null) { + return; + } + PageCacheWarmupConfig pageCacheWarmupConfig = tableConfig.getPageCacheWarmupConfig(); + PageCacheWarmupConfig.Spec spec = pageCacheWarmupConfig != null ? pageCacheWarmupConfig.getOnRefresh() : null; + if (spec == null || !spec.isEnabled()) { + return; + } + + LOGGER.info("Starting page cache warmup for table: {}, maxWarmupDurationMs: {}", tableNameWithType, + _maxPageCacheWarmupDurationMs); + _controllerMetrics.addMeteredGlobalValue(ControllerMeter.PAGE_CACHE_WARMUP_REQUESTS, 1); + + PinotFS pinotFS = PinotFSFactory.create(URIUtils.getUri(_pageCacheWarmupQueriesDataDir).getScheme()); + File tableDir = new File(_pageCacheWarmupQueriesDataDir, tableNameWithType); + File[] files = tableDir.listFiles(File::isFile); Review Comment: Could we keep this lookup entirely within `PinotFS`? `_pageCacheWarmupQueriesDataDir` is resolved to a `PinotFS` above, and the upload/REST-read paths support filesystem URI schemes, but `java.io.File.listFiles()` can only enumerate local/POSIX paths. With an S3/HDFS-style data directory this will return `null`, causing `onRefresh` warmup to be silently skipped; `queryFile.toURI()` would also produce a `file://` URI. This is especially relevant for multiple controllers, where the query files may need shared storage. `PinotFS` supports local disk through `LocalPinotFS`, so this could use `pinotFS.listFiles(tableDirUri, false)`, filter entries with `!pinotFS.isDirectory(uri)`, select by `pinotFS.lastModified(uri)`, and then call `pinotFS.open(uri)`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
