cshuo commented on code in PR #18419:
URL: https://github.com/apache/hudi/pull/18419#discussion_r4213943611
##########
hudi-common/src/main/java/org/apache/hudi/core/io/storage/HFileReaderFactory.java:
##########
@@ -48,44 +52,87 @@ public class HFileReaderFactory {
private final HoodieMetadataConfig metadataConfig;
private final TypedProperties properties;
private final Either<StoragePath, byte[]> fileSource;
+ private final Option<StoragePathInfo> pathInfoOpt;
private Option<Long> fileSizeOpt;
private HFileReaderFactory(HoodieStorage storage,
TypedProperties properties,
Either<StoragePath, byte[]> fileSource,
+ Option<StoragePathInfo> pathInfoOpt,
Option<Long> fileSizeOpt) {
this.storage = storage;
this.metadataConfig =
HoodieMetadataConfig.newBuilder().withProperties(properties).build();
this.properties = properties;
this.fileSource = fileSource;
+ this.pathInfoOpt = pathInfoOpt;
this.fileSizeOpt = fileSizeOpt;
}
public HFileReader createHFileReader() throws IOException {
+ if (!shouldEnableBlockCaching()) {
+ final long fileSize = getFileSize();
+ return new HFileReaderImpl(createInputStream(fileSize), fileSize);
+ }
+
+ int blockCacheSize = ConfigUtils.getIntWithAltKeys(
+ properties, HoodieReaderConfig.HFILE_BLOCK_CACHE_SIZE);
+ int loadOnOpenCacheSize = ConfigUtils.getIntWithAltKeys(
+ properties, HoodieReaderConfig.HFILE_LOAD_ON_OPEN_CACHE_SIZE);
+ int cacheTtlMinutes = ConfigUtils.getIntWithAltKeys(
+ properties, HoodieReaderConfig.HFILE_BLOCK_CACHE_TTL_MINUTES);
+ long cacheMaxWeightBytes = (long) ConfigUtils.getIntWithAltKeys(
+ properties, HoodieReaderConfig.HFILE_BLOCK_CACHE_MAX_WEIGHT_MB) *
1024L * 1024L;
+ HFileReaderCacheManager cacheManager = HFileReaderCacheManager.getInstance(
+ blockCacheSize, cacheMaxWeightBytes, loadOnOpenCacheSize,
cacheTtlMinutes, isCacheExplicitlyConfigured());
+ // The caching reader opens the stream only on a cache miss, so both the
size lookup and the
+ // open are deferred; their IOExceptions are unwrapped again by the reader.
+ final Lazy<Long> lazyFileSize = Lazy.lazily(() -> {
+ try {
+ return getFileSize();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to determine the HFile size.", e);
+ }
+ });
+ final Lazy<SeekableDataInputStream> lazyStream = Lazy.lazily(() -> {
+ try {
+ return createInputStream(lazyFileSize.get());
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to create input stream.", e);
+ }
+ });
+ return new CachingHFileReaderImpl(lazyStream, lazyFileSize,
getFileIdentity(), fileSource.isLeft(), cacheManager);
+ }
+
+ private long getFileSize() throws IOException {
if (fileSizeOpt.isEmpty()) {
fileSizeOpt = Option.of(determineFileSize());
}
- final long fileSize = fileSizeOpt.get();
- final SeekableDataInputStream inputStream = createInputStream(fileSize);
-
- if (shouldEnableBlockCaching()) {
- int blockCacheSize = ConfigUtils.getIntWithAltKeys(
- properties, HoodieReaderConfig.HFILE_BLOCK_CACHE_SIZE);
- int cacheTtlMinutes = ConfigUtils.getIntWithAltKeys(
- properties, HoodieReaderConfig.HFILE_BLOCK_CACHE_TTL_MINUTES);
- String filePath = getFilePath();
- return new CachingHFileReaderImpl(inputStream, fileSize, filePath,
blockCacheSize, cacheTtlMinutes);
- }
-
- return new HFileReaderImpl(inputStream, fileSize);
+ return fileSizeOpt.get();
}
private boolean shouldEnableBlockCaching() {
return ConfigUtils.getBooleanWithAltKeys(properties,
HoodieReaderConfig.HFILE_BLOCK_CACHE_ENABLED);
}
- private String getFilePath() {
+ private boolean isCacheExplicitlyConfigured() {
Review Comment:
isCacheExplicitlyConfigured() checks only whether configuration keys exist.
However, getHFileCacheConfigs(emptyMap), buildFileGroupReaderProperties(), and
createMetadataWriteConfig() all write default size, TTL, and weight values into
the properties.
Consequently, an ordinary Spark or metadata reader using default settings
can initialize the manager with explicitlyConfigured=true.
##########
hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileReaderCacheManager.java:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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.hudi.io.hfile;
+
+import org.apache.hudi.common.util.Option;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Central manager for all shared caches used by {@link
CachingHFileReaderImpl}.
+ *
+ * <p>The manager intentionally keeps the block cache and the trailer and
"load-on-open" cache
+ * separate, because they have different keys, values, and reuse patterns:
+ * <ul>
+ * <li>{@link HFileBlockCache} stores independently addressable blocks keyed
by file, offset, and size.</li>
+ * <li>The trailer and "load-on-open" cache stores one {@link
HFileTrailerAndLoadOnOpenBlocks} per
+ * file, keyed by file identity only.</li>
+ * </ul>
+ *
+ * <p>One manager is shared by all readers in the JVM. Its configuration comes
from the first
+ * reader that asks for it with an explicit configuration; readers running on
default configuration
+ * neither pin nor change it, so a default-configured reader created first
does not stop a later,
+ * explicitly configured one from taking effect.
+ */
+@Slf4j
+public final class HFileReaderCacheManager {
+
+ private static volatile HFileReaderCacheManager INSTANCE;
+ private static final Object INSTANCE_LOCK = new Object();
+
+ private final HFileBlockCache blockCache;
+ private final Cache<String, HFileTrailerAndLoadOnOpenBlocks>
trailerAndLoadOnOpenCache;
+ private final int blockCacheSize;
+ private final long cacheMaxWeightBytes;
+ private final int loadOnOpenCacheSize;
+ private final int cacheTtlMinutes;
+ private volatile boolean explicitlyConfigured;
+ private final AtomicBoolean ignoredConfigWarned = new AtomicBoolean();
+
+ private HFileReaderCacheManager(int blockCacheSize,
+ long cacheMaxWeightBytes,
+ int loadOnOpenCacheSize,
+ int cacheTtlMinutes,
+ boolean explicitlyConfigured) {
+ this.blockCacheSize = blockCacheSize;
+ this.cacheMaxWeightBytes = cacheMaxWeightBytes;
+ this.loadOnOpenCacheSize = loadOnOpenCacheSize;
+ this.cacheTtlMinutes = cacheTtlMinutes;
+ this.explicitlyConfigured = explicitlyConfigured;
+ log.info("Initializing global HFile caches: block cache size: {} blocks,
trailer and load-on-open cache size: {} files, "
+ + "maxWeightBytes: {}, TTL: {} minutes, explicitly configured:
{}.",
+ blockCacheSize, loadOnOpenCacheSize, cacheMaxWeightBytes,
cacheTtlMinutes, explicitlyConfigured);
+ this.blockCache = new HFileBlockCache(blockCacheSize, cacheMaxWeightBytes,
cacheTtlMinutes, TimeUnit.MINUTES);
+ Caffeine<Object, Object> builder = Caffeine.newBuilder()
+ .expireAfterAccess(Duration.ofMinutes(cacheTtlMinutes))
+ .recordStats();
+ if (cacheMaxWeightBytes > 0L) {
+ this.trailerAndLoadOnOpenCache =
builder.maximumWeight(cacheMaxWeightBytes)
+ .weigher((String key, HFileTrailerAndLoadOnOpenBlocks blocks) ->
(int) Math.min(Integer.MAX_VALUE, Math.max(1L, blocks.heapSize())))
+ .build();
+ } else {
+ this.trailerAndLoadOnOpenCache =
builder.maximumSize(loadOnOpenCacheSize).build();
+ }
+ }
+
+ /**
+ * Returns the shared manager, creating it on first use. A positive {@code
cacheMaxWeightBytes}
+ * bounds each of the two caches by retained bytes instead of by entry count.
+ *
+ * <p>An explicitly configured request replaces a manager that was created
on default
+ * configuration, dropping its cached entries. Once the manager is
explicitly configured, a
+ * differing explicit configuration is ignored with a single warning.
+ */
+ public static HFileReaderCacheManager getInstance(int blockCacheSize,
+ long cacheMaxWeightBytes,
+ int loadOnOpenCacheSize,
+ int cacheTtlMinutes,
+ boolean
explicitlyConfigured) {
+ HFileReaderCacheManager instance = INSTANCE;
+ if (instance != null && (!explicitlyConfigured ||
instance.explicitlyConfigured)) {
+ if (explicitlyConfigured && !instance.hasConfig(blockCacheSize,
cacheMaxWeightBytes, loadOnOpenCacheSize, cacheTtlMinutes)) {
+ instance.warnConfigIgnoredOnce(blockCacheSize, cacheMaxWeightBytes,
loadOnOpenCacheSize, cacheTtlMinutes);
+ }
+ return instance;
+ }
+ synchronized (INSTANCE_LOCK) {
+ instance = INSTANCE;
+ if (instance == null) {
+ INSTANCE = instance = new HFileReaderCacheManager(
+ blockCacheSize, cacheMaxWeightBytes, loadOnOpenCacheSize,
cacheTtlMinutes, explicitlyConfigured);
+ } else if (explicitlyConfigured && !instance.explicitlyConfigured) {
+ if (instance.hasConfig(blockCacheSize, cacheMaxWeightBytes,
loadOnOpenCacheSize, cacheTtlMinutes)) {
+ instance.explicitlyConfigured = true;
+ } else {
+ log.info("Replacing the global HFile caches created on default
configuration with explicitly configured ones.");
+ INSTANCE = instance = new HFileReaderCacheManager(
Review Comment:
When an explicitly configured reader replaces static INSTANCE, any existing
reader or iterator retains its final cacheManager reference to the previous
manager. Its entries are not cleared, and subsequent reads continue populating
the old cache.
Both cache sets can therefore remain active simultaneously. The explicit
byte budget does not govern the old cache, while globalStatsString() and
reset() operate only on the new instance. This also contradicts the comment
that replacement drops the previous entries.
--
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]
