You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ac...@apache.org on 2020/06/30 04:24:12 UTC
[phoenix] branch 4.x updated: PHOENIX-5975 Index
rebuild/verification page size should be configurable from IndexTool
This is an automated email from the ASF dual-hosted git repository.
achouhan 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 54329fa PHOENIX-5975 Index rebuild/verification page size should be configurable from IndexTool
54329fa is described below
commit 54329fae44a699c2857af2a64602b9b79a862e05
Author: Abhishek Singh Chouhan <ac...@apache.org>
AuthorDate: Mon Jun 29 13:41:19 2020 -0700
PHOENIX-5975 Index rebuild/verification page size should be configurable from IndexTool
---
.../end2end/IndexToolForNonTxGlobalIndexIT.java | 47 ++++++++++++++++++++++
.../org/apache/phoenix/end2end/IndexToolIT.java | 10 ++++-
.../phoenix/compile/ServerBuildIndexCompiler.java | 10 +++++
.../coprocessor/BaseScannerRegionObserver.java | 1 +
.../coprocessor/GlobalIndexRegionScanner.java | 11 ++++-
.../PhoenixServerBuildIndexInputFormat.java | 8 ++++
.../index/PhoenixServerBuildIndexMapper.java | 16 ++++++++
7 files changed, 100 insertions(+), 3 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 0216355..07ca656 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
@@ -18,6 +18,8 @@
package org.apache.phoenix.end2end;
import com.google.common.collect.Maps;
+
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HColumnDescriptor;
@@ -1077,6 +1079,51 @@ public class IndexToolForNonTxGlobalIndexIT extends BaseUniqueNamesOwnClusterIT
}
@Test
+ public void testOverrideIndexRebuildPageSizeFromIndexTool() throws Exception {
+ String schemaName = generateUniqueName();
+ String dataTableName = generateUniqueName();
+ String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName);
+ String indexTableName = generateUniqueName();
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+
+ try(Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String stmString1 =
+ "CREATE TABLE " + dataTableFullName
+ + " (ID INTEGER NOT NULL PRIMARY KEY, NAME VARCHAR, ZIP INTEGER) "
+ + tableDDLOptions;
+ conn.createStatement().execute(stmString1);
+ String upsertQuery = String.format("UPSERT INTO %s VALUES(?, ?, ?)", dataTableFullName);
+ PreparedStatement stmt1 = conn.prepareStatement(upsertQuery);
+
+ // Insert NROWS rows
+ final int NROWS = 16;
+ for (int i = 0; i < NROWS; i++) {
+ IndexToolIT.upsertRow(stmt1, i);
+ }
+ conn.commit();
+
+ String stmtString2 =
+ String.format(
+ "CREATE INDEX %s ON %s (NAME) INCLUDE (ZIP) ASYNC ", indexTableName, dataTableFullName);
+ conn.createStatement().execute(stmtString2);
+
+ // Run the index MR job and verify that the index table is built correctly
+ Configuration conf = new Configuration(getUtility().getConfiguration());
+ conf.set(QueryServices.INDEX_REBUILD_PAGE_SIZE_IN_ROWS, Long.toString(2));
+ IndexTool indexTool = IndexToolIT.runIndexTool(conf, directApi, useSnapshot, schemaName, dataTableName, indexTableName, null, 0, IndexTool.IndexVerifyType.BEFORE, new String[0]);
+ assertEquals(NROWS, indexTool.getJob().getCounters().findCounter(INPUT_RECORDS).getValue());
+ assertEquals(NROWS, indexTool.getJob().getCounters().findCounter(SCANNED_DATA_ROW_COUNT).getValue());
+ assertEquals(NROWS, indexTool.getJob().getCounters().findCounter(REBUILT_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, indexTool.getJob().getCounters().findCounter(BEFORE_REBUILD_VALID_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, indexTool.getJob().getCounters().findCounter(BEFORE_REBUILD_EXPIRED_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, indexTool.getJob().getCounters().findCounter(BEFORE_REBUILD_INVALID_INDEX_ROW_COUNT).getValue());
+ assertEquals(NROWS, indexTool.getJob().getCounters().findCounter(BEFORE_REBUILD_MISSING_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, indexTool.getJob().getCounters().findCounter(BEFORE_REBUILD_BEYOND_MAXLOOKBACK_MISSING_INDEX_ROW_COUNT).getValue());
+ assertEquals(0, indexTool.getJob().getCounters().findCounter(BEFORE_REBUILD_BEYOND_MAXLOOKBACK_INVALID_INDEX_ROW_COUNT).getValue());
+ }
+ }
+
+ @Test
public void testUpdatablePKFilterViewIndexRebuild() throws Exception {
if (!mutable) {
return;
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
index 5310618..63258b8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
@@ -773,8 +773,16 @@ public class IndexToolIT extends BaseUniqueNamesOwnClusterIT {
String dataTableName, String indexTableName, String tenantId,
int expectedStatus, IndexTool.IndexVerifyType verifyType,
String... additionalArgs) throws Exception {
- IndexTool indexingTool = new IndexTool();
Configuration conf = new Configuration(getUtility().getConfiguration());
+ return runIndexTool(conf, directApi, useSnapshot, schemaName, dataTableName, indexTableName,
+ tenantId, expectedStatus, verifyType, additionalArgs);
+ }
+
+ public static IndexTool runIndexTool(Configuration conf, boolean directApi, boolean useSnapshot, String schemaName,
+ String dataTableName, String indexTableName, String tenantId,
+ int expectedStatus, IndexTool.IndexVerifyType verifyType,
+ String... additionalArgs) throws Exception {
+ IndexTool indexingTool = new IndexTool();
conf.set(QueryServices.TRANSACTIONS_ENABLED, Boolean.TRUE.toString());
indexingTool.setConf(conf);
final String[] cmdArgs = getArgValues(directApi, useSnapshot, schemaName, dataTableName,
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/ServerBuildIndexCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/ServerBuildIndexCompiler.java
index f221696..0359bfa 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/ServerBuildIndexCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/ServerBuildIndexCompiler.java
@@ -20,6 +20,7 @@ package org.apache.phoenix.compile;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.execute.BaseQueryPlan;
@@ -29,6 +30,7 @@ import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixStatement;
+import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TableRef;
@@ -125,6 +127,14 @@ public class ServerBuildIndexCompiler {
scan.setAttribute(BaseScannerRegionObserver.REBUILD_INDEXES, TRUE_BYTES);
ScanUtil.setClientVersion(scan, MetaDataProtocol.PHOENIX_VERSION);
scan.setAttribute(BaseScannerRegionObserver.INDEX_REBUILD_PAGING, TRUE_BYTES);
+ // Serialize page row size only if we're overriding, else use server side value
+ String rebuildPageRowSize =
+ connection.getQueryServices().getProps()
+ .get(QueryServices.INDEX_REBUILD_PAGE_SIZE_IN_ROWS);
+ if (rebuildPageRowSize != null) {
+ scan.setAttribute(BaseScannerRegionObserver.INDEX_REBUILD_PAGE_ROWS,
+ Bytes.toBytes(Long.valueOf(rebuildPageRowSize)));
+ }
BaseQueryPlan.serializeViewConstantsIntoScan(scan, dataTable);
addEmptyColumnToScan(scan, indexMaintainer.getDataEmptyKeyValueCF(), indexMaintainer.getEmptyKeyValueQualifier());
}
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 1909d8f..8950417 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
@@ -86,6 +86,7 @@ abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
public static final String LOCAL_INDEX_BUILD = "_LocalIndexBuild";
// The number of index rows to be rebuild in one RPC call
public static final String INDEX_REBUILD_PAGING = "_IndexRebuildPaging";
+ public static final String INDEX_REBUILD_PAGE_ROWS = "_IndexRebuildPageRows";
// Index verification type done by the index tool
public static final String INDEX_REBUILD_VERIFY_TYPE = "_IndexRebuildVerifyType";
public static final String INDEX_RETRY_VERIFY = "_IndexRetryVerify";
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java
index 726708a..8e1cad6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GlobalIndexRegionScanner.java
@@ -101,8 +101,15 @@ public abstract class GlobalIndexRegionScanner extends BaseRegionScanner {
super(innerScanner);
final Configuration config = env.getConfiguration();
if (scan.getAttribute(BaseScannerRegionObserver.INDEX_REBUILD_PAGING) != null) {
- pageSizeInRows = config.getLong(INDEX_REBUILD_PAGE_SIZE_IN_ROWS,
- QueryServicesOptions.DEFAULT_INDEX_REBUILD_PAGE_SIZE_IN_ROWS);
+ byte[] pageSizeFromScan =
+ scan.getAttribute(BaseScannerRegionObserver.INDEX_REBUILD_PAGE_ROWS);
+ if (pageSizeFromScan != null) {
+ pageSizeInRows = Bytes.toLong(pageSizeFromScan);
+ } else {
+ pageSizeInRows =
+ config.getLong(INDEX_REBUILD_PAGE_SIZE_IN_ROWS,
+ QueryServicesOptions.DEFAULT_INDEX_REBUILD_PAGE_SIZE_IN_ROWS);
+ }
}
maxBatchSize = config.getInt(MUTATE_BATCH_SIZE_ATTRIB, QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE);
maxBatchSizeBytes = config.getLong(MUTATE_BATCH_SIZE_BYTES_ATTRIB,
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 db59d4e..056daf6 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
@@ -34,6 +34,7 @@ import org.apache.phoenix.coprocessor.IndexRebuildRegionScanner;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.util.ConnectionUtil;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.*;
import org.apache.phoenix.util.*;
import org.slf4j.Logger;
@@ -105,6 +106,13 @@ public class PhoenixServerBuildIndexInputFormat<T extends DBWritable> extends Ph
try {
scan.setTimeRange(startTime, scn);
scan.setAttribute(BaseScannerRegionObserver.INDEX_REBUILD_PAGING, TRUE_BYTES);
+ // Serialize page row size only if we're overriding, else use server side value
+ String rebuildPageRowSize =
+ configuration.get(QueryServices.INDEX_REBUILD_PAGE_SIZE_IN_ROWS);
+ if (rebuildPageRowSize != null) {
+ scan.setAttribute(BaseScannerRegionObserver.INDEX_REBUILD_PAGE_ROWS,
+ Bytes.toBytes(Long.valueOf(rebuildPageRowSize)));
+ }
scan.setAttribute(BaseScannerRegionObserver.INDEX_REBUILD_VERIFY_TYPE, getIndexVerifyType(configuration).toBytes());
scan.setAttribute(BaseScannerRegionObserver.INDEX_RETRY_VERIFY, Bytes.toBytes(lastVerifyTimeValue));
scan.setAttribute(BaseScannerRegionObserver.INDEX_REBUILD_DISABLE_LOGGING_VERIFY_TYPE,
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixServerBuildIndexMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixServerBuildIndexMapper.java
index 5987a02..7a75cf6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixServerBuildIndexMapper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixServerBuildIndexMapper.java
@@ -25,6 +25,7 @@ import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.phoenix.mapreduce.PhoenixJobCounters;
+import org.apache.phoenix.query.QueryServices;
/**
* Mapper that does not do much as regions servers actually build the index from the data table regions directly
@@ -32,15 +33,30 @@ import org.apache.phoenix.mapreduce.PhoenixJobCounters;
public class PhoenixServerBuildIndexMapper extends
Mapper<NullWritable, PhoenixServerBuildIndexDBWritable, ImmutableBytesWritable, IntWritable> {
+ private long rebuildPageRowSize;
+
@Override
protected void setup(final Context context) throws IOException, InterruptedException {
super.setup(context);
+ String rebuildPageRowSizeConf =
+ context.getConfiguration().get(QueryServices.INDEX_REBUILD_PAGE_SIZE_IN_ROWS);
+ if (rebuildPageRowSizeConf != null) {
+ this.rebuildPageRowSize = Long.valueOf(rebuildPageRowSizeConf);
+ } else {
+ this.rebuildPageRowSize = -1L;
+ }
}
@Override
protected void map(NullWritable key, PhoenixServerBuildIndexDBWritable record, Context context)
throws IOException, InterruptedException {
context.getCounter(PhoenixJobCounters.INPUT_RECORDS).increment(record.getRowCount());
+ if (this.rebuildPageRowSize != -1) {
+ if (record.getRowCount() > this.rebuildPageRowSize) {
+ throw new IOException("Rebuilt/Verified rows greater than page size. Rebuilt rows: "
+ + record.getRowCount() + " Page size: " + this.rebuildPageRowSize);
+ }
+ }
// Make sure progress is reported to Application Master.
context.progress();
}