hudi-agent commented on code in PR #18419:
URL: https://github.com/apache/hudi/pull/18419#discussion_r4163653751
##########
hudi-io/src/main/java/org/apache/hudi/io/hfile/CachingHFileReaderImpl.java:
##########
@@ -19,125 +19,140 @@
package org.apache.hudi.io.hfile;
+import org.apache.hudi.common.util.Lazy;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.io.SeekableDataInputStream;
-import lombok.extern.slf4j.Slf4j;
-
import java.io.IOException;
-import java.util.concurrent.TimeUnit;
+import java.nio.ByteBuffer;
+import java.util.List;
/**
- * HFile reader implementation with integrated caching functionality. This
extends BaseHFileReaderImpl and overrides the block instantiation method to add
caching capabilities.
- * <p>
- * Uses a shared static cache across all instances to maximize cache hits when
multiple readers access the same file.
+ * HFile reader implementation with integrated caching functionality.
+ * Uses shared caches across all instances to maximize cache hits when
multiple readers access the same file.
*/
-@Slf4j
public class CachingHFileReaderImpl extends HFileReaderImpl {
- private static volatile HFileBlockCache GLOBAL_BLOCK_CACHE;
- // Store first config values to check against cache config
- private static volatile Integer INITIAL_CACHE_SIZE;
- private static volatile Integer INITIAL_CACHE_TTL;
- private static final Object CACHE_LOCK = new Object();
-
private final String filePath;
-
- public CachingHFileReaderImpl(SeekableDataInputStream stream, long fileSize,
String filePath, int cacheSize, int cacheTtlMinutes) {
- super(stream, fileSize);
+ private final HFileReaderCacheManager cacheManager;
+
+ public CachingHFileReaderImpl(Lazy<SeekableDataInputStream> lazyStream,
+ Lazy<Long> lazyFileSize,
+ String filePath,
+ int blockCacheSize,
+ long cacheMaxWeightBytes,
+ int loadOnOpenCacheSize,
+ int cacheTtlMinutes) {
+ super(lazyStream, lazyFileSize);
this.filePath = filePath;
- // Initialize global cache with provided config (ignored if already
initialized)
- getGlobalCache(cacheSize, cacheTtlMinutes);
+ this.cacheManager = HFileReaderCacheManager.getInstance(
+ blockCacheSize, cacheMaxWeightBytes, loadOnOpenCacheSize,
cacheTtlMinutes);
}
- /**
- * Gets or creates the global cache shared by all CachingHFileReaderImpl
instances.
- * Thread-safe singleton pattern with double-checked locking.
- */
- private static HFileBlockCache getGlobalCache(int cacheSize, int
cacheTtlMinutes) {
- if (GLOBAL_BLOCK_CACHE == null) {
- synchronized (CACHE_LOCK) {
- if (GLOBAL_BLOCK_CACHE == null) {
- log.info("Initializing global HFileBlockCache with size: {}, TTL: {}
minutes.",
- cacheSize, cacheTtlMinutes);
- // Store the config used for initialization
- INITIAL_CACHE_SIZE = cacheSize;
- INITIAL_CACHE_TTL = cacheTtlMinutes;
- GLOBAL_BLOCK_CACHE = new HFileBlockCache(
- cacheSize,
- cacheTtlMinutes,
- TimeUnit.MINUTES);
- } else if (!INITIAL_CACHE_SIZE.equals(cacheSize) ||
!INITIAL_CACHE_TTL.equals(cacheTtlMinutes)) {
- // Log a warning if a different config is provided after
initialization
- log.warn("HFile block cache is already initialized. The provided
configuration is being ignored. "
- + "Existing config: [Size: {}, TTL: {} mins], Ignored
config: [Size: {}, TTL: {} mins].",
- INITIAL_CACHE_SIZE, INITIAL_CACHE_TTL,
- cacheSize, cacheTtlMinutes);
- }
- }
+ @Override
+ public synchronized void initializeMetadata() throws IOException {
+ if (this.isMetadataInitialized) {
+ return;
}
- return GLOBAL_BLOCK_CACHE;
+
+ HFileReaderCacheManager.LoadOnOpenBlocks loadOnOpenBlocks =
getOrComputeLoadOnOpenData();
+ this.trailer = loadOnOpenBlocks.trailer;
+ this.context = HFileContext.builder()
+ .compressionCodec(trailer.getCompressionCodec())
+ .build();
+ this.dataBlockIndexEntryMap =
+ readDataBlockIndex(loadOnOpenBlocks.rootDataIndexBlock,
trailer.getDataIndexCount(), trailer.getNumDataIndexLevels());
+ this.metaBlockIndexEntryMap =
+
loadOnOpenBlocks.metaRootIndexBlock.readBlockIndex(trailer.getMetaIndexCount(),
true);
+ this.fileInfo = loadOnOpenBlocks.fileInfoBlock.readFileInfo();
+ this.isMetadataInitialized = true;
}
@Override
- public HFileDataBlock instantiateHFileDataBlock(BlockIndexEntry blockToRead)
throws IOException {
- HFileBlockCache.BlockCacheKey cacheKey = new HFileBlockCache.BlockCacheKey(
- filePath, blockToRead.getOffset(), blockToRead.getSize());
-
- try {
- HFileBlock block = GLOBAL_BLOCK_CACHE.getOrCompute(cacheKey, () ->
super.instantiateHFileDataBlock(blockToRead));
- return (HFileDataBlock) block;
- } catch (IOException | RuntimeException e) {
- throw e;
- } catch (Exception e) {
- throw new IOException("Failed to load HFile block", e);
+ public Option<ByteBuffer> getMetaBlock(String metaBlockName) throws
IOException {
+ initializeMetadata();
+ BlockIndexEntry blockIndexEntry = metaBlockIndexEntryMap.get(new
UTF8StringKey(metaBlockName));
+ if (blockIndexEntry == null) {
+ return Option.empty();
}
+ HFileMetaBlock block = getOrComputeBlock(
+ blockIndexEntry.getOffset(), blockIndexEntry.getSize(),
HFileBlockType.META, HFileMetaBlock.class);
+ return Option.of(block.readContent());
}
@Override
- public void close() throws IOException {
- // NOTE: Do not clear the shared cache when closing individual readers
- // The cache is shared across all instances
- super.close();
+ public HFileDataBlock instantiateHFileDataBlock(BlockIndexEntry blockToRead)
throws IOException {
+ return getOrComputeBlock(
+ blockToRead.getOffset(), blockToRead.getSize(), HFileBlockType.DATA,
HFileDataBlock.class);
+ }
+
+ @Override
+ protected List<BlockIndexEntry> readDataBlockIndexEntries(BlockIndexEntry
indexEntry,
+ HFileBlockType
blockType) throws IOException {
+ HFileLeafIndexBlock block = getOrComputeBlock(indexEntry.getOffset(),
indexEntry.getSize(), blockType, HFileLeafIndexBlock.class);
+ return block.readBlockIndex();
}
- /**
- * Gets current cache size from the global cache.
- *
- * @return number of cached blocks
- */
public long getCacheSize() {
- return GLOBAL_BLOCK_CACHE != null ? GLOBAL_BLOCK_CACHE.size() : 0;
+ return cacheManager.getBlockCacheSize();
}
- /**
- * Clears the global block cache.
- */
public void clearCache() {
- if (GLOBAL_BLOCK_CACHE != null) {
- GLOBAL_BLOCK_CACHE.clear();
- }
+ cacheManager.clear();
}
- /**
- * Gets cache statistics for monitoring optimization effectiveness.
- *
- * @return string representation of cache statistics
- */
public String getCacheStats() {
- return "HFileReader Cache Stats - Size: " + (GLOBAL_BLOCK_CACHE != null ?
GLOBAL_BLOCK_CACHE.size() : 0);
+ return cacheManager.getStats();
}
/**
- * Clears the global cache. Should only be used for testing.
+ * Clears the global caches. Should only be used for testing.
*/
public static void resetGlobalCache() {
- synchronized (CACHE_LOCK) {
- if (GLOBAL_BLOCK_CACHE != null) {
- GLOBAL_BLOCK_CACHE.clear();
- GLOBAL_BLOCK_CACHE = null;
+ HFileReaderCacheManager.reset();
+ }
+
+ private HFileReaderCacheManager.LoadOnOpenBlocks
getOrComputeLoadOnOpenData() throws IOException {
+ return cacheManager.getOrComputeLoadOnOpenData(filePath, () -> {
Review Comment:
🤖 For `withContent(...)` readers (inline HFile log blocks), `filePath` is
only the 32-bit murmur hash of the bytes. A collision used to also need
matching block (offset, size). Now the hash alone swaps in another block's
trailer/index/file info, and the follow-up block lookups hit that file's cached
blocks, so wrong MDT records come back silently. Could we add the content
length to the key or use a wider hash, or skip the load-on-open cache for
in-memory content, where there's no I/O to save?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileBlockCache.java:
##########
@@ -30,16 +33,27 @@
/**
* Least Frequently Used (LFU) cache for HFile blocks to improve read
performance by avoiding repeated block reads.
* Uses Caffeine cache with configurable size and TTL. Thread-safe for
concurrent access.
+ * The cache is bounded by entry count, or by retained bytes when a positive
max weight is configured.
*/
public class HFileBlockCache {
private final Cache<BlockCacheKey, HFileBlock> cache;
public HFileBlockCache(int maxCacheSize, long expireAfterWrite, TimeUnit
timeUnit) {
- this.cache = Caffeine.newBuilder()
- .maximumSize(maxCacheSize)
+ this(maxCacheSize, 0L, expireAfterWrite, timeUnit);
+ }
+
+ public HFileBlockCache(int maxCacheSize, long maxWeightBytes, long
expireAfterWrite, TimeUnit timeUnit) {
+ Caffeine<Object, Object> builder = Caffeine.newBuilder()
.expireAfterAccess(Duration.ofMillis(timeUnit.toMillis(expireAfterWrite)))
- .build();
+ .recordStats();
+ if (maxWeightBytes > 0L) {
+ this.cache = builder.maximumWeight(maxWeightBytes)
+ .weigher((BlockCacheKey key, HFileBlock block) -> Math.max(1,
block.heapSize()))
Review Comment:
🤖 `heapSize()` only counts header + uncompressed bytes. For compressed
blocks (the HFile default is GZ), each cached block also keeps its whole
on-disk reader buffer alive through `compressedByteBuff`, so the cache can go
over `max.weight.mb` by roughly the compressed size (up to ~2x for poorly
compressible data). Since each block-cache entry owns its own reader buffer,
could the weigher sum `retainedBuffers()`? Or `unpack()` could drop
`compressedByteBuff` once it's done, which would also shrink both caches.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileReaderCacheManager.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Central manager for all shared caches used by {@link
CachingHFileReaderImpl}.
+ *
+ * <p>The manager intentionally keeps block-level cache and load-on-open
metadata 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>Load-on-open cache stores file-scoped metadata bundles keyed only by
file identity.</li>
+ * </ul>
+ */
+@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, LoadOnOpenBlocks> loadOnOpenDataCache;
+ private final int blockCacheSize;
+ private final long cacheMaxWeightBytes;
+ private final int loadOnOpenDataCacheSize;
+ private final int cacheTtlMinutes;
+
+ private HFileReaderCacheManager(int blockCacheSize,
+ long cacheMaxWeightBytes,
+ int loadOnOpenDataCacheSize,
+ int cacheTtlMinutes) {
+ this.blockCacheSize = blockCacheSize;
+ this.cacheMaxWeightBytes = cacheMaxWeightBytes;
+ this.loadOnOpenDataCacheSize = loadOnOpenDataCacheSize;
+ this.cacheTtlMinutes = cacheTtlMinutes;
+ log.info("Initializing global HFileBlockCache with size: {} blocks,
maxWeightBytes: {}, TTL: {} minutes.",
+ blockCacheSize, cacheMaxWeightBytes, cacheTtlMinutes);
+ log.info("Initializing global load-on-open data cache with size: {} files,
maxWeightBytes: {}, TTL: {} minutes.",
+ loadOnOpenDataCacheSize, cacheMaxWeightBytes, cacheTtlMinutes);
+ this.blockCache = new HFileBlockCache(blockCacheSize, cacheMaxWeightBytes,
cacheTtlMinutes, TimeUnit.MINUTES);
+ Caffeine<Object, Object> builder = Caffeine.newBuilder()
+ .expireAfterAccess(Duration.ofMinutes(cacheTtlMinutes))
+ .recordStats();
+ if (cacheMaxWeightBytes > 0L) {
+ this.loadOnOpenDataCache = builder.maximumWeight(cacheMaxWeightBytes)
+ .weigher((String key, LoadOnOpenBlocks blocks) -> (int)
Math.min(Integer.MAX_VALUE, Math.max(1L, blocks.heapSize())))
+ .build();
+ } else {
+ this.loadOnOpenDataCache =
builder.maximumSize(loadOnOpenDataCacheSize).build();
+ }
+ }
+
+ /**
+ * Returns the shared manager, creating it with the given configuration on
first use. The
+ * configuration of later calls is ignored; a positive {@code
cacheMaxWeightBytes} bounds each
+ * of the two caches by retained bytes instead of by entry count.
+ */
+ public static HFileReaderCacheManager getInstance(int blockCacheSize,
+ long cacheMaxWeightBytes,
+ int
loadOnOpenDataCacheSize,
+ int cacheTtlMinutes) {
+ if (INSTANCE == null) {
+ synchronized (INSTANCE_LOCK) {
+ if (INSTANCE == null) {
+ INSTANCE = new HFileReaderCacheManager(blockCacheSize,
cacheMaxWeightBytes, loadOnOpenDataCacheSize, cacheTtlMinutes);
+ } else {
+ INSTANCE.warnIfConfigIgnored(blockCacheSize, cacheMaxWeightBytes,
loadOnOpenDataCacheSize, cacheTtlMinutes);
+ }
+ }
+ } else {
+ INSTANCE.warnIfConfigIgnored(blockCacheSize, cacheMaxWeightBytes,
loadOnOpenDataCacheSize, cacheTtlMinutes);
Review Comment:
🤖 This `warnIfConfigIgnored` call now runs on every `CachingHFileReaderImpl`
construction. The old `getGlobalCache` only warned inside the one-time init
race. Paths like
`HoodieHFileDataBlock.lookupRecords`/`deserializeRecords(byte[], type)` and
`HFileUtils.readFooter` build readers with default props, so once a user tunes
any cache config, every reader on the other side logs one or two WARNs (per
lookup or log block). Could we warn once, e.g. behind an AtomicBoolean?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileBlock.java:
##########
@@ -217,6 +219,33 @@ public int getOnDiskSizeWithHeader() {
return onDiskSizeWithoutHeader + HFILEBLOCK_HEADER_SIZE;
}
+ /**
+ * Returns this block's content span for byte-weighted caching. The backing
array can hold
+ * several blocks, so weighing its full length would over-count a block
sliced from it.
+ */
Review Comment:
🤖 nit: the fallback logic here (falling back to `byteBuff.length` or
`getOnDiskSizeWithHeader()` when `contentSpan <= HFILEBLOCK_HEADER_SIZE`) is a
bit hard to follow — might help to add a short comment explaining when that
branch is actually hit.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-io/src/main/java/org/apache/hudi/io/hfile/HFileReaderCacheManager.java:
##########
@@ -0,0 +1,245 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Central manager for all shared caches used by {@link
CachingHFileReaderImpl}.
+ *
+ * <p>The manager intentionally keeps block-level cache and load-on-open
metadata 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>Load-on-open cache stores file-scoped metadata bundles keyed only by
file identity.</li>
+ * </ul>
+ */
+@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, LoadOnOpenBlocks> loadOnOpenDataCache;
+ private final int blockCacheSize;
+ private final long cacheMaxWeightBytes;
+ private final int loadOnOpenDataCacheSize;
+ private final int cacheTtlMinutes;
+
+ private HFileReaderCacheManager(int blockCacheSize,
+ long cacheMaxWeightBytes,
+ int loadOnOpenDataCacheSize,
+ int cacheTtlMinutes) {
+ this.blockCacheSize = blockCacheSize;
+ this.cacheMaxWeightBytes = cacheMaxWeightBytes;
+ this.loadOnOpenDataCacheSize = loadOnOpenDataCacheSize;
+ this.cacheTtlMinutes = cacheTtlMinutes;
+ log.info("Initializing global HFileBlockCache with size: {} blocks,
maxWeightBytes: {}, TTL: {} minutes.",
+ blockCacheSize, cacheMaxWeightBytes, cacheTtlMinutes);
+ log.info("Initializing global load-on-open data cache with size: {} files,
maxWeightBytes: {}, TTL: {} minutes.",
+ loadOnOpenDataCacheSize, cacheMaxWeightBytes, cacheTtlMinutes);
+ this.blockCache = new HFileBlockCache(blockCacheSize, cacheMaxWeightBytes,
cacheTtlMinutes, TimeUnit.MINUTES);
+ Caffeine<Object, Object> builder = Caffeine.newBuilder()
+ .expireAfterAccess(Duration.ofMinutes(cacheTtlMinutes))
+ .recordStats();
+ if (cacheMaxWeightBytes > 0L) {
+ this.loadOnOpenDataCache = builder.maximumWeight(cacheMaxWeightBytes)
+ .weigher((String key, LoadOnOpenBlocks blocks) -> (int)
Math.min(Integer.MAX_VALUE, Math.max(1L, blocks.heapSize())))
+ .build();
+ } else {
+ this.loadOnOpenDataCache =
builder.maximumSize(loadOnOpenDataCacheSize).build();
+ }
+ }
+
+ /**
+ * Returns the shared manager, creating it with the given configuration on
first use. The
+ * configuration of later calls is ignored; a positive {@code
cacheMaxWeightBytes} bounds each
+ * of the two caches by retained bytes instead of by entry count.
+ */
+ public static HFileReaderCacheManager getInstance(int blockCacheSize,
+ long cacheMaxWeightBytes,
+ int
loadOnOpenDataCacheSize,
+ int cacheTtlMinutes) {
Review Comment:
🤖 nit: `getInstance` calls `warnIfConfigIgnored` in both the branch where
INSTANCE was just created inside the sync block and the outer else-branch —
worth collapsing into a single call after the synchronized block to avoid the
duplicated path.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-io/src/main/java/org/apache/hudi/io/hfile/CachingHFileReaderImpl.java:
##########
@@ -19,125 +19,140 @@
package org.apache.hudi.io.hfile;
+import org.apache.hudi.common.util.Lazy;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.io.SeekableDataInputStream;
-import lombok.extern.slf4j.Slf4j;
-
import java.io.IOException;
-import java.util.concurrent.TimeUnit;
+import java.nio.ByteBuffer;
+import java.util.List;
/**
- * HFile reader implementation with integrated caching functionality. This
extends BaseHFileReaderImpl and overrides the block instantiation method to add
caching capabilities.
- * <p>
- * Uses a shared static cache across all instances to maximize cache hits when
multiple readers access the same file.
+ * HFile reader implementation with integrated caching functionality.
+ * Uses shared caches across all instances to maximize cache hits when
multiple readers access the same file.
*/
-@Slf4j
public class CachingHFileReaderImpl extends HFileReaderImpl {
- private static volatile HFileBlockCache GLOBAL_BLOCK_CACHE;
- // Store first config values to check against cache config
- private static volatile Integer INITIAL_CACHE_SIZE;
- private static volatile Integer INITIAL_CACHE_TTL;
- private static final Object CACHE_LOCK = new Object();
-
private final String filePath;
-
- public CachingHFileReaderImpl(SeekableDataInputStream stream, long fileSize,
String filePath, int cacheSize, int cacheTtlMinutes) {
- super(stream, fileSize);
+ private final HFileReaderCacheManager cacheManager;
+
+ public CachingHFileReaderImpl(Lazy<SeekableDataInputStream> lazyStream,
+ Lazy<Long> lazyFileSize,
+ String filePath,
+ int blockCacheSize,
+ long cacheMaxWeightBytes,
+ int loadOnOpenCacheSize,
+ int cacheTtlMinutes) {
+ super(lazyStream, lazyFileSize);
this.filePath = filePath;
- // Initialize global cache with provided config (ignored if already
initialized)
- getGlobalCache(cacheSize, cacheTtlMinutes);
+ this.cacheManager = HFileReaderCacheManager.getInstance(
+ blockCacheSize, cacheMaxWeightBytes, loadOnOpenCacheSize,
cacheTtlMinutes);
}
- /**
- * Gets or creates the global cache shared by all CachingHFileReaderImpl
instances.
- * Thread-safe singleton pattern with double-checked locking.
- */
- private static HFileBlockCache getGlobalCache(int cacheSize, int
cacheTtlMinutes) {
- if (GLOBAL_BLOCK_CACHE == null) {
- synchronized (CACHE_LOCK) {
- if (GLOBAL_BLOCK_CACHE == null) {
- log.info("Initializing global HFileBlockCache with size: {}, TTL: {}
minutes.",
- cacheSize, cacheTtlMinutes);
- // Store the config used for initialization
- INITIAL_CACHE_SIZE = cacheSize;
- INITIAL_CACHE_TTL = cacheTtlMinutes;
- GLOBAL_BLOCK_CACHE = new HFileBlockCache(
- cacheSize,
- cacheTtlMinutes,
- TimeUnit.MINUTES);
- } else if (!INITIAL_CACHE_SIZE.equals(cacheSize) ||
!INITIAL_CACHE_TTL.equals(cacheTtlMinutes)) {
- // Log a warning if a different config is provided after
initialization
- log.warn("HFile block cache is already initialized. The provided
configuration is being ignored. "
- + "Existing config: [Size: {}, TTL: {} mins], Ignored
config: [Size: {}, TTL: {} mins].",
- INITIAL_CACHE_SIZE, INITIAL_CACHE_TTL,
- cacheSize, cacheTtlMinutes);
- }
- }
+ @Override
+ public synchronized void initializeMetadata() throws IOException {
+ if (this.isMetadataInitialized) {
+ return;
}
- return GLOBAL_BLOCK_CACHE;
+
+ HFileReaderCacheManager.LoadOnOpenBlocks loadOnOpenBlocks =
getOrComputeLoadOnOpenData();
+ this.trailer = loadOnOpenBlocks.trailer;
+ this.context = HFileContext.builder()
+ .compressionCodec(trailer.getCompressionCodec())
+ .build();
+ this.dataBlockIndexEntryMap =
+ readDataBlockIndex(loadOnOpenBlocks.rootDataIndexBlock,
trailer.getDataIndexCount(), trailer.getNumDataIndexLevels());
+ this.metaBlockIndexEntryMap =
+
loadOnOpenBlocks.metaRootIndexBlock.readBlockIndex(trailer.getMetaIndexCount(),
true);
+ this.fileInfo = loadOnOpenBlocks.fileInfoBlock.readFileInfo();
+ this.isMetadataInitialized = true;
}
@Override
- public HFileDataBlock instantiateHFileDataBlock(BlockIndexEntry blockToRead)
throws IOException {
- HFileBlockCache.BlockCacheKey cacheKey = new HFileBlockCache.BlockCacheKey(
- filePath, blockToRead.getOffset(), blockToRead.getSize());
-
- try {
- HFileBlock block = GLOBAL_BLOCK_CACHE.getOrCompute(cacheKey, () ->
super.instantiateHFileDataBlock(blockToRead));
- return (HFileDataBlock) block;
- } catch (IOException | RuntimeException e) {
- throw e;
- } catch (Exception e) {
- throw new IOException("Failed to load HFile block", e);
+ public Option<ByteBuffer> getMetaBlock(String metaBlockName) throws
IOException {
+ initializeMetadata();
+ BlockIndexEntry blockIndexEntry = metaBlockIndexEntryMap.get(new
UTF8StringKey(metaBlockName));
+ if (blockIndexEntry == null) {
+ return Option.empty();
}
+ HFileMetaBlock block = getOrComputeBlock(
+ blockIndexEntry.getOffset(), blockIndexEntry.getSize(),
HFileBlockType.META, HFileMetaBlock.class);
+ return Option.of(block.readContent());
}
@Override
- public void close() throws IOException {
- // NOTE: Do not clear the shared cache when closing individual readers
- // The cache is shared across all instances
- super.close();
+ public HFileDataBlock instantiateHFileDataBlock(BlockIndexEntry blockToRead)
throws IOException {
+ return getOrComputeBlock(
+ blockToRead.getOffset(), blockToRead.getSize(), HFileBlockType.DATA,
HFileDataBlock.class);
+ }
+
+ @Override
+ protected List<BlockIndexEntry> readDataBlockIndexEntries(BlockIndexEntry
indexEntry,
+ HFileBlockType
blockType) throws IOException {
+ HFileLeafIndexBlock block = getOrComputeBlock(indexEntry.getOffset(),
indexEntry.getSize(), blockType, HFileLeafIndexBlock.class);
+ return block.readBlockIndex();
}
- /**
- * Gets current cache size from the global cache.
- *
- * @return number of cached blocks
- */
public long getCacheSize() {
- return GLOBAL_BLOCK_CACHE != null ? GLOBAL_BLOCK_CACHE.size() : 0;
+ return cacheManager.getBlockCacheSize();
}
- /**
- * Clears the global block cache.
- */
public void clearCache() {
- if (GLOBAL_BLOCK_CACHE != null) {
- GLOBAL_BLOCK_CACHE.clear();
- }
+ cacheManager.clear();
}
- /**
- * Gets cache statistics for monitoring optimization effectiveness.
- *
- * @return string representation of cache statistics
- */
public String getCacheStats() {
- return "HFileReader Cache Stats - Size: " + (GLOBAL_BLOCK_CACHE != null ?
GLOBAL_BLOCK_CACHE.size() : 0);
+ return cacheManager.getStats();
}
/**
- * Clears the global cache. Should only be used for testing.
+ * Clears the global caches. Should only be used for testing.
*/
public static void resetGlobalCache() {
- synchronized (CACHE_LOCK) {
- if (GLOBAL_BLOCK_CACHE != null) {
- GLOBAL_BLOCK_CACHE.clear();
- GLOBAL_BLOCK_CACHE = null;
+ HFileReaderCacheManager.reset();
+ }
+
+ private HFileReaderCacheManager.LoadOnOpenBlocks
getOrComputeLoadOnOpenData() throws IOException {
+ return cacheManager.getOrComputeLoadOnOpenData(filePath, () -> {
+ SeekableDataInputStream stream = getStream();
+ long fileSize = lazyFileSize.get();
+ HFileTrailer loadedTrailer = readTrailer(stream, fileSize);
+ HFileContext loadOnOpenContext = HFileContext.builder()
+ .compressionCodec(loadedTrailer.getCompressionCodec())
+ .build();
+ HFileBlockReader blockReader = new HFileBlockReader(loadOnOpenContext,
stream,
+ loadedTrailer.getLoadOnOpenDataOffset(), fileSize -
HFileTrailer.getTrailerSize());
+ HFileRootIndexBlock rootDataIndexBlock = (HFileRootIndexBlock)
blockReader.nextBlock(HFileBlockType.ROOT_INDEX);
+ HFileRootIndexBlock metaRootIndexBlock = (HFileRootIndexBlock)
blockReader.nextBlock(HFileBlockType.ROOT_INDEX);
+ HFileFileInfoBlock fileInfoBlock = (HFileFileInfoBlock)
blockReader.nextBlock(HFileBlockType.FILE_INFO);
+ return new HFileReaderCacheManager.LoadOnOpenBlocks(
+ loadedTrailer, rootDataIndexBlock, metaRootIndexBlock,
fileInfoBlock);
+ });
+ }
+
+ private <T extends HFileBlock> T getOrComputeBlock(long offset,
+ int size,
+ HFileBlockType
expectedBlockType,
+ Class<T> blockClass)
throws IOException {
+ return cacheManager.getOrComputeBlock(filePath, offset, size, blockClass,
() -> {
+ HFileBlockReader blockReader = new HFileBlockReader(context,
getStream(), offset, offset + size);
+ return blockReader.nextBlock(expectedBlockType);
+ });
+ }
+
+ /**
Review Comment:
🤖 nit: `getStream()` unwrapping `HoodieIOException` back to the original
`IOException` is a subtle contract bridge — could you add a one-line comment
near the `Lazy<SeekableDataInputStream>`/`Lazy<Long>` fields in HFileReaderImpl
pointing here, so future readers know why the lambda wraps in HoodieIOException
instead of just throwing IOException directly?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
