You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by an...@apache.org on 2016/12/26 06:29:42 UTC
[1/2] phoenix git commit: PHOENIX-2890 Extend IndexTool to allow
incremental index rebuilds
Repository: phoenix
Updated Branches:
refs/heads/master 70dc3836a -> 83827cd8c
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
index 82b353c..e594e0d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
@@ -17,10 +17,17 @@
*/
package org.apache.phoenix.mapreduce.index;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_SCHEM;
+
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.util.ArrayList;
import java.util.List;
import org.apache.commons.cli.CommandLine;
@@ -36,11 +43,15 @@ import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
+import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.util.Bytes;
@@ -48,9 +59,13 @@ import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.phoenix.compile.PostIndexDDLCompiler;
+import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
+import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
+import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.CsvBulkImportUtil;
import org.apache.phoenix.mapreduce.util.ColumnInfoToStringEncoderDecoder;
@@ -62,9 +77,9 @@ import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.ColumnInfo;
import org.apache.phoenix.util.IndexUtil;
-import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.SchemaUtil;
@@ -72,6 +87,8 @@ import org.apache.phoenix.util.TransactionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.collect.Lists;
+
/**
* An MR job to populate the index table from the data table.
*
@@ -85,7 +102,11 @@ public class IndexTool extends Configured implements Tool {
private static final Option DATA_TABLE_OPTION = new Option("dt", "data-table", true,
"Data table name (mandatory)");
private static final Option INDEX_TABLE_OPTION = new Option("it", "index-table", true,
- "Index table name(mandatory)");
+ "Index table name(not required in case of partial rebuilding)");
+
+ private static final Option PARTIAL_REBUILD_OPTION = new Option("pr", "partial-rebuild", false,
+ "To build indexes for a data table from least disabledTimeStamp");
+
private static final Option DIRECT_API_OPTION = new Option("direct", "direct", false,
"If specified, we avoid the bulk load (optional)");
private static final Option RUN_FOREGROUND_OPTION =
@@ -105,6 +126,7 @@ public class IndexTool extends Configured implements Tool {
options.addOption(SCHEMA_NAME_OPTION);
options.addOption(DATA_TABLE_OPTION);
options.addOption(INDEX_TABLE_OPTION);
+ options.addOption(PARTIAL_REBUILD_OPTION);
options.addOption(DIRECT_API_OPTION);
options.addOption(RUN_FOREGROUND_OPTION);
options.addOption(OUTPUT_PATH_OPTION);
@@ -139,18 +161,18 @@ public class IndexTool extends Configured implements Tool {
+ "parameter");
}
- if (!cmdLine.hasOption(INDEX_TABLE_OPTION.getOpt())) {
- throw new IllegalStateException(INDEX_TABLE_OPTION.getLongOpt() + " is a mandatory "
- + "parameter");
- }
-
- if (!cmdLine.hasOption(OUTPUT_PATH_OPTION.getOpt())) {
- throw new IllegalStateException(OUTPUT_PATH_OPTION.getLongOpt() + " is a mandatory "
- + "parameter");
- }
-
- if (!cmdLine.hasOption(DIRECT_API_OPTION.getOpt())
- && cmdLine.hasOption(RUN_FOREGROUND_OPTION.getOpt())) {
+ if (!(cmdLine.hasOption(PARTIAL_REBUILD_OPTION.getOpt()) || cmdLine.hasOption(DIRECT_API_OPTION.getOpt()))
+ && !cmdLine.hasOption(OUTPUT_PATH_OPTION.getOpt())) {
+ throw new IllegalStateException(OUTPUT_PATH_OPTION.getLongOpt() + " is a mandatory " + "parameter");
+ }
+
+ if (cmdLine.hasOption(PARTIAL_REBUILD_OPTION.getOpt()) && cmdLine.hasOption(INDEX_TABLE_OPTION.getOpt())) {
+ throw new IllegalStateException("Index name should not be passed with " + PARTIAL_REBUILD_OPTION.getLongOpt());
+ }
+
+ if (!(cmdLine.hasOption(DIRECT_API_OPTION.getOpt())) && cmdLine.hasOption(INDEX_TABLE_OPTION.getOpt())
+ && cmdLine.hasOption(RUN_FOREGROUND_OPTION
+ .getOpt())) {
throw new IllegalStateException(RUN_FOREGROUND_OPTION.getLongOpt()
+ " is applicable only for " + DIRECT_API_OPTION.getLongOpt());
}
@@ -167,50 +189,155 @@ public class IndexTool extends Configured implements Tool {
formatter.printHelp("help", options);
System.exit(exitCode);
}
+
+ class JobFactory {
+ Connection connection;
+ Configuration configuration;
+ private Path outputPath;
- @Override
- public int run(String[] args) throws Exception {
- Connection connection = null;
- try {
- CommandLine cmdLine = null;
- try {
- cmdLine = parseOptions(args);
- } catch (IllegalStateException e) {
- printHelpAndExit(e.getMessage(), getOptions());
+ public JobFactory(Connection connection, Configuration configuration, Path outputPath) {
+ this.connection = connection;
+ this.configuration = configuration;
+ this.outputPath = outputPath;
+
+ }
+
+ public Job getJob(String schemaName, String indexTable, String dataTable, boolean useDirectApi, boolean isPartialBuild) throws Exception {
+ if (isPartialBuild) {
+ return configureJobForPartialBuild(schemaName, dataTable);
+ } else {
+ return configureJobForAysncIndex(schemaName, indexTable, dataTable, useDirectApi);
}
- final Configuration configuration = HBaseConfiguration.addHbaseResources(getConf());
- final String schemaName = cmdLine.getOptionValue(SCHEMA_NAME_OPTION.getOpt());
- final String dataTable = cmdLine.getOptionValue(DATA_TABLE_OPTION.getOpt());
- final String indexTable = cmdLine.getOptionValue(INDEX_TABLE_OPTION.getOpt());
+ }
+
+ private Job configureJobForPartialBuild(String schemaName, String dataTable) throws Exception {
final String qDataTable = SchemaUtil.getQualifiedTableName(schemaName, dataTable);
- final String qIndexTable = SchemaUtil.getQualifiedTableName(schemaName, indexTable);
-
+ final PTable pdataTable = PhoenixRuntime.getTable(connection, qDataTable);
connection = ConnectionUtil.getInputConnection(configuration);
- if (!isValidIndexTable(connection, qDataTable, indexTable)) {
- throw new IllegalArgumentException(String.format(
- " %s is not an index table for %s ", qIndexTable, qDataTable));
+ long minDisableTimestamp = HConstants.LATEST_TIMESTAMP;
+ PTable indexWithMinDisableTimestamp = null;
+
+ //Get Indexes in building state, minDisabledTimestamp
+ List<String> disableIndexes = new ArrayList<String>();
+ List<PTable> disabledPIndexes = new ArrayList<PTable>();
+ for (PTable index : pdataTable.getIndexes()) {
+ if (index.getIndexState().equals(PIndexState.BUILDING)) {
+ disableIndexes.add(index.getTableName().getString());
+ disabledPIndexes.add(index);
+ if (minDisableTimestamp > index.getIndexDisableTimestamp()) {
+ minDisableTimestamp = index.getIndexDisableTimestamp();
+ indexWithMinDisableTimestamp = index;
+ }
+ }
+ }
+
+ if (indexWithMinDisableTimestamp == null) {
+ throw new Exception("There is no index for a datatable to be rebuild:" + qDataTable);
+ }
+ if (minDisableTimestamp == 0) {
+ throw new Exception("It seems Index " + indexWithMinDisableTimestamp
+ + " has disable timestamp as 0 , please run IndexTool with IndexName to build it first");
+ // TODO probably we can initiate the job by ourself or can skip them while making the list for partial build with a warning
+ }
+
+ long maxTimestamp = getMaxRebuildAsyncDate(schemaName, disableIndexes);
+
+ //serialize index maintaienr in job conf with Base64 TODO: Need to find better way to serialize them in conf.
+ List<IndexMaintainer> maintainers = Lists.newArrayListWithExpectedSize(disabledPIndexes.size());
+ for (PTable index : disabledPIndexes) {
+ maintainers.add(index.getIndexMaintainer(pdataTable, connection.unwrap(PhoenixConnection.class)));
+ }
+ ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable(ByteUtil.EMPTY_BYTE_ARRAY);
+ IndexMaintainer.serializeAdditional(pdataTable, indexMetaDataPtr, disabledPIndexes, connection.unwrap(PhoenixConnection.class));
+ PhoenixConfigurationUtil.setIndexMaintainers(configuration, indexMetaDataPtr);
+
+ //Prepare raw scan
+ Scan scan = IndexManagementUtil.newLocalStateScan(maintainers);
+ scan.setTimeRange(minDisableTimestamp - 1, maxTimestamp);
+ scan.setRaw(true);
+ scan.setCacheBlocks(false);
+ if (pdataTable.isTransactional()) {
+ long maxTimeRange = pdataTable.getTimeStamp() + 1;
+ scan.setAttribute(BaseScannerRegionObserver.TX_SCN,
+ Bytes.toBytes(Long.valueOf(Long.toString(TransactionUtil.convertToNanoseconds(maxTimeRange)))));
+ }
+
+
+ String physicalTableName=pdataTable.getPhysicalName().getString();
+ final String jobName = String.format("Phoenix Indexes build for " + pdataTable.getName().toString());
+
+ PhoenixConfigurationUtil.setInputTableName(configuration, qDataTable);
+ PhoenixConfigurationUtil.setPhysicalTableName(configuration, physicalTableName);
+
+ //TODO: update disable indexes
+ PhoenixConfigurationUtil.setDisableIndexes(configuration, StringUtils.join(",",disableIndexes));
+
+ final Job job = Job.getInstance(configuration, jobName);
+ if (outputPath != null) {
+ FileOutputFormat.setOutputPath(job, outputPath);
+ }
+ job.setJarByClass(IndexTool.class);
+ TableMapReduceUtil.initTableMapperJob(physicalTableName, scan, PhoenixIndexPartialBuildMapper.class, null,
+ null, job);
+ TableMapReduceUtil.initCredentials(job);
+ TableInputFormat.configureSplitTable(job, TableName.valueOf(physicalTableName));
+ return configureSubmittableJobUsingDirectApi(job, true);
+ }
+
+ private long getMaxRebuildAsyncDate(String schemaName, List<String> disableIndexes) throws SQLException {
+ Long maxRebuilAsyncDate=HConstants.LATEST_TIMESTAMP;
+ Long maxDisabledTimeStamp=0L;
+ if (disableIndexes == null || disableIndexes.isEmpty()) { return 0; }
+ List<String> quotedIndexes = new ArrayList<String>(disableIndexes.size());
+ for (String index : disableIndexes) {
+ quotedIndexes.add("'" + index + "'");
+ }
+ ResultSet rs = connection.createStatement()
+ .executeQuery("SELECT MAX(" + ASYNC_REBUILD_TIMESTAMP + "),MAX("+INDEX_DISABLE_TIMESTAMP+") FROM " + SYSTEM_CATALOG_NAME + " ("
+ + ASYNC_REBUILD_TIMESTAMP + " BIGINT) WHERE " + TABLE_SCHEM
+ + (schemaName != null && schemaName.length() > 0 ? "='" + schemaName + "'" : " IS NULL")
+ + " and " + TABLE_NAME + " IN (" + StringUtils.join(",", quotedIndexes) + ")");
+ if (rs.next()) {
+ maxRebuilAsyncDate = rs.getLong(1);
+ maxDisabledTimeStamp = rs.getLong(2);
+ }
+ // Do check if table is disabled again after user invoked async rebuilding during the run of the job
+ if (maxRebuilAsyncDate > maxDisabledTimeStamp) {
+ return maxRebuilAsyncDate;
+ } else {
+ throw new RuntimeException(
+ "Inconsistent state we have one or more index tables which are disabled after the async is called!!");
}
+
+ }
+ private Job configureJobForAysncIndex(String schemaName, String indexTable, String dataTable, boolean useDirectApi)
+ throws Exception {
+ final String qDataTable = SchemaUtil.getQualifiedTableName(schemaName, dataTable);
+ final String qIndexTable;
+ if (schemaName != null && !schemaName.isEmpty()) {
+ qIndexTable = SchemaUtil.getQualifiedTableName(schemaName, indexTable);
+ } else {
+ qIndexTable = indexTable;
+ }
final PTable pdataTable = PhoenixRuntime.getTable(connection, qDataTable);
+
final PTable pindexTable = PhoenixRuntime.getTable(connection, qIndexTable);
-
+
+ long maxTimeRange = pindexTable.getTimeStamp() + 1;
// this is set to ensure index tables remains consistent post population.
- long maxTimeRange = pindexTable.getTimeStamp()+1;
+
if (pdataTable.isTransactional()) {
configuration.set(PhoenixConfigurationUtil.TX_SCN_VALUE,
Long.toString(TransactionUtil.convertToNanoseconds(maxTimeRange)));
}
configuration.set(PhoenixConfigurationUtil.CURRENT_SCN_VALUE,
Long.toString(maxTimeRange));
-
+
// check if the index type is LOCAL, if so, derive and set the physicalIndexName that is
// computed from the qDataTable name.
String physicalIndexTable = pindexTable.getPhysicalName().getString();
- boolean isLocalIndexBuild = false;
- if (IndexType.LOCAL.equals(pindexTable.getIndexType())) {
- physicalIndexTable = qDataTable;
- isLocalIndexBuild = true;
- }
+
final PhoenixConnection pConnection = connection.unwrap(PhoenixConnection.class);
final PostIndexDDLCompiler ddlCompiler =
@@ -224,18 +351,15 @@ public class IndexTool extends Configured implements Tool {
configuration.set(PhoenixConfigurationUtil.UPSERT_STATEMENT, upsertQuery);
PhoenixConfigurationUtil.setPhysicalTableName(configuration, physicalIndexTable);
- PhoenixConfigurationUtil.setOutputTableName(configuration, indexTable);
+ PhoenixConfigurationUtil.setDisableIndexes(configuration, indexTable);
PhoenixConfigurationUtil.setUpsertColumnNames(configuration,
indexColumns.toArray(new String[indexColumns.size()]));
final List<ColumnInfo> columnMetadataList =
PhoenixRuntime.generateColumnInfo(connection, qIndexTable, indexColumns);
ColumnInfoToStringEncoderDecoder.encode(configuration, columnMetadataList);
-
- final Path outputPath = CsvBulkImportUtil
- .getOutputPath(new Path(cmdLine.getOptionValue(OUTPUT_PATH_OPTION.getOpt())), physicalIndexTable);
FileSystem.get(configuration).delete(outputPath, true);
- final String jobName = String.format(INDEX_JOB_NAME_TEMPLATE, dataTable, indexTable);
+ final String jobName = String.format(INDEX_JOB_NAME_TEMPLATE, pdataTable.getName().toString(), indexTable);
final Job job = Job.getInstance(configuration, jobName);
job.setJarByClass(IndexTool.class);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
@@ -245,17 +369,146 @@ public class IndexTool extends Configured implements Tool {
selectQuery);
TableMapReduceUtil.initCredentials(job);
- boolean useDirectApi = cmdLine.hasOption(DIRECT_API_OPTION.getOpt());
+
if (useDirectApi) {
- configureSubmittableJobUsingDirectApi(job, outputPath,
- cmdLine.hasOption(RUN_FOREGROUND_OPTION.getOpt()));
+ return configureSubmittableJobUsingDirectApi(job, false);
} else {
- configureRunnableJobUsingBulkLoad(job, outputPath, isLocalIndexBuild);
- // Without direct API, we need to update the index state to ACTIVE from client.
- IndexToolUtil.updateIndexState(connection, qDataTable, indexTable,
- PIndexState.ACTIVE);
+ return configureRunnableJobUsingBulkLoad(job, outputPath);
+
+ }
+
+ }
+
+ /**
+ * Submits the job and waits for completion.
+ * @param job
+ * @param outputPath
+ * @return
+ * @throws Exception
+ */
+ private Job configureRunnableJobUsingBulkLoad(Job job, Path outputPath) throws Exception {
+ job.setMapperClass(PhoenixIndexImportMapper.class);
+ job.setMapOutputKeyClass(ImmutableBytesWritable.class);
+ job.setMapOutputValueClass(KeyValue.class);
+ final Configuration configuration = job.getConfiguration();
+ final String physicalIndexTable =
+ PhoenixConfigurationUtil.getPhysicalTableName(configuration);
+ final HTable htable = new HTable(configuration, physicalIndexTable);
+ HFileOutputFormat.configureIncrementalLoad(job, htable);
+ return job;
+
+ }
+
+ /**
+ * Uses the HBase Front Door Api to write to index table. Submits the job and either returns or
+ * waits for the job completion based on runForeground parameter.
+ *
+ * @param job
+ * @param outputPath
+ * @param runForeground - if true, waits for job completion, else submits and returns
+ * immediately.
+ * @return
+ * @throws Exception
+ */
+ private Job configureSubmittableJobUsingDirectApi(Job job, boolean isPartialRebuild)
+ throws Exception {
+ if (!isPartialRebuild) {
+ //Don't configure mapper for partial build as it is configured already
+ job.setMapperClass(PhoenixIndexImportDirectMapper.class);
+ }
+ job.setReducerClass(PhoenixIndexImportDirectReducer.class);
+ Configuration conf = job.getConfiguration();
+ HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf));
+ // Set the Physical Table name for use in DirectHTableWriter#write(Mutation)
+ conf.set(TableOutputFormat.OUTPUT_TABLE,
+ PhoenixConfigurationUtil.getPhysicalTableName(job.getConfiguration()));
+ //Set the Output classes
+ job.setMapOutputKeyClass(ImmutableBytesWritable.class);
+ job.setMapOutputValueClass(IntWritable.class);
+ job.setOutputKeyClass(NullWritable.class);
+ job.setOutputValueClass(NullWritable.class);
+ TableMapReduceUtil.addDependencyJars(job);
+ job.setNumReduceTasks(1);
+ return job;
+ }
+
+ }
+
+ @Override
+ public int run(String[] args) throws Exception {
+ Connection connection = null;
+ HTable htable = null;
+ try {
+ CommandLine cmdLine = null;
+ try {
+ cmdLine = parseOptions(args);
+ } catch (IllegalStateException e) {
+ printHelpAndExit(e.getMessage(), getOptions());
+ }
+ final Configuration configuration = HBaseConfiguration.addHbaseResources(getConf());
+ final String schemaName = cmdLine.getOptionValue(SCHEMA_NAME_OPTION.getOpt());
+ final String dataTable = cmdLine.getOptionValue(DATA_TABLE_OPTION.getOpt());
+ final String indexTable = cmdLine.getOptionValue(INDEX_TABLE_OPTION.getOpt());
+ final boolean isPartialBuild = cmdLine.hasOption(PARTIAL_REBUILD_OPTION.getOpt());
+ final String qDataTable = SchemaUtil.getQualifiedTableName(schemaName, dataTable);
+ boolean useDirectApi = cmdLine.hasOption(DIRECT_API_OPTION.getOpt());
+ String basePath=cmdLine.getOptionValue(OUTPUT_PATH_OPTION.getOpt());
+ boolean isForeground = cmdLine.hasOption(RUN_FOREGROUND_OPTION.getOpt());
+ connection = ConnectionUtil.getInputConnection(configuration);
+ byte[][] splitKeysBeforeJob = null;
+ boolean isLocalIndexBuild = false;
+ PTable pindexTable = null;
+ if (indexTable != null) {
+ if (!isValidIndexTable(connection, qDataTable,indexTable)) {
+ throw new IllegalArgumentException(String.format(
+ " %s is not an index table for %s ", indexTable, qDataTable));
+ }
+ pindexTable = PhoenixRuntime.getTable(connection, schemaName != null && !schemaName.isEmpty()
+ ? SchemaUtil.getQualifiedTableName(schemaName, indexTable) : indexTable);
+ htable = (HTable)connection.unwrap(PhoenixConnection.class).getQueryServices()
+ .getTable(pindexTable.getPhysicalName().getBytes());
+ if (IndexType.LOCAL.equals(pindexTable.getIndexType())) {
+ isLocalIndexBuild = true;
+ splitKeysBeforeJob = htable.getRegionLocator().getStartKeys();
+ }
+ }
+
+ PTable pdataTable = PhoenixRuntime.getTableNoCache(connection, qDataTable);
+ Path outputPath = null;
+ if (basePath != null) {
+ outputPath = CsvBulkImportUtil.getOutputPath(new Path(basePath), pindexTable == null
+ ? pdataTable.getPhysicalName().getString() : pindexTable.getPhysicalName().getString());
+ FileSystem.get(configuration).delete(outputPath, true);
+ }
+
+ Job job = new JobFactory(connection, configuration, outputPath).getJob(schemaName, indexTable, dataTable,
+ useDirectApi, isPartialBuild);
+ if (!isForeground && useDirectApi) {
+ LOG.info("Running Index Build in Background - Submit async and exit");
+ job.submit();
+ return 0;
+ }
+ LOG.info("Running Index Build in Foreground. Waits for the build to complete. This may take a long time!.");
+ boolean result = job.waitForCompletion(true);
+
+ if (result) {
+ if (!useDirectApi && indexTable != null) {
+ if (isLocalIndexBuild) {
+ validateSplitForLocalIndex(splitKeysBeforeJob, htable);
+ }
+ LOG.info("Loading HFiles from {}", outputPath);
+ LoadIncrementalHFiles loader = new LoadIncrementalHFiles(configuration);
+ loader.doBulkLoad(outputPath, htable);
+ htable.close();
+ // Without direct API, we need to update the index state to ACTIVE from client.
+ IndexToolUtil.updateIndexState(connection, qDataTable, indexTable, PIndexState.ACTIVE);
+ FileSystem.get(configuration).delete(outputPath, true);
+ }
+ return 0;
+ } else {
+ LOG.error("IndexTool job failed! Check logs for errors..");
+ return -1;
}
- return 0;
} catch (Exception ex) {
LOG.error("An exception occurred while performing the indexing job: "
+ ExceptionUtils.getMessage(ex) + " at:\n" + ExceptionUtils.getStackTrace(ex));
@@ -265,6 +518,9 @@ public class IndexTool extends Configured implements Tool {
if (connection != null) {
connection.close();
}
+ if (htable != null) {
+ htable.close();
+ }
} catch (SQLException sqle) {
LOG.error("Failed to close connection ", sqle.getMessage());
throw new RuntimeException("Failed to close connection");
@@ -272,91 +528,18 @@ public class IndexTool extends Configured implements Tool {
}
}
- /**
- * Submits the job and waits for completion.
- * @param job
- * @param outputPath
- * @return
- * @throws Exception
- */
- private void configureRunnableJobUsingBulkLoad(Job job, Path outputPath, boolean isLocalIndexBuild) throws Exception {
- job.setMapperClass(PhoenixIndexImportMapper.class);
- job.setMapOutputKeyClass(ImmutableBytesWritable.class);
- job.setMapOutputValueClass(KeyValue.class);
- final Configuration configuration = job.getConfiguration();
- final String physicalIndexTable =
- PhoenixConfigurationUtil.getPhysicalTableName(configuration);
- final HTable htable = new HTable(configuration, physicalIndexTable);
- HFileOutputFormat.configureIncrementalLoad(job, htable);
- byte[][] splitKeysBeforeJob = null;
- if(isLocalIndexBuild) {
- splitKeysBeforeJob = htable.getRegionLocator().getStartKeys();
- }
- boolean status = job.waitForCompletion(true);
- if (!status) {
- LOG.error("IndexTool job failed!");
- htable.close();
- throw new Exception("IndexTool job failed: " + job.toString());
- } else {
- if (isLocalIndexBuild
- && !IndexUtil.matchingSplitKeys(splitKeysBeforeJob, htable.getRegionLocator()
- .getStartKeys())) {
- String errMsg = "The index to build is local index and the split keys are not matching"
- + " before and after running the job. Please rerun the job otherwise"
- + " there may be inconsistencies between actual data and index data";
- LOG.error(errMsg);
- throw new Exception(errMsg);
- }
- }
-
- LOG.info("Loading HFiles from {}", outputPath);
- LoadIncrementalHFiles loader = new LoadIncrementalHFiles(configuration);
- loader.doBulkLoad(outputPath, htable);
- htable.close();
-
- FileSystem.get(configuration).delete(outputPath, true);
- }
- /**
- * Uses the HBase Front Door Api to write to index table. Submits the job and either returns or
- * waits for the job completion based on runForeground parameter.
- *
- * @param job
- * @param outputPath
- * @param runForeground - if true, waits for job completion, else submits and returns
- * immediately.
- * @return
- * @throws Exception
- */
- private void configureSubmittableJobUsingDirectApi(Job job, Path outputPath, boolean runForeground)
- throws Exception {
- job.setMapperClass(PhoenixIndexImportDirectMapper.class);
- job.setReducerClass(PhoenixIndexImportDirectReducer.class);
- Configuration conf = job.getConfiguration();
- HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf));
- // Set the Physical Table name for use in DirectHTableWriter#write(Mutation)
- conf.set(TableOutputFormat.OUTPUT_TABLE,
- PhoenixConfigurationUtil.getPhysicalTableName(job.getConfiguration()));
-
- //Set the Output classes
- job.setMapOutputValueClass(IntWritable.class);
- job.setOutputKeyClass(NullWritable.class);
- job.setOutputValueClass(NullWritable.class);
- TableMapReduceUtil.addDependencyJars(job);
- job.setNumReduceTasks(1);
-
- if (!runForeground) {
- LOG.info("Running Index Build in Background - Submit async and exit");
- job.submit();
- return;
- }
- LOG.info("Running Index Build in Foreground. Waits for the build to complete. This may take a long time!.");
- boolean result = job.waitForCompletion(true);
- if (!result) {
- LOG.error("IndexTool job failed!");
- throw new Exception("IndexTool job failed: " + job.toString());
+
+ private boolean validateSplitForLocalIndex(byte[][] splitKeysBeforeJob, HTable htable) throws Exception {
+ if (splitKeysBeforeJob != null
+ && !IndexUtil.matchingSplitKeys(splitKeysBeforeJob, htable.getRegionLocator().getStartKeys())) {
+ String errMsg = "The index to build is local index and the split keys are not matching"
+ + " before and after running the job. Please rerun the job otherwise"
+ + " there may be inconsistencies between actual data and index data";
+ LOG.error(errMsg);
+ throw new Exception(errMsg);
}
- FileSystem.get(conf).delete(outputPath, true);
+ return true;
}
/**
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexToolUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexToolUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexToolUtil.java
index 1058670..2dc7551 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexToolUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexToolUtil.java
@@ -48,11 +48,13 @@ public class IndexToolUtil {
*/
public static void updateIndexState(Configuration configuration,PIndexState state) throws SQLException {
final String masterTable = PhoenixConfigurationUtil.getInputTableName(configuration);
- final String indexTable = PhoenixConfigurationUtil.getOutputTableName(configuration);
+ final String[] indexTables = PhoenixConfigurationUtil.getDisableIndexes(configuration).split(",");
final Properties overrideProps = new Properties();
final Connection connection = ConnectionUtil.getOutputConnection(configuration, overrideProps);
try {
- updateIndexState(connection, masterTable, indexTable , state);
+ for (String indexTable : indexTables) {
+ updateIndexState(connection, masterTable, indexTable, state);
+ }
} finally {
if(connection != null) {
connection.close();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java
index 9c64efc..15e55dd 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java
@@ -99,7 +99,7 @@ public class PhoenixIndexImportDirectMapper extends
this.pStatement = connection.prepareStatement(upsertQuery);
} catch (SQLException e) {
- throw new RuntimeException(e.getMessage());
+ throw new RuntimeException(e);
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexPartialBuildMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexPartialBuildMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexPartialBuildMapper.java
new file mode 100644
index 0000000..47a38a7
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexPartialBuildMapper.java
@@ -0,0 +1,182 @@
+/*
+ * 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.mapreduce.index;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Properties;
+import java.util.UUID;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.mapreduce.TableMapper;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.phoenix.cache.ServerCacheClient;
+import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.index.PhoenixIndexCodec;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.mapreduce.PhoenixJobCounters;
+import org.apache.phoenix.mapreduce.util.ConnectionUtil;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.query.ConnectionQueryServices;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.util.ByteUtil;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Mapper that hands over rows from data table to the index table.
+ */
+public class PhoenixIndexPartialBuildMapper extends TableMapper<ImmutableBytesWritable, IntWritable> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PhoenixIndexPartialBuildMapper.class);
+
+ private PhoenixConnection connection;
+
+ private DirectHTableWriter writer;
+
+ private int batchSize;
+
+ private List<Mutation> mutations ;
+
+ private ImmutableBytesPtr maintainers;
+
+ @Override
+ protected void setup(final Context context) throws IOException, InterruptedException {
+ super.setup(context);
+ final Configuration configuration = context.getConfiguration();
+ writer = new DirectHTableWriter(configuration);
+
+ try {
+ final Properties overrideProps = new Properties();
+ String scn = configuration.get(PhoenixConfigurationUtil.CURRENT_SCN_VALUE);
+ String txScnValue = configuration.get(PhoenixConfigurationUtil.TX_SCN_VALUE);
+ if(txScnValue==null && scn!=null) {
+ overrideProps.put(PhoenixRuntime.CURRENT_SCN_ATTRIB, scn);
+ }
+ connection = ConnectionUtil.getOutputConnection(configuration, overrideProps).unwrap(PhoenixConnection.class);
+ connection.setAutoCommit(false);
+ // Get BatchSize
+ ConnectionQueryServices services = connection.getQueryServices();
+ int maxSize =
+ services.getProps().getInt(QueryServices.MAX_MUTATION_SIZE_ATTRIB,
+ QueryServicesOptions.DEFAULT_MAX_MUTATION_SIZE);
+ batchSize = Math.min(connection.getMutateBatchSize(), maxSize);
+ LOG.info("Mutation Batch Size = " + batchSize);
+ this.mutations = Lists.newArrayListWithExpectedSize(batchSize);
+ maintainers=new ImmutableBytesPtr(PhoenixConfigurationUtil.getIndexMaintainers(configuration));
+ } catch (SQLException e) {
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ protected void map(ImmutableBytesWritable row, Result value, Context context)
+ throws IOException, InterruptedException {
+ context.getCounter(PhoenixJobCounters.INPUT_RECORDS).increment(1);
+ try {
+ byte[] attribValue = ByteUtil.copyKeyBytesIfNecessary(maintainers);
+ byte[] uuidValue = ServerCacheClient.generateId();
+ Put put = null;
+ Delete del = null;
+ for (Cell cell : value.rawCells()) {
+ if (KeyValue.Type.codeToType(cell.getTypeByte()) == KeyValue.Type.Put) {
+ if (put == null) {
+ put = new Put(CellUtil.cloneRow(cell));
+ put.setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue);
+ put.setAttribute(PhoenixIndexCodec.INDEX_MD, attribValue);
+ put.setAttribute(BaseScannerRegionObserver.IGNORE_NEWER_MUTATIONS, PDataType.TRUE_BYTES);
+ mutations.add(put);
+ }
+ put.add(cell);
+ } else {
+ if (del == null) {
+ del = new Delete(CellUtil.cloneRow(cell));
+ del.setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue);
+ del.setAttribute(PhoenixIndexCodec.INDEX_MD, attribValue);
+ del.setAttribute(BaseScannerRegionObserver.IGNORE_NEWER_MUTATIONS, PDataType.TRUE_BYTES);
+ mutations.add(del);
+ }
+ del.addDeleteMarker(cell);
+ }
+ }
+ // Write Mutation Batch
+ if (context.getCounter(PhoenixJobCounters.INPUT_RECORDS).getValue() % batchSize == 0) {
+ writeBatch(mutations, context);
+ mutations.clear();
+ }
+ // Make sure progress is reported to Application Master.
+ context.progress();
+ } catch (SQLException e) {
+ LOG.error(" Error {} while read/write of a record ", e.getMessage());
+ context.getCounter(PhoenixJobCounters.FAILED_RECORDS).increment(1);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void writeBatch(List<Mutation> mutations, Context context)
+ throws IOException, SQLException, InterruptedException {
+ writer.write(mutations);
+ context.getCounter(PhoenixJobCounters.OUTPUT_RECORDS).increment(mutations.size());
+ }
+
+ @Override
+ protected void cleanup(Context context) throws IOException, InterruptedException {
+ try {
+ // Write the last & final Mutation Batch
+ if (!mutations.isEmpty()) {
+ writeBatch(mutations, context);
+ }
+ // We are writing some dummy key-value as map output here so that we commit only one
+ // output to reducer.
+ context.write(new ImmutableBytesWritable(UUID.randomUUID().toString().getBytes()),
+ new IntWritable(0));
+ super.cleanup(context);
+ } catch (SQLException e) {
+ LOG.error(" Error {} while read/write of a record ", e.getMessage());
+ context.getCounter(PhoenixJobCounters.FAILED_RECORDS).increment(1);
+ throw new RuntimeException(e);
+ } finally {
+ if (connection != null) {
+ try {
+ connection.close();
+ } catch (SQLException e) {
+ LOG.error("Error {} while closing connection in the PhoenixIndexMapper class ",
+ e.getMessage());
+ }
+ }
+ if (writer != null) {
+ writer.close();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/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 2264acd..f3e4450 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
@@ -30,7 +30,9 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
+import org.apache.hadoop.hbase.util.Base64;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat.NullDBWritable;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;
@@ -98,6 +100,12 @@ public final class PhoenixConfigurationUtil {
public static final String MAPREDUCE_OUTPUT_CLUSTER_QUORUM = "phoneix.mapreduce.output.cluster.quorum";
+ public static final String INDEX_DISABLED_TIMESTAMP_VALUE = "phoenix.mr.index.disableTimestamp";
+
+ public static final String INDEX_MAINTAINERS = "phoenix.mr.index.maintainers";
+
+ public static final String DISABLED_INDEXES = "phoenix.mr.index.disabledIndexes";
+
public enum SchemaType {
TABLE,
QUERY;
@@ -428,4 +436,27 @@ public final class PhoenixConfigurationUtil {
return ReflectionUtils.newInstance(processorClass, conf);
}
+
+ public static byte[] getIndexMaintainers(final Configuration configuration){
+ Preconditions.checkNotNull(configuration);
+ return Base64.decode(configuration.get(INDEX_MAINTAINERS));
+ }
+
+ public static void setIndexMaintainers(final Configuration configuration,
+ final ImmutableBytesWritable indexMetaDataPtr) {
+ Preconditions.checkNotNull(configuration);
+ Preconditions.checkNotNull(indexMetaDataPtr);
+ configuration.set(INDEX_MAINTAINERS, Base64.encodeBytes(indexMetaDataPtr.get()));
+ }
+
+ public static void setDisableIndexes(Configuration configuration, String indexName) {
+ Preconditions.checkNotNull(configuration);
+ Preconditions.checkNotNull(indexName);
+ configuration.set(DISABLED_INDEXES, indexName);
+ }
+
+ public static String getDisableIndexes(Configuration configuration) {
+ Preconditions.checkNotNull(configuration);
+ return configuration.get(DISABLED_INDEXES);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/parse/AlterIndexStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/AlterIndexStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/AlterIndexStatement.java
index fcf817a..11328c2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/AlterIndexStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/AlterIndexStatement.java
@@ -23,12 +23,14 @@ public class AlterIndexStatement extends SingleTableStatement {
private final String dataTableName;
private final boolean ifExists;
private final PIndexState indexState;
+ private boolean async;
- public AlterIndexStatement(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState indexState) {
+ public AlterIndexStatement(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState indexState, boolean async) {
super(indexTableNode,0);
this.dataTableName = dataTableName;
this.ifExists = ifExists;
this.indexState = indexState;
+ this.async = async;
}
public String getTableName() {
@@ -48,4 +50,8 @@ public class AlterIndexStatement extends SingleTableStatement {
return indexState;
}
+ public boolean isAsync() {
+ return async;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
index 0c60e39..4b65c6a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
@@ -374,8 +374,12 @@ public class ParseNodeFactory {
return new DropIndexStatement(indexName, tableName, ifExists);
}
+ public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState state, boolean async) {
+ return new AlterIndexStatement(indexTableNode, dataTableName, ifExists, state, async);
+ }
+
public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState state) {
- return new AlterIndexStatement(indexTableNode, dataTableName, ifExists, state);
+ return new AlterIndexStatement(indexTableNode, dataTableName, ifExists, state, false);
}
public TraceStatement trace(boolean isTraceOn, double samplingRate) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
index 4a565be..1e002d2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -127,10 +127,13 @@ public interface QueryServices extends SQLCloseable {
// A master switch if to enable auto rebuild an index which failed to be updated previously
public static final String INDEX_FAILURE_HANDLING_REBUILD_ATTRIB = "phoenix.index.failure.handling.rebuild";
+ public static final String INDEX_FAILURE_HANDLING_REBUILD_PERIOD = "phoenix.index.failure.handling.rebuild.period";
// Time interval to check if there is an index needs to be rebuild
public static final String INDEX_FAILURE_HANDLING_REBUILD_INTERVAL_ATTRIB =
"phoenix.index.failure.handling.rebuild.interval";
+
+ public static final String INDEX_FAILURE_HANDLING_REBUILD_NUMBER_OF_BATCHES_PER_TABLE = "phoenix.index.rebuild.batch.perTable";
// A master switch if to block writes when index build failed
public static final String INDEX_FAILURE_BLOCK_WRITE = "phoenix.index.failure.block.write";
@@ -227,6 +230,7 @@ public interface QueryServices extends SQLCloseable {
public static final String CLIENT_CACHE_ENCODING = "phoenix.table.client.cache.encoding";
public static final String AUTO_UPGRADE_ENABLED = "phoenix.autoupgrade.enabled";
+
/**
* Get executor service used for parallel scans
*/
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 7f183e9..a3b9b32 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -27,6 +27,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.APPEND_ONLY_SCHEMA
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ARG_POSITION;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ARRAY_SIZE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ASYNC_CREATED_DATE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.AUTO_PARTITION_SEQ;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.BASE_COLUMN_COUNT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CLASS_NAME;
@@ -304,16 +305,27 @@ public class MetaDataClient {
TENANT_ID + "," +
TABLE_SCHEM + "," +
TABLE_NAME + "," +
- INDEX_STATE +
+ INDEX_STATE + "," +
+ ASYNC_REBUILD_TIMESTAMP + " " + PLong.INSTANCE.getSqlTypeName() +
+ ") VALUES (?, ?, ?, ?, ?)";
+
+ private static final String UPDATE_INDEX_REBUILD_ASYNC_STATE =
+ "UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " +
+ TENANT_ID + "," +
+ TABLE_SCHEM + "," +
+ TABLE_NAME + "," +
+ ASYNC_REBUILD_TIMESTAMP + " " + PLong.INSTANCE.getSqlTypeName() +
") VALUES (?, ?, ?, ?)";
+
private static final String UPDATE_INDEX_STATE_TO_ACTIVE =
"UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " +
TENANT_ID + "," +
TABLE_SCHEM + "," +
TABLE_NAME + "," +
INDEX_STATE + "," +
- INDEX_DISABLE_TIMESTAMP +
- ") VALUES (?, ?, ?, ?, ?)";
+ INDEX_DISABLE_TIMESTAMP +","+
+ ASYNC_REBUILD_TIMESTAMP + " " + PLong.INSTANCE.getSqlTypeName() +
+ ") VALUES (?, ?, ?, ?, ?, ?)";
//TODO: merge INSERT_COLUMN_CREATE_TABLE and INSERT_COLUMN_ALTER_TABLE column when
// the new major release is out.
private static final String INSERT_COLUMN_CREATE_TABLE =
@@ -3469,7 +3481,13 @@ public class MetaDataClient {
String dataTableName = statement.getTableName();
String schemaName = statement.getTable().getName().getSchemaName();
String indexName = statement.getTable().getName().getTableName();
+ boolean isAsync = statement.isAsync();
PIndexState newIndexState = statement.getIndexState();
+ if (isAsync && newIndexState != PIndexState.REBUILD) { throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.ASYNC_NOT_ALLOWED)
+ .setMessage(" ASYNC building of index is allowed only with REBUILD index state")
+ .setSchemaName(schemaName).setTableName(indexName).build().buildException(); }
+
if (newIndexState == PIndexState.REBUILD) {
newIndexState = PIndexState.BUILDING;
}
@@ -3480,15 +3498,16 @@ public class MetaDataClient {
try {
if(newIndexState == PIndexState.ACTIVE){
tableUpsert = connection.prepareStatement(UPDATE_INDEX_STATE_TO_ACTIVE);
- } else {
+ }else{
tableUpsert = connection.prepareStatement(UPDATE_INDEX_STATE);
}
tableUpsert.setString(1, connection.getTenantId() == null ? null : connection.getTenantId().getString());
tableUpsert.setString(2, schemaName);
tableUpsert.setString(3, indexName);
tableUpsert.setString(4, newIndexState.getSerializedValue());
+ tableUpsert.setLong(5, 0);
if(newIndexState == PIndexState.ACTIVE){
- tableUpsert.setLong(5, 0);
+ tableUpsert.setLong(6, 0);
}
tableUpsert.execute();
} finally {
@@ -3515,9 +3534,25 @@ public class MetaDataClient {
addTableToCache(result);
// Set so that we get the table below with the potentially modified rowKeyOrderOptimizable flag set
indexRef.setTable(result.getTable());
+ if (newIndexState == PIndexState.BUILDING && isAsync) {
+ try {
+ tableUpsert = connection.prepareStatement(UPDATE_INDEX_REBUILD_ASYNC_STATE);
+ tableUpsert.setString(1,
+ connection.getTenantId() == null ? null : connection.getTenantId().getString());
+ tableUpsert.setString(2, schemaName);
+ tableUpsert.setString(3, indexName);
+ tableUpsert.setLong(4, result.getTable().getTimeStamp());
+ tableUpsert.execute();
+ connection.commit();
+ } finally {
+ if (tableUpsert != null) {
+ tableUpsert.close();
+ }
+ }
+ }
}
}
- if (newIndexState == PIndexState.BUILDING) {
+ if (newIndexState == PIndexState.BUILDING && !isAsync) {
PTable index = indexRef.getTable();
// First delete any existing rows of the index
Long scn = connection.getSCN();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/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 9622880..2a0f3b9 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,9 +42,13 @@ import org.apache.hadoop.hbase.client.Mutation;
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;
+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.util.Bytes;
import org.apache.hadoop.io.WritableUtils;
@@ -54,6 +58,10 @@ import org.apache.phoenix.compile.IndexStatementRewriter;
import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.compile.WhereCompiler;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
+import org.apache.phoenix.coprocessor.MetaDataProtocol.MetaDataMutationResult;
+import org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataResponse;
+import org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataService;
+import org.apache.phoenix.coprocessor.generated.MetaDataProtos.UpdateIndexStateRequest;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.execute.TupleProjector;
@@ -67,10 +75,12 @@ import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.jdbc.PhoenixStatement;
import org.apache.phoenix.parse.ParseNode;
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.schema.ColumnFamilyNotFoundException;
import org.apache.phoenix.schema.ColumnNotFoundException;
@@ -78,6 +88,7 @@ import org.apache.phoenix.schema.ColumnRef;
import org.apache.phoenix.schema.KeyValueSchema;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PColumnFamily;
+import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.TableNotFoundException;
@@ -87,12 +98,13 @@ import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.schema.types.PBinary;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PDecimal;
+import org.apache.phoenix.schema.types.PLong;
import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.schema.types.PVarchar;
+import org.apache.tephra.TxConstants;
import com.google.common.collect.Lists;
-
-import org.apache.tephra.TxConstants;
+import com.google.protobuf.ServiceException;
public class IndexUtil {
public static final String INDEX_COLUMN_NAME_SEP = ":";
@@ -673,13 +685,48 @@ public class IndexUtil {
HConstants.NO_NONCE, HConstants.NO_NONCE);
}
+ public static MetaDataMutationResult disableIndexWithTimestamp(String indexTableName, long minTimeStamp,
+ HTableInterface metaTable, boolean blockWriteRebuildIndex) throws ServiceException, Throwable {
+ byte[] indexTableKey = SchemaUtil.getTableKeyFromFullName(indexTableName);
+ // Mimic the Put that gets generated by the client on an update of the index state
+ Put put = new Put(indexTableKey);
+ if (blockWriteRebuildIndex)
+ put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES,
+ PIndexState.ACTIVE.getSerializedBytes());
+ else
+ put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES,
+ PIndexState.DISABLE.getSerializedBytes());
+ put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES,
+ PLong.INSTANCE.toBytes(minTimeStamp));
+ put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP_BYTES,
+ PLong.INSTANCE.toBytes(0));
+ final List<Mutation> tableMetadata = Collections.<Mutation> singletonList(put);
+
+ final Map<byte[], MetaDataResponse> results = metaTable.coprocessorService(MetaDataService.class, indexTableKey,
+ indexTableKey, new Batch.Call<MetaDataService, MetaDataResponse>() {
+ @Override
+ public MetaDataResponse call(MetaDataService instance) throws IOException {
+ ServerRpcController controller = new ServerRpcController();
+ BlockingRpcCallback<MetaDataResponse> rpcCallback = new BlockingRpcCallback<MetaDataResponse>();
+ UpdateIndexStateRequest.Builder builder = UpdateIndexStateRequest.newBuilder();
+ for (Mutation m : tableMetadata) {
+ MutationProto mp = ProtobufUtil.toProto(m);
+ builder.addTableMetadataMutations(mp.toByteString());
+ }
+ instance.updateIndexState(controller, builder.build(), rpcCallback);
+ if (controller.getFailedOn() != null) { throw controller.getFailedOn(); }
+ return rpcCallback.get();
+ }
+ });
+ if (results.isEmpty()) { throw new IOException("Didn't get expected result size"); }
+ MetaDataResponse tmpResponse = results.values().iterator().next();
+ return MetaDataMutationResult.constructFromProto(tmpResponse);
+ }
+
public static boolean matchingSplitKeys(byte[][] splitKeys1, byte[][] splitKeys2) throws IOException {
- if (splitKeys1 != null && splitKeys2 != null
- && splitKeys1.length == splitKeys2.length) {
+ if (splitKeys1 != null && splitKeys2 != null && splitKeys1.length == splitKeys2.length) {
for (int i = 0; i < splitKeys1.length; i++) {
- if (Bytes.compareTo(splitKeys1[i], splitKeys2[i]) != 0) {
- return false;
- }
+ if (Bytes.compareTo(splitKeys1[i], splitKeys2[i]) != 0) { return false; }
}
} else {
return false;
[2/2] phoenix git commit: PHOENIX-2890 Extend IndexTool to allow
incremental index rebuilds
Posted by an...@apache.org.
PHOENIX-2890 Extend IndexTool to allow incremental index rebuilds
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/83827cd8
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/83827cd8
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/83827cd8
Branch: refs/heads/master
Commit: 83827cd8c2876c6b6dccf3a5678889b40a76261b
Parents: 70dc383
Author: Ankit Singhal <an...@gmail.com>
Authored: Mon Dec 26 11:59:19 2016 +0530
Committer: Ankit Singhal <an...@gmail.com>
Committed: Mon Dec 26 11:59:19 2016 +0530
----------------------------------------------------------------------
.../phoenix/end2end/AutomaticRebuildIT.java | 219 +++++++++
.../end2end/IndexToolForPartialBuildIT.java | 298 ++++++++++++
...olForPartialBuildWithNamespaceEnabledIT.java | 70 +++
.../phoenix/end2end/index/IndexMetadataIT.java | 58 +++
.../end2end/index/MutableIndexFailureIT.java | 10 +-
phoenix-core/src/main/antlr3/PhoenixSQL.g | 4 +-
.../coprocessor/MetaDataEndpointImpl.java | 9 +-
.../coprocessor/MetaDataRegionObserver.java | 291 +++++++-----
.../phoenix/exception/SQLExceptionCode.java | 3 +-
.../index/PhoenixIndexFailurePolicy.java | 52 +--
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 2 +
.../apache/phoenix/jdbc/PhoenixStatement.java | 8 +-
.../phoenix/mapreduce/index/IndexTool.java | 455 +++++++++++++------
.../phoenix/mapreduce/index/IndexToolUtil.java | 6 +-
.../index/PhoenixIndexImportDirectMapper.java | 2 +-
.../index/PhoenixIndexPartialBuildMapper.java | 182 ++++++++
.../util/PhoenixConfigurationUtil.java | 31 ++
.../phoenix/parse/AlterIndexStatement.java | 8 +-
.../apache/phoenix/parse/ParseNodeFactory.java | 6 +-
.../org/apache/phoenix/query/QueryServices.java | 4 +
.../apache/phoenix/schema/MetaDataClient.java | 47 +-
.../java/org/apache/phoenix/util/IndexUtil.java | 61 ++-
22 files changed, 1504 insertions(+), 322 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/it/java/org/apache/phoenix/end2end/AutomaticRebuildIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/AutomaticRebuildIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/AutomaticRebuildIT.java
new file mode 100644
index 0000000..cbb7745
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/AutomaticRebuildIT.java
@@ -0,0 +1,219 @@
+/*
+ * 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.end2end;
+
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.HBaseIOException;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.SimpleRegionObserver;
+import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.StringUtil;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Tests for the {@link AutomaticRebuildIT}
+ */
+@RunWith(Parameterized.class)
+public class AutomaticRebuildIT extends BaseOwnClusterIT {
+
+ private final boolean localIndex;
+ protected boolean isNamespaceEnabled = false;
+ protected final String tableDDLOptions;
+
+ public AutomaticRebuildIT(boolean localIndex) {
+ this.localIndex = localIndex;
+ StringBuilder optionBuilder = new StringBuilder();
+ optionBuilder.append(" SPLIT ON(1,2)");
+ this.tableDDLOptions = optionBuilder.toString();
+ }
+
+ @BeforeClass
+ public static void doSetup() throws Exception {
+ Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(7);
+ serverProps.put(QueryServices.EXTRA_JDBC_ARGUMENTS_ATTRIB, QueryServicesOptions.DEFAULT_EXTRA_JDBC_ARGUMENTS);
+ serverProps.put("hbase.coprocessor.region.classes", FailingRegionObserver.class.getName());
+ serverProps.put(" yarn.scheduler.capacity.maximum-am-resource-percent", "1.0");
+ serverProps.put(HConstants.HBASE_CLIENT_RETRIES_NUMBER, "2");
+ serverProps.put(HConstants.HBASE_RPC_TIMEOUT_KEY, "10000");
+ serverProps.put("hbase.client.pause", "5000");
+ serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_PERIOD, "1000");
+ serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_NUMBER_OF_BATCHES_PER_TABLE, "5");
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(1);
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()),
+ new ReadOnlyProps(clientProps.entrySet().iterator()));
+ }
+
+ @Parameters(name = "localIndex = {0}")
+ public static Collection<Boolean[]> data() {
+ return Arrays.asList(new Boolean[][] { { false }, { true } });
+ }
+
+ @Test
+ public void testSecondaryAutomaticRebuildIndex() throws Exception {
+ String schemaName = generateUniqueName();
+ String dataTableName = generateUniqueName();
+ String fullTableName = SchemaUtil.getTableName(schemaName, dataTableName);
+ final String indxTable = String.format("%s_%s", dataTableName, FailingRegionObserver.INDEX_NAME);
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ props.setProperty(QueryServices.TRANSACTIONS_ENABLED, Boolean.TRUE.toString());
+ props.setProperty(QueryServices.EXPLAIN_ROW_COUNT_ATTRIB, Boolean.FALSE.toString());
+ props.setProperty(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.toString(isNamespaceEnabled));
+ final Connection conn = DriverManager.getConnection(getUrl(), props);
+ Statement stmt = conn.createStatement();
+ try {
+ if (isNamespaceEnabled) {
+ conn.createStatement().execute("CREATE SCHEMA IF NOT EXISTS " + schemaName);
+ }
+ stmt.execute(String.format(
+ "CREATE TABLE %s (ID BIGINT NOT NULL, NAME VARCHAR, ZIP INTEGER CONSTRAINT PK PRIMARY KEY(ID ROW_TIMESTAMP)) %s",
+ fullTableName, tableDDLOptions));
+ String upsertQuery = String.format("UPSERT INTO %s VALUES(?, ?, ?)", fullTableName);
+ PreparedStatement stmt1 = conn.prepareStatement(upsertQuery);
+ FailingRegionObserver.FAIL_WRITE = false;
+ // insert two rows
+ upsertRow(stmt1, 1000);
+ upsertRow(stmt1, 2000);
+
+ conn.commit();
+ stmt.execute(String.format("CREATE %s INDEX %s ON %s (LPAD(UPPER(NAME),11,'x')||'_xyz') ",
+ (localIndex ? "LOCAL" : ""), indxTable, fullTableName));
+ FailingRegionObserver.FAIL_WRITE = true;
+ upsertRow(stmt1, 3000);
+ upsertRow(stmt1, 4000);
+ upsertRow(stmt1, 5000);
+ try {
+ conn.commit();
+ fail();
+ } catch (SQLException e) {
+ } catch (Exception e) {
+ }
+ FailingRegionObserver.FAIL_WRITE = false;
+ ResultSet rs = conn.getMetaData().getTables(null, StringUtil.escapeLike(schemaName), indxTable,
+ new String[] { PTableType.INDEX.toString() });
+ assertTrue(rs.next());
+ assertEquals(indxTable, rs.getString(3));
+ String indexState = rs.getString("INDEX_STATE");
+ assertEquals(PIndexState.DISABLE.toString(), indexState);
+ assertFalse(rs.next());
+ upsertRow(stmt1, 6000);
+ upsertRow(stmt1, 7000);
+ conn.commit();
+ int maxTries = 4, nTries = 0;
+ boolean isInactive = false;
+ do {
+ rs = conn.createStatement()
+ .executeQuery(String.format("SELECT " + PhoenixDatabaseMetaData.INDEX_STATE + ","
+ + PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP + " FROM "
+ + PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME + " ("
+ + PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP + " bigint) where "
+ + PhoenixDatabaseMetaData.TABLE_SCHEM + "='" + schemaName + "' and "
+ + PhoenixDatabaseMetaData.TABLE_NAME + "='" + indxTable + "'"));
+ rs.next();
+ if (PIndexState.INACTIVE.getSerializedValue().equals(rs.getString(1)) && rs.getLong(2) > 3000) {
+ isInactive = true;
+ break;
+ }
+ Thread.sleep(10 * 1000); // sleep 10 secs
+ } while (++nTries < maxTries);
+ assertTrue(isInactive);
+ nTries = 0;
+ boolean isActive = false;
+ do {
+ Thread.sleep(15 * 1000); // sleep 15 secs
+ rs = conn.getMetaData().getTables(null, StringUtil.escapeLike(schemaName), indxTable,
+ new String[] { PTableType.INDEX.toString() });
+ assertTrue(rs.next());
+ if (PIndexState.ACTIVE.toString().equals(rs.getString("INDEX_STATE"))) {
+ isActive = true;
+ break;
+ }
+ } while (++nTries < maxTries);
+ assertTrue(isActive);
+
+ } finally {
+ conn.close();
+ }
+ }
+
+ public static void upsertRow(PreparedStatement stmt, int i) throws SQLException {
+ // insert row
+ stmt.setInt(1, i);
+ stmt.setString(2, "uname" + String.valueOf(i));
+ stmt.setInt(3, 95050 + i);
+ stmt.executeUpdate();
+ }
+
+ public static class FailingRegionObserver extends SimpleRegionObserver {
+ public static volatile boolean FAIL_WRITE = false;
+ public static final String INDEX_NAME = "IDX";
+
+ @Override
+ public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
+ MiniBatchOperationInProgress<Mutation> miniBatchOp) throws HBaseIOException {
+ if (c.getEnvironment().getRegionInfo().getTable().getNameAsString().contains(INDEX_NAME) && FAIL_WRITE) {
+ throw new DoNotRetryIOException();
+ }
+ Mutation operation = miniBatchOp.getOperation(0);
+ Set<byte[]> keySet = operation.getFamilyMap().keySet();
+ for (byte[] family : keySet) {
+ if (Bytes.toString(family).startsWith(QueryConstants.LOCAL_INDEX_COLUMN_FAMILY_PREFIX) && FAIL_WRITE) {
+ throw new DoNotRetryIOException();
+ }
+ }
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java
new file mode 100644
index 0000000..116c47f
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java
@@ -0,0 +1,298 @@
+/*
+ * 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.end2end;
+
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.HBaseIOException;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.SimpleRegionObserver;
+import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.mapreduce.index.IndexTool;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.StringUtil;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+/**
+ * Tests for the {@link IndexToolForPartialBuildIT}
+ */
+@RunWith(Parameterized.class)
+public class IndexToolForPartialBuildIT extends BaseOwnClusterIT {
+
+ private final boolean localIndex;
+ protected boolean isNamespaceEnabled = false;
+ protected final String tableDDLOptions;
+
+ public IndexToolForPartialBuildIT(boolean localIndex) {
+
+ this.localIndex = localIndex;
+ StringBuilder optionBuilder = new StringBuilder();
+ optionBuilder.append(" SPLIT ON(1,2)");
+ this.tableDDLOptions = optionBuilder.toString();
+ }
+
+ @BeforeClass
+ public static void doSetup() throws Exception {
+ Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(7);
+ serverProps.put(QueryServices.EXTRA_JDBC_ARGUMENTS_ATTRIB, QueryServicesOptions.DEFAULT_EXTRA_JDBC_ARGUMENTS);
+ serverProps.put("hbase.coprocessor.region.classes", FailingRegionObserver.class.getName());
+ serverProps.put(" yarn.scheduler.capacity.maximum-am-resource-percent", "1.0");
+ serverProps.put(HConstants.HBASE_CLIENT_RETRIES_NUMBER, "2");
+ serverProps.put(HConstants.HBASE_RPC_TIMEOUT_KEY, "10000");
+ serverProps.put("hbase.client.pause", "5000");
+ serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_ATTRIB, Boolean.FALSE.toString());
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(1);
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()), new ReadOnlyProps(clientProps.entrySet().iterator()));
+ }
+
+ @Parameters(name="localIndex = {0}")
+ public static Collection<Boolean[]> data() {
+ return Arrays.asList(new Boolean[][] {
+ { false},{ true }
+ });
+ }
+
+ @Test
+ public void testSecondaryIndex() throws Exception {
+ String schemaName = generateUniqueName();
+ String dataTableName = generateUniqueName();
+ String fullTableName = SchemaUtil.getTableName(schemaName, dataTableName);
+ final String indxTable = String.format("%s_%s", dataTableName, FailingRegionObserver.INDEX_NAME);
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ props.setProperty(QueryServices.TRANSACTIONS_ENABLED, Boolean.TRUE.toString());
+ props.setProperty(QueryServices.EXPLAIN_ROW_COUNT_ATTRIB, Boolean.FALSE.toString());
+ props.setProperty(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.toString(isNamespaceEnabled));
+ final Connection conn = DriverManager.getConnection(getUrl(), props);
+ Statement stmt = conn.createStatement();
+ try {
+ if (isNamespaceEnabled) {
+ conn.createStatement().execute("CREATE SCHEMA IF NOT EXISTS " + schemaName);
+ }
+ stmt.execute(
+ String.format("CREATE TABLE %s (ID BIGINT NOT NULL, NAME VARCHAR, ZIP INTEGER CONSTRAINT PK PRIMARY KEY(ID ROW_TIMESTAMP)) %s",
+ fullTableName, tableDDLOptions));
+ String upsertQuery = String.format("UPSERT INTO %s VALUES(?, ?, ?)", fullTableName);
+ PreparedStatement stmt1 = conn.prepareStatement(upsertQuery);
+ FailingRegionObserver.FAIL_WRITE = false;
+ // insert two rows
+ upsertRow(stmt1, 1000);
+ upsertRow(stmt1, 2000);
+
+ conn.commit();
+ stmt.execute(String.format("CREATE %s INDEX %s ON %s (LPAD(UPPER(NAME),11,'x')||'_xyz') ",
+ (localIndex ? "LOCAL" : ""), indxTable, fullTableName));
+ FailingRegionObserver.FAIL_WRITE = true;
+ upsertRow(stmt1, 3000);
+ upsertRow(stmt1, 4000);
+ upsertRow(stmt1, 5000);
+ try {
+ conn.commit();
+ fail();
+ } catch (SQLException e) {} catch (Exception e) {}
+ conn.createStatement()
+ .execute(String.format("ALTER INDEX %s on %s REBUILD ASYNC", indxTable, fullTableName));
+
+ FailingRegionObserver.FAIL_WRITE = false;
+ ResultSet rs = conn.getMetaData().getTables(null, StringUtil.escapeLike(schemaName), indxTable,
+ new String[] { PTableType.INDEX.toString() });
+ assertTrue(rs.next());
+ assertEquals(indxTable, rs.getString(3));
+ String indexState = rs.getString("INDEX_STATE");
+ assertEquals(PIndexState.BUILDING.toString(), indexState);
+ assertFalse(rs.next());
+ upsertRow(stmt1, 6000);
+ upsertRow(stmt1, 7000);
+ conn.commit();
+
+ rs = conn.createStatement()
+ .executeQuery(String.format("SELECT " + PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP + ","
+ + PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP + " FROM "
+ + PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME + " ("
+ + PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP + " bigint) where "
+ + PhoenixDatabaseMetaData.TABLE_SCHEM + "='" + schemaName + "' and "
+ + PhoenixDatabaseMetaData.TABLE_NAME + "='" + indxTable + "'"));
+ rs.next();
+ PTable pindexTable = PhoenixRuntime.getTable(conn, SchemaUtil.getTableName(schemaName, indxTable));
+ assertEquals(PIndexState.BUILDING, pindexTable.getIndexState());
+ assertEquals(rs.getLong(1), pindexTable.getTimeStamp());
+ //assert disabled timestamp
+ assertEquals(rs.getLong(2), 3000);
+
+ String selectSql = String.format("SELECT LPAD(UPPER(NAME),11,'x')||'_xyz',ID FROM %s", fullTableName);
+ rs = conn.createStatement().executeQuery("EXPLAIN " + selectSql);
+ String actualExplainPlan = QueryUtil.getExplainPlan(rs);
+
+ // assert we are pulling from data table.
+ assertExplainPlan(actualExplainPlan, schemaName, dataTableName, null, false, isNamespaceEnabled);
+
+ rs = stmt1.executeQuery(selectSql);
+ for (int i = 1; i <= 7; i++) {
+ assertTrue(rs.next());
+ assertEquals("xxUNAME" + i*1000 + "_xyz", rs.getString(1));
+ }
+
+ // Validate Index table data till disabled timestamp
+ rs = stmt1.executeQuery(String.format("SELECT * FROM %s", SchemaUtil.getTableName(schemaName, indxTable)));
+ for (int i = 1; i <= 2; i++) {
+ assertTrue(rs.next());
+ assertEquals("xxUNAME" + i*1000 + "_xyz", rs.getString(1));
+ }
+ assertFalse(rs.next());
+ // run the index MR job.
+ final IndexTool indexingTool = new IndexTool();
+ Configuration conf = new Configuration(getUtility().getConfiguration());
+ conf.set(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.toString(isNamespaceEnabled));
+ indexingTool.setConf(conf);
+
+ final String[] cmdArgs = getArgValues(schemaName, dataTableName);
+ int status = indexingTool.run(cmdArgs);
+ assertEquals(0, status);
+
+ // insert two more rows
+ upsertRow(stmt1, 8000);
+ upsertRow(stmt1, 9000);
+ conn.commit();
+
+ // assert we are pulling from index table.
+ rs = conn.createStatement().executeQuery("EXPLAIN " + selectSql);
+ actualExplainPlan = QueryUtil.getExplainPlan(rs);
+ assertExplainPlan(actualExplainPlan, schemaName, dataTableName, indxTable, localIndex, isNamespaceEnabled);
+
+ rs = stmt.executeQuery(selectSql);
+
+ for (int i = 1; i <= 9; i++) {
+ assertTrue(rs.next());
+ assertEquals("xxUNAME" + i*1000 + "_xyz", rs.getString(1));
+ }
+
+ assertFalse(rs.next());
+
+ // conn.createStatement().execute(String.format("DROP INDEX %s ON %s", indxTable, fullTableName));
+ } finally {
+ conn.close();
+ }
+ }
+
+ public static void assertExplainPlan(final String actualExplainPlan, String schemaName, String dataTable,
+ String indxTable, boolean isLocal, boolean isNamespaceMapped) {
+
+ String expectedExplainPlan = "";
+ if (indxTable != null) {
+ if (isLocal) {
+ final String localIndexName = SchemaUtil
+ .getPhysicalHBaseTableName(SchemaUtil.getTableName(schemaName, dataTable), isNamespaceMapped,
+ PTableType.INDEX)
+ .getString();
+ expectedExplainPlan = String.format("CLIENT PARALLEL 3-WAY RANGE SCAN OVER %s [1]", localIndexName);
+ } else {
+ expectedExplainPlan = String.format("CLIENT PARALLEL 1-WAY FULL SCAN OVER %s",
+ SchemaUtil.getPhysicalHBaseTableName(SchemaUtil.getTableName(schemaName, indxTable),
+ isNamespaceMapped, PTableType.INDEX));
+ }
+ } else {
+ expectedExplainPlan = String.format("CLIENT PARALLEL 1-WAY FULL SCAN OVER %s",
+ SchemaUtil.getPhysicalHBaseTableName(SchemaUtil.getTableName(schemaName, dataTable),
+ isNamespaceMapped, PTableType.TABLE));
+ }
+ assertTrue(actualExplainPlan.contains(expectedExplainPlan));
+ }
+
+ public String[] getArgValues(String schemaName, String dataTable) {
+ final List<String> args = Lists.newArrayList();
+ if (schemaName!=null) {
+ args.add("-s");
+ args.add(schemaName);
+ }
+ args.add("-dt");
+ args.add(dataTable);
+ args.add("-pr");
+ args.add("-op");
+ args.add("/tmp/output/partialTable_"+localIndex);
+ return args.toArray(new String[0]);
+ }
+
+ public static void upsertRow(PreparedStatement stmt, int i) throws SQLException {
+ // insert row
+ stmt.setInt(1, i);
+ stmt.setString(2, "uname" + String.valueOf(i));
+ stmt.setInt(3, 95050 + i);
+ stmt.executeUpdate();
+ }
+
+
+ public static class FailingRegionObserver extends SimpleRegionObserver {
+ public static volatile boolean FAIL_WRITE = false;
+ public static final String INDEX_NAME = "IDX";
+ @Override
+ public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c, MiniBatchOperationInProgress<Mutation> miniBatchOp) throws HBaseIOException {
+ if (c.getEnvironment().getRegionInfo().getTable().getNameAsString().contains(INDEX_NAME) && FAIL_WRITE) {
+ throw new DoNotRetryIOException();
+ }
+ Mutation operation = miniBatchOp.getOperation(0);
+ Set<byte[]> keySet = operation.getFamilyMap().keySet();
+ for(byte[] family: keySet) {
+ if(Bytes.toString(family).startsWith(QueryConstants.LOCAL_INDEX_COLUMN_FAMILY_PREFIX) && FAIL_WRITE) {
+ throw new DoNotRetryIOException();
+ }
+ }
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildWithNamespaceEnabledIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildWithNamespaceEnabledIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildWithNamespaceEnabledIT.java
new file mode 100644
index 0000000..5e16b05
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildWithNamespaceEnabledIT.java
@@ -0,0 +1,70 @@
+/*
+ * 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.end2end;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.junit.BeforeClass;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Tests for the {@link IndexToolForPartialBuildWithNamespaceEnabled}
+ */
+@RunWith(Parameterized.class)
+public class IndexToolForPartialBuildWithNamespaceEnabled extends IndexToolForPartialBuildIT {
+
+
+ public IndexToolForPartialBuildWithNamespaceEnabled(boolean localIndex, boolean isNamespaceEnabled) {
+ super(localIndex);
+ this.isNamespaceEnabled=isNamespaceEnabled;
+ }
+
+ @BeforeClass
+ public static void doSetup() throws Exception {
+ Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(7);
+ serverProps.put(QueryServices.EXTRA_JDBC_ARGUMENTS_ATTRIB, QueryServicesOptions.DEFAULT_EXTRA_JDBC_ARGUMENTS);
+ serverProps.put("hbase.coprocessor.region.classes", FailingRegionObserver.class.getName());
+ serverProps.put(HConstants.HBASE_CLIENT_RETRIES_NUMBER, "2");
+ serverProps.put(HConstants.HBASE_RPC_TIMEOUT_KEY, "10000");
+ serverProps.put("hbase.client.pause", "5000");
+ serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_PERIOD, "2000");
+ serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_INTERVAL_ATTRIB, "1000");
+ serverProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, "true");
+ Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(1);
+ clientProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, "true");
+ setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()), new ReadOnlyProps(clientProps.entrySet().iterator()));
+ }
+
+ @Parameters(name="localIndex = {0} , isNamespaceEnabled = {1}")
+ public static Collection<Boolean[]> data() {
+ return Arrays.asList(new Boolean[][] {
+ { false, true},{ true, false }
+ });
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexMetadataIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexMetadataIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexMetadataIT.java
index f0c670b..63a6bd6 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexMetadataIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexMetadataIT.java
@@ -31,6 +31,7 @@ import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
import java.sql.Types;
import java.util.Properties;
@@ -43,10 +44,13 @@ import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.AmbiguousColumnException;
import org.apache.phoenix.schema.ColumnAlreadyExistsException;
import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableKey;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.schema.types.PDate;
+import org.apache.phoenix.schema.types.PLong;
+import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.StringUtil;
@@ -216,6 +220,15 @@ public class IndexMetadataIT extends ParallelStatsDisabledIT {
assertFalse(rs.next());
assertActiveIndex(conn, INDEX_DATA_SCHEMA, indexDataTable);
+
+ ddl = "ALTER INDEX " + indexName + " ON " + INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + indexDataTable + " REBUILD ASYNC";
+ conn.createStatement().execute(ddl);
+ // Verify the metadata for index is correct.
+ rs = conn.getMetaData().getTables(null, StringUtil.escapeLike(INDEX_DATA_SCHEMA), indexName , new String[] {PTableType.INDEX.toString()});
+ assertTrue(rs.next());
+ assertEquals(indexName , rs.getString(3));
+ assertEquals(PIndexState.BUILDING.toString(), rs.getString("INDEX_STATE"));
+ assertFalse(rs.next());
ddl = "DROP INDEX " + indexName + " ON " + INDEX_DATA_SCHEMA + QueryConstants.NAME_SEPARATOR + indexDataTable;
stmt = conn.prepareStatement(ddl);
@@ -568,4 +581,49 @@ public class IndexMetadataIT extends ParallelStatsDisabledIT {
assertTrue(d2.after(d1));
assertFalse(rs.next());
}
+
+ @Test
+ public void testAsyncRebuildTimestamp() throws Exception {
+ long startTimestamp = System.currentTimeMillis();
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ Connection conn = DriverManager.getConnection(getUrl(), props);
+ conn.setAutoCommit(false);
+ String testTable = generateUniqueName();
+
+
+ String ddl = "create table " + testTable + " (k varchar primary key, v1 varchar, v2 varchar, v3 varchar)";
+ Statement stmt = conn.createStatement();
+ stmt.execute(ddl);
+ String indexName = "R_ASYNCIND_" + generateUniqueName();
+
+ ddl = "CREATE INDEX " + indexName + "1 ON " + testTable + " (v1) ";
+ stmt.execute(ddl);
+ ddl = "CREATE INDEX " + indexName + "2 ON " + testTable + " (v2) ";
+ stmt.execute(ddl);
+ ddl = "CREATE INDEX " + indexName + "3 ON " + testTable + " (v3)";
+ stmt.execute(ddl);
+ conn.createStatement().execute("ALTER INDEX "+indexName+"1 ON " + testTable +" DISABLE ");
+ conn.createStatement().execute("ALTER INDEX "+indexName+"2 ON " + testTable +" REBUILD ");
+ conn.createStatement().execute("ALTER INDEX "+indexName+"3 ON " + testTable +" REBUILD ASYNC");
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "select table_name, " + PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP + " " +
+ "from system.catalog (" + PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP + " " + PLong.INSTANCE.getSqlTypeName() + ") " +
+ "where " + PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP + " !=0 and table_name like 'R_ASYNCIND_%' " +
+ "order by table_name");
+ assertTrue(rs.next());
+ assertEquals(indexName + "3", rs.getString(1));
+ long asyncTimestamp = rs.getLong(2);
+ assertTrue("Async timestamp is recent timestamp", asyncTimestamp > startTimestamp);
+ PTable table = PhoenixRuntime.getTable(conn, indexName+"3");
+ assertEquals(table.getTimeStamp(), asyncTimestamp);
+ assertFalse(rs.next());
+ conn.createStatement().execute("ALTER INDEX "+indexName+"3 ON " + testTable +" DISABLE");
+ rs = conn.createStatement().executeQuery(
+ "select table_name, " + PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP + " " +
+ "from system.catalog (" + PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP + " " + PLong.INSTANCE.getSqlTypeName() + ") " +
+ "where " + PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP + " !=0 and table_name like 'ASYNCIND_%' " +
+ "order by table_name" );
+ assertFalse(rs.next());
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
index 687b2c2..e9205c9 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
@@ -102,7 +102,7 @@ public class MutableIndexFailureIT extends BaseTest {
this.tableDDLOptions = " SALT_BUCKETS=2 " + (transactional ? ", TRANSACTIONAL=true " : "");
this.tableName = (localIndex ? "L_" : "") + TestUtil.DEFAULT_DATA_TABLE_NAME + (transactional ? "_TXN" : "")
+ (isNamespaceMapped ? "_NM" : "");
- this.indexName = INDEX_NAME;
+ this.indexName = FailingRegionObserver.INDEX_NAME;
fullTableName = SchemaUtil.getTableName(schema, tableName);
this.fullIndexName = SchemaUtil.getTableName(schema, indexName);
this.isNamespaceMapped = isNamespaceMapped;
@@ -155,7 +155,7 @@ public class MutableIndexFailureIT extends BaseTest {
rs = conn.createStatement().executeQuery(query);
assertFalse(rs.next());
- FAIL_WRITE = false;
+ FailingRegionObserver.FAIL_WRITE = false;
conn.createStatement().execute(
"CREATE " + (localIndex ? "LOCAL " : "") + "INDEX " + indexName + " ON " + fullTableName + " (v1) INCLUDE (v2)");
// Create other index which should be local/global if the other index is global/local to
@@ -202,7 +202,7 @@ public class MutableIndexFailureIT extends BaseTest {
assertEquals("z", rs.getString(2));
assertFalse(rs.next());
- FAIL_WRITE = true;
+ FailingRegionObserver.FAIL_WRITE = true;
updateTable(conn, fullTableName);
updateTable(conn, secondTableName);
// Verify the metadata for index is correct.
@@ -259,7 +259,7 @@ public class MutableIndexFailureIT extends BaseTest {
}
// re-enable index table
- FAIL_WRITE = false;
+ FailingRegionObserver.FAIL_WRITE = false;
waitForIndexToBeActive(conn,indexName);
waitForIndexToBeActive(conn,indexName+"_2");
waitForIndexToBeActive(conn,secondIndexName);
@@ -391,6 +391,8 @@ public class MutableIndexFailureIT extends BaseTest {
}
public static class FailingRegionObserver extends SimpleRegionObserver {
+ public static volatile boolean FAIL_WRITE = false;
+ public static final String INDEX_NAME = "IDX";
@Override
public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c, MiniBatchOperationInProgress<Mutation> miniBatchOp) throws HBaseIOException {
if (c.getEnvironment().getRegionInfo().getTable().getNameAsString().contains(INDEX_NAME) && FAIL_WRITE) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/antlr3/PhoenixSQL.g
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/antlr3/PhoenixSQL.g b/phoenix-core/src/main/antlr3/PhoenixSQL.g
index 3e09766..07a51ce 100644
--- a/phoenix-core/src/main/antlr3/PhoenixSQL.g
+++ b/phoenix-core/src/main/antlr3/PhoenixSQL.g
@@ -567,8 +567,8 @@ drop_index_node returns [DropIndexStatement ret]
// Parse a alter index statement
alter_index_node returns [AlterIndexStatement ret]
- : ALTER INDEX (IF ex=EXISTS)? i=index_name ON t=from_table_name s=(USABLE | UNUSABLE | REBUILD | DISABLE | ACTIVE)
- {ret = factory.alterIndex(factory.namedTable(null, TableName.create(t.getSchemaName(), i.getName())), t.getTableName(), ex!=null, PIndexState.valueOf(SchemaUtil.normalizeIdentifier(s.getText()))); }
+ : ALTER INDEX (IF ex=EXISTS)? i=index_name ON t=from_table_name s=(USABLE | UNUSABLE | REBUILD | DISABLE | ACTIVE) (async=ASYNC)?
+ {ret = factory.alterIndex(factory.namedTable(null, TableName.create(t.getSchemaName(), i.getName())), t.getTableName(), ex!=null, PIndexState.valueOf(SchemaUtil.normalizeIdentifier(s.getText())), async!=null); }
;
// Parse a trace statement.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index 38dc494..a0681fb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@ -3296,11 +3296,13 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
List<Cell> newKVs = tableMetadata.get(0).getFamilyCellMap().get(TABLE_FAMILY_BYTES);
Cell newKV = null;
int disableTimeStampKVIndex = -1;
+ int indexStateKVIndex = 0;
int index = 0;
for(Cell cell : newKVs){
if(Bytes.compareTo(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
INDEX_STATE_BYTES, 0, INDEX_STATE_BYTES.length) == 0){
newKV = cell;
+ indexStateKVIndex = index;
} else if (Bytes.compareTo(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
INDEX_DISABLE_TIMESTAMP_BYTES, 0, INDEX_DISABLE_TIMESTAMP_BYTES.length) == 0){
disableTimeStampKVIndex = index;
@@ -3378,11 +3380,11 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
if ((currentState == PIndexState.UNUSABLE && newState == PIndexState.ACTIVE)
|| (currentState == PIndexState.ACTIVE && newState == PIndexState.UNUSABLE)) {
newState = PIndexState.INACTIVE;
- newKVs.set(0, KeyValueUtil.newKeyValue(key, TABLE_FAMILY_BYTES,
+ newKVs.set(indexStateKVIndex, KeyValueUtil.newKeyValue(key, TABLE_FAMILY_BYTES,
INDEX_STATE_BYTES, timeStamp, Bytes.toBytes(newState.getSerializedValue())));
} else if (currentState == PIndexState.INACTIVE && newState == PIndexState.USABLE) {
newState = PIndexState.ACTIVE;
- newKVs.set(0, KeyValueUtil.newKeyValue(key, TABLE_FAMILY_BYTES,
+ newKVs.set(indexStateKVIndex, KeyValueUtil.newKeyValue(key, TABLE_FAMILY_BYTES,
INDEX_STATE_BYTES, timeStamp, Bytes.toBytes(newState.getSerializedValue())));
}
@@ -3414,7 +3416,8 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
if(dataTableKey != null) {
metaDataCache.invalidate(new ImmutableBytesPtr(dataTableKey));
}
- if (setRowKeyOrderOptimizableCell || disableTimeStampKVIndex != -1) {
+ if (setRowKeyOrderOptimizableCell || disableTimeStampKVIndex != -1
+ || currentState == PIndexState.DISABLE || newState == PIndexState.BUILDING) {
returnTable = doGetTable(key, HConstants.LATEST_TIMESTAMP, rowLock);
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataRegionObserver.java
index e790b59..a60de03 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataRegionObserver.java
@@ -17,12 +17,19 @@
*/
package org.apache.phoenix.coprocessor;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_SCHEM;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TENANT_ID;
import static org.apache.phoenix.schema.types.PDataType.TRUE_BYTES;
import java.io.IOException;
+import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -45,6 +52,7 @@ 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.filter.CompareFilter;
+import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
@@ -69,6 +77,7 @@ import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.MetaDataClient;
import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TableNotFoundException;
@@ -97,6 +106,7 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
private boolean enableRebuildIndex = QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD;
private long rebuildIndexTimeInterval = QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_INTERVAL;
private boolean blockWriteRebuildIndex = false;
+ private static Map<PName, Long> batchExecutedPerTableMap = new HashMap<PName, Long>();
@Override
public void preClose(final ObserverContext<RegionCoprocessorEnvironment> c,
@@ -125,6 +135,7 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_INTERVAL);
blockWriteRebuildIndex = env.getConfiguration().getBoolean(QueryServices.INDEX_FAILURE_BLOCK_WRITE,
QueryServicesOptions.DEFAULT_INDEX_FAILURE_BLOCK_WRITE);
+
}
@Override
@@ -195,9 +206,15 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
// running
private final static AtomicInteger inProgress = new AtomicInteger(0);
RegionCoprocessorEnvironment env;
+ private long rebuildIndexBatchSize = HConstants.LATEST_TIMESTAMP;
+ private long configuredBatches = 10;
public BuildIndexScheduleTask(RegionCoprocessorEnvironment env) {
this.env = env;
+ this.rebuildIndexBatchSize = env.getConfiguration().getLong(
+ QueryServices.INDEX_FAILURE_HANDLING_REBUILD_PERIOD, HConstants.LATEST_TIMESTAMP);
+ this.configuredBatches = env.getConfiguration().getLong(
+ QueryServices.INDEX_FAILURE_HANDLING_REBUILD_NUMBER_OF_BATCHES_PER_TABLE, configuredBatches);
}
@Override
@@ -228,6 +245,7 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
PhoenixDatabaseMetaData.INDEX_STATE_BYTES);
scan.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES);
+ PreparedStatement updateDisabledTimeStampSmt = null;
Map<PTable, List<PTable>> dataTableToIndexesMap = null;
MetaDataClient client = null;
@@ -243,8 +261,13 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
Result r = Result.create(results);
byte[] disabledTimeStamp = r.getValue(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES);
+ byte[] indexState = r.getValue(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.INDEX_STATE_BYTES);
+
+ if (disabledTimeStamp == null || disabledTimeStamp.length == 0 || (indexState != null
+ && PIndexState.BUILDING == PIndexState.fromSerializedValue(Bytes.toString(indexState)))) {
- if (disabledTimeStamp == null || disabledTimeStamp.length == 0) {
+ // Don't rebuild the building index , because they are marked for aysnc
continue;
}
@@ -255,8 +278,6 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
}
byte[] dataTable = r.getValue(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
PhoenixDatabaseMetaData.DATA_TABLE_NAME_BYTES);
- byte[] indexState = r.getValue(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
- PhoenixDatabaseMetaData.INDEX_STATE_BYTES);
if ((dataTable == null || dataTable.length == 0) || (indexState == null || indexState.length == 0)) {
// data table name can't be empty
continue;
@@ -317,109 +338,169 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
indexesToPartiallyRebuild.add(indexPTable);
} while (hasMore);
- if (dataTableToIndexesMap != null) {
- long overlapTime = env.getConfiguration().getLong(
- QueryServices.INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_TIME_ATTRIB,
- QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_TIME);
- for (Map.Entry<PTable, List<PTable>> entry : dataTableToIndexesMap.entrySet()) {
- PTable dataPTable = entry.getKey();
- List<PTable> indexesToPartiallyRebuild = entry.getValue();
- try {
- long earliestDisableTimestamp = Long.MAX_VALUE;
- List<IndexMaintainer> maintainers = Lists
- .newArrayListWithExpectedSize(indexesToPartiallyRebuild.size());
- for (PTable index : indexesToPartiallyRebuild) {
- long disabledTimeStampVal = index.getIndexDisableTimestamp();
- if (disabledTimeStampVal > 0) {
- if (disabledTimeStampVal < earliestDisableTimestamp) {
- earliestDisableTimestamp = disabledTimeStampVal;
- }
-
- maintainers.add(index.getIndexMaintainer(dataPTable, conn));
- }
- }
- // No indexes are disabled, so skip this table
- if (earliestDisableTimestamp == Long.MAX_VALUE) {
- continue;
- }
-
- long timeStamp = Math.max(0, earliestDisableTimestamp - overlapTime);
- LOG.info("Starting to build " + dataPTable + " indexes " + indexesToPartiallyRebuild
- + " from timestamp=" + timeStamp);
- TableRef tableRef = new TableRef(null, dataPTable, HConstants.LATEST_TIMESTAMP, false);
- // TODO Need to set high timeout
- PostDDLCompiler compiler = new PostDDLCompiler(conn);
- MutationPlan plan = compiler.compile(Collections.singletonList(tableRef), null, null, null,
- HConstants.LATEST_TIMESTAMP);
- Scan dataTableScan = IndexManagementUtil.newLocalStateScan(plan.getContext().getScan(),
- maintainers);
- dataTableScan.setTimeRange(timeStamp, HConstants.LATEST_TIMESTAMP);
- dataTableScan.setCacheBlocks(false);
- dataTableScan.setAttribute(BaseScannerRegionObserver.REBUILD_INDEXES, TRUE_BYTES);
-
- ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable(
- ByteUtil.EMPTY_BYTE_ARRAY);
- IndexMaintainer.serializeAdditional(dataPTable, indexMetaDataPtr, indexesToPartiallyRebuild,
- conn);
- byte[] attribValue = ByteUtil.copyKeyBytesIfNecessary(indexMetaDataPtr);
-
- dataTableScan.setAttribute(PhoenixIndexCodec.INDEX_MD, attribValue);
- MutationState mutationState = plan.execute();
- long rowCount = mutationState.getUpdateCount();
- LOG.info(rowCount + " rows of index which are rebuild");
- for (PTable indexPTable : indexesToPartiallyRebuild) {
- String indexTableFullName = SchemaUtil.getTableName(indexPTable.getSchemaName()
- .getString(), indexPTable.getTableName().getString());
- updateIndexState(conn, indexTableFullName, env, PIndexState.INACTIVE, PIndexState.ACTIVE);
- }
- } catch (Exception e) { // Log, but try next table's indexes
- LOG.warn("Unable to rebuild " + dataPTable + " indexes " + indexesToPartiallyRebuild
- + ". Will try again next on next scheduled invocation.", e);
- }
- }
- }
- } catch (Throwable t) {
- LOG.warn("ScheduledBuildIndexTask failed!", t);
- } finally {
- inProgress.decrementAndGet();
- if (scanner != null) {
- try {
- scanner.close();
- } catch (IOException ignored) {
- LOG.debug("ScheduledBuildIndexTask can't close scanner.", ignored);
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException ignored) {
- LOG.debug("ScheduledBuildIndexTask can't close connection", ignored);
- }
- }
- }
+ if (dataTableToIndexesMap != null) {
+ long overlapTime = env.getConfiguration().getLong(
+ QueryServices.INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_TIME_ATTRIB,
+ QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_TIME);
+ for (Map.Entry<PTable, List<PTable>> entry : dataTableToIndexesMap.entrySet()) {
+ PTable dataPTable = entry.getKey();
+ List<PTable> indexesToPartiallyRebuild = entry.getValue();
+ ReadOnlyProps props = new ReadOnlyProps(env.getConfiguration().iterator());
+ try (HTableInterface metaTable = env.getTable(
+ SchemaUtil.getPhysicalName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, props))) {
+ long earliestDisableTimestamp = Long.MAX_VALUE;
+ List<IndexMaintainer> maintainers = Lists
+ .newArrayListWithExpectedSize(indexesToPartiallyRebuild.size());
+ for (PTable index : indexesToPartiallyRebuild) {
+ long disabledTimeStampVal = index.getIndexDisableTimestamp();
+ if (disabledTimeStampVal > 0) {
+ if (disabledTimeStampVal < earliestDisableTimestamp) {
+ earliestDisableTimestamp = disabledTimeStampVal;
+ }
+
+ maintainers.add(index.getIndexMaintainer(dataPTable, conn));
+ }
+ }
+ // No indexes are disabled, so skip this table
+ if (earliestDisableTimestamp == Long.MAX_VALUE) {
+ continue;
+ }
+ long timeStamp = Math.max(0, earliestDisableTimestamp - overlapTime);
+ LOG.info("Starting to build " + dataPTable + " indexes " + indexesToPartiallyRebuild
+ + " from timestamp=" + timeStamp);
+
+ TableRef tableRef = new TableRef(null, dataPTable, HConstants.LATEST_TIMESTAMP, false);
+ // TODO Need to set high timeout
+ PostDDLCompiler compiler = new PostDDLCompiler(conn);
+ MutationPlan plan = compiler.compile(Collections.singletonList(tableRef), null, null, null,
+ HConstants.LATEST_TIMESTAMP);
+ Scan dataTableScan = IndexManagementUtil.newLocalStateScan(plan.getContext().getScan(),
+ maintainers);
+
+ long scanEndTime = getTimestampForBatch(timeStamp,
+ batchExecutedPerTableMap.get(dataPTable.getName()));
+ dataTableScan.setTimeRange(timeStamp, scanEndTime);
+ dataTableScan.setCacheBlocks(false);
+ dataTableScan.setAttribute(BaseScannerRegionObserver.REBUILD_INDEXES, TRUE_BYTES);
+
+ ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable(
+ ByteUtil.EMPTY_BYTE_ARRAY);
+ IndexMaintainer.serializeAdditional(dataPTable, indexMetaDataPtr, indexesToPartiallyRebuild,
+ conn);
+ byte[] attribValue = ByteUtil.copyKeyBytesIfNecessary(indexMetaDataPtr);
+ dataTableScan.setAttribute(PhoenixIndexCodec.INDEX_MD, attribValue);
+ MutationState mutationState = plan.execute();
+ long rowCount = mutationState.getUpdateCount();
+ LOG.info(rowCount + " rows of index which are rebuild");
+ for (PTable indexPTable : indexesToPartiallyRebuild) {
+ String indexTableFullName = SchemaUtil.getTableName(
+ indexPTable.getSchemaName().getString(),
+ indexPTable.getTableName().getString());
+ if (scanEndTime == HConstants.LATEST_TIMESTAMP) {
+ updateIndexState(conn, indexTableFullName, env, PIndexState.INACTIVE,
+ PIndexState.ACTIVE);
+ batchExecutedPerTableMap.remove(dataPTable.getName());
+ } else {
+
+ updateDisableTimestamp(conn, indexTableFullName, env, scanEndTime, metaTable);
+ Long noOfBatches = batchExecutedPerTableMap.get(dataPTable.getName());
+ if (noOfBatches == null) {
+ noOfBatches = 0l;
+ }
+ batchExecutedPerTableMap.put(dataPTable.getName(), ++noOfBatches);
+ // clearing cache to get the updated
+ // disabled timestamp
+ new MetaDataClient(conn).updateCache(dataPTable.getSchemaName().getString(),
+ dataPTable.getTableName().getString());
+ new MetaDataClient(conn).updateCache(indexPTable.getSchemaName().getString(),
+ indexPTable.getTableName().getString());
+ LOG.info(
+ "During Round-robin build: Successfully updated index disabled timestamp for "
+ + indexTableFullName + " to " + scanEndTime);
+ }
+
+ }
+ } catch (Exception e) { // Log, but try next table's
+ // indexes
+ LOG.warn("Unable to rebuild " + dataPTable + " indexes " + indexesToPartiallyRebuild
+ + ". Will try again next on next scheduled invocation.", e);
+ }
+ }
+ }
+ } catch (Throwable t) {
+ LOG.warn("ScheduledBuildIndexTask failed!", t);
+ } finally {
+ inProgress.decrementAndGet();
+ if (scanner != null) {
+ try {
+ scanner.close();
+ } catch (IOException ignored) {
+ LOG.debug("ScheduledBuildIndexTask can't close scanner.", ignored);
+ }
+ }
+ if (conn != null) {
+ try {
+ conn.close();
+ } catch (SQLException ignored) {
+ LOG.debug("ScheduledBuildIndexTask can't close connection", ignored);
+ }
+ }
+ }
}
- }
-
- private static void updateIndexState(PhoenixConnection conn, String indexTableName, RegionCoprocessorEnvironment env, PIndexState oldState,
- PIndexState newState) throws ServiceException, Throwable {
- byte[] indexTableKey = SchemaUtil.getTableKeyFromFullName(indexTableName);
- String schemaName = SchemaUtil.getSchemaNameFromFullName(indexTableName);
- String indexName = SchemaUtil.getTableNameFromFullName(indexTableName);
- // Mimic the Put that gets generated by the client on an update of the index state
- Put put = new Put(indexTableKey);
- put.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES,
- newState.getSerializedBytes());
- if (newState == PIndexState.ACTIVE) {
- put.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES,
- PLong.INSTANCE.toBytes(0));
+
+ private long getTimestampForBatch(long disabledTimeStamp, Long noOfBatches) {
+ if (disabledTimeStamp < 0 || rebuildIndexBatchSize > (HConstants.LATEST_TIMESTAMP
+ - disabledTimeStamp)) { return HConstants.LATEST_TIMESTAMP; }
+ long timestampForNextBatch = disabledTimeStamp + rebuildIndexBatchSize;
+ if (timestampForNextBatch < 0 || timestampForNextBatch > System.currentTimeMillis()
+ || (noOfBatches != null && noOfBatches > configuredBatches)) {
+ // if timestampForNextBatch cross current time , then we should
+ // build the complete index
+ timestampForNextBatch = HConstants.LATEST_TIMESTAMP;
+ }
+ return timestampForNextBatch;
}
- final List<Mutation> tableMetadata = Collections.<Mutation> singletonList(put);
- MetaDataMutationResult result = conn.getQueryServices().updateIndexState(tableMetadata, null);
- MutationCode code = result.getMutationCode();
- if (code == MutationCode.TABLE_NOT_FOUND) { throw new TableNotFoundException(schemaName, indexName); }
- if (code == MutationCode.UNALLOWED_TABLE_MUTATION) { throw new SQLExceptionInfo.Builder(
- SQLExceptionCode.INVALID_INDEX_STATE_TRANSITION)
- .setMessage(" currentState=" + oldState + ". requestedState=" + newState).setSchemaName(schemaName)
- .setTableName(indexName).build().buildException(); }
}
+
+ private static void updateIndexState(PhoenixConnection conn, String indexTableName,
+ RegionCoprocessorEnvironment env, PIndexState oldState, PIndexState newState)
+ throws ServiceException, Throwable {
+ byte[] indexTableKey = SchemaUtil.getTableKeyFromFullName(indexTableName);
+ String schemaName = SchemaUtil.getSchemaNameFromFullName(indexTableName);
+ String indexName = SchemaUtil.getTableNameFromFullName(indexTableName);
+ // Mimic the Put that gets generated by the client on an update of the
+ // index state
+ Put put = new Put(indexTableKey);
+ put.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES,
+ newState.getSerializedBytes());
+ if (newState == PIndexState.ACTIVE) {
+ put.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES, PLong.INSTANCE.toBytes(0));
+ put.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP_BYTES, PLong.INSTANCE.toBytes(0));
+ }
+ final List<Mutation> tableMetadata = Collections.<Mutation> singletonList(put);
+ MetaDataMutationResult result = conn.getQueryServices().updateIndexState(tableMetadata, null);
+ MutationCode code = result.getMutationCode();
+ if (code == MutationCode.TABLE_NOT_FOUND) {
+ throw new TableNotFoundException(schemaName, indexName);
+ }
+ if (code == MutationCode.UNALLOWED_TABLE_MUTATION) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.INVALID_INDEX_STATE_TRANSITION)
+ .setMessage(" currentState=" + oldState + ". requestedState=" + newState).setSchemaName(schemaName)
+ .setTableName(indexName).build().buildException();
+ }
+ }
+
+ private static void updateDisableTimestamp(PhoenixConnection conn, String indexTableName,
+ RegionCoprocessorEnvironment env, long disabledTimestamp, HTableInterface metaTable) throws IOException {
+ byte[] indexTableKey = SchemaUtil.getTableKeyFromFullName(indexTableName);
+ Put put = new Put(indexTableKey);
+ put.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES,
+ PLong.INSTANCE.toBytes(disabledTimestamp));
+ metaTable.checkAndPut(indexTableKey, PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.INDEX_STATE_BYTES, CompareOp.EQUAL, PIndexState.INACTIVE.getSerializedBytes(),
+ put);
+
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
index fb4e3c3..fde403c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@ -421,7 +421,8 @@ public enum SQLExceptionCode {
724, "43M07", "Schema name not allowed!!"), CREATE_SCHEMA_NOT_ALLOWED(725, "43M08",
"Cannot create schema because config " + QueryServices.IS_NAMESPACE_MAPPING_ENABLED
+ " for enabling name space mapping isn't enabled."), INCONSISTENET_NAMESPACE_MAPPING_PROPERTIES(
- 726, "43M10", " Inconsistent namespace mapping properites..");
+ 726, "43M10", " Inconsistent namespace mapping properites.."), ASYNC_NOT_ALLOWED(
+ 727, "43M11", " ASYNC option is not allowed.. ");
private final int errorCode;
private final String sqlState;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
index eb73d6b..e515dbb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
@@ -35,30 +35,21 @@ import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.Stoppable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.client.Put;
-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.phoenix.coprocessor.MetaDataProtocol.MetaDataMutationResult;
import org.apache.phoenix.coprocessor.MetaDataProtocol.MutationCode;
-import org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataResponse;
-import org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataService;
-import org.apache.phoenix.coprocessor.generated.MetaDataProtos.UpdateIndexStateRequest;
import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
import org.apache.phoenix.hbase.index.write.DelegateIndexFailurePolicy;
import org.apache.phoenix.hbase.index.write.KillServerOnFailurePolicy;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
-import org.apache.phoenix.protobuf.ProtobufUtil;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
-import org.apache.phoenix.schema.types.PLong;
+import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.QueryUtil;
@@ -153,47 +144,10 @@ public class PhoenixIndexFailurePolicy extends DelegateIndexFailurePolicy {
String indexTableName = tableTimeElement.getKey();
long minTimeStamp = tableTimeElement.getValue();
// Disable the index by using the updateIndexState method of MetaDataProtocol end point coprocessor.
- byte[] indexTableKey = SchemaUtil.getTableKeyFromFullName(indexTableName);
HTableInterface systemTable = env.getTable(SchemaUtil
.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, env.getConfiguration()));
- // Mimic the Put that gets generated by the client on an update of the index state
- Put put = new Put(indexTableKey);
- if (blockWriteRebuildIndex)
- put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES,
- PIndexState.ACTIVE.getSerializedBytes());
- else
- put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES,
- PIndexState.DISABLE.getSerializedBytes());
- put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES,
- PLong.INSTANCE.toBytes(minTimeStamp));
- final List<Mutation> tableMetadata = Collections.<Mutation>singletonList(put);
-
- final Map<byte[], MetaDataResponse> results =
- systemTable.coprocessorService(MetaDataService.class, indexTableKey, indexTableKey,
- new Batch.Call<MetaDataService, MetaDataResponse>() {
- @Override
- public MetaDataResponse call(MetaDataService instance) throws IOException {
- ServerRpcController controller = new ServerRpcController();
- BlockingRpcCallback<MetaDataResponse> rpcCallback =
- new BlockingRpcCallback<MetaDataResponse>();
- UpdateIndexStateRequest.Builder builder = UpdateIndexStateRequest.newBuilder();
- for (Mutation m : tableMetadata) {
- MutationProto mp = ProtobufUtil.toProto(m);
- builder.addTableMetadataMutations(mp.toByteString());
- }
- instance.updateIndexState(controller, builder.build(), rpcCallback);
- if (controller.getFailedOn() != null) {
- throw controller.getFailedOn();
- }
- return rpcCallback.get();
- }
- });
- if (results.isEmpty()) {
- throw new IOException("Didn't get expected result size");
- }
- MetaDataResponse tmpResponse = results.values().iterator().next();
- MetaDataMutationResult result = MetaDataMutationResult.constructFromProto(tmpResponse);
-
+ MetaDataMutationResult result = IndexUtil.disableIndexWithTimestamp(indexTableName, minTimeStamp,
+ systemTable, blockWriteRebuildIndex);
if (result.getMutationCode() == MutationCode.TABLE_NOT_FOUND) {
LOG.info("Index " + indexTableName + " has been dropped. Ignore uncommitted mutations");
continue;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 54080d1..5142b57 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -279,6 +279,8 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final String LAST_STATS_UPDATE_TIME = "LAST_STATS_UPDATE_TIME";
public static final byte[] LAST_STATS_UPDATE_TIME_BYTES = Bytes.toBytes(LAST_STATS_UPDATE_TIME);
public static final String GUIDE_POST_KEY = "GUIDE_POST_KEY";
+ public static final String ASYNC_REBUILD_TIMESTAMP = "ASYNC_REBUILD_TIMESTAMP";
+ public static final byte[] ASYNC_REBUILD_TIMESTAMP_BYTES = Bytes.toBytes(ASYNC_REBUILD_TIMESTAMP);
public static final String PARENT_TENANT_ID = "PARENT_TENANT_ID";
public static final byte[] PARENT_TENANT_ID_BYTES = Bytes.toBytes(PARENT_TENANT_ID);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/83827cd8/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index d57c250..f3c6d30 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -989,8 +989,8 @@ public class PhoenixStatement implements Statement, SQLCloseable {
private static class ExecutableAlterIndexStatement extends AlterIndexStatement implements CompilableStatement {
- public ExecutableAlterIndexStatement(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState state) {
- super(indexTableNode, dataTableName, ifExists, state);
+ public ExecutableAlterIndexStatement(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState state, boolean async) {
+ super(indexTableNode, dataTableName, ifExists, state, async);
}
@SuppressWarnings("unchecked")
@@ -1302,8 +1302,8 @@ public class PhoenixStatement implements Statement, SQLCloseable {
}
@Override
- public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState state) {
- return new ExecutableAlterIndexStatement(indexTableNode, dataTableName, ifExists, state);
+ public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState state, boolean async) {
+ return new ExecutableAlterIndexStatement(indexTableNode, dataTableName, ifExists, state, async);
}
@Override