From dd1110c719575eb111a96bdb5e85d651afe545c1 Mon Sep 17 00:00:00 2001 From: arbaazkhan1 Date: Thu, 13 Aug 2026 18:00:01 -0400 Subject: [PATCH 1/2] New option to cache data --- .../apache/accumulo/core/conf/Property.java | 23 +- .../cache/impl/BlockCacheConfiguration.java | 23 +- .../cache/impl/CompressedBlockCache.java | 173 ++++++++++++ .../blockfile/impl/BasicCacheProvider.java | 11 + .../blockfile/impl/CachableBlockFile.java | 62 +++++ .../file/blockfile/impl/CacheProvider.java | 6 + .../blockfile/impl/ScanCacheProvider.java | 17 +- .../cache/CompressedBlockCacheTest.java | 253 ++++++++++++++++++ .../tserver/TabletServerResourceManager.java | 58 +++- 9 files changed, 606 insertions(+), 20 deletions(-) create mode 100644 core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/impl/CompressedBlockCache.java create mode 100644 core/src/test/java/org/apache/accumulo/core/file/blockfile/cache/CompressedBlockCacheTest.java diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java index b148c1a60e5..b5d4da2fe60 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java @@ -594,6 +594,11 @@ Each key is the name of the pool (can be assigned any string). Each value is a J "Properties in this category affect the behavior of the scan servers.", "2.1.0"), SSERV_DATACACHE_SIZE("sserver.cache.data.size", "10%", PropertyType.MEMORY, "Specifies the size of the cache for RFile data blocks on each scan server.", "2.1.0"), + SSERV_COMPRESSED_DATACACHE_SIZE("sserver.cache.compressed.data.size", "0", PropertyType.MEMORY, + "Specifies the size of the secondary compressed cache for compressed RFile data blocks." + + "When set to a non-zero value a secondary block cache is created that stores block in compressed form." + + "A value of 0 disables the secondary compressed cache (default).", + "4.1.0"), SSERV_INDEXCACHE_SIZE("sserver.cache.index.size", "25%", PropertyType.MEMORY, "Specifies the size of the cache for RFile index blocks on each scan server.", "2.1.0"), SSERV_SUMMARYCACHE_SIZE("sserver.cache.summary.size", "10%", PropertyType.MEMORY, @@ -679,6 +684,11 @@ Each key is the name of the pool (can be assigned any string). Each value is a J "Specifies a default blocksize for the tserver caches.", "1.3.5"), TSERV_DATACACHE_SIZE("tserver.cache.data.size", "10%", PropertyType.MEMORY, "Specifies the size of the cache for RFile data blocks.", "1.3.5"), + TSERV_COMPRESSED_DATACACHE_SIZE("tserver.cache.data.compressed.size", "0", PropertyType.MEMORY, + "Specifies the size of the secondary compressed cache for compressed RFile data blocks." + + "When set to a non-zero value a secondary block cache is created that stores block in compressed form." + + "A value of 0 disables the secondary compressed cache (default).", + "4.1.0"), TSERV_INDEXCACHE_SIZE("tserver.cache.index.size", "25%", PropertyType.MEMORY, "Specifies the size of the cache for RFile index blocks.", "1.3.5"), TSERV_SUMMARYCACHE_SIZE("tserver.cache.summary.size", "10%", PropertyType.MEMORY, @@ -1720,9 +1730,10 @@ public static boolean isValidTablePropertyKey(String key) { // SSERV options SSERV_CACHED_TABLET_METADATA_REFRESH_PERCENT, SSERV_THREADCHECK, SSERV_CLIENTPORT, - SSERV_DATACACHE_SIZE, SSERV_INDEXCACHE_SIZE, SSERV_SUMMARYCACHE_SIZE, SSERV_DEFAULT_BLOCKSIZE, - SSERV_SCAN_REFERENCE_EXPIRATION_TIME, SSERV_CACHED_TABLET_METADATA_EXPIRATION, - SSERV_MINTHREADS, SSERV_MINTHREADS_TIMEOUT, SSERV_WAL_SORT_MAX_CONCURRENT, SSERV_GROUP_NAME, + SSERV_DATACACHE_SIZE, SSERV_COMPRESSED_DATACACHE_SIZE, SSERV_INDEXCACHE_SIZE, + SSERV_SUMMARYCACHE_SIZE, SSERV_DEFAULT_BLOCKSIZE, SSERV_SCAN_REFERENCE_EXPIRATION_TIME, + SSERV_CACHED_TABLET_METADATA_EXPIRATION, SSERV_MINTHREADS, SSERV_MINTHREADS_TIMEOUT, + SSERV_WAL_SORT_MAX_CONCURRENT, SSERV_GROUP_NAME, // TSERV options TSERV_TOTAL_MUTATION_QUEUE_MAX, TSERV_WAL_MAX_SIZE, TSERV_WAL_MAX_AGE, @@ -1731,9 +1742,9 @@ public static boolean isValidTablePropertyKey(String key) { TSERV_SCAN_RESULTS_MAX_TIMEOUT, TSERV_MINC_MAXCONCURRENT, TSERV_THREADCHECK, TSERV_LOG_BUSY_TABLETS_COUNT, TSERV_LOG_BUSY_TABLETS_INTERVAL, TSERV_WAL_SORT_MAX_CONCURRENT, TSERV_SLOW_FILEPERMIT_MILLIS, TSERV_WAL_BLOCKSIZE, TSERV_CLIENTPORT, TSERV_DATACACHE_SIZE, - TSERV_INDEXCACHE_SIZE, TSERV_SUMMARYCACHE_SIZE, TSERV_DEFAULT_BLOCKSIZE, TSERV_MINTHREADS, - TSERV_MINTHREADS_TIMEOUT, TSERV_NATIVEMAP_ENABLED, TSERV_MAXMEM, TSERV_SCAN_MAX_OPENFILES, - TSERV_ONDEMAND_UNLOADER_INTERVAL, TSERV_GROUP_NAME, + TSERV_COMPRESSED_DATACACHE_SIZE, TSERV_INDEXCACHE_SIZE, TSERV_SUMMARYCACHE_SIZE, + TSERV_DEFAULT_BLOCKSIZE, TSERV_MINTHREADS, TSERV_MINTHREADS_TIMEOUT, TSERV_NATIVEMAP_ENABLED, + TSERV_MAXMEM, TSERV_SCAN_MAX_OPENFILES, TSERV_ONDEMAND_UNLOADER_INTERVAL, TSERV_GROUP_NAME, // GC options GC_CANDIDATE_BATCH_SIZE, GC_CYCLE_START, GC_PORT, diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/impl/BlockCacheConfiguration.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/impl/BlockCacheConfiguration.java index efcb2997e16..a518a7e24c0 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/impl/BlockCacheConfiguration.java +++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/impl/BlockCacheConfiguration.java @@ -23,6 +23,7 @@ import java.util.Map; import org.apache.accumulo.core.conf.AccumuloConfiguration; +import org.apache.accumulo.core.conf.ConfigurationTypeHelper; import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.core.spi.cache.BlockCacheManager.Configuration; import org.apache.accumulo.core.spi.cache.CacheType; @@ -41,22 +42,24 @@ public class BlockCacheConfiguration implements Configuration { private final long dataMaxSize; private final long summaryMaxSize; + private final long compressedDataMaxSize; public static BlockCacheConfiguration forTabletServer(AccumuloConfiguration conf) { return new BlockCacheConfiguration(conf, Property.TSERV_PREFIX, Property.TSERV_INDEXCACHE_SIZE, Property.TSERV_DATACACHE_SIZE, Property.TSERV_SUMMARYCACHE_SIZE, - Property.TSERV_DEFAULT_BLOCKSIZE); + Property.TSERV_DEFAULT_BLOCKSIZE, Property.TSERV_COMPRESSED_DATACACHE_SIZE); } public static BlockCacheConfiguration forScanServer(AccumuloConfiguration conf) { return new BlockCacheConfiguration(conf, Property.SSERV_PREFIX, Property.SSERV_INDEXCACHE_SIZE, Property.SSERV_DATACACHE_SIZE, Property.SSERV_SUMMARYCACHE_SIZE, - Property.SSERV_DEFAULT_BLOCKSIZE); + Property.SSERV_DEFAULT_BLOCKSIZE, Property.SSERV_COMPRESSED_DATACACHE_SIZE); } private BlockCacheConfiguration(AccumuloConfiguration conf, Property serverPrefix, Property indexCacheSizeProperty, Property dataCacheSizeProperty, - Property summaryCacheSizeProperty, Property defaultBlockSizeProperty) { + Property summaryCacheSizeProperty, Property defaultBlockSizeProperty, + Property compressedDataCacheSizeProperty) { this.serverPrefix = serverPrefix; this.genProps = conf.getAllPropertiesWithPrefix(serverPrefix); @@ -64,6 +67,13 @@ private BlockCacheConfiguration(AccumuloConfiguration conf, Property serverPrefi this.dataMaxSize = conf.getAsBytes(dataCacheSizeProperty); this.summaryMaxSize = conf.getAsBytes(summaryCacheSizeProperty); this.blockSize = conf.getAsBytes(defaultBlockSizeProperty); + + String compressedSizeStr = conf.get(compressedDataCacheSizeProperty); + if (compressedSizeStr == null) { + compressedSizeStr = compressedDataCacheSizeProperty.getDefaultValue(); + } + + this.compressedDataMaxSize = ConfigurationTypeHelper.getMemoryAsBytes(compressedSizeStr); } @Override @@ -83,7 +93,8 @@ public long getBlockSize() { @Override public String toString() { return "indexMaxSize: " + indexMaxSize + "dataMaxSize: " + dataMaxSize + "summaryMaxSize: " - + summaryMaxSize + ", blockSize: " + getBlockSize(); + + summaryMaxSize + ", blockSize: " + getBlockSize() + ", compressedDataMaxSize: " + + compressedDataMaxSize; } @Override @@ -121,4 +132,8 @@ public static String getCachePropertyBase(Property serverPrefix) { return serverPrefix.getKey() + "cache.config."; } + public long getCompressedDataMaxSize() { + return compressedDataMaxSize; + } + } diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/impl/CompressedBlockCache.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/impl/CompressedBlockCache.java new file mode 100644 index 00000000000..d3680e7ff06 --- /dev/null +++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/impl/CompressedBlockCache.java @@ -0,0 +1,173 @@ +/* + * 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 + * + * https://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.accumulo.core.file.blockfile.cache.impl; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Map; +import java.util.function.Supplier; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; + +import org.apache.accumulo.core.spi.cache.BlockCache; +import org.apache.accumulo.core.spi.cache.CacheEntry; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class CompressedBlockCache implements BlockCache { + + private static final Logger log = LoggerFactory.getLogger(CompressedBlockCache.class); + private final BlockCache delegate; + + public CompressedBlockCache(BlockCache delegate) { + this.delegate = delegate; + } + + private static byte[] compress(byte[] uncompressed) { + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(uncompressed.length / 2 + 16); + GZIPOutputStream gzip = new GZIPOutputStream(baos)) { + gzip.write(uncompressed); + gzip.finish(); + return baos.toByteArray(); + } catch (IOException e) { + log.warn("Failed to compress block for cache storage", e); + return null; + } + } + + private final byte[] uncompress(byte[] compressed) { + try (ByteArrayInputStream bais = new ByteArrayInputStream(compressed); + GZIPInputStream gzip = new GZIPInputStream(bais); + ByteArrayOutputStream baos = new ByteArrayOutputStream(compressed.length * 3)) { + byte[] buffer = new byte[8192]; + int len; + while ((len = gzip.read(buffer)) != -1) { + baos.write(buffer, 0, len); + } + return baos.toByteArray(); + } catch (IOException e) { + log.warn("Failed to decompress block from cache", e); + return null; + } + } + + @Override + public CacheEntry cacheBlock(String blockName, byte[] buf) { + byte[] compressed = compress(buf); + if (compressed == null) { + // compression failed, skip caching + return null; + } + + if (log.isTraceEnabled()) { + log.trace("Caching block {} compressed: {} -> {} bytes (ratio {:.2f}", blockName, buf.length, + compressed.length, (double) buf.length / compressed.length); + } + CacheEntry entry = delegate.cacheBlock(blockName, compressed); + if (entry == null) { + return null; + } + return new DecompressingCacheEntry(entry, buf); + } + + @Override + public CacheEntry getBlock(String blockName) { + CacheEntry entry = delegate.getBlock(blockName); + if (entry == null) { + return null; + } + byte[] uncompressed = uncompress(entry.getBuffer()); + return new DecompressingCacheEntry(entry, uncompressed); + } + + @Override + public CacheEntry getBlock(String blockName, Loader loader) { + CacheEntry existing = delegate.getBlock(blockName); + if (existing != null) { + byte[] uncompressed = uncompress(existing.getBuffer()); + return new DecompressingCacheEntry(existing, uncompressed); + } + + Loader compressingLoader = new Loader() { + @Override + public Map getDependencies() { + return loader.getDependencies(); + } + + @Override + public byte[] load(int maxSize, Map dependencies) { + byte[] uncompressed = loader.load(maxSize, dependencies); + if (uncompressed == null) { + return null; + } + byte[] compressed = compress(uncompressed); + return compressed; + } + }; + + CacheEntry entry = delegate.getBlock(blockName, compressingLoader); + if (entry == null) { + return null; + } + byte[] uncompressed = uncompress(entry.getBuffer()); + return new DecompressingCacheEntry(entry, uncompressed); + } + + @Override + public long getMaxHeapSize() { + return 0; + } + + @Override + public long getMaxSize() { + return delegate.getMaxSize(); + } + + @Override + public Stats getStats() { + return delegate.getStats(); + } + + private static final class DecompressingCacheEntry implements CacheEntry { + + private final CacheEntry compressedEntry; + private final byte[] uncompressedBuffer; + + public DecompressingCacheEntry(CacheEntry compressedEntry, byte[] uncompressedBuffer) { + this.compressedEntry = compressedEntry; + this.uncompressedBuffer = uncompressedBuffer; + } + + @Override + public byte[] getBuffer() { + return uncompressedBuffer; + } + + @Override + public T getIndex(Supplier supplier) { + return null; + } + + @Override + public void indexWeightChanged() { + compressedEntry.indexWeightChanged(); + } + } +} diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/BasicCacheProvider.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/BasicCacheProvider.java index e8890640ade..c8a30fd9027 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/BasicCacheProvider.java +++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/BasicCacheProvider.java @@ -24,10 +24,17 @@ public class BasicCacheProvider implements CacheProvider { private final BlockCache indexCache; private final BlockCache dataCache; + private final BlockCache compressedDataCache; public BasicCacheProvider(BlockCache indexCache, BlockCache dataCache) { + this(indexCache, dataCache, null); + } + + public BasicCacheProvider(BlockCache indexCache, BlockCache dataCache, + BlockCache compressedDataCache) { this.indexCache = indexCache; this.dataCache = dataCache; + this.compressedDataCache = compressedDataCache; } @Override @@ -40,4 +47,8 @@ public BlockCache getIndexCache() { return indexCache; } + @Override + public BlockCache getCompressedDataCache() { + return compressedDataCache; + } } diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java index 39ee8efcf89..1c059cdab3b 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java +++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java @@ -456,6 +456,15 @@ public CachedBlockRead getMetaBlock(long offset, long compressedSize, long rawSi * It is intended that once the BlockRead object is returned to the caller, that the caller will * read the entire block and then call close on the BlockRead class. * + *

+ * Lookup order: + *

    + *
  1. Primary (uncompressed) data cache
    2. + *
  2. Secondary (compressed) data cache
    3. + *
  3. Disk
    4. + *
+ *

+ * * NOTE: In the case of multi-read threads: This method can do redundant work where an entry is * read from disk and other threads check the cache before it has been inserted. */ @@ -470,6 +479,21 @@ public CachedBlockRead getDataBlock(int blockIndex) throws IOException { } } + // Check secondary + BlockCache _cCache = cacheProvider.getCompressedDataCache(); + if (_cCache != null) { + String _lookup = this.cacheId + "O" + blockIndex; + CacheEntry ce = _cCache.getBlock(_lookup); + if (ce != null) { + // promote the block to the primary cache + byte[] uncompressed = ce.getBuffer(); + if (_dCache != null) { + _dCache.cacheBlock(_lookup, uncompressed); + } + return new CachedBlockRead(new SimpleCacheEntry(uncompressed), uncompressed); + } + } + BlockReader _currBlock = getBCFile().getDataBlock(blockIndex); incrementCacheBypass(CacheType.DATA); return new CachedBlockRead(_currBlock); @@ -487,6 +511,21 @@ public CachedBlockRead getDataBlock(long offset, long compressedSize, long rawSi } } + // Check secondary compressed cache + BlockCache _cCache = cacheProvider.getCompressedDataCache(); + if (_cCache != null) { + String _lookup = this.cacheId + "R" + offset; + CacheEntry ce = _cCache.getBlock(_lookup); + if (ce != null) { + // promote the block to the primary cache + byte[] uncompressed = ce.getBuffer(); + if (_dCache != null) { + _dCache.cacheBlock(_lookup, uncompressed); + } + return new CachedBlockRead(new SimpleCacheEntry(uncompressed), uncompressed); + } + } + BlockReader _currBlock = getBCFile().getDataBlock(offset, compressedSize, rawSize); incrementCacheBypass(CacheType.DATA); return new CachedBlockRead(_currBlock); @@ -524,6 +563,29 @@ public void setCacheProvider(CacheProvider cacheProvider) { } + private static final class SimpleCacheEntry implements CacheEntry { + private final byte[] buffer; + + SimpleCacheEntry(byte[] buffer) { + this.buffer = buffer; + } + + @Override + public byte[] getBuffer() { + return buffer; + } + + @Override + public T getIndex(Supplier supplier) { + return null; + } + + @Override + public void indexWeightChanged() { + // no-op not backed by a real cache slot + } + } + public static class CachedBlockRead extends DataInputStream { private static InputStream wrapForTrace(InputStream inputStream) { diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CacheProvider.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CacheProvider.java index 4a64f9f2a1b..e4cbd49f74e 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CacheProvider.java +++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CacheProvider.java @@ -26,4 +26,10 @@ public interface CacheProvider { BlockCache getDataCache(); BlockCache getIndexCache(); + + // Returns an optional secondary data cache that stores block in compressed for. When non-null + // blocks that miss the primary data cache will be looked up here before falling through to disk. + default BlockCache getCompressedDataCache() { + return null; + } } diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/ScanCacheProvider.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/ScanCacheProvider.java index 45a375aae10..50277e3cb69 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/ScanCacheProvider.java +++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/ScanCacheProvider.java @@ -30,8 +30,10 @@ public final class ScanCacheProvider implements CacheProvider { private final BlockCache indexCache; private final BlockCache dataCache; + private final BlockCache compressedDataCache; + public ScanCacheProvider(AccumuloConfiguration tableConfig, ScanDispatch dispatch, - BlockCache indexCache, BlockCache dataCache) { + BlockCache indexCache, BlockCache dataCache, BlockCache compressedDataCache) { var loggingIndexCache = BlockCacheUtil.instrument(CacheType.INDEX, indexCache); var loggingDataCache = BlockCacheUtil.instrument(CacheType.DATA, dataCache); @@ -57,16 +59,20 @@ public ScanCacheProvider(AccumuloConfiguration tableConfig, ScanDispatch dispatc switch (dispatch.getDataCacheUsage()) { case ENABLED: this.dataCache = loggingDataCache; + this.compressedDataCache = compressedDataCache; break; case DISABLED: this.dataCache = null; + this.compressedDataCache = null; break; case OPPORTUNISTIC: this.dataCache = new OpportunisticBlockCache(loggingDataCache); + this.compressedDataCache = null; break; case TABLE: - this.dataCache = - tableConfig.getBoolean(Property.TABLE_BLOCKCACHE_ENABLED) ? loggingDataCache : null; + boolean tableDataCacheEnabled = tableConfig.getBoolean(Property.TABLE_BLOCKCACHE_ENABLED); + this.dataCache = tableDataCacheEnabled ? loggingDataCache : null; + this.compressedDataCache = tableDataCacheEnabled ? compressedDataCache : null; break; default: throw new IllegalStateException(); @@ -82,4 +88,9 @@ public BlockCache getDataCache() { public BlockCache getIndexCache() { return indexCache; } + + @Override + public BlockCache getCompressedDataCache() { + return compressedDataCache; + } } diff --git a/core/src/test/java/org/apache/accumulo/core/file/blockfile/cache/CompressedBlockCacheTest.java b/core/src/test/java/org/apache/accumulo/core/file/blockfile/cache/CompressedBlockCacheTest.java new file mode 100644 index 00000000000..6fdd75bda56 --- /dev/null +++ b/core/src/test/java/org/apache/accumulo/core/file/blockfile/cache/CompressedBlockCacheTest.java @@ -0,0 +1,253 @@ +/* + * 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 + * + * https://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.accumulo.core.file.blockfile.cache; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Supplier; + +import org.apache.accumulo.core.file.blockfile.cache.impl.CompressedBlockCache; +import org.apache.accumulo.core.spi.cache.BlockCache; +import org.apache.accumulo.core.spi.cache.CacheEntry; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class CompressedBlockCacheTest { + private static class MapBlockCache implements BlockCache { + private final ConcurrentHashMap map = new ConcurrentHashMap<>(); + + @Override + public CacheEntry cacheBlock(String blockName, byte[] buf) { + map.put(blockName, buf.clone()); + return wrap(buf); + } + + @Override + public CacheEntry getBlock(String blockName) { + byte[] data = map.get(blockName); + return data == null ? null : wrap(data); + } + + @Override + public CacheEntry getBlock(String blockName, Loader loader) { + byte[] data = map.computeIfAbsent(blockName, k -> { + byte[] loaded = loader.load(Integer.MAX_VALUE, Collections.emptyMap()); + return loaded; + }); + return data == null ? null : wrap(data); + } + + private CacheEntry wrap(byte[] data) { + return new CacheEntry() { + @Override + public byte[] getBuffer() { + return data; + } + + @Override + public T getIndex(Supplier supplier) { + return null; + } + + @Override + public void indexWeightChanged() {} + }; + } + + @Override + public long getMaxHeapSize() { + return Long.MAX_VALUE; + } + + @Override + public long getMaxSize() { + return Long.MAX_VALUE; + } + + @Override + public Stats getStats() { + return new Stats() { + @Override + public long hitCount() { + return 0; + } + + @Override + public long requestCount() { + return 0; + } + + @Override + public long evictionCount() { + return 0; + } + }; + } + + int size() { + return map.size(); + } + + byte[] getRaw(String key) { + return map.get(key); + } + } + + private MapBlockCache underlying; + private CompressedBlockCache cache; + + @BeforeEach + public void setUp() { + underlying = new MapBlockCache(); + cache = new CompressedBlockCache(underlying); + } + + @Test + public void testCacheAndRetrieve() { + // Store a block with highly compressible data (lots of repeated bytes) + byte[] data = new byte[1000]; + Arrays.fill(data, (byte) 'A'); + + CacheEntry stored = cache.cacheBlock("block1", data); + assertNotNull(stored); + // The returned entry should present the original uncompressed data + assertArrayEquals(data, stored.getBuffer()); + + // The underlying delegate should have stored compressed bytes (smaller than original) + byte[] compressed = underlying.getRaw("block1"); + assertNotNull(compressed); + // Verify compression actually reduced the size for highly repetitive data + assert compressed.length < data.length + : "Expected compressed size < original for repetitive data"; + } + + @Test + public void testRetrieveMiss() { + CacheEntry ce = cache.getBlock("nonexistent"); + assertNull(ce); + } + + @Test + public void testRetrieveHit() { + byte[] data = new byte[500]; + Arrays.fill(data, (byte) 'Z'); + + cache.cacheBlock("blockZ", data); + + CacheEntry ce = cache.getBlock("blockZ"); + assertNotNull(ce); + assertArrayEquals(data, ce.getBuffer()); + } + + @Test + public void testGetBlockWithLoader() { + byte[] data = new byte[800]; + Arrays.fill(data, (byte) 0x42); + + BlockCache.Loader loader = new BlockCache.Loader() { + @Override + public Map getDependencies() { + return Collections.emptyMap(); + } + + @Override + public byte[] load(int maxSize, Map dependencies) { + return data.clone(); + } + }; + + // First call: cache miss → loader invoked → compressed and stored + CacheEntry ce = cache.getBlock("blockLoader", loader); + assertNotNull(ce); + assertArrayEquals(data, ce.getBuffer()); + assert underlying.size() == 1 : "Expected one entry in underlying cache"; + + // Second call: cache hit → decompressed from stored entry + CacheEntry ce2 = cache.getBlock("blockLoader", loader); + assertNotNull(ce2); + assertArrayEquals(data, ce2.getBuffer()); + } + + @Test + public void testGetIndexReturnsNull() { + byte[] data = new byte[100]; + cache.cacheBlock("blockIdx", data); + + CacheEntry ce = cache.getBlock("blockIdx"); + assertNotNull(ce); + // Index is not supported for secondary compressed cache entries + assertNull(ce.getIndex(() -> null)); + } + + @Test + public void testLoaderNullReturnIsHandled() { + BlockCache.Loader nullLoader = new BlockCache.Loader() { + @Override + public Map getDependencies() { + return Collections.emptyMap(); + } + + @Override + public byte[] load(int maxSize, Map dependencies) { + return null; // block too large or otherwise unavailable + } + }; + + CacheEntry ce = cache.getBlock("blockNull", nullLoader); + assertNull(ce); + assert underlying.size() == 0 : "No entries should be stored when loader returns null"; + } + + @Test + public void testConcurrentAccess() throws InterruptedException { + final int THREADS = 20; + final int OPS_PER_THREAD = 50; + byte[] data = new byte[256]; + Arrays.fill(data, (byte) 0xFF); + + Thread[] threads = new Thread[THREADS]; + for (int i = 0; i < THREADS; i++) { + final String key = "block-" + (i % 5); // 5 distinct keys → lots of contention + threads[i] = new Thread(() -> { + for (int j = 0; j < OPS_PER_THREAD; j++) { + cache.cacheBlock(key, data); + CacheEntry ce = cache.getBlock(key); + if (ce != null) { + assertArrayEquals(data, ce.getBuffer()); + } + } + }); + } + + for (Thread t : threads) { + t.start(); + } + for (Thread t : threads) { + t.join(); + } + // If we reach here without exceptions, thread safety is satisfied + } +} diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java index e6779ada309..b1cb6167c85 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java @@ -69,7 +69,10 @@ import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.core.data.TableId; import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.file.blockfile.cache.impl.BlockCacheConfiguration; import org.apache.accumulo.core.file.blockfile.cache.impl.BlockCacheManagerFactory; +import org.apache.accumulo.core.file.blockfile.cache.impl.CompressedBlockCache; +import org.apache.accumulo.core.file.blockfile.cache.tinylfu.TinyLfuBlockCache; import org.apache.accumulo.core.file.blockfile.impl.ScanCacheProvider; import org.apache.accumulo.core.metadata.schema.Ample; import org.apache.accumulo.core.spi.cache.BlockCache; @@ -139,6 +142,7 @@ public class TabletServerResourceManager { private final BlockCache _dCache; private final BlockCache _iCache; private final BlockCache _sCache; + private final BlockCache _compressedDCache; private final ServerContext context; private Cache fileLenCache; @@ -286,32 +290,68 @@ public TabletServerResourceManager(ServerContext context, TabletHostingServer ts throw new IllegalStateException("Error creating BlockCacheManager", e); } - cacheManager.start(tserver.getBlockCacheConfiguration(acuConf)); + BlockCacheConfiguration blockCacheConf = + (BlockCacheConfiguration) tserver.getBlockCacheConfiguration(acuConf); + cacheManager.start(blockCacheConf); _iCache = cacheManager.getBlockCache(CacheType.INDEX); _dCache = cacheManager.getBlockCache(CacheType.DATA); _sCache = cacheManager.getBlockCache(CacheType.SUMMARY); + // Build the optional secondary compressed data cache. The BlockCacheManager only manages + // INDEX, DATA, and SUMMARY — it has no concept of a compressed secondary cache — so we + // construct it directly here using the same TinyLfuBlockCache implementation, then wrap + // it in CompressedBlockCache so blocks are stored at compressed size. + long compressedCacheSize = blockCacheConf.getCompressedDataMaxSize(); + if (compressedCacheSize > 0) { + // Reuse the same block-size hint that the primary data cache uses. + BlockCacheManager.Configuration compressedConf = new BlockCacheManager.Configuration() { + @Override + public long getMaxSize(CacheType type) { + return compressedCacheSize; + } + + @Override + public long getBlockSize() { + return blockCacheConf.getBlockSize(); + } + + @Override + public Map getProperties(String prefix, CacheType type) { + return Collections.emptyMap(); + } + }; + BlockCache underlying = new TinyLfuBlockCache(compressedConf, CacheType.DATA); + _compressedDCache = new CompressedBlockCache(underlying); + log.info("Secondary compressed data cache enabled, max size: {} bytes", compressedCacheSize); + } else { + _compressedDCache = null; + } + long dCacheSize = _dCache == null ? 0 : _dCache.getMaxHeapSize(); long iCacheSize = _iCache == null ? 0 : _iCache.getMaxHeapSize(); long sCacheSize = _sCache == null ? 0 : _sCache.getMaxHeapSize(); + long compressedCacheHeapSize = _compressedDCache == null ? 0 : compressedCacheSize; Runtime runtime = Runtime.getRuntime(); if (usingNativeMap) { // Still check block cache sizes when using native maps. - if (dCacheSize + iCacheSize + sCacheSize + totalQueueSize > runtime.maxMemory()) { + if (dCacheSize + iCacheSize + sCacheSize + compressedCacheHeapSize + totalQueueSize + > runtime.maxMemory()) { throw new IllegalArgumentException(String.format( "Block cache sizes %,d and mutation queue size %,d is too large for this JVM" + " configuration %,d", - dCacheSize + iCacheSize + sCacheSize, totalQueueSize, runtime.maxMemory())); + dCacheSize + iCacheSize + sCacheSize, compressedCacheHeapSize, totalQueueSize, + runtime.maxMemory())); } - } else if (maxMemory + dCacheSize + iCacheSize + sCacheSize + totalQueueSize - > runtime.maxMemory()) { + } else if (maxMemory + dCacheSize + iCacheSize + sCacheSize + compressedCacheHeapSize + + totalQueueSize > runtime.maxMemory()) { throw new IllegalArgumentException(String.format( "Maximum tablet server" + " map memory %,d block cache sizes %,d and mutation queue size %,d is" + " too large for this JVM configuration %,d", - maxMemory, dCacheSize + iCacheSize + sCacheSize, totalQueueSize, runtime.maxMemory())); + maxMemory, dCacheSize + iCacheSize + sCacheSize, compressedCacheHeapSize, totalQueueSize, + runtime.maxMemory())); } runtime.gc(); @@ -708,7 +748,7 @@ public synchronized ScanFileManager newScanFileManager(ScanDispatch scanDispatch } return fileManager.newScanFileManager(extent, - new ScanCacheProvider(tableConf, scanDispatch, _iCache, _dCache)); + new ScanCacheProvider(tableConf, scanDispatch, _iCache, _dCache, _compressedDCache)); } // END methods that Tablets call to manage their set of open data files @@ -885,6 +925,10 @@ public BlockCache getSummaryCache() { return _sCache; } + public BlockCache getCompressedDataCache() { + return _compressedDCache; + } + public Cache getFileLenCache() { return fileLenCache; } From 42f2212bd2152c30249f470d3f2c3193362e200a Mon Sep 17 00:00:00 2001 From: arbaazkhan1 Date: Fri, 14 Aug 2026 12:33:03 -0400 Subject: [PATCH 2/2] fixed comments --- .../blockfile/cache/CompressedBlockCacheTest.java | 11 ++--------- .../accumulo/tserver/TabletServerResourceManager.java | 6 +----- 2 files changed, 3 insertions(+), 14 deletions(-) diff --git a/core/src/test/java/org/apache/accumulo/core/file/blockfile/cache/CompressedBlockCacheTest.java b/core/src/test/java/org/apache/accumulo/core/file/blockfile/cache/CompressedBlockCacheTest.java index 6fdd75bda56..e60e482c101 100644 --- a/core/src/test/java/org/apache/accumulo/core/file/blockfile/cache/CompressedBlockCacheTest.java +++ b/core/src/test/java/org/apache/accumulo/core/file/blockfile/cache/CompressedBlockCacheTest.java @@ -127,19 +127,15 @@ public void setUp() { @Test public void testCacheAndRetrieve() { - // Store a block with highly compressible data (lots of repeated bytes) byte[] data = new byte[1000]; Arrays.fill(data, (byte) 'A'); CacheEntry stored = cache.cacheBlock("block1", data); assertNotNull(stored); - // The returned entry should present the original uncompressed data assertArrayEquals(data, stored.getBuffer()); - // The underlying delegate should have stored compressed bytes (smaller than original) byte[] compressed = underlying.getRaw("block1"); assertNotNull(compressed); - // Verify compression actually reduced the size for highly repetitive data assert compressed.length < data.length : "Expected compressed size < original for repetitive data"; } @@ -179,13 +175,11 @@ public byte[] load(int maxSize, Map dependencies) { } }; - // First call: cache miss → loader invoked → compressed and stored CacheEntry ce = cache.getBlock("blockLoader", loader); assertNotNull(ce); assertArrayEquals(data, ce.getBuffer()); assert underlying.size() == 1 : "Expected one entry in underlying cache"; - // Second call: cache hit → decompressed from stored entry CacheEntry ce2 = cache.getBlock("blockLoader", loader); assertNotNull(ce2); assertArrayEquals(data, ce2.getBuffer()); @@ -198,7 +192,6 @@ public void testGetIndexReturnsNull() { CacheEntry ce = cache.getBlock("blockIdx"); assertNotNull(ce); - // Index is not supported for secondary compressed cache entries assertNull(ce.getIndex(() -> null)); } @@ -212,7 +205,7 @@ public Map getDependencies() { @Override public byte[] load(int maxSize, Map dependencies) { - return null; // block too large or otherwise unavailable + return null; } }; @@ -230,7 +223,7 @@ public void testConcurrentAccess() throws InterruptedException { Thread[] threads = new Thread[THREADS]; for (int i = 0; i < THREADS; i++) { - final String key = "block-" + (i % 5); // 5 distinct keys → lots of contention + final String key = "block-" + (i % 5); threads[i] = new Thread(() -> { for (int j = 0; j < OPS_PER_THREAD; j++) { cache.cacheBlock(key, data); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java index b1cb6167c85..eed6345f1ee 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java @@ -298,13 +298,9 @@ public TabletServerResourceManager(ServerContext context, TabletHostingServer ts _dCache = cacheManager.getBlockCache(CacheType.DATA); _sCache = cacheManager.getBlockCache(CacheType.SUMMARY); - // Build the optional secondary compressed data cache. The BlockCacheManager only manages - // INDEX, DATA, and SUMMARY — it has no concept of a compressed secondary cache — so we - // construct it directly here using the same TinyLfuBlockCache implementation, then wrap - // it in CompressedBlockCache so blocks are stored at compressed size. + // Build the optional secondary compressed data cache. long compressedCacheSize = blockCacheConf.getCompressedDataMaxSize(); if (compressedCacheSize > 0) { - // Reuse the same block-size hint that the primary data cache uses. BlockCacheManager.Configuration compressedConf = new BlockCacheManager.Configuration() { @Override public long getMaxSize(CacheType type) {