New option to cache data

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,

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,29 +42,38 @@ 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);
     this.indexMaxSize = conf.getAsBytes(indexCacheSizeProperty);
     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;
+  }
+
 }

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<String,Loader> getDependencies() {
+        return loader.getDependencies();
+      }
+
+      @Override
+      public byte[] load(int maxSize, Map<String,byte[]> 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 extends Weighable> T getIndex(Supplier<T> supplier) {
+      return null;
+    }
+
+    @Override
+    public void indexWeightChanged() {
+      compressedEntry.indexWeightChanged();
+    }
+  }
+}

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;
+  }
 }