You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2017/06/02 23:45:31 UTC
[1/2] phoenix git commit: PHOENIX-3744 Support snapshot scanners for
MR-based Non-aggregate queries (Akshita Malhotra)
Repository: phoenix
Updated Branches:
refs/heads/master ed77c36b0 -> faadb5448
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/iterate/SnapshotScanner.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SnapshotScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SnapshotScanner.java
new file mode 100644
index 0000000..31f04f2
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SnapshotScanner.java
@@ -0,0 +1,165 @@
+package org.apache.phoenix.iterate;
+
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.*;
+import org.apache.hadoop.hbase.client.*;
+
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.regionserver.*;
+import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.util.*;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+
+public class SnapshotScanner extends AbstractClientScanner {
+
+ private static final Log LOG = LogFactory.getLog(SnapshotScanner.class);
+
+ private RegionScanner scanner = null;
+ private HRegion region;
+ List<Cell> values;
+
+ public SnapshotScanner(Configuration conf, FileSystem fs, Path rootDir,
+ HTableDescriptor htd, HRegionInfo hri, Scan scan) throws Throwable{
+
+ scan.setIsolationLevel(IsolationLevel.READ_UNCOMMITTED);
+ values = new ArrayList<>();
+ this.region = HRegion.openHRegion(conf, fs, rootDir, hri, htd, null, null, null);
+
+ // process the region scanner for non-aggregate queries
+ PTable.QualifierEncodingScheme encodingScheme = EncodedColumnsUtil.getQualifierEncodingScheme(scan);
+ boolean useNewValueColumnQualifier = EncodedColumnsUtil.useNewValueColumnQualifier(scan);
+
+ RegionCoprocessorEnvironment snapshotEnv = getSnapshotContextEnvironment(conf);
+
+ RegionScannerFactory regionScannerFactory;
+ if (scan.getAttribute(BaseScannerRegionObserver.NON_AGGREGATE_QUERY) != null) {
+ regionScannerFactory = new NonAggregateRegionScannerFactory(snapshotEnv, useNewValueColumnQualifier, encodingScheme);
+ } else {
+ /* future work : Snapshot M/R jobs for aggregate queries*/
+ throw new UnsupportedOperationException("Snapshot M/R jobs not available for aggregate queries");
+ }
+
+ this.scanner = regionScannerFactory.getRegionScanner(scan, region.getScanner(scan));
+ region.startRegionOperation();
+ }
+
+
+ @Override
+ public Result next() throws IOException {
+ values.clear();
+ scanner.nextRaw(values);
+ if (values.isEmpty()) {
+ //we are done
+ return null;
+ }
+
+ return Result.create(values);
+ }
+
+ @Override
+ public void close() {
+ if (this.scanner != null) {
+ try {
+ this.scanner.close();
+ this.scanner = null;
+ } catch (IOException e) {
+ LOG.warn("Exception while closing scanner", e);
+ }
+ }
+ if (this.region != null) {
+ try {
+ this.region.closeRegionOperation();
+ this.region.close(true);
+ this.region = null;
+ } catch (IOException e) {
+ LOG.warn("Exception while closing scanner", e);
+ }
+ }
+ }
+
+ @Override
+ public boolean renewLease() {
+ return false;
+ }
+
+ private RegionCoprocessorEnvironment getSnapshotContextEnvironment(final Configuration conf) {
+ return new RegionCoprocessorEnvironment() {
+ @Override
+ public Region getRegion() {
+ return region;
+ }
+
+ @Override
+ public HRegionInfo getRegionInfo() {
+ return region.getRegionInfo();
+ }
+
+ @Override
+ public RegionServerServices getRegionServerServices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ConcurrentMap<String, Object> getSharedData() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getVersion() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getHBaseVersion() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Coprocessor getInstance() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getPriority() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getLoadSequence() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ return conf;
+ }
+
+ @Override
+ public HTableInterface getTable(TableName tableName) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public HTableInterface getTable(TableName tableName, ExecutorService executorService)
+ throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
new file mode 100644
index 0000000..a3f6224
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
@@ -0,0 +1,138 @@
+package org.apache.phoenix.iterate;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.snapshot.RestoreSnapshotHelper;
+import org.apache.hadoop.hbase.util.FSUtils;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.monitoring.ScanMetricsHolder;
+import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.util.ServerUtil;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+public class TableSnapshotResultIterator implements ResultIterator {
+
+ private final Scan scan;
+ private ResultIterator scanIterator;
+ private Configuration configuration;
+ private final ScanMetricsHolder scanMetricsHolder;
+ private Tuple lastTuple = null;
+ private static final ResultIterator UNINITIALIZED_SCANNER = ResultIterator.EMPTY_ITERATOR;
+ private ArrayList<HRegionInfo> regions;
+ private HTableDescriptor htd;
+ private String snapshotName;
+
+ private Path restoreDir;
+ private Path rootDir;
+ private FileSystem fs;
+ private int currentRegion;
+ private boolean closed = false;
+
+ public TableSnapshotResultIterator(Configuration configuration, Scan scan, ScanMetricsHolder scanMetricsHolder)
+ throws IOException {
+ this.configuration = configuration;
+ this.currentRegion = -1;
+ this.scan = scan;
+ this.scanMetricsHolder = scanMetricsHolder;
+ this.scanIterator = UNINITIALIZED_SCANNER;
+ this.restoreDir = new Path(configuration.get(PhoenixConfigurationUtil.RESTORE_DIR_KEY));
+ this.snapshotName = configuration.get(
+ PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY);
+ this.rootDir = FSUtils.getRootDir(configuration);
+ this.fs = rootDir.getFileSystem(configuration);
+ init();
+ }
+
+ private void init() throws IOException {
+ RestoreSnapshotHelper.RestoreMetaChanges meta =
+ RestoreSnapshotHelper.copySnapshotForScanner(this.configuration, this.fs,
+ this.rootDir, this.restoreDir, this.snapshotName);
+ List restoredRegions = meta.getRegionsToAdd();
+ this.htd = meta.getTableDescriptor();
+ this.regions = new ArrayList(restoredRegions.size());
+ Iterator i$ = restoredRegions.iterator();
+
+ while(i$.hasNext()) {
+ HRegionInfo hri = (HRegionInfo)i$.next();
+ if(CellUtil.overlappingKeys(this.scan.getStartRow(), this.scan.getStopRow(),
+ hri.getStartKey(), hri.getEndKey())) {
+ this.regions.add(hri);
+ }
+ }
+
+ Collections.sort(this.regions);
+ }
+
+ public boolean initSnapshotScanner() throws SQLException {
+ if (closed) {
+ return true;
+ }
+ ResultIterator delegate = this.scanIterator;
+ if (delegate == UNINITIALIZED_SCANNER) {
+ ++this.currentRegion;
+ if (this.currentRegion >= this.regions.size())
+ return false;
+ try {
+ HRegionInfo hri = regions.get(this.currentRegion);
+ this.scanIterator =
+ new ScanningResultIterator(new SnapshotScanner(configuration, fs, restoreDir, htd, hri, scan),
+ scan, scanMetricsHolder);
+ } catch (Throwable e) {
+ throw ServerUtil.parseServerException(e);
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public Tuple next() throws SQLException {
+ while (true) {
+ if (!initSnapshotScanner())
+ return null;
+ try {
+ lastTuple = scanIterator.next();
+ if (lastTuple != null) {
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ lastTuple.getKey(ptr);
+ return lastTuple;
+ }
+ } finally {
+ if (lastTuple == null) {
+ scanIterator.close();
+ scanIterator = UNINITIALIZED_SCANNER;
+ }
+ }
+ }
+ }
+
+ @Override
+ public void close() throws SQLException {
+ closed = true; // ok to say closed even if the below code throws an exception
+ try {
+ scanIterator.close();
+ fs.delete(this.restoreDir, true);
+ } catch (IOException e) {
+ throw ServerUtil.parseServerException(e);
+ } finally {
+ scanIterator = UNINITIALIZED_SCANNER;
+ }
+ }
+
+ @Override
+ public void explain(List<String> planSteps) {
+ // noop
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
index d4d6734..ec1b451 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
@@ -35,15 +35,9 @@ import org.apache.hadoop.util.ReflectionUtils;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
-import org.apache.phoenix.iterate.ConcatResultIterator;
-import org.apache.phoenix.iterate.LookAheadResultIterator;
-import org.apache.phoenix.iterate.MapReduceParallelScanGrouper;
-import org.apache.phoenix.iterate.PeekingResultIterator;
-import org.apache.phoenix.iterate.ResultIterator;
-import org.apache.phoenix.iterate.RoundRobinResultIterator;
-import org.apache.phoenix.iterate.SequenceResultIterator;
-import org.apache.phoenix.iterate.TableResultIterator;
+import org.apache.phoenix.iterate.*;
import org.apache.phoenix.jdbc.PhoenixResultSet;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
import org.apache.phoenix.monitoring.ReadMetricQueue;
import org.apache.phoenix.monitoring.ScanMetricsHolder;
import org.apache.phoenix.query.ConnectionQueryServices;
@@ -110,6 +104,7 @@ public class PhoenixRecordReader<T extends DBWritable> extends RecordReader<Null
StatementContext ctx = queryPlan.getContext();
ReadMetricQueue readMetrics = ctx.getReadMetricsQueue();
String tableName = queryPlan.getTableRef().getTable().getPhysicalName().getString();
+ String snapshotName = this.configuration.get(PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY);
// Clear the table region boundary cache to make sure long running jobs stay up to date
byte[] tableNameBytes = queryPlan.getTableRef().getTable().getPhysicalName().getBytes();
@@ -121,15 +116,25 @@ public class PhoenixRecordReader<T extends DBWritable> extends RecordReader<Null
for (Scan scan : scans) {
// For MR, skip the region boundary check exception if we encounter a split. ref: PHOENIX-2599
scan.setAttribute(BaseScannerRegionObserver.SKIP_REGION_BOUNDARY_CHECK, Bytes.toBytes(true));
+
+ PeekingResultIterator peekingResultIterator;
ScanMetricsHolder scanMetricsHolder =
- ScanMetricsHolder.getInstance(readMetrics, tableName, scan,
- isRequestMetricsEnabled);
- final TableResultIterator tableResultIterator =
- new TableResultIterator(
- queryPlan.getContext().getConnection().getMutationState(), scan,
- scanMetricsHolder, renewScannerLeaseThreshold, queryPlan,
- MapReduceParallelScanGrouper.getInstance());
- PeekingResultIterator peekingResultIterator = LookAheadResultIterator.wrap(tableResultIterator);
+ ScanMetricsHolder.getInstance(readMetrics, tableName, scan,
+ isRequestMetricsEnabled);
+ if (snapshotName != null) {
+ // result iterator to read snapshots
+ final TableSnapshotResultIterator tableSnapshotResultIterator = new TableSnapshotResultIterator(configuration, scan,
+ scanMetricsHolder);
+ peekingResultIterator = LookAheadResultIterator.wrap(tableSnapshotResultIterator);
+ } else {
+ final TableResultIterator tableResultIterator =
+ new TableResultIterator(
+ queryPlan.getContext().getConnection().getMutationState(), scan,
+ scanMetricsHolder, renewScannerLeaseThreshold, queryPlan,
+ MapReduceParallelScanGrouper.getInstance());
+ peekingResultIterator = LookAheadResultIterator.wrap(tableResultIterator);
+ }
+
iterators.add(peekingResultIterator);
}
ResultIterator iterator = queryPlan.useRoundRobinIterator() ? RoundRobinResultIterator.newIterator(iterators, queryPlan) : ConcatResultIterator.newIterator(iterators);
@@ -139,13 +144,14 @@ public class PhoenixRecordReader<T extends DBWritable> extends RecordReader<Null
this.resultIterator = iterator;
// Clone the row projector as it's not thread safe and would be used simultaneously by
// multiple threads otherwise.
+
this.resultSet = new PhoenixResultSet(this.resultIterator, queryPlan.getProjector().cloneIfNecessary(), queryPlan.getContext());
} catch (SQLException e) {
LOG.error(String.format(" Error [%s] initializing PhoenixRecordReader. ",e.getMessage()));
Throwables.propagate(e);
}
}
-
+
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
if (key == null) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
index 1d2cbbe..1302f85 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
@@ -111,6 +111,10 @@ public final class PhoenixConfigurationUtil {
public static final boolean DEFAULT_SPLIT_BY_STATS = true;
+ public static final String SNAPSHOT_NAME_KEY = "phoenix.mapreduce.snapshot.name";
+
+ public static final String RESTORE_DIR_KEY = "phoenix.tableSnapshot.restore.dir";
+
public enum SchemaType {
TABLE,
QUERY;
@@ -192,6 +196,18 @@ public final class PhoenixConfigurationUtil {
public static void setUpsertColumnNames(final Configuration configuration,final String[] columns) {
setValues(configuration, columns, MAPREDUCE_UPSERT_COLUMN_COUNT, MAPREDUCE_UPSERT_COLUMN_VALUE_PREFIX);
}
+
+ public static void setSnapshotNameKey(final Configuration configuration, final String snapshotName) {
+ Preconditions.checkNotNull(configuration);
+ Preconditions.checkNotNull(snapshotName);
+ configuration.set(SNAPSHOT_NAME_KEY, snapshotName);
+ }
+
+ public static void setRestoreDirKey(final Configuration configuration, final String restoreDir) {
+ Preconditions.checkNotNull(configuration);
+ Preconditions.checkNotNull(restoreDir);
+ configuration.set(RESTORE_DIR_KEY, restoreDir);
+ }
public static List<String> getUpsertColumnNames(final Configuration configuration) {
return getValues(configuration, MAPREDUCE_UPSERT_COLUMN_COUNT, MAPREDUCE_UPSERT_COLUMN_VALUE_PREFIX);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
index 98f0364..b0981ef 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
@@ -18,12 +18,16 @@
package org.apache.phoenix.mapreduce.util;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.mapreduce.PhoenixOutputFormat;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.SchemaType;
+import java.io.IOException;
+import java.util.UUID;
+
/**
* Utility class for setting Configuration parameters for the Map Reduce job
*/
@@ -63,6 +67,66 @@ public final class PhoenixMapReduceUtil {
PhoenixConfigurationUtil.setSchemaType(configuration, SchemaType.QUERY);
}
+ /**
+ *
+ * @param job
+ * @param inputClass DBWritable class
+ * @param snapshotName The name of a snapshot (of a table) to read from
+ * @param tableName Input table name
+ * @param restoreDir a temporary dir to copy the snapshot files into
+ * @param conditions Condition clause to be added to the WHERE clause. Can be <tt>null</tt> if there are no conditions.
+ * @param fieldNames fields being projected for the SELECT query.
+ */
+ public static void setInput(final Job job, final Class<? extends DBWritable> inputClass, final String snapshotName, String tableName,
+ Path restoreDir, final String conditions, final String... fieldNames) throws
+ IOException {
+ final Configuration configuration = setSnapshotInput(job, inputClass, snapshotName, tableName, restoreDir);
+ if(conditions != null) {
+ PhoenixConfigurationUtil.setInputTableConditions(configuration, conditions);
+ }
+ PhoenixConfigurationUtil.setSelectColumnNames(configuration, fieldNames);
+ }
+
+ /**
+ *
+ * @param job
+ * @param inputClass DBWritable class
+ * @param snapshotName The name of a snapshot (of a table) to read from
+ * @param tableName Input table name
+ * @param restoreDir a temporary dir to copy the snapshot files into
+ * @param inputQuery The select query
+ */
+ public static void setInput(final Job job, final Class<? extends DBWritable> inputClass, final String snapshotName, String tableName,
+ Path restoreDir, String inputQuery) throws
+ IOException {
+ final Configuration configuration = setSnapshotInput(job, inputClass, snapshotName, tableName, restoreDir);
+ if(inputQuery != null) {
+ PhoenixConfigurationUtil.setInputQuery(configuration, inputQuery);
+ }
+
+ }
+
+ /**
+ *
+ * @param job
+ * @param inputClass DBWritable class
+ * @param snapshotName The name of a snapshot (of a table) to read from
+ * @param tableName Input table name
+ * @param restoreDir a temporary dir to copy the snapshot files into
+ */
+ private static Configuration setSnapshotInput(Job job, Class<? extends DBWritable> inputClass, String snapshotName,
+ String tableName, Path restoreDir) {
+ job.setInputFormatClass(PhoenixInputFormat.class);
+ final Configuration configuration = job.getConfiguration();
+ PhoenixConfigurationUtil.setInputClass(configuration, inputClass);
+ PhoenixConfigurationUtil.setSnapshotNameKey(configuration, snapshotName);
+ PhoenixConfigurationUtil.setInputTableName(configuration, tableName);
+
+ PhoenixConfigurationUtil.setRestoreDirKey(configuration, new Path(restoreDir, UUID.randomUUID().toString()).toString());
+ PhoenixConfigurationUtil.setSchemaType(configuration, SchemaType.QUERY);
+ return configuration;
+ }
+
private static Configuration setInput(final Job job, final Class<? extends DBWritable> inputClass, final String tableName){
job.setInputFormatClass(PhoenixInputFormat.class);
final Configuration configuration = job.getConfiguration();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
index 986debd..2f65647 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
@@ -48,7 +48,6 @@ import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
-import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;
@@ -504,7 +503,7 @@ public class IndexUtil {
return QueryUtil.getViewStatement(index.getSchemaName().getString(), index.getTableName().getString(), buf.toString());
}
- public static void wrapResultUsingOffset(final ObserverContext<RegionCoprocessorEnvironment> c,
+ public static void wrapResultUsingOffset(final RegionCoprocessorEnvironment environment,
List<Cell> result, final int offset, ColumnReference[] dataColumns,
TupleProjector tupleProjector, Region dataRegion, IndexMaintainer indexMaintainer,
byte[][] viewConstants, ImmutableBytesWritable ptr) throws IOException {
@@ -529,11 +528,11 @@ public class IndexUtil {
joinResult = dataRegion.get(get);
} else {
TableName dataTable =
- TableName.valueOf(MetaDataUtil.getLocalIndexUserTableName(c.getEnvironment()
- .getRegion().getTableDesc().getNameAsString()));
+ TableName.valueOf(MetaDataUtil.getLocalIndexUserTableName(
+ environment.getRegion().getTableDesc().getNameAsString()));
HTableInterface table = null;
try {
- table = c.getEnvironment().getTable(dataTable);
+ table = environment.getTable(dataTable);
joinResult = table.get(get);
} finally {
if (table != null) table.close();
[2/2] phoenix git commit: PHOENIX-3744 Support snapshot scanners for
MR-based Non-aggregate queries (Akshita Malhotra)
Posted by ja...@apache.org.
PHOENIX-3744 Support snapshot scanners for MR-based Non-aggregate queries (Akshita Malhotra)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/faadb544
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/faadb544
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/faadb544
Branch: refs/heads/master
Commit: faadb544857845d9520e34359c6db341ade1fe02
Parents: ed77c36
Author: James Taylor <ja...@apache.org>
Authored: Fri Jun 2 16:29:18 2017 -0700
Committer: James Taylor <ja...@apache.org>
Committed: Fri Jun 2 16:29:18 2017 -0700
----------------------------------------------------------------------
.../end2end/TableSnapshotReadsMapReduceIT.java | 204 +++++++++++
.../coprocessor/BaseScannerRegionObserver.java | 287 +--------------
.../phoenix/coprocessor/ScanRegionObserver.java | 329 +----------------
.../phoenix/iterate/BaseResultIterators.java | 15 +-
.../iterate/DefaultParallelScanGrouper.java | 64 ++--
.../iterate/MapReduceParallelScanGrouper.java | 67 +++-
.../NonAggregateRegionScannerFactory.java | 356 +++++++++++++++++++
.../phoenix/iterate/ParallelScanGrouper.java | 9 +-
.../phoenix/iterate/RegionScannerFactory.java | 322 +++++++++++++++++
.../apache/phoenix/iterate/SnapshotScanner.java | 165 +++++++++
.../iterate/TableSnapshotResultIterator.java | 138 +++++++
.../phoenix/mapreduce/PhoenixRecordReader.java | 40 ++-
.../util/PhoenixConfigurationUtil.java | 16 +
.../mapreduce/util/PhoenixMapReduceUtil.java | 64 ++++
.../java/org/apache/phoenix/util/IndexUtil.java | 9 +-
15 files changed, 1424 insertions(+), 661 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/it/java/org/apache/phoenix/end2end/TableSnapshotReadsMapReduceIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/TableSnapshotReadsMapReduceIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/TableSnapshotReadsMapReduceIT.java
new file mode 100644
index 0000000..01d92bd
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/TableSnapshotReadsMapReduceIT.java
@@ -0,0 +1,204 @@
+package org.apache.phoenix.end2end;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.mapreduce.index.PhoenixIndexDBWritable;
+import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;
+import org.junit.*;
+
+import java.io.IOException;
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.UUID;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TableSnapshotReadsMapReduceIT extends ParallelStatsDisabledIT {
+ private final static String SNAPSHOT_NAME = "FOO";
+ private static final String FIELD1 = "FIELD1";
+ private static final String FIELD2 = "FIELD2";
+ private static final String FIELD3 = "FIELD3";
+ private String CREATE_TABLE = "CREATE TABLE IF NOT EXISTS %s ( " +
+ " FIELD1 VARCHAR NOT NULL , FIELD2 VARCHAR , FIELD3 INTEGER CONSTRAINT pk PRIMARY KEY (FIELD1 ))";
+ private String UPSERT = "UPSERT into %s values (?, ?, ?)";
+
+ private static List<List<Object>> result;
+ private long timestamp;
+ private String tableName;
+
+
+ @Test
+ public void testMapReduceSnapshots() throws Exception {
+ // create table
+ Connection conn = DriverManager.getConnection(getUrl());
+ tableName = generateUniqueName();
+ conn.createStatement().execute(String.format(CREATE_TABLE, tableName));
+ conn.commit();
+
+ // configure Phoenix M/R job to read snapshot
+ final Configuration conf = getUtility().getConfiguration();
+ Job job = Job.getInstance(conf);
+ Path tmpDir = getUtility().getRandomDir();
+
+ PhoenixMapReduceUtil.setInput(job,PhoenixIndexDBWritable.class,SNAPSHOT_NAME,tableName,tmpDir, null, FIELD1, FIELD2, FIELD3);
+
+ // configure and test job
+ configureJob(job, tableName, null, null);
+ }
+
+ @Test
+ public void testMapReduceSnapshotsWithCondition() throws Exception {
+ // create table
+ Connection conn = DriverManager.getConnection(getUrl());
+ tableName = generateUniqueName();
+ conn.createStatement().execute(String.format(CREATE_TABLE, tableName));
+ conn.commit();
+
+ // configure Phoenix M/R job to read snapshot
+ final Configuration conf = getUtility().getConfiguration();
+ Job job = Job.getInstance(conf);
+ Path tmpDir = getUtility().getRandomDir();
+ PhoenixMapReduceUtil.setInput(job,PhoenixIndexDBWritable.class,SNAPSHOT_NAME,tableName,tmpDir, FIELD3 + " > 0001", FIELD1, FIELD2, FIELD3);
+
+ // configure and test job
+ configureJob(job, tableName, null, "FIELD3 > 0001");
+
+ }
+
+ @Test
+ public void testMapReduceSnapshotWithLimit() throws Exception {
+ // create table
+ Connection conn = DriverManager.getConnection(getUrl());
+ tableName = generateUniqueName();
+ conn.createStatement().execute(String.format(CREATE_TABLE, tableName));
+ conn.commit();
+
+ // configure Phoenix M/R job to read snapshot
+ final Configuration conf = getUtility().getConfiguration();
+ Job job = Job.getInstance(conf);
+ Path tmpDir = getUtility().getRandomDir();
+ // Running limit with order by on non pk column
+ String inputQuery = "SELECT * FROM " + tableName + " ORDER BY FIELD2 LIMIT 1";
+ PhoenixMapReduceUtil.setInput(job,PhoenixIndexDBWritable.class,SNAPSHOT_NAME,tableName,tmpDir,inputQuery);
+
+ // configure and test job
+ configureJob(job, tableName, inputQuery, null);
+ }
+
+ private void configureJob(Job job, String tableName, String inputQuery, String condition) throws Exception {
+ try {
+ upsertAndSnapshot(tableName);
+ result = new ArrayList<>();
+
+ job.setMapperClass(TableSnapshotMapper.class);
+ job.setMapOutputKeyClass(ImmutableBytesWritable.class);
+ job.setMapOutputValueClass(NullWritable.class);
+ job.setOutputFormatClass(NullOutputFormat.class);
+
+ Assert.assertTrue(job.waitForCompletion(true));
+
+ // verify the result, should match the values at the corresponding timestamp
+ Properties props = new Properties();
+ props.setProperty("CurrentSCN", Long.toString(timestamp));
+
+ StringBuilder selectQuery = new StringBuilder("SELECT * FROM " + tableName);
+ if (condition != null) {
+ selectQuery.append(" WHERE " + condition);
+ }
+ if (inputQuery == null)
+ inputQuery = selectQuery.toString();
+
+ ResultSet rs = DriverManager.getConnection(getUrl(), props).createStatement().executeQuery(inputQuery);
+
+ for (List<Object> r : result) {
+ assertTrue("No data stored in the table!", rs.next());
+ int i = 0;
+ String field1 = rs.getString(i + 1);
+ assertEquals("Got the incorrect value for field1", r.get(i++), field1);
+ String field2 = rs.getString(i + 1);
+ assertEquals("Got the incorrect value for field2", r.get(i++), field2);
+ int field3 = rs.getInt(i + 1);
+ assertEquals("Got the incorrect value for field3", r.get(i++), field3);
+ }
+
+ assertFalse("Should only have stored" + result.size() + "rows in the table for the timestamp!", rs.next());
+ } finally {
+ deleteSnapshotAndTable(tableName);
+ }
+ }
+
+ private void upsertData(String tableName) throws SQLException {
+ Connection conn = DriverManager.getConnection(getUrl());
+ PreparedStatement stmt = conn.prepareStatement(String.format(UPSERT, tableName));
+ upsertData(stmt, "CCCC", "SSDD", 0001);
+ upsertData(stmt, "CCCC", "HDHG", 0005);
+ upsertData(stmt, "BBBB", "JSHJ", 0002);
+ upsertData(stmt, "AAAA", "JHHD", 0003);
+ conn.commit();
+ timestamp = System.currentTimeMillis();
+ }
+
+ private void upsertData(PreparedStatement stmt, String field1, String field2, int field3) throws SQLException {
+ stmt.setString(1, field1);
+ stmt.setString(2, field2);
+ stmt.setInt(3, field3);
+ stmt.execute();
+ }
+
+ public void upsertAndSnapshot(String tableName) throws Exception {
+ upsertData(tableName);
+
+ Connection conn = DriverManager.getConnection(getUrl());
+ HBaseAdmin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin();
+ admin.snapshot(SNAPSHOT_NAME, TableName.valueOf(tableName));
+ // call flush to create new files in the region
+ admin.flush(tableName);
+
+ List<HBaseProtos.SnapshotDescription> snapshots = admin.listSnapshots();
+ Assert.assertEquals(tableName, snapshots.get(0).getTable());
+
+ // upsert data after snapshot
+ PreparedStatement stmt = conn.prepareStatement(String.format(UPSERT, tableName));
+ upsertData(stmt, "DDDD", "SNFB", 0004);
+ conn.commit();
+ }
+
+ public void deleteSnapshotAndTable(String tableName) throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+ HBaseAdmin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin();
+ admin.deleteSnapshot(SNAPSHOT_NAME);
+
+ conn.createStatement().execute("DROP TABLE " + tableName);
+ conn.close();
+
+ }
+
+ public static class TableSnapshotMapper extends Mapper<NullWritable, PhoenixIndexDBWritable, ImmutableBytesWritable, NullWritable> {
+
+ @Override
+ protected void map(NullWritable key, PhoenixIndexDBWritable record, Context context)
+ throws IOException, InterruptedException {
+ final List<Object> values = record.getValues();
+ result.add(values);
+
+ // write dummy data
+ context.write(new ImmutableBytesWritable(UUID.randomUUID().toString().getBytes()),
+ NullWritable.get());
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
index e042150..9dc1e47413 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
@@ -19,23 +19,20 @@ package org.apache.phoenix.coprocessor;
import java.io.IOException;
import java.util.List;
-import java.util.ListIterator;
-import java.util.Set;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.DoNotRetryIOException;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.HRegionInfo;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.KeyValue.Type;
+
import org.apache.hadoop.hbase.NotServingRegionException;
-import org.apache.hadoop.hbase.client.Result;
+
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.io.TimeRange;
import org.apache.hadoop.hbase.regionserver.Region;
@@ -45,28 +42,15 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.htrace.Span;
import org.apache.htrace.Trace;
import org.apache.phoenix.execute.TupleProjector;
-import org.apache.phoenix.expression.Expression;
-import org.apache.phoenix.expression.KeyValueColumnExpression;
import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.index.IndexMaintainer;
-import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.schema.KeyValueSchema;
import org.apache.phoenix.schema.PTable.QualifierEncodingScheme;
import org.apache.phoenix.schema.StaleRegionBoundaryCacheException;
-import org.apache.phoenix.schema.ValueBitSet;
-import org.apache.phoenix.schema.tuple.MultiKeyValueTuple;
-import org.apache.phoenix.schema.tuple.PositionBasedMultiKeyValueTuple;
-import org.apache.phoenix.schema.tuple.PositionBasedResultTuple;
-import org.apache.phoenix.schema.tuple.ResultTuple;
-import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.util.EncodedColumnsUtil;
-import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.ServerUtil;
-import org.apache.tephra.Transaction;
-
-import com.google.common.collect.ImmutableList;
-
+import org.apache.phoenix.iterate.RegionScannerFactory;
+import org.apache.phoenix.iterate.NonAggregateRegionScannerFactory;
abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
@@ -336,261 +320,12 @@ abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
final Region dataRegion, final IndexMaintainer indexMaintainer,
final byte[][] viewConstants, final TupleProjector projector,
final ImmutableBytesWritable ptr, final boolean useQualiferAsListIndex) {
- return getWrappedScanner(c, s, null, null, offset, scan, dataColumns, tupleProjector,
- dataRegion, indexMaintainer, null, viewConstants, null, null, projector, ptr, useQualiferAsListIndex);
- }
-
- /**
- * Return wrapped scanner that catches unexpected exceptions (i.e. Phoenix bugs) and
- * re-throws as DoNotRetryIOException to prevent needless retrying hanging the query
- * for 30 seconds. Unfortunately, until HBASE-7481 gets fixed, there's no way to do
- * the same from a custom filter.
- * @param arrayKVRefs
- * @param arrayFuncRefs
- * @param offset starting position in the rowkey.
- * @param scan
- * @param tupleProjector
- * @param dataRegion
- * @param indexMaintainer
- * @param tx current transaction
- * @param viewConstants
- */
- RegionScanner getWrappedScanner(final ObserverContext<RegionCoprocessorEnvironment> c,
- final RegionScanner s, final Set<KeyValueColumnExpression> arrayKVRefs,
- final Expression[] arrayFuncRefs, final int offset, final Scan scan,
- final ColumnReference[] dataColumns, final TupleProjector tupleProjector,
- final Region dataRegion, final IndexMaintainer indexMaintainer,
- Transaction tx,
- final byte[][] viewConstants, final KeyValueSchema kvSchema,
- final ValueBitSet kvSchemaBitSet, final TupleProjector projector,
- final ImmutableBytesWritable ptr, final boolean useQualifierAsListIndex) {
- return new RegionScanner() {
-
- private boolean hasReferences = checkForReferenceFiles();
- private HRegionInfo regionInfo = c.getEnvironment().getRegionInfo();
- private byte[] actualStartKey = getActualStartKey();
-
- // If there are any reference files after local index region merge some cases we might
- // get the records less than scan start row key. This will happen when we replace the
- // actual region start key with merge region start key. This method gives whether are
- // there any reference files in the region or not.
- private boolean checkForReferenceFiles() {
- if(!ScanUtil.isLocalIndex(scan)) return false;
- for (byte[] family : scan.getFamilies()) {
- if (c.getEnvironment().getRegion().getStore(family).hasReferences()) {
- return true;
- }
- }
- return false;
- }
-
- // Get the actual scan start row of local index. This will be used to compare the row
- // key of the results less than scan start row when there are references.
- public byte[] getActualStartKey() {
- return ScanUtil.isLocalIndex(scan) ? ScanUtil.getActualStartRow(scan, regionInfo)
- : null;
- }
-
- @Override
- public boolean next(List<Cell> results) throws IOException {
- try {
- return s.next(results);
- } catch (Throwable t) {
- ServerUtil.throwIOException(c.getEnvironment().getRegion().getRegionInfo().getRegionNameAsString(), t);
- return false; // impossible
- }
- }
-
- @Override
- public boolean next(List<Cell> result, ScannerContext scannerContext) throws IOException {
- try {
- return s.next(result, scannerContext);
- } catch (Throwable t) {
- ServerUtil.throwIOException(c.getEnvironment().getRegion().getRegionInfo().getRegionNameAsString(), t);
- return false; // impossible
- }
- }
-
- @Override
- public void close() throws IOException {
- s.close();
- }
-
- @Override
- public HRegionInfo getRegionInfo() {
- return s.getRegionInfo();
- }
- @Override
- public boolean isFilterDone() throws IOException {
- return s.isFilterDone();
- }
-
- @Override
- public boolean reseek(byte[] row) throws IOException {
- return s.reseek(row);
- }
-
- @Override
- public long getMvccReadPoint() {
- return s.getMvccReadPoint();
- }
-
- @Override
- public boolean nextRaw(List<Cell> result) throws IOException {
- try {
- boolean next = s.nextRaw(result);
- Cell arrayElementCell = null;
- if (result.size() == 0) {
- return next;
- }
- if (arrayFuncRefs != null && arrayFuncRefs.length > 0 && arrayKVRefs.size() > 0) {
- int arrayElementCellPosition = replaceArrayIndexElement(arrayKVRefs, arrayFuncRefs, result);
- arrayElementCell = result.get(arrayElementCellPosition);
- }
- if (ScanUtil.isLocalIndex(scan) && !ScanUtil.isAnalyzeTable(scan)) {
- if(hasReferences && actualStartKey!=null) {
- next = scanTillScanStartRow(s, arrayKVRefs, arrayFuncRefs, result,
- null, arrayElementCell);
- if (result.isEmpty()) {
- return next;
- }
- }
- IndexUtil.wrapResultUsingOffset(c, result, offset, dataColumns,
- tupleProjector, dataRegion, indexMaintainer, viewConstants, ptr);
- }
- if (projector != null) {
- Tuple toProject = useQualifierAsListIndex ? new PositionBasedResultTuple(result) : new ResultTuple(Result.create(result));
- Tuple tuple = projector.projectResults(toProject, useNewValueColumnQualifier);
- result.clear();
- result.add(tuple.getValue(0));
- if (arrayElementCell != null) {
- result.add(arrayElementCell);
- }
- }
- // There is a scanattribute set to retrieve the specific array element
- return next;
- } catch (Throwable t) {
- ServerUtil.throwIOException(c.getEnvironment().getRegion().getRegionInfo().getRegionNameAsString(), t);
- return false; // impossible
- }
- }
-
- @Override
- public boolean nextRaw(List<Cell> result, ScannerContext scannerContext)
- throws IOException {
- try {
- boolean next = s.nextRaw(result, scannerContext);
- Cell arrayElementCell = null;
- if (result.size() == 0) {
- return next;
- }
- if (arrayFuncRefs != null && arrayFuncRefs.length > 0 && arrayKVRefs.size() > 0) {
- int arrayElementCellPosition = replaceArrayIndexElement(arrayKVRefs, arrayFuncRefs, result);
- arrayElementCell = result.get(arrayElementCellPosition);
- }
- if ((offset > 0 || ScanUtil.isLocalIndex(scan)) && !ScanUtil.isAnalyzeTable(scan)) {
- if(hasReferences && actualStartKey!=null) {
- next = scanTillScanStartRow(s, arrayKVRefs, arrayFuncRefs, result,
- scannerContext, arrayElementCell);
- if (result.isEmpty()) {
- return next;
- }
- }
- IndexUtil.wrapResultUsingOffset(c, result, offset, dataColumns,
- tupleProjector, dataRegion, indexMaintainer, viewConstants, ptr);
- }
- if (projector != null) {
- Tuple toProject = useQualifierAsListIndex ? new PositionBasedMultiKeyValueTuple(result) : new ResultTuple(Result.create(result));
- Tuple tuple = projector.projectResults(toProject, useNewValueColumnQualifier);
- result.clear();
- result.add(tuple.getValue(0));
- if(arrayElementCell != null)
- result.add(arrayElementCell);
- }
- // There is a scanattribute set to retrieve the specific array element
- return next;
- } catch (Throwable t) {
- ServerUtil.throwIOException(c.getEnvironment().getRegion().getRegionInfo().getRegionNameAsString(), t);
- return false; // impossible
- }
- }
+ RegionScannerFactory regionScannerFactory = new NonAggregateRegionScannerFactory(c.getEnvironment(),
+ useNewValueColumnQualifier, encodingScheme);
- /**
- * When there is a merge in progress while scanning local indexes we might get the key values less than scan start row.
- * In that case we need to scan until get the row key more or equal to scan start key.
- * TODO try to fix this case in LocalIndexStoreFileScanner when there is a merge.
- */
- private boolean scanTillScanStartRow(final RegionScanner s,
- final Set<KeyValueColumnExpression> arrayKVRefs,
- final Expression[] arrayFuncRefs, List<Cell> result,
- ScannerContext scannerContext, Cell arrayElementCell) throws IOException {
- boolean next = true;
- Cell firstCell = result.get(0);
- while (Bytes.compareTo(firstCell.getRowArray(), firstCell.getRowOffset(),
- firstCell.getRowLength(), actualStartKey, 0, actualStartKey.length) < 0) {
- result.clear();
- if(scannerContext == null) {
- next = s.nextRaw(result);
- } else {
- next = s.nextRaw(result, scannerContext);
- }
- if (result.isEmpty()) {
- return next;
- }
- if (arrayFuncRefs != null && arrayFuncRefs.length > 0 && arrayKVRefs.size() > 0) {
- int arrayElementCellPosition = replaceArrayIndexElement(arrayKVRefs, arrayFuncRefs, result);
- arrayElementCell = result.get(arrayElementCellPosition);
- }
- firstCell = result.get(0);
- }
- return next;
- }
-
- private int replaceArrayIndexElement(final Set<KeyValueColumnExpression> arrayKVRefs,
- final Expression[] arrayFuncRefs, List<Cell> result) {
- // make a copy of the results array here, as we're modifying it below
- MultiKeyValueTuple tuple = new MultiKeyValueTuple(ImmutableList.copyOf(result));
- // The size of both the arrays would be same?
- // Using KeyValueSchema to set and retrieve the value
- // collect the first kv to get the row
- Cell rowKv = result.get(0);
- for (KeyValueColumnExpression kvExp : arrayKVRefs) {
- if (kvExp.evaluate(tuple, ptr)) {
- ListIterator<Cell> itr = result.listIterator();
- while (itr.hasNext()) {
- Cell kv = itr.next();
- if (Bytes.equals(kvExp.getColumnFamily(), 0, kvExp.getColumnFamily().length,
- kv.getFamilyArray(), kv.getFamilyOffset(), kv.getFamilyLength())
- && Bytes.equals(kvExp.getColumnQualifier(), 0, kvExp.getColumnQualifier().length,
- kv.getQualifierArray(), kv.getQualifierOffset(), kv.getQualifierLength())) {
- // remove the kv that has the full array values.
- itr.remove();
- break;
- }
- }
- }
- }
- byte[] value = kvSchema.toBytes(tuple, arrayFuncRefs,
- kvSchemaBitSet, ptr);
- // Add a dummy kv with the exact value of the array index
- result.add(new KeyValue(rowKv.getRowArray(), rowKv.getRowOffset(), rowKv.getRowLength(),
- QueryConstants.ARRAY_VALUE_COLUMN_FAMILY, 0, QueryConstants.ARRAY_VALUE_COLUMN_FAMILY.length,
- QueryConstants.ARRAY_VALUE_COLUMN_QUALIFIER, 0,
- QueryConstants.ARRAY_VALUE_COLUMN_QUALIFIER.length, HConstants.LATEST_TIMESTAMP,
- Type.codeToType(rowKv.getTypeByte()), value, 0, value.length));
- return result.size() - 1;
- }
-
- @Override
- public long getMaxResultSize() {
- return s.getMaxResultSize();
- }
-
- @Override
- public int getBatch() {
- return s.getBatch();
- }
- };
+ return regionScannerFactory.getWrappedScanner(c.getEnvironment(), s, null, null, offset, scan, dataColumns, tupleProjector,
+ dataRegion, indexMaintainer, null, viewConstants, null, null, projector, ptr, useQualiferAsListIndex);
}
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
index 02b05f9..b006ef6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
@@ -17,64 +17,19 @@
*/
package org.apache.phoenix.coprocessor;
-import static org.apache.phoenix.util.EncodedColumnsUtil.getMinMaxQualifiersFromScan;
-
-import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
-import java.sql.SQLException;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Set;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.regionserver.Region;
+
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.io.WritableUtils;
-import org.apache.phoenix.cache.GlobalCache;
-import org.apache.phoenix.cache.TenantCache;
-import org.apache.phoenix.execute.MutationState;
-import org.apache.phoenix.execute.TupleProjector;
-import org.apache.phoenix.expression.Expression;
-import org.apache.phoenix.expression.KeyValueColumnExpression;
import org.apache.phoenix.expression.OrderByExpression;
-import org.apache.phoenix.expression.SingleCellColumnExpression;
-import org.apache.phoenix.expression.function.ArrayIndexFunction;
-import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
-import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
-import org.apache.phoenix.index.IndexMaintainer;
-import org.apache.phoenix.iterate.OffsetResultIterator;
-import org.apache.phoenix.iterate.OrderedResultIterator;
-import org.apache.phoenix.iterate.RegionScannerResultIterator;
-import org.apache.phoenix.iterate.ResultIterator;
-import org.apache.phoenix.join.HashJoinInfo;
-import org.apache.phoenix.memory.MemoryManager.MemoryChunk;
-import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.schema.KeyValueSchema;
-import org.apache.phoenix.schema.KeyValueSchema.KeyValueSchemaBuilder;
-import org.apache.phoenix.schema.PTable.ImmutableStorageScheme;
-import org.apache.phoenix.schema.PTable.QualifierEncodingScheme;
-import org.apache.phoenix.schema.ValueBitSet;
-import org.apache.phoenix.schema.tuple.ResultTuple;
-import org.apache.phoenix.schema.tuple.Tuple;
-import org.apache.phoenix.schema.types.PInteger;
-import org.apache.phoenix.util.EncodedColumnsUtil;
-import org.apache.phoenix.util.IndexUtil;
-import org.apache.phoenix.util.ScanUtil;
-import org.apache.phoenix.util.ServerUtil;
-import org.apache.tephra.Transaction;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
-
+import org.apache.phoenix.iterate.NonAggregateRegionScannerFactory;
/**
*
@@ -87,9 +42,7 @@ import com.google.common.collect.Sets;
* @since 0.1
*/
public class ScanRegionObserver extends BaseScannerRegionObserver {
- private ImmutableBytesWritable ptr = new ImmutableBytesWritable();
- private KeyValueSchema kvSchema = null;
- private ValueBitSet kvSchemaBitSet;
+
public static void serializeIntoScan(Scan scan, int thresholdBytes, int limit, List<OrderByExpression> orderByExpressions, int estimatedRowSize) {
ByteArrayOutputStream stream = new ByteArrayOutputStream(); // TODO: size?
try {
@@ -113,282 +66,10 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
}
}
- private static OrderedResultIterator deserializeFromScan(Scan scan, RegionScanner s) {
- byte[] topN = scan.getAttribute(BaseScannerRegionObserver.TOPN);
- if (topN == null) {
- return null;
- }
- ByteArrayInputStream stream = new ByteArrayInputStream(topN); // TODO: size?
- try {
- DataInputStream input = new DataInputStream(stream);
- int thresholdBytes = WritableUtils.readVInt(input);
- int limit = WritableUtils.readVInt(input);
- int estimatedRowSize = WritableUtils.readVInt(input);
- int size = WritableUtils.readVInt(input);
- List<OrderByExpression> orderByExpressions = Lists.newArrayListWithExpectedSize(size);
- for (int i = 0; i < size; i++) {
- OrderByExpression orderByExpression = new OrderByExpression();
- orderByExpression.readFields(input);
- orderByExpressions.add(orderByExpression);
- }
- QualifierEncodingScheme encodingScheme = EncodedColumnsUtil.getQualifierEncodingScheme(scan);
- ResultIterator inner = new RegionScannerResultIterator(s, EncodedColumnsUtil.getMinMaxQualifiersFromScan(scan), encodingScheme);
- return new OrderedResultIterator(inner, orderByExpressions, thresholdBytes, limit >= 0 ? limit : null, null,
- estimatedRowSize);
- } catch (IOException e) {
- throw new RuntimeException(e);
- } finally {
- try {
- stream.close();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- }
-
- private Expression[] deserializeArrayPostionalExpressionInfoFromScan(Scan scan, RegionScanner s,
- Set<KeyValueColumnExpression> arrayKVRefs) {
- byte[] specificArrayIdx = scan.getAttribute(BaseScannerRegionObserver.SPECIFIC_ARRAY_INDEX);
- if (specificArrayIdx == null) {
- return null;
- }
- KeyValueSchemaBuilder builder = new KeyValueSchemaBuilder(0);
- ByteArrayInputStream stream = new ByteArrayInputStream(specificArrayIdx);
- try {
- DataInputStream input = new DataInputStream(stream);
- int arrayKVRefSize = WritableUtils.readVInt(input);
- for (int i = 0; i < arrayKVRefSize; i++) {
- ImmutableStorageScheme scheme = EncodedColumnsUtil.getImmutableStorageScheme(scan);
- KeyValueColumnExpression kvExp = scheme != ImmutableStorageScheme.ONE_CELL_PER_COLUMN ? new SingleCellColumnExpression()
- : new KeyValueColumnExpression();
- kvExp.readFields(input);
- arrayKVRefs.add(kvExp);
- }
- int arrayKVFuncSize = WritableUtils.readVInt(input);
- Expression[] arrayFuncRefs = new Expression[arrayKVFuncSize];
- for (int i = 0; i < arrayKVFuncSize; i++) {
- ArrayIndexFunction arrayIdxFunc = new ArrayIndexFunction();
- arrayIdxFunc.readFields(input);
- arrayFuncRefs[i] = arrayIdxFunc;
- builder.addField(arrayIdxFunc);
- }
- kvSchema = builder.build();
- kvSchemaBitSet = ValueBitSet.newInstance(kvSchema);
- return arrayFuncRefs;
- } catch (IOException e) {
- throw new RuntimeException(e);
- } finally {
- try {
- stream.close();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- }
-
@Override
protected RegionScanner doPostScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c, final Scan scan, final RegionScanner s) throws Throwable {
- int offset = 0;
- if (ScanUtil.isLocalIndex(scan)) {
- /*
- * For local indexes, we need to set an offset on row key expressions to skip
- * the region start key.
- */
- Region region = c.getEnvironment().getRegion();
- offset = region.getRegionInfo().getStartKey().length != 0 ? region.getRegionInfo().getStartKey().length :
- region.getRegionInfo().getEndKey().length;
- ScanUtil.setRowKeyOffset(scan, offset);
- }
- byte[] scanOffsetBytes = scan.getAttribute(BaseScannerRegionObserver.SCAN_OFFSET);
- Integer scanOffset = null;
- if (scanOffsetBytes != null) {
- scanOffset = (Integer) PInteger.INSTANCE.toObject(scanOffsetBytes);
- }
- RegionScanner innerScanner = s;
-
- Set<KeyValueColumnExpression> arrayKVRefs = Sets.newHashSet();
- Expression[] arrayFuncRefs = deserializeArrayPostionalExpressionInfoFromScan(
- scan, innerScanner, arrayKVRefs);
- TupleProjector tupleProjector = null;
- Region dataRegion = null;
- IndexMaintainer indexMaintainer = null;
- byte[][] viewConstants = null;
- Transaction tx = null;
- ColumnReference[] dataColumns = IndexUtil.deserializeDataTableColumnsToJoin(scan);
- if (dataColumns != null) {
- tupleProjector = IndexUtil.getTupleProjector(scan, dataColumns);
- dataRegion = c.getEnvironment().getRegion();
- boolean useProto = false;
- byte[] localIndexBytes = scan.getAttribute(LOCAL_INDEX_BUILD_PROTO);
- useProto = localIndexBytes != null;
- if (localIndexBytes == null) {
- localIndexBytes = scan.getAttribute(LOCAL_INDEX_BUILD);
- }
- List<IndexMaintainer> indexMaintainers = localIndexBytes == null ? null : IndexMaintainer.deserialize(localIndexBytes, useProto);
- indexMaintainer = indexMaintainers.get(0);
- viewConstants = IndexUtil.deserializeViewConstantsFromScan(scan);
- byte[] txState = scan.getAttribute(BaseScannerRegionObserver.TX_STATE);
- tx = MutationState.decodeTransaction(txState);
- }
-
- final TupleProjector p = TupleProjector.deserializeProjectorFromScan(scan);
- final HashJoinInfo j = HashJoinInfo.deserializeHashJoinFromScan(scan);
- boolean useQualifierAsIndex = EncodedColumnsUtil.useQualifierAsIndex(getMinMaxQualifiersFromScan(scan)) && scan.getAttribute(BaseScannerRegionObserver.TOPN) != null;
- innerScanner =
- getWrappedScanner(c, innerScanner, arrayKVRefs, arrayFuncRefs, offset, scan,
- dataColumns, tupleProjector, dataRegion, indexMaintainer, tx,
- viewConstants, kvSchema, kvSchemaBitSet, j == null ? p : null, ptr, useQualifierAsIndex);
-
- final ImmutableBytesPtr tenantId = ScanUtil.getTenantId(scan);
- if (j != null) {
- innerScanner = new HashJoinRegionScanner(innerScanner, p, j, tenantId, c.getEnvironment(), useQualifierAsIndex, useNewValueColumnQualifier);
- }
- if (scanOffset != null) {
- innerScanner = getOffsetScanner(c, innerScanner,
- new OffsetResultIterator(new RegionScannerResultIterator(innerScanner, getMinMaxQualifiersFromScan(scan), encodingScheme), scanOffset),
- scan.getAttribute(QueryConstants.LAST_SCAN) != null);
- }
- final OrderedResultIterator iterator = deserializeFromScan(scan, innerScanner);
- if (iterator == null) {
- return innerScanner;
- }
- // TODO:the above wrapped scanner should be used here also
- return getTopNScanner(c, innerScanner, iterator, tenantId);
- }
-
- private RegionScanner getOffsetScanner(final ObserverContext<RegionCoprocessorEnvironment> c, final RegionScanner s,
- final OffsetResultIterator iterator, final boolean isLastScan) throws IOException {
- final Tuple firstTuple;
- final Region region = c.getEnvironment().getRegion();
- region.startRegionOperation();
- try {
- Tuple tuple = iterator.next();
- if (tuple == null && !isLastScan) {
- List<KeyValue> kvList = new ArrayList<KeyValue>(1);
- KeyValue kv = new KeyValue(QueryConstants.OFFSET_ROW_KEY_BYTES, QueryConstants.OFFSET_FAMILY,
- QueryConstants.OFFSET_COLUMN, PInteger.INSTANCE.toBytes(iterator.getRemainingOffset()));
- kvList.add(kv);
- Result r = new Result(kvList);
- firstTuple = new ResultTuple(r);
- } else {
- firstTuple = tuple;
- }
- } catch (Throwable t) {
- ServerUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), t);
- return null;
- } finally {
- region.closeRegionOperation();
- }
- return new BaseRegionScanner(s) {
- private Tuple tuple = firstTuple;
-
- @Override
- public boolean isFilterDone() {
- return tuple == null;
- }
-
- @Override
- public boolean next(List<Cell> results) throws IOException {
- try {
- if (isFilterDone()) { return false; }
- for (int i = 0; i < tuple.size(); i++) {
- results.add(tuple.getValue(i));
- }
- tuple = iterator.next();
- return !isFilterDone();
- } catch (Throwable t) {
- ServerUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), t);
- return false;
- }
- }
-
- @Override
- public void close() throws IOException {
- try {
- s.close();
- } finally {
- try {
- if (iterator != null) {
- iterator.close();
- }
- } catch (SQLException e) {
- ServerUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), e);
- }
- }
- }
- };
- }
-
- /**
- * Return region scanner that does TopN.
- * We only need to call startRegionOperation and closeRegionOperation when
- * getting the first Tuple (which forces running through the entire region)
- * since after this everything is held in memory
- */
- private RegionScanner getTopNScanner(final ObserverContext<RegionCoprocessorEnvironment> c, final RegionScanner s, final OrderedResultIterator iterator, ImmutableBytesPtr tenantId) throws Throwable {
- final Tuple firstTuple;
- TenantCache tenantCache = GlobalCache.getTenantCache(c.getEnvironment(), tenantId);
- long estSize = iterator.getEstimatedByteSize();
- final MemoryChunk chunk = tenantCache.getMemoryManager().allocate(estSize);
- final Region region = c.getEnvironment().getRegion();
- region.startRegionOperation();
- try {
- // Once we return from the first call to next, we've run through and cached
- // the topN rows, so we no longer need to start/stop a region operation.
- firstTuple = iterator.next();
- // Now that the topN are cached, we can resize based on the real size
- long actualSize = iterator.getByteSize();
- chunk.resize(actualSize);
- } catch (Throwable t) {
- ServerUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), t);
- return null;
- } finally {
- region.closeRegionOperation();
- }
- return new BaseRegionScanner(s) {
- private Tuple tuple = firstTuple;
-
- @Override
- public boolean isFilterDone() {
- return tuple == null;
- }
-
- @Override
- public boolean next(List<Cell> results) throws IOException {
- try {
- if (isFilterDone()) {
- return false;
- }
-
- for (int i = 0; i < tuple.size(); i++) {
- results.add(tuple.getValue(i));
- }
-
- tuple = iterator.next();
- return !isFilterDone();
- } catch (Throwable t) {
- ServerUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), t);
- return false;
- }
- }
-
- @Override
- public void close() throws IOException {
- try {
- s.close();
- } finally {
- try {
- if(iterator != null) {
- iterator.close();
- }
- } catch (SQLException e) {
- ServerUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), e);
- } finally {
- chunk.close();
- }
- }
- }
- };
+ NonAggregateRegionScannerFactory nonAggregateROUtil = new NonAggregateRegionScannerFactory(c.getEnvironment(), useNewValueColumnQualifier, encodingScheme);
+ return nonAggregateROUtil.getRegionScanner(scan, s);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
index b37514c..848281f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
@@ -501,6 +501,11 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
return scans;
}
+ private List<HRegionLocation> getRegionBoundaries(ParallelScanGrouper scanGrouper)
+ throws SQLException{
+ return scanGrouper.getRegionBoundaries(context, physicalTableName);
+ }
+
private static List<byte[]> toBoundaries(List<HRegionLocation> regionLocations) {
int nBoundaries = regionLocations.size() - 1;
List<byte[]> ranges = Lists.newArrayListWithExpectedSize(nBoundaries);
@@ -564,7 +569,9 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
private List<Scan> addNewScan(List<List<Scan>> parallelScans, List<Scan> scans, Scan scan, byte[] startKey, boolean crossedRegionBoundary, HRegionLocation regionLocation) {
boolean startNewScan = scanGrouper.shouldStartNewScan(plan, scans, startKey, crossedRegionBoundary);
if (scan != null) {
- scan.setAttribute(BaseScannerRegionObserver.SCAN_REGION_SERVER, regionLocation.getServerName().getVersionedBytes());
+ if (regionLocation.getServerName() != null) {
+ scan.setAttribute(BaseScannerRegionObserver.SCAN_REGION_SERVER, regionLocation.getServerName().getVersionedBytes());
+ }
scans.add(scan);
}
if (startNewScan && !scans.isEmpty()) {
@@ -592,8 +599,7 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
* @throws SQLException
*/
private List<List<Scan>> getParallelScans(Scan scan) throws SQLException {
- List<HRegionLocation> regionLocations = context.getConnection().getQueryServices()
- .getAllTableRegions(physicalTableName);
+ List<HRegionLocation> regionLocations = getRegionBoundaries(scanGrouper);
List<byte[]> regionBoundaries = toBoundaries(regionLocations);
int regionIndex = 0;
int stopIndex = regionBoundaries.size();
@@ -645,8 +651,7 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
* @throws SQLException
*/
private List<List<Scan>> getParallelScans(byte[] startKey, byte[] stopKey) throws SQLException {
- List<HRegionLocation> regionLocations = context.getConnection().getQueryServices()
- .getAllTableRegions(physicalTableName);
+ List<HRegionLocation> regionLocations = getRegionBoundaries(scanGrouper);
List<byte[]> regionBoundaries = toBoundaries(regionLocations);
ScanRanges scanRanges = context.getScanRanges();
PTable table = getTable();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelScanGrouper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelScanGrouper.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelScanGrouper.java
index 5c7136f..2cd96df 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelScanGrouper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelScanGrouper.java
@@ -17,46 +17,58 @@
*/
package org.apache.phoenix.iterate;
+import java.sql.SQLException;
import java.util.List;
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.phoenix.compile.QueryPlan;
+import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.SaltingUtil;
+import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.util.ScanUtil;
/**
* Default implementation that creates a scan group if a plan is row key ordered (which requires a merge sort),
- * or if a scan crosses a region boundary and the table is salted or a local index.
+ * or if a scan crosses a region boundary and the table is salted or a local index.
*/
public class DefaultParallelScanGrouper implements ParallelScanGrouper {
-
- private static final DefaultParallelScanGrouper INSTANCE = new DefaultParallelScanGrouper();
- public static DefaultParallelScanGrouper getInstance() {
- return INSTANCE;
- }
-
- private DefaultParallelScanGrouper() {}
-
- @Override
- public boolean shouldStartNewScan(QueryPlan plan, List<Scan> scans, byte[] startKey, boolean crossedRegionBoundary) {
- PTable table = plan.getTableRef().getTable();
- boolean startNewScanGroup = false;
- if (!plan.isRowKeyOrdered()) {
- startNewScanGroup = true;
- } else if (crossedRegionBoundary) {
- if (table.getIndexType() == IndexType.LOCAL) {
- startNewScanGroup = true;
- } else if (table.getBucketNum() != null) {
- startNewScanGroup = scans.isEmpty() ||
- ScanUtil.crossesPrefixBoundary(startKey,
- ScanUtil.getPrefix(scans.get(scans.size()-1).getStartRow(), SaltingUtil.NUM_SALTING_BYTES),
- SaltingUtil.NUM_SALTING_BYTES);
- }
- }
- return startNewScanGroup;
+ private static DefaultParallelScanGrouper INSTANCE = new DefaultParallelScanGrouper();
+
+
+ public DefaultParallelScanGrouper() {
+ }
+
+ public static DefaultParallelScanGrouper getInstance() {
+ return INSTANCE;
+ }
+
+
+ @Override
+ public boolean shouldStartNewScan(QueryPlan plan, List<Scan> scans, byte[] startKey, boolean crossedRegionBoundary) {
+ PTable table = plan.getTableRef().getTable();
+ boolean startNewScanGroup = false;
+ if (!plan.isRowKeyOrdered()) {
+ startNewScanGroup = true;
+ } else if (crossedRegionBoundary) {
+ if (table.getIndexType() == IndexType.LOCAL) {
+ startNewScanGroup = true;
+ } else if (table.getBucketNum() != null) {
+ startNewScanGroup = scans.isEmpty() ||
+ ScanUtil.crossesPrefixBoundary(startKey,
+ ScanUtil.getPrefix(scans.get(scans.size()-1).getStartRow(), SaltingUtil.NUM_SALTING_BYTES),
+ SaltingUtil.NUM_SALTING_BYTES);
+ }
}
+ return startNewScanGroup;
+ }
+ @Override
+ public List<HRegionLocation> getRegionBoundaries(StatementContext context, byte[] tableName) throws SQLException{
+ return context.getConnection().getQueryServices().getAllTableRegions(tableName);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/iterate/MapReduceParallelScanGrouper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/MapReduceParallelScanGrouper.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/MapReduceParallelScanGrouper.java
index bf2666d..593608f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/MapReduceParallelScanGrouper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/MapReduceParallelScanGrouper.java
@@ -17,24 +17,39 @@
*/
package org.apache.phoenix.iterate;
+import java.sql.SQLException;
import java.util.List;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.hbase.protobuf.generated.SnapshotProtos;
+import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
+import org.apache.hadoop.hbase.snapshot.SnapshotManifest;
import org.apache.phoenix.compile.QueryPlan;
+import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
/**
* Scan grouper that creates a scan group if a plan is row key ordered or if a
* scan crosses region boundaries
*/
public class MapReduceParallelScanGrouper implements ParallelScanGrouper {
-
+
private static final MapReduceParallelScanGrouper INSTANCE = new MapReduceParallelScanGrouper();
- public static MapReduceParallelScanGrouper getInstance() {
- return INSTANCE;
- }
-
- private MapReduceParallelScanGrouper() {}
+ public static MapReduceParallelScanGrouper getInstance() {
+ return INSTANCE;
+ }
+
+ private MapReduceParallelScanGrouper() {}
@Override
public boolean shouldStartNewScan(QueryPlan plan, List<Scan> scans,
@@ -42,4 +57,44 @@ public class MapReduceParallelScanGrouper implements ParallelScanGrouper {
return !plan.isRowKeyOrdered() || crossedRegionBoundary;
}
+ @Override
+ public List<HRegionLocation> getRegionBoundaries(StatementContext context, byte[] tableName) throws SQLException {
+ String snapshotName;
+ Configuration conf = context.getConnection().getQueryServices().getConfiguration();
+ if((snapshotName = getSnapshotName(conf)) != null) {
+ try {
+ Path rootDir = new Path(conf.get(HConstants.HBASE_DIR));
+ FileSystem fs = rootDir.getFileSystem(conf);
+ Path snapshotDir = SnapshotDescriptionUtils.getCompletedSnapshotDir(snapshotName, rootDir);
+ HBaseProtos.SnapshotDescription snapshotDescription = SnapshotDescriptionUtils.readSnapshotInfo(fs, snapshotDir);
+ SnapshotManifest manifest = SnapshotManifest.open(conf, fs, snapshotDir, snapshotDescription);
+ return getRegionLocationsFromManifest(manifest);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ else {
+ return context.getConnection().getQueryServices().getAllTableRegions(tableName);
+ }
+ }
+
+ private List<HRegionLocation> getRegionLocationsFromManifest(SnapshotManifest manifest) {
+ List<SnapshotProtos.SnapshotRegionManifest> regionManifests = manifest.getRegionManifests();
+ Preconditions.checkNotNull(regionManifests);
+
+ List<HRegionLocation> regionLocations = Lists.newArrayListWithCapacity(regionManifests.size());
+
+ for (SnapshotProtos.SnapshotRegionManifest regionManifest : regionManifests) {
+ regionLocations.add(new HRegionLocation(
+ HRegionInfo.convert(regionManifest.getRegionInfo()), null));
+ }
+
+ return regionLocations;
+ }
+
+ private String getSnapshotName(Configuration conf) {
+ return conf.get(PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/iterate/NonAggregateRegionScannerFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/NonAggregateRegionScannerFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/NonAggregateRegionScannerFactory.java
new file mode 100644
index 0000000..096ff25
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/NonAggregateRegionScannerFactory.java
@@ -0,0 +1,356 @@
+package org.apache.phoenix.iterate;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.regionserver.Region;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.phoenix.cache.GlobalCache;
+import org.apache.phoenix.cache.TenantCache;
+import org.apache.phoenix.coprocessor.BaseRegionScanner;
+import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
+import org.apache.phoenix.coprocessor.HashJoinRegionScanner;
+import org.apache.phoenix.execute.MutationState;
+import org.apache.phoenix.execute.TupleProjector;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.expression.KeyValueColumnExpression;
+import org.apache.phoenix.expression.OrderByExpression;
+import org.apache.phoenix.expression.SingleCellColumnExpression;
+import org.apache.phoenix.expression.function.ArrayIndexFunction;
+import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.join.HashJoinInfo;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.memory.MemoryManager;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.KeyValueSchema;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.ValueBitSet;
+import org.apache.phoenix.schema.tuple.ResultTuple;
+import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.schema.types.PInteger;
+import org.apache.phoenix.util.EncodedColumnsUtil;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.ScanUtil;
+import org.apache.phoenix.util.ServerUtil;
+import org.apache.tephra.Transaction;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.phoenix.util.EncodedColumnsUtil.getMinMaxQualifiersFromScan;
+
+public class NonAggregateRegionScannerFactory extends RegionScannerFactory {
+
+ private ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ private KeyValueSchema kvSchema = null;
+ private ValueBitSet kvSchemaBitSet;
+
+ public NonAggregateRegionScannerFactory(RegionCoprocessorEnvironment env, boolean useNewValueColumnQualifier,
+ PTable.QualifierEncodingScheme encodingScheme) {
+ this.env = env;
+ this.useNewValueColumnQualifier = useNewValueColumnQualifier;
+ this.encodingScheme = encodingScheme;
+ }
+
+ @Override
+ public RegionScanner getRegionScanner(final Scan scan, final RegionScanner s) throws Throwable {
+
+ int offset = 0;
+ if (ScanUtil.isLocalIndex(scan)) {
+ /*
+ * For local indexes, we need to set an offset on row key expressions to skip
+ * the region start key.
+ */
+ Region region = getRegion();
+ offset = region.getRegionInfo().getStartKey().length != 0 ?
+ region.getRegionInfo().getStartKey().length :
+ region.getRegionInfo().getEndKey().length;
+ ScanUtil.setRowKeyOffset(scan, offset);
+ }
+ byte[] scanOffsetBytes = scan.getAttribute(BaseScannerRegionObserver.SCAN_OFFSET);
+ Integer scanOffset = null;
+ if (scanOffsetBytes != null) {
+ scanOffset = (Integer)PInteger.INSTANCE.toObject(scanOffsetBytes);
+ }
+ RegionScanner innerScanner = s;
+
+ Set<KeyValueColumnExpression> arrayKVRefs = Sets.newHashSet();
+ Expression[] arrayFuncRefs = deserializeArrayPostionalExpressionInfoFromScan(scan, innerScanner, arrayKVRefs);
+ TupleProjector tupleProjector = null;
+ Region dataRegion = null;
+ IndexMaintainer indexMaintainer = null;
+ byte[][] viewConstants = null;
+ Transaction tx = null;
+ ColumnReference[] dataColumns = IndexUtil.deserializeDataTableColumnsToJoin(scan);
+ if (dataColumns != null) {
+ tupleProjector = IndexUtil.getTupleProjector(scan, dataColumns);
+ dataRegion = env.getRegion();
+ boolean useProto = false;
+ byte[] localIndexBytes = scan.getAttribute(BaseScannerRegionObserver.LOCAL_INDEX_BUILD_PROTO);
+ useProto = localIndexBytes != null;
+ if (localIndexBytes == null) {
+ localIndexBytes = scan.getAttribute(BaseScannerRegionObserver.LOCAL_INDEX_BUILD);
+ }
+ List<IndexMaintainer> indexMaintainers =
+ localIndexBytes == null ? null : IndexMaintainer.deserialize(localIndexBytes, useProto);
+ indexMaintainer = indexMaintainers.get(0);
+ viewConstants = IndexUtil.deserializeViewConstantsFromScan(scan);
+ byte[] txState = scan.getAttribute(BaseScannerRegionObserver.TX_STATE);
+ tx = MutationState.decodeTransaction(txState);
+ }
+
+ final TupleProjector p = TupleProjector.deserializeProjectorFromScan(scan);
+ final HashJoinInfo j = HashJoinInfo.deserializeHashJoinFromScan(scan);
+ boolean useQualifierAsIndex = EncodedColumnsUtil.useQualifierAsIndex(getMinMaxQualifiersFromScan(scan))
+ && scan.getAttribute(BaseScannerRegionObserver.TOPN) != null;
+ // setting dataRegion in case of a non-coprocessor environment
+ if (dataRegion == null &&
+ env.getConfiguration().get(PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY) != null) {
+ dataRegion = env.getRegion();
+ }
+ innerScanner = getWrappedScanner(env, innerScanner, arrayKVRefs, arrayFuncRefs, offset, scan, dataColumns,
+ tupleProjector, dataRegion, indexMaintainer, tx, viewConstants, kvSchema, kvSchemaBitSet, j == null ? p : null,
+ ptr, useQualifierAsIndex);
+
+ final ImmutableBytesPtr tenantId = ScanUtil.getTenantId(scan);
+ if (j != null) {
+ innerScanner = new HashJoinRegionScanner(innerScanner, p, j, tenantId, env, useQualifierAsIndex,
+ useNewValueColumnQualifier);
+ }
+ if (scanOffset != null) {
+ innerScanner = getOffsetScanner(innerScanner, new OffsetResultIterator(
+ new RegionScannerResultIterator(innerScanner, getMinMaxQualifiersFromScan(scan), encodingScheme), scanOffset),
+ scan.getAttribute(QueryConstants.LAST_SCAN) != null);
+ }
+ final OrderedResultIterator iterator = deserializeFromScan(scan, innerScanner);
+ if (iterator == null) {
+ return innerScanner;
+ }
+ // TODO:the above wrapped scanner should be used here also
+ return getTopNScanner(env, innerScanner, iterator, tenantId);
+ }
+
+ private static OrderedResultIterator deserializeFromScan(Scan scan, RegionScanner s) {
+ byte[] topN = scan.getAttribute(BaseScannerRegionObserver.TOPN);
+ if (topN == null) {
+ return null;
+ }
+ ByteArrayInputStream stream = new ByteArrayInputStream(topN); // TODO: size?
+ try {
+ DataInputStream input = new DataInputStream(stream);
+ int thresholdBytes = WritableUtils.readVInt(input);
+ int limit = WritableUtils.readVInt(input);
+ int estimatedRowSize = WritableUtils.readVInt(input);
+ int size = WritableUtils.readVInt(input);
+ List<OrderByExpression> orderByExpressions = Lists.newArrayListWithExpectedSize(size);
+ for (int i = 0; i < size; i++) {
+ OrderByExpression orderByExpression = new OrderByExpression();
+ orderByExpression.readFields(input);
+ orderByExpressions.add(orderByExpression);
+ }
+ PTable.QualifierEncodingScheme encodingScheme = EncodedColumnsUtil.getQualifierEncodingScheme(scan);
+ ResultIterator inner = new RegionScannerResultIterator(s, EncodedColumnsUtil.getMinMaxQualifiersFromScan(scan), encodingScheme);
+ return new OrderedResultIterator(inner, orderByExpressions, thresholdBytes, limit >= 0 ? limit : null, null,
+ estimatedRowSize);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ stream.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ private Expression[] deserializeArrayPostionalExpressionInfoFromScan(Scan scan, RegionScanner s,
+ Set<KeyValueColumnExpression> arrayKVRefs) {
+ byte[] specificArrayIdx = scan.getAttribute(BaseScannerRegionObserver.SPECIFIC_ARRAY_INDEX);
+ if (specificArrayIdx == null) {
+ return null;
+ }
+ KeyValueSchema.KeyValueSchemaBuilder builder = new KeyValueSchema.KeyValueSchemaBuilder(0);
+ ByteArrayInputStream stream = new ByteArrayInputStream(specificArrayIdx);
+ try {
+ DataInputStream input = new DataInputStream(stream);
+ int arrayKVRefSize = WritableUtils.readVInt(input);
+ for (int i = 0; i < arrayKVRefSize; i++) {
+ PTable.ImmutableStorageScheme scheme = EncodedColumnsUtil.getImmutableStorageScheme(scan);
+ KeyValueColumnExpression kvExp = scheme != PTable.ImmutableStorageScheme.ONE_CELL_PER_COLUMN ? new SingleCellColumnExpression()
+ : new KeyValueColumnExpression();
+ kvExp.readFields(input);
+ arrayKVRefs.add(kvExp);
+ }
+ int arrayKVFuncSize = WritableUtils.readVInt(input);
+ Expression[] arrayFuncRefs = new Expression[arrayKVFuncSize];
+ for (int i = 0; i < arrayKVFuncSize; i++) {
+ ArrayIndexFunction arrayIdxFunc = new ArrayIndexFunction();
+ arrayIdxFunc.readFields(input);
+ arrayFuncRefs[i] = arrayIdxFunc;
+ builder.addField(arrayIdxFunc);
+ }
+ kvSchema = builder.build();
+ kvSchemaBitSet = ValueBitSet.newInstance(kvSchema);
+ return arrayFuncRefs;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ stream.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+
+ private RegionScanner getOffsetScanner(final RegionScanner s,
+ final OffsetResultIterator iterator, final boolean isLastScan) throws IOException {
+ final Tuple firstTuple;
+ final Region region = getRegion();
+ region.startRegionOperation();
+ try {
+ Tuple tuple = iterator.next();
+ if (tuple == null && !isLastScan) {
+ List<KeyValue> kvList = new ArrayList<KeyValue>(1);
+ KeyValue kv = new KeyValue(QueryConstants.OFFSET_ROW_KEY_BYTES, QueryConstants.OFFSET_FAMILY,
+ QueryConstants.OFFSET_COLUMN, PInteger.INSTANCE.toBytes(iterator.getRemainingOffset()));
+ kvList.add(kv);
+ Result r = new Result(kvList);
+ firstTuple = new ResultTuple(r);
+ } else {
+ firstTuple = tuple;
+ }
+ } catch (Throwable t) {
+ ServerUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), t);
+ return null;
+ } finally {
+ region.closeRegionOperation();
+ }
+ return new BaseRegionScanner(s) {
+ private Tuple tuple = firstTuple;
+
+ @Override
+ public boolean isFilterDone() {
+ return tuple == null;
+ }
+
+ @Override
+ public boolean next(List<Cell> results) throws IOException {
+ try {
+ if (isFilterDone()) { return false; }
+ for (int i = 0; i < tuple.size(); i++) {
+ results.add(tuple.getValue(i));
+ }
+ tuple = iterator.next();
+ return !isFilterDone();
+ } catch (Throwable t) {
+ ServerUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), t);
+ return false;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ s.close();
+ } finally {
+ try {
+ if (iterator != null) {
+ iterator.close();
+ }
+ } catch (SQLException e) {
+ ServerUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), e);
+ }
+ }
+ }
+ };
+ }
+
+ /**
+ * Return region scanner that does TopN.
+ * We only need to call startRegionOperation and closeRegionOperation when
+ * getting the first Tuple (which forces running through the entire region)
+ * since after this everything is held in memory
+ */
+ private RegionScanner getTopNScanner(RegionCoprocessorEnvironment env, final RegionScanner s,
+ final OrderedResultIterator iterator, ImmutableBytesPtr tenantId) throws Throwable {
+
+ final Tuple firstTuple;
+ TenantCache tenantCache = GlobalCache.getTenantCache(env, tenantId);
+ long estSize = iterator.getEstimatedByteSize();
+ final MemoryManager.MemoryChunk chunk = tenantCache.getMemoryManager().allocate(estSize);
+ final Region region = getRegion();
+ region.startRegionOperation();
+ try {
+ // Once we return from the first call to next, we've run through and cached
+ // the topN rows, so we no longer need to start/stop a region operation.
+ firstTuple = iterator.next();
+ // Now that the topN are cached, we can resize based on the real size
+ long actualSize = iterator.getByteSize();
+ chunk.resize(actualSize);
+ } catch (Throwable t) {
+ ServerUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), t);
+ return null;
+ } finally {
+ region.closeRegionOperation();
+ }
+ return new BaseRegionScanner(s) {
+ private Tuple tuple = firstTuple;
+
+ @Override
+ public boolean isFilterDone() {
+ return tuple == null;
+ }
+
+ @Override
+ public boolean next(List<Cell> results) throws IOException {
+ try {
+ if (isFilterDone()) {
+ return false;
+ }
+
+ for (int i = 0; i < tuple.size(); i++) {
+ results.add(tuple.getValue(i));
+ }
+
+ tuple = iterator.next();
+ return !isFilterDone();
+ } catch (Throwable t) {
+ ServerUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), t);
+ return false;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ s.close();
+ } finally {
+ try {
+ if(iterator != null) {
+ iterator.close();
+ }
+ } catch (SQLException e) {
+ ServerUtil.throwIOException(region.getRegionInfo().getRegionNameAsString(), e);
+ } finally {
+ chunk.close();
+ }
+ }
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelScanGrouper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelScanGrouper.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelScanGrouper.java
index 0becf4f..5e237a5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelScanGrouper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelScanGrouper.java
@@ -17,10 +17,13 @@
*/
package org.apache.phoenix.iterate;
+import java.sql.SQLException;
import java.util.List;
+import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.phoenix.compile.QueryPlan;
+import org.apache.phoenix.compile.StatementContext;
/**
* Interface for a parallel scan grouper
@@ -29,7 +32,7 @@ public interface ParallelScanGrouper {
/**
* Determines whether to create a new group of parallel scans.
- *
+ *
* @param scans current scan group
* @param plan current query plan
* @param startKey start key of scan
@@ -37,5 +40,7 @@ public interface ParallelScanGrouper {
* @return true if we should create a new group of scans
*/
boolean shouldStartNewScan(QueryPlan plan, List<Scan> scans, byte[] startKey, boolean crossedRegionBoundary);
-
+
+ List<HRegionLocation> getRegionBoundaries(StatementContext context, byte[] tableName) throws SQLException;
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/faadb544/phoenix-core/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java
new file mode 100644
index 0000000..252773c
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java
@@ -0,0 +1,322 @@
+package org.apache.phoenix.iterate;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.hadoop.hbase.*;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.regionserver.Region;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.hadoop.hbase.regionserver.ScannerContext;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.execute.TupleProjector;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.expression.KeyValueColumnExpression;
+import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
+import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.KeyValueSchema;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.ValueBitSet;
+import org.apache.phoenix.schema.tuple.*;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.ScanUtil;
+import org.apache.phoenix.util.ServerUtil;
+import org.apache.tephra.Transaction;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Set;
+
+public abstract class RegionScannerFactory {
+
+ protected RegionCoprocessorEnvironment env;
+ protected boolean useNewValueColumnQualifier;
+ protected PTable.QualifierEncodingScheme encodingScheme;
+
+ /**
+ * Returns the region based on the value of the
+ * region context
+ * @return
+ */
+ public Region getRegion() {
+ return env.getRegion();
+ }
+
+ /**
+ * Returns a processed region scanner based on the query
+ * conditions. Thie functionality is abstracted out of
+ * the non-aggregate region observer class for better
+ * usage
+ * @param scan input scan
+ * @param s input region scanner
+ * @return
+ * @throws Throwable
+ */
+ public abstract RegionScanner getRegionScanner(final Scan scan, final RegionScanner s) throws Throwable;
+
+ /**
+ * Return wrapped scanner that catches unexpected exceptions (i.e. Phoenix bugs) and
+ * re-throws as DoNotRetryIOException to prevent needless retrying hanging the query
+ * for 30 seconds. Unfortunately, until HBASE-7481 gets fixed, there's no way to do
+ * the same from a custom filter.
+ * @param arrayKVRefs
+ * @param arrayFuncRefs
+ * @param offset starting position in the rowkey.
+ * @param scan
+ * @param tupleProjector
+ * @param dataRegion
+ * @param indexMaintainer
+ * @param tx current transaction
+ * @param viewConstants
+ */
+ public RegionScanner getWrappedScanner(final RegionCoprocessorEnvironment env,
+ final RegionScanner s, final Set<KeyValueColumnExpression> arrayKVRefs,
+ final Expression[] arrayFuncRefs, final int offset, final Scan scan,
+ final ColumnReference[] dataColumns, final TupleProjector tupleProjector,
+ final Region dataRegion, final IndexMaintainer indexMaintainer,
+ Transaction tx,
+ final byte[][] viewConstants, final KeyValueSchema kvSchema,
+ final ValueBitSet kvSchemaBitSet, final TupleProjector projector,
+ final ImmutableBytesWritable ptr, final boolean useQualifierAsListIndex) {
+ return new RegionScanner() {
+
+ private boolean hasReferences = checkForReferenceFiles();
+ private HRegionInfo regionInfo = env.getRegionInfo();
+ private byte[] actualStartKey = getActualStartKey();
+
+ // If there are any reference files after local index region merge some cases we might
+ // get the records less than scan start row key. This will happen when we replace the
+ // actual region start key with merge region start key. This method gives whether are
+ // there any reference files in the region or not.
+ private boolean checkForReferenceFiles() {
+ if(!ScanUtil.isLocalIndex(scan)) return false;
+ for (byte[] family : scan.getFamilies()) {
+ if (getRegion().getStore(family).hasReferences()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ // Get the actual scan start row of local index. This will be used to compare the row
+ // key of the results less than scan start row when there are references.
+ public byte[] getActualStartKey() {
+ return ScanUtil.isLocalIndex(scan) ? ScanUtil.getActualStartRow(scan, regionInfo)
+ : null;
+ }
+
+ @Override
+ public boolean next(List<Cell> results) throws IOException {
+ try {
+ return s.next(results);
+ } catch (Throwable t) {
+ ServerUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), t);
+ return false; // impossible
+ }
+ }
+
+ @Override
+ public boolean next(List<Cell> result, ScannerContext scannerContext) throws IOException {
+ try {
+ return s.next(result, scannerContext);
+ } catch (Throwable t) {
+ ServerUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), t);
+ return false; // impossible
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ s.close();
+ }
+
+ @Override
+ public HRegionInfo getRegionInfo() {
+ return s.getRegionInfo();
+ }
+
+ @Override
+ public boolean isFilterDone() throws IOException {
+ return s.isFilterDone();
+ }
+
+ @Override
+ public boolean reseek(byte[] row) throws IOException {
+ return s.reseek(row);
+ }
+
+ @Override
+ public long getMvccReadPoint() {
+ return s.getMvccReadPoint();
+ }
+
+ @Override
+ public boolean nextRaw(List<Cell> result) throws IOException {
+ try {
+ boolean next = s.nextRaw(result);
+ Cell arrayElementCell = null;
+ if (result.size() == 0) {
+ return next;
+ }
+ if (arrayFuncRefs != null && arrayFuncRefs.length > 0 && arrayKVRefs.size() > 0) {
+ int arrayElementCellPosition = replaceArrayIndexElement(arrayKVRefs, arrayFuncRefs, result);
+ arrayElementCell = result.get(arrayElementCellPosition);
+ }
+ if (ScanUtil.isLocalIndex(scan) && !ScanUtil.isAnalyzeTable(scan)) {
+ if(hasReferences && actualStartKey!=null) {
+ next = scanTillScanStartRow(s, arrayKVRefs, arrayFuncRefs, result,
+ null, arrayElementCell);
+ if (result.isEmpty()) {
+ return next;
+ }
+ }
+ /* In the following, c is only used when data region is null.
+ dataRegion will never be null in case of non-coprocessor call,
+ therefore no need to refactor
+ */
+ IndexUtil.wrapResultUsingOffset(env, result, offset, dataColumns,
+ tupleProjector, dataRegion, indexMaintainer, viewConstants, ptr);
+ }
+ if (projector != null) {
+ Tuple toProject = useQualifierAsListIndex ? new PositionBasedResultTuple(result) : new ResultTuple(
+ Result.create(result));
+ Tuple tuple = projector.projectResults(toProject, useNewValueColumnQualifier);
+ result.clear();
+ result.add(tuple.getValue(0));
+ if (arrayElementCell != null) {
+ result.add(arrayElementCell);
+ }
+ }
+ // There is a scanattribute set to retrieve the specific array element
+ return next;
+ } catch (Throwable t) {
+ ServerUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), t);
+ return false; // impossible
+ }
+ }
+
+ @Override
+ public boolean nextRaw(List<Cell> result, ScannerContext scannerContext)
+ throws IOException {
+ try {
+ boolean next = s.nextRaw(result, scannerContext);
+ Cell arrayElementCell = null;
+ if (result.size() == 0) {
+ return next;
+ }
+ if (arrayFuncRefs != null && arrayFuncRefs.length > 0 && arrayKVRefs.size() > 0) {
+ int arrayElementCellPosition = replaceArrayIndexElement(arrayKVRefs, arrayFuncRefs, result);
+ arrayElementCell = result.get(arrayElementCellPosition);
+ }
+ if ((offset > 0 || ScanUtil.isLocalIndex(scan)) && !ScanUtil.isAnalyzeTable(scan)) {
+ if(hasReferences && actualStartKey!=null) {
+ next = scanTillScanStartRow(s, arrayKVRefs, arrayFuncRefs, result,
+ scannerContext, arrayElementCell);
+ if (result.isEmpty()) {
+ return next;
+ }
+ }
+ /* In the following, c is only used when data region is null.
+ dataRegion will never be null in case of non-coprocessor call,
+ therefore no need to refactor
+ */
+ IndexUtil.wrapResultUsingOffset(env, result, offset, dataColumns,
+ tupleProjector, dataRegion, indexMaintainer, viewConstants, ptr);
+ }
+ if (projector != null) {
+ Tuple toProject = useQualifierAsListIndex ? new PositionBasedMultiKeyValueTuple(result) : new ResultTuple(Result.create(result));
+ Tuple tuple = projector.projectResults(toProject, useNewValueColumnQualifier);
+ result.clear();
+ result.add(tuple.getValue(0));
+ if(arrayElementCell != null)
+ result.add(arrayElementCell);
+ }
+ // There is a scanattribute set to retrieve the specific array element
+ return next;
+ } catch (Throwable t) {
+ ServerUtil.throwIOException(getRegion().getRegionInfo().getRegionNameAsString(), t);
+ return false; // impossible
+ }
+ }
+
+ /**
+ * When there is a merge in progress while scanning local indexes we might get the key values less than scan start row.
+ * In that case we need to scan until get the row key more or equal to scan start key.
+ * TODO try to fix this case in LocalIndexStoreFileScanner when there is a merge.
+ */
+ private boolean scanTillScanStartRow(final RegionScanner s,
+ final Set<KeyValueColumnExpression> arrayKVRefs,
+ final Expression[] arrayFuncRefs, List<Cell> result,
+ ScannerContext scannerContext, Cell arrayElementCell) throws IOException {
+ boolean next = true;
+ Cell firstCell = result.get(0);
+ while (Bytes.compareTo(firstCell.getRowArray(), firstCell.getRowOffset(),
+ firstCell.getRowLength(), actualStartKey, 0, actualStartKey.length) < 0) {
+ result.clear();
+ if(scannerContext == null) {
+ next = s.nextRaw(result);
+ } else {
+ next = s.nextRaw(result, scannerContext);
+ }
+ if (result.isEmpty()) {
+ return next;
+ }
+ if (arrayFuncRefs != null && arrayFuncRefs.length > 0 && arrayKVRefs.size() > 0) {
+ int arrayElementCellPosition = replaceArrayIndexElement(arrayKVRefs, arrayFuncRefs, result);
+ arrayElementCell = result.get(arrayElementCellPosition);
+ }
+ firstCell = result.get(0);
+ }
+ return next;
+ }
+
+ private int replaceArrayIndexElement(final Set<KeyValueColumnExpression> arrayKVRefs,
+ final Expression[] arrayFuncRefs, List<Cell> result) {
+ // make a copy of the results array here, as we're modifying it below
+ MultiKeyValueTuple tuple = new MultiKeyValueTuple(ImmutableList.copyOf(result));
+ // The size of both the arrays would be same?
+ // Using KeyValueSchema to set and retrieve the value
+ // collect the first kv to get the row
+ Cell rowKv = result.get(0);
+ for (KeyValueColumnExpression kvExp : arrayKVRefs) {
+ if (kvExp.evaluate(tuple, ptr)) {
+ ListIterator<Cell> itr = result.listIterator();
+ while (itr.hasNext()) {
+ Cell kv = itr.next();
+ if (Bytes.equals(kvExp.getColumnFamily(), 0, kvExp.getColumnFamily().length,
+ kv.getFamilyArray(), kv.getFamilyOffset(), kv.getFamilyLength())
+ && Bytes.equals(kvExp.getColumnQualifier(), 0, kvExp.getColumnQualifier().length,
+ kv.getQualifierArray(), kv.getQualifierOffset(), kv.getQualifierLength())) {
+ // remove the kv that has the full array values.
+ itr.remove();
+ break;
+ }
+ }
+ }
+ }
+ byte[] value = kvSchema.toBytes(tuple, arrayFuncRefs,
+ kvSchemaBitSet, ptr);
+ // Add a dummy kv with the exact value of the array index
+ result.add(new KeyValue(rowKv.getRowArray(), rowKv.getRowOffset(), rowKv.getRowLength(),
+ QueryConstants.ARRAY_VALUE_COLUMN_FAMILY, 0, QueryConstants.ARRAY_VALUE_COLUMN_FAMILY.length,
+ QueryConstants.ARRAY_VALUE_COLUMN_QUALIFIER, 0,
+ QueryConstants.ARRAY_VALUE_COLUMN_QUALIFIER.length, HConstants.LATEST_TIMESTAMP,
+ KeyValue.Type.codeToType(rowKv.getTypeByte()), value, 0, value.length));
+ return result.size() - 1;
+ }
+
+ @Override
+ public long getMaxResultSize() {
+ return s.getMaxResultSize();
+ }
+
+ @Override
+ public int getBatch() {
+ return s.getBatch();
+ }
+ };
+ }
+}