| /* |
| * 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.phoenix.iterate; |
| |
| import static org.apache.phoenix.util.EncodedColumnsUtil.getMinMaxQualifiersFromScan; |
| import static org.apache.phoenix.util.ScanUtil.getDummyResult; |
| import static org.apache.phoenix.util.ScanUtil.getPageSizeMsForRegionScanner; |
| import static org.apache.phoenix.util.ScanUtil.isDummy; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.DataInputStream; |
| import java.io.IOException; |
| import java.sql.SQLException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Set; |
| |
| import org.apache.hadoop.hbase.Cell; |
| import org.apache.hadoop.hbase.CellUtil; |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.KeyValue; |
| import org.apache.hadoop.hbase.client.Result; |
| import org.apache.hadoop.hbase.client.Scan; |
| import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; |
| import org.apache.hadoop.hbase.io.ImmutableBytesWritable; |
| import org.apache.hadoop.hbase.regionserver.Region; |
| import org.apache.hadoop.hbase.regionserver.RegionScanner; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.hadoop.io.WritableUtils; |
| import org.apache.phoenix.cache.GlobalCache; |
| import org.apache.phoenix.cache.TenantCache; |
| import org.apache.phoenix.coprocessor.BaseRegionScanner; |
| import org.apache.phoenix.coprocessor.HashJoinRegionScanner; |
| import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants; |
| import org.apache.phoenix.coprocessorclient.MetaDataProtocol; |
| import org.apache.phoenix.execute.TupleProjector; |
| import org.apache.phoenix.expression.Expression; |
| import org.apache.phoenix.expression.KeyValueColumnExpression; |
| import org.apache.phoenix.expression.OrderByExpression; |
| import org.apache.phoenix.expression.SingleCellColumnExpression; |
| import org.apache.phoenix.expression.function.ArrayIndexFunction; |
| import org.apache.phoenix.expression.function.JsonQueryFunction; |
| import org.apache.phoenix.expression.function.JsonValueFunction; |
| import org.apache.phoenix.expression.function.ScalarFunction; |
| import org.apache.phoenix.hbase.index.covered.update.ColumnReference; |
| import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; |
| import org.apache.phoenix.hbase.index.util.VersionUtil; |
| import org.apache.phoenix.index.IndexMaintainer; |
| import org.apache.phoenix.join.HashJoinInfo; |
| import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil; |
| import org.apache.phoenix.memory.MemoryManager; |
| import org.apache.phoenix.query.QueryConstants; |
| import org.apache.phoenix.query.QueryServices; |
| import org.apache.phoenix.query.QueryServicesOptions; |
| import org.apache.phoenix.schema.KeyValueSchema; |
| import org.apache.phoenix.schema.PTable; |
| import org.apache.phoenix.schema.ValueBitSet; |
| import org.apache.phoenix.schema.tuple.ResultTuple; |
| import org.apache.phoenix.schema.tuple.Tuple; |
| import org.apache.phoenix.schema.types.PInteger; |
| import org.apache.phoenix.transaction.PhoenixTransactionContext; |
| import org.apache.phoenix.transaction.TransactionFactory; |
| import org.apache.phoenix.util.ByteUtil; |
| import org.apache.phoenix.util.ClientUtil; |
| import org.apache.phoenix.util.EncodedColumnsUtil; |
| import org.apache.phoenix.util.IndexUtil; |
| import org.apache.phoenix.util.ScanUtil; |
| |
| import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting; |
| import org.apache.phoenix.thirdparty.com.google.common.collect.Lists; |
| import org.apache.phoenix.thirdparty.com.google.common.collect.Sets; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| public class NonAggregateRegionScannerFactory extends RegionScannerFactory { |
| |
| private static final Logger LOGGER = |
| LoggerFactory.getLogger(NonAggregateRegionScannerFactory.class); |
| |
| public NonAggregateRegionScannerFactory(RegionCoprocessorEnvironment env) { |
| this.env = env; |
| } |
| |
| @Override |
| public RegionScanner getRegionScanner(final Scan scan, final RegionScanner s) throws Throwable { |
| ImmutableBytesWritable ptr = new ImmutableBytesWritable(); |
| int offset = 0; |
| if (ScanUtil.isLocalIndex(scan)) { |
| /* |
| * For local indexes, we need to set an offset on row key expressions to skip |
| * the region start key. |
| */ |
| Region region = getRegion(); |
| offset = region.getRegionInfo().getStartKey().length != 0 ? |
| region.getRegionInfo().getStartKey().length : |
| region.getRegionInfo().getEndKey().length; |
| ScanUtil.setRowKeyOffset(scan, offset); |
| } |
| byte[] scanOffsetBytes = scan.getAttribute(BaseScannerRegionObserverConstants.SCAN_OFFSET); |
| Integer scanOffset = null; |
| if (scanOffsetBytes != null) { |
| scanOffset = (Integer)PInteger.INSTANCE.toObject(scanOffsetBytes); |
| } |
| RegionScanner innerScanner = s; |
| PTable.QualifierEncodingScheme encodingScheme = EncodedColumnsUtil.getQualifierEncodingScheme(scan); |
| boolean useNewValueColumnQualifier = EncodedColumnsUtil.useNewValueColumnQualifier(scan); |
| |
| Set<KeyValueColumnExpression> serverParsedKVRefs = Sets.newHashSet(); |
| KeyValueSchema kvSchema = null; |
| ValueBitSet kvSchemaBitSet = null; |
| List<Expression> resultList = getServerParsedExpressions(scan, serverParsedKVRefs); |
| Expression[] serverParsedFuncRefs = resultList.toArray(new Expression[0]); |
| if (serverParsedFuncRefs != null && serverParsedFuncRefs.length > 0) { |
| KeyValueSchema.KeyValueSchemaBuilder |
| builder = |
| new KeyValueSchema.KeyValueSchemaBuilder(0); |
| for (Expression expression : serverParsedFuncRefs) { |
| builder.addField(expression); |
| } |
| kvSchema = builder.build(); |
| kvSchemaBitSet = ValueBitSet.newInstance(kvSchema); |
| } |
| TupleProjector tupleProjector = null; |
| Region dataRegion = null; |
| IndexMaintainer indexMaintainer = null; |
| byte[][] viewConstants = null; |
| PhoenixTransactionContext tx = null; |
| ColumnReference[] dataColumns = IndexUtil.deserializeDataTableColumnsToJoin(scan); |
| if (dataColumns != null || ScanUtil.isUncoveredGlobalIndex(scan)) { |
| if (dataColumns != null) { |
| tupleProjector = IndexUtil.getTupleProjector(scan, dataColumns); |
| } |
| dataRegion = env.getRegion(); |
| int clientVersion = ScanUtil.getClientVersion(scan); |
| List<IndexMaintainer> indexMaintainers = |
| IndexUtil.deSerializeIndexMaintainersFromScan(scan); |
| indexMaintainer = indexMaintainers.get(0); |
| viewConstants = IndexUtil.deserializeViewConstantsFromScan(scan); |
| byte[] txState = scan.getAttribute(BaseScannerRegionObserverConstants.TX_STATE); |
| tx = TransactionFactory.getTransactionContext(txState, clientVersion); |
| } |
| |
| final TupleProjector p = TupleProjector.deserializeProjectorFromScan(scan); |
| final HashJoinInfo j = HashJoinInfo.deserializeHashJoinFromScan(scan); |
| boolean useQualifierAsIndex = EncodedColumnsUtil.useQualifierAsIndex(getMinMaxQualifiersFromScan(scan)) |
| && scan.getAttribute(BaseScannerRegionObserverConstants.TOPN) != null; |
| // setting dataRegion in case of a non-coprocessor environment |
| if (dataRegion == null && |
| env.getConfiguration().get(PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY) != null) { |
| dataRegion = env.getRegion(); |
| } |
| innerScanner = |
| getWrappedScanner(env, innerScanner, serverParsedKVRefs, serverParsedFuncRefs, |
| offset, scan, dataColumns, tupleProjector, dataRegion, indexMaintainer, tx, |
| viewConstants, kvSchema, kvSchemaBitSet, j == null ? p : null, ptr, |
| useQualifierAsIndex); |
| |
| final ImmutableBytesPtr tenantId = ScanUtil.getTenantId(scan); |
| if (j != null) { |
| innerScanner = |
| new HashJoinRegionScanner(env, innerScanner, scan, serverParsedKVRefs, |
| serverParsedFuncRefs, p, j, tenantId, useQualifierAsIndex, |
| useNewValueColumnQualifier); |
| } |
| if (scanOffset != null) { |
| final boolean isIncompatibleClient = |
| ScanUtil.isIncompatibleClientForServerReturnValidRowKey(scan); |
| innerScanner = getOffsetScanner( |
| innerScanner, |
| new OffsetResultIterator( |
| new RegionScannerResultIterator( |
| innerScanner, |
| getMinMaxQualifiersFromScan(scan), |
| encodingScheme), |
| scanOffset, |
| getPageSizeMsForRegionScanner(scan), |
| isIncompatibleClient), |
| scan.getAttribute(QueryConstants.LAST_SCAN) != null, |
| isIncompatibleClient, |
| scan); |
| } |
| boolean spoolingEnabled = |
| env.getConfiguration().getBoolean( |
| QueryServices.SERVER_ORDERBY_SPOOLING_ENABLED_ATTRIB, |
| QueryServicesOptions.DEFAULT_SERVER_ORDERBY_SPOOLING_ENABLED); |
| long thresholdBytes = |
| env.getConfiguration() |
| .getLongBytes(QueryServices.SERVER_SPOOL_THRESHOLD_BYTES_ATTRIB, |
| QueryServicesOptions.DEFAULT_SERVER_SPOOL_THRESHOLD_BYTES); |
| final OrderedResultIterator iterator = |
| deserializeFromScan(scan, innerScanner, spoolingEnabled, thresholdBytes); |
| if (iterator == null) { |
| return innerScanner; |
| } |
| // TODO:the above wrapped scanner should be used here also |
| return getTopNScanner(env, innerScanner, iterator, tenantId); |
| } |
| |
| private List<Expression> getServerParsedExpressions(Scan scan, |
| Set<KeyValueColumnExpression> serverParsedKVRefs) { |
| Expression[] serverParsedArrayFuncRefs = null; |
| if (scan.getAttribute(BaseScannerRegionObserverConstants.SPECIFIC_ARRAY_INDEX) != null) { |
| serverParsedArrayFuncRefs = |
| deserializeServerParsedPositionalExpressionInfoFromScan(scan, |
| BaseScannerRegionObserverConstants.SPECIFIC_ARRAY_INDEX, serverParsedKVRefs); |
| } |
| List<Expression> resultList = new ArrayList<>(); |
| if (serverParsedArrayFuncRefs != null) { |
| Collections.addAll(resultList, serverParsedArrayFuncRefs); |
| } |
| Expression[] serverParsedJsonValueFuncRefs = null; |
| if (scan.getAttribute(BaseScannerRegionObserverConstants.JSON_VALUE_FUNCTION) != null) { |
| serverParsedJsonValueFuncRefs = |
| deserializeServerParsedPositionalExpressionInfoFromScan(scan, |
| BaseScannerRegionObserverConstants.JSON_VALUE_FUNCTION, serverParsedKVRefs); |
| } |
| if (serverParsedJsonValueFuncRefs != null) { |
| Collections.addAll(resultList, serverParsedJsonValueFuncRefs); |
| } |
| Expression[] serverParsedJsonQueryFuncRefs = null; |
| if (scan.getAttribute(BaseScannerRegionObserverConstants.JSON_QUERY_FUNCTION) != null) { |
| serverParsedJsonQueryFuncRefs = |
| deserializeServerParsedPositionalExpressionInfoFromScan(scan, |
| BaseScannerRegionObserverConstants.JSON_QUERY_FUNCTION, serverParsedKVRefs); |
| } |
| if (serverParsedJsonQueryFuncRefs != null) { |
| Collections.addAll(resultList, serverParsedJsonQueryFuncRefs); |
| } |
| return resultList; |
| } |
| |
| @VisibleForTesting |
| static OrderedResultIterator deserializeFromScan(Scan scan, RegionScanner s, |
| boolean spoolingEnabled, long thresholdBytes) { |
| byte[] topN = scan.getAttribute(BaseScannerRegionObserverConstants.TOPN); |
| if (topN == null) { |
| return null; |
| } |
| int clientVersion = ScanUtil.getClientVersion(scan); |
| // Client including and after 4.15 and 5.1 are not going to serialize thresholdBytes |
| // so we need to decode this only for older clients to not break wire compat |
| boolean shouldDecodeSpoolThreshold = |
| (scan.getAttribute(BaseScannerRegionObserverConstants.CLIENT_VERSION) == null) |
| || (VersionUtil.decodeMajorVersion(clientVersion) > 5) |
| || (VersionUtil.decodeMajorVersion(clientVersion) == 5 |
| && clientVersion < MetaDataProtocol.MIN_5_x_DISABLE_SERVER_SPOOL_THRESHOLD) |
| || (VersionUtil.decodeMajorVersion(clientVersion) == 4 |
| && clientVersion < MetaDataProtocol.MIN_4_x_DISABLE_SERVER_SPOOL_THRESHOLD); |
| ByteArrayInputStream stream = new ByteArrayInputStream(topN); // TODO: size? |
| try { |
| DataInputStream input = new DataInputStream(stream); |
| if (shouldDecodeSpoolThreshold) { |
| // Read off the scan but ignore, we won't honor client sent thresholdbytes, but the |
| // one set on server |
| WritableUtils.readVInt(input); |
| } |
| int limit = WritableUtils.readVInt(input); |
| int estimatedRowSize = WritableUtils.readVInt(input); |
| int size = WritableUtils.readVInt(input); |
| List<OrderByExpression> orderByExpressions = Lists.newArrayListWithExpectedSize(size); |
| for (int i = 0; i < size; i++) { |
| OrderByExpression orderByExpression = new OrderByExpression(); |
| orderByExpression.readFields(input); |
| orderByExpressions.add(orderByExpression); |
| } |
| PTable.QualifierEncodingScheme encodingScheme = |
| EncodedColumnsUtil.getQualifierEncodingScheme(scan); |
| ResultIterator inner = new RegionScannerResultIterator(s, |
| EncodedColumnsUtil.getMinMaxQualifiersFromScan(scan), encodingScheme); |
| return new OrderedResultIterator(inner, orderByExpressions, spoolingEnabled, |
| thresholdBytes, limit >= 0 ? limit : null, null, estimatedRowSize, |
| getPageSizeMsForRegionScanner(scan), scan, s.getRegionInfo()); |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } finally { |
| try { |
| stream.close(); |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| } |
| |
| private Expression[] deserializeServerParsedPositionalExpressionInfoFromScan(Scan scan, |
| String scanAttribute, Set<KeyValueColumnExpression> serverParsedKVRefs) { |
| byte[] specificArrayIdx = scan.getAttribute(scanAttribute); |
| if (specificArrayIdx == null) { |
| return null; |
| } |
| ByteArrayInputStream stream = new ByteArrayInputStream(specificArrayIdx); |
| try { |
| DataInputStream input = new DataInputStream(stream); |
| int kvRefSize = WritableUtils.readVInt(input); |
| for (int i = 0; i < kvRefSize; i++) { |
| PTable.ImmutableStorageScheme scheme = EncodedColumnsUtil.getImmutableStorageScheme(scan); |
| KeyValueColumnExpression kvExp = scheme != PTable.ImmutableStorageScheme.ONE_CELL_PER_COLUMN ? new SingleCellColumnExpression(scheme) |
| : new KeyValueColumnExpression(); |
| kvExp.readFields(input); |
| serverParsedKVRefs.add(kvExp); |
| } |
| int kvFuncSize = WritableUtils.readVInt(input); |
| Expression[] funcRefs = new Expression[kvFuncSize]; |
| for (int i = 0; i < kvFuncSize; i++) { |
| ScalarFunction func = null; |
| if (scanAttribute.equals(BaseScannerRegionObserverConstants.SPECIFIC_ARRAY_INDEX)) { |
| func = new ArrayIndexFunction(); |
| } else if (scanAttribute.equals(BaseScannerRegionObserverConstants.JSON_VALUE_FUNCTION)) { |
| func = new JsonValueFunction(); |
| } else if (scanAttribute.equals(BaseScannerRegionObserverConstants.JSON_QUERY_FUNCTION)) { |
| func = new JsonQueryFunction(); |
| } |
| if (func != null) { |
| func.readFields(input); |
| funcRefs[i] = func; |
| } |
| } |
| return funcRefs; |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } finally { |
| try { |
| stream.close(); |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| } |
| |
| private RegionScanner getOffsetScanner(final RegionScanner s, |
| final OffsetResultIterator iterator, |
| final boolean isLastScan, |
| final boolean incompatibleClient, |
| final Scan scan) |
| throws IOException { |
| final Tuple firstTuple; |
| final Region region = getRegion(); |
| region.startRegionOperation(); |
| final byte[] initStartRowKey = scan.getStartRow().length > 0 ? scan.getStartRow() : |
| (scan.isReversed() ? region.getRegionInfo().getEndKey() : |
| region.getRegionInfo().getStartKey()); |
| byte[] prevScanStartRowKey = |
| scan.getAttribute(BaseScannerRegionObserverConstants.SCAN_ACTUAL_START_ROW); |
| // If the region has moved after server has returned dummy or valid row to client, |
| // prevScanStartRowKey would be different from actual scan start rowkey. |
| // If the region moves after dummy was returned, we do not need to set row count to |
| // offset. However, if the region moves after valid row was returned, we do need to |
| // set row count to offset because we return valid row only after offset num of rows |
| // are skipped. |
| if (Bytes.compareTo(prevScanStartRowKey, initStartRowKey) != 0 && Bytes.compareTo( |
| ByteUtil.concat(prevScanStartRowKey, ByteUtil.ZERO_BYTE), |
| initStartRowKey) != 0) { |
| iterator.setRowCountToOffset(); |
| } |
| try { |
| Tuple tuple = iterator.next(); |
| if (tuple == null && !isLastScan) { |
| List<Cell> kvList = new ArrayList<>(1); |
| KeyValue kv; |
| byte[] remainingOffset = |
| PInteger.INSTANCE.toBytes(iterator.getRemainingOffset()); |
| if (incompatibleClient) { |
| kv = new KeyValue( |
| QueryConstants.OFFSET_ROW_KEY_BYTES, |
| QueryConstants.OFFSET_FAMILY, |
| QueryConstants.OFFSET_COLUMN, |
| remainingOffset); |
| } else { |
| Tuple lastScannedTuple = iterator.getLastScannedTuple(); |
| if (lastScannedTuple != null) { |
| kv = getOffsetKvWithLastScannedRowKey(remainingOffset, lastScannedTuple); |
| } else { |
| byte[] rowKey; |
| byte[] startKey = scan.getStartRow().length > 0 ? scan.getStartRow() : |
| region.getRegionInfo().getStartKey(); |
| byte[] endKey = scan.getStopRow().length > 0 ? scan.getStopRow() : |
| region.getRegionInfo().getEndKey(); |
| rowKey = ByteUtil.getLargestPossibleRowKeyInRange(startKey, endKey); |
| if (rowKey == null) { |
| if (scan.includeStartRow()) { |
| rowKey = startKey; |
| } else if (scan.includeStopRow()) { |
| rowKey = endKey; |
| } else { |
| rowKey = HConstants.EMPTY_END_ROW; |
| } |
| } |
| kv = new KeyValue( |
| rowKey, |
| QueryConstants.OFFSET_FAMILY, |
| QueryConstants.OFFSET_COLUMN, |
| remainingOffset); |
| } |
| } |
| kvList.add(kv); |
| Result r = Result.create(kvList); |
| firstTuple = new ResultTuple(r); |
| } else { |
| firstTuple = tuple; |
| } |
| } catch (Throwable t) { |
| ClientUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), t); |
| return null; |
| } finally { |
| region.closeRegionOperation(); |
| } |
| |
| return new BaseRegionScanner(s) { |
| private Tuple tuple = firstTuple; |
| private byte[] previousResultRowKey; |
| |
| @Override |
| public boolean isFilterDone() { |
| return tuple == null; |
| } |
| |
| @Override |
| public boolean next(List<Cell> results) throws IOException { |
| try { |
| if (isFilterDone()) { |
| return false; |
| } |
| Tuple nextTuple = iterator.next(); |
| if (tuple.size() > 0 && !isDummy(tuple)) { |
| for (int i = 0; i < tuple.size(); i++) { |
| results.add(tuple.getValue(i)); |
| if (i == 0) { |
| previousResultRowKey = CellUtil.cloneRow(tuple.getValue(i)); |
| } |
| } |
| } else { |
| if (nextTuple == null) { |
| byte[] remainingOffset = |
| PInteger.INSTANCE.toBytes(iterator.getRemainingOffset()); |
| KeyValue kv; |
| if (incompatibleClient) { |
| kv = new KeyValue( |
| QueryConstants.OFFSET_ROW_KEY_BYTES, |
| QueryConstants.OFFSET_FAMILY, |
| QueryConstants.OFFSET_COLUMN, |
| remainingOffset); |
| } else { |
| kv = getOffsetKvWithLastScannedRowKey(remainingOffset, tuple); |
| } |
| results.add(kv); |
| } else { |
| updateDummyWithPrevRowKey(results, initStartRowKey, |
| previousResultRowKey); |
| } |
| } |
| tuple = nextTuple; |
| return !isFilterDone(); |
| } catch (Throwable t) { |
| LOGGER.error("Error while iterating Offset scanner.", t); |
| ClientUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), t); |
| return false; |
| } |
| } |
| |
| @Override |
| public void close() throws IOException { |
| try { |
| s.close(); |
| } finally { |
| try { |
| iterator.close(); |
| } catch (SQLException e) { |
| ClientUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), e); |
| } |
| } |
| } |
| }; |
| } |
| |
| /** |
| * Add dummy cell to the result list based on either the previous rowkey returned to the |
| * client or the start rowkey of the scan or region start key. |
| * |
| * @param result result row. |
| * @param initStartRowKey scan start rowkey. |
| * @param previousResultRowKey previous result rowkey returned to client. |
| */ |
| private void updateDummyWithPrevRowKey(final List<Cell> result, |
| final byte[] initStartRowKey, |
| final byte[] previousResultRowKey) { |
| result.clear(); |
| if (previousResultRowKey != null) { |
| getDummyResult(previousResultRowKey, result); |
| } else { |
| getDummyResult(initStartRowKey, result); |
| } |
| } |
| |
| private static KeyValue getOffsetKvWithLastScannedRowKey(byte[] value, Tuple tuple) { |
| ImmutableBytesWritable ptr = new ImmutableBytesWritable(); |
| tuple.getKey(ptr); |
| byte[] rowKey = new byte[ptr.getLength()]; |
| System.arraycopy(ptr.get(), ptr.getOffset(), rowKey, 0, |
| rowKey.length); |
| return new KeyValue( |
| rowKey, |
| QueryConstants.OFFSET_FAMILY, |
| QueryConstants.OFFSET_COLUMN, |
| value); |
| } |
| |
| /** |
| * Return region scanner that does TopN. |
| * We only need to call startRegionOperation and closeRegionOperation when |
| * getting the first Tuple (which forces running through the entire region) |
| * since after this everything is held in memory |
| */ |
| private RegionScanner getTopNScanner(RegionCoprocessorEnvironment env, final RegionScanner s, |
| final OrderedResultIterator iterator, ImmutableBytesPtr tenantId) throws Throwable { |
| |
| final Tuple firstTuple; |
| TenantCache tenantCache = GlobalCache.getTenantCache(env, tenantId); |
| long estSize = iterator.getEstimatedByteSize(); |
| final MemoryManager.MemoryChunk chunk = tenantCache.getMemoryManager().allocate(estSize); |
| final Region region = getRegion(); |
| region.startRegionOperation(); |
| try { |
| // Once we return from the first call to next, we've run through and cached |
| // the topN rows, so we no longer need to start/stop a region operation. |
| firstTuple = iterator.next(); |
| // Now that the topN are cached, we can resize based on the real size |
| long actualSize = iterator.getByteSize(); |
| chunk.resize(actualSize); |
| } catch (Throwable t) { |
| ClientUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), t); |
| return null; |
| } finally { |
| region.closeRegionOperation(); |
| } |
| return new BaseRegionScanner(s) { |
| private Tuple tuple = firstTuple; |
| |
| @Override |
| public boolean isFilterDone() { |
| return tuple == null; |
| } |
| |
| @Override |
| public boolean next(List<Cell> results) throws IOException { |
| try { |
| if (isFilterDone()) { |
| return false; |
| } |
| if (isDummy(tuple)) { |
| ScanUtil.getDummyResult(CellUtil.cloneRow(tuple.getValue(0)), results); |
| } else { |
| for (int i = 0; i < tuple.size(); i++) { |
| results.add(tuple.getValue(i)); |
| } |
| } |
| tuple = iterator.next(); |
| return !isFilterDone(); |
| } catch (Throwable t) { |
| ClientUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), t); |
| return false; |
| } |
| } |
| |
| @Override |
| public void close() throws IOException { |
| try { |
| s.close(); |
| } finally { |
| try { |
| if(iterator != null) { |
| iterator.close(); |
| } |
| } catch (SQLException e) { |
| ClientUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), e); |
| } finally { |
| chunk.close(); |
| } |
| } |
| } |
| }; |
| } |
| } |
