Author: mbautin Date: Fri Jan 27 01:23:37 2012 New Revision: 1236483 URL: http://svn.apache.org/viewvc?rev=1236483&view=rev Log: [jira] [HBASE-5274] Filter out expired scanners on compaction as well
Summary: This is a followup for D1017 to make it similar to D909 (89-fb). The fix for 89-fb used the TTL-based scanner filtering logic on both normal scanners and compactions, while the trunk fix D1017 did not. This is just the delta between the two diffs that brings filtering expired store files on compaction to trunk. Test Plan: Unit tests Reviewers: Liyin, JIRA, lhofhansl, Kannan Reviewed By: Liyin CC: Liyin, tedyu, Kannan, mbautin, lhofhansl Differential Revision: https://reviews.facebook.net/D1473 Modified: hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java hbase/trunk/src/test/java/org/apache/hadoop/hbase/io/hfile/TestScannerSelectionUsingTTL.java hbase/trunk/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java Modified: hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java URL: http://svn.apache.org/viewvc/hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java?rev=1236483&r1=1236482&r2=1236483&view=diff ============================================================================== --- hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java (original) +++ hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java Fri Jan 27 01:23:37 2012 @@ -845,6 +845,11 @@ public class Store extends SchemaConfigu } } + /** + * Get all scanners with no filtering based on TTL (that happens further down + * the line). + * @return all scanners for this store + */ protected List<KeyValueScanner> getScanners(boolean cacheBlocks, boolean isGet, boolean isCompaction, @@ -964,10 +969,10 @@ public class Store extends SchemaConfigu + StringUtils.humanReadableInt(storeSize)); } - /* - * Compact the most recent N files. Essentially a hook for testing. + /** + * Compact the most recent N files. Used in testing. */ - protected void compactRecent(int N) throws IOException { + public void compactRecentForTesting(int N) throws IOException { List<StoreFile> filesToCompact; long maxId; boolean isMajor; @@ -1926,7 +1931,8 @@ public class Store extends SchemaConfigu ////////////////////////////////////////////////////////////////////////////// /** - * Return a scanner for both the memstore and the HStore files + * Return a scanner for both the memstore and the HStore files. Assumes we + * are not in a compaction. * @throws IOException */ public StoreScanner getScanner(Scan scan, Modified: hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java URL: http://svn.apache.org/viewvc/hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java?rev=1236483&r1=1236482&r2=1236483&view=diff ============================================================================== --- hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java (original) +++ hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java Fri Jan 27 01:23:37 2012 @@ -56,6 +56,8 @@ class StoreScanner extends NonLazyKeyVal private final boolean isGet; private final boolean explicitColumnQuery; private final boolean useRowColBloom; + private final Scan scan; + private final NavigableSet<byte[]> columns; private final long oldestUnexpiredTS; private final int minVersions; @@ -77,6 +79,8 @@ class StoreScanner extends NonLazyKeyVal isGet = scan.isGetScan(); int numCol = columns == null ? 0 : columns.size(); explicitColumnQuery = numCol > 0; + this.scan = scan; + this.columns = columns; oldestUnexpiredTS = EnvironmentEdgeManager.currentTimeMillis() - ttl; this.minVersions = minVersions; @@ -88,7 +92,8 @@ class StoreScanner extends NonLazyKeyVal } /** - * Opens a scanner across memstore, snapshot, and all StoreFiles. + * Opens a scanner across memstore, snapshot, and all StoreFiles. Assumes we + * are not in a compaction. * * @param store who we scan * @param scan the spec @@ -109,7 +114,7 @@ class StoreScanner extends NonLazyKeyVal oldestUnexpiredTS); // Pass columns to try to filter out unnecessary StoreFiles. - List<KeyValueScanner> scanners = getScanners(scan, columns); + List<KeyValueScanner> scanners = getScannersNoCompaction(); // Seek all scanners to the start of the Row (or if the exact matching row // key does not exist, then to the start of the next matching Row). @@ -150,6 +155,9 @@ class StoreScanner extends NonLazyKeyVal matcher = new ScanQueryMatcher(scan, store.scanInfo, null, scanType, smallestReadPoint, earliestPutTs, oldestUnexpiredTS); + // Filter the list of scanners using Bloom filters, time range, TTL, etc. + scanners = selectScannersFrom(scanners); + // Seek all scanners to the initial key for(KeyValueScanner scanner : scanners) { scanner.seek(matcher.getStartKey()); @@ -159,7 +167,7 @@ class StoreScanner extends NonLazyKeyVal heap = new KeyValueHeap(scanners, store.comparator); } - // Constructor for testing. + /** Constructor for testing. */ StoreScanner(final Scan scan, Store.ScanInfo scanInfo, StoreScanner.ScanType scanType, final NavigableSet<byte[]> columns, final List<KeyValueScanner> scanners) throws IOException { @@ -202,18 +210,22 @@ class StoreScanner extends NonLazyKeyVal tableName, family) + "getsize"; } - /* - * @return List of scanners ordered properly. + /** + * Get a filtered list of scanners. Assumes we are not in a compaction. + * @return list of scanners to seek */ - private List<KeyValueScanner> getScanners() throws IOException { - return this.store.getScanners(cacheBlocks, isGet, false, null); + private List<KeyValueScanner> getScannersNoCompaction() throws IOException { + final boolean isCompaction = false; + return selectScannersFrom(store.getScanners(cacheBlocks, isGet, + isCompaction, matcher)); } - /* - * @return List of scanners to seek, possibly filtered by StoreFile. + /** + * Filters the given list of scanners using Bloom filter, time range, and + * TTL. */ - private List<KeyValueScanner> getScanners(Scan scan, - final NavigableSet<byte[]> columns) throws IOException { + private List<KeyValueScanner> selectScannersFrom( + final List<? extends KeyValueScanner> allScanners) { boolean memOnly; boolean filesOnly; if (scan instanceof InternalScan) { @@ -224,11 +236,9 @@ class StoreScanner extends NonLazyKeyVal memOnly = false; filesOnly = false; } - List<KeyValueScanner> allStoreScanners = - this.store.getScanners(cacheBlocks, isGet, false, this.matcher); List<KeyValueScanner> scanners = - new ArrayList<KeyValueScanner>(allStoreScanners.size()); + new ArrayList<KeyValueScanner>(allScanners.size()); // We can only exclude store files based on TTL if minVersions is set to 0. // Otherwise, we might have to return KVs that have technically expired. @@ -236,7 +246,7 @@ class StoreScanner extends NonLazyKeyVal Long.MIN_VALUE; // include only those scan files which pass all filters - for (KeyValueScanner kvs : allStoreScanners) { + for (KeyValueScanner kvs : allScanners) { boolean isFile = kvs.isFileScanner(); if ((!isFile && filesOnly) || (isFile && memOnly)) { continue; @@ -246,7 +256,6 @@ class StoreScanner extends NonLazyKeyVal scanners.add(kvs); } } - return scanners; } @@ -281,7 +290,7 @@ class StoreScanner extends NonLazyKeyVal public synchronized boolean seek(KeyValue key) throws IOException { if (this.heap == null) { - List<KeyValueScanner> scanners = getScanners(); + List<KeyValueScanner> scanners = getScannersNoCompaction(); heap = new KeyValueHeap(scanners, store.comparator); } @@ -479,7 +488,7 @@ class StoreScanner extends NonLazyKeyVal /* When we have the scan object, should we not pass it to getScanners() * to get a limited set of scanners? We did so in the constructor and we * could have done it now by storing the scan object from the constructor */ - List<KeyValueScanner> scanners = getScanners(); + List<KeyValueScanner> scanners = getScannersNoCompaction(); for(KeyValueScanner scanner : scanners) { scanner.seek(lastTopKey); Modified: hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java URL: http://svn.apache.org/viewvc/hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java?rev=1236483&r1=1236482&r2=1236483&view=diff ============================================================================== --- hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java (original) +++ hbase/trunk/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java Fri Jan 27 01:23:37 2012 @@ -742,7 +742,7 @@ public class SchemaMetrics { return metricsSnapshot; } - private static long getLong(Map<String, Long> m, String k) { + public static long getLong(Map<String, Long> m, String k) { Long l = m.get(k); return l != null ? l : 0; } Modified: hbase/trunk/src/test/java/org/apache/hadoop/hbase/io/hfile/TestScannerSelectionUsingTTL.java URL: http://svn.apache.org/viewvc/hbase/trunk/src/test/java/org/apache/hadoop/hbase/io/hfile/TestScannerSelectionUsingTTL.java?rev=1236483&r1=1236482&r2=1236483&view=diff ============================================================================== --- hbase/trunk/src/test/java/org/apache/hadoop/hbase/io/hfile/TestScannerSelectionUsingTTL.java (original) +++ hbase/trunk/src/test/java/org/apache/hadoop/hbase/io/hfile/TestScannerSelectionUsingTTL.java Fri Jan 27 01:23:37 2012 @@ -20,9 +20,9 @@ import static org.junit.Assert.assertEqu import java.io.IOException; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; import java.util.List; +import java.util.Map; import java.util.Set; import org.apache.commons.logging.Log; @@ -35,9 +35,12 @@ import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.SmallTests; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.io.hfile.BlockType.BlockCategory; import org.apache.hadoop.hbase.regionserver.HRegion; import org.apache.hadoop.hbase.regionserver.InternalScanner; import org.apache.hadoop.hbase.regionserver.StoreFile.BloomType; +import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics; +import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics.BlockMetricType; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Threads; import org.junit.Test; @@ -70,19 +73,27 @@ public class TestScannerSelectionUsingTT private static final int NUM_ROWS = 8; private static final int NUM_COLS_PER_ROW = 5; - public final int numFreshFiles; + public final int numFreshFiles, totalNumFiles; + + /** Whether we are specifying the exact files to compact */ + private final boolean explicitCompaction; @Parameters - public static Collection<Object[]> parametersNumFreshFiles() { - return Arrays.asList(new Object[][]{ - new Object[] { new Integer(1) }, - new Object[] { new Integer(2) }, - new Object[] { new Integer(3) } - }); + public static Collection<Object[]> parameters() { + List<Object[]> params = new ArrayList<Object[]>(); + for (int numFreshFiles = 1; numFreshFiles <= 3; ++numFreshFiles) { + for (boolean explicitCompaction : new boolean[] { false, true }) { + params.add(new Object[] { numFreshFiles, explicitCompaction }); + } + } + return params; } - public TestScannerSelectionUsingTTL(int numFreshFiles) { + public TestScannerSelectionUsingTTL(int numFreshFiles, + boolean explicitCompaction) { this.numFreshFiles = numFreshFiles; + this.totalNumFiles = numFreshFiles + NUM_EXPIRED_FILES; + this.explicitCompaction = explicitCompaction; } @Test @@ -101,7 +112,7 @@ public class TestScannerSelectionUsingTT HRegion.createHRegion(info, TEST_UTIL.getClusterTestDir(), TEST_UTIL.getConfiguration(), htd); - for (int iFile = 0; iFile < NUM_EXPIRED_FILES + numFreshFiles; ++iFile) { + for (int iFile = 0; iFile < totalNumFiles; ++iFile) { if (iFile == NUM_EXPIRED_FILES) { Threads.sleepWithoutInterrupt(TTL_MS); } @@ -135,11 +146,28 @@ public class TestScannerSelectionUsingTT assertEquals(NUM_ROWS, numReturnedRows); Set<String> accessedFiles = cache.getCachedFileNamesForTest(); LOG.debug("Files accessed during scan: " + accessedFiles); - assertEquals("If " + (NUM_EXPIRED_FILES + numFreshFiles) + " files are " - + "accessed instead of " + numFreshFiles + ", we are " - + "not filtering expired files out.", numFreshFiles, - accessedFiles.size()); + Map<String, Long> metricsBeforeCompaction = + SchemaMetrics.getMetricsSnapshot(); + + // Exercise both compaction codepaths. + if (explicitCompaction) { + region.getStore(FAMILY_BYTES).compactRecentForTesting(totalNumFiles); + } else { + region.compactStores(); + } + + SchemaMetrics.validateMetricChanges(metricsBeforeCompaction); + Map<String, Long> compactionMetrics = + SchemaMetrics.diffMetrics(metricsBeforeCompaction, + SchemaMetrics.getMetricsSnapshot()); + long compactionDataBlocksRead = SchemaMetrics.getLong( + compactionMetrics, + SchemaMetrics.getInstance(TABLE, FAMILY).getBlockMetricName( + BlockCategory.DATA, true, BlockMetricType.READ_COUNT)); + assertEquals("Invalid number of blocks accessed during compaction. " + + "We only expect non-expired files to be accessed.", + numFreshFiles, compactionDataBlocksRead); region.close(); } Modified: hbase/trunk/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java URL: http://svn.apache.org/viewvc/hbase/trunk/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java?rev=1236483&r1=1236482&r2=1236483&view=diff ============================================================================== --- hbase/trunk/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java (original) +++ hbase/trunk/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java Fri Jan 27 01:23:37 2012 @@ -391,7 +391,7 @@ public class TestCompaction extends HBas Store store2 = this.r.stores.get(fam2); int numFiles1 = store2.getStorefiles().size(); assertTrue("Was expecting to see 4 store files", numFiles1 > compactionThreshold); // > 3 - store2.compactRecent(compactionThreshold); // = 3 + store2.compactRecentForTesting(compactionThreshold); // = 3 int numFiles2 = store2.getStorefiles().size(); // Check that we did compact assertTrue("Number of store files should go down", numFiles1 > numFiles2);
