You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@phoenix.apache.org by ankitsinghal <gi...@git.apache.org> on 2016/09/27 09:27:02 UTC
[GitHub] phoenix pull request #210: PHOENIX 2890 Extend IndexTool to allow incrementa...
GitHub user ankitsinghal opened a pull request:
https://github.com/apache/phoenix/pull/210
PHOENIX 2890 Extend IndexTool to allow incremental index rebuilds
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/ankitsinghal/phoenix PHOENIX-2890
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/phoenix/pull/210.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #210
----
commit 5cbf908184900e3708f5cf9471a529b8cf579e97
Author: Ankit Singhal <an...@gmail.com>
Date: 2016-09-27T09:25:59Z
PHOENIX 2890 Extend IndexTool to allow incremental index rebuilds
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83792778
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java ---
@@ -0,0 +1,264 @@
+/*
+ * 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.UUID;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.phoenix.end2end.index.MutableIndexFailureIT.FailingRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.mapreduce.index.IndexTool;
+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_BATCH_SIZE_ATTRIB, "1000");
+ serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_INTERVAL_ATTRIB, "2000");
+ 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 + " 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());
+
+ 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.
+ assertTrue(actualExplainPlan.contains(
--- End diff --
Use assertExplainPlan method to check the plan.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83793514
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildWithNamespaceEnabled.java ---
@@ -0,0 +1,71 @@
+/*
+ * 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.end2end.index.MutableIndexFailureIT.FailingRegionObserver;
+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_BATCH_SIZE_ATTRIB, "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 }
--- End diff --
We can make IndexToolForPartialBuildIT itself to run with namespace enabled and disabled. We don't need extra IT test.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83824633
--- Diff: phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java ---
@@ -85,7 +102,7 @@
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)");
--- End diff --
Better to have an argument for partial rebuilding than go by rebuild if we don't mention index table.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83827883
--- Diff: phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java ---
@@ -167,50 +180,152 @@ private void printHelpAndExit(Options options, int exitCode) {
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) throws Exception {
+ if (indexTable == null) {
+ 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);
+ job.setJarByClass(IndexTool.class);
+ FileOutputFormat.setOutputPath(job, outputPath);
+ 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
--- End diff --
Make sure this should not happen if that's the case we may lose some data.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83792475
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java ---
@@ -0,0 +1,264 @@
+/*
+ * 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.UUID;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.phoenix.end2end.index.MutableIndexFailureIT.FailingRegionObserver;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.mapreduce.index.IndexTool;
+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_BATCH_SIZE_ATTRIB, "1000");
+ serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_INTERVAL_ATTRIB, "2000");
+ 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 + " 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());
+
+ 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.
+ assertTrue(actualExplainPlan.contains(
+ String.format("CLIENT PARALLEL 1-WAY FULL SCAN OVER %s",
+ SchemaUtil.getPhysicalHBaseTableName(fullTableName, isNamespaceEnabled, PTableType.TABLE))));
+
+ 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();
+
+ rs = stmt1.executeQuery("SELECT LPAD(UPPER(NAME),11,'x')||'_xyz',ID FROM " + fullTableName);
--- End diff --
This is not needed. We are not using the result for anything.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r90192647
--- Diff: phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java ---
@@ -167,50 +180,152 @@ private void printHelpAndExit(Options options, int exitCode) {
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) throws Exception {
+ if (indexTable == null) {
+ 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);
--- End diff --
@ankitsinghal The rebuilding also a kind of scan from other clients so HBase should cope up with the caches and GCs. But setting it to true makes the job run faster. We can check the performance difference and decide on this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by ankitsinghal <gi...@git.apache.org>.
Github user ankitsinghal commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r91454910
--- Diff: phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java ---
@@ -85,7 +102,7 @@
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)");
--- End diff --
This was taken care in the last commit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by ankitsinghal <gi...@git.apache.org>.
Github user ankitsinghal commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r91454865
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexMetadataIT.java ---
@@ -216,6 +219,15 @@ public void testIndexCreateDrop() throws Exception {
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"));
--- End diff --
This was taken care in the last commit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83832270
--- Diff: phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java ---
@@ -305,16 +305,27 @@
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() +
--- End diff --
Here ASYNC_REBUILD_TIMESTAMP is dynamic column. Are there any problems if we use dynamic columns for system tables? why can't you make normal column?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83795010
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexMetadataIT.java ---
@@ -216,6 +219,15 @@ public void testIndexCreateDrop() throws Exception {
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"));
--- End diff --
This is making active index to rebuild can we add an assertion what is time from which we rebuild?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix issue #210: PHOENIX-2890 Extend IndexTool to allow incremental index...
Posted by ankitsinghal <gi...@git.apache.org>.
Github user ankitsinghal commented on the issue:
https://github.com/apache/phoenix/pull/210
Thank you so much @chrajeshbabu for the review, Raised PHOENIX-3525 for the same.
@JamesRTaylor ,do you want to take a look or it's ok to commit now?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r90188400
--- Diff: phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java ---
@@ -305,16 +305,27 @@
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() +
--- End diff --
Ok. It's fine. Thanks Ankit
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83795678
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexMetadataIT.java ---
@@ -568,4 +580,52 @@ public void testAsyncCreatedDate() throws Exception {
assertTrue(d2.after(d1));
assertFalse(rs.next());
}
+
+ @Test
+ public void testAsyncRebuildTimestamp() throws Exception {
+ long l0 = 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)";
+ PreparedStatement stmt = conn.prepareStatement(ddl);
+ stmt.execute();
+ String indexName = "R_ASYNCIND_" + generateUniqueName();
+
+ ddl = "CREATE INDEX " + indexName + "1 ON " + testTable + " (v1) ";
+ stmt = conn.prepareStatement(ddl);
--- End diff --
You can use createStatement than prepareStatement here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by ankitsinghal <gi...@git.apache.org>.
Github user ankitsinghal commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r91454867
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java ---
@@ -385,6 +383,8 @@ private void updateTable(Connection conn, String tableName) throws SQLException
}
public static class FailingRegionObserver extends SimpleRegionObserver {
+ public static volatile boolean FAIL_WRITE = false;
--- End diff --
This was taken care in the last commit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83795282
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexMetadataIT.java ---
@@ -568,4 +580,52 @@ public void testAsyncCreatedDate() throws Exception {
assertTrue(d2.after(d1));
assertFalse(rs.next());
}
+
+ @Test
+ public void testAsyncRebuildTimestamp() throws Exception {
+ long l0 = System.currentTimeMillis();
--- End diff --
can you use meaning full names for variables l0 and l1.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83793592
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/OrderByIT.java ---
@@ -500,14 +500,12 @@ public void testOrderByWithExpression() throws Exception {
stmt.execute();
conn.commit();
- String query = "SELECT col1+col2, col4, a_string FROM " + tableName + " ORDER BY 1, 2";
--- End diff --
why this change needed?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r90192895
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java ---
@@ -217,39 +223,47 @@ public void testSecondaryIndex() throws Exception {
assertFalse(rs.next());
- conn.createStatement().execute(String.format("DROP INDEX %s ON %s", indxTable, fullTableName));
+ // conn.createStatement().execute(String.format("DROP INDEX %s ON %s", indxTable, fullTableName));
--- End diff --
We can remove this commented code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by ankitsinghal <gi...@git.apache.org>.
Github user ankitsinghal commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r88209639
--- Diff: phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildWithNamespaceEnabled.java ---
@@ -0,0 +1,71 @@
+/*
+ * 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.end2end.index.MutableIndexFailureIT.FailingRegionObserver;
+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_BATCH_SIZE_ATTRIB, "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 }
--- End diff --
As discussed with you, it's difficult to do .. As mini cluster can be shared across test and in parallel environment restarting a cluster may fail other tests.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by ankitsinghal <gi...@git.apache.org>.
Github user ankitsinghal commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r91459367
--- Diff: phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java ---
@@ -167,50 +180,152 @@ private void printHelpAndExit(Options options, int exitCode) {
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) throws Exception {
+ if (indexTable == null) {
+ 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);
--- End diff --
Yes @chrajeshbabu , I did a below test and found the performance of both the scans(with or without cache) are same, file descriptor count during the test is almost same.
// with cache on
- create a new table
- Insert 500k records
- Flush the data
- scan with cache on
// with cache off
- create a new table again
- Insert 500k records
- Flush the data
- scan with cache off
Do you want me to do some other tests to verify the GC impact and all? or is it fine for now? as the cache is also kept off in org.apache.hadoop.hbase.mapreduce.Export by default.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] phoenix pull request #210: PHOENIX-2890 Extend IndexTool to allow incrementa...
Posted by chrajeshbabu <gi...@git.apache.org>.
Github user chrajeshbabu commented on a diff in the pull request:
https://github.com/apache/phoenix/pull/210#discussion_r83827393
--- Diff: phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java ---
@@ -167,50 +180,152 @@ private void printHelpAndExit(Options options, int exitCode) {
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) throws Exception {
+ if (indexTable == null) {
+ 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);
--- End diff --
Why not cache blocks? For every next call to server we need to read data from disk in this case. But mostly recent data reside all together in blocks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---