| // 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.impala.planner; |
| |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.ListIterator; |
| import java.util.Map; |
| |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.HRegionLocation; |
| import org.apache.hadoop.hbase.filter.CompareFilter; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.impala.analysis.Analyzer; |
| import org.apache.impala.analysis.BinaryPredicate; |
| import org.apache.impala.analysis.Expr; |
| import org.apache.impala.analysis.LiteralExpr; |
| import org.apache.impala.analysis.SlotDescriptor; |
| import org.apache.impala.analysis.StringLiteral; |
| import org.apache.impala.analysis.TupleDescriptor; |
| import org.apache.impala.catalog.FeHBaseTable; |
| import org.apache.impala.catalog.FeTable; |
| import org.apache.impala.catalog.HBaseColumn; |
| import org.apache.impala.catalog.PrimitiveType; |
| import org.apache.impala.catalog.Type; |
| import org.apache.impala.common.ImpalaException; |
| import org.apache.impala.common.Pair; |
| import org.apache.impala.thrift.TExplainLevel; |
| import org.apache.impala.thrift.THBaseFilter; |
| import org.apache.impala.thrift.THBaseKeyRange; |
| import org.apache.impala.thrift.THBaseScanNode; |
| import org.apache.impala.thrift.TNetworkAddress; |
| import org.apache.impala.thrift.TPlanNode; |
| import org.apache.impala.thrift.TPlanNodeType; |
| import org.apache.impala.thrift.TQueryOptions; |
| import org.apache.impala.thrift.TScanRange; |
| import org.apache.impala.thrift.TScanRangeLocation; |
| import org.apache.impala.thrift.TScanRangeLocationList; |
| import org.apache.impala.thrift.TScanRangeSpec; |
| import org.apache.impala.util.BitUtil; |
| import org.apache.impala.util.ExecutorMembershipSnapshot; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.base.Objects; |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.Lists; |
| |
| /** |
| * Full scan of an HBase table. |
| * Only families/qualifiers specified in TupleDescriptor will be retrieved in the backend. |
| */ |
| public class HBaseScanNode extends ScanNode { |
| // The suggested value for "hbase.client.scan.setCaching", which batches maxCaching |
| // rows per fetch request to the HBase region server. If the value is too high, |
| // then the hbase region server will have a hard time (GC pressure and long response |
| // times). If the value is too small, then there will be extra trips to the hbase |
| // region server. |
| // Default to 1024 and update it based on row size estimate such that each batch size |
| // won't exceed 500MB. |
| private final static int MAX_HBASE_FETCH_BATCH_SIZE = 500 * 1024 * 1024; |
| private final static int DEFAULT_SUGGESTED_CACHING = 1024; |
| |
| // Used for memory estimation when the column max size stat is missing (happens only |
| // in case of string type columns). |
| private final static int DEFAULT_STRING_COL_BYTES = 32 * 1024; |
| |
| // Used for memory estimation to clamp the max estimate to 128 MB in case of |
| // missing stats. |
| private final static int DEFAULT_MAX_ESTIMATE_BYTES = 128 * 1024 * 1024; |
| |
| // Used for memory estimation to clamp the min estimate to 4 KB which is min |
| // block size that can be allocated by the mem-pool. |
| private final static int DEFAULT_MIN_ESTIMATE_BYTES = 4 * 1024; |
| |
| private final static Logger LOG = LoggerFactory.getLogger(HBaseScanNode.class); |
| private final TupleDescriptor desc_; |
| |
| // One range per clustering column. The range bounds are expected to be constants. |
| // A null entry means there's no range restriction for that particular key. |
| // If keyRanges is non-null it always contains as many entries as there are clustering |
| // cols. Don't replace this variable after init(). |
| private List<ValueRange> keyRanges_ = new ArrayList<>(); |
| |
| // The list of conjuncts used to create the key ranges. Used if we must estimate |
| // cardinality based on row count stats. Don't replace this variable after init(). |
| private List<Expr> keyConjuncts_ = new ArrayList<>(); |
| |
| // derived from keyRanges_; empty means unbounded; |
| // initialize start/stopKey_ to be unbounded. |
| private byte[] startKey_ = HConstants.EMPTY_START_ROW; |
| private byte[] stopKey_ = HConstants.EMPTY_END_ROW; |
| |
| // True if this scan node is not going to scan anything. If the row key filter |
| // evaluates to null, or if the lower bound > upper bound, then this scan node won't |
| // scan at all. |
| private boolean isEmpty_ = false; |
| |
| // List of HBase Filters for generating thrift message. Filled in finalize(). |
| private final List<THBaseFilter> filters_ = new ArrayList<>(); |
| |
| private int suggestedCaching_ = DEFAULT_SUGGESTED_CACHING; |
| |
| public HBaseScanNode(PlanNodeId id, TupleDescriptor desc) { |
| super(id, desc, "SCAN HBASE"); |
| desc_ = desc; |
| } |
| |
| @Override |
| public void init(Analyzer analyzer) throws ImpalaException { |
| FeTable table = desc_.getTable(); |
| // determine scan predicates for clustering cols |
| for (int i = 0; i < table.getNumClusteringCols(); ++i) { |
| SlotDescriptor slotDesc = analyzer.getColumnSlot( |
| desc_, table.getColumns().get(i)); |
| if (slotDesc == null || !slotDesc.getType().isStringType()) { |
| // the hbase row key is mapped to a non-string type |
| // (since it's stored in ASCII it will be lexicographically ordered, |
| // and non-string comparisons won't work) |
| keyRanges_.add(null); |
| } else { |
| keyRanges_.add(createHBaseValueRange(slotDesc)); |
| } |
| } |
| |
| checkForSupportedFileFormats(); |
| assignConjuncts(analyzer); |
| conjuncts_ = orderConjunctsByCost(conjuncts_); |
| setStartStopKey(analyzer); |
| // Convert predicates to HBase filters_. |
| createHBaseFilters(analyzer); |
| |
| // materialize slots in remaining conjuncts_ |
| analyzer.materializeSlots(conjuncts_); |
| computeMemLayout(analyzer); |
| computeScanRangeLocations(analyzer); |
| Preconditions.checkState(!scanRangeSpecs_.isSetSplit_specs()); |
| |
| // Make sure key ranges are not changed any more. So startKey_ and stopKey_ are |
| // stable too. These invariants make it safe to reuse row estimation results in |
| // computeStats. |
| keyRanges_ = Collections.unmodifiableList(keyRanges_); |
| keyConjuncts_ = Collections.unmodifiableList(keyConjuncts_); |
| |
| // Call computeStats() after materializing slots and computing the mem layout. |
| computeStats(analyzer); |
| } |
| |
| /** |
| * Transform '=', '<[=]' and '>[=]' comparisons for given slot into |
| * ValueRange. Also removes those predicates which were used for the construction |
| * of ValueRange from 'conjuncts_'. Only looks at comparisons w/ string constants |
| * (ie, the bounds of the result can be evaluated with Expr::GetValue(NULL)). |
| * HBase row key filtering works only if the row key is mapped to a string column and |
| * the expression is a string constant expression. |
| * If there are multiple competing comparison predicates that could be used |
| * to construct a ValueRange, only the first one from each category is chosen. |
| */ |
| private ValueRange createHBaseValueRange(SlotDescriptor d) { |
| ListIterator<Expr> i = conjuncts_.listIterator(); |
| ValueRange result = null; |
| while (i.hasNext()) { |
| Expr e = i.next(); |
| if (!(e instanceof BinaryPredicate)) continue; |
| BinaryPredicate comp = (BinaryPredicate) e; |
| if ((comp.getOp() == BinaryPredicate.Operator.NE) |
| || (comp.getOp() == BinaryPredicate.Operator.DISTINCT_FROM) |
| || (comp.getOp() == BinaryPredicate.Operator.NOT_DISTINCT)) { |
| continue; |
| } |
| Expr slotBinding = comp.getSlotBinding(d.getId()); |
| if (slotBinding == null || !slotBinding.isConstant() || |
| !slotBinding.getType().equals(Type.STRING)) { |
| continue; |
| } |
| |
| if (comp.getOp() == BinaryPredicate.Operator.EQ) { |
| i.remove(); |
| keyConjuncts_.add(e); |
| return ValueRange.createEqRange(slotBinding); |
| } |
| |
| if (result == null) result = new ValueRange(); |
| |
| // TODO: do we need copies here? |
| if (comp.getOp() == BinaryPredicate.Operator.GT |
| || comp.getOp() == BinaryPredicate.Operator.GE) { |
| if (result.getLowerBound() == null) { |
| result.setLowerBound(slotBinding); |
| result.setLowerBoundInclusive(comp.getOp() == BinaryPredicate.Operator.GE); |
| i.remove(); |
| keyConjuncts_.add(e); |
| } |
| } else { |
| if (result.getUpperBound() == null) { |
| result.setUpperBound(slotBinding); |
| result.setUpperBoundInclusive(comp.getOp() == BinaryPredicate.Operator.LE); |
| i.remove(); |
| keyConjuncts_.add(e); |
| } |
| } |
| } |
| return result; |
| } |
| |
| /** |
| * Convert keyRanges_ to startKey_ and stopKey_. |
| * If ValueRange is not null, transform it into start/stopKey_ by evaluating the |
| * expression. Analysis has checked that the expression is string type. If the |
| * expression evaluates to null, then there's nothing to scan because Hbase row key |
| * cannot be null. |
| * At present, we only do row key filtering for string-mapped keys. String-mapped keys |
| * are always encoded as ASCII. |
| * ValueRange is null if there is no predicate on the row-key. |
| */ |
| private void setStartStopKey(Analyzer analyzer) throws ImpalaException { |
| Preconditions.checkNotNull(keyRanges_); |
| Preconditions.checkState(keyRanges_.size() == 1); |
| |
| ValueRange rowRange = keyRanges_.get(0); |
| if (rowRange != null) { |
| if (rowRange.getLowerBound() != null) { |
| Preconditions.checkState(rowRange.getLowerBound().isConstant()); |
| Preconditions.checkState( |
| rowRange.getLowerBound().getType().equals(Type.STRING)); |
| LiteralExpr val = LiteralExpr.createBounded(rowRange.getLowerBound(), |
| analyzer.getQueryCtx(), StringLiteral.MAX_STRING_LEN); |
| // TODO: Make this a Preconditions.checkState(). If we get here, |
| // and the value is not a string literal, then we've got a predicate |
| // that we removed from the conjunct list, but which we won't evaluate |
| // as a key. That is, we'll produce wrong query results. |
| if (val instanceof StringLiteral) { |
| StringLiteral litVal = (StringLiteral) val; |
| startKey_ = convertToBytes(litVal.getUnescapedValue(), |
| !rowRange.getLowerBoundInclusive()); |
| } else { |
| // lower bound is null. |
| isEmpty_ = true; |
| return; |
| } |
| } |
| if (rowRange.getUpperBound() != null) { |
| Preconditions.checkState(rowRange.getUpperBound().isConstant()); |
| Preconditions.checkState( |
| rowRange.getUpperBound().getType().equals(Type.STRING)); |
| LiteralExpr val = LiteralExpr.createBounded(rowRange.getUpperBound(), |
| analyzer.getQueryCtx(), StringLiteral.MAX_STRING_LEN); |
| if (val instanceof StringLiteral) { |
| StringLiteral litVal = (StringLiteral) val; |
| stopKey_ = convertToBytes(litVal.getUnescapedValue(), |
| rowRange.getUpperBoundInclusive()); |
| } else { |
| // lower bound is null. |
| isEmpty_ = true; |
| return; |
| } |
| } |
| } |
| |
| boolean endKeyIsEndOfTable = Bytes.equals(stopKey_, HConstants.EMPTY_END_ROW); |
| if ((Bytes.compareTo(startKey_, stopKey_) > 0) && !endKeyIsEndOfTable) { |
| // Lower bound is greater than upper bound. |
| isEmpty_ = true; |
| } |
| } |
| |
| /** |
| * Also sets suggestedCaching_. |
| */ |
| @Override |
| public void computeStats(Analyzer analyzer) { |
| super.computeStats(analyzer); |
| FeHBaseTable tbl = (FeHBaseTable) desc_.getTable(); |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("computing stats for HbaseScan on " + tbl.getHBaseTableName()); |
| } |
| ValueRange rowRange = keyRanges_.get(0); |
| if (isEmpty_) { |
| cardinality_ = 0; |
| } else if (rowRange != null && rowRange.isEqRange()) { |
| cardinality_ = 1; |
| } else if (inputCardinality_ >= 0) { |
| // We have run computeStats successfully. Don't need to estimate cardinality again |
| // (IMPALA-8912). Check some invariants if computeStats has been called. |
| Preconditions.checkState(numNodes_ > 0); |
| Preconditions.checkState(cardinality_ >= 0); |
| cardinality_ = inputCardinality_; |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("Reuse last stats: inputCardinality_=" + inputCardinality_); |
| } |
| } else { |
| Pair<Long, Long> estimate; |
| if (analyzer.getQueryOptions().isDisable_hbase_num_rows_estimate()) { |
| estimate = new Pair<>(-1L, -1L); |
| } else { |
| // Set maxCaching so that each fetch from hbase won't return a batch of more than |
| // MAX_HBASE_FETCH_BATCH_SIZE bytes. |
| // May return -1 for the estimate if insufficient data is available. |
| estimate = tbl.getEstimatedRowStats(startKey_, stopKey_); |
| } |
| if (estimate.first == -1) { |
| // No useful estimate. Rely on HMS row count stats. |
| // This works only if HBase stats are available in HMS. This is true |
| // for the Impala tests, and may be true for some applications. |
| cardinality_ = tbl.getTTableStats().getNum_rows(); |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("Fallback to use table stats in HMS: num_rows=" + cardinality_); |
| } |
| // TODO: What do do if neither HBase nor HMS provide a row count estimate? |
| // Is there some third, ulitimate fallback? |
| // Apply estimated key range selectivity from original key conjuncts |
| if (cardinality_ > 0 && keyConjuncts_ != null) { |
| cardinality_ = |
| applySelectivity(cardinality_, computeCombinedSelectivity(keyConjuncts_)); |
| } |
| } else { |
| // Use the HBase sampling scan to estimate cardinality. Note that, |
| // in tests, this estimate has proven to be very rough: off by |
| // 2x or more. |
| cardinality_ = estimate.first; |
| if (estimate.second > 0) { |
| suggestedCaching_ = (int) |
| Math.max(MAX_HBASE_FETCH_BATCH_SIZE / estimate.second, 1); |
| } |
| } |
| } |
| inputCardinality_ = cardinality_; |
| |
| if (cardinality_ > 0) { |
| cardinality_ = applyConjunctsSelectivity(cardinality_); |
| } else { |
| // Safe guard for cardinality_ < -1, e.g. when hbase sampling fails and numRows |
| // in HMS is abnormally set to be < -1. |
| cardinality_ = Math.max(-1, cardinality_); |
| } |
| cardinality_ = capCardinalityAtLimit(cardinality_); |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("computeStats HbaseScan: cardinality=" + cardinality_); |
| } |
| |
| // Assume that each executor in the cluster gets a scan range, unless there are fewer |
| // scan ranges than nodes. |
| numNodes_ = Math.max(1, Math.min(scanRangeSpecs_.getConcrete_rangesSize(), |
| ExecutorMembershipSnapshot.getCluster().numExecutors())); |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("computeStats HbaseScan: #nodes=" + numNodes_); |
| } |
| } |
| |
| @Override |
| protected String debugString() { |
| FeHBaseTable tbl = (FeHBaseTable) desc_.getTable(); |
| return Objects.toStringHelper(this) |
| .add("tid", desc_.getId().asInt()) |
| .add("hiveTblName", tbl.getFullName()) |
| .add("hbaseTblName", tbl.getHBaseTableName()) |
| .add("startKey", ByteBuffer.wrap(startKey_).toString()) |
| .add("stopKey", ByteBuffer.wrap(stopKey_).toString()) |
| .add("isEmpty", isEmpty_) |
| .addValue(super.debugString()) |
| .toString(); |
| } |
| |
| // We convert predicates of the form <slotref> op <constant> where slotref is of |
| // type string to HBase filters. All these predicates are also evaluated at |
| // the HBaseScanNode. To properly filter out NULL values HBaseScanNode treats all |
| // predicates as disjunctive, thereby requiring re-evaluation when there are multiple |
| // attributes. We explicitly materialize the referenced slots, otherwise our hbase |
| // scans don't return correct data. |
| // TODO: expand this to generate nested filter lists for arbitrary conjunctions |
| // and disjunctions. |
| private void createHBaseFilters(Analyzer analyzer) { |
| for (Expr e: conjuncts_) { |
| // We only consider binary predicates |
| if (!(e instanceof BinaryPredicate)) continue; |
| BinaryPredicate bp = (BinaryPredicate) e; |
| CompareFilter.CompareOp hbaseOp = impalaOpToHBaseOp(bp.getOp()); |
| // Ignore unsupported ops |
| if (hbaseOp == null) continue; |
| |
| for (SlotDescriptor slot: desc_.getSlots()) { |
| // Only push down predicates on string columns |
| if (slot.getType().getPrimitiveType() != PrimitiveType.STRING) continue; |
| |
| Expr bindingExpr = bp.getSlotBinding(slot.getId()); |
| if (bindingExpr == null || !(bindingExpr instanceof StringLiteral)) continue; |
| |
| StringLiteral literal = (StringLiteral) bindingExpr; |
| HBaseColumn col = (HBaseColumn) slot.getColumn(); |
| // IMPALA-7929: Since the qualifier can be null (e.g. for the key column of an |
| // HBase table), the qualifier field must be optional in order to express the |
| // null value. Constructors in Thrift do not set optional fields, so the qualifier |
| // must be set separately. |
| THBaseFilter thbf = new THBaseFilter(col.getColumnFamily(), |
| (byte) hbaseOp.ordinal(), literal.getUnescapedValue()); |
| thbf.setQualifier(col.getColumnQualifier()); |
| filters_.add(thbf); |
| analyzer.materializeSlots(Lists.newArrayList(e)); |
| } |
| } |
| } |
| |
| @Override |
| protected void toThrift(TPlanNode msg) { |
| msg.node_type = TPlanNodeType.HBASE_SCAN_NODE; |
| FeHBaseTable tbl = (FeHBaseTable) desc_.getTable(); |
| msg.hbase_scan_node = |
| new THBaseScanNode(desc_.getId().asInt(), tbl.getHBaseTableName()); |
| if (!filters_.isEmpty()) { |
| msg.hbase_scan_node.setFilters(filters_); |
| } |
| msg.hbase_scan_node.setSuggested_max_caching(suggestedCaching_); |
| } |
| |
| /** |
| * We create a TScanRange for each region server that contains at least one |
| * relevant region, and the created TScanRange will contain all the relevant regions |
| * of that region server. |
| */ |
| private void computeScanRangeLocations(Analyzer analyzer) { |
| scanRangeSpecs_ = new TScanRangeSpec(); |
| |
| // For empty scan node, return an empty list. |
| if (isEmpty_) return; |
| |
| // Retrieve relevant HBase regions and their region servers |
| FeHBaseTable tbl = (FeHBaseTable) desc_.getTable(); |
| List<HRegionLocation> regionsLoc; |
| try { |
| regionsLoc = FeHBaseTable.Util.getRegionsInRange(tbl, startKey_, stopKey_); |
| } catch (IOException e) { |
| throw new RuntimeException( |
| "couldn't retrieve HBase table (" + tbl.getHBaseTableName() + ") info:\n" |
| + e.getMessage(), e); |
| } |
| |
| // Convert list of HRegionLocation to Map<hostport, List<HRegionLocation>>. |
| // The List<HRegionLocations>'s end up being sorted by start key/end key, because |
| // regionsLoc is sorted that way. |
| Map<String, List<HRegionLocation>> locationMap = new HashMap<>(); |
| for (HRegionLocation regionLoc: regionsLoc) { |
| String locHostPort = regionLoc.getHostnamePort(); |
| if (locationMap.containsKey(locHostPort)) { |
| locationMap.get(locHostPort).add(regionLoc); |
| } else { |
| locationMap.put(locHostPort, Lists.newArrayList(regionLoc)); |
| } |
| } |
| |
| for (Map.Entry<String, List<HRegionLocation>> locEntry: locationMap.entrySet()) { |
| // HBaseTableScanner(backend) initializes a result scanner for each key range. |
| // To minimize # of result scanner re-init, create only a single HBaseKeyRange |
| // for all adjacent regions on this server. |
| THBaseKeyRange keyRange = null; |
| byte[] prevEndKey = null; |
| for (HRegionLocation regionLoc: locEntry.getValue()) { |
| byte[] curRegStartKey = regionLoc.getRegionInfo().getStartKey(); |
| byte[] curRegEndKey = regionLoc.getRegionInfo().getEndKey(); |
| if (prevEndKey != null && |
| Bytes.compareTo(prevEndKey, curRegStartKey) == 0) { |
| // the current region starts where the previous one left off; |
| // extend the key range |
| setKeyRangeEnd(keyRange, curRegEndKey); |
| } else { |
| // create a new HBaseKeyRange (and TScanRange2/TScanRangeLocationList to go |
| // with it). |
| keyRange = new THBaseKeyRange(); |
| setKeyRangeStart(keyRange, curRegStartKey); |
| setKeyRangeEnd(keyRange, curRegEndKey); |
| |
| TScanRangeLocationList scanRangeLocation = new TScanRangeLocationList(); |
| TNetworkAddress networkAddress = addressToTNetworkAddress(locEntry.getKey()); |
| scanRangeLocation.addToLocations( |
| new TScanRangeLocation(analyzer.getHostIndex().getIndex(networkAddress))); |
| |
| TScanRange scanRange = new TScanRange(); |
| scanRange.setHbase_key_range(keyRange); |
| scanRangeLocation.setScan_range(scanRange); |
| |
| scanRangeSpecs_.addToConcrete_ranges(scanRangeLocation); |
| } |
| prevEndKey = curRegEndKey; |
| } |
| } |
| } |
| |
| /** |
| * Set the start key of keyRange using the provided key, bounded by startKey_ |
| * @param keyRange the keyRange to be updated |
| * @param rangeStartKey the start key value to be set to |
| */ |
| private void setKeyRangeStart(THBaseKeyRange keyRange, byte[] rangeStartKey) { |
| keyRange.unsetStartKey(); |
| // use the max(startKey, rangeStartKey) for scan start |
| if (!Bytes.equals(rangeStartKey, HConstants.EMPTY_START_ROW) || |
| !Bytes.equals(startKey_, HConstants.EMPTY_START_ROW)) { |
| byte[] partStart = (Bytes.compareTo(rangeStartKey, startKey_) < 0) ? |
| startKey_ : rangeStartKey; |
| keyRange.setStartKey(Bytes.toString(partStart)); |
| } |
| } |
| |
| /** |
| * Set the end key of keyRange using the provided key, bounded by stopKey_ |
| * @param keyRange the keyRange to be updated |
| * @param rangeEndKey the end key value to be set to |
| */ |
| private void setKeyRangeEnd(THBaseKeyRange keyRange, byte[] rangeEndKey) { |
| keyRange.unsetStopKey(); |
| // use the min(stopkey, regionStopKey) for scan stop |
| if (!Bytes.equals(rangeEndKey, HConstants.EMPTY_END_ROW) || |
| !Bytes.equals(stopKey_, HConstants.EMPTY_END_ROW)) { |
| if (Bytes.equals(stopKey_, HConstants.EMPTY_END_ROW)) { |
| keyRange.setStopKey(Bytes.toString(rangeEndKey)); |
| } else if (Bytes.equals(rangeEndKey, HConstants.EMPTY_END_ROW)) { |
| keyRange.setStopKey(Bytes.toString(stopKey_)); |
| } else { |
| byte[] partEnd = (Bytes.compareTo(rangeEndKey, stopKey_) < 0) ? |
| rangeEndKey : stopKey_; |
| keyRange.setStopKey(Bytes.toString(partEnd)); |
| } |
| } |
| } |
| |
| @Override |
| protected String getNodeExplainString(String prefix, String detailPrefix, |
| TExplainLevel detailLevel) { |
| FeHBaseTable table = (FeHBaseTable) desc_.getTable(); |
| StringBuilder output = new StringBuilder(); |
| if (isEmpty_) { |
| output.append(prefix + "empty scan node\n"); |
| return output.toString(); |
| } |
| String aliasStr = ""; |
| if (!table.getFullName().equalsIgnoreCase(desc_.getAlias()) && |
| !table.getName().equalsIgnoreCase(desc_.getAlias())) { |
| aliasStr = " " + desc_.getAlias(); |
| } |
| output.append(String.format("%s%s:%s [%s%s]\n", prefix, id_.toString(), |
| displayName_, table.getFullName(), aliasStr)); |
| if (detailLevel.ordinal() >= TExplainLevel.STANDARD.ordinal()) { |
| if (!keyConjuncts_.isEmpty()) { |
| output.append(detailPrefix + "key predicates: " + |
| Expr.getExplainString(keyConjuncts_, detailLevel) + "\n"); |
| } |
| if (!Bytes.equals(startKey_, HConstants.EMPTY_START_ROW)) { |
| output.append(detailPrefix + "start key: " + printKey(startKey_) + "\n"); |
| } |
| if (!Bytes.equals(stopKey_, HConstants.EMPTY_END_ROW)) { |
| output.append(detailPrefix + "stop key: " + printKey(stopKey_) + "\n"); |
| } |
| if (!filters_.isEmpty()) { |
| output.append(detailPrefix + "hbase filters:"); |
| if (filters_.size() == 1) { |
| THBaseFilter filter = filters_.get(0); |
| output.append(" " + filter.family + ":" + filter.qualifier + " " + |
| CompareFilter.CompareOp.values()[filter.op_ordinal].toString() + " " + |
| "'" + filter.filter_constant + "'"); |
| } else { |
| for (int i = 0; i < filters_.size(); ++i) { |
| THBaseFilter filter = filters_.get(i); |
| output.append("\n " + filter.family + ":" + filter.qualifier + " " + |
| CompareFilter.CompareOp.values()[filter.op_ordinal].toString() + " " + |
| "'" + filter.filter_constant + "'"); |
| } |
| } |
| output.append('\n'); |
| } |
| if (!conjuncts_.isEmpty()) { |
| output.append(detailPrefix |
| + "predicates: " + Expr.getExplainString(conjuncts_, detailLevel) + "\n"); |
| } |
| } |
| if (detailLevel.ordinal() >= TExplainLevel.EXTENDED.ordinal()) { |
| output.append(getStatsExplainString(detailPrefix)); |
| output.append("\n"); |
| } |
| return output.toString(); |
| } |
| |
| /** |
| * Convert key into byte array and append a '\0' if 'nextKey' is true. |
| */ |
| private byte[] convertToBytes(String rowKey, boolean nextKey) { |
| byte[] keyBytes = Bytes.toBytes(rowKey); |
| if (!nextKey) { |
| return keyBytes; |
| } else { |
| // append \0 |
| return Arrays.copyOf(keyBytes, keyBytes.length + 1); |
| } |
| } |
| |
| /** |
| * Prints non-printable characters in escaped octal, otherwise outputs |
| * the characters. |
| */ |
| public static String printKey(byte[] key) { |
| StringBuilder result = new StringBuilder(); |
| for (int i = 0; i < key.length; ++i) { |
| if (!Character.isISOControl(key[i])) { |
| result.append((char) key[i]); |
| } else { |
| result.append("\\"); |
| result.append(Integer.toOctalString(key[i])); |
| } |
| } |
| return result.toString(); |
| } |
| |
| private static CompareFilter.CompareOp impalaOpToHBaseOp( |
| BinaryPredicate.Operator impalaOp) { |
| switch(impalaOp) { |
| case EQ: return CompareFilter.CompareOp.EQUAL; |
| case NE: return CompareFilter.CompareOp.NOT_EQUAL; |
| case GT: return CompareFilter.CompareOp.GREATER; |
| case GE: return CompareFilter.CompareOp.GREATER_OR_EQUAL; |
| case LT: return CompareFilter.CompareOp.LESS; |
| case LE: return CompareFilter.CompareOp.LESS_OR_EQUAL; |
| // TODO: Add support for pushing LIKE/REGEX down to HBase with a different Filter. |
| default: throw new IllegalArgumentException( |
| "HBase: Unsupported Impala compare operator: " + impalaOp); |
| } |
| } |
| |
| @Override |
| public void computeNodeResourceProfile(TQueryOptions queryOptions) { |
| FeHBaseTable tbl = (FeHBaseTable) desc_.getTable(); |
| // The first column in an HBase table is always the key column. |
| HBaseColumn keyCol = (HBaseColumn) tbl.getColumns().get(0); |
| |
| List<HBaseColumn> colsToFetchFromHBase = new ArrayList<>(); |
| for (SlotDescriptor slot : desc_.getSlots()) { |
| HBaseColumn col = (HBaseColumn) tbl.getColumn(slot.getLabel()); |
| // Will add key column separately, since its always fetched. |
| if (col.getColumnFamily().equals(FeHBaseTable.Util.ROW_KEY_COLUMN_FAMILY)) continue; |
| colsToFetchFromHBase.add(col); |
| } |
| // Add the key column. |
| colsToFetchFromHBase.add(keyCol); |
| long mem_estimate = memoryEstimateForFetchingColumns(colsToFetchFromHBase); |
| mem_estimate = Math.max(mem_estimate, DEFAULT_MIN_ESTIMATE_BYTES); |
| nodeResourceProfile_ = ResourceProfile.noReservation(mem_estimate); |
| } |
| |
| /** |
| * Returns an estimate of memory required by the HBase scan node for fetching |
| * the given list of HBase columns. Primarily used as a helper function but |
| * also exposed at the package level for testing. |
| */ |
| protected static long memoryEstimateForFetchingColumns(List<HBaseColumn> columns) { |
| // In HBase, every column value is stored in the following format: |
| // http://hbase.apache.org/0.94/book/regions.arch.html#keyvalue |
| // and out of this only rowKey per row and (columnFamily, columnQualifier and |
| // columnValue) per column per row are allocated. The following estimations are based |
| // on that and its interaction with the mem-pool. Currently, we do an |
| // allocate-clear cycle on mem-pool for each row fetched from HBase (See |
| // HBaseTableScanner::Next()). To get an approx upper limit on mem |
| // allocation, we take the max row size possible and assume all possible |
| // chunk sizes below it have been allocated in previous row iterations. So, |
| // for a value of n bytes the max possible mem allocation for a mem pool |
| // that only uses power of 2 chunk sizes will be: |
| // (2^(ceil((log(n))+1) - 1) ~ 2 * BitUtil.roundUpToPowerOf2(n) |
| long maxRowSize = 0; |
| boolean isMissingStats = false; |
| for (HBaseColumn col : columns) { |
| long colMaxSize = col.getStats().getMaxSize(); |
| if (col.getType().isStringType()) { |
| if (colMaxSize == -1) { |
| colMaxSize = DEFAULT_STRING_COL_BYTES; |
| isMissingStats = true; |
| } |
| // Round off string col size to next power of 2. For strings less than |
| // 512 KB (MemPool::MAX_CHUNK_SIZE) this ensures enough contribution to |
| // the max row size so that the final round off can accommodate any |
| // fluctuations in size. For larger strings, this approximates that it |
| // completely takes up the new chunk allocated for it. |
| colMaxSize = BitUtil.roundUpToPowerOf2(colMaxSize); |
| } |
| Preconditions.checkState(colMaxSize != -1); |
| maxRowSize += colMaxSize; |
| if (!col.getColumnFamily().equals(FeHBaseTable.Util.ROW_KEY_COLUMN_FAMILY)) { |
| // For non-key columns, their respective column family and column qualifier |
| // strings need to be fetched and mem needs to be allocated for that too. |
| maxRowSize += col.getColumnFamily().length() + col.getColumnQualifier().length(); |
| } |
| } |
| // Use the max allocation assuming all possible chunk sizes below it have |
| // been allocated. |
| long mem_estimate = BitUtil.roundUpToPowerOf2(maxRowSize) * 2; |
| // We use a default of 32 KB for string cols that dont have max length set. |
| // Assuming such large cols are uncommon we set an upper limit to avoid extreme |
| // overestimation. |
| if (isMissingStats) mem_estimate = Math.min(mem_estimate, DEFAULT_MAX_ESTIMATE_BYTES); |
| return mem_estimate; |
| } |
| |
| @Override |
| public boolean hasStorageLayerConjuncts() { return !filters_.isEmpty(); } |
| } |
