You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by sk...@apache.org on 2020/06/17 22:06:47 UTC
[phoenix] branch 4.x updated: PHOENIX-5783: Implement starttime in
IndexTool for rebuild and verification (#789)
This is an automated email from the ASF dual-hosted git repository.
skadam 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 751f944 PHOENIX-5783: Implement starttime in IndexTool for rebuild and verification (#789)
751f944 is described below
commit 751f944b4b403ce2d13efb5dddf9539c09414b8a
Author: Swaroopa Kadam <sw...@gmail.com>
AuthorDate: Wed Jun 17 15:06:37 2020 -0700
PHOENIX-5783: Implement starttime in IndexTool for rebuild and verification (#789)
---
.../end2end/IndexToolForNonTxGlobalIndexIT.java | 221 +++++++++++++++++++++
.../phoenix/end2end/IndexToolTimeRangeIT.java | 2 -
.../index/IndexVerificationOutputRepositoryIT.java | 1 +
.../index/IndexVerificationResultRepositoryIT.java | 1 +
.../coprocessor/IndexRebuildRegionScanner.java | 85 +++++++-
.../PhoenixServerBuildIndexInputFormat.java | 3 +-
6 files changed, 305 insertions(+), 8 deletions(-)
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForNonTxGlobalIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForNonTxGlobalIndexIT.java
index 9415018..a81ead8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForNonTxGlobalIndexIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForNonTxGlobalIndexIT.java
@@ -48,6 +48,7 @@ import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.apache.phoenix.util.IndexScrutiny;
+import org.apache.phoenix.util.ManualEnvironmentEdge;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.SchemaUtil;
@@ -159,6 +160,7 @@ public class IndexToolForNonTxGlobalIndexIT extends BaseUniqueNamesOwnClusterIT
deleteAllRows(conn,
TableName.valueOf(IndexVerificationResultRepository.RESULT_TABLE_NAME));
}
+ EnvironmentEdgeManager.reset();
}
@Test
@@ -592,6 +594,225 @@ public class IndexToolForNonTxGlobalIndexIT extends BaseUniqueNamesOwnClusterIT
}
@Test
+ public void testIndexToolForIncrementalVerify() throws Exception {
+ ManualEnvironmentEdge customeEdge = new ManualEnvironmentEdge();
+ String schemaName = generateUniqueName();
+ String dataTableName = generateUniqueName();
+ String viewName = generateUniqueName();
+ String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName);
+ String viewFullName = SchemaUtil.getTableName(schemaName, viewName);
+ String indexTableName = generateUniqueName();
+ String viewIndexName = generateUniqueName();
+ long waitForUpsert = 2;
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ conn.createStatement().execute("CREATE TABLE "+dataTableFullName+" "
+ + "(key1 BIGINT NOT NULL, key2 BIGINT NOT NULL, val1 VARCHAR, val2 BIGINT, "
+ + "val3 BIGINT, val4 DOUBLE, val5 BIGINT, val6 VARCHAR "
+ + "CONSTRAINT my_pk PRIMARY KEY(key1, key2)) "+tableDDLOptions);
+ conn.createStatement().execute("CREATE VIEW "+viewFullName+" AS SELECT * FROM "+dataTableFullName);
+
+ conn.createStatement().execute(String.format(
+ "CREATE INDEX "+viewIndexName+" ON "+viewFullName+" (val3) INCLUDE(val5)"));
+
+ conn.createStatement().execute(String.format(
+ "CREATE INDEX "+indexTableName+" ON "+dataTableFullName+" (val3) INCLUDE(val5)"));
+
+ customeEdge.setValue(EnvironmentEdgeManager.currentTimeMillis());
+ EnvironmentEdgeManager.injectEdge(customeEdge);
+ long t0 = customeEdge.currentTime();
+ customeEdge.incrementValue(waitForUpsert);
+ conn.createStatement().execute("UPSERT INTO "+dataTableFullName+"(key1, key2, val1, val2) VALUES (4,5,'abc',3)");
+ customeEdge.incrementValue(waitForUpsert);
+ long t1 = customeEdge.currentTime();
+ customeEdge.incrementValue(waitForUpsert);
+ conn.createStatement().execute("UPSERT INTO "+dataTableFullName+"(key1, key2, val1, val2) VALUES (1,2,'abc',3)");
+ customeEdge.incrementValue(waitForUpsert);
+ long t2 = customeEdge.currentTime();
+ customeEdge.incrementValue(waitForUpsert);
+ conn.createStatement().execute("UPSERT INTO "+dataTableFullName+"(key1, key2, val3, val4) VALUES (1,2,4,1.2)");
+ customeEdge.incrementValue(waitForUpsert);
+ long t3 = customeEdge.currentTime();
+ customeEdge.incrementValue(waitForUpsert);
+ conn.createStatement().execute("UPSERT INTO "+dataTableFullName+"(key1, key2, val5, val6) VALUES (1,2,5,'def')");
+ customeEdge.incrementValue(waitForUpsert);
+ long t4 = customeEdge.currentTime();
+ customeEdge.incrementValue(waitForUpsert);
+ conn.createStatement().execute("DELETE FROM "+dataTableFullName+" WHERE key1=4");
+ customeEdge.incrementValue(waitForUpsert);
+ long t5 = customeEdge.currentTime();
+ customeEdge.incrementValue(10);
+ long t6 = customeEdge.currentTime();
+ IndexTool it;
+ if(!mutable) {
+ // job with 2 rows
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t0),"-et", String.valueOf(t2));
+ verifyCounters(it, 2, 2);
+
+ // only one row
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t1),"-et", String.valueOf(t2));
+ verifyCounters(it, 1, 1);
+
+ // no rows
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t5),"-et", String.valueOf(t6));
+ verifyCounters(it, 0, 0);
+
+ //view index
+ // job with 2 rows
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, viewName, viewIndexName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t0),"-et", String.valueOf(t2));
+ verifyCounters(it, 2, 2);
+
+ // only one row
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, viewName, viewIndexName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t1),"-et", String.valueOf(t2));
+ verifyCounters(it, 1, 1);
+
+ // no rows
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, viewName, viewIndexName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t5),"-et", String.valueOf(t6));
+ verifyCounters(it, 0, 0);
+
+ return;
+ }
+ // regular job without delete row
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t0),"-et", String.valueOf(t4));
+ verifyCounters(it, 2, 3);
+
+ // job with 2 rows
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t0),"-et", String.valueOf(t2));
+ verifyCounters(it, 2, 2);
+
+ // job with update on only one row
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t1),"-et", String.valueOf(t3));
+ verifyCounters(it, 1, 2);
+
+ // job with update on only one row
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t2),"-et", String.valueOf(t4));
+ verifyCounters(it, 1, 2);
+
+ // job with update on only one row
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t4),"-et", String.valueOf(t5));
+ verifyCounters(it, 1, 1);
+
+ // job with no new updates on any row
+ it = IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t5),"-et", String.valueOf(t6));
+ verifyCounters(it, 0, 0);
+ }
+ }
+
+ @Test
+ public void testIndexToolForIncrementalVerify_viewIndex() throws Exception {
+ ManualEnvironmentEdge customeEdge = new ManualEnvironmentEdge();
+ String schemaName = generateUniqueName();
+ String dataTableName = generateUniqueName();
+ String viewName = generateUniqueName();
+ String viewFullName = SchemaUtil.getTableName(schemaName, viewName);
+ String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName);
+ String viewIndexName = generateUniqueName();
+ long waitForUpsert = 2;
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ conn.setAutoCommit(true);
+ conn.createStatement().execute("CREATE TABLE " + dataTableFullName + " "
+ + "(key1 BIGINT NOT NULL, key2 BIGINT NOT NULL, val1 VARCHAR, val2 BIGINT, "
+ + "val3 BIGINT, val4 DOUBLE, val5 BIGINT, val6 VARCHAR "
+ + "CONSTRAINT my_pk PRIMARY KEY(key1, key2)) COLUMN_ENCODED_BYTES=0");
+ conn.createStatement().execute(
+ "CREATE VIEW " + viewFullName + " AS SELECT * FROM " + dataTableFullName + " WHERE val6 = 'def'");
+ conn.createStatement().execute(String.format(
+ "CREATE INDEX " + viewIndexName + " ON " + viewFullName
+ + " (val3) INCLUDE(val5)"));
+ customeEdge.setValue(EnvironmentEdgeManager.currentTimeMillis());
+ EnvironmentEdgeManager.injectEdge(customeEdge);
+
+ long t1 = customeEdge.currentTime();
+ customeEdge.incrementValue(waitForUpsert);
+ conn.createStatement()
+ .execute("UPSERT INTO " + viewFullName + " VALUES (5,6,'abc',8,4,1.3,6,'def')");
+ customeEdge.incrementValue(waitForUpsert);
+ long t2 = customeEdge.currentTime();
+ customeEdge.incrementValue(waitForUpsert);
+ conn.createStatement()
+ .execute("UPSERT INTO " + viewFullName + " VALUES (1,2,'abc',3,4,1.2,5,'def')");
+ customeEdge.incrementValue(waitForUpsert);
+ long t3 = customeEdge.currentTime();
+ customeEdge.incrementValue(waitForUpsert);
+ conn.createStatement().execute("DELETE FROM " + viewFullName + " WHERE key1=5");
+ customeEdge.incrementValue(waitForUpsert);
+ long t4 = customeEdge.currentTime();
+ customeEdge.incrementValue(10);
+ long t5 = customeEdge.currentTime();
+ IndexTool it;
+ // regular job without delete row
+ it =
+ IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, viewName,
+ viewIndexName, null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t1),
+ "-et", String.valueOf(t4));
+ verifyCounters(it, 2, 2);
+
+ // job with 1 row
+ it =
+ IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, viewName,
+ viewIndexName, null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t1),
+ "-et", String.valueOf(t2));
+ verifyCounters(it, 1, 1);
+
+ // job with update on only one row
+ it =
+ IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, viewName,
+ viewIndexName, null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t2),
+ "-et", String.valueOf(t3));
+ verifyCounters(it, 1, 1);
+
+ // job with update on 2 rows
+ it =
+ IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, viewName,
+ viewIndexName, null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t1),
+ "-et", String.valueOf(t3));
+ verifyCounters(it, 2, 2);
+
+ // job with update on only one row
+ it =
+ IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, viewName,
+ viewIndexName, null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t3),
+ "-et", String.valueOf(t4));
+ verifyCounters(it, 1, 1);
+
+ // job with no new updates on any row
+ it =
+ IndexToolIT.runIndexTool(directApi, useSnapshot, schemaName, viewName,
+ viewIndexName, null, 0, IndexTool.IndexVerifyType.ONLY, "-st", String.valueOf(t4),
+ "-et", String.valueOf(t5));
+ verifyCounters(it, 0, 0);
+ }
+ }
+
+ private void verifyCounters(IndexTool it, int scanned, int valid) throws IOException {
+ assertEquals(scanned, it.getJob().getCounters().findCounter(SCANNED_DATA_ROW_COUNT).getValue());
+ assertEquals(0, it.getJob().getCounters().findCounter(REBUILT_INDEX_ROW_COUNT).getValue());
+ assertEquals(valid, it.getJob().getCounters().findCounter(BEFORE_REBUILD_VALID_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, it.getJob().getCounters().findCounter(BEFORE_REBUILD_EXPIRED_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, it.getJob().getCounters().findCounter(BEFORE_REBUILD_INVALID_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, it.getJob().getCounters().findCounter(BEFORE_REBUILD_MISSING_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, it.getJob().getCounters().findCounter(BEFORE_REBUILD_BEYOND_MAXLOOKBACK_MISSING_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, it.getJob().getCounters().findCounter(BEFORE_REBUILD_BEYOND_MAXLOOKBACK_INVALID_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, it.getJob().getCounters().findCounter(BEFORE_REBUILD_UNVERIFIED_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, it.getJob().getCounters().findCounter(BEFORE_REBUILD_OLD_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, it.getJob().getCounters().findCounter(BEFORE_REBUILD_UNKNOWN_INDEX_ROW_COUNT).getValue());
+ }
+
+ @Test
public void testDisableOutputLogging() throws Exception {
if (!mutable || useSnapshot) {
return;
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolTimeRangeIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolTimeRangeIT.java
index 3deedad..d188fe3 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolTimeRangeIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolTimeRangeIT.java
@@ -137,7 +137,6 @@ public class IndexToolTimeRangeIT extends BaseUniqueNamesOwnClusterIT {
Assert.assertEquals(5, countRowsInIndex());
}
- @Ignore("Until PHOENIX-5783 is fixed")
@Test
public void testValidTimeRange_startTimeInBetween() throws Exception {
String [] args = {"--delete-all-and-rebuild",
@@ -166,7 +165,6 @@ public class IndexToolTimeRangeIT extends BaseUniqueNamesOwnClusterIT {
Assert.assertEquals(5, countRowsInIndex());
}
- @Ignore("Until PHOENIX-5783 is fixed")
@Test
public void testValidTimeRange_onlyStartTimePassed() throws Exception {
//starttime passed of last upsert
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexVerificationOutputRepositoryIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexVerificationOutputRepositoryIT.java
index e772fa7..3df0318 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexVerificationOutputRepositoryIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexVerificationOutputRepositoryIT.java
@@ -290,5 +290,6 @@ public class IndexVerificationOutputRepositoryIT extends ParallelStatsDisabledIT
((HBaseAdmin) admin).deleteTable(OUTPUT_TABLE_NAME_BYTES);
}
}
+ EnvironmentEdgeManager.reset();
}
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexVerificationResultRepositoryIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexVerificationResultRepositoryIT.java
index 84df8b9..43e5ace 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexVerificationResultRepositoryIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexVerificationResultRepositoryIT.java
@@ -195,5 +195,6 @@ public class IndexVerificationResultRepositoryIT extends ParallelStatsDisabledIT
((HBaseAdmin) admin).deleteTable(RESULT_TABLE_NAME_BYTES);
}
}
+ EnvironmentEdgeManager.reset();
}
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/IndexRebuildRegionScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/IndexRebuildRegionScanner.java
index c828819..6f82db3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/IndexRebuildRegionScanner.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/IndexRebuildRegionScanner.java
@@ -53,12 +53,15 @@ import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
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.filter.AllVersionsIndexRebuildFilter;
import org.apache.phoenix.mapreduce.index.IndexVerificationResultRepository;
import org.apache.phoenix.query.HBaseFactoryProvider;
+import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.compile.ScanRanges;
@@ -100,6 +103,10 @@ public class IndexRebuildRegionScanner extends GlobalIndexRegionScanner {
private boolean shouldVerifyCheckDone = false;
final private RegionCoprocessorEnvironment env;
private byte[][] indexRegionEndKeys;
+ private byte[] nextStartKey;
+ private boolean hasMoreIncr = false;
+ private long minTimestamp = 0;
+
@VisibleForTesting
public IndexRebuildRegionScanner(final RegionScanner innerScanner, final Region region, final Scan scan,
final RegionCoprocessorEnvironment env) throws IOException {
@@ -129,6 +136,8 @@ public class IndexRebuildRegionScanner extends GlobalIndexRegionScanner {
verificationResult = new IndexToolVerificationResult(scan);
verificationResultRepository =
new IndexVerificationResultRepository(indexMaintainer.getIndexTableName(), hTableFactory);
+ nextStartKey = null;
+ minTimestamp = scan.getTimeRange().getMin();
}
}
@@ -1207,15 +1216,20 @@ public class IndexRebuildRegionScanner extends GlobalIndexRegionScanner {
int dataRowCount = 0;
int indexMutationCount = 0;
region.startRegionOperation();
+ RegionScanner localScanner = null;
try {
- synchronized (innerScanner) {
- if(!shouldVerify()) {
+ localScanner = getLocalScanner();
+ if (localScanner == null) {
+ return false;
+ }
+ synchronized (localScanner) {
+ if (!shouldVerify()) {
skipped = true;
return false;
}
do {
List<Cell> row = new ArrayList<Cell>();
- hasMore = innerScanner.nextRaw(row);
+ hasMore = localScanner.nextRaw(row);
if (!row.isEmpty()) {
lastCell = row.get(0); // lastCell is any cell from the last visited row
Put put = null;
@@ -1260,6 +1274,9 @@ public class IndexRebuildRegionScanner extends GlobalIndexRegionScanner {
throw e;
} finally {
region.closeRegionOperation();
+ if (localScanner!=null && localScanner!=innerScanner) {
+ localScanner.close();
+ }
}
if (indexRowKeyforReadRepair != null) {
dataRowCount = singleRowRebuildReturnCode;
@@ -1274,7 +1291,67 @@ public class IndexRebuildRegionScanner extends GlobalIndexRegionScanner {
SINGLE_COLUMN, AGG_TIMESTAMP, rowCountBytes, 0, rowCountBytes.length);
}
results.add(aggKeyValue);
- return hasMore;
+ return hasMore || hasMoreIncr;
+ }
+
+ private RegionScanner getLocalScanner() throws IOException {
+ // override the filter to skip scan and open new scanner
+ // when lower bound of timerange is passed or newStartKey was populated
+ // from previous call to next()
+ if(minTimestamp!= 0) {
+ Scan incrScan = new Scan(scan);
+ incrScan.setTimeRange(minTimestamp, scan.getTimeRange().getMax());
+ incrScan.setRaw(true);
+ incrScan.setMaxVersions();
+ incrScan.getFamilyMap().clear();
+ incrScan.setCacheBlocks(false);
+ for (byte[] family : scan.getFamilyMap().keySet()) {
+ incrScan.addFamily(family);
+ }
+ // For rebuilds we use count (*) as query for regular tables which ends up setting the FKOF on scan
+ // This filter doesn't give us all columns and skips to the next row as soon as it finds 1 col
+ // For rebuilds we need all columns and all versions
+ if (scan.getFilter() instanceof FirstKeyOnlyFilter) {
+ incrScan.setFilter(null);
+ } else if (scan.getFilter() != null) {
+ // Override the filter so that we get all versions
+ incrScan.setFilter(new AllVersionsIndexRebuildFilter(scan.getFilter()));
+ }
+ if(nextStartKey != null) {
+ incrScan.setStartRow(nextStartKey);
+ }
+ List<KeyRange> keys = new ArrayList<>();
+ try(RegionScanner scanner = region.getScanner(incrScan)) {
+ List<Cell> row = new ArrayList<>();
+ int rowCount = 0;
+ // collect row keys that have been modified in the given time-range
+ // up to the size of page to build skip scan filter
+ do {
+ hasMoreIncr = scanner.nextRaw(row);
+ if (!row.isEmpty()) {
+ keys.add(PVarbinary.INSTANCE.getKeyRange(CellUtil.cloneRow(row.get(0))));
+ rowCount++;
+ }
+ row.clear();
+ } while (hasMoreIncr && rowCount < pageSizeInRows);
+ }
+ if (!hasMoreIncr && keys.isEmpty()) {
+ return null;
+ }
+ if (keys.isEmpty()) {
+ return innerScanner;
+ }
+ nextStartKey = ByteUtil.calculateTheClosestNextRowKeyForPrefix(keys.get(keys.size() - 1).getLowerRange());
+ ScanRanges scanRanges = ScanRanges.createPointLookup(keys);
+ scanRanges.initializeScan(incrScan);
+ SkipScanFilter skipScanFilter = scanRanges.getSkipScanFilter();
+ incrScan.setFilter(new SkipScanFilter(skipScanFilter, true));
+ //putting back the min time to 0 for index and data reads
+ incrScan.setTimeRange(0, scan.getTimeRange().getMax());
+ scan.setTimeRange(0, scan.getTimeRange().getMax());
+ return region.getScanner(incrScan);
+ }
+ return innerScanner;
}
@Override
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixServerBuildIndexInputFormat.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixServerBuildIndexInputFormat.java
index 696659b..60327f2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixServerBuildIndexInputFormat.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixServerBuildIndexInputFormat.java
@@ -75,11 +75,10 @@ public class PhoenixServerBuildIndexInputFormat<T extends DBWritable> extends Ph
}
final String txnScnValue = configuration.get(PhoenixConfigurationUtil.TX_SCN_VALUE);
final String currentScnValue = getCurrentScnValue(configuration);
+ final String startTimeValue = getIndexToolStartTime(configuration);
final String tenantId = configuration.get(PhoenixConfigurationUtil.MAPREDUCE_TENANT_ID);
final String lastVerifyTime = getIndexToolLastVerifyTime(configuration);
- //until PHOENIX-5783 is fixed; we'll continue with startTime = 0
- final String startTimeValue = null;
final Properties overridingProps = new Properties();
if (txnScnValue==null && currentScnValue!=null) {
overridingProps.put(PhoenixRuntime.CURRENT_SCN_ATTRIB, currentScnValue);