You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by sa...@apache.org on 2016/09/20 01:50:32 UTC
[1/2] phoenix git commit: PHOENIX-3174 Make minor upgrade a manual
step
Repository: phoenix
Updated Branches:
refs/heads/4.x-HBase-0.98 97a625030 -> 887b57cf9
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 8395c6d..0d398cf 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -64,6 +64,7 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.concurrent.GuardedBy;
@@ -132,6 +133,8 @@ import org.apache.phoenix.coprocessor.generated.MetaDataProtos.UpdateIndexStateR
import org.apache.phoenix.exception.PhoenixIOException;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.exception.UpgradeInProgressException;
+import org.apache.phoenix.exception.UpgradeNotRequiredException;
import org.apache.phoenix.execute.MutationState;
import org.apache.phoenix.hbase.index.IndexRegionSplitPolicy;
import org.apache.phoenix.hbase.index.Indexer;
@@ -267,6 +270,8 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
private final List<LinkedBlockingQueue<WeakReference<PhoenixConnection>>> connectionQueues;
private ScheduledExecutorService renewLeaseExecutor;
private final boolean renewLeaseEnabled;
+ private final boolean isAutoUpgradeEnabled;
+ private final AtomicBoolean upgradeRequired = new AtomicBoolean(false);
private static interface FeatureSupported {
boolean isSupported(ConnectionQueryServices services);
@@ -343,6 +348,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
connectionQueues = ImmutableList.copyOf(list);
// A little bit of a smell to leak `this` here, but should not be a problem
this.tableStatsCache = new TableStatsCache(this, config);
+ this.isAutoUpgradeEnabled = config.getBoolean(AUTO_UPGRADE_ENABLED, QueryServicesOptions.DEFAULT_AUTO_UPGRADE_ENABLED);
}
@Override
@@ -2307,29 +2313,16 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
return null;
}
checkClosed();
- PhoenixConnection metaConnection = null;
- boolean success = false;
- String snapshotName = null;
- String sysCatalogTableName = null;
try {
openConnection();
- String noUpgradeProp = props.getProperty(PhoenixRuntime.NO_UPGRADE_ATTRIB);
- boolean upgradeSystemTables = !Boolean.TRUE.equals(Boolean.valueOf(noUpgradeProp));
- Properties scnProps = PropertiesUtil.deepCopy(props);
- scnProps.setProperty(
- PhoenixRuntime.CURRENT_SCN_ATTRIB,
- Long.toString(MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP));
- scnProps.remove(PhoenixRuntime.TENANT_ID_ATTRIB);
- String globalUrl = JDBCUtil.removeProperty(url, PhoenixRuntime.TENANT_ID_ATTRIB);
- metaConnection = new PhoenixConnection(
- ConnectionQueryServicesImpl.this, globalUrl, scnProps, newEmptyMetaData());
+ boolean isDoNotUpgradePropSet = UpgradeUtil.isNoUpgradeSet(props);
try (HBaseAdmin admin = getAdmin()) {
boolean mappedSystemCatalogExists = admin
.tableExists(SchemaUtil.getPhysicalTableName(SYSTEM_CATALOG_NAME_BYTES, true));
if (SchemaUtil.isNamespaceMappingEnabled(PTableType.SYSTEM,
ConnectionQueryServicesImpl.this.getProps())) {
if (admin.tableExists(SYSTEM_CATALOG_NAME_BYTES)) {
- //check if the server is already updated and have namespace config properly set.
+ //check if the server is already updated and have namespace config properly set.
checkClientServerCompatibility(SYSTEM_CATALOG_NAME_BYTES);
}
ensureSystemTablesUpgraded(ConnectionQueryServicesImpl.this.getProps());
@@ -2342,232 +2335,32 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
+ IS_NAMESPACE_MAPPING_ENABLED + " enabled")
.build().buildException(); }
}
- try {
- metaConnection.createStatement().executeUpdate(QueryConstants.CREATE_TABLE_METADATA);
- } catch (NewerTableAlreadyExistsException ignore) {
- // Ignore, as this will happen if the SYSTEM.CATALOG already exists at this fixed timestamp.
- // A TableAlreadyExistsException is not thrown, since the table only exists *after* this fixed timestamp.
- } catch (TableAlreadyExistsException e) {
- if (upgradeSystemTables) {
- long currentServerSideTableTimeStamp = e.getTable().getTimeStamp();
- sysCatalogTableName = e.getTable().getPhysicalName().getString();
- if (currentServerSideTableTimeStamp < MIN_SYSTEM_TABLE_TIMESTAMP && acquireUpgradeMutex(currentServerSideTableTimeStamp, e.getTable().getPhysicalName().getBytes())) {
- snapshotName = getUpgradeSnapshotName(sysCatalogTableName, currentServerSideTableTimeStamp);
- createSnapshot(snapshotName, sysCatalogTableName);
- }
- String columnsToAdd = "";
- // This will occur if we have an older SYSTEM.CATALOG and we need to update it to include
- // any new columns we've added.
- if(currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0) {
- // We know that we always need to add the STORE_NULLS column for 4.3 release
- columnsToAdd = addColumn(columnsToAdd, PhoenixDatabaseMetaData.STORE_NULLS + " " + PBoolean.INSTANCE.getSqlTypeName());
- try (HBaseAdmin admin = getAdmin()) {
- HTableDescriptor[] localIndexTables = admin.listTables(MetaDataUtil.LOCAL_INDEX_TABLE_PREFIX+".*");
- for (HTableDescriptor table : localIndexTables) {
- if (table.getValue(MetaDataUtil.PARENT_TABLE_KEY) == null
- && table.getValue(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_NAME) != null) {
- table.setValue(MetaDataUtil.PARENT_TABLE_KEY,
- MetaDataUtil.getUserTableName(table
- .getNameAsString()));
- // Explicitly disable, modify and enable the table to ensure co-location of data
- // and index regions. If we just modify the table descriptor when online schema
- // change enabled may reopen the region in same region server instead of following data region.
- admin.disableTable(table.getTableName());
- admin.modifyTable(table.getTableName(), table);
- admin.enableTable(table.getTableName());
- }
- }
- }
- }
-
- // If the server side schema is before MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0 then
- // we need to add INDEX_TYPE and INDEX_DISABLE_TIMESTAMP columns too.
- // TODO: Once https://issues.apache.org/jira/browse/PHOENIX-1614 is fixed,
- // we should just have a ALTER TABLE ADD IF NOT EXISTS statement with all
- // the column names that have been added to SYSTEM.CATALOG since 4.0.
- if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0) {
- columnsToAdd = addColumn(columnsToAdd, PhoenixDatabaseMetaData.INDEX_TYPE + " " + PUnsignedTinyint.INSTANCE.getSqlTypeName()
- + ", " + PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP + " " + PLong.INSTANCE.getSqlTypeName());
- }
-
- // If we have some new columns from 4.1-4.3 to add, add them now.
- if (!columnsToAdd.isEmpty()) {
- // Ugh..need to assign to another local variable to keep eclipse happy.
- PhoenixConnection newMetaConnection = addColumnsIfNotExists(metaConnection,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0, columnsToAdd);
- metaConnection = newMetaConnection;
- }
-
- if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_5_0) {
- columnsToAdd = PhoenixDatabaseMetaData.BASE_COLUMN_COUNT + " "
- + PInteger.INSTANCE.getSqlTypeName();
- try {
- metaConnection = addColumn(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_5_0, columnsToAdd, false);
- upgradeTo4_5_0(metaConnection);
- } catch (ColumnAlreadyExistsException ignored) {
- /*
- * Upgrade to 4.5 is a slightly special case. We use the fact that the column
- * BASE_COLUMN_COUNT is already part of the meta-data schema as the signal that
- * the server side upgrade has finished or is in progress.
- */
- logger.debug("No need to run 4.5 upgrade");
- }
- Properties props = PropertiesUtil.deepCopy(metaConnection.getClientInfo());
- props.remove(PhoenixRuntime.CURRENT_SCN_ATTRIB);
- props.remove(PhoenixRuntime.TENANT_ID_ATTRIB);
- PhoenixConnection conn = new PhoenixConnection(ConnectionQueryServicesImpl.this, metaConnection.getURL(), props, metaConnection.getMetaDataCache());
- try {
- List<String> tablesNeedingUpgrade = UpgradeUtil.getPhysicalTablesWithDescRowKey(conn);
- if (!tablesNeedingUpgrade.isEmpty()) {
- logger.warn("The following tables require upgrade due to a bug causing the row key to be incorrect for descending columns and ascending BINARY columns (PHOENIX-2067 and PHOENIX-2120):\n" + Joiner.on(' ').join(tablesNeedingUpgrade) + "\nTo upgrade issue the \"bin/psql.py -u\" command.");
- }
- List<String> unsupportedTables = UpgradeUtil.getPhysicalTablesWithDescVarbinaryRowKey(conn);
- if (!unsupportedTables.isEmpty()) {
- logger.warn("The following tables use an unsupported VARBINARY DESC construct and need to be changed:\n" + Joiner.on(' ').join(unsupportedTables));
- }
- } catch (Exception ex) {
- logger.error("Unable to determine tables requiring upgrade due to PHOENIX-2067", ex);
- } finally {
- conn.close();
- }
- }
- // Add these columns one at a time, each with different timestamps so that if folks have
- // run the upgrade code already for a snapshot, we'll still enter this block (and do the
- // parts we haven't yet done).
- if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_6_0) {
- columnsToAdd = PhoenixDatabaseMetaData.IS_ROW_TIMESTAMP + " " + PBoolean.INSTANCE.getSqlTypeName();
- metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_6_0, columnsToAdd);
- }
- if(currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0) {
- // Drop old stats table so that new stats table is created
- metaConnection = dropStatsTable(metaConnection,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 - 4);
- metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 - 3,
- PhoenixDatabaseMetaData.TRANSACTIONAL + " " + PBoolean.INSTANCE.getSqlTypeName());
- metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 - 2,
- PhoenixDatabaseMetaData.UPDATE_CACHE_FREQUENCY + " " + PLong.INSTANCE.getSqlTypeName());
- metaConnection = setImmutableTableIndexesImmutable(metaConnection,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 - 1);
- metaConnection = updateSystemCatalogTimestamp(metaConnection,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0);
- ConnectionQueryServicesImpl.this.removeTable(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME, null, MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0);
- clearCache();
- }
-
- if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0) {
- metaConnection = addColumnsIfNotExists(metaConnection,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0 - 2,
- PhoenixDatabaseMetaData.IS_NAMESPACE_MAPPED + " "
- + PBoolean.INSTANCE.getSqlTypeName());
- metaConnection = addColumnsIfNotExists(metaConnection,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0 - 1,
- PhoenixDatabaseMetaData.AUTO_PARTITION_SEQ + " "
- + PVarchar.INSTANCE.getSqlTypeName());
- metaConnection = addColumnsIfNotExists(metaConnection,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0,
- PhoenixDatabaseMetaData.APPEND_ONLY_SCHEMA + " "
- + PBoolean.INSTANCE.getSqlTypeName());
- metaConnection = UpgradeUtil.disableViewIndexes(metaConnection);
- if(getProps().getBoolean(QueryServices.LOCAL_INDEX_CLIENT_UPGRADE_ATTRIB,
- QueryServicesOptions.DEFAULT_LOCAL_INDEX_CLIENT_UPGRADE)) {
- metaConnection = UpgradeUtil.upgradeLocalIndexes(metaConnection);
- }
- ConnectionQueryServicesImpl.this.removeTable(null,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME, null,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0);
- clearCache();
- }
-
- }
- }
-
- int nSaltBuckets = ConnectionQueryServicesImpl.this.props.getInt(QueryServices.SEQUENCE_SALT_BUCKETS_ATTRIB,
- QueryServicesOptions.DEFAULT_SEQUENCE_TABLE_SALT_BUCKETS);
- try {
- String createSequenceTable = Sequence.getCreateTableStatement(nSaltBuckets);
- metaConnection.createStatement().executeUpdate(createSequenceTable);
- nSequenceSaltBuckets = nSaltBuckets;
- } catch (NewerTableAlreadyExistsException e) {
- // Ignore, as this will happen if the SYSTEM.SEQUENCE already exists at this fixed timestamp.
- // A TableAlreadyExistsException is not thrown, since the table only exists *after* this fixed timestamp.
- nSequenceSaltBuckets = getSaltBuckets(e);
- } catch (TableAlreadyExistsException e) {
- if (upgradeSystemTables) {
- // This will occur if we have an older SYSTEM.SEQUENCE and we need to update it to include
- // any new columns we've added.
+ Properties scnProps = PropertiesUtil.deepCopy(props);
+ scnProps.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB,
+ Long.toString(MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP));
+ scnProps.remove(PhoenixRuntime.TENANT_ID_ATTRIB);
+ String globalUrl = JDBCUtil.removeProperty(url, PhoenixRuntime.TENANT_ID_ATTRIB);
+ try (PhoenixConnection metaConnection = new PhoenixConnection(ConnectionQueryServicesImpl.this, globalUrl,
+ scnProps, newEmptyMetaData())) {
+ try {
+ metaConnection.createStatement().executeUpdate(QueryConstants.CREATE_TABLE_METADATA);
+ } catch (NewerTableAlreadyExistsException ignore) {
+ // Ignore, as this will happen if the SYSTEM.CATALOG already exists at this fixed
+ // timestamp. A TableAlreadyExistsException is not thrown, since the table only exists
+ // *after* this fixed timestamp.
+ } catch (TableAlreadyExistsException e) {
long currentServerSideTableTimeStamp = e.getTable().getTimeStamp();
- if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0) {
- // If the table time stamp is before 4.1.0 then we need to add below columns
- // to the SYSTEM.SEQUENCE table.
- String columnsToAdd = PhoenixDatabaseMetaData.MIN_VALUE + " " + PLong.INSTANCE.getSqlTypeName()
- + ", " + PhoenixDatabaseMetaData.MAX_VALUE + " " + PLong.INSTANCE.getSqlTypeName()
- + ", " + PhoenixDatabaseMetaData.CYCLE_FLAG + " " + PBoolean.INSTANCE.getSqlTypeName()
- + ", " + PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG + " " + PBoolean.INSTANCE.getSqlTypeName();
- addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP, columnsToAdd);
- }
- // If the table timestamp is before 4.2.1 then run the upgrade script
- if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_2_1) {
- if (UpgradeUtil.upgradeSequenceTable(metaConnection, nSaltBuckets, e.getTable())) {
- metaConnection.removeTable(null,
- PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_SCHEMA,
- PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_TABLE,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP);
- clearTableFromCache(ByteUtil.EMPTY_BYTE_ARRAY,
- PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_SCHEMA_BYTES,
- PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_TABLE_BYTES,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP);
- clearTableRegionCache(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
- }
- nSequenceSaltBuckets = nSaltBuckets;
- } else {
- nSequenceSaltBuckets = getSaltBuckets(e);
+ if (currentServerSideTableTimeStamp < MIN_SYSTEM_TABLE_TIMESTAMP) {
+ ConnectionQueryServicesImpl.this.upgradeRequired.set(true);
}
}
- }
- try {
- metaConnection.createStatement().executeUpdate(
- QueryConstants.CREATE_STATS_TABLE_METADATA);
- } catch (NewerTableAlreadyExistsException ignore) {
- } catch(TableAlreadyExistsException e) {
- if (upgradeSystemTables) {
- long currentServerSideTableTimeStamp = e.getTable().getTimeStamp();
- if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0) {
- metaConnection = addColumnsIfNotExists(
- metaConnection,
- SYSTEM_STATS_NAME,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP,
- PhoenixDatabaseMetaData.GUIDE_POSTS_ROW_COUNT + " "
- + PLong.INSTANCE.getSqlTypeName());
- }
+ if (!ConnectionQueryServicesImpl.this.upgradeRequired.get()) {
+ createOtherSystemTables(metaConnection);
+ } else if (isAutoUpgradeEnabled && !isDoNotUpgradePropSet) {
+ upgradeSystemTables(url, props);
}
}
- try {
- metaConnection.createStatement().executeUpdate(
- QueryConstants.CREATE_FUNCTION_METADATA);
- } catch (NewerTableAlreadyExistsException e) {
- } catch (TableAlreadyExistsException e) {
- }
- if (SchemaUtil.isNamespaceMappingEnabled(PTableType.SYSTEM,
- ConnectionQueryServicesImpl.this.getProps())) {
- try {
- metaConnection.createStatement().executeUpdate("CREATE SCHEMA IF NOT EXISTS "
- + PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA);
- } catch (NewerSchemaAlreadyExistsException e) {}
- }
- success = true;
scheduleRenewLeaseTasks();
- } catch (UpgradeInProgressException e) {
- // don't set it as initializationException because otherwise client won't be able to retry
- throw e;
} catch (Exception e) {
if (e instanceof SQLException) {
initializationException = (SQLException)e;
@@ -2576,222 +2369,518 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
initializationException = new SQLException(e);
}
} finally {
- try {
- if (metaConnection != null) metaConnection.close();
- } catch (SQLException e) {
- if (initializationException != null) {
- initializationException.setNextException(e);
- } else {
- initializationException = e;
- }
- } finally {
- try {
- restoreFromSnapshot(sysCatalogTableName, snapshotName, success);
- } catch (SQLException e) {
- if (initializationException != null) {
- initializationException.setNextException(e);
- } else {
- initializationException = e;
- }
- }
- try {
- if (initializationException != null) {
- throw initializationException;
- }
- } finally {
- initialized = true;
- }
- }
+ initialized = true;
}
- }
+ }
return null;
}
-
- private void createSnapshot(String snapshotName, String tableName)
- throws SQLException {
- HBaseAdmin admin = null;
- SQLException sqlE = null;
- try {
- admin = getAdmin();
- admin.snapshot(snapshotName, tableName);
- logger.info("Successfully created snapshot " + snapshotName + " for "
- + tableName);
- } catch (Exception e) {
- sqlE = new SQLException(e);
- } finally {
- try {
- if (admin != null) {
- admin.close();
- }
- } catch (Exception e) {
- SQLException adminCloseEx = new SQLException(e);
- if (sqlE == null) {
- sqlE = adminCloseEx;
- } else {
- sqlE.setNextException(adminCloseEx);
- }
- } finally {
- if (sqlE != null) {
- throw sqlE;
+ });
+ } catch (Exception e) {
+ Throwables.propagateIfInstanceOf(e, SQLException.class);
+ Throwables.propagate(e);
+ }
+ }
+
+ private void createOtherSystemTables(PhoenixConnection metaConnection) throws SQLException {
+ try {
+ metaConnection.createStatement().execute(QueryConstants.CREATE_SEQUENCE_METADATA);
+ } catch (TableAlreadyExistsException ignore) {}
+ try {
+ metaConnection.createStatement().execute(QueryConstants.CREATE_STATS_TABLE_METADATA);
+ } catch (TableAlreadyExistsException ignore) {}
+ try {
+ metaConnection.createStatement().execute(QueryConstants.CREATE_FUNCTION_METADATA);
+ } catch (TableAlreadyExistsException ignore) {}
+ }
+
+ /**
+ * There is no other locking needed here since only one connection (on the same or different JVM) will be able to
+ * acquire the upgrade mutex via {@link #acquireUpgradeMutex(long, byte[])}.
+ */
+ @Override
+ public void upgradeSystemTables(final String url, final Properties props) throws SQLException {
+ PhoenixConnection metaConnection = null;
+ boolean success = false;
+ String snapshotName = null;
+ String sysCatalogTableName = null;
+ SQLException toThrow = null;
+ try {
+ if (!ConnectionQueryServicesImpl.this.upgradeRequired.get()) {
+ throw new UpgradeNotRequiredException();
+ }
+ Properties scnProps = PropertiesUtil.deepCopy(props);
+ scnProps.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB,
+ Long.toString(MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP));
+ scnProps.remove(PhoenixRuntime.TENANT_ID_ATTRIB);
+ String globalUrl = JDBCUtil.removeProperty(url, PhoenixRuntime.TENANT_ID_ATTRIB);
+ metaConnection = new PhoenixConnection(ConnectionQueryServicesImpl.this, globalUrl,
+ scnProps, newEmptyMetaData());
+ metaConnection.setRunningUpgrade(true);
+ try {
+ metaConnection.createStatement().executeUpdate(QueryConstants.CREATE_TABLE_METADATA);
+ } catch (NewerTableAlreadyExistsException ignore) {
+ // Ignore, as this will happen if the SYSTEM.CATALOG already exists at this fixed
+ // timestamp. A TableAlreadyExistsException is not thrown, since the table only exists
+ // *after* this fixed timestamp.
+ } catch (TableAlreadyExistsException e) {
+ long currentServerSideTableTimeStamp = e.getTable().getTimeStamp();
+ sysCatalogTableName = e.getTable().getPhysicalName().getString();
+ if (currentServerSideTableTimeStamp < MIN_SYSTEM_TABLE_TIMESTAMP
+ && acquireUpgradeMutex(currentServerSideTableTimeStamp, e.getTable()
+ .getPhysicalName().getBytes())) {
+ snapshotName = getUpgradeSnapshotName(sysCatalogTableName,
+ currentServerSideTableTimeStamp);
+ createSnapshot(snapshotName, sysCatalogTableName);
+ }
+ String columnsToAdd = "";
+ // This will occur if we have an older SYSTEM.CATALOG and we need to update it to
+ // include any new columns we've added.
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0) {
+ // We know that we always need to add the STORE_NULLS column for 4.3 release
+ columnsToAdd = addColumn(columnsToAdd, PhoenixDatabaseMetaData.STORE_NULLS
+ + " " + PBoolean.INSTANCE.getSqlTypeName());
+ try (HBaseAdmin admin = getAdmin()) {
+ HTableDescriptor[] localIndexTables = admin
+ .listTables(MetaDataUtil.LOCAL_INDEX_TABLE_PREFIX + ".*");
+ for (HTableDescriptor table : localIndexTables) {
+ if (table.getValue(MetaDataUtil.PARENT_TABLE_KEY) == null
+ && table.getValue(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_NAME) != null) {
+ table.setValue(MetaDataUtil.PARENT_TABLE_KEY,
+ MetaDataUtil.getUserTableName(table.getNameAsString()));
+ // Explicitly disable, modify and enable the table to ensure
+ // co-location of data and index regions. If we just modify the
+ // table descriptor when online schema change enabled may reopen
+ // the region in same region server instead of following data region.
+ admin.disableTable(table.getTableName());
+ admin.modifyTable(table.getTableName(), table);
+ admin.enableTable(table.getTableName());
}
}
}
}
- private void restoreFromSnapshot(String tableName, String snapshotName,
- boolean success) throws SQLException {
- boolean snapshotRestored = false;
- boolean tableDisabled = false;
- if (!success && snapshotName != null) {
- SQLException sqlE = null;
- HBaseAdmin admin = null;
- try {
- logger.warn("Starting restore of " + tableName + " using snapshot "
- + snapshotName + " because upgrade failed");
- admin = getAdmin();
- admin.disableTable(tableName);
- tableDisabled = true;
- admin.restoreSnapshot(snapshotName);
- snapshotRestored = true;
- logger.warn("Successfully restored " + tableName + " using snapshot "
- + snapshotName);
- } catch (Exception e) {
- sqlE = new SQLException(e);
- } finally {
- if (admin != null && tableDisabled) {
- try {
- admin.enableTable(tableName);
- if (snapshotRestored) {
- logger.warn("Successfully restored and enabled " + tableName + " using snapshot "
- + snapshotName);
- } else {
- logger.warn("Successfully enabled " + tableName + " after restoring using snapshot "
- + snapshotName + " failed. ");
- }
- } catch (Exception e1) {
- SQLException enableTableEx = new SQLException(e1);
- if (sqlE == null) {
- sqlE = enableTableEx;
- } else {
- sqlE.setNextException(enableTableEx);
- }
- logger.error("Failure in enabling "
- + tableName
- + (snapshotRestored ? " after successfully restoring using snapshot"
- + snapshotName
- : " after restoring using snapshot "
- + snapshotName + " failed. "));
- } finally {
- try {
- admin.close();
- } catch (Exception e2) {
- SQLException adminCloseEx = new SQLException(e2);
- if (sqlE == null) {
- sqlE = adminCloseEx;
- } else {
- sqlE.setNextException(adminCloseEx);
- }
- } finally {
- if (sqlE != null) {
- throw sqlE;
- }
- }
- }
- }
- }
- }
+ // If the server side schema is before MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0 then
+ // we need to add INDEX_TYPE and INDEX_DISABLE_TIMESTAMP columns too.
+ // TODO: Once https://issues.apache.org/jira/browse/PHOENIX-1614 is fixed,
+ // we should just have a ALTER TABLE ADD IF NOT EXISTS statement with all
+ // the column names that have been added to SYSTEM.CATALOG since 4.0.
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0) {
+ columnsToAdd = addColumn(columnsToAdd, PhoenixDatabaseMetaData.INDEX_TYPE + " "
+ + PUnsignedTinyint.INSTANCE.getSqlTypeName() + ", "
+ + PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP + " "
+ + PLong.INSTANCE.getSqlTypeName());
}
- private void ensureSystemTablesUpgraded(ReadOnlyProps props)
- throws SQLException, IOException, IllegalArgumentException, InterruptedException {
- if (!SchemaUtil.isNamespaceMappingEnabled(PTableType.SYSTEM, props)) { return; }
- HTableInterface metatable = null;
- try (HBaseAdmin admin = getAdmin()) {
- ensureNamespaceCreated(QueryConstants.SYSTEM_SCHEMA_NAME);
- List<HTableDescriptor> tables = Arrays
- .asList(admin.listTables(QueryConstants.SYSTEM_SCHEMA_NAME + "\\..*"));
- List<String> tableNames = getTableNames(tables);
- if (tableNames.size() == 0) { return; }
- if (tableNames.size() > 4) { throw new IllegalArgumentException(
- "Expected 4 system table only but found " + tableNames.size() + ":" + tableNames); }
- byte[] mappedSystemTable = SchemaUtil
- .getPhysicalName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, props).getName();
- metatable = getTable(mappedSystemTable);
- if (tableNames.contains(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME)) {
- if (!admin.tableExists(mappedSystemTable)) {
- UpgradeUtil.mapTableToNamespace(admin, metatable,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME, props, null, PTableType.SYSTEM,
- null);
- ConnectionQueryServicesImpl.this.removeTable(null,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME, null,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0);
- }
- tableNames.remove(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME);
- }
- for (String table : tableNames) {
- UpgradeUtil.mapTableToNamespace(admin, metatable, table, props, null, PTableType.SYSTEM,
- null);
- ConnectionQueryServicesImpl.this.removeTable(null, table, null,
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0);
+ // If we have some new columns from 4.1-4.3 to add, add them now.
+ if (!columnsToAdd.isEmpty()) {
+ // Ugh..need to assign to another local variable to keep eclipse happy.
+ PhoenixConnection newMetaConnection = addColumnsIfNotExists(metaConnection,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0, columnsToAdd);
+ metaConnection = newMetaConnection;
+ }
+
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_5_0) {
+ columnsToAdd = PhoenixDatabaseMetaData.BASE_COLUMN_COUNT + " "
+ + PInteger.INSTANCE.getSqlTypeName();
+ try {
+ metaConnection = addColumn(metaConnection,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_5_0, columnsToAdd,
+ false);
+ upgradeTo4_5_0(metaConnection);
+ } catch (ColumnAlreadyExistsException ignored) {
+ /*
+ * Upgrade to 4.5 is a slightly special case. We use the fact that the
+ * column BASE_COLUMN_COUNT is already part of the meta-data schema as the
+ * signal that the server side upgrade has finished or is in progress.
+ */
+ logger.debug("No need to run 4.5 upgrade");
+ }
+ Properties p = PropertiesUtil.deepCopy(metaConnection.getClientInfo());
+ p.remove(PhoenixRuntime.CURRENT_SCN_ATTRIB);
+ p.remove(PhoenixRuntime.TENANT_ID_ATTRIB);
+ PhoenixConnection conn = new PhoenixConnection(
+ ConnectionQueryServicesImpl.this, metaConnection.getURL(), p,
+ metaConnection.getMetaDataCache());
+ try {
+ List<String> tablesNeedingUpgrade = UpgradeUtil
+ .getPhysicalTablesWithDescRowKey(conn);
+ if (!tablesNeedingUpgrade.isEmpty()) {
+ logger.warn("The following tables require upgrade due to a bug causing the row key to be incorrect for descending columns and ascending BINARY columns (PHOENIX-2067 and PHOENIX-2120):\n"
+ + Joiner.on(' ').join(tablesNeedingUpgrade)
+ + "\nTo upgrade issue the \"bin/psql.py -u\" command.");
}
- if (!tableNames.isEmpty()) {
- clearCache();
+ List<String> unsupportedTables = UpgradeUtil
+ .getPhysicalTablesWithDescVarbinaryRowKey(conn);
+ if (!unsupportedTables.isEmpty()) {
+ logger.warn("The following tables use an unsupported VARBINARY DESC construct and need to be changed:\n"
+ + Joiner.on(' ').join(unsupportedTables));
}
+ } catch (Exception ex) {
+ logger.error(
+ "Unable to determine tables requiring upgrade due to PHOENIX-2067",
+ ex);
} finally {
- if (metatable != null) {
- metatable.close();
- }
+ conn.close();
+ }
+ }
+ // Add these columns one at a time, each with different timestamps so that if folks
+ // have
+ // run the upgrade code already for a snapshot, we'll still enter this block (and do
+ // the
+ // parts we haven't yet done).
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_6_0) {
+ columnsToAdd = PhoenixDatabaseMetaData.IS_ROW_TIMESTAMP + " "
+ + PBoolean.INSTANCE.getSqlTypeName();
+ metaConnection = addColumnsIfNotExists(metaConnection,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_6_0, columnsToAdd);
+ }
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0) {
+ // Drop old stats table so that new stats table is created
+ metaConnection = dropStatsTable(metaConnection,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 - 4);
+ metaConnection = addColumnsIfNotExists(
+ metaConnection,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 - 3,
+ PhoenixDatabaseMetaData.TRANSACTIONAL + " "
+ + PBoolean.INSTANCE.getSqlTypeName());
+ metaConnection = addColumnsIfNotExists(
+ metaConnection,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 - 2,
+ PhoenixDatabaseMetaData.UPDATE_CACHE_FREQUENCY + " "
+ + PLong.INSTANCE.getSqlTypeName());
+ metaConnection = setImmutableTableIndexesImmutable(metaConnection,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0 - 1);
+ metaConnection = updateSystemCatalogTimestamp(metaConnection,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0);
+ ConnectionQueryServicesImpl.this.removeTable(null,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME, null,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0);
+ clearCache();
+ }
+
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0) {
+ metaConnection = addColumnsIfNotExists(
+ metaConnection,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0 - 2,
+ PhoenixDatabaseMetaData.IS_NAMESPACE_MAPPED + " "
+ + PBoolean.INSTANCE.getSqlTypeName());
+ metaConnection = addColumnsIfNotExists(
+ metaConnection,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0 - 1,
+ PhoenixDatabaseMetaData.AUTO_PARTITION_SEQ + " "
+ + PVarchar.INSTANCE.getSqlTypeName());
+ metaConnection = addColumnsIfNotExists(
+ metaConnection,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0,
+ PhoenixDatabaseMetaData.APPEND_ONLY_SCHEMA + " "
+ + PBoolean.INSTANCE.getSqlTypeName());
+ metaConnection = UpgradeUtil.disableViewIndexes(metaConnection);
+ if (getProps().getBoolean(QueryServices.LOCAL_INDEX_CLIENT_UPGRADE_ATTRIB,
+ QueryServicesOptions.DEFAULT_LOCAL_INDEX_CLIENT_UPGRADE)) {
+ metaConnection = UpgradeUtil.upgradeLocalIndexes(metaConnection);
}
+ ConnectionQueryServicesImpl.this.removeTable(null,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME, null,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_8_0);
+ clearCache();
}
+ }
+
- /**
- * Acquire distributed mutex of sorts to make sure only one JVM is able to run the upgrade code by
- * making use of HBase's checkAndPut api.
- * <p>
- * This method was added as part of 4.8.1 release. For clients upgrading to 4.8.1, the old value in the
- * version cell will be null i.e. the {@value QueryConstants#UPGRADE_MUTEX} column will be non-existent. For client's
- * upgrading to a release newer than 4.8.1 the existing version cell will be non-null. The client which
- * wins the race will end up setting the version cell to the {@value MetaDataProtocol#MIN_SYSTEM_TABLE_TIMESTAMP}
- * for the release.
- * </p>
- *
- * @return true if client won the race, false otherwise
- * @throws IOException
- * @throws SQLException
- */
- private boolean acquireUpgradeMutex(long currentServerSideTableTimestamp, byte[] sysCatalogTableName) throws IOException,
- SQLException {
- Preconditions.checkArgument(currentServerSideTableTimestamp < MIN_SYSTEM_TABLE_TIMESTAMP);
- try (HTableInterface sysCatalogTable = getTable(sysCatalogTableName)) {
- byte[] row = SchemaUtil.getTableKey(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE);
- byte[] family = PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES;
- byte[] qualifier = QueryConstants.UPGRADE_MUTEX;
- byte[] oldValue = currentServerSideTableTimestamp < MIN_SYSTEM_TABLE_TIMESTAMP_4_8_1 ? null
- : PLong.INSTANCE.toBytes(currentServerSideTableTimestamp);
- byte[] newValue = PLong.INSTANCE.toBytes(MIN_SYSTEM_TABLE_TIMESTAMP);
- // Note that the timestamp for this put doesn't really matter since UPGRADE_MUTEX column isn't used
- // to calculate SYSTEM.CATALOG's server side timestamp.
- Put put = new Put(row);
- put.add(family, qualifier, newValue);
- boolean acquired = sysCatalogTable.checkAndPut(row, family, qualifier, oldValue, put);
- if (!acquired) { throw new UpgradeInProgressException(
- getVersion(currentServerSideTableTimestamp), getVersion(MIN_SYSTEM_TABLE_TIMESTAMP)); }
- return true;
+ int nSaltBuckets = ConnectionQueryServicesImpl.this.props.getInt(
+ QueryServices.SEQUENCE_SALT_BUCKETS_ATTRIB,
+ QueryServicesOptions.DEFAULT_SEQUENCE_TABLE_SALT_BUCKETS);
+ try {
+ String createSequenceTable = Sequence.getCreateTableStatement(nSaltBuckets);
+ metaConnection.createStatement().executeUpdate(createSequenceTable);
+ nSequenceSaltBuckets = nSaltBuckets;
+ } catch (NewerTableAlreadyExistsException e) {
+ // Ignore, as this will happen if the SYSTEM.SEQUENCE already exists at this fixed
+ // timestamp.
+ // A TableAlreadyExistsException is not thrown, since the table only exists *after* this
+ // fixed timestamp.
+ nSequenceSaltBuckets = getSaltBuckets(e);
+ } catch (TableAlreadyExistsException e) {
+ // This will occur if we have an older SYSTEM.SEQUENCE and we need to update it to
+ // include
+ // any new columns we've added.
+ long currentServerSideTableTimeStamp = e.getTable().getTimeStamp();
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0) {
+ // If the table time stamp is before 4.1.0 then we need to add below columns
+ // to the SYSTEM.SEQUENCE table.
+ String columnsToAdd = PhoenixDatabaseMetaData.MIN_VALUE + " "
+ + PLong.INSTANCE.getSqlTypeName() + ", "
+ + PhoenixDatabaseMetaData.MAX_VALUE + " "
+ + PLong.INSTANCE.getSqlTypeName() + ", "
+ + PhoenixDatabaseMetaData.CYCLE_FLAG + " "
+ + PBoolean.INSTANCE.getSqlTypeName() + ", "
+ + PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG + " "
+ + PBoolean.INSTANCE.getSqlTypeName();
+ addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP, columnsToAdd);
+ }
+ // If the table timestamp is before 4.2.1 then run the upgrade script
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_2_1) {
+ if (UpgradeUtil.upgradeSequenceTable(metaConnection, nSaltBuckets, e.getTable())) {
+ metaConnection.removeTable(null,
+ PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_SCHEMA,
+ PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_TABLE,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP);
+ clearTableFromCache(ByteUtil.EMPTY_BYTE_ARRAY,
+ PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_SCHEMA_BYTES,
+ PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_TABLE_BYTES,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP);
+ clearTableRegionCache(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_NAME_BYTES);
}
+ nSequenceSaltBuckets = nSaltBuckets;
+ } else {
+ nSequenceSaltBuckets = getSaltBuckets(e);
}
- });
+ }
+ try {
+ metaConnection.createStatement().executeUpdate(
+ QueryConstants.CREATE_STATS_TABLE_METADATA);
+ } catch (NewerTableAlreadyExistsException ignore) {} catch (TableAlreadyExistsException e) {
+ long currentServerSideTableTimeStamp = e.getTable().getTimeStamp();
+ if (currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0) {
+ metaConnection = addColumnsIfNotExists(
+ metaConnection,
+ SYSTEM_STATS_NAME,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP,
+ PhoenixDatabaseMetaData.GUIDE_POSTS_ROW_COUNT + " "
+ + PLong.INSTANCE.getSqlTypeName());
+ }
+ }
+ try {
+ metaConnection.createStatement().executeUpdate(QueryConstants.CREATE_FUNCTION_METADATA);
+ } catch (NewerTableAlreadyExistsException e) {} catch (TableAlreadyExistsException e) {}
+ if (SchemaUtil.isNamespaceMappingEnabled(PTableType.SYSTEM,
+ ConnectionQueryServicesImpl.this.getProps())) {
+ try {
+ metaConnection.createStatement().executeUpdate(
+ "CREATE SCHEMA IF NOT EXISTS "
+ + PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA);
+ } catch (NewerSchemaAlreadyExistsException e) {}
+ }
+ ConnectionQueryServicesImpl.this.upgradeRequired.set(false);
+ success = true;
+ } catch (UpgradeInProgressException | UpgradeNotRequiredException e) {
+ // don't set it as initializationException because otherwise client won't be able to retry
+ throw e;
} catch (Exception e) {
- Throwables.propagateIfInstanceOf(e, SQLException.class);
- Throwables.propagate(e);
+ if (e instanceof SQLException) {
+ toThrow = (SQLException)e;
+ } else {
+ // wrap every other exception into a SQLException
+ toThrow = new SQLException(e);
+ }
+ } finally {
+ try {
+ if (metaConnection != null) {
+ metaConnection.close();
+ }
+ } catch (SQLException e) {
+ if (toThrow != null) {
+ toThrow.setNextException(e);
+ } else {
+ toThrow = e;
+ }
+ } finally {
+ try {
+ restoreFromSnapshot(sysCatalogTableName, snapshotName, success);
+ } catch (SQLException e) {
+ if (toThrow != null) {
+ toThrow.setNextException(e);
+ } else {
+ toThrow = e;
+ }
+ }
+ if (toThrow != null) { throw toThrow; }
+ }
+
+ }
+ }
+
+ private void createSnapshot(String snapshotName, String tableName)
+ throws SQLException {
+ HBaseAdmin admin = null;
+ SQLException sqlE = null;
+ try {
+ admin = getAdmin();
+ admin.snapshot(snapshotName, tableName);
+ logger.info("Successfully created snapshot " + snapshotName + " for "
+ + tableName);
+ } catch (Exception e) {
+ sqlE = new SQLException(e);
+ } finally {
+ try {
+ if (admin != null) {
+ admin.close();
+ }
+ } catch (Exception e) {
+ SQLException adminCloseEx = new SQLException(e);
+ if (sqlE == null) {
+ sqlE = adminCloseEx;
+ } else {
+ sqlE.setNextException(adminCloseEx);
+ }
+ } finally {
+ if (sqlE != null) {
+ throw sqlE;
+ }
+ }
}
}
- private static class UpgradeInProgressException extends SQLException {
- public UpgradeInProgressException(String upgradeFrom, String upgradeTo) {
- super("Cluster is being concurrently upgraded from " + upgradeFrom + " to " + upgradeTo
- + ". Please retry establishing connection.", SQLExceptionCode.CONCURRENT_UPGRADE_IN_PROGRESS
- .getSQLState(), SQLExceptionCode.CONCURRENT_UPGRADE_IN_PROGRESS.getErrorCode());
+ private void restoreFromSnapshot(String tableName, String snapshotName,
+ boolean success) throws SQLException {
+ boolean snapshotRestored = false;
+ boolean tableDisabled = false;
+ if (!success && snapshotName != null) {
+ SQLException sqlE = null;
+ HBaseAdmin admin = null;
+ try {
+ logger.warn("Starting restore of " + tableName + " using snapshot "
+ + snapshotName + " because upgrade failed");
+ admin = getAdmin();
+ admin.disableTable(tableName);
+ tableDisabled = true;
+ admin.restoreSnapshot(snapshotName);
+ snapshotRestored = true;
+ logger.warn("Successfully restored " + tableName + " using snapshot "
+ + snapshotName);
+ } catch (Exception e) {
+ sqlE = new SQLException(e);
+ } finally {
+ if (admin != null && tableDisabled) {
+ try {
+ admin.enableTable(tableName);
+ if (snapshotRestored) {
+ logger.warn("Successfully restored and enabled " + tableName + " using snapshot "
+ + snapshotName);
+ } else {
+ logger.warn("Successfully enabled " + tableName + " after restoring using snapshot "
+ + snapshotName + " failed. ");
+ }
+ } catch (Exception e1) {
+ SQLException enableTableEx = new SQLException(e1);
+ if (sqlE == null) {
+ sqlE = enableTableEx;
+ } else {
+ sqlE.setNextException(enableTableEx);
+ }
+ logger.error("Failure in enabling "
+ + tableName
+ + (snapshotRestored ? " after successfully restoring using snapshot"
+ + snapshotName
+ : " after restoring using snapshot "
+ + snapshotName + " failed. "));
+ } finally {
+ try {
+ admin.close();
+ } catch (Exception e2) {
+ SQLException adminCloseEx = new SQLException(e2);
+ if (sqlE == null) {
+ sqlE = adminCloseEx;
+ } else {
+ sqlE.setNextException(adminCloseEx);
+ }
+ } finally {
+ if (sqlE != null) {
+ throw sqlE;
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ private void ensureSystemTablesUpgraded(ReadOnlyProps props)
+ throws SQLException, IOException, IllegalArgumentException, InterruptedException {
+ if (!SchemaUtil.isNamespaceMappingEnabled(PTableType.SYSTEM, props)) { return; }
+ HTableInterface metatable = null;
+ try (HBaseAdmin admin = getAdmin()) {
+ ensureNamespaceCreated(QueryConstants.SYSTEM_SCHEMA_NAME);
+ List<HTableDescriptor> tables = Arrays
+ .asList(admin.listTables(QueryConstants.SYSTEM_SCHEMA_NAME + "\\..*"));
+ List<String> tableNames = getTableNames(tables);
+ if (tableNames.size() == 0) { return; }
+ if (tableNames.size() > 4) { throw new IllegalArgumentException(
+ "Expected 4 system table only but found " + tableNames.size() + ":" + tableNames); }
+ byte[] mappedSystemTable = SchemaUtil
+ .getPhysicalName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, props).getName();
+ metatable = getTable(mappedSystemTable);
+ if (tableNames.contains(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME)) {
+ if (!admin.tableExists(mappedSystemTable)) {
+ UpgradeUtil.mapTableToNamespace(admin, metatable,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME, props, null, PTableType.SYSTEM,
+ null);
+ ConnectionQueryServicesImpl.this.removeTable(null,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME, null,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0);
+ }
+ tableNames.remove(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME);
+ }
+ for (String table : tableNames) {
+ UpgradeUtil.mapTableToNamespace(admin, metatable, table, props, null, PTableType.SYSTEM,
+ null);
+ ConnectionQueryServicesImpl.this.removeTable(null, table, null,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_1_0);
+ }
+ if (!tableNames.isEmpty()) {
+ clearCache();
+ }
+ } finally {
+ if (metatable != null) {
+ metatable.close();
+ }
+ }
+ }
+
+ /**
+ * Acquire distributed mutex of sorts to make sure only one JVM is able to run the upgrade code by
+ * making use of HBase's checkAndPut api.
+ * <p>
+ * This method was added as part of 4.8.1 release. For clients upgrading to 4.8.1, the old value in the
+ * cell will be null i.e. the {@value QueryConstants#UPGRADE_MUTEX} column will be non-existent. For client's
+ * upgrading to a release newer than 4.8.1 the existing cell value will be non-null. The client which
+ * wins the race will end up setting the cell value to the {@value MetaDataProtocol#MIN_SYSTEM_TABLE_TIMESTAMP}
+ * for the release.
+ * </p>
+ *
+ * @return true if client won the race, false otherwise
+ * @throws IOException
+ * @throws SQLException
+ */
+ @VisibleForTesting
+ public boolean acquireUpgradeMutex(long currentServerSideTableTimestamp, byte[] sysCatalogTableName) throws IOException,
+ SQLException {
+ Preconditions.checkArgument(currentServerSideTableTimestamp < MIN_SYSTEM_TABLE_TIMESTAMP);
+ try (HTableInterface sysCatalogTable = getTable(sysCatalogTableName)) {
+ byte[] row = SchemaUtil.getTableKey(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE);
+ byte[] family = PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES;
+ byte[] qualifier = QueryConstants.UPGRADE_MUTEX;
+ byte[] oldValue = currentServerSideTableTimestamp < MIN_SYSTEM_TABLE_TIMESTAMP_4_8_1 ? null
+ : PLong.INSTANCE.toBytes(currentServerSideTableTimestamp);
+ byte[] newValue = PLong.INSTANCE.toBytes(MIN_SYSTEM_TABLE_TIMESTAMP);
+ // Note that the timestamp for this put doesn't really matter since UPGRADE_MUTEX column isn't used
+ // to calculate SYSTEM.CATALOG's server side timestamp.
+ Put put = new Put(row);
+ put.add(family, qualifier, newValue);
+ boolean acquired = sysCatalogTable.checkAndPut(row, family, qualifier, oldValue, put);
+ if (!acquired) { throw new UpgradeInProgressException(
+ getVersion(currentServerSideTableTimestamp), getVersion(MIN_SYSTEM_TABLE_TIMESTAMP)); }
+ return true;
}
}
@@ -3878,4 +3967,9 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
public void invalidateStats(ImmutableBytesPtr tableName) {
this.tableStatsCache.invalidate(Objects.requireNonNull(tableName));
}
+
+ @Override
+ public boolean isUpgradeRequired() {
+ return upgradeRequired.get();
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
index 560b5d9..427f36f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
@@ -650,4 +650,12 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
public void invalidateStats(ImmutableBytesPtr tableName) {
this.tableStatsCache.invalidate(Objects.requireNonNull(tableName));
}
+
+ @Override
+ public boolean isUpgradeRequired() {
+ return false;
+ }
+
+ @Override
+ public void upgradeSystemTables(String url, Properties props) throws SQLException {}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
index 99ad59c..91ae4ed 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
@@ -345,4 +345,14 @@ public class DelegateConnectionQueryServices extends DelegateQueryServices imple
public void invalidateStats(ImmutableBytesPtr tableName) {
getDelegate().invalidateStats(tableName);
}
+
+ @Override
+ public boolean isUpgradeRequired() {
+ return getDelegate().isUpgradeRequired();
+ }
+
+ @Override
+ public void upgradeSystemTables(String url, Properties props) throws SQLException {
+ getDelegate().upgradeSystemTables(url, props);
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/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 e8de5a8..28844e1 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
@@ -223,6 +223,7 @@ public interface QueryServices extends SQLCloseable {
public static final String INDEX_ASYNC_BUILD_ENABLED = "phoenix.index.async.build.enabled";
public static final String CLIENT_CACHE_ENCODING = "phoenix.table.client.cache.encoding";
+ public static final String AUTO_UPGRADE_ENABLED = "phoenix.autoupgrade.enabled";
/**
* Get executor service used for parallel scans
*/
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
index 09f4769..0302286 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
@@ -20,6 +20,7 @@ package org.apache.phoenix.query;
import static org.apache.hadoop.hbase.HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD;
import static org.apache.phoenix.query.QueryServices.ALLOW_ONLINE_TABLE_SCHEMA_UPDATE;
import static org.apache.phoenix.query.QueryServices.ALLOW_VIEWS_ADD_NEW_CF_BASE_TABLE;
+import static org.apache.phoenix.query.QueryServices.AUTO_UPGRADE_ENABLED;
import static org.apache.phoenix.query.QueryServices.CALL_QUEUE_PRODUCER_ATTRIB_NAME;
import static org.apache.phoenix.query.QueryServices.CALL_QUEUE_ROUND_ROBIN_ATTRIB;
import static org.apache.phoenix.query.QueryServices.COLLECT_REQUEST_LEVEL_METRICS;
@@ -254,6 +255,7 @@ public class QueryServicesOptions {
public static final boolean DEFAULT_INDEX_ASYNC_BUILD_ENABLED = true;
public static final String DEFAULT_CLIENT_CACHE_ENCODING = PTableRefFactory.Encoding.OBJECT.toString();
+ public static final boolean DEFAULT_AUTO_UPGRADE_ENABLED = true;
@SuppressWarnings("serial")
public static final Set<String> DEFAULT_QUERY_SERVER_SKIP_WORDS = new HashSet<String>() {
@@ -331,7 +333,8 @@ public class QueryServicesOptions {
.setIfUnset(RENEW_LEASE_THREAD_POOL_SIZE, DEFAULT_RENEW_LEASE_THREAD_POOL_SIZE)
.setIfUnset(IS_NAMESPACE_MAPPING_ENABLED, DEFAULT_IS_NAMESPACE_MAPPING_ENABLED)
.setIfUnset(IS_SYSTEM_TABLE_MAPPED_TO_NAMESPACE, DEFAULT_IS_SYSTEM_TABLE_MAPPED_TO_NAMESPACE)
- .setIfUnset(LOCAL_INDEX_CLIENT_UPGRADE_ATTRIB, DEFAULT_LOCAL_INDEX_CLIENT_UPGRADE);
+ .setIfUnset(LOCAL_INDEX_CLIENT_UPGRADE_ATTRIB, DEFAULT_LOCAL_INDEX_CLIENT_UPGRADE)
+ .setIfUnset(AUTO_UPGRADE_ENABLED, DEFAULT_AUTO_UPGRADE_ENABLED);
// HBase sets this to 1, so we reset it to something more appropriate.
// Hopefully HBase will change this, because we can't know if a user set
// it to 1, so we'll change it.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 28ed11d..73f1501 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -3406,7 +3406,7 @@ public class MetaDataClient {
String indexTenantId = entry.getKey();
Properties props = new Properties(connection.getClientInfo());
props.setProperty(PhoenixRuntime.TENANT_ID_ATTRIB, indexTenantId);
- try (PhoenixConnection tenantConn = DriverManager.getConnection(connection.getURL(), props).unwrap(PhoenixConnection.class)) {
+ try (PhoenixConnection tenantConn = new PhoenixConnection(connection, connection.getQueryServices(), props)) {
PostDDLCompiler dropCompiler = new PostDDLCompiler(tenantConn);
tenantConn.getQueryServices().updateData(dropCompiler.compile(entry.getValue(), null, null, Collections.<PColumn>emptyList(), ts));
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/trace/PhoenixMetricsSink.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/trace/PhoenixMetricsSink.java b/phoenix-core/src/main/java/org/apache/phoenix/trace/PhoenixMetricsSink.java
index a8e80ab..fea6d61 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/trace/PhoenixMetricsSink.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/trace/PhoenixMetricsSink.java
@@ -38,16 +38,23 @@ import org.apache.commons.configuration.SubsetConfiguration;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.TableNotDisabledException;
import org.apache.hadoop.metrics2.AbstractMetric;
import org.apache.hadoop.metrics2.MetricsRecord;
import org.apache.hadoop.metrics2.MetricsSink;
import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.phoenix.compile.MutationPlan;
+import org.apache.phoenix.execute.MutationState;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.jdbc.PhoenixPreparedStatement;
import org.apache.phoenix.metrics.MetricInfo;
import org.apache.phoenix.metrics.Metrics;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.trace.util.Tracing;
+import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.QueryUtil;
import com.google.common.annotations.VisibleForTesting;
@@ -93,7 +100,7 @@ public class PhoenixMetricsSink implements MetricsSink {
private Connection conn;
private String table;
-
+
public PhoenixMetricsSink() {
LOG.info("Writing tracing metrics to phoenix table");
@@ -133,14 +140,25 @@ public class PhoenixMetricsSink implements MetricsSink {
}
}
}
-
+
private void initializeInternal(Connection conn, String tableName) throws SQLException {
this.conn = conn;
-
// ensure that the target table already exists
- createTable(conn, tableName);
+ if (!traceTableExists(conn, tableName)) {
+ createTable(conn, tableName);
+ }
+ this.table = tableName;
}
-
+
+ private boolean traceTableExists(Connection conn, String traceTableName) throws SQLException {
+ try {
+ PhoenixRuntime.getTable(conn, traceTableName);
+ return true;
+ } catch (TableNotFoundException e) {
+ return false;
+ }
+ }
+
/**
* Used for <b>TESTING ONLY</b>
* Initialize the connection and setup the table to use the
@@ -183,10 +201,8 @@ public class PhoenixMetricsSink implements MetricsSink {
// tables created as transactional tables, make these table non
// transactional
PhoenixDatabaseMetaData.TRANSACTIONAL + "=" + Boolean.FALSE;
-;
PreparedStatement stmt = conn.prepareStatement(ddl);
stmt.execute();
- this.table = table;
}
@Override
@@ -281,7 +297,12 @@ public class PhoenixMetricsSink implements MetricsSink {
for (String tag : variableValues) {
ps.setString(index++, tag);
}
- ps.execute();
+ // Not going through the standard route of using statement.execute() as that code path
+ // is blocked if the metadata hasn't been been upgraded to the new minor release.
+ MutationPlan plan = ps.unwrap(PhoenixPreparedStatement.class).compileMutation(stmt);
+ MutationState state = conn.unwrap(PhoenixConnection.class).getMutationState();
+ MutationState newState = plan.execute();
+ state.join(newState);
} catch (SQLException e) {
LOG.error("Could not write metric: \n" + record + " to prepared statement:\n" + stmt,
e);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
index 685879d..6027b95 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
@@ -174,7 +174,6 @@ public class PhoenixRuntime {
AUTO_COMMIT_ATTRIB,
CONSISTENCY_ATTRIB,
REQUEST_METRIC_ATTRIB,
- NO_UPGRADE_ATTRIB
};
/**
@@ -218,6 +217,7 @@ public class PhoenixRuntime {
props.setProperty(QueryServices.LOCAL_INDEX_CLIENT_UPGRADE_ATTRIB, "false");
}
conn = DriverManager.getConnection(jdbcUrl, props).unwrap(PhoenixConnection.class);
+ conn.setRunningUpgrade(true);
if (execCmd.isMapNamespace()) {
String srcTable = execCmd.getSrcTable();
System.out.println("Starting upgrading table:" + srcTable + "... please don't kill it in between!!");
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
index 9f9dfc6..74b3132 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
@@ -311,7 +311,7 @@ public final class QueryUtil {
}
/**
- * @return {@link PhoenixConnection} with NO_UPGRADE_ATTRIB set so that we don't initiate server upgrade
+ * @return {@link PhoenixConnection} with {@value UpgradeUtil#RUN_UPGRADE} set so that we don't initiate server upgrade
*/
public static Connection getConnectionOnServer(Configuration conf) throws ClassNotFoundException,
SQLException {
@@ -319,12 +319,12 @@ public final class QueryUtil {
}
/**
- * @return {@link PhoenixConnection} with NO_UPGRADE_ATTRIB set so that we don't initiate server upgrade
+ * @return {@link PhoenixConnection} with {@value UpgradeUtil#DO_NOT_UPGRADE} set so that we don't initiate metadata upgrade.
*/
public static Connection getConnectionOnServer(Properties props, Configuration conf)
throws ClassNotFoundException,
SQLException {
- props.setProperty(PhoenixRuntime.NO_UPGRADE_ATTRIB, Boolean.TRUE.toString());
+ UpgradeUtil.doNotUpgradeOnFirstConnection(props);
return getConnection(props, conf);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
index b205c4a..c175adc 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
@@ -132,7 +132,12 @@ public class UpgradeUtil {
private static final Logger logger = LoggerFactory.getLogger(UpgradeUtil.class);
private static final byte[] SEQ_PREFIX_BYTES = ByteUtil.concat(QueryConstants.SEPARATOR_BYTE_ARRAY, Bytes.toBytes("_SEQ_"));
public static final byte[] UPGRADE_TO_4_7_COLUMN_NAME = Bytes.toBytes("UPGRADE_TO_4_7");
-
+ /**
+ * Attribute for Phoenix's internal purposes only. When this attribute is set on a phoenix connection, then
+ * the upgrade code for upgrading the cluster to the new minor release is not triggered. Note that presence
+ * of this attribute overrides a true value for {@value QueryServices#AUTO_UPGRADE_ENABLED}.
+ */
+ private static final String DO_NOT_UPGRADE = "DoNotUpgrade";
public static String UPSERT_BASE_COLUMN_COUNT_IN_HEADER_ROW = "UPSERT "
+ "INTO SYSTEM.CATALOG "
+ "(TENANT_ID, TABLE_SCHEM, TABLE_NAME, COLUMN_NAME, COLUMN_FAMILY, BASE_COLUMN_COUNT) "
@@ -174,7 +179,7 @@ public class UpgradeUtil {
+ " FROM " + SYSTEM_CATALOG_SCHEMA + "." + SYSTEM_CATALOG_TABLE + " WHERE " + COLUMN_FAMILY + " = ? AND "
+ LINK_TYPE + " = " + LinkType.PHYSICAL_TABLE.getSerializedValue() + " AND ( " + TABLE_TYPE + "=" + "'"
+ PTableType.VIEW.getSerializedValue() + "' OR " + TABLE_TYPE + " IS NULL) ORDER BY "+TENANT_ID;
-
+
private UpgradeUtil() {
}
@@ -1498,7 +1503,7 @@ public class UpgradeUtil {
}
throw new SQLException(buf.toString());
}
- PhoenixConnection upgradeConn = new PhoenixConnection(conn, true);
+ PhoenixConnection upgradeConn = new PhoenixConnection(conn, true, true);
try {
upgradeConn.setAutoCommit(true);
for (PTable table : tablesNeedingUpgrading) {
@@ -1900,4 +1905,12 @@ public class UpgradeUtil {
String upgradingFrom = getVersion(currentSystemTableTimestamp);
return "SNAPSHOT_" + tableString + "_" + upgradingFrom + "_TO_" + CURRENT_CLIENT_VERSION + "_" + date;
}
+
+ public static boolean isNoUpgradeSet(Properties props) {
+ return Boolean.compare(true, Boolean.valueOf(props.getProperty(DO_NOT_UPGRADE))) == 0;
+ }
+
+ public static void doNotUpgradeOnFirstConnection(Properties props) {
+ props.setProperty(DO_NOT_UPGRADE, String.valueOf(true));
+ }
}
\ No newline at end of file
[2/2] phoenix git commit: PHOENIX-3174 Make minor upgrade a manual
step
Posted by sa...@apache.org.
PHOENIX-3174 Make minor upgrade a manual step
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/887b57cf
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/887b57cf
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/887b57cf
Branch: refs/heads/4.x-HBase-0.98
Commit: 887b57cf9a02f8fdfb21e655f10029531568bf54
Parents: 97a6250
Author: Samarth <sa...@salesforce.com>
Authored: Mon Sep 19 18:50:13 2016 -0700
Committer: Samarth <sa...@salesforce.com>
Committed: Mon Sep 19 18:50:13 2016 -0700
----------------------------------------------------------------------
.../org/apache/phoenix/end2end/UpgradeIT.java | 156 ++-
phoenix-core/src/main/antlr3/PhoenixSQL.g | 10 +-
.../coprocessor/MetaDataEndpointImpl.java | 16 +-
.../coprocessor/MetaDataRegionObserver.java | 50 +-
.../phoenix/exception/SQLExceptionCode.java | 2 +
.../exception/UpgradeInProgressException.java | 28 +
.../exception/UpgradeNotRequiredException.java | 27 +
.../exception/UpgradeRequiredException.java | 29 +
.../apache/phoenix/jdbc/PhoenixConnection.java | 27 +-
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 2 +-
.../apache/phoenix/jdbc/PhoenixStatement.java | 72 +-
.../index/automation/PhoenixMRJobSubmitter.java | 4 +
.../phoenix/parse/ExecuteUpgradeStatement.java | 34 +
.../apache/phoenix/parse/ParseNodeFactory.java | 4 +
.../phoenix/query/ConnectionQueryServices.java | 3 +
.../query/ConnectionQueryServicesImpl.java | 952 ++++++++++---------
.../query/ConnectionlessQueryServicesImpl.java | 8 +
.../query/DelegateConnectionQueryServices.java | 10 +
.../org/apache/phoenix/query/QueryServices.java | 1 +
.../phoenix/query/QueryServicesOptions.java | 5 +-
.../apache/phoenix/schema/MetaDataClient.java | 2 +-
.../phoenix/trace/PhoenixMetricsSink.java | 37 +-
.../org/apache/phoenix/util/PhoenixRuntime.java | 2 +-
.../java/org/apache/phoenix/util/QueryUtil.java | 6 +-
.../org/apache/phoenix/util/UpgradeUtil.java | 19 +-
25 files changed, 1029 insertions(+), 477 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java
index b47738d..bdd94a2 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java
@@ -24,7 +24,9 @@ import static org.apache.phoenix.query.QueryConstants.DIVERGED_VIEW_BASE_COLUMN_
import static org.apache.phoenix.util.UpgradeUtil.SELECT_BASE_COLUMN_COUNT_FROM_HEADER_ROW;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.io.IOException;
import java.sql.Connection;
@@ -33,6 +35,9 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.HBaseAdmin;
@@ -41,8 +46,14 @@ import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.RowMutations;
import org.apache.hadoop.hbase.snapshot.SnapshotCreationException;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.coprocessor.MetaDataProtocol;
+import org.apache.phoenix.exception.UpgradeInProgressException;
+import org.apache.phoenix.exception.UpgradeRequiredException;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.query.ConnectionQueryServices;
+import org.apache.phoenix.query.ConnectionQueryServicesImpl;
+import org.apache.phoenix.query.DelegateConnectionQueryServices;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.PName;
@@ -51,7 +62,6 @@ import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.UpgradeUtil;
-import org.junit.Ignore;
import org.junit.Test;
public class UpgradeIT extends BaseHBaseManagedTimeIT {
@@ -576,6 +586,150 @@ public class UpgradeIT extends BaseHBaseManagedTimeIT {
}
}
}
+
+ @Test
+ public void testUpgradeRequiredPreventsSQL() throws SQLException {
+ String tableName = generateRandomString();
+ try (Connection conn = getConnection(false, null)) {
+ conn.createStatement().execute(
+ "CREATE TABLE " + tableName
+ + " (PK1 VARCHAR NOT NULL, PK2 VARCHAR, KV1 VARCHAR, KV2 VARCHAR CONSTRAINT PK PRIMARY KEY(PK1, PK2))");
+ final ConnectionQueryServices delegate = conn.unwrap(PhoenixConnection.class).getQueryServices();
+ ConnectionQueryServices servicesWithUpgrade = new DelegateConnectionQueryServices(delegate) {
+ @Override
+ public boolean isUpgradeRequired() {
+ return true;
+ }
+ };
+ try (PhoenixConnection phxConn = new PhoenixConnection(servicesWithUpgrade,
+ conn.unwrap(PhoenixConnection.class), HConstants.LATEST_TIMESTAMP)) {
+ try {
+ phxConn.createStatement().execute(
+ "CREATE TABLE " + generateRandomString()
+ + " (k1 VARCHAR NOT NULL, k2 VARCHAR, CONSTRAINT PK PRIMARY KEY(K1,K2))");
+ fail("CREATE TABLE should have failed with UpgradeRequiredException");
+ } catch (UpgradeRequiredException expected) {
+
+ }
+ try {
+ phxConn.createStatement().execute("SELECT * FROM " + tableName);
+ fail("SELECT should have failed with UpgradeRequiredException");
+ } catch (UpgradeRequiredException expected) {
+
+ }
+ try {
+ phxConn.createStatement().execute("DELETE FROM " + tableName);
+ fail("DELETE should have failed with UpgradeRequiredException");
+ } catch (UpgradeRequiredException expected) {
+
+ }
+ try {
+ phxConn.createStatement().execute(
+ "CREATE INDEX " + tableName + "_IDX ON " + tableName + " (KV1) INCLUDE (KV2)" );
+ fail("CREATE INDEX should have failed with UpgradeRequiredException");
+ } catch (UpgradeRequiredException expected) {
+
+ }
+ try {
+ phxConn.createStatement().execute(
+ "UPSERT INTO " + tableName + " VALUES ('PK1', 'PK2', 'KV1', 'KV2')" );
+ fail("UPSERT VALUES should have failed with UpgradeRequiredException");
+ } catch (UpgradeRequiredException expected) {
+
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testUpgradingConnectionBypassesUpgradeRequiredCheck() throws Exception {
+ String tableName = generateRandomString();
+ try (Connection conn = getConnection(false, null)) {
+ conn.createStatement()
+ .execute(
+ "CREATE TABLE "
+ + tableName
+ + " (PK1 VARCHAR NOT NULL, PK2 VARCHAR, KV1 VARCHAR, KV2 VARCHAR CONSTRAINT PK PRIMARY KEY(PK1, PK2))");
+ final ConnectionQueryServices delegate = conn.unwrap(PhoenixConnection.class).getQueryServices();
+ ConnectionQueryServices servicesWithUpgrade = new DelegateConnectionQueryServices(delegate) {
+ @Override
+ public boolean isUpgradeRequired() {
+ return true;
+ }
+ };
+ try (PhoenixConnection phxConn = new PhoenixConnection(servicesWithUpgrade,
+ conn.unwrap(PhoenixConnection.class), HConstants.LATEST_TIMESTAMP)) {
+ // Because upgrade is required, this SQL should fail.
+ try {
+ phxConn.createStatement().executeQuery("SELECT * FROM " + tableName);
+ fail("SELECT should have failed with UpgradeRequiredException");
+ } catch (UpgradeRequiredException expected) {
+
+ }
+ // Marking connection as the one running upgrade should let SQL execute fine.
+ phxConn.setRunningUpgrade(true);
+ phxConn.createStatement().execute(
+ "UPSERT INTO " + tableName + " VALUES ('PK1', 'PK2', 'KV1', 'KV2')" );
+ phxConn.commit();
+ try (ResultSet rs = phxConn.createStatement().executeQuery("SELECT * FROM " + tableName)) {
+ assertTrue(rs.next());
+ assertFalse(rs.next());
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testConcurrentUpgradeThrowsUprgadeInProgressException() throws Exception {
+ final AtomicBoolean mutexStatus1 = new AtomicBoolean(false);
+ final AtomicBoolean mutexStatus2 = new AtomicBoolean(false);
+ final CountDownLatch latch = new CountDownLatch(2);
+ final AtomicInteger numExceptions = new AtomicInteger(0);
+ try (Connection conn = getConnection(false, null)) {
+ final ConnectionQueryServices services = conn.unwrap(PhoenixConnection.class).getQueryServices();
+ Thread t1 = new Thread(new AcquireMutexRunnable(mutexStatus1, services, latch, numExceptions));
+ t1.setDaemon(true);
+ Thread t2 = new Thread(new AcquireMutexRunnable(mutexStatus2, services, latch, numExceptions));
+ t2.setDaemon(true);;
+ t1.start();
+ t2.start();
+ latch.await();
+ assertTrue("One of the threads should have acquired the mutex", mutexStatus1.get() || mutexStatus2.get());
+ assertNotEquals("One and only one thread should have acquired the mutex ", mutexStatus1.get(), mutexStatus2.get());
+ assertEquals("One and only one thread should have caught UpgradeRequiredException ", 1, numExceptions.get());
+ }
+ }
+
+ private static class AcquireMutexRunnable implements Runnable {
+
+ private final AtomicBoolean acquireStatus;
+ private final ConnectionQueryServices services;
+ private final CountDownLatch latch;
+ private final AtomicInteger numExceptions;
+ public AcquireMutexRunnable(AtomicBoolean acquireStatus, ConnectionQueryServices services, CountDownLatch latch, AtomicInteger numExceptions) {
+ this.acquireStatus = acquireStatus;
+ this.services = services;
+ this.latch = latch;
+ this.numExceptions = numExceptions;
+ }
+ @Override
+ public void run() {
+ try {
+ ((ConnectionQueryServicesImpl)services).acquireUpgradeMutex(
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES);
+ acquireStatus.set(true);
+ } catch (UpgradeInProgressException e) {
+ numExceptions.incrementAndGet();
+ }
+ catch (IOException | SQLException ignore) {
+
+ } finally {
+ latch.countDown();
+ }
+ }
+
+ }
private Connection createTenantConnection(String tenantId) throws SQLException {
Properties props = new Properties();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/antlr3/PhoenixSQL.g
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/antlr3/PhoenixSQL.g b/phoenix-core/src/main/antlr3/PhoenixSQL.g
index 6716fdb..a6f9ef4 100644
--- a/phoenix-core/src/main/antlr3/PhoenixSQL.g
+++ b/phoenix-core/src/main/antlr3/PhoenixSQL.g
@@ -132,6 +132,8 @@ tokens
ROW = 'row';
ROWS = 'rows';
ONLY = 'only';
+ EXECUTE = 'execute';
+ UPGRADE = 'upgrade';
}
@@ -417,6 +419,7 @@ oneStatement returns [BindableStatement ret]
| s=drop_schema_node
| s=use_schema_node
| s=update_statistics_node
+ | s=execute_upgrade_node
| s=explain_node) { $ret = s; }
;
finally{ contextStack.pop(); }
@@ -568,7 +571,7 @@ trace_node returns [TraceStatement ret]
{ret = factory.trace(Tracing.isTraceOn(flag.getText()), s == null ? Tracing.isTraceOn(flag.getText()) ? 1.0 : 0.0 : (((BigDecimal)s.getValue())).doubleValue());}
;
-// Parse a trace statement.
+// Parse a create function statement.
create_function_node returns [CreateFunctionStatement ret]
: CREATE (OR replace=REPLACE)? (temp=TEMPORARY)? FUNCTION function=identifier
(LPAREN args=zero_or_more_data_types RPAREN)
@@ -611,6 +614,11 @@ update_statistics_node returns [UpdateStatisticsStatement ret]
{ret = factory.updateStatistics(factory.namedTable(null, t), s == null ? StatisticsCollectionScope.getDefault() : StatisticsCollectionScope.valueOf(SchemaUtil.normalizeIdentifier(s.getText())), p);}
;
+execute_upgrade_node returns [ExecuteUpgradeStatement ret]
+ : EXECUTE UPGRADE
+ {ret = factory.executeUpgrade();}
+ ;
+
prop_name returns [String ret]
: p=identifier {$ret = SchemaUtil.normalizeIdentifier(p); }
;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index 06c35d1..96d8575 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@ -79,7 +79,6 @@ import static org.apache.phoenix.util.SchemaUtil.getVarChars;
import java.io.IOException;
import java.sql.DriverManager;
-import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
@@ -127,6 +126,7 @@ import org.apache.phoenix.cache.GlobalCache;
import org.apache.phoenix.cache.GlobalCache.FunctionBytesPtr;
import org.apache.phoenix.compile.ColumnResolver;
import org.apache.phoenix.compile.FromCompiler;
+import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.compile.ScanRanges;
import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.compile.WhereCompiler;
@@ -162,8 +162,10 @@ import org.apache.phoenix.hbase.index.util.GenericKeyValueBuilder;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.iterate.ResultIterator;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.jdbc.PhoenixResultSet;
import org.apache.phoenix.jdbc.PhoenixStatement;
import org.apache.phoenix.metrics.Metrics;
import org.apache.phoenix.parse.LiteralParseNode;
@@ -212,7 +214,6 @@ import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.KeyValueUtil;
import org.apache.phoenix.util.MetaDataUtil;
-import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.SchemaUtil;
@@ -1415,12 +1416,17 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
if (parentTable!=null && parentTable.getAutoPartitionSeqName()!=null) {
long autoPartitionNum = 1;
final Properties props = new Properties();
- props.setProperty(PhoenixRuntime.NO_UPGRADE_ATTRIB, Boolean.TRUE.toString());
+ UpgradeUtil.doNotUpgradeOnFirstConnection(props);
try (PhoenixConnection connection = DriverManager.getConnection(MetaDataUtil.getJdbcUrl(env), props).unwrap(PhoenixConnection.class);
- Statement stmt = connection.createStatement()) {
+ Statement stmt = connection.createStatement()) {
String seqName = parentTable.getAutoPartitionSeqName();
+ // Not going through the standard route of using statement.execute() as that code path
+ // is blocked if the metadata hasn't been been upgraded to the new minor release.
String seqNextValueSql = String.format("SELECT NEXT VALUE FOR %s", seqName);
- ResultSet rs = stmt.executeQuery(seqNextValueSql);
+ PhoenixStatement ps = stmt.unwrap(PhoenixStatement.class);
+ QueryPlan plan = ps.compileQuery(seqNextValueSql);
+ ResultIterator resultIterator = plan.iterator();
+ PhoenixResultSet rs = ps.newResultSet(resultIterator, plan.getProjector(), plan.getContext());
rs.next();
autoPartitionNum = rs.getLong(1);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/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 aef5c69..2ab5df6 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
@@ -18,8 +18,10 @@
package org.apache.phoenix.coprocessor;
import java.io.IOException;
+import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -54,20 +56,23 @@ import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.phoenix.cache.GlobalCache;
import org.apache.phoenix.cache.ServerCacheClient;
+import org.apache.phoenix.coprocessor.MetaDataProtocol.MetaDataMutationResult;
+import org.apache.phoenix.coprocessor.MetaDataProtocol.MutationCode;
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.jdbc.PhoenixDriver;
-import org.apache.phoenix.parse.AlterIndexStatement;
-import org.apache.phoenix.parse.NamedTableNode;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.MetaDataClient;
import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.SortOrder;
+import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PLong;
import org.apache.phoenix.util.ByteUtil;
@@ -80,6 +85,7 @@ import org.apache.phoenix.util.UpgradeUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import com.google.protobuf.ServiceException;
/**
@@ -297,11 +303,7 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
// Allow index to begin incremental maintenance as index is back online and we
// cannot transition directly from DISABLED -> ACTIVE
if (Bytes.compareTo(PIndexState.DISABLE.getSerializedBytes(), indexState) == 0) {
- AlterIndexStatement statement = new AlterIndexStatement(
- NamedTableNode.create(indexPTable.getSchemaName().getString(), indexPTable.getTableName().getString()),
- dataPTable.getTableName().getString(),
- false, PIndexState.INACTIVE);
- client.alterIndex(statement);
+ updateIndexState(indexTableFullName, env, PIndexState.DISABLE, PIndexState.INACTIVE);
}
List<PTable> indexesToPartiallyRebuild = dataTableToIndexesMap.get(dataPTable);
if (indexesToPartiallyRebuild == null) {
@@ -393,11 +395,9 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
}
}
for (PTable indexPTable : indexesToPartiallyRebuild) {
- AlterIndexStatement statement = new AlterIndexStatement(
- NamedTableNode.create(indexPTable.getSchemaName().getString(), indexPTable
- .getTableName().getString()), dataPTable.getTableName().getString(),
- false, PIndexState.ACTIVE);
- client.alterIndex(statement);
+ String indexTableFullName = SchemaUtil.getTableName(indexPTable.getSchemaName()
+ .getString(), indexPTable.getTableName().getString());
+ updateIndexState(indexTableFullName, env, PIndexState.INACTIVE, PIndexState.ACTIVE);
}
} catch (Exception e) { // Log, but try next table's indexes
LOG.warn("Unable to rebuild " + dataPTable + " indexes " + indexesToPartiallyRebuild
@@ -426,4 +426,30 @@ public class MetaDataRegionObserver extends BaseRegionObserver {
}
}
}
+
+
+ private static void updateIndexState(String indexTableName, RegionCoprocessorEnvironment env, PIndexState oldState,
+ PIndexState newState) throws ServiceException, Throwable {
+ byte[] indexTableKey = SchemaUtil.getTableKeyFromFullName(indexTableName);
+ String schemaName = SchemaUtil.getSchemaNameFromFullName(indexTableName);
+ String indexName = SchemaUtil.getSchemaNameFromFullName(indexTableName);
+ // Mimic the Put that gets generated by the client on an update of the index state
+ Put put = new Put(indexTableKey);
+ put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES,
+ newState.getSerializedBytes());
+ if (newState == PIndexState.ACTIVE) {
+ put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES,
+ PLong.INSTANCE.toBytes(0));
+ }
+ final List<Mutation> tableMetadata = Collections.<Mutation> singletonList(put);
+ Connection conn = QueryUtil.getConnection(env.getConfiguration());
+ MetaDataMutationResult result = conn.unwrap(PhoenixConnection.class).getQueryServices()
+ .updateIndexState(tableMetadata, null);
+ MutationCode code = result.getMutationCode();
+ if (code == MutationCode.TABLE_NOT_FOUND) { throw new TableNotFoundException(schemaName, indexName); }
+ if (code == MutationCode.UNALLOWED_TABLE_MUTATION) { throw new SQLExceptionInfo.Builder(
+ SQLExceptionCode.INVALID_INDEX_STATE_TRANSITION)
+ .setMessage(" currentState=" + oldState + ". requestedState=" + newState).setSchemaName(schemaName)
+ .setTableName(indexName).build().buildException(); }
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
index 0ccecae..e6a26ae 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@ -371,6 +371,8 @@ public enum SQLExceptionCode {
INDEX_METADATA_NOT_FOUND(2008, "INT10", "Unable to find cached index metadata. "),
UNKNOWN_ERROR_CODE(2009, "INT11", "Unknown error code."),
CONCURRENT_UPGRADE_IN_PROGRESS(2010, "INT12", ""),
+ UPGRADE_REQUIRED(2011, "INT13", ""),
+ UPGRADE_NOT_REQUIRED(2012, "INT14", ""),
OPERATION_TIMED_OUT(6000, "TIM01", "Operation timed out.", new Factory() {
@Override
public SQLException newException(SQLExceptionInfo info) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeInProgressException.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeInProgressException.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeInProgressException.java
new file mode 100644
index 0000000..5b15216
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeInProgressException.java
@@ -0,0 +1,28 @@
+/*
+ * 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.exception;
+
+import java.sql.SQLException;
+
+public class UpgradeInProgressException extends SQLException {
+ public UpgradeInProgressException(String upgradeFrom, String upgradeTo) {
+ super("Cluster is being concurrently upgraded from " + upgradeFrom + " to " + upgradeTo
+ + ". Please retry establishing connection.", SQLExceptionCode.CONCURRENT_UPGRADE_IN_PROGRESS
+ .getSQLState(), SQLExceptionCode.CONCURRENT_UPGRADE_IN_PROGRESS.getErrorCode());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeNotRequiredException.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeNotRequiredException.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeNotRequiredException.java
new file mode 100644
index 0000000..0490319
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeNotRequiredException.java
@@ -0,0 +1,27 @@
+/*
+ * 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.exception;
+
+import java.sql.SQLException;
+
+public class UpgradeNotRequiredException extends SQLException {
+ public UpgradeNotRequiredException() {
+ super("Operation not allowed since cluster has already been upgraded. ", SQLExceptionCode.UPGRADE_NOT_REQUIRED
+ .getSQLState(), SQLExceptionCode.UPGRADE_NOT_REQUIRED.getErrorCode());
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeRequiredException.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeRequiredException.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeRequiredException.java
new file mode 100644
index 0000000..005a1bd
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/UpgradeRequiredException.java
@@ -0,0 +1,29 @@
+/*
+ * 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.exception;
+
+import java.sql.SQLException;
+
+public class UpgradeRequiredException extends SQLException {
+
+ public UpgradeRequiredException() {
+ super("Operation not allowed since cluster hasn't been upgraded. Call EXECUTE UPGRADE. ",
+ SQLExceptionCode.UPGRADE_REQUIRED.getSQLState(), SQLExceptionCode.UPGRADE_REQUIRED.getErrorCode());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
index 7ff4fdf..7cdab20 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
@@ -161,6 +161,7 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
private ParallelIteratorFactory parallelIteratorFactory;
private final LinkedBlockingQueue<WeakReference<TableResultIterator>> scannerQueue;
private TableResultIteratorFactory tableResultIteratorFactory;
+ private boolean isRunningUpgrade;
static {
Tracing.addTraceMetricsSource();
@@ -172,8 +173,8 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
return props;
}
- public PhoenixConnection(PhoenixConnection connection, boolean isDescRowKeyOrderUpgrade) throws SQLException {
- this(connection.getQueryServices(), connection.getURL(), connection.getClientInfo(), connection.metaData, connection.getMutationState(), isDescRowKeyOrderUpgrade);
+ public PhoenixConnection(PhoenixConnection connection, boolean isDescRowKeyOrderUpgrade, boolean isRunningUpgrade) throws SQLException {
+ this(connection.getQueryServices(), connection.getURL(), connection.getClientInfo(), connection.metaData, connection.getMutationState(), isDescRowKeyOrderUpgrade, isRunningUpgrade);
this.isAutoCommit = connection.isAutoCommit;
this.isAutoFlush = connection.isAutoFlush;
this.sampler = connection.sampler;
@@ -181,11 +182,11 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
}
public PhoenixConnection(PhoenixConnection connection) throws SQLException {
- this(connection, connection.isDescVarLengthRowKeyUpgrade());
+ this(connection, connection.isDescVarLengthRowKeyUpgrade(), connection.isRunningUpgrade());
}
public PhoenixConnection(PhoenixConnection connection, MutationState mutationState) throws SQLException {
- this(connection.getQueryServices(), connection.getURL(), connection.getClientInfo(), connection.getMetaDataCache(), mutationState, connection.isDescVarLengthRowKeyUpgrade());
+ this(connection.getQueryServices(), connection.getURL(), connection.getClientInfo(), connection.getMetaDataCache(), mutationState, connection.isDescVarLengthRowKeyUpgrade(), connection.isRunningUpgrade());
}
public PhoenixConnection(PhoenixConnection connection, long scn) throws SQLException {
@@ -193,7 +194,7 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
}
public PhoenixConnection(ConnectionQueryServices services, PhoenixConnection connection, long scn) throws SQLException {
- this(services, connection.getURL(), newPropsWithSCN(scn,connection.getClientInfo()), connection.metaData, connection.getMutationState(), connection.isDescVarLengthRowKeyUpgrade());
+ this(services, connection.getURL(), newPropsWithSCN(scn,connection.getClientInfo()), connection.metaData, connection.getMutationState(), connection.isDescVarLengthRowKeyUpgrade(), connection.isRunningUpgrade());
this.isAutoCommit = connection.isAutoCommit;
this.isAutoFlush = connection.isAutoFlush;
this.sampler = connection.sampler;
@@ -201,14 +202,14 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
}
public PhoenixConnection(ConnectionQueryServices services, String url, Properties info, PMetaData metaData) throws SQLException {
- this(services, url, info, metaData, null, false);
+ this(services, url, info, metaData, null, false, false);
}
public PhoenixConnection(PhoenixConnection connection, ConnectionQueryServices services, Properties info) throws SQLException {
- this(services, connection.url, info, connection.metaData, null, connection.isDescVarLengthRowKeyUpgrade());
+ this(services, connection.url, info, connection.metaData, null, connection.isDescVarLengthRowKeyUpgrade(), connection.isRunningUpgrade());
}
- public PhoenixConnection(ConnectionQueryServices services, String url, Properties info, PMetaData metaData, MutationState mutationState, boolean isDescVarLengthRowKeyUpgrade) throws SQLException {
+ public PhoenixConnection(ConnectionQueryServices services, String url, Properties info, PMetaData metaData, MutationState mutationState, boolean isDescVarLengthRowKeyUpgrade, boolean isRunningUpgrade) throws SQLException {
this.url = url;
this.isDescVarLengthRowKeyUpgrade = isDescVarLengthRowKeyUpgrade;
// Copy so client cannot change
@@ -294,6 +295,7 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
this.customTracingAnnotations = getImmutableCustomTracingAnnotations();
this.scannerQueue = new LinkedBlockingQueue<>();
this.tableResultIteratorFactory = new DefaultTableResultIteratorFactory();
+ this.isRunningUpgrade = isRunningUpgrade;
GLOBAL_OPEN_PHOENIX_CONNECTIONS.increment();
}
@@ -1055,4 +1057,13 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
getQueryServices().removeSchema(schema, schemaTimeStamp);
}
+
+ public boolean isRunningUpgrade() {
+ return isRunningUpgrade;
+ }
+
+ public void setRunningUpgrade(boolean isRunningUpgrade) {
+ this.isRunningUpgrade = isRunningUpgrade;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 17c2a1c..5a4a791 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -515,7 +515,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public PhoenixStatement newStatement(PhoenixConnection connection) {
return new PhoenixStatement(connection) {
@Override
- protected PhoenixResultSet newResultSet(ResultIterator iterator, RowProjector projector,
+ public PhoenixResultSet newResultSet(ResultIterator iterator, RowProjector projector,
StatementContext context) throws SQLException {
return new PhoenixResultSet(new TenantColumnFilteringIterator(iterator, projector),
projector, context);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index a7b2587..da7c7e8 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -37,6 +37,7 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
@@ -77,6 +78,7 @@ import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.exception.BatchUpdateExecution;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.exception.UpgradeRequiredException;
import org.apache.phoenix.execute.MutationState;
import org.apache.phoenix.expression.RowKeyColumnExpression;
import org.apache.phoenix.iterate.MaterializedResultIterator;
@@ -102,6 +104,7 @@ import org.apache.phoenix.parse.DropIndexStatement;
import org.apache.phoenix.parse.DropSchemaStatement;
import org.apache.phoenix.parse.DropSequenceStatement;
import org.apache.phoenix.parse.DropTableStatement;
+import org.apache.phoenix.parse.ExecuteUpgradeStatement;
import org.apache.phoenix.parse.ExplainStatement;
import org.apache.phoenix.parse.FilterableStatement;
import org.apache.phoenix.parse.HintNode;
@@ -189,7 +192,8 @@ public class PhoenixStatement implements Statement, SQLCloseable {
public enum Operation {
QUERY("queried", false),
DELETE("deleted", true),
- UPSERT("upserted", true);
+ UPSERT("upserted", true),
+ UPGRADE("upgrade", true);
private final String toString;
private final boolean isMutation;
@@ -238,7 +242,7 @@ public class PhoenixStatement implements Statement, SQLCloseable {
return resultSets;
}
- protected PhoenixResultSet newResultSet(ResultIterator iterator, RowProjector projector, StatementContext context) throws SQLException {
+ public PhoenixResultSet newResultSet(ResultIterator iterator, RowProjector projector, StatementContext context) throws SQLException {
return new PhoenixResultSet(iterator, projector, context);
}
@@ -265,7 +269,12 @@ public class PhoenixStatement implements Statement, SQLCloseable {
public PhoenixResultSet call() throws SQLException {
final long startTime = System.currentTimeMillis();
try {
- QueryPlan plan = stmt.compilePlan(PhoenixStatement.this, Sequence.ValueOp.VALIDATE_SEQUENCE);
+ PhoenixConnection conn = getConnection();
+ if (conn.getQueryServices().isUpgradeRequired() && !conn.isRunningUpgrade()
+ && stmt.getOperation() != Operation.UPGRADE) {
+ throw new UpgradeRequiredException();
+ }
+ QueryPlan plan = stmt.compilePlan(PhoenixStatement.this, Sequence.ValueOp.VALIDATE_SEQUENCE);
// Send mutations to hbase, so they are visible to subsequent reads.
// Use original plan for data table so that data and immutable indexes will be sent
// TODO: for joins, we need to iterate through all tables, but we need the original table,
@@ -330,6 +339,11 @@ public class PhoenixStatement implements Statement, SQLCloseable {
@Override
public Integer call() throws SQLException {
try {
+ PhoenixConnection conn = getConnection();
+ if (conn.getQueryServices().isUpgradeRequired() && !conn.isRunningUpgrade()
+ && stmt.getOperation() != Operation.UPGRADE) {
+ throw new UpgradeRequiredException();
+ }
MutationState state = connection.getMutationState();
MutationPlan plan = stmt.compilePlan(PhoenixStatement.this, Sequence.ValueOp.VALIDATE_SEQUENCE);
if (plan.getTargetRef() != null && plan.getTargetRef().getTable() != null && plan.getTargetRef().getTable().isTransactional()) {
@@ -1032,6 +1046,53 @@ public class PhoenixStatement implements Statement, SQLCloseable {
}
}
+
+ private static class ExecutableExecuteUpgradeStatement extends ExecuteUpgradeStatement implements CompilableStatement {
+ @SuppressWarnings("unchecked")
+ @Override
+ public MutationPlan compilePlan(final PhoenixStatement stmt, Sequence.ValueOp seqAction) throws SQLException {
+ return new MutationPlan() {
+
+ @Override
+ public Set<TableRef> getSourceRefs() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public ParameterMetaData getParameterMetaData() {
+ return PhoenixParameterMetaData.EMPTY_PARAMETER_META_DATA;
+ }
+
+ @Override
+ public Operation getOperation() {
+ return Operation.UPGRADE;
+ }
+
+ @Override
+ public ExplainPlan getExplainPlan() throws SQLException {
+ return new ExplainPlan(Collections.singletonList("EXECUTE UPGRADE"));
+ }
+
+ @Override
+ public StatementContext getContext() {
+ return new StatementContext(stmt);
+ }
+
+ @Override
+ public TableRef getTargetRef() {
+ return TableRef.EMPTY_TABLE_REF;
+ }
+
+ @Override
+ public MutationState execute() throws SQLException {
+ PhoenixConnection phxConn = stmt.getConnection();
+ Properties props = new Properties();
+ phxConn.getQueryServices().upgradeSystemTables(phxConn.getURL(), props);
+ return MutationState.emptyMutationState(-1, phxConn);
+ }
+ };
+ }
+ }
private static class ExecutableAddColumnStatement extends AddColumnStatement implements CompilableStatement {
@@ -1209,6 +1270,11 @@ public class PhoenixStatement implements Statement, SQLCloseable {
public UpdateStatisticsStatement updateStatistics(NamedTableNode table, StatisticsCollectionScope scope, Map<String,Object> props) {
return new ExecutableUpdateStatisticsStatement(table, scope, props);
}
+
+ @Override
+ public ExecuteUpgradeStatement executeUpgrade() {
+ return new ExecutableExecuteUpgradeStatement();
+ }
}
static class PhoenixStatementParser extends SQLParser {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/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 f6f2482..9c447e8 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
@@ -28,6 +28,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -52,6 +53,7 @@ import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.types.PDate;
import org.apache.phoenix.util.PhoenixMRJobUtil;
+import org.apache.phoenix.util.UpgradeUtil;
import org.apache.phoenix.util.PhoenixMRJobUtil.MR_SCHEDULER_TYPE;
import org.apache.phoenix.util.ZKBasedMasterElectionUtil;
import org.codehaus.jettison.json.JSONArray;
@@ -158,6 +160,8 @@ public class PhoenixMRJobSubmitter {
}
public Map<String, PhoenixAsyncIndex> getCandidateJobs() throws SQLException {
+ Properties props = new Properties();
+ UpgradeUtil.doNotUpgradeOnFirstConnection(props);
Connection con = DriverManager.getConnection("jdbc:phoenix:" + zkQuorum);
Statement s = con.createStatement();
ResultSet rs = s.executeQuery(CANDIDATE_INDEX_INFO_QUERY);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/parse/ExecuteUpgradeStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ExecuteUpgradeStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ExecuteUpgradeStatement.java
new file mode 100644
index 0000000..29edf8f
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ExecuteUpgradeStatement.java
@@ -0,0 +1,34 @@
+/*
+ * 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.parse;
+
+import org.apache.phoenix.jdbc.PhoenixStatement.Operation;
+
+public class ExecuteUpgradeStatement implements BindableStatement {
+
+ @Override
+ public int getBindCount() {
+ return 0;
+ }
+
+ @Override
+ public Operation getOperation() {
+ return Operation.UPGRADE;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
index 282ec62..d633299 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
@@ -389,6 +389,10 @@ public class ParseNodeFactory {
public UpdateStatisticsStatement updateStatistics(NamedTableNode table, StatisticsCollectionScope scope, Map<String,Object> props) {
return new UpdateStatisticsStatement(table, scope, props);
}
+
+ public ExecuteUpgradeStatement executeUpgrade() {
+ return new ExecuteUpgradeStatement();
+ }
public FunctionParseNode functionDistinct(String name, List<ParseNode> args) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/887b57cf/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
index 7154d58..6f8b528 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
@@ -139,4 +139,7 @@ public interface ConnectionQueryServices extends QueryServices, MetaDataMutated
* @param tableName The table to remove stats for
*/
void invalidateStats(ImmutableBytesPtr tableName);
+
+ boolean isUpgradeRequired();
+ void upgradeSystemTables(String url, Properties props) throws SQLException;
}
\ No newline at end of file