| // @@@ START COPYRIGHT @@@ |
| // |
| // 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. |
| // |
| // @@@ END COPYRIGHT @@@ |
| |
| package org.trafodion.sql; |
| import org.trafodion.sql.*; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.List; |
| import java.util.HashMap; |
| import java.util.Map; |
| import java.util.NavigableSet; |
| |
| import org.apache.commons.codec.binary.Hex; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.nio.ByteBuffer; |
| import java.nio.LongBuffer; |
| import java.nio.ByteOrder; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.KeyValue; |
| import org.apache.hadoop.hbase.Cell; |
| import org.apache.hadoop.hbase.client.Delete; |
| import org.apache.hadoop.hbase.client.Get; |
| import org.apache.hadoop.hbase.client.Put; |
| import org.apache.hadoop.hbase.client.Result; |
| import org.apache.hadoop.hbase.client.ResultScanner; |
| import org.apache.hadoop.hbase.client.Scan; |
| import org.apache.hadoop.hbase.client.Connection; |
| import org.apache.hadoop.hbase.client.Table; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.hadoop.hbase.client.coprocessor.AggregationClient; |
| import org.apache.hadoop.hbase.client.transactional.RMInterface; |
| import org.apache.hadoop.hbase.client.transactional.TransactionalAggregationClient; |
| import org.apache.hadoop.hbase.client.transactional.TransactionalTable; |
| import org.apache.hadoop.hbase.client.transactional.TransactionState; |
| |
| import org.apache.log4j.Logger; |
| |
| // H98 coprocessor needs |
| import java.util.*; |
| import org.apache.hadoop.hbase.*; |
| import org.apache.hadoop.hbase.client.*; |
| import org.apache.hadoop.hbase.client.coprocessor.*; |
| import org.apache.hadoop.hbase.coprocessor.*; |
| import org.apache.hadoop.hbase.ipc.*; |
| import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.*; |
| import org.apache.hadoop.hbase.util.*; |
| |
| //import org.apache.hadoop.hbase.client.coprocessor.AggregationClient; |
| import org.apache.hadoop.hbase.coprocessor.ColumnInterpreter; |
| import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter; |
| |
| // classes to do hbase pushdown filtering |
| import org.apache.hadoop.hbase.filter.Filter; |
| import org.apache.hadoop.hbase.filter.SingleColumnValueFilter; |
| import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp; |
| import org.apache.hadoop.hbase.filter.FilterList; |
| import org.apache.hadoop.hbase.filter.RandomRowFilter; |
| import org.apache.hadoop.hbase.filter.FamilyFilter; |
| import org.apache.hadoop.hbase.filter.QualifierFilter; |
| import org.apache.hadoop.hbase.filter.SingleColumnValueExcludeFilter; |
| import org.apache.hadoop.hbase.filter.BinaryComparator; |
| import org.apache.hadoop.hbase.filter.BinaryPrefixComparator; |
| import org.apache.hadoop.hbase.filter.ByteArrayComparable; |
| import org.apache.hadoop.hbase.filter.NullComparator; |
| |
| import org.apache.hadoop.hbase.client.TableSnapshotScanner; |
| import org.apache.hadoop.hbase.client.Admin; |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.TableName; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.FileUtil; |
| import java.util.UUID; |
| import java.security.InvalidParameterException; |
| |
| public class HTableClient { |
| private static final int GET_ROW = 1; |
| private static final int BATCH_GET = 2; |
| private static final int SCAN_FETCH = 3; |
| private boolean useTRex; |
| private boolean useTRexScanner; |
| private static boolean envUseTRex; |
| private static boolean envUseTRexScanner; |
| private String tableName; |
| private static Connection connection; |
| private ResultScanner scanner = null; |
| private ScanHelper scanHelper = null; |
| Result[] getResultSet = null; |
| RMInterface table = null; |
| private boolean writeToWAL = false; |
| int numRowsCached = 1; |
| int numColsInScan = 0; |
| int[] kvValLen = null; |
| int[] kvValOffset = null; |
| int[] kvQualLen = null; |
| int[] kvQualOffset = null; |
| int[] kvFamLen = null; |
| int[] kvFamOffset = null; |
| long[] kvTimestamp = null; |
| byte[][] kvBuffer = null; |
| byte[][] rowIDs = null; |
| int[] kvsPerRow = null; |
| byte[][] kvFamArray = null; |
| byte[][] kvQualArray = null; |
| static ExecutorService executorService = null; |
| Future future = null; |
| boolean preFetch = false; |
| int fetchType = 0; |
| long jniObject = 0; |
| SnapshotScanHelper snapHelper = null; |
| |
| class SnapshotScanHelper |
| { |
| Path snapRestorePath = null; |
| Admin admin = null; |
| Configuration conf = null; |
| SnapshotDescription snpDesc = null; |
| String tmpLocation = null; |
| FileSystem fs = null; |
| |
| SnapshotScanHelper( Configuration cnfg , String tmpLoc, String snapName) |
| throws IOException |
| { |
| conf = cnfg; |
| tmpLocation = tmpLoc; |
| setSnapshotDescription(snapName); |
| Path rootDir = new Path(conf.get(HConstants.HBASE_DIR)); |
| fs = rootDir.getFileSystem(conf); |
| setSnapRestorePath(); |
| } |
| |
| String getTmpLocation() |
| { |
| return tmpLocation; |
| } |
| String getSnapshotName() |
| { |
| if (snpDesc == null) |
| return null; |
| return snpDesc.getName(); |
| } |
| |
| void setSnapRestorePath() throws IOException |
| { |
| String restoreDirStr = tmpLocation + getSnapshotDescription().getName(); ; |
| snapRestorePath = new Path(restoreDirStr); |
| snapRestorePath = snapRestorePath.makeQualified(fs.getUri(), snapRestorePath); |
| } |
| |
| Path getSnapRestorePath() throws IOException |
| { |
| return snapRestorePath; |
| } |
| |
| boolean snapshotExists() throws IOException |
| { |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.snapshotExists() called. "); |
| Admin admin = connection.getAdmin(); |
| boolean retcode = !(admin.listSnapshots(snpDesc.getName()).isEmpty()); |
| admin.close(); |
| return retcode; |
| } |
| |
| void deleteSnapshot() throws IOException |
| { |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.deleteSnapshot() called. "); |
| if (snapshotExists()) |
| { |
| Admin admin = connection.getAdmin(); |
| admin.deleteSnapshot(snpDesc.getName()); |
| admin.close(); |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.deleteSnapshot(). snapshot: " + snpDesc.getName() + " deleted."); |
| } |
| else |
| { |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.deleteSnapshot(). snapshot: " + snpDesc.getName() + " does not exist."); |
| } |
| |
| } |
| void deleteRestorePath() throws IOException |
| { |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.deleteRestorePath() called. "); |
| if (fs.exists(snapRestorePath)) |
| { |
| fs.delete(snapRestorePath, true); |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.deleteRestorePath(). restorePath: " + snapRestorePath + " deleted."); |
| } |
| else |
| { |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.deleteRestorePath(). restorePath: " + snapRestorePath + " does not exist."); |
| } |
| } |
| |
| void createTableSnapshotScanner(int timeout, int slp, long nbre, Scan scan) throws InterruptedException, IOException |
| { |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.createTableSnapshotScanner() called. "); |
| int xx=0; |
| IOException ioExc = null; |
| while (xx < timeout) |
| { |
| xx++; |
| scanner = null; |
| try |
| { |
| ioExc = null; |
| scanner = new TableSnapshotScanner(connection.getConfiguration(), snapHelper.getSnapRestorePath(), snapHelper.getSnapshotName(), scan); |
| } |
| catch(IOException e ) |
| { |
| ioExc = e; |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.createTableSnapshotScanner(). espNumber: " + nbre + |
| " snapshot " + snpDesc.getName() + " TableSnapshotScanner Exception :" + e); |
| Thread.sleep(slp); |
| continue; |
| } |
| if (logger.isTraceEnabled()) logger.trace("[Snapshot Scan] SnapshotScanHelper.createTableSnapshotScanner(). espNumber: " + |
| nbre + " snapshot " + snpDesc.getName() + " TableSnapshotScanner Done - Scanner:" + scanner ); |
| break; |
| } |
| if (ioExc != null) |
| throw ioExc; |
| } |
| |
| void setSnapshotDescription( String snapName) |
| { |
| if (snapName == null ) |
| throw new InvalidParameterException ("snapshotName is null."); |
| |
| SnapshotDescription.Builder builder = SnapshotDescription.newBuilder(); |
| builder.setTable(Bytes.toString(table.getTableName())); |
| builder.setName(snapName); |
| builder.setType(SnapshotDescription.Type.FLUSH); |
| snpDesc = builder.build(); |
| } |
| SnapshotDescription getSnapshotDescription() |
| { |
| return snpDesc; |
| } |
| |
| public void release() throws IOException |
| { |
| if (logger.isTraceEnabled()) logger.trace("HTableClient.release(" + (tableName == null ? " tableName is null " : tableName) + ") called."); |
| if (admin != null) |
| { |
| admin.close(); |
| admin = null; |
| } |
| } |
| } |
| |
| public HTableClient(Connection connection) { |
| this.connection = connection; |
| } |
| |
| class ScanHelper implements Callable { |
| public Result[] call() throws IOException { |
| return scanner.next(numRowsCached); |
| } |
| } |
| |
| static Logger logger = Logger.getLogger(HTableClient.class.getName());; |
| |
| static public byte[] getFamily(byte[] qc) { |
| byte[] family = null; |
| |
| if (qc != null && qc.length > 0) { |
| int pos = Bytes.indexOf(qc, (byte) ':'); |
| if (pos == -1) |
| family = Bytes.toBytes("cf1"); |
| else |
| family = Arrays.copyOfRange(qc, 0, pos); |
| } |
| return family; |
| } |
| |
| static public byte[] getName(byte[] qc) { |
| byte[] name = null; |
| |
| if (qc != null && qc.length > 0) { |
| int pos = Bytes.indexOf(qc, (byte) ':'); |
| if (pos == -1) |
| name = qc; |
| else |
| name = Arrays.copyOfRange(qc, pos + 1, qc.length); |
| } |
| return name; |
| } |
| |
| public boolean setWriteBufferSize(long writeBufferSize) throws IOException { |
| if (logger.isDebugEnabled()) logger.debug("Enter HTableClient::setWriteBufferSize, size : " + writeBufferSize); |
| table.setWriteBufferSize(writeBufferSize); |
| return true; |
| } |
| public long getWriteBufferSize() { |
| if (logger.isDebugEnabled()) logger.debug("Enter HTableClient::getWriteBufferSize, size return : " + table.getWriteBufferSize()); |
| return table.getWriteBufferSize(); |
| } |
| public boolean setWriteToWAL(boolean v) { |
| if (logger.isDebugEnabled()) logger.debug("Enter HTableClient::setWriteToWALL, size : " + v); |
| writeToWAL = v; |
| return true; |
| } |
| |
| public boolean init(String tblName, |
| boolean useTRex) throws IOException |
| { |
| if (logger.isDebugEnabled()) logger.debug("Enter HTableClient::init, tableName: " + tblName); |
| this.useTRex = useTRex; |
| tableName = tblName; |
| |
| if ( !this.useTRex ) { |
| this.useTRexScanner = false; |
| } |
| else { |
| |
| // If the parameter useTRex is false, then do not go thru this logic |
| this.useTRex = envUseTRex; |
| this.useTRexScanner = envUseTRexScanner; |
| } |
| table = new RMInterface(tblName, connection); |
| if (logger.isDebugEnabled()) logger.debug("Exit HTableClient::init, useTRex: " + this.useTRex + ", useTRexScanner: " |
| + this.useTRexScanner + ", table object: " + table); |
| return true; |
| } |
| |
| String getTableName() { |
| return tableName; |
| } |
| |
| String getHTableName() { |
| if (table == null) |
| return null; |
| else |
| return new String(table.getTableName()); |
| } |
| |
| private enum Op { |
| EQUAL, EQUAL_NULL, NOT_EQUAL, NOT_EQUAL_NULL, LESS, LESS_NULL, LESS_OR_EQUAL, LESS_OR_EQUAL_NULL, GREATER, GREATER_NULL, |
| GREATER_OR_EQUAL, GREATER_OR_EQUAL_NULL, NO_OP, NO_OP_NULL,IS_NULL, IS_NULL_NULL, IS_NOT_NULL, IS_NOT_NULL_NULL, AND, OR}; |
| |
| private Filter SingleColumnValueExcludeOrNotFilter(byte[] columnToFilter, |
| CompareOp op, |
| ByteArrayComparable comparator, |
| HashMap<String,Object> columnsToRemove, |
| Boolean... filterIfMissing){ |
| Filter result; |
| boolean fMissing = filterIfMissing.length>0?filterIfMissing[0]:false;//default to false |
| if ((columnsToRemove == null) || !columnsToRemove.containsKey(new String(columnToFilter))){ |
| result = new SingleColumnValueFilter(getFamily(columnToFilter), getName(columnToFilter), op, comparator); |
| ((SingleColumnValueFilter)result).setFilterIfMissing(fMissing); |
| } |
| else{ |
| result= new SingleColumnValueExcludeFilter(getFamily(columnToFilter), getName(columnToFilter), op, comparator); |
| ((SingleColumnValueExcludeFilter)result).setFilterIfMissing(fMissing); |
| } |
| return result; |
| } |
| |
| // construct the hbase filter |
| // optimizes for OR and AND associativity |
| // optimizes for detection of a<? and a>? on nullable and non nullable column equivalent to a<>? |
| // optimize for null check factorization (A not null and (A <op> ?)) or (A not null and A <op2> ?) -> A not null and (A <op> ? or A <op2> ?) |
| // this is an important optimzation for IN statement on non null column |
| // uses the columnToRemove parametter to know if we need to use the SingleColumnValue Exclude or not method to limit returned columns |
| |
| private Filter constructV2Filter(Object[] colNamesToFilter, |
| Object[] compareOpList, |
| Object[] colValuesToCompare, |
| HashMap<String,Object> columnsToRemove){ |
| LinkedList linkedList = new LinkedList(); |
| //populate the list with nodes in reverse polish notation order. |
| int k=0;//column index |
| int kk=0;//value index |
| for (int i=1; i<compareOpList.length; i++){ // skip first one containing "V2" marker |
| String opStr = new String((byte[])compareOpList[i]); |
| switch(Op.valueOf(opStr)){ |
| |
| case EQUAL: |
| linkedList.addLast(SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove |
| )); |
| k++;kk++; |
| break; |
| case EQUAL_NULL: |
| linkedList.addLast(new FilterList(FilterList.Operator.MUST_PASS_ALL, //AND between if not null and the actual |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryPrefixComparator(new byte[]{0x00}),//check for null indicator = 0 representing non null |
| columnsToRemove, |
| true //filterIfMissing |
| ), |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove))); |
| k++;kk++; |
| break; |
| case NOT_EQUAL: |
| linkedList.addLast(SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.NOT_EQUAL, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove)); |
| k++;kk++; |
| break; |
| case NOT_EQUAL_NULL: |
| linkedList.addLast(new FilterList(FilterList.Operator.MUST_PASS_ALL, //AND between if not null and the actual |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryPrefixComparator(new byte[]{0x00}),//check for null indicator = 0 representing non null |
| columnsToRemove, |
| true), //filterIfMissing, |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.NOT_EQUAL, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove))); |
| k++;kk++; |
| break; |
| case LESS: |
| linkedList.addLast(SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.LESS, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove)); |
| k++;kk++; |
| break; |
| case LESS_NULL: |
| linkedList.addLast(new FilterList(FilterList.Operator.MUST_PASS_ALL, //AND between if not null and the actual |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryPrefixComparator(new byte[]{0x00}),//check for null indicator = 0 representing non null |
| columnsToRemove, |
| true), //filterIfMissing, |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.LESS, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove))); |
| k++;kk++; |
| break; |
| case LESS_OR_EQUAL: |
| linkedList.addLast(SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.LESS_OR_EQUAL, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove)); |
| k++;kk++; |
| break; |
| case LESS_OR_EQUAL_NULL: |
| linkedList.addLast(new FilterList(FilterList.Operator.MUST_PASS_ALL, //AND between if not null and the actual |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryPrefixComparator(new byte[]{0x00}),//check for null indicator = 0 representing non null |
| columnsToRemove, |
| true), //filterIfMissing, |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.LESS_OR_EQUAL, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove))); |
| k++;kk++; |
| break; |
| case GREATER: |
| linkedList.addLast(SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.GREATER, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove)); |
| k++;kk++; |
| break; |
| case GREATER_NULL: |
| linkedList.addLast(new FilterList(FilterList.Operator.MUST_PASS_ALL, //AND between if not null and the actual |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryPrefixComparator(new byte[]{0x00}),//check for null indicator = 0 representing non null |
| columnsToRemove, |
| true), //filterIfMissing, |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.GREATER, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove))); |
| k++;kk++; |
| break; |
| case GREATER_OR_EQUAL: |
| linkedList.addLast(SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.GREATER_OR_EQUAL, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove)); |
| k++;kk++; |
| break; |
| case GREATER_OR_EQUAL_NULL: |
| linkedList.addLast(new FilterList(FilterList.Operator.MUST_PASS_ALL, //AND between if not null and the actual |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryPrefixComparator(new byte[]{0x00}),//check for null indicator = 0 representing non null |
| columnsToRemove, |
| true), //filterIfMissing, |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.GREATER_OR_EQUAL, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove))); |
| k++;kk++; |
| break; |
| case NO_OP: |
| linkedList.addLast(SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.NO_OP, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove)); |
| k++;kk++; |
| break; |
| case NO_OP_NULL: |
| linkedList.addLast(new FilterList(FilterList.Operator.MUST_PASS_ALL, //AND between if not null and the actual |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryPrefixComparator(new byte[]{0x00}),//check for null indicator = 0 representing non null |
| columnsToRemove, |
| true), //filterIfMissing, |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.NO_OP, |
| new BinaryComparator((byte[])colValuesToCompare[kk]), |
| columnsToRemove))); |
| k++;kk++; |
| break; |
| case IS_NULL: |
| // is null on a non nullable column! |
| linkedList.addLast(SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.NO_OP, //exclude everything |
| new BinaryPrefixComparator((new byte[]{})), |
| columnsToRemove)); |
| k++; |
| break; |
| case IS_NULL_NULL: |
| // is_null on nullable column: is absent OR has the first byte set to FF indicating NULL. |
| linkedList.addLast( |
| new FilterList(FilterList.Operator.MUST_PASS_ONE, //OR |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new NullComparator(),//is absent? |
| columnsToRemove), |
| SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.EQUAL, |
| new BinaryPrefixComparator(new byte[]{-1}),//0xFF has null prefix indicator |
| columnsToRemove))); |
| k++; |
| break; |
| case IS_NOT_NULL: |
| // is not null on a non nullable column! |
| // do nothing, always true |
| k++; |
| break; |
| case IS_NOT_NULL_NULL: |
| // is_not_null on nullable column: is not absent AND has the first byte not set to FF indicating NULL. |
| linkedList.addLast(SingleColumnValueExcludeOrNotFilter( |
| (byte[])colNamesToFilter[k], |
| CompareOp.NOT_EQUAL, |
| new BinaryPrefixComparator(new byte[]{-1}),// 0xFF has null prefix indicator |
| columnsToRemove, |
| true));//filter if missing (if absent null) |
| k++; |
| break; |
| case AND: |
| linkedList.addLast("AND"); |
| break; |
| case OR: |
| linkedList.addLast("OR"); |
| break; |
| default: |
| }//switch |
| }//for |
| //evaluate the reverse polish notation list |
| while (linkedList.size()>1){// evaluate until only one element is left in the list |
| //look for first operator (AND or OR) |
| int j=0; |
| while (j<linkedList.size() && !(linkedList.get(j) instanceof String)){ |
| j++; |
| } |
| //here j points on the first operator; (all operands are of type Filter) |
| if (j==linkedList.size()){logger.error("j==linkedList.size()");return null;} // should not happen |
| Filter leftOperand; |
| Filter rightOperand; |
| switch(Op.valueOf((String)linkedList.get(j))){ |
| case AND: |
| FilterList filterListAnd = new FilterList(FilterList.Operator.MUST_PASS_ALL); //AND filterList |
| //left operand |
| leftOperand = (Filter)linkedList.get(j-2); |
| if (leftOperand instanceof FilterList && ((FilterList)leftOperand).getOperator()==FilterList.Operator.MUST_PASS_ALL){//associativity of AND optimization |
| //for(Filter f:((FilterList)leftOperand).getFilters()) |
| // filterListAnd.addFilter(f); |
| filterListAnd = (FilterList)leftOperand; //more efficient than the 2 lines above (kept commented out for code lisibility) |
| }else{ |
| filterListAnd.addFilter(leftOperand); |
| } |
| // right operand |
| rightOperand = (Filter)linkedList.get(j-1); |
| if (rightOperand instanceof FilterList && ((FilterList)rightOperand).getOperator()==FilterList.Operator.MUST_PASS_ALL){//associativity of AND optimization |
| for(Filter f:((FilterList)rightOperand).getFilters()) |
| filterListAnd.addFilter(f); |
| }else{ |
| filterListAnd.addFilter(rightOperand); |
| } |
| // setup evaluated filter |
| linkedList.set(j,filterListAnd); // replace the operator with the constructer filter |
| linkedList.remove(j-1);// remove right operand |
| linkedList.remove(j-2);// remove left operand. warning order matter |
| break; |
| case OR: |
| FilterList filterListOr = new FilterList(FilterList.Operator.MUST_PASS_ONE); //OR filterList |
| leftOperand = (Filter)linkedList.get(j-2); |
| rightOperand = (Filter)linkedList.get(j-1); |
| //begin detection of null check factorization (A not null and (A <op> ?)) or (A not null and A <op2> ?) -> A not null and (A <op> ? or A <op2> ?) |
| //the code is doing more than just nullcheck, but any factorization where left operands are identical |
| if (leftOperand instanceof FilterList && rightOperand instanceof FilterList && |
| ((FilterList)leftOperand).getOperator() == FilterList.Operator.MUST_PASS_ALL && |
| ((FilterList)rightOperand).getOperator() == FilterList.Operator.MUST_PASS_ALL && |
| ((FilterList)leftOperand).getFilters().size() == 2 && |
| ((FilterList)rightOperand).getFilters().size() == 2 && |
| ((FilterList)leftOperand).getFilters().get(0) instanceof SingleColumnValueFilter && //cannot be SingleColumnValueExcludeFilter when we have the optimization scenario |
| ((FilterList)rightOperand).getFilters().get(0) instanceof SingleColumnValueFilter){//cannot be SingleColumnValueExcludeFilter when we have the optimization scenario |
| SingleColumnValueFilter scvfLeft = (SingleColumnValueFilter)((FilterList)leftOperand).getFilters().get(0); |
| SingleColumnValueFilter scvfRight = (SingleColumnValueFilter)((FilterList)rightOperand).getFilters().get(0); |
| if (scvfLeft.getOperator() == scvfRight.getOperator() && //more general case than just for null check (identical operands) |
| Arrays.equals(scvfLeft.getQualifier(),scvfRight.getQualifier()) && |
| Arrays.equals(scvfLeft.getFamily(),scvfRight.getFamily()) && |
| Arrays.equals(scvfLeft.getComparator().getValue(),scvfRight.getComparator().getValue()) && |
| (scvfLeft.getFilterIfMissing() == scvfRight.getFilterIfMissing())){ |
| Filter left = ((FilterList)leftOperand).getFilters().get(1); |
| Filter right = ((FilterList)rightOperand).getFilters().get(1); |
| if (left instanceof FilterList && ((FilterList)left).getOperator()==FilterList.Operator.MUST_PASS_ONE){//associativity of OR optimization |
| //for(Filter f:((FilterList)left).getFilters()) |
| // filterListOr.addFilter(f); |
| filterListOr = (FilterList)left; // more efficient than the 2 lines above (kept commented out for code lisibility) |
| }else{ |
| filterListOr.addFilter(left); |
| } |
| // right operand |
| if (right instanceof FilterList && ((FilterList)right).getOperator()==FilterList.Operator.MUST_PASS_ONE){//associativity of OR optimization |
| for(Filter f:((FilterList)right).getFilters()) |
| filterListOr.addFilter(f); |
| }else{ |
| filterListOr.addFilter(right); |
| } |
| linkedList.set(j,new FilterList(FilterList.Operator.MUST_PASS_ALL,scvfLeft,filterListOr));//resulting factorized AND filter |
| linkedList.remove(j-1);// remove right operand |
| linkedList.remove(j-2);// remove left operand. warning order matter |
| break; |
| } |
| } |
| //end detection of null (and more) check factorization |
| //begin detection of RangeSpec a<>? transformed to a<? or a>? to convert it back to a <> ? when we push down |
| //check for <> on non nullable columns |
| if (leftOperand instanceof SingleColumnValueFilter && rightOperand instanceof SingleColumnValueFilter){ |
| SingleColumnValueFilter leftscvf = (SingleColumnValueFilter)leftOperand; |
| SingleColumnValueFilter rightscvf = (SingleColumnValueFilter)rightOperand; |
| if (leftscvf.getOperator() == CompareOp.LESS && rightscvf.getOperator()== CompareOp.GREATER && |
| Arrays.equals(leftscvf.getQualifier(), rightscvf.getQualifier()) && |
| Arrays.equals(leftscvf.getFamily(), rightscvf.getFamily()) && |
| Arrays.equals(leftscvf.getComparator().getValue(),rightscvf.getComparator().getValue()) |
| ){ |
| // setup evaluated filter |
| linkedList.set(j,new SingleColumnValueFilter(leftscvf.getFamily(), leftscvf.getQualifier(), CompareOp.NOT_EQUAL, leftscvf.getComparator())); // replace the operator with the constructer filter |
| linkedList.remove(j-1);// remove right operand |
| linkedList.remove(j-2);// remove left operand. warning order matter |
| break; |
| } |
| } |
| //check for <> on nullable column |
| if( leftOperand instanceof FilterList && rightOperand instanceof FilterList){ |
| //no need to check FilterList size, as all possible case FilterList size is at least 2. |
| if (((FilterList)leftOperand).getFilters().get(1) instanceof SingleColumnValueFilter && |
| ((FilterList)rightOperand).getFilters().get(1) instanceof SingleColumnValueFilter){ |
| SingleColumnValueFilter leftscvf = (SingleColumnValueFilter)((FilterList)leftOperand).getFilters().get(1); |
| SingleColumnValueFilter rightscvf = (SingleColumnValueFilter)((FilterList)rightOperand).getFilters().get(1); |
| if (leftscvf.getOperator() == CompareOp.LESS && rightscvf.getOperator()== CompareOp.GREATER && |
| Arrays.equals(leftscvf.getQualifier(), rightscvf.getQualifier()) && |
| Arrays.equals(leftscvf.getFamily(), rightscvf.getFamily()) && |
| Arrays.equals(leftscvf.getComparator().getValue(),rightscvf.getComparator().getValue()) |
| ){ |
| // setup evaluated filter |
| SingleColumnValueFilter nullCheck = new SingleColumnValueFilter(// null checker |
| leftscvf.getFamily(), leftscvf.getQualifier(), |
| CompareOp.EQUAL, |
| new BinaryPrefixComparator(new byte[]{0x00})); |
| nullCheck.setFilterIfMissing(true); |
| linkedList.set(j,new FilterList(FilterList.Operator.MUST_PASS_ALL, //AND between if not null and the actual |
| nullCheck, |
| new SingleColumnValueFilter( |
| leftscvf.getFamily(), leftscvf.getQualifier(), |
| CompareOp.NOT_EQUAL, |
| leftscvf.getComparator()))); |
| linkedList.remove(j-1);// remove right operand |
| linkedList.remove(j-2);// remove left operand. warning order matter |
| break; |
| } |
| } |
| } |
| //end detection of RangeSpec a<>? |
| //now general case... |
| //left operand |
| if (leftOperand instanceof FilterList && ((FilterList)leftOperand).getOperator()==FilterList.Operator.MUST_PASS_ONE){//associativity of OR optimization |
| //for(Filter f:((FilterList)leftOperand).getFilters()) |
| // filterListOr.addFilter(f); |
| filterListOr = (FilterList)leftOperand; // more efficient than the 2 lines above (kept commented out for code lisibility) |
| }else{ |
| filterListOr.addFilter(leftOperand); |
| } |
| // right operand |
| if (rightOperand instanceof FilterList && ((FilterList)rightOperand).getOperator()==FilterList.Operator.MUST_PASS_ONE){//associativity of OR optimization |
| for(Filter f:((FilterList)rightOperand).getFilters()) |
| filterListOr.addFilter(f); |
| }else{ |
| filterListOr.addFilter(rightOperand); |
| } |
| // setup evaluated filter |
| linkedList.set(j,filterListOr); // replace the operator with the constructer filter |
| linkedList.remove(j-1);// remove right operand |
| linkedList.remove(j-2);// remove left operand. warning order matter |
| break; |
| default: |
| logger.error("operator different than OR or AND???"); |
| return null;//should never happen |
| } |
| } |
| // after evaluation, the linkedList contains only one element containing the filter built |
| return (Filter)linkedList.pop(); |
| } |
| |
| |
| public boolean startScan(long transID, byte[] startRow, byte[] stopRow, |
| Object[] columns, long timestamp, |
| boolean cacheBlocks, boolean smallScanner, int numCacheRows, |
| Object[] colNamesToFilter, |
| Object[] compareOpList, |
| Object[] colValuesToCompare, |
| float dopParallelScanner, |
| float samplePercent, |
| boolean inPreFetch, |
| boolean useSnapshotScan, |
| int snapTimeout, |
| String snapName, |
| String tmpLoc, |
| int espNum, |
| int versions) |
| throws IOException { |
| |
| Scan scan; |
| |
| if (logger.isTraceEnabled()) logger.trace("Enter startScan() " + tableName + " txid: " + transID + " startRow=" |
| + ((startRow != null) ? (Bytes.equals(startRow, HConstants.EMPTY_START_ROW) ? "INFINITE" : Hex.encodeHexString(startRow)) : "NULL") |
| + " stopRow=" + ((stopRow != null) ? (Bytes.equals(stopRow, HConstants.EMPTY_START_ROW) ? "INFINITE" : Hex.encodeHexString(stopRow)) : "NULL") |
| + " CacheBlocks: " + cacheBlocks + " numCacheRows: " + numCacheRows + " Bulkread: " + useSnapshotScan); |
| |
| if (startRow != null && startRow.toString() == "") |
| startRow = null; |
| if (stopRow != null && stopRow.toString() == "") |
| stopRow = null; |
| |
| if (startRow != null && stopRow != null) |
| scan = new Scan(startRow, stopRow); |
| else |
| scan = new Scan(); |
| |
| if (versions != 0) |
| { |
| if (versions == -1) |
| scan.setMaxVersions(); |
| else if (versions == -2) |
| { |
| scan.setMaxVersions(); |
| scan.setRaw(true); |
| columns = null; |
| } |
| else if (versions > 0) |
| { |
| scan.setMaxVersions(versions); |
| } |
| } |
| |
| if (cacheBlocks == true) { |
| scan.setCacheBlocks(true); |
| } |
| else |
| scan.setCacheBlocks(false); |
| |
| scan.setSmall(smallScanner); |
| scan.setCaching(numCacheRows); |
| numRowsCached = numCacheRows; |
| if (columns != null) { |
| numColsInScan = columns.length; |
| for (int i = 0; i < columns.length ; i++) { |
| byte[] col = (byte[])columns[i]; |
| scan.addColumn(getFamily(col), getName(col)); |
| } |
| } |
| else |
| numColsInScan = 0; |
| if (colNamesToFilter != null) { |
| FilterList list; |
| boolean narrowDownResultColumns = false; //to check if we need a narrow down column filter (V2 only feature) |
| if (compareOpList == null)return false; |
| if (new String((byte[])compareOpList[0]).equals("V2")){ // are we dealing with predicate pushdown V2 |
| list = new FilterList(FilterList.Operator.MUST_PASS_ALL); |
| HashMap<String,Object> columnsToRemove = new HashMap<String,Object>(); |
| //if columnsToRemove not null, we are narrowing down using the SingleColumnValue[Exclude]Filter method |
| //else we will use the explicit FamilyFilter and QualifierFilter |
| //the simplified logic is that we can use the first method if and only if each and every column in the |
| //pushed down predicate shows up only once. |
| for (int i = 0; i < colNamesToFilter.length; i++) { |
| byte[] colName = (byte[])colNamesToFilter[i]; |
| |
| // check if the filter column is already part of the column list, if not add it if we are limiting columns (not *) |
| if(columns!=null && columns.length > 0){// if not * |
| boolean columnAlreadyIn = false; //assume column not yet in the scan object |
| for (int k=0; k<columns.length;k++){ |
| if (Arrays.equals(colName, (byte[])columns[k])){ |
| columnAlreadyIn = true;//found already exist |
| break;//no need to look further |
| } |
| } |
| if (!columnAlreadyIn){// column was not already in, so add it |
| scan.addColumn(getFamily(colName),getName(colName)); |
| narrowDownResultColumns = true; //since we added a column for predicate eval, we need to remove it later out of result set |
| String strColName = new String(colName); |
| if (columnsToRemove != null && columnsToRemove.containsKey(strColName)){// if we already added this column, it means it shows up more than once |
| columnsToRemove = null; // therefore, use the FamilyFilter/QualifierFilter method |
| }else if (columnsToRemove != null)// else |
| columnsToRemove.put(strColName,null); // add it to the list of column that should be nuked with the Exclude version of the SingleColumnValueFilter |
| } |
| } |
| } |
| if (columnsToRemove != null) |
| { //we are almost done checking if Exclude version of SingleColumnnValueFilter can be used. Th elast check s about to know if there is a IS_NULL_NULL |
| //operation that cannot be using the Exclude method, as it is transformed in a filterList with OR, therefore we cannot guaranty that the SingleColumnValueExcludeFilter |
| //performing the exclusion will be reached. |
| boolean is_null_nullFound = false; |
| for (Object o:compareOpList ){ |
| if (new String((byte[])o).equals("IS_NULL_NULL")){ |
| is_null_nullFound = true; |
| break; |
| } |
| } |
| if (is_null_nullFound){ |
| columnsToRemove = null; // disable Exclude method version of SingleColumnnValueFilter |
| }else |
| narrowDownResultColumns = false; // we will use the Exclude version of SingleColumnnValueFilter, so bypass the Family/QualifierFilter method |
| } |
| Filter f =constructV2Filter(colNamesToFilter,compareOpList,colValuesToCompare, columnsToRemove); |
| if (f==null) return false; // error logging done inside constructV2Filter |
| list.addFilter(f); |
| }//end V2 |
| else{// deal with V1 |
| list = new FilterList(FilterList.Operator.MUST_PASS_ALL); |
| |
| for (int i = 0; i < colNamesToFilter.length; i++) { |
| byte[] colName = (byte[])colNamesToFilter[i]; |
| byte[] coByte = (byte[])compareOpList[i]; |
| byte[] colVal = (byte[])colValuesToCompare[i]; |
| |
| if ((coByte == null) || (colVal == null)) { |
| return false; |
| } |
| String coStr = new String(coByte); |
| CompareOp co = CompareOp.valueOf(coStr); |
| |
| SingleColumnValueFilter filter1 = |
| new SingleColumnValueFilter(getFamily(colName), getName(colName), |
| co, colVal); |
| list.addFilter(filter1); |
| } |
| }//end V1 |
| // if we added a column for predicate eval, we need to filter down result columns |
| FilterList resultColumnsOnlyFilter = null; |
| if (narrowDownResultColumns){ |
| HashMap<String,ArrayList<byte[]>> hm = new HashMap<String,ArrayList<byte[]>>(3);//use to deal with multiple family table |
| // initialize hm with list of columns requested for output |
| for (int i=0; i<columns.length; i++){ // if we are here we know columns is not null |
| if (hm.containsKey(new String(getFamily((byte[])columns[i])))){ |
| hm.get(new String(getFamily((byte[])columns[i]))).add((byte[])columns[i]); |
| }else{ |
| ArrayList<byte[]> al = new ArrayList<byte[]>(); |
| al.add((byte[])columns[i]); |
| hm.put(new String(getFamily((byte[])columns[i])), al); |
| } |
| } |
| |
| if (hm.size()==1){//only one column family |
| resultColumnsOnlyFilter = new FilterList(FilterList.Operator.MUST_PASS_ALL); |
| if (columns.length == 1){ |
| resultColumnsOnlyFilter.addFilter(new QualifierFilter(CompareOp.EQUAL, new BinaryComparator(getName((byte[])columns[0])))); |
| }else{// more than one column |
| FilterList flColumns = new FilterList(FilterList.Operator.MUST_PASS_ONE); |
| for(int i=0; i<columns.length;i++) |
| flColumns.addFilter(new QualifierFilter(CompareOp.EQUAL, new BinaryComparator(getName((byte[])columns[i])))); |
| resultColumnsOnlyFilter.addFilter(flColumns); |
| } |
| // note the optimization puting family check at the end |
| resultColumnsOnlyFilter.addFilter(new FamilyFilter(CompareOp.EQUAL, new BinaryComparator(getFamily((byte[])columns[0])))); |
| }else{//more than one column family |
| resultColumnsOnlyFilter = new FilterList(FilterList.Operator.MUST_PASS_ONE); |
| for (Map.Entry<String,ArrayList<byte[]>> entry : hm.entrySet()){//for each column family |
| ArrayList<byte[]> alb = entry.getValue(); |
| if (alb.size() == 1){// when only one column for the family |
| resultColumnsOnlyFilter.addFilter( |
| new FilterList(FilterList.Operator.MUST_PASS_ALL, |
| new QualifierFilter(CompareOp.EQUAL, new BinaryComparator(getName(alb.get(0)))), |
| new FamilyFilter(CompareOp.EQUAL, new BinaryComparator(getFamily(alb.get(0))))) |
| ); |
| }else{// when multiple columns for the family |
| FamilyFilter familyFilter = null; |
| FilterList filterListCol = new FilterList(FilterList.Operator.MUST_PASS_ONE); |
| for(int j = 0; j<alb.size(); j++){ |
| if (familyFilter == null) |
| familyFilter = new FamilyFilter(CompareOp.EQUAL, new BinaryComparator(getFamily(alb.get(0)))); |
| filterListCol.addFilter(new QualifierFilter(CompareOp.EQUAL, new BinaryComparator(getName(alb.get(j))))); |
| } |
| resultColumnsOnlyFilter.addFilter(new FilterList(FilterList.Operator.MUST_PASS_ALL,filterListCol,familyFilter)); |
| } |
| } |
| } |
| list.addFilter(resultColumnsOnlyFilter); // add column limiting filter |
| }//end narrowDownResultColumns |
| if (samplePercent > 0.0f) |
| list.addFilter(new RandomRowFilter(samplePercent)); |
| // last optimization is making sure we remove top level filter list if it is singleton MUST_PASS_ALL filterlist |
| if (list.getFilters().size()==1){ |
| scan.setFilter(list.getFilters().get(0)); |
| if (logger.isTraceEnabled()) logger.trace("Pushed down filter:"+list.getFilters().get(0)); |
| }else{ |
| scan.setFilter(list); |
| if (logger.isTraceEnabled()) logger.trace("Pushed down filter:"+list ); |
| } |
| } else if (samplePercent > 0.0f) { |
| scan.setFilter(new RandomRowFilter(samplePercent)); |
| } |
| |
| if (!useSnapshotScan || transID != 0) |
| { |
| if (useTRexScanner && (transID != 0)) { |
| scanner = table.getScanner(transID, scan); |
| } else { |
| scanner = table.getScanner(scan,dopParallelScanner); |
| } |
| if (logger.isTraceEnabled()) logger.trace("startScan(). After getScanner. Scanner: " + scanner+ " dop:"+ |
| dopParallelScanner + "TransID " + transID + " " + useTRexScanner + " " + getTableName()); |
| } |
| else |
| { |
| snapHelper = new SnapshotScanHelper(connection.getConfiguration(), tmpLoc,snapName); |
| |
| if (logger.isTraceEnabled()) |
| logger.trace("[Snapshot Scan] HTableClient.startScan(). useSnapshotScan: " + useSnapshotScan + |
| " espNumber: " + espNum + |
| " tmpLoc: " + snapHelper.getTmpLocation() + |
| " snapshot name: " + snapHelper.getSnapshotName()); |
| |
| if (!snapHelper.snapshotExists()) |
| throw new IOException ("Snapshot " + snapHelper.getSnapshotName() + " does not exist."); |
| |
| try { |
| snapHelper.createTableSnapshotScanner(snapTimeout, 5, espNum, scan); |
| } |
| catch (InterruptedException ie) { |
| throw new IOException(ie); |
| } |
| } |
| |
| if (useSnapshotScan) |
| preFetch = false; |
| else |
| preFetch = inPreFetch; |
| if (preFetch) |
| { |
| scanHelper = new ScanHelper(); |
| future = executorService.submit(scanHelper); |
| } |
| fetchType = SCAN_FETCH; |
| if (logger.isTraceEnabled()) logger.trace("Exit startScan()."); |
| return true; |
| } |
| |
| public int startGet(long transID, byte[] rowID, |
| Object[] columns, |
| long timestamp) throws IOException { |
| |
| if (logger.isTraceEnabled()) logger.trace("Enter startGet(" + tableName + |
| " #cols: " + ((columns == null) ? 0:columns.length ) + |
| " rowID: " + new String(rowID)); |
| fetchType = GET_ROW; |
| Get get = new Get(rowID); |
| if (columns != null) |
| { |
| for (int i = 0; i < columns.length; i++) { |
| byte[] col = (byte[]) columns[i]; |
| get.addColumn(getFamily(col), getName(col)); |
| } |
| numColsInScan = columns.length; |
| } |
| else |
| numColsInScan = 0; |
| |
| Result getResult; |
| if (useTRex && (transID != 0)) { |
| getResult = table.get(transID, get); |
| } else { |
| getResult = table.get(get); |
| } |
| if (getResult == null |
| || getResult.isEmpty()) { |
| setJavaObject(jniObject); |
| return 0; |
| } |
| if (logger.isTraceEnabled()) logger.trace("startGet, result: " + getResult); |
| pushRowsToJni(getResult); |
| return 1; |
| |
| } |
| |
| // The TransactionalTable class is missing the batch get operation, |
| // so work around it. |
| private Result[] batchGet(long transactionID, List<Get> gets) |
| throws IOException { |
| if (logger.isTraceEnabled()) logger.trace("Enter batchGet(multi-row) " + tableName); |
| Result [] results = new Result[gets.size()]; |
| int i=0; |
| for (Get g : gets) { |
| Result r = table.get(transactionID, g); |
| results[i++] = r; |
| } |
| return results; |
| } |
| |
| public int startGet(long transID, Object[] rows, |
| Object[] columns, long timestamp) |
| throws IOException { |
| |
| if (logger.isTraceEnabled()) logger.trace("Enter startGet(multi-row) " + tableName); |
| |
| List<Get> listOfGets = new ArrayList<Get>(); |
| for (int i = 0; i < rows.length; i++) { |
| byte[] rowID = (byte[])rows[i]; |
| Get get = new Get(rowID); |
| listOfGets.add(get); |
| if (columns != null) |
| { |
| for (int j = 0; j < columns.length; j++ ) { |
| byte[] col = (byte[])columns[j]; |
| get.addColumn(getFamily(col), getName(col)); |
| } |
| } |
| } |
| if (columns != null) |
| numColsInScan = columns.length; |
| else |
| numColsInScan = 0; |
| if (useTRex && (transID != 0)) { |
| getResultSet = batchGet(transID, listOfGets); |
| fetchType = GET_ROW; |
| } else { |
| getResultSet = table.get(listOfGets); |
| fetchType = BATCH_GET; |
| } |
| if (getResultSet != null && getResultSet.length > 0) { |
| pushRowsToJni(getResultSet); |
| return getResultSet.length; |
| } |
| else { |
| setJavaObject(jniObject); |
| return 0; |
| } |
| } |
| |
| public int getRows(long transID, short rowIDLen, Object rowIDs, |
| Object[] columns) |
| throws IOException { |
| |
| if (logger.isTraceEnabled()) logger.trace("Enter getRows " + tableName); |
| |
| ByteBuffer bbRowIDs = (ByteBuffer)rowIDs; |
| List<Get> listOfGets = new ArrayList<Get>(); |
| short numRows = bbRowIDs.getShort(); |
| short actRowIDLen ; |
| byte rowIDSuffix; |
| byte[] rowID; |
| |
| for (int i = 0; i < numRows; i++) { |
| rowIDSuffix = bbRowIDs.get(); |
| if (rowIDSuffix == '1') |
| actRowIDLen = (short)(rowIDLen+1); |
| else |
| actRowIDLen = rowIDLen; |
| rowID = new byte[actRowIDLen]; |
| bbRowIDs.get(rowID, 0, actRowIDLen); |
| Get get = new Get(rowID); |
| listOfGets.add(get); |
| if (columns != null) { |
| for (int j = 0; j < columns.length; j++ ) { |
| byte[] col = (byte[])columns[j]; |
| get.addColumn(getFamily(col), getName(col)); |
| } |
| } |
| } |
| if (columns != null) |
| numColsInScan = columns.length; |
| else |
| numColsInScan = 0; |
| if (useTRex && (transID != 0)) { |
| getResultSet = batchGet(transID, listOfGets); |
| fetchType = GET_ROW; |
| } else { |
| getResultSet = table.get(listOfGets); |
| fetchType = BATCH_GET; |
| } |
| if (getResultSet.length != numRows) |
| throw new IOException("Number of rows retunred is not equal to requested number of rows"); |
| pushRowsToJni(getResultSet); |
| return getResultSet.length; |
| } |
| |
| public int fetchRows() throws IOException, |
| InterruptedException, ExecutionException { |
| int rowsReturned = 0; |
| |
| if (logger.isTraceEnabled()) logger.trace("Enter fetchRows(). Table: " + tableName); |
| if (getResultSet != null) |
| { |
| rowsReturned = pushRowsToJni(getResultSet); |
| getResultSet = null; |
| return rowsReturned; |
| } |
| else |
| { |
| if (scanner == null) { |
| throw new IOException("HTableClient.FetchRows() called before scanOpen()."); |
| } |
| Result[] result = null; |
| if (preFetch) |
| { |
| result = (Result[])future.get(); |
| rowsReturned = pushRowsToJni(result); |
| future = null; |
| if ((rowsReturned <= 0 || rowsReturned < numRowsCached)) |
| return rowsReturned; |
| future = executorService.submit(scanHelper); |
| } |
| else |
| { |
| result = scanner.next(numRowsCached); |
| rowsReturned = pushRowsToJni(result); |
| } |
| return rowsReturned; |
| } |
| } |
| |
| protected int pushRowsToJni(Result[] result) |
| throws IOException { |
| if (result == null || result.length == 0) |
| return 0; |
| int rowsReturned = result.length; |
| int numTotalCells = 0; |
| if (numColsInScan == 0) |
| { |
| for (int i = 0; i < result.length; i++) { |
| numTotalCells += result[i].size(); |
| } |
| } |
| else |
| // There can be maximum of 2 versions per kv |
| // So, allocate place holder to keep cell info |
| // for that many KVs |
| numTotalCells = 2 * rowsReturned * numColsInScan; |
| int numColsReturned; |
| Cell[] kvList; |
| Cell kv; |
| |
| if (kvValLen == null || |
| (kvValLen != null && numTotalCells > kvValLen.length)) |
| { |
| kvValLen = new int[numTotalCells]; |
| kvValOffset = new int[numTotalCells]; |
| kvQualLen = new int[numTotalCells]; |
| kvQualOffset = new int[numTotalCells]; |
| kvFamLen = new int[numTotalCells]; |
| kvFamOffset = new int[numTotalCells]; |
| kvTimestamp = new long[numTotalCells]; |
| kvBuffer = new byte[numTotalCells][]; |
| kvFamArray = new byte[numTotalCells][]; |
| kvQualArray = new byte[numTotalCells][]; |
| } |
| |
| if (rowIDs == null || (rowIDs != null && |
| rowsReturned > rowIDs.length)) |
| { |
| rowIDs = new byte[rowsReturned][]; |
| kvsPerRow = new int[rowsReturned]; |
| } |
| int cellNum = 0; |
| boolean colFound = false; |
| for (int rowNum = 0; rowNum < rowsReturned ; rowNum++) |
| { |
| rowIDs[rowNum] = result[rowNum].getRow(); |
| kvList = result[rowNum].rawCells(); |
| numColsReturned = kvList.length; |
| if ((cellNum + numColsReturned) > numTotalCells) |
| throw new IOException("Insufficient cell array pre-allocated"); |
| kvsPerRow[rowNum] = numColsReturned; |
| for (int colNum = 0 ; colNum < numColsReturned ; colNum++, cellNum++) |
| { |
| kv = kvList[colNum]; |
| kvValLen[cellNum] = kv.getValueLength(); |
| kvValOffset[cellNum] = kv.getValueOffset(); |
| kvQualLen[cellNum] = kv.getQualifierLength(); |
| kvQualOffset[cellNum] = kv.getQualifierOffset(); |
| kvFamLen[cellNum] = kv.getFamilyLength(); |
| kvFamOffset[cellNum] = kv.getFamilyOffset(); |
| kvTimestamp[cellNum] = kv.getTimestamp(); |
| kvBuffer[cellNum] = kv.getValueArray(); |
| kvFamArray[cellNum] = kv.getFamilyArray(); |
| kvQualArray[cellNum] = kv.getQualifierArray(); |
| colFound = true; |
| } |
| } |
| int cellsReturned; |
| if (colFound) |
| cellsReturned = cellNum++; |
| else |
| cellsReturned = 0; |
| if (cellsReturned == 0) |
| setResultInfo(jniObject, null, null, |
| null, null, null, null, |
| null, null, null, null, rowIDs, kvsPerRow, cellsReturned, rowsReturned); |
| else |
| setResultInfo(jniObject, kvValLen, kvValOffset, |
| kvQualLen, kvQualOffset, kvFamLen, kvFamOffset, |
| kvTimestamp, kvBuffer, kvFamArray, kvQualArray, rowIDs, kvsPerRow, cellsReturned, rowsReturned); |
| return rowsReturned; |
| } |
| |
| protected int pushRowsToJni(Result result) |
| throws IOException { |
| int rowsReturned = 1; |
| int numTotalCells; |
| |
| if (numColsInScan == 0) |
| numTotalCells = result.size(); |
| else |
| // There can be maximum of 2 versions per kv |
| // So, allocate place holder to keep cell info |
| // for that many KVs |
| numTotalCells = 2 * rowsReturned * numColsInScan; |
| int numColsReturned; |
| Cell[] kvList; |
| Cell kv; |
| |
| if (kvValLen == null || |
| (kvValLen != null && numTotalCells > kvValLen.length)) |
| { |
| kvValLen = new int[numTotalCells]; |
| kvValOffset = new int[numTotalCells]; |
| kvQualLen = new int[numTotalCells]; |
| kvQualOffset = new int[numTotalCells]; |
| kvFamLen = new int[numTotalCells]; |
| kvFamOffset = new int[numTotalCells]; |
| kvTimestamp = new long[numTotalCells]; |
| kvBuffer = new byte[numTotalCells][]; |
| kvFamArray = new byte[numTotalCells][]; |
| kvQualArray = new byte[numTotalCells][]; |
| } |
| if (rowIDs == null) |
| { |
| rowIDs = new byte[rowsReturned][]; |
| kvsPerRow = new int[rowsReturned]; |
| } |
| kvList = result.rawCells(); |
| if (kvList == null) |
| numColsReturned = 0; |
| else |
| numColsReturned = kvList.length; |
| if ((numColsReturned) > numTotalCells) |
| throw new IOException("Insufficient cell array pre-allocated"); |
| rowIDs[0] = result.getRow(); |
| kvsPerRow[0] = numColsReturned; |
| for (int colNum = 0 ; colNum < numColsReturned ; colNum++) |
| { |
| kv = kvList[colNum]; |
| kvValLen[colNum] = kv.getValueLength(); |
| kvValOffset[colNum] = kv.getValueOffset(); |
| kvQualLen[colNum] = kv.getQualifierLength(); |
| kvQualOffset[colNum] = kv.getQualifierOffset(); |
| kvFamLen[colNum] = kv.getFamilyLength(); |
| kvFamOffset[colNum] = kv.getFamilyOffset(); |
| kvTimestamp[colNum] = kv.getTimestamp(); |
| kvBuffer[colNum] = kv.getValueArray(); |
| kvFamArray[colNum] = kv.getFamilyArray(); |
| kvQualArray[colNum] = kv.getQualifierArray(); |
| } |
| if (numColsReturned == 0) |
| setResultInfo(jniObject, null, null, |
| null, null, null, null, |
| null, null, null, null, rowIDs, kvsPerRow, numColsReturned, rowsReturned); |
| else |
| setResultInfo(jniObject, kvValLen, kvValOffset, |
| kvQualLen, kvQualOffset, kvFamLen, kvFamOffset, |
| kvTimestamp, kvBuffer, kvFamArray, kvQualArray, rowIDs, kvsPerRow, numColsReturned, rowsReturned); |
| return rowsReturned; |
| } |
| |
| public boolean deleteRow(final long transID, byte[] rowID, |
| Object[] columns, |
| long timestamp, |
| boolean asyncOperation, |
| final boolean useRegionXn) throws IOException { |
| |
| if (logger.isTraceEnabled()) logger.trace("Enter deleteRow transID " + transID |
| + " (" + new String(rowID) + ", " + timestamp + ") " + tableName); |
| |
| final Delete del; |
| if (timestamp == -1) |
| del = new Delete(rowID); |
| else |
| del = new Delete(rowID, timestamp); |
| |
| if (columns != null) { |
| for (int i = 0; i < columns.length ; i++) { |
| byte[] col = (byte[]) columns[i]; |
| del.deleteColumns(getFamily(col), getName(col)); |
| } |
| } |
| if (asyncOperation) { |
| future = executorService.submit(new Callable() { |
| public Object call() throws IOException { |
| boolean res = true; |
| if (useTRex && (transID != 0)) { |
| table.delete(transID, del); |
| } |
| else if (useRegionXn){ |
| table.deleteRegionTx(del, /* auto-commit */ true); |
| } |
| else { |
| table.delete(del); |
| } |
| return new Boolean(res); |
| } |
| }); |
| return true; |
| } |
| else { |
| if (useTRex && (transID != 0)) { |
| table.delete(transID, del); |
| } |
| else if (useRegionXn){ |
| table.deleteRegionTx(del, /* auto-commit */ true); |
| } |
| else { |
| table.delete(del); |
| } |
| } |
| if (logger.isTraceEnabled()) logger.trace("Exit deleteRow"); |
| return true; |
| } |
| |
| public boolean deleteRows(final long transID, short rowIDLen, Object rowIDs, |
| long timestamp, |
| boolean asyncOperation) throws IOException { |
| |
| if (logger.isTraceEnabled()) logger.trace("Enter deleteRowsInt() transID " |
| + transID + " " + tableName); |
| |
| final List<Delete> listOfDeletes = new ArrayList<Delete>(); |
| listOfDeletes.clear(); |
| ByteBuffer bbRowIDs = (ByteBuffer)rowIDs; |
| short numRows = bbRowIDs.getShort(); |
| byte[] rowID; |
| byte rowIDSuffix; |
| short actRowIDLen; |
| |
| for (short rowNum = 0; rowNum < numRows; rowNum++) { |
| rowIDSuffix = bbRowIDs.get(); |
| if (rowIDSuffix == '1') |
| actRowIDLen = (short)(rowIDLen+1); |
| else |
| actRowIDLen = rowIDLen; |
| rowID = new byte[actRowIDLen]; |
| bbRowIDs.get(rowID, 0, actRowIDLen); |
| |
| Delete del; |
| if (timestamp == -1) |
| del = new Delete(rowID); |
| else |
| del = new Delete(rowID, timestamp); |
| listOfDeletes.add(del); |
| } |
| if (asyncOperation) { |
| future = executorService.submit(new Callable() { |
| public Object call() throws IOException { |
| boolean res = true; |
| if (useTRex && (transID != 0)) |
| table.delete(transID, listOfDeletes); |
| else |
| table.delete(listOfDeletes); |
| return new Boolean(res); |
| } |
| }); |
| return true; |
| } |
| else { |
| if (useTRex && (transID != 0)) |
| table.delete(transID, listOfDeletes); |
| else |
| table.delete(listOfDeletes); |
| } |
| if (logger.isTraceEnabled()) logger.trace("Exit deleteRows"); |
| return true; |
| } |
| |
| public byte[] intToByteArray(int value) { |
| return new byte[] { |
| (byte)(value >>> 24), |
| (byte)(value >>> 16), |
| (byte)(value >>> 8), |
| (byte)value}; |
| } |
| |
| public boolean checkAndDeleteRow(long transID, byte[] rowID, |
| byte[] columnToCheck, byte[] colValToCheck, |
| long timestamp, final boolean useRegionXn) throws IOException { |
| |
| if (logger.isTraceEnabled()) logger.trace("Enter checkAndDeleteRow transID " + transID |
| + " (" + new String(rowID) + ", " + new String(columnToCheck) + ", " + new String(colValToCheck) + ", " + timestamp + ") " + tableName); |
| Delete del; |
| if (timestamp == -1) |
| del = new Delete(rowID); |
| else |
| del = new Delete(rowID, timestamp); |
| |
| byte[] family = null; |
| byte[] qualifier = null; |
| |
| if (columnToCheck.length > 0) { |
| family = getFamily(columnToCheck); |
| qualifier = getName(columnToCheck); |
| } |
| |
| boolean res; |
| if (useTRex && (transID != 0)) { |
| res = table.checkAndDelete(transID, rowID, family, qualifier, colValToCheck, del); |
| } |
| else if (useRegionXn){ |
| res = table.checkAndDeleteRegionTx(rowID, family, qualifier, colValToCheck, |
| del, /* autoCommit */ true); |
| } |
| else { |
| res = table.checkAndDelete(rowID, family, qualifier, colValToCheck, del); |
| } |
| |
| if (res == false) |
| return false; |
| return true; |
| } |
| |
| public boolean putRow(final long transID, final byte[] rowID, Object row, |
| byte[] columnToCheck, final byte[] colValToCheck, |
| final short colIndexToCheck, |
| final boolean checkAndPut, boolean asyncOperation, |
| final boolean useRegionXn) throws IOException, InterruptedException, |
| ExecutionException |
| { |
| if (logger.isTraceEnabled()) logger.trace("Enter putRow() " + tableName + |
| " transID: " + transID + |
| " useTRex: " + useTRex + |
| " useRegionXn: " + useRegionXn); |
| |
| final Put put; |
| ByteBuffer bb; |
| short numCols; |
| short colNameLen; |
| int colValueLen; |
| byte[] family = null; |
| byte[] qualifier = null; |
| byte[] colName, colValue; |
| |
| bb = (ByteBuffer)row; |
| put = new Put(rowID); |
| numCols = bb.getShort(); |
| for (short colIndex = 0; colIndex < numCols; colIndex++) |
| { |
| colNameLen = bb.getShort(); |
| colName = new byte[colNameLen]; |
| bb.get(colName, 0, colNameLen); |
| colValueLen = bb.getInt(); |
| colValue = new byte[colValueLen]; |
| bb.get(colValue, 0, colValueLen); |
| put.add(getFamily(colName), getName(colName), colValue); |
| if (checkAndPut && colIndex == colIndexToCheck) { |
| family = getFamily(colName); |
| qualifier = getName(colName); |
| } |
| } |
| if (columnToCheck != null && columnToCheck.length > 0) { |
| family = getFamily(columnToCheck); |
| qualifier = getName(columnToCheck); |
| } |
| final byte[] family1 = family; |
| final byte[] qualifier1 = qualifier; |
| if (asyncOperation) { |
| future = executorService.submit(new Callable() { |
| public Object call() throws IOException { |
| boolean res = true; |
| |
| if (checkAndPut) { |
| if (useTRex && (transID != 0)){ |
| res = table.checkAndPut(transID, rowID, |
| family1, qualifier1, colValToCheck, put); |
| } |
| else if (useRegionXn){ |
| if (logger.isTraceEnabled()) logger.trace("checkAndPutRegionTx with regionTX "); |
| res = table.checkAndPutRegionTx(rowID, |
| family1, qualifier1, colValToCheck, put, /* auto-commit */ true); |
| |
| } |
| else { |
| res = table.checkAndPut(rowID, |
| family1, qualifier1, colValToCheck, put); |
| } |
| } |
| else { |
| if (useTRex && (transID != 0)){ |
| table.put(transID, put); |
| } |
| else if (useRegionXn){ |
| if (logger.isTraceEnabled()) logger.trace("putRow using putRegionTx"); |
| table.putRegionTx(put, /* auto-commit */ true); |
| }else{ |
| table.put(put); |
| } |
| } |
| return new Boolean(res); |
| } |
| }); |
| return true; |
| } else { |
| boolean result = true; |
| if (checkAndPut) { |
| if (useTRex && (transID != 0)){ |
| result = table.checkAndPut(transID, rowID, |
| family1, qualifier1, colValToCheck, put); |
| } |
| else if (useRegionXn){ |
| if (logger.isTraceEnabled()) logger.trace("checkAndPutRegionTx using regionTX "); |
| result = table.checkAndPutRegionTx(rowID, family1, qualifier1, |
| colValToCheck, put, /* auto-commit */ true); |
| } |
| else { |
| result = table.checkAndPut(rowID, |
| family1, qualifier1, colValToCheck, put); |
| } |
| } |
| else { |
| if (useTRex && (transID != 0)){ |
| table.put(transID, put); |
| } |
| else if (useRegionXn){ |
| if (logger.isTraceEnabled()) logger.trace("putRow using putRegionTx"); |
| table.putRegionTx(put, true /* also commit */); |
| }else{ |
| table.put(put); |
| } |
| } |
| return result; |
| } |
| } |
| |
| /* public boolean insertRow(long transID, byte[] rowID, |
| Object row, |
| long timestamp, |
| boolean asyncOperation) throws IOException, InterruptedException, ExecutionException { |
| return putRow(transID, rowID, row, null, null, 0, |
| false, asyncOperation, false); |
| } */ |
| |
| public boolean putRows(final long transID, short rowIDLen, Object rowIDs, |
| Object rows, |
| long timestamp, boolean asyncOperation) |
| throws IOException, InterruptedException, ExecutionException { |
| |
| if (logger.isTraceEnabled()) logger.trace("Enter putRows() " + tableName + |
| " transID: " + transID + |
| " useTRex: " + useTRex); |
| Put put; |
| ByteBuffer bbRows, bbRowIDs; |
| short numCols, numRows; |
| short colNameLen; |
| int colValueLen; |
| byte[] colName, colValue, rowID; |
| byte rowIDSuffix; |
| short actRowIDLen; |
| bbRowIDs = (ByteBuffer)rowIDs; |
| bbRows = (ByteBuffer)rows; |
| |
| final List<Put> listOfPuts = new ArrayList<Put>(); |
| numRows = bbRowIDs.getShort(); |
| |
| for (short rowNum = 0; rowNum < numRows; rowNum++) { |
| rowIDSuffix = bbRowIDs.get(); |
| if (rowIDSuffix == '1') |
| actRowIDLen = (short)(rowIDLen+1); |
| else |
| actRowIDLen = rowIDLen; |
| rowID = new byte[actRowIDLen]; |
| bbRowIDs.get(rowID, 0, actRowIDLen); |
| put = new Put(rowID); |
| numCols = bbRows.getShort(); |
| for (short colIndex = 0; colIndex < numCols; colIndex++) |
| { |
| colNameLen = bbRows.getShort(); |
| colName = new byte[colNameLen]; |
| bbRows.get(colName, 0, colNameLen); |
| colValueLen = bbRows.getInt(); |
| colValue = new byte[colValueLen]; |
| bbRows.get(colValue, 0, colValueLen); |
| put.add(getFamily(colName), getName(colName), colValue); |
| } |
| if (writeToWAL) |
| put.setWriteToWAL(writeToWAL); |
| listOfPuts.add(put); |
| } |
| if (asyncOperation) { |
| future = executorService.submit(new Callable() { |
| public Object call() throws IOException { |
| boolean res = true; |
| if (useTRex && (transID != 0)) |
| table.put(transID, listOfPuts); |
| else |
| table.put(listOfPuts); |
| return new Boolean(res); |
| } |
| }); |
| } |
| else { |
| if (useTRex && (transID != 0)) |
| table.put(transID, listOfPuts); |
| else |
| table.put(listOfPuts); |
| } |
| return true; |
| } |
| |
| public boolean completeAsyncOperation(int timeout, boolean resultArray[]) |
| throws InterruptedException, ExecutionException |
| { |
| if (timeout == -1) { |
| if (! future.isDone()) |
| return false; |
| } |
| try { |
| Boolean result = (Boolean)future.get(timeout, TimeUnit.MILLISECONDS); |
| // Need to enhance to return the result |
| // for each Put object |
| for (int i = 0; i < resultArray.length; i++) |
| resultArray[i] = result.booleanValue(); |
| future = null; |
| } catch(TimeoutException te) { |
| return false; |
| } |
| return true; |
| } |
| |
| /* public boolean checkAndInsertRow(long transID, byte[] rowID, |
| Object row, |
| long timestamp, |
| boolean asyncOperation) throws IOException, InterruptedException, ExecutionException { |
| return putRow(transID, rowID, row, null, null, 0, |
| true, asyncOperation, false); |
| } */ |
| |
| public boolean checkAndUpdateRow(long transID, byte[] rowID, |
| Object columns, byte[] columnToCheck, byte[] colValToCheck, |
| long timestamp, boolean asyncOperation) throws IOException, InterruptedException, |
| ExecutionException, Throwable { |
| short colIndexToCheck = 0; // overridden by columnToCheck |
| return putRow(transID, rowID, columns, columnToCheck, |
| colValToCheck, colIndexToCheck, |
| true, asyncOperation, false); |
| } |
| |
| public byte[] coProcAggr(long transID, int aggrType, |
| byte[] startRowID, |
| byte[] stopRowID, byte[] colFamily, byte[] colName, |
| boolean cacheBlocks, int numCacheRows) |
| throws IOException, Throwable { |
| |
| Configuration customConf = connection.getConfiguration(); |
| long rowCount = 0; |
| |
| if (transID > 0) { |
| TransactionalAggregationClient aggregationClient = |
| new TransactionalAggregationClient(customConf, connection); |
| Scan scan = new Scan(); |
| scan.addFamily(colFamily); |
| scan.setCacheBlocks(false); |
| final ColumnInterpreter<Long, Long, EmptyMsg, LongMsg, LongMsg> ci = |
| new LongColumnInterpreter(); |
| byte[] tname = getTableName().getBytes(); |
| TransactionalTable lv_ttable = new TransactionalTable(getTableName(), connection); |
| TransactionState ts = table.registerTransaction(lv_ttable, transID, startRowID); |
| |
| rowCount = aggregationClient.rowCount(transID, ts.getStartId(), |
| org.apache.hadoop.hbase.TableName.valueOf(getTableName()), |
| ci, |
| scan); |
| } |
| else { |
| AggregationClient aggregationClient = |
| new AggregationClient(customConf); |
| Scan scan = new Scan(); |
| scan.addFamily(colFamily); |
| scan.setCacheBlocks(false); |
| final ColumnInterpreter<Long, Long, EmptyMsg, LongMsg, LongMsg> ci = |
| new LongColumnInterpreter(); |
| byte[] tname = getTableName().getBytes(); |
| rowCount = aggregationClient.rowCount( |
| org.apache.hadoop.hbase.TableName.valueOf(getTableName()), |
| ci, |
| scan); |
| } |
| |
| byte[] rcBytes = |
| ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN).putLong(rowCount).array(); |
| return rcBytes; |
| } |
| |
| |
| public boolean release(boolean cleanJniObject) throws IOException { |
| |
| boolean retcode = false; |
| // Complete the pending IO |
| if (future != null) { |
| try { |
| future.get(30, TimeUnit.SECONDS); |
| } catch(TimeoutException e) { |
| logger.error("Asynchronous Thread is Cancelled (timeout), " + e); |
| retcode = true; |
| future.cancel(true); // Interrupt the thread |
| } catch(InterruptedException e) { |
| logger.error("Asynchronous Thread is Cancelled (interrupt), " + e); |
| retcode = true; |
| future.cancel(true); // Interrupt the thread |
| } catch (ExecutionException ee) |
| { |
| } |
| future = null; |
| } |
| if (scanner != null) { |
| if (logger.isTraceEnabled()) logger.trace("scanner.close() " + tableName + " " + scanner + " " |
| + retcode ); |
| scanner.close(); |
| scanner = null; |
| } |
| if (snapHelper !=null) |
| { |
| snapHelper.release(); |
| snapHelper = null; |
| } |
| cleanScan(); |
| getResultSet = null; |
| if (cleanJniObject) { |
| if (jniObject != 0) |
| cleanup(jniObject); |
| tableName = null; |
| } |
| scanHelper = null; |
| jniObject = 0; |
| table.close(); |
| return retcode; |
| } |
| |
| public boolean close(boolean clearRegionCache, boolean cleanJniObject) throws IOException { |
| if (logger.isTraceEnabled()) logger.trace("Enter close() " + tableName); |
| if (table != null) |
| { |
| /* |
| if (clearRegionCache) |
| { |
| connection.clearRegionCache(tableName.getBytes()); |
| } |
| */ |
| table.close(); |
| table = null; |
| } |
| return true; |
| } |
| |
| public byte[][] getStartKeys() throws IOException |
| { |
| return table.getStartKeys(); |
| } |
| |
| public byte[][] getEndKeys() throws IOException |
| { |
| return table.getEndKeys(); |
| } |
| |
| private void cleanScan() |
| { |
| if (fetchType == GET_ROW || fetchType == BATCH_GET) |
| return; |
| numRowsCached = 1; |
| numColsInScan = 0; |
| kvValLen = null; |
| kvValOffset = null; |
| kvQualLen = null; |
| kvQualOffset = null; |
| kvFamLen = null; |
| kvFamOffset = null; |
| kvTimestamp = null; |
| kvBuffer = null; |
| rowIDs = null; |
| kvsPerRow = null; |
| } |
| |
| protected void setJniObject(long inJniObject) { |
| jniObject = inJniObject; |
| } |
| |
| private native int setResultInfo(long jniObject, |
| int[] kvValLen, int[] kvValOffset, |
| int[] kvQualLen, int[] kvQualOffset, |
| int[] kvFamLen, int[] kvFamOffset, |
| long[] timestamp, |
| byte[][] kvBuffer, |
| byte[][] kvFamArray, |
| byte[][] kvQualArray, |
| byte[][] rowIDs, |
| int[] kvsPerRow, int numCellsReturned, |
| int rowsReturned); |
| |
| private native void cleanup(long jniObject); |
| |
| protected native int setJavaObject(long jniObject); |
| |
| static { |
| envUseTRex = true; |
| envUseTRexScanner = true; |
| String useTransactions = System.getenv("USE_TRANSACTIONS"); |
| if (useTransactions != null) { |
| int lv_useTransactions = (Integer.parseInt(useTransactions)); |
| if (lv_useTransactions == 0) |
| envUseTRex = false; |
| } |
| String useTransactionsScanner = System.getenv("USE_TRANSACTIONS_SCANNER"); |
| if (useTransactionsScanner != null) { |
| int lv_useTransactionsScanner = (Integer.parseInt(useTransactionsScanner)); |
| if (lv_useTransactionsScanner == 0) |
| envUseTRexScanner = false; |
| } |
| executorService = Executors.newCachedThreadPool(); |
| System.loadLibrary("executor"); |
| } |
| } |
