This is an automated email from the ASF dual-hosted git repository.
skadam pushed a commit to branch 4.14-HBase-1.3
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/4.14-HBase-1.3 by this push:
new 1e534c2 PHOENIX-5415: NPE in getting conf from addHbaseResources in
IndexUpgradeTool (#551)
1e534c2 is described below
commit 1e534c2c2f9df6b8da04ef305832fca58502389c
Author: Swaroopa Kadam <swaroopa.kada...@gmail.com>
AuthorDate: Sun Jul 28 13:59:51 2019 -0700
PHOENIX-5415: NPE in getting conf from addHbaseResources in
IndexUpgradeTool (#551)
---
.../phoenix/mapreduce/index/IndexUpgradeTool.java | 148 +++++++++++++--------
1 file changed, 93 insertions(+), 55 deletions(-)
diff --git
a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexUpgradeTool.java
b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexUpgradeTool.java
index 7855c40..10d9caf 100644
---
a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexUpgradeTool.java
+++
b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexUpgradeTool.java
@@ -18,6 +18,7 @@
package org.apache.phoenix.mapreduce.index;
import com.google.common.annotations.VisibleForTesting;
+import com.google.gson.Gson;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.HelpFormatter;
@@ -32,6 +33,8 @@ import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
import org.apache.phoenix.hbase.index.IndexRegionObserver;
import org.apache.phoenix.hbase.index.Indexer;
import org.apache.phoenix.hbase.index.covered.NonTxIndexBuilder;
@@ -55,6 +58,7 @@ import java.util.HashSet;
import java.util.logging.Logger;
import org.apache.hadoop.conf.Configuration;
import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.StringUtil;
import java.io.IOException;
import java.nio.file.Files;
@@ -70,7 +74,7 @@ import java.util.logging.SimpleFormatter;
import static org.apache.phoenix.query.QueryServicesOptions
.GLOBAL_INDEX_CHECKER_ENABLED_MAP_EXPIRATION_MIN;
-public class IndexUpgradeTool extends Configured {
+public class IndexUpgradeTool extends Configured implements Tool {
private static final Logger LOGGER =
Logger.getLogger(IndexUpgradeTool.class.getName());
@@ -145,21 +149,6 @@ public class IndexUpgradeTool extends Configured {
return operation;
}
- public static void main (String[] args) {
- CommandLine cmdLine = null;
-
- IndexUpgradeTool iut = new IndexUpgradeTool();
- try {
- cmdLine = iut.parseOptions(args);
- LOGGER.info("Index Upgrade tool initiated: "+ StringUtils.join(
args, ","));
- } catch (IllegalStateException e) {
- iut.printHelpAndExit(e.getMessage(), iut.getOptions());
- }
- iut.initializeTool(cmdLine);
- iut.prepareToolSetup();
- iut.executeTool();
- }
-
public IndexUpgradeTool(String mode, String tables, String inputFile,
String outputFile, boolean dryRun, IndexTool indexTool) {
this.operation = mode;
@@ -172,6 +161,21 @@ public class IndexUpgradeTool extends Configured {
public IndexUpgradeTool () { }
+ @Override
+ public int run(String[] args) throws Exception {
+ CommandLine cmdLine = null;
+ try {
+ cmdLine = parseOptions(args);
+ LOGGER.info("Index Upgrade tool initiated: " + StringUtils.join(
args, ","));
+ } catch (IllegalStateException e) {
+ printHelpAndExit(e.getMessage(), getOptions());
+ }
+ initializeTool(cmdLine);
+ prepareToolSetup();
+ executeTool();
+ return 0;
+ }
+
/**
* Parses the commandline arguments, throws IllegalStateException if
mandatory arguments are
* missing.
@@ -324,12 +328,12 @@ public class IndexUpgradeTool extends Configured {
try (Admin admin = queryServices.getAdmin()) {
PTable dataTable = PhoenixRuntime.getTableNoCache(conn,
dataTableFullName);
- LOGGER.fine("Executing " + operation + " for " +
dataTableFullName);
+ LOGGER.info("Executing " + operation + " for " +
dataTableFullName);
boolean mutable = !(dataTable.isImmutableRows());
if (!mutable) {
- LOGGER.fine("Data table is immutable, waiting for "
- + GLOBAL_INDEX_CHECKER_ENABLED_MAP_EXPIRATION_MIN
+ 1
+ LOGGER.info("Data table is immutable, waiting for "
+ + (GLOBAL_INDEX_CHECKER_ENABLED_MAP_EXPIRATION_MIN
+ 1)
+ " minutes for client cache to expire");
if (!test) {
Thread.sleep(
@@ -340,8 +344,10 @@ public class IndexUpgradeTool extends Configured {
modifyTable(admin, dataTableFullName, indexes);
enableTable(admin, dataTableFullName, indexes);
rebuildIndexes(conn, conf, dataTableFullName);
+ LOGGER.info("Completed " + operation + " for " +
dataTableFullName);
} catch (IOException | SQLException | InterruptedException e) {
- LOGGER.severe("Something went wrong while executing " +
operation + " steps " + e);
+ LOGGER.severe("Something went wrong while executing " +
operation
+ + " for " + dataTableFullName + " steps " + e);
return -1;
}
}
@@ -356,7 +362,7 @@ public class IndexUpgradeTool extends Configured {
}
LOGGER.info("Disabled data table " + dataTable);
} else {
- LOGGER.info( "Data table " + dataTable +" is already disabled");
+ LOGGER.info( "Data table " + dataTable + " is already disabled");
}
for (String indexName : indexes) {
if (admin.isTableEnabled(TableName.valueOf(indexName))) {
@@ -365,7 +371,7 @@ public class IndexUpgradeTool extends Configured {
}
LOGGER.info("Disabled index table " + indexName);
} else {
- LOGGER.info( "Index table " + indexName +" is already
disabled");
+ LOGGER.info( "Index table " + indexName + " is already
disabled");
}
}
}
@@ -389,7 +395,7 @@ public class IndexUpgradeTool extends Configured {
}
LOGGER.info("Enabled data table " + dataTable);
} else {
- LOGGER.info( "Data table " + dataTable +" is already enabled");
+ LOGGER.info( "Data table " + dataTable + " is already enabled");
}
for (String indexName : indexes) {
if(!admin.isTableEnabled(TableName.valueOf(indexName))) {
@@ -398,7 +404,7 @@ public class IndexUpgradeTool extends Configured {
}
LOGGER.info("Enabled index table " + indexName);
} else {
- LOGGER.info( "Index table " + indexName +" is already
enabled");
+ LOGGER.info( "Index table " + indexName + " is already
enabled");
}
}
}
@@ -436,9 +442,9 @@ public class IndexUpgradeTool extends Configured {
tableDesc.addCoprocessor(coprocName,
null,
QueryServicesOptions.DEFAULT_COPROCESSOR_PRIORITY, prop);
}
- LOGGER.info("Loaded "+coprocName+" coprocessor on table " +
tableName);
+ LOGGER.info("Loaded " + coprocName+ " coprocessor on table " +
tableName);
} else {
- LOGGER.info(coprocName+" coprocessor on table " + tableName + "is
already loaded");
+ LOGGER.info(coprocName + " coprocessor on table " + tableName + "
is already loaded");
}
}
@@ -450,7 +456,8 @@ public class IndexUpgradeTool extends Configured {
}
LOGGER.info("Unloaded "+ coprocName +"coprocessor on table " +
tableName);
} else {
- LOGGER.info(coprocName+" coprocessor on table " + tableName + " is
already unloaded");
+ LOGGER.info(coprocName + " coprocessor on table " + tableName
+ + " is already unloaded");
}
}
@@ -478,7 +485,7 @@ public class IndexUpgradeTool extends Configured {
String baseTable = indexInfo.getBaseTable();
String schema = indexInfo.getSchemaName();
String outFile = "/tmp/index_rebuild_" +schema+"_"+ indexName +
- (GLOBAL_INDEX_ID.equals(tenantId)?"":"_"+tenantId)
+ (GLOBAL_INDEX_ID.equals(tenantId) ? "" : "_" + tenantId)
+"_"+ UUID.randomUUID().toString();
String[] args = getIndexToolArgValues(schema, baseTable,
indexName, outFile, tenantId);
@@ -533,7 +540,7 @@ public class IndexUpgradeTool extends Configured {
//for upgrade or rollback
tablesAndIndexes.put(physicalTableName, physicalIndexes);
} else {
- LOGGER.info("Skipping Table " + tableName + " because it
is "+
+ LOGGER.info("Skipping Table " + tableName + " because it
is " +
(dataTable.isTransactional() ? "transactional" :
"not a data table"));
}
}
@@ -549,6 +556,7 @@ public class IndexUpgradeTool extends Configured {
private void prepareToRebuildIndexes(Connection conn, String
dataTableFullName) {
try {
+ Gson gson = new Gson();
HashMap<String, IndexInfo> rebuildIndexes = new HashMap<>();
HashSet<String> physicalIndexes =
tablesAndIndexes.get(dataTableFullName);
@@ -556,7 +564,6 @@ public class IndexUpgradeTool extends Configured {
String viewIndexPhysicalName = MetaDataUtil
.getViewIndexPhysicalName(dataTableFullName);
boolean hasViewIndex =
physicalIndexes.contains(viewIndexPhysicalName);
-
String schemaName =
SchemaUtil.getSchemaNameFromFullName(dataTableFullName);
String tableName =
SchemaUtil.getTableNameFromFullName(dataTableFullName);
@@ -572,40 +579,66 @@ public class IndexUpgradeTool extends Configured {
}
if (hasViewIndex) {
- ResultSet
- rs =
- conn.createStatement().executeQuery(
- "SELECT DISTINCT TABLE_NAME, TENANT_ID FROM "
- + "SYSTEM.CATALOG WHERE COLUMN_FAMILY
= \'"
- + viewIndexPhysicalName
- + "\' AND TABLE_TYPE = \'i\' AND " +
"LINK_TYPE = "
- +
PTable.LinkType.PHYSICAL_TABLE.getSerializedValue());
+
+ String viewSql = "SELECT DISTINCT TABLE_NAME, TENANT_ID FROM "
+ + "SYSTEM.CATALOG "
+ + "WHERE COLUMN_FAMILY = \'" + dataTableFullName + "\'
"
+ + (!StringUtil.EMPTY_STRING.equals(schemaName) ? "AND
TABLE_SCHEM = \'"
+ + schemaName + "\' " : "")
+ + "AND LINK_TYPE = "
+ + PTable.LinkType.PHYSICAL_TABLE.getSerializedValue();
+
+ ResultSet rs = conn.createStatement().executeQuery(viewSql);
+
while (rs.next()) {
- String viewIndexName = rs.getString(1);
+ String viewName = rs.getString(1);
String tenantId = rs.getString(2);
- ResultSet
- innerRS =
- conn.createStatement().executeQuery(
- "SELECT DISTINCT TABLE_NAME FROM "
- + "SYSTEM.CATALOG WHERE
COLUMN_FAMILY = \'"
- + viewIndexName
- + "\' AND TABLE_TYPE = \'i\' AND "
+ "LINK_TYPE = "
- +
PTable.LinkType.INDEX_TABLE.getSerializedValue());
- innerRS.next();
- String viewName = innerRS.getString(1);
- IndexInfo indexInfo = new IndexInfo(schemaName, viewName,
tenantId == null ? GLOBAL_INDEX_ID: tenantId,
- viewIndexName);
- rebuildIndexes.put(viewIndexName, indexInfo);
+
+ ArrayList<String> viewIndexes = findViewIndexes(conn,
schemaName, viewName,
+ tenantId);
+ for (String viewIndex : viewIndexes) {
+ IndexInfo indexInfo = new IndexInfo(schemaName,
viewName,
+ tenantId == null ? GLOBAL_INDEX_ID :
tenantId, viewIndex);
+ rebuildIndexes.put(viewIndex, indexInfo);
+ }
}
}
- //for rebuilding indexes in case of upgrade.
- rebuildMap.put(dataTableFullName, rebuildIndexes);
+ //for rebuilding indexes in case of upgrade and if there are
indexes on the table/view.
+ if (!rebuildIndexes.isEmpty()) {
+ rebuildMap.put(dataTableFullName, rebuildIndexes);
+ String json = gson.toJson(rebuildMap);
+ LOGGER.info("Index rebuild map " + json);
+ } else {
+ LOGGER.info("No indexes to rebuild for table " +
dataTableFullName);
+ }
+
} catch (SQLException e) {
- LOGGER.severe("Failed to prepare the map for index rebuilds "+e);
+ LOGGER.severe("Failed to prepare the map for index rebuilds " + e);
throw new RuntimeException("Failed to prepare the map for index
rebuilds");
}
}
+ private ArrayList<String> findViewIndexes(Connection conn, String
schemaName, String viewName,
+ String tenantId) throws SQLException {
+
+ String viewIndexesSql = "SELECT DISTINCT COLUMN_FAMILY FROM "
+ + "SYSTEM.CATALOG "
+ + "WHERE TABLE_NAME = \'" + viewName + "\'"
+ + (!StringUtil.EMPTY_STRING.equals(schemaName) ? "AND
TABLE_SCHEM = \'"
+ + schemaName + "\' " : "")
+ + "AND LINK_TYPE = " +
PTable.LinkType.INDEX_TABLE.getSerializedValue()
+ + (tenantId != null ? " AND TENANT_ID = \'" + tenantId + "\'"
: "");
+ ArrayList<String> viewIndexes = new ArrayList<>();
+ ResultSet
+ rs =
+ conn.createStatement().executeQuery(viewIndexesSql);
+ while(rs.next()) {
+ String viewIndexName = rs.getString(1);
+ viewIndexes.add(viewIndexName);
+ }
+ return viewIndexes;
+ }
+
private class IndexInfo {
final private String schemaName;
final private String baseTable;
@@ -635,4 +668,9 @@ public class IndexUpgradeTool extends Configured {
return indexName;
}
}
+
+ public static void main (String[] args) throws Exception {
+ int result = ToolRunner.run(new IndexUpgradeTool(), args);
+ System.exit(result);
+ }
}
\ No newline at end of file
