You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by td...@apache.org on 2016/07/11 22:15:14 UTC
phoenix git commit: Added ability to run async indexes when hbase
cluster is in non-distributed mode or when mr is in local mode
Repository: phoenix
Updated Branches:
refs/heads/4.x-HBase-0.98 cf4e33b33 -> 04df7bca0
Added ability to run async indexes when hbase cluster is in non-distributed mode or when mr is in local mode
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/04df7bca
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/04df7bca
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/04df7bca
Branch: refs/heads/4.x-HBase-0.98
Commit: 04df7bca07607f9db7f7bba5f726bff6013b81e0
Parents: cf4e33b
Author: tejamobref <te...@mobref.io>
Authored: Tue Jul 12 01:37:12 2016 +0530
Committer: Thomas D'Silva <td...@salesforce.com>
Committed: Mon Jul 11 15:15:09 2016 -0700
----------------------------------------------------------------------
.../phoenix/end2end/index/AsyncIndexIT.java | 180 +++++++++++++++++++
.../coprocessor/MetaDataRegionObserver.java | 113 ++++++++++--
.../phoenix/jdbc/PhoenixEmbeddedDriver.java | 4 +-
.../index/automation/PhoenixMRJobSubmitter.java | 13 +-
.../apache/phoenix/query/QueryConstants.java | 18 +-
.../org/apache/phoenix/query/QueryServices.java | 4 +-
6 files changed, 308 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04df7bca/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/AsyncIndexIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/AsyncIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/AsyncIndexIT.java
new file mode 100644
index 0000000..43d1bd9
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/AsyncIndexIT.java
@@ -0,0 +1,180 @@
+/*
+ * 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.index;
+
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ASYNC_CREATED_DATE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_FAMILY;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TABLE_NAME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE;
+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.TABLE_TYPE;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.query.BaseTest;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.types.PDate;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.junit.After;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class AsyncIndexIT extends BaseTest {
+
+ private static final String PERSON_TABLE_NAME = "PERSON";
+ private static final String PERSON_TABLE_NAME_WITH_SCHEMA = "TEST.PERSON";
+ private static final String TEST_SCHEMA = "TEST";
+
+ private static final String PERSON_TABLE_ASYNC_INDEX_INFO_QUERY = "SELECT "
+ + DATA_TABLE_NAME + ", " + TABLE_SCHEM + ", "
+ + TABLE_NAME + " FROM " + SYSTEM_CATALOG_SCHEMA + ".\""
+ + SYSTEM_CATALOG_TABLE + "\""
+ + " (" + ASYNC_CREATED_DATE + " "
+ + PDate.INSTANCE.getSqlTypeName() + ") " + " WHERE "
+ + COLUMN_NAME + " IS NULL and " + COLUMN_FAMILY + " IS NULL and "
+ + ASYNC_CREATED_DATE + " IS NOT NULL and "
+ + TABLE_TYPE + " = '" + PTableType.INDEX.getSerializedValue()
+ + "' and DATA_TABLE_NAME='" + PERSON_TABLE_NAME
+ + "' and TABLE_SCHEM='" + TEST_SCHEMA + "' and "
+ + PhoenixDatabaseMetaData.INDEX_STATE + " = '"
+ + PIndexState.BUILDING.getSerializedValue() + "'";
+
+ private void dropTable(Statement stmt) throws SQLException, IOException {
+ stmt.execute("DROP TABLE IF EXISTS " + PERSON_TABLE_NAME_WITH_SCHEMA);
+ }
+
+ private void createTableAndLoadData(Statement stmt) throws SQLException {
+ String ddl = "CREATE TABLE " + PERSON_TABLE_NAME_WITH_SCHEMA + " (ID INTEGER NOT NULL PRIMARY KEY, " +
+ "FNAME VARCHAR, LNAME VARCHAR)";
+
+ stmt.execute(ddl);
+ stmt.execute("UPSERT INTO " + PERSON_TABLE_NAME_WITH_SCHEMA + " values(1, 'FIRST', 'F')");
+ stmt.execute("UPSERT INTO " + PERSON_TABLE_NAME_WITH_SCHEMA + " values(2, 'SECOND', 'S')");
+ }
+
+ private void createAsyncIndex(Statement stmt) throws SQLException {
+ stmt.execute("CREATE INDEX FNAME_INDEX ON " + PERSON_TABLE_NAME_WITH_SCHEMA + "(FNAME) ASYNC");
+ }
+
+ private void dropAsyncIndex(Statement stmt) throws SQLException {
+ stmt.execute("DROP INDEX IF EXISTS FNAME_INDEX ON " + PERSON_TABLE_NAME_WITH_SCHEMA);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ tearDownMiniCluster();
+ }
+
+ private void retryWithSleep(int maxRetries, int sleepInSecs, Statement stmt) throws Exception {
+ ResultSet rs = stmt.executeQuery(PERSON_TABLE_ASYNC_INDEX_INFO_QUERY);
+ // Wait for max of 5 retries with each retry of 5 sec sleep
+ int retries = 0;
+ while(retries <= maxRetries) {
+ Thread.sleep(sleepInSecs * 1000);
+ rs = stmt.executeQuery(PERSON_TABLE_ASYNC_INDEX_INFO_QUERY);
+ if (!rs.next()) {
+ break;
+ }
+ retries++;
+ }
+ }
+
+ @Test
+ public void testAsyncIndexBuilderNonDistributed() throws Exception {
+ setUpTestDriver(ReadOnlyProps.EMPTY_PROPS);
+ Connection conn = DriverManager.getConnection(getUrl());
+ Statement stmt = conn.createStatement();
+ createTableAndLoadData(stmt);
+ createAsyncIndex(stmt);
+
+ ResultSet rs = stmt.executeQuery(PERSON_TABLE_ASYNC_INDEX_INFO_QUERY);
+ assertTrue(rs.next());
+
+ retryWithSleep(5, 5, stmt);
+
+ rs = stmt.executeQuery(PERSON_TABLE_ASYNC_INDEX_INFO_QUERY);
+ assertFalse(rs.next());
+
+ dropAsyncIndex(stmt);
+ dropTable(stmt);
+ }
+
+ @Test
+ public void testAsyncIndexBuilderNonDistributedMapreduceYarn() throws Exception {
+ Map<String,String> props = new HashMap<>();
+ props.put(QueryServices.MAPRED_FRAMEWORK_NAME, "yarn");
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+
+ Connection conn = DriverManager.getConnection(getUrl());
+ Statement stmt = conn.createStatement();
+ createTableAndLoadData(stmt);
+ createAsyncIndex(stmt);
+
+ ResultSet rs = stmt.executeQuery(PERSON_TABLE_ASYNC_INDEX_INFO_QUERY);
+ assertTrue(rs.next());
+
+ retryWithSleep(5, 5, stmt);
+
+ rs = stmt.executeQuery(PERSON_TABLE_ASYNC_INDEX_INFO_QUERY);
+ assertFalse(rs.next());
+
+ dropAsyncIndex(stmt);
+ dropTable(stmt);
+ }
+
+ @Test
+ public void testAsyncIndexBuilderDistributed() throws Exception {
+ Map<String,String> props = new HashMap<>();
+ props.put(QueryServices.HBASE_CLUSTER_DISTRIBUTED_ATTRIB, "true");
+ props.put(QueryServices.MAPRED_FRAMEWORK_NAME, "yarn");
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+
+ Connection conn = DriverManager.getConnection(getUrl());
+ Statement stmt = conn.createStatement();
+ createTableAndLoadData(stmt);
+ createAsyncIndex(stmt);
+
+ ResultSet rs = stmt.executeQuery(PERSON_TABLE_ASYNC_INDEX_INFO_QUERY);
+ assertTrue(rs.next());
+
+ retryWithSleep(5, 5, stmt);
+
+ rs = stmt.executeQuery(PERSON_TABLE_ASYNC_INDEX_INFO_QUERY);
+ assertTrue(rs.next());
+
+ dropAsyncIndex(stmt);
+ dropTable(stmt);
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04df7bca/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 8d7d8e5..bf396a9 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,8 +17,12 @@
*/
package org.apache.phoenix.coprocessor;
+import static org.apache.phoenix.query.QueryConstants.ASYNC_INDEX_INFO_QUERY;
+
import java.io.IOException;
+import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -30,6 +34,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
@@ -87,10 +92,12 @@ import com.google.common.collect.Lists;
*/
public class MetaDataRegionObserver extends BaseRegionObserver {
public static final Log LOG = LogFactory.getLog(MetaDataRegionObserver.class);
- protected ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
+ protected ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2);
private boolean enableRebuildIndex = QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD;
private long rebuildIndexTimeInterval = QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_INTERVAL;
private boolean blockWriteRebuildIndex = false;
+ private final String HBASE_CLUSTER_DISTRIBUTED_CONFIG = "true";
+ private final String MAPRED_FRAMEWORK_YARN_CONFIG = "yarn";
@Override
public void preClose(final ObserverContext<RegionCoprocessorEnvironment> c,
@@ -158,23 +165,109 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
};
(new Thread(r)).start();
- if (!enableRebuildIndex && !blockWriteRebuildIndex) {
- LOG.info("Failure Index Rebuild is skipped by configuration.");
- return;
- }
// turn off verbose deprecation logging
Logger deprecationLogger = Logger.getLogger("org.apache.hadoop.conf.Configuration.deprecation");
if (deprecationLogger != null) {
deprecationLogger.setLevel(Level.WARN);
}
+
try {
Class.forName(PhoenixDriver.class.getName());
- // starts index rebuild schedule work
- BuildIndexScheduleTask task = new BuildIndexScheduleTask(e.getEnvironment());
- // run scheduled task every 10 secs
- executor.scheduleAtFixedRate(task, 10000, rebuildIndexTimeInterval, TimeUnit.MILLISECONDS);
} catch (ClassNotFoundException ex) {
- LOG.error("BuildIndexScheduleTask cannot start!", ex);
+ LOG.error("Phoenix Driver class is not found. Fix the classpath.", ex);
+ }
+
+ Configuration conf = env.getConfiguration();
+ String hbaseClusterDistributedMode = conf.get(QueryServices.HBASE_CLUSTER_DISTRIBUTED_ATTRIB);
+ String mapredFrameworkName = conf.get(QueryServices.MAPRED_FRAMEWORK_NAME);
+
+ // In case of non-distributed mode of hbase service or local mode of map reduce service, add timer task to rebuild the async indexes
+ if ((hbaseClusterDistributedMode != null && !hbaseClusterDistributedMode.equals(HBASE_CLUSTER_DISTRIBUTED_CONFIG)) ||
+ (mapredFrameworkName != null && !mapredFrameworkName.equals(MAPRED_FRAMEWORK_YARN_CONFIG)))
+ {
+ LOG.info("Enabling Async Index rebuilder");
+ AsyncIndexRebuilderTask asyncIndexRebuilderTask = new AsyncIndexRebuilderTask(e.getEnvironment());
+ // run async index rebuilder task every 10 secs to rebuild any newly created async indexes
+ executor.scheduleAtFixedRate(asyncIndexRebuilderTask, 10000, rebuildIndexTimeInterval, TimeUnit.MILLISECONDS);
+ }
+
+ if (!enableRebuildIndex && !blockWriteRebuildIndex) {
+ LOG.info("Failure Index Rebuild is skipped by configuration.");
+ return;
+ }
+
+ // starts index rebuild schedule work
+ BuildIndexScheduleTask task = new BuildIndexScheduleTask(e.getEnvironment());
+ // run scheduled task every 10 secs
+ executor.scheduleAtFixedRate(task, 10000, rebuildIndexTimeInterval, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Task runs periodically to re-build async indexes when hbase is running in non-distributed mode or
+ * when mapreduce is running in local mode
+ *
+ */
+ public static class AsyncIndexRebuilderTask extends TimerTask {
+ RegionCoprocessorEnvironment env;
+ PhoenixConnection conn = null;
+
+ public AsyncIndexRebuilderTask(RegionCoprocessorEnvironment env) {
+ this.env = env;
+ }
+
+ @Override
+ public void run() {
+ try {
+ if (conn == null) {
+ conn = QueryUtil.getConnectionOnServer(env.getConfiguration()).unwrap(PhoenixConnection.class);
+ }
+ Statement s = conn.createStatement();
+ ResultSet rs = s.executeQuery(ASYNC_INDEX_INFO_QUERY);
+
+ PhoenixConnection alterIndexConnection = null;
+ while (rs.next()) {
+ String tableName = rs.getString(PhoenixDatabaseMetaData.DATA_TABLE_NAME);
+ String tableSchema = rs.getString(PhoenixDatabaseMetaData.TABLE_SCHEM);
+ String indexName = rs.getString(PhoenixDatabaseMetaData.TABLE_NAME);
+ String tableNameWithSchema = SchemaUtil.getTableName(tableSchema, tableName);
+
+ final PTable pindexTable = PhoenixRuntime.getTable(conn, SchemaUtil.getTableName(tableSchema, indexName));
+ // this is set to ensure index tables remains consistent post population.
+ long maxTimeRange = pindexTable.getTimeStamp()+1;
+
+ try {
+ final Properties props = new Properties();
+ props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(maxTimeRange));
+ alterIndexConnection = QueryUtil.getConnectionOnServer(props, env.getConfiguration()).unwrap(PhoenixConnection.class);
+
+ // Alter index query for rebuilding async indexes
+ String alterIndexQuery = String.format("ALTER INDEX IF EXISTS %s ON %s REBUILD", indexName, tableNameWithSchema);
+
+ LOG.info("Executing Rebuild Index Query:" + alterIndexQuery);
+ alterIndexConnection.createStatement().execute(alterIndexQuery);
+ } catch (Throwable t) {
+ LOG.error("AsyncIndexRebuilderTask failed during rebuilding index!", t);
+ } finally {
+ if (alterIndexConnection != null) {
+ try {
+ alterIndexConnection.close();
+ } catch (SQLException ignored) {
+ LOG.debug("AsyncIndexRebuilderTask can't close alterIndexConnection", ignored);
+ }
+ }
+ }
+ }
+ } catch (Throwable t) {
+ LOG.error("AsyncIndexRebuilderTask failed!", t);
+ } finally {
+ if (conn != null) {
+ try {
+ conn.close();
+ } catch (SQLException ignored) {
+ LOG.debug("AsyncIndexRebuilderTask can't close connection", ignored);
+ }
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04df7bca/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java
index 40e03b9..d2dd94f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java
@@ -290,7 +290,7 @@ public abstract class PhoenixEmbeddedDriver implements Driver, SQLCloseable {
// Normalize connInfo so that a url explicitly specifying versus implicitly inheriting
// the default values will both share the same ConnectionQueryServices.
if (zookeeperQuorum == null) {
- zookeeperQuorum = props.get(QueryServices.ZOOKEEPER_QUARUM_ATTRIB);
+ zookeeperQuorum = props.get(QueryServices.ZOOKEEPER_QUORUM_ATTRIB);
if (zookeeperQuorum == null) {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.MALFORMED_CONNECTION_URL)
.setMessage(this.toString()).build().buildException();
@@ -357,7 +357,7 @@ public abstract class PhoenixEmbeddedDriver implements Driver, SQLCloseable {
public ReadOnlyProps asProps() {
Map<String, String> connectionProps = Maps.newHashMapWithExpectedSize(3);
if (getZookeeperQuorum() != null) {
- connectionProps.put(QueryServices.ZOOKEEPER_QUARUM_ATTRIB, getZookeeperQuorum());
+ connectionProps.put(QueryServices.ZOOKEEPER_QUORUM_ATTRIB, getZookeeperQuorum());
}
if (getPort() != null) {
connectionProps.put(QueryServices.ZOOKEEPER_PORT_ATTRIB, getPort().toString());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04df7bca/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
index 9b8d5c8..99f84ec 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
@@ -17,6 +17,8 @@
*/
package org.apache.phoenix.mapreduce.index.automation;
+import static org.apache.phoenix.query.QueryConstants.ASYNC_INDEX_INFO_QUERY;
+
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -67,15 +69,6 @@ public class PhoenixMRJobSubmitter {
"/phoenix/automated-mr-index-build-leader-election";
private static final String AUTO_INDEX_BUILD_LOCK_NAME = "ActiveStandbyElectorLock";
- private static final String CANDIDATE_INDEX_INFO_QUERY = "SELECT "
- + PhoenixDatabaseMetaData.INDEX_TYPE + ", " + PhoenixDatabaseMetaData.DATA_TABLE_NAME
- + ", " + PhoenixDatabaseMetaData.TABLE_SCHEM + ", "
- + PhoenixDatabaseMetaData.TABLE_NAME + " FROM "
- + PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME + " WHERE "
- + PhoenixDatabaseMetaData.COLUMN_NAME + " is null and "
- + PhoenixDatabaseMetaData.TABLE_TYPE + " = '" + PTableType.INDEX.getSerializedValue()
- + "' and " + PhoenixDatabaseMetaData.INDEX_STATE + " = '"
- + PIndexState.BUILDING.getSerializedValue() + "'";
// TODO - Move this to a property?
private static final int JOB_SUBMIT_POOL_TIMEOUT = 5;
private Configuration conf;
@@ -151,7 +144,7 @@ public class PhoenixMRJobSubmitter {
public Map<String, PhoenixAsyncIndex> getCandidateJobs() throws SQLException {
Connection con = DriverManager.getConnection("jdbc:phoenix:" + zkQuorum);
Statement s = con.createStatement();
- ResultSet rs = s.executeQuery(CANDIDATE_INDEX_INFO_QUERY);
+ ResultSet rs = s.executeQuery(ASYNC_INDEX_INFO_QUERY);
Map<String, PhoenixAsyncIndex> candidateIndexes = new HashMap<String, PhoenixAsyncIndex>();
while (rs.next()) {
PhoenixAsyncIndex indexInfo = new PhoenixAsyncIndex();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04df7bca/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
index 9f8f58c..8bc1c5b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@ -21,6 +21,7 @@ package org.apache.phoenix.query;
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.AUTO_PARTITION_SEQ;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.BASE_COLUMN_COUNT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.BUFFER_LENGTH;
@@ -115,9 +116,12 @@ import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.schema.MetaDataSplitPolicy;
+import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PNameFactory;
+import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.SortOrder;
+import org.apache.phoenix.schema.types.PDate;
import org.apache.phoenix.util.ByteUtil;
@@ -167,7 +171,19 @@ public interface QueryConstants {
public static final byte[] TRUE = new byte[] {1};
-
+ public static final String ASYNC_INDEX_INFO_QUERY = "SELECT "
+ + DATA_TABLE_NAME + ", " + TABLE_SCHEM + ", "
+ + TABLE_NAME + ", " + ASYNC_CREATED_DATE
+ + " FROM " + SYSTEM_CATALOG_SCHEMA + ".\""
+ + SYSTEM_CATALOG_TABLE + "\""
+ + " (" + ASYNC_CREATED_DATE + " "
+ + PDate.INSTANCE.getSqlTypeName() + ") " + " WHERE "
+ + COLUMN_NAME + " IS NULL and " + COLUMN_FAMILY + " IS NULL and "
+ + ASYNC_CREATED_DATE + " IS NOT NULL and "
+ + TABLE_TYPE + " = '" + PTableType.INDEX.getSerializedValue()
+ + "' and " + PhoenixDatabaseMetaData.INDEX_STATE + " = '"
+ + PIndexState.BUILDING.getSerializedValue() + "'";
+
/**
* Separator used between variable length keys for a composite key.
* Variable length data types may not use this byte value.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04df7bca/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 6df3d1e..3f3518b 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
@@ -107,7 +107,7 @@ public interface QueryServices extends SQLCloseable {
public static final String REGIONSERVER_LEASE_PERIOD_ATTRIB = "hbase.regionserver.lease.period";
public static final String RPC_TIMEOUT_ATTRIB = "hbase.rpc.timeout";
public static final String DYNAMIC_JARS_DIR_KEY = "hbase.dynamic.jars.dir";
- public static final String ZOOKEEPER_QUARUM_ATTRIB = "hbase.zookeeper.quorum";
+ public static final String ZOOKEEPER_QUORUM_ATTRIB = "hbase.zookeeper.quorum";
public static final String ZOOKEEPER_PORT_ATTRIB = "hbase.zookeeper.property.clientPort";
public static final String ZOOKEEPER_ROOT_NODE_ATTRIB = "zookeeper.znode.parent";
public static final String DISTINCT_VALUE_COMPRESS_THRESHOLD_ATTRIB = "phoenix.distinct.value.compress.threshold";
@@ -116,6 +116,8 @@ public interface QueryServices extends SQLCloseable {
public static final String MAX_SERVER_METADATA_CACHE_TIME_TO_LIVE_MS_ATTRIB = "phoenix.coprocessor.maxMetaDataCacheTimeToLiveMs";
public static final String MAX_SERVER_METADATA_CACHE_SIZE_ATTRIB = "phoenix.coprocessor.maxMetaDataCacheSize";
public static final String MAX_CLIENT_METADATA_CACHE_SIZE_ATTRIB = "phoenix.client.maxMetaDataCacheSize";
+ public static final String HBASE_CLUSTER_DISTRIBUTED_ATTRIB = "hbase.cluster.distributed";
+ public static final String MAPRED_FRAMEWORK_NAME = "mapreduce.framework.name";
public static final String AUTO_UPGRADE_WHITELIST_ATTRIB = "phoenix.client.autoUpgradeWhiteList";
// Mainly for testing to force spilling