You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ka...@apache.org on 2022/03/18 02:56:00 UTC
[phoenix] branch 4.x updated: PHOENIX-6663 Use batching when joining data table rows with uncovered local index rows (#1403)
This is an automated email from the ASF dual-hosted git repository.
kadir pushed a commit to branch 4.x
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/4.x by this push:
new 42e2e6f PHOENIX-6663 Use batching when joining data table rows with uncovered local index rows (#1403)
42e2e6f is described below
commit 42e2e6f1b26f4f6d22f89d061cce09e5d110513d
Author: kadirozde <37...@users.noreply.github.com>
AuthorDate: Thu Mar 17 19:17:38 2022 -0700
PHOENIX-6663 Use batching when joining data table rows with uncovered local index rows (#1403)
---
.../UncoveredGlobalIndexRegionScanner.java | 241 ++-------------
.../coprocessor/UncoveredIndexRegionScanner.java | 330 +++++++++++++++++++++
.../UncoveredLocalIndexRegionScanner.java | 128 ++++++++
.../phoenix/iterate/RegionScannerFactory.java | 42 +--
.../java/org/apache/phoenix/util/IndexUtil.java | 67 +----
5 files changed, 521 insertions(+), 287 deletions(-)
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java
index d2f8b2a..96997fe 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java
@@ -19,50 +19,41 @@ package org.apache.phoenix.coprocessor;
import static org.apache.phoenix.coprocessor.BaseScannerRegionObserver.PHYSICAL_DATA_TABLE_NAME;
import static org.apache.phoenix.hbase.index.write.AbstractParallelWriterIndexCommitter.INDEX_WRITER_KEEP_ALIVE_TIME_CONF_KEY;
-import static org.apache.phoenix.query.QueryServices.INDEX_PAGE_SIZE_IN_ROWS;
-import static org.apache.phoenix.util.ScanUtil.getDummyResult;
-import static org.apache.phoenix.util.ScanUtil.isDummy;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.Collection;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.execute.TupleProjector;
-import org.apache.phoenix.hbase.index.parallel.*;
+import org.apache.phoenix.hbase.index.parallel.EarlyExitFailure;
+import org.apache.phoenix.hbase.index.parallel.Task;
+import org.apache.phoenix.hbase.index.parallel.TaskBatch;
+import org.apache.phoenix.hbase.index.parallel.TaskRunner;
+import org.apache.phoenix.hbase.index.parallel.ThreadPoolBuilder;
+import org.apache.phoenix.hbase.index.parallel.ThreadPoolManager;
+import org.apache.phoenix.hbase.index.parallel.WaitForCompletionTaskRunner;
import org.apache.phoenix.hbase.index.table.HTableFactory;
import org.apache.phoenix.hbase.index.write.IndexWriterUtils;
import org.apache.phoenix.index.IndexMaintainer;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.regionserver.Region;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.query.HBaseFactoryProvider;
-import org.apache.phoenix.query.QueryServicesOptions;
-import org.apache.phoenix.schema.tuple.ResultTuple;
-import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
-import org.apache.phoenix.compile.ScanRanges;
-import org.apache.phoenix.filter.SkipScanFilter;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
-import org.apache.phoenix.query.KeyRange;
-import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.util.EnvironmentEdgeManager;
-import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.ServerUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -72,44 +63,20 @@ import org.slf4j.LoggerFactory;
* data table row keys from them. Using the data table row keys, the data table rows are scanned
* using the HBase client available to region servers.
*/
-public class UncoveredGlobalIndexRegionScanner extends BaseRegionScanner {
+public class UncoveredGlobalIndexRegionScanner extends UncoveredIndexRegionScanner {
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(UncoveredGlobalIndexRegionScanner.class);
public static final String NUM_CONCURRENT_INDEX_THREADS_CONF_KEY = "index.threads.max";
public static final int DEFAULT_CONCURRENT_INDEX_THREADS = 16;
public static final String INDEX_ROW_COUNTS_PER_TASK_CONF_KEY = "index.row.count.per.task";
public static final int DEFAULT_INDEX_ROW_COUNTS_PER_TASK = 2048;
- /**
- * The states of the processing a page of index rows
- */
- private enum State {
- INITIAL, SCANNING_INDEX, SCANNING_DATA, SCANNING_DATA_INTERRUPTED, READY
- }
- State state = State.INITIAL;
- private static final Logger LOGGER =
- LoggerFactory.getLogger(UncoveredGlobalIndexRegionScanner.class);
- protected final byte[][] viewConstants;
- protected final RegionCoprocessorEnvironment env;
protected byte[][] regionEndKeys;
protected final Table dataHTable;
- protected final int pageSizeInRows;
protected final int rowCountPerTask;
- protected final Scan scan;
- protected final Scan dataTableScan;
- protected final RegionScanner innerScanner;
- protected final Region region;
- protected final IndexMaintainer indexMaintainer;
- protected final TupleProjector tupleProjector;
- protected final ImmutableBytesWritable ptr;
protected final TaskRunner pool;
protected String exceptionMessage;
protected final HTableFactory hTableFactory;
- protected List<List<Cell>> indexRows = null;
- protected Map<ImmutableBytesPtr, Result> dataRows = null;
- protected Iterator<List<Cell>> indexRowIterator = null;
- protected Map<byte[], byte[]> indexToDataRowKeyMap = null;
- protected int indexRowCount = 0;
- protected final long pageSizeMs;
- protected byte[] lastIndexRowKey = null;
public UncoveredGlobalIndexRegionScanner(final RegionScanner innerScanner,
final Region region,
@@ -122,29 +89,9 @@ public class UncoveredGlobalIndexRegionScanner extends BaseRegionScanner {
final ImmutableBytesWritable ptr,
final long pageSizeMs)
throws IOException {
- super(innerScanner);
+ super(innerScanner, region, scan, env, dataTableScan, tupleProjector, indexMaintainer,
+ viewConstants, ptr, pageSizeMs);
final Configuration config = env.getConfiguration();
-
- byte[] pageSizeFromScan =
- scan.getAttribute(BaseScannerRegionObserver.INDEX_PAGE_ROWS);
- if (pageSizeFromScan != null) {
- pageSizeInRows = (int) Bytes.toLong(pageSizeFromScan);
- } else {
- pageSizeInRows = (int)
- config.getLong(INDEX_PAGE_SIZE_IN_ROWS,
- QueryServicesOptions.DEFAULT_INDEX_PAGE_SIZE_IN_ROWS);
- }
-
- this.indexMaintainer = indexMaintainer;
- this.viewConstants = viewConstants;
- this.scan = scan;
- this.dataTableScan = dataTableScan;
- this.innerScanner = innerScanner;
- this.region = region;
- this.env = env;
- this.ptr = ptr;
- this.tupleProjector = tupleProjector;
- this.pageSizeMs = pageSizeMs;
hTableFactory = IndexWriterUtils.getDefaultDelegateHTableFactory(env);
rowCountPerTask = config.getInt(INDEX_ROW_COUNTS_PER_TASK_CONF_KEY,
DEFAULT_INDEX_ROW_COUNTS_PER_TASK);
@@ -164,21 +111,6 @@ public class UncoveredGlobalIndexRegionScanner extends BaseRegionScanner {
}
@Override
- public long getMvccReadPoint() {
- return innerScanner.getMvccReadPoint();
- }
-
- @Override
- public HRegionInfo getRegionInfo() {
- return region.getRegionInfo();
- }
-
- @Override
- public boolean isFilterDone() {
- return false;
- }
-
- @Override
public void close() throws IOException {
innerScanner.close();
hTableFactory.shutdown();
@@ -188,40 +120,27 @@ public class UncoveredGlobalIndexRegionScanner extends BaseRegionScanner {
this.pool.stop("UncoveredGlobalIndexRegionScanner is closing");
}
- @Override
- public long getMaxResultSize() {
- return innerScanner.getMaxResultSize();
- }
-
- @Override
- public int getBatch() {
- return innerScanner.getBatch();
- }
-
- private void scanDataRows(Set<byte[]> dataRowKeys, long startTime) throws IOException {
- List<KeyRange> keys = new ArrayList<>(dataRowKeys.size());
- for (byte[] dataRowKey: dataRowKeys) {
- keys.add(PVarbinary.INSTANCE.getKeyRange(dataRowKey));
- }
- ScanRanges scanRanges = ScanRanges.createPointLookup(keys);
- Scan dataScan = new Scan(dataTableScan);
- dataScan.setTimeRange(scan.getTimeRange().getMin(), scan.getTimeRange().getMax());
- scanRanges.initializeScan(dataScan);
- SkipScanFilter skipScanFilter = scanRanges.getSkipScanFilter();
- dataScan.setFilter(new SkipScanFilter(skipScanFilter, false));
+ protected void scanDataRows(Collection<byte[]> dataRowKeys, long startTime) throws IOException {
+ Scan dataScan = prepareDataTableScan(dataRowKeys);
try (ResultScanner resultScanner = dataHTable.getScanner(dataScan)) {
for (Result result = resultScanner.next(); (result != null);
result = resultScanner.next()) {
+ if (ScanUtil.isDummy(result)) {
+ state = State.SCANNING_DATA_INTERRUPTED;
+ break;
+ }
dataRows.put(new ImmutableBytesPtr(result.getRow()), result);
if ((EnvironmentEdgeManager.currentTimeMillis() - startTime) >= pageSizeMs) {
- LOGGER.info("One of the scan tasks in UncoveredGlobalIndexRegionScanner"
- + " for region " + region.getRegionInfo().getRegionNameAsString()
- + " could not complete on time (in " + pageSizeMs+ " ms) and"
- + " will be resubmitted");
state = State.SCANNING_DATA_INTERRUPTED;
break;
}
}
+ if (state == State.SCANNING_DATA_INTERRUPTED) {
+ LOGGER.info("One of the scan tasks in UncoveredGlobalIndexRegionScanner"
+ + " for region " + region.getRegionInfo().getRegionNameAsString()
+ + " could not complete on time (in " + pageSizeMs + " ms) and"
+ + " will be resubmitted");
+ }
} catch (Throwable t) {
exceptionMessage = "scanDataRows fails for at least one task";
ServerUtil.throwIOException(dataHTable.getName().toString(), t);
@@ -275,8 +194,8 @@ public class UncoveredGlobalIndexRegionScanner extends BaseRegionScanner {
}
}
- private void scanDataTableRows(long startTime)
- throws IOException {
+ @Override
+ protected void scanDataTableRows(long startTime) throws IOException {
if (indexToDataRowKeyMap.size() == 0) {
state = State.READY;
return;
@@ -299,108 +218,4 @@ public class UncoveredGlobalIndexRegionScanner extends BaseRegionScanner {
state = State.READY;
}
}
-
- /**
- * A page of index rows are scanned and then their corresponding data table rows are retrieved
- * from the data table regions in parallel. These data rows are then joined with index rows.
- * The join is for adding uncovered columns to index rows.
- *
- * This implementation conforms to server paging such that if the server side operation takes
- * more than pageSizeInMs, a dummy result is returned to signal the client that more work
- * to do on the server side. This is done to prevent RPC timeouts.
- *
- * @param result
- * @return boolean to indicate if there are more rows to scan
- * @throws IOException
- */
- @Override
- public boolean next(List<Cell> result) throws IOException {
- long startTime = EnvironmentEdgeManager.currentTimeMillis();
- boolean hasMore;
- region.startRegionOperation();
- try {
- synchronized (innerScanner) {
- if (state == State.READY && !indexRowIterator.hasNext()) {
- state = State.INITIAL;
- }
- if (state == State.INITIAL) {
- indexRowCount = 0;
- indexRows = new ArrayList();
- dataRows = Maps.newConcurrentMap();
- indexToDataRowKeyMap = Maps.newTreeMap(Bytes.BYTES_COMPARATOR);
- state = State.SCANNING_INDEX;
- }
- if (state == State.SCANNING_INDEX) {
- do {
- List<Cell> row = new ArrayList<Cell>();
- hasMore = innerScanner.nextRaw(row);
- if (!row.isEmpty()) {
- if (isDummy(row)) {
- result.addAll(row);
- // We got a dummy request from lower layers. This means that
- // the scan took more than pageSizeMs. Just return true here.
- // The client will drop this dummy request and continue to scan.
- // Then the lower layer scanner will continue
- // wherever it stopped due to this dummy request
- return true;
- }
- lastIndexRowKey = CellUtil.cloneRow(row.get(0));
- indexToDataRowKeyMap.put(lastIndexRowKey, indexMaintainer.buildDataRowKey(
- new ImmutableBytesWritable(lastIndexRowKey), viewConstants));
- indexRows.add(row);
- indexRowCount++;
- if (hasMore && (EnvironmentEdgeManager.currentTimeMillis() - startTime)
- >= pageSizeMs) {
- getDummyResult(lastIndexRowKey, result);
- // We do not need to change the state, State.SCANNING_INDEX
- // since we will continue scanning the index table after
- // the client drops the dummy request and then calls the next
- // method on its ResultScanner within ScanningResultIterator
- return true;
- }
- }
- } while (hasMore && indexRowCount < pageSizeInRows);
- state = State.SCANNING_DATA;
- }
- if (state == State.SCANNING_DATA) {
- scanDataTableRows(startTime);
- indexRowIterator = indexRows.iterator();
- }
- if (state == State.READY) {
- if (indexRowIterator.hasNext()) {
- List<Cell> indexRow = indexRowIterator.next();
- result.addAll(indexRow);
- try {
- Result dataRow = dataRows.get(new ImmutableBytesPtr(
- indexToDataRowKeyMap.get(CellUtil.cloneRow(indexRow.get(0)))));
- if (dataRow != null) {
- IndexUtil.addTupleAsOneCell(result, new ResultTuple(dataRow),
- tupleProjector, ptr);
- } else {
- // The data row does not exist, we should skip this index row. This can happen
- // if index row is replicated but the data row has not been
- result.clear();
- }
- } catch (Throwable e) {
- LOGGER.error("Exception in UncoveredGlobalIndexRegionScanner for region "
- + region.getRegionInfo().getRegionNameAsString(), e);
- throw e;
- }
- return true;
- } else {
- return false;
- }
- } else {
- getDummyResult(lastIndexRowKey, result);
- return true;
- }
- }
- } catch (Throwable e) {
- LOGGER.error("Exception in UncoveredGlobalIndexRegionScanner for region "
- + region.getRegionInfo().getRegionNameAsString(), e);
- throw e;
- } finally {
- region.closeRegionOperation();
- }
- }
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredIndexRegionScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredIndexRegionScanner.java
new file mode 100644
index 0000000..50b26d3
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredIndexRegionScanner.java
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.coprocessor;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.regionserver.Region;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.compile.ScanRanges;
+import org.apache.phoenix.execute.TupleProjector;
+import org.apache.phoenix.filter.SkipScanFilter;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.tuple.ResultTuple;
+import org.apache.phoenix.schema.types.PVarbinary;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.ScanUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.phoenix.query.QueryServices.INDEX_PAGE_SIZE_IN_ROWS;
+import static org.apache.phoenix.util.ScanUtil.getDummyResult;
+import static org.apache.phoenix.util.ScanUtil.isDummy;
+
+public abstract class UncoveredIndexRegionScanner extends BaseRegionScanner {
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(UncoveredIndexRegionScanner.class);
+ /**
+ * The states of the processing a page of index rows
+ */
+ protected enum State {
+ INITIAL, SCANNING_INDEX, SCANNING_DATA, SCANNING_DATA_INTERRUPTED, READY
+ }
+ protected State state = State.INITIAL;
+ protected final byte[][] viewConstants;
+ protected final RegionCoprocessorEnvironment env;
+ protected byte[][] regionEndKeys;
+ protected final int pageSizeInRows;
+ protected final Scan scan;
+ protected final Scan dataTableScan;
+ protected final RegionScanner innerScanner;
+ protected final Region region;
+ protected final IndexMaintainer indexMaintainer;
+ protected final TupleProjector tupleProjector;
+ protected final ImmutableBytesWritable ptr;
+ protected String exceptionMessage;
+ protected List<List<Cell>> indexRows = null;
+ protected Map<ImmutableBytesPtr, Result> dataRows = null;
+ protected Iterator<List<Cell>> indexRowIterator = null;
+ protected Map<byte[], byte[]> indexToDataRowKeyMap = null;
+ protected int indexRowCount = 0;
+ protected final long pageSizeMs;
+ protected byte[] lastIndexRowKey = null;
+
+ public UncoveredIndexRegionScanner(final RegionScanner innerScanner,
+ final Region region,
+ final Scan scan,
+ final RegionCoprocessorEnvironment env,
+ final Scan dataTableScan,
+ final TupleProjector tupleProjector,
+ final IndexMaintainer indexMaintainer,
+ final byte[][] viewConstants,
+ final ImmutableBytesWritable ptr,
+ final long pageSizeMs) {
+ super(innerScanner);
+ final Configuration config = env.getConfiguration();
+
+ byte[] pageSizeFromScan =
+ scan.getAttribute(BaseScannerRegionObserver.INDEX_PAGE_ROWS);
+ if (pageSizeFromScan != null) {
+ pageSizeInRows = (int) Bytes.toLong(pageSizeFromScan);
+ } else {
+ pageSizeInRows = (int)
+ config.getLong(INDEX_PAGE_SIZE_IN_ROWS,
+ QueryServicesOptions.DEFAULT_INDEX_PAGE_SIZE_IN_ROWS);
+ }
+
+ this.indexMaintainer = indexMaintainer;
+ this.viewConstants = viewConstants;
+ this.scan = scan;
+ this.dataTableScan = dataTableScan;
+ this.innerScanner = innerScanner;
+ this.region = region;
+ this.env = env;
+ this.ptr = ptr;
+ this.tupleProjector = tupleProjector;
+ this.pageSizeMs = pageSizeMs;
+ }
+
+ @Override
+ public long getMvccReadPoint() {
+ return innerScanner.getMvccReadPoint();
+ }
+
+ @Override
+ public HRegionInfo getRegionInfo() {
+ return region.getRegionInfo();
+ }
+
+ @Override
+ public boolean isFilterDone() {
+ return false;
+ }
+
+ @Override
+ public void close() throws IOException {
+ innerScanner.close();
+ }
+
+ @Override
+ public long getMaxResultSize() {
+ return innerScanner.getMaxResultSize();
+ }
+
+ @Override
+ public int getBatch() {
+ return innerScanner.getBatch();
+ }
+
+ protected abstract void scanDataTableRows(long startTime) throws IOException;
+
+ protected Scan prepareDataTableScan(Collection<byte[]> dataRowKeys) throws IOException {
+ List<KeyRange> keys = new ArrayList<>(dataRowKeys.size());
+ for (byte[] dataRowKey : dataRowKeys) {
+ keys.add(PVarbinary.INSTANCE.getKeyRange(dataRowKey));
+ }
+ ScanRanges scanRanges = ScanRanges.createPointLookup(keys);
+ Scan dataScan = new Scan(dataTableScan);
+ dataScan.setTimeRange(scan.getTimeRange().getMin(), scan.getTimeRange().getMax());
+ scanRanges.initializeScan(dataScan);
+ SkipScanFilter skipScanFilter = scanRanges.getSkipScanFilter();
+ dataScan.setFilter(new SkipScanFilter(skipScanFilter, false));
+ scan.setAttribute(BaseScannerRegionObserver.SERVER_PAGE_SIZE_MS,
+ Bytes.toBytes(Long.valueOf(pageSizeMs)));
+ return dataScan;
+ }
+
+ protected boolean scanIndexTableRows(List<Cell> result,
+ final long startTime,
+ final byte[] actualStartKey,
+ final int offset) throws IOException {
+ boolean hasMore = false;
+ if (actualStartKey != null) {
+ do {
+ hasMore = innerScanner.nextRaw(result);
+ if (result.isEmpty()) {
+ return hasMore;
+ }
+ if (ScanUtil.isDummy(result)) {
+ return true;
+ }
+ Cell firstCell = result.get(0);
+ if (Bytes.compareTo(firstCell.getRowArray(), firstCell.getRowOffset(),
+ firstCell.getRowLength(), actualStartKey, 0, actualStartKey.length) < 0) {
+ result.clear();
+ if (EnvironmentEdgeManager.currentTimeMillis() - startTime >= pageSizeMs) {
+ byte[] rowKey = CellUtil.cloneRow(firstCell);
+ ScanUtil.getDummyResult(rowKey, result);
+ return true;
+ }
+ } else {
+ break;
+ }
+ } while (hasMore);
+ }
+
+ do {
+ List<Cell> row = new ArrayList<Cell>();
+ if (result.isEmpty()) {
+ hasMore = innerScanner.nextRaw(row);
+ } else {
+ row.addAll(result);
+ result.clear();
+ }
+ if (!row.isEmpty()) {
+ if (isDummy(row)) {
+ result.addAll(row);
+ // We got a dummy request from lower layers. This means that
+ // the scan took more than pageSizeMs. Just return true here.
+ // The client will drop this dummy request and continue to scan.
+ // Then the lower layer scanner will continue
+ // wherever it stopped due to this dummy request
+ return true;
+ }
+ Cell firstCell = row.get(0);
+ byte[] indexRowKey = firstCell.getRowArray();
+ ptr.set(indexRowKey, firstCell.getRowOffset() + offset,
+ firstCell.getRowLength() - offset);
+ lastIndexRowKey = ptr.copyBytes();
+ indexToDataRowKeyMap.put(offset == 0 ? lastIndexRowKey :
+ CellUtil.cloneRow(firstCell), indexMaintainer.buildDataRowKey(
+ new ImmutableBytesWritable(lastIndexRowKey),
+ viewConstants));
+ indexRows.add(row);
+ indexRowCount++;
+ if (hasMore && (EnvironmentEdgeManager.currentTimeMillis() - startTime)
+ >= pageSizeMs) {
+ getDummyResult(lastIndexRowKey, result);
+ // We do not need to change the state, State.SCANNING_INDEX
+ // since we will continue scanning the index table after
+ // the client drops the dummy request and then calls the next
+ // method on its ResultScanner within ScanningResultIterator
+ return true;
+ }
+ }
+ } while (hasMore && indexRowCount < pageSizeInRows);
+ return hasMore;
+ }
+
+ protected boolean scanIndexTableRows(List<Cell> result,
+ final long startTime) throws IOException {
+ return scanIndexTableRows(result, startTime, null, 0);
+ }
+
+ private boolean getNextCoveredIndexRow(List<Cell> result) {
+ if (indexRowIterator.hasNext()) {
+ List<Cell> indexRow = indexRowIterator.next();
+ result.addAll(indexRow);
+ try {
+ Result dataRow = dataRows.get(new ImmutableBytesPtr(
+ indexToDataRowKeyMap.get(CellUtil.cloneRow(indexRow.get(0)))));
+ if (dataRow != null) {
+ IndexUtil.addTupleAsOneCell(result, new ResultTuple(dataRow),
+ tupleProjector, ptr);
+ } else {
+ // The data row satisfying the scan does not exist. This could be because
+ // the data row may not include the columns corresponding to the uncovered
+ // index columns either. Just return the index row. Nothing to do here
+ }
+ } catch (Throwable e) {
+ LOGGER.error("Exception in UncoveredIndexRegionScanner for region "
+ + region.getRegionInfo().getRegionNameAsString(), e);
+ throw e;
+ }
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ /**
+ * A page of index rows are scanned and then their corresponding data table rows are retrieved
+ * from the data table regions in parallel. These data rows are then joined with index rows.
+ * The join is for adding uncovered columns to index rows.
+ *
+ * This implementation conforms to server paging such that if the server side operation takes
+ * more than pageSizeInMs, a dummy result is returned to signal the client that more work
+ * to do on the server side. This is done to prevent RPC timeouts.
+ *
+ * @param result
+ * @return boolean to indicate if there are more rows to scan
+ * @throws IOException
+ */
+ @Override
+ public boolean next(List<Cell> result) throws IOException {
+ long startTime = EnvironmentEdgeManager.currentTimeMillis();
+ boolean hasMore;
+ region.startRegionOperation();
+ try {
+ synchronized (innerScanner) {
+ if (state == State.READY && !indexRowIterator.hasNext()) {
+ state = State.INITIAL;
+ }
+ if (state == State.INITIAL) {
+ indexRowCount = 0;
+ indexRows = new ArrayList<>();
+ dataRows = Maps.newConcurrentMap();
+ indexToDataRowKeyMap = Maps.newTreeMap(Bytes.BYTES_COMPARATOR);
+ state = State.SCANNING_INDEX;
+ }
+ if (state == State.SCANNING_INDEX) {
+ hasMore = scanIndexTableRows(result, startTime);
+ if (isDummy(result)) {
+ return hasMore;
+ }
+ state = State.SCANNING_DATA;
+ }
+ if (state == State.SCANNING_DATA) {
+ scanDataTableRows(startTime);
+ indexRowIterator = indexRows.iterator();
+ }
+ if (state == State.READY) {
+ return getNextCoveredIndexRow(result);
+ } else {
+ getDummyResult(lastIndexRowKey, result);
+ return true;
+ }
+ }
+ } catch (Throwable e) {
+ LOGGER.error("Exception in UncoveredIndexRegionScanner for region "
+ + region.getRegionInfo().getRegionNameAsString(), e);
+ throw e;
+ } finally {
+ region.closeRegionOperation();
+ }
+ }
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredLocalIndexRegionScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredLocalIndexRegionScanner.java
new file mode 100644
index 0000000..59bd793
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UncoveredLocalIndexRegionScanner.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.coprocessor;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+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.phoenix.execute.TupleProjector;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.ScanUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+
+public class UncoveredLocalIndexRegionScanner extends UncoveredIndexRegionScanner {
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(UncoveredLocalIndexRegionScanner.class);
+ final int offset;
+ final byte[] actualStartKey;
+
+ public UncoveredLocalIndexRegionScanner(final RegionScanner innerScanner,
+ final Region region,
+ final Scan scan,
+ final RegionCoprocessorEnvironment env,
+ final Scan dataTableScan,
+ final TupleProjector tupleProjector,
+ final IndexMaintainer indexMaintainer,
+ final byte[][] viewConstants,
+ final ImmutableBytesWritable ptr,
+ final long pageSizeMs,
+ final int offset,
+ final byte[] actualStartKey) {
+ super(innerScanner, region, scan, env, dataTableScan, tupleProjector, indexMaintainer,
+ viewConstants, ptr, pageSizeMs);
+ this.offset = offset;
+ this.actualStartKey = actualStartKey;
+ }
+
+ protected void scanDataRows(Collection<byte[]> dataRowKeys, long startTime) throws IOException {
+ Scan dataScan = prepareDataTableScan(dataRowKeys);
+ try (RegionScanner regionScanner = region.getScanner(dataScan)) {
+ boolean hasMore;
+ do {
+ List<Cell> row = new ArrayList<Cell>();
+ hasMore = regionScanner.nextRaw(row);
+
+ if (!row.isEmpty()) {
+ if (ScanUtil.isDummy(row)) {
+ state = State.SCANNING_DATA_INTERRUPTED;
+ break;
+ }
+ Cell firstCell = row.get(0);
+ dataRows.put(new ImmutableBytesPtr(CellUtil.cloneRow(firstCell)),
+ Result.create(row));
+ if (hasMore &&
+ (EnvironmentEdgeManager.currentTimeMillis() - startTime) >=
+ pageSizeMs) {
+ state = State.SCANNING_DATA_INTERRUPTED;
+ break;
+ }
+ }
+ } while (hasMore);
+ if (state == State.SCANNING_DATA_INTERRUPTED) {
+ LOGGER.info("Data table scan is interrupted in "
+ + "UncoveredLocalIndexRegionScanner for region "
+ + region.getRegionInfo().getRegionNameAsString()
+ + " as it could not complete on time (in " + pageSizeMs + " ms), and"
+ + " it will be resubmitted");
+ }
+ }
+ }
+
+ @Override
+ protected void scanDataTableRows(long startTime)
+ throws IOException {
+ if (indexToDataRowKeyMap.size() == 0) {
+ state = State.READY;
+ return;
+ }
+ scanDataRows(indexToDataRowKeyMap.values(), startTime);
+ if (state == State.SCANNING_DATA_INTERRUPTED) {
+ state = State.SCANNING_DATA;
+ } else {
+ state = State.READY;
+ }
+ }
+
+ @Override
+ protected boolean scanIndexTableRows(List<Cell> result,
+ final long startTime) throws IOException {
+ return scanIndexTableRows(result, startTime, actualStartKey, offset);
+ }
+
+ @Override
+ public boolean next(List<Cell> result) throws IOException {
+ boolean hasMore = super.next(result);
+ IndexUtil.wrapResultUsingOffset(result, offset);
+ return hasMore;
+ }
+}
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
index 069d35d..af28d78 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/RegionScannerFactory.java
@@ -28,6 +28,7 @@ import static org.apache.phoenix.util.ScanUtil.isDummy;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.phoenix.coprocessor.UncoveredGlobalIndexRegionScanner;
+import org.apache.phoenix.coprocessor.UncoveredLocalIndexRegionScanner;
import org.apache.phoenix.schema.KeyValueSchema;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PColumnImpl;
@@ -165,18 +166,28 @@ public abstract class RegionScannerFactory {
if (limitBytes != null) {
extraLimit = Bytes.toLong(limitBytes);
}
- if (ScanUtil.isUncoveredGlobalIndex(scan) && tupleProjector != null) {
- PTable.ImmutableStorageScheme storageScheme = indexMaintainer.getIndexStorageScheme();
+ if (ScanUtil.isLocalOrUncoveredGlobalIndex(scan) && tupleProjector != null) {
+ PTable.ImmutableStorageScheme storageScheme =
+ indexMaintainer.getIndexStorageScheme();
Scan dataTableScan = new Scan();
for (int i = 0; i < dataColumns.length; i++) {
- if (storageScheme == PTable.ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS) {
- scan.addFamily(dataColumns[i].getFamily());
+ if (storageScheme ==
+ PTable.ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS) {
+ dataTableScan.addFamily(dataColumns[i].getFamily());
} else {
- scan.addColumn(dataColumns[i].getFamily(), dataColumns[i].getQualifier());
+ dataTableScan.addColumn(dataColumns[i].getFamily(),
+ dataColumns[i].getQualifier());
}
}
- s = new UncoveredGlobalIndexRegionScanner(regionScanner, dataRegion, scan, env,
- dataTableScan, tupleProjector, indexMaintainer, viewConstants, ptr, pageSizeMs);
+ if (ScanUtil.isLocalIndex(scan)) {
+ s = new UncoveredLocalIndexRegionScanner(regionScanner, dataRegion, scan, env,
+ dataTableScan, tupleProjector, indexMaintainer, viewConstants, ptr,
+ pageSizeMs, offset, actualStartKey);
+ } else {
+ s = new UncoveredGlobalIndexRegionScanner(regionScanner, dataRegion, scan, env,
+ dataTableScan, tupleProjector, indexMaintainer, viewConstants, ptr,
+ pageSizeMs);
+ }
}
}
}
@@ -244,20 +255,14 @@ public abstract class RegionScannerFactory {
}
if ((ScanUtil.isLocalOrUncoveredGlobalIndex(scan))
&& !ScanUtil.isAnalyzeTable(scan)) {
- if (ScanUtil.isLocalIndex(scan)) {
+ if (ScanUtil.isLocalIndex(scan) && tupleProjector == null) {
if (actualStartKey != null) {
- next = scanTillScanStartRow(s, arrayKVRefs, arrayFuncRefs, result,
- null);
+ next = scanTillScanStartRow(s, result, null);
if (result.isEmpty() || ScanUtil.isDummy(result)) {
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, scan, offset, dataColumns,
- tupleProjector, dataRegion, indexMaintainer, viewConstants, ptr);
+ IndexUtil.wrapResultUsingOffset(result, offset);
}
if (extraWhere != null) {
@@ -371,9 +376,8 @@ public abstract class RegionScannerFactory {
* 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) throws IOException {
+ List<Cell> result,
+ ScannerContext scannerContext) throws IOException {
boolean next = true;
Cell firstCell = result.get(0);
long startTime = EnvironmentEdgeManager.currentTimeMillis();
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 4906bb4..b8542c4 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
@@ -42,24 +42,24 @@ import java.util.ListIterator;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.PhoenixTagType;
import org.apache.hadoop.hbase.Tag;
+import org.apache.hadoop.hbase.TagRewriteCell;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
-import org.apache.phoenix.hbase.index.table.HTableFactory;
+import org.apache.hadoop.hbase.regionserver.Region;
+import org.apache.hadoop.hbase.regionserver.Store;
import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.thirdparty.com.google.common.cache.Cache;
import org.apache.phoenix.thirdparty.com.google.common.cache.CacheBuilder;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.HRegionLocation;
-import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.TagRewriteCell;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
@@ -70,13 +70,10 @@ import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
-import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;
import org.apache.hadoop.hbase.ipc.ServerRpcController;
import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto;
-import org.apache.hadoop.hbase.regionserver.Region;
-import org.apache.hadoop.hbase.regionserver.Store;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.compile.ColumnResolver;
@@ -115,7 +112,6 @@ import org.apache.phoenix.parse.SQLParser;
import org.apache.phoenix.parse.SelectStatement;
import org.apache.phoenix.protobuf.ProtobufUtil;
import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.ColumnFamilyNotFoundException;
import org.apache.phoenix.schema.ColumnNotFoundException;
@@ -134,7 +130,6 @@ import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.ValueSchema.Field;
-import org.apache.phoenix.schema.tuple.ResultTuple;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.schema.types.PBinary;
import org.apache.phoenix.schema.types.PDataType;
@@ -577,6 +572,7 @@ public class IndexUtil {
whereNode.toSQL(indexResolver, buf);
return QueryUtil.getViewStatement(index.getSchemaName().getString(), index.getTableName().getString(), buf.toString());
}
+
public static void addTupleAsOneCell(List<Cell> result,
Tuple tuple,
TupleProjector tupleProjector,
@@ -592,47 +588,8 @@ public class IndexUtil {
VALUE_COLUMN_QUALIFIER, firstCell.getTimestamp(), value, 0, value.length);
result.add(keyValue);
}
- public static void wrapResultUsingOffset(final RegionCoprocessorEnvironment environment,
- List<Cell> result, final Scan scan, final int offset, ColumnReference[] dataColumns,
- TupleProjector tupleProjector, Region dataRegion, IndexMaintainer indexMaintainer,
- byte[][] viewConstants, ImmutableBytesWritable ptr) throws IOException {
- if (tupleProjector != null) {
- // Join back to data table here by issuing a local get projecting
- // all of the cq:cf from the KeyValueColumnExpression into the Get.
- Cell firstCell = result.get(0);
- byte[] indexRowKey = firstCell.getRowArray();
- ptr.set(indexRowKey, firstCell.getRowOffset() + offset, firstCell.getRowLength() - offset);
- byte[] dataRowKey = indexMaintainer.buildDataRowKey(ptr, viewConstants);
- Get get = new Get(dataRowKey);
- ImmutableStorageScheme storageScheme = indexMaintainer.getIndexStorageScheme();
- for (int i = 0; i < dataColumns.length; i++) {
- if (storageScheme == ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS) {
- get.addFamily(dataColumns[i].getFamily());
- } else {
- get.addColumn(dataColumns[i].getFamily(), dataColumns[i].getQualifier());
- }
- }
- Result joinResult = null;
- if (ScanUtil.isLocalIndex(scan)) {
- if (dataRegion != null) {
- joinResult = dataRegion.get(get);
- } else {
- TableName dataTable =
- TableName.valueOf(MetaDataUtil.getLocalIndexUserTableName(
- environment.getRegion().getTableDesc().getNameAsString()));
- try (Table table = environment.getTable(dataTable)) {
- joinResult = table.get(get);
- }
- }
- }
- // at this point join result has data from the data table. We now need to take this result and
- // add it to the cells that we are returning.
- // TODO: handle null case (but shouldn't happen)
- Tuple joinTuple = new ResultTuple(joinResult);
- // This will create a byte[] that captures all of the values from the data table
- addTupleAsOneCell(result, joinTuple, tupleProjector, ptr);
- }
-
+
+ public static void wrapResultUsingOffset(List<Cell> result, final int offset) throws IOException {
ListIterator<Cell> itr = result.listIterator();
while (itr.hasNext()) {
final Cell cell = itr.next();
@@ -756,7 +713,7 @@ public class IndexUtil {
itr.set(newCell);
}
}
-
+
public static String getIndexColumnExpressionStr(PColumn col) {
return col.getExpressionStr() == null ? IndexUtil.getCaseSensitiveDataColumnFullName(col.getName().getString())
: col.getExpressionStr();