wchevreuil commented on code in PR #7291: URL: https://github.com/apache/hbase/pull/7291#discussion_r2340919078
########## hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RowCacheService.java: ########## @@ -0,0 +1,361 @@ +/* + * 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.hadoop.hbase.regionserver; + +import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_ACTIVATE_MIN_HFILES_DEFAULT; +import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_ACTIVATE_MIN_HFILES_KEY; + +import java.io.IOException; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellScanner; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.CheckAndMutate; +import org.apache.hadoop.hbase.client.CheckAndMutateResult; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Consistency; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Mutation; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.io.hfile.RowCacheKey; +import org.apache.hadoop.hbase.ipc.RpcCallContext; +import org.apache.hadoop.hbase.quotas.ActivePolicyEnforcement; +import org.apache.hadoop.hbase.quotas.OperationQuota; + +import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException; + +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileRequest; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileResponse; + +/** + * It is responsible for populating the row cache and retrieving rows from it. + */ +class RowCacheService { + /** + * A barrier that prevents the row cache from being populated during table operations, such as + * bulk loads. It is implemented as a counter to address issues that arise when the same table is + * updated concurrently. + */ + private final Map<TableName, AtomicInteger> tableLevelBarrierMap = new ConcurrentHashMap<>(); + /** + * A barrier that prevents the row cache from being populated during row mutations. It is + * implemented as a counter to address issues that arise when the same row is mutated + * concurrently. + */ + private final Map<RowCacheKey, AtomicInteger> rowLevelBarrierMap = new ConcurrentHashMap<>(); + private int activateMinHFiles; + + @FunctionalInterface + interface RowOperation<R> { + R execute() throws IOException; + } + + RowCacheService(Configuration conf) { + updateConf(conf); + } + + synchronized void updateConf(Configuration conf) { + this.activateMinHFiles = + conf.getInt(ROW_CACHE_ACTIVATE_MIN_HFILES_KEY, ROW_CACHE_ACTIVATE_MIN_HFILES_DEFAULT); + } + + RegionScannerImpl getScanner(HRegion region, Get get, Scan scan, List<Cell> results) + throws IOException { + if (!canCacheRow(get, region)) { + return getScannerInternal(region, scan, results); + } + + RowCacheKey key = new RowCacheKey(region, get.getRow()); + + // Try get from row cache + if (tryGetFromCache(region, key, get, results)) { + // Cache is hit, and then no scanner is created + return null; + } + + RegionScannerImpl scanner = getScannerInternal(region, scan, results); + + // The row cache is ineffective when the number of store files is small. If the number + // of store files falls below the minimum threshold, rows will not be cached + if (hasSufficientHFiles(region)) { + populateCache(region, results, key); + } + + return scanner; + } + + private RegionScannerImpl getScannerInternal(HRegion region, Scan scan, List<Cell> results) + throws IOException { + RegionScannerImpl scanner = region.getScanner(scan); + scanner.next(results); + return scanner; + } + + private boolean tryGetFromCache(HRegion region, RowCacheKey key, Get get, List<Cell> results) { + RowCells row = + (RowCells) region.getBlockCache().getBlock(key, get.getCacheBlocks(), false, true); Review Comment: RowCacheKey uses the region encoded name for indexing, whilst BlockCacheKey uses (store file name + offset). If the given row is already cached in a L2 cache block, this call will fail to fetch it and we'll cache it on the L1 too. ########## hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RowCacheService.java: ########## @@ -0,0 +1,361 @@ +/* + * 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.hadoop.hbase.regionserver; + +import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_ACTIVATE_MIN_HFILES_DEFAULT; +import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_ACTIVATE_MIN_HFILES_KEY; + +import java.io.IOException; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellScanner; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.CheckAndMutate; +import org.apache.hadoop.hbase.client.CheckAndMutateResult; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Consistency; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Mutation; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.io.hfile.RowCacheKey; +import org.apache.hadoop.hbase.ipc.RpcCallContext; +import org.apache.hadoop.hbase.quotas.ActivePolicyEnforcement; +import org.apache.hadoop.hbase.quotas.OperationQuota; + +import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException; + +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileRequest; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileResponse; + +/** + * It is responsible for populating the row cache and retrieving rows from it. + */ +class RowCacheService { + /** + * A barrier that prevents the row cache from being populated during table operations, such as + * bulk loads. It is implemented as a counter to address issues that arise when the same table is + * updated concurrently. + */ + private final Map<TableName, AtomicInteger> tableLevelBarrierMap = new ConcurrentHashMap<>(); + /** + * A barrier that prevents the row cache from being populated during row mutations. It is + * implemented as a counter to address issues that arise when the same row is mutated + * concurrently. + */ + private final Map<RowCacheKey, AtomicInteger> rowLevelBarrierMap = new ConcurrentHashMap<>(); + private int activateMinHFiles; + + @FunctionalInterface + interface RowOperation<R> { + R execute() throws IOException; + } + + RowCacheService(Configuration conf) { + updateConf(conf); + } + + synchronized void updateConf(Configuration conf) { + this.activateMinHFiles = + conf.getInt(ROW_CACHE_ACTIVATE_MIN_HFILES_KEY, ROW_CACHE_ACTIVATE_MIN_HFILES_DEFAULT); + } + + RegionScannerImpl getScanner(HRegion region, Get get, Scan scan, List<Cell> results) + throws IOException { + if (!canCacheRow(get, region)) { + return getScannerInternal(region, scan, results); + } + + RowCacheKey key = new RowCacheKey(region, get.getRow()); + + // Try get from row cache + if (tryGetFromCache(region, key, get, results)) { + // Cache is hit, and then no scanner is created + return null; + } + + RegionScannerImpl scanner = getScannerInternal(region, scan, results); + + // The row cache is ineffective when the number of store files is small. If the number + // of store files falls below the minimum threshold, rows will not be cached + if (hasSufficientHFiles(region)) { + populateCache(region, results, key); + } + + return scanner; + } + + private RegionScannerImpl getScannerInternal(HRegion region, Scan scan, List<Cell> results) + throws IOException { + RegionScannerImpl scanner = region.getScanner(scan); + scanner.next(results); Review Comment: Here, if the row we are reading is not in the memstore anymore, but rather on a store file, we are likely to cache the block containing this row once we read it from the store file in the `HFileReaderImpl.read()` call. ########## hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RowCacheService.java: ########## @@ -0,0 +1,361 @@ +/* + * 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.hadoop.hbase.regionserver; + +import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_ACTIVATE_MIN_HFILES_DEFAULT; +import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_ACTIVATE_MIN_HFILES_KEY; + +import java.io.IOException; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellScanner; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.CheckAndMutate; +import org.apache.hadoop.hbase.client.CheckAndMutateResult; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Consistency; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Mutation; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.io.hfile.RowCacheKey; +import org.apache.hadoop.hbase.ipc.RpcCallContext; +import org.apache.hadoop.hbase.quotas.ActivePolicyEnforcement; +import org.apache.hadoop.hbase.quotas.OperationQuota; + +import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException; + +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileRequest; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileResponse; + +/** + * It is responsible for populating the row cache and retrieving rows from it. + */ +class RowCacheService { + /** + * A barrier that prevents the row cache from being populated during table operations, such as + * bulk loads. It is implemented as a counter to address issues that arise when the same table is + * updated concurrently. + */ + private final Map<TableName, AtomicInteger> tableLevelBarrierMap = new ConcurrentHashMap<>(); + /** + * A barrier that prevents the row cache from being populated during row mutations. It is + * implemented as a counter to address issues that arise when the same row is mutated + * concurrently. + */ + private final Map<RowCacheKey, AtomicInteger> rowLevelBarrierMap = new ConcurrentHashMap<>(); + private int activateMinHFiles; + + @FunctionalInterface + interface RowOperation<R> { + R execute() throws IOException; + } + + RowCacheService(Configuration conf) { + updateConf(conf); + } + + synchronized void updateConf(Configuration conf) { + this.activateMinHFiles = + conf.getInt(ROW_CACHE_ACTIVATE_MIN_HFILES_KEY, ROW_CACHE_ACTIVATE_MIN_HFILES_DEFAULT); + } + + RegionScannerImpl getScanner(HRegion region, Get get, Scan scan, List<Cell> results) + throws IOException { + if (!canCacheRow(get, region)) { + return getScannerInternal(region, scan, results); + } + + RowCacheKey key = new RowCacheKey(region, get.getRow()); + + // Try get from row cache + if (tryGetFromCache(region, key, get, results)) { + // Cache is hit, and then no scanner is created + return null; + } + + RegionScannerImpl scanner = getScannerInternal(region, scan, results); + + // The row cache is ineffective when the number of store files is small. If the number Review Comment: Can you elaborate more on this? Is it really a matter of number of files or total store file size? For a single CF table, where a given region, after major compaction, has a 10GB store file, wouldn't this be more efficient? ########## hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RowCacheService.java: ########## @@ -0,0 +1,361 @@ +/* + * 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.hadoop.hbase.regionserver; + +import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_ACTIVATE_MIN_HFILES_DEFAULT; +import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_ACTIVATE_MIN_HFILES_KEY; + +import java.io.IOException; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellScanner; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.CheckAndMutate; +import org.apache.hadoop.hbase.client.CheckAndMutateResult; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Consistency; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Mutation; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.io.hfile.RowCacheKey; +import org.apache.hadoop.hbase.ipc.RpcCallContext; +import org.apache.hadoop.hbase.quotas.ActivePolicyEnforcement; +import org.apache.hadoop.hbase.quotas.OperationQuota; + +import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException; + +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileRequest; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileResponse; + +/** + * It is responsible for populating the row cache and retrieving rows from it. + */ +class RowCacheService { + /** + * A barrier that prevents the row cache from being populated during table operations, such as + * bulk loads. It is implemented as a counter to address issues that arise when the same table is + * updated concurrently. + */ + private final Map<TableName, AtomicInteger> tableLevelBarrierMap = new ConcurrentHashMap<>(); + /** + * A barrier that prevents the row cache from being populated during row mutations. It is + * implemented as a counter to address issues that arise when the same row is mutated + * concurrently. + */ + private final Map<RowCacheKey, AtomicInteger> rowLevelBarrierMap = new ConcurrentHashMap<>(); + private int activateMinHFiles; + + @FunctionalInterface + interface RowOperation<R> { + R execute() throws IOException; + } + + RowCacheService(Configuration conf) { + updateConf(conf); + } + + synchronized void updateConf(Configuration conf) { + this.activateMinHFiles = + conf.getInt(ROW_CACHE_ACTIVATE_MIN_HFILES_KEY, ROW_CACHE_ACTIVATE_MIN_HFILES_DEFAULT); + } + + RegionScannerImpl getScanner(HRegion region, Get get, Scan scan, List<Cell> results) + throws IOException { + if (!canCacheRow(get, region)) { + return getScannerInternal(region, scan, results); + } + + RowCacheKey key = new RowCacheKey(region, get.getRow()); + + // Try get from row cache + if (tryGetFromCache(region, key, get, results)) { + // Cache is hit, and then no scanner is created + return null; + } + + RegionScannerImpl scanner = getScannerInternal(region, scan, results); + + // The row cache is ineffective when the number of store files is small. If the number + // of store files falls below the minimum threshold, rows will not be cached + if (hasSufficientHFiles(region)) { + populateCache(region, results, key); Review Comment: What if the row is still in the memstore cache? We'll be placing it on both caches? -- 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]
