| /* |
| * 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.schema.transform; |
| |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.phoenix.coprocessorclient.TableInfo; |
| import org.apache.phoenix.schema.PIndexState; |
| import org.apache.phoenix.thirdparty.com.google.common.base.Strings; |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.phoenix.jdbc.PhoenixConnection; |
| import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; |
| import org.apache.phoenix.mapreduce.index.IndexScrutinyTool; |
| import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil; |
| import org.apache.phoenix.query.QueryServicesOptions; |
| import org.apache.phoenix.schema.MetaDataClient; |
| import org.apache.phoenix.schema.PColumn; |
| import org.apache.phoenix.schema.PColumnImpl; |
| import org.apache.phoenix.schema.PName; |
| import org.apache.phoenix.schema.PNameFactory; |
| import org.apache.phoenix.schema.PTable; |
| import org.apache.phoenix.schema.PTableImpl; |
| import org.apache.phoenix.util.EncodedColumnsUtil; |
| import org.apache.phoenix.util.EnvironmentEdgeManager; |
| import org.apache.phoenix.util.QueryUtil; |
| import org.apache.phoenix.util.SchemaUtil; |
| import org.apache.phoenix.util.TableViewFinderResult; |
| import org.apache.phoenix.util.UpgradeUtil; |
| import org.apache.phoenix.util.ViewUtil; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.sql.Connection; |
| import java.sql.PreparedStatement; |
| import java.sql.SQLException; |
| import java.sql.Timestamp; |
| import java.sql.Types; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| |
| import static org.apache.phoenix.coprocessorclient.MetaDataProtocol.MIN_TABLE_TIMESTAMP; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ENCODING_SCHEME; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_STORAGE_SCHEME; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PHYSICAL_TABLE_NAME; |
| import static org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.MAPREDUCE_TENANT_ID; |
| import static org.apache.phoenix.query.QueryConstants.DEFAULT_COLUMN_FAMILY; |
| import static org.apache.phoenix.query.QueryConstants.ENCODED_CQ_COUNTER_INITIAL_VALUE; |
| import static org.apache.phoenix.query.QueryServices.MUTATE_BATCH_SIZE_ATTRIB; |
| import static org.apache.phoenix.schema.ColumnMetaDataOps.addColumnMutation; |
| import static org.apache.phoenix.schema.MetaDataClient.UPDATE_INDEX_STATE_TO_ACTIVE; |
| import static org.apache.phoenix.schema.PTable.ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS; |
| import static org.apache.phoenix.schema.PTableType.INDEX; |
| |
| public class Transform extends TransformClient { |
| private static final Logger LOGGER = LoggerFactory.getLogger(Transform.class); |
| |
| public static PTable getTransformingNewTable(PhoenixConnection connection, PTable oldTable) throws SQLException{ |
| SystemTransformRecord transformRecord = TransformClient.getTransformRecord(connection, oldTable.getType(), oldTable.getSchemaName() |
| , oldTable.getTableName(), oldTable.getType()==INDEX? oldTable.getParentTableName():null, oldTable.getTenantId() |
| , oldTable.getBaseTableLogicalName()); |
| |
| PTable transformingNewTable = null; |
| if (transformRecord != null && transformRecord.isActive()) { |
| // New table will behave like an index |
| PName newTableNameWithoutSchema = PNameFactory.newName(SchemaUtil.getTableNameFromFullName(transformRecord.getNewPhysicalTableName())); |
| if (!newTableNameWithoutSchema.equals(oldTable.getPhysicalName(true))) { |
| transformingNewTable = connection.getTableNoCache( |
| transformRecord.getNewPhysicalTableName()); |
| } |
| } |
| return transformingNewTable; |
| } |
| |
| public static void updateNewTableState(PhoenixConnection connection, SystemTransformRecord systemTransformRecord, |
| PIndexState state) |
| throws SQLException { |
| String schema = SchemaUtil.getSchemaNameFromFullName(systemTransformRecord.getNewPhysicalTableName()); |
| String tableName = SchemaUtil.getTableNameFromFullName(systemTransformRecord.getNewPhysicalTableName()); |
| try (PreparedStatement tableUpsert = connection.prepareStatement(UPDATE_INDEX_STATE_TO_ACTIVE)){ |
| tableUpsert.setString(1, systemTransformRecord.getTenantId() == null ? null : |
| systemTransformRecord.getTenantId()); |
| tableUpsert.setString(2, schema); |
| tableUpsert.setString(3, tableName); |
| tableUpsert.setString(4, state.getSerializedValue()); |
| tableUpsert.setLong(5, 0); |
| tableUpsert.setLong(6, 0); |
| tableUpsert.execute(); |
| } |
| // Update cache |
| UpgradeUtil.clearCache(connection, connection.getTenantId(), schema, tableName, |
| systemTransformRecord.getLogicalParentName(), MIN_TABLE_TIMESTAMP); |
| } |
| |
| public static void removeTransformRecord( |
| SystemTransformRecord transformRecord, PhoenixConnection connection) throws SQLException { |
| connection.prepareStatement("DELETE FROM " |
| + PhoenixDatabaseMetaData.SYSTEM_TRANSFORM_NAME + " WHERE " + |
| (Strings.isNullOrEmpty(transformRecord.getSchemaName()) ? "" : |
| (PhoenixDatabaseMetaData.TABLE_SCHEM + " ='" + transformRecord.getSchemaName() + "' AND ")) + |
| PhoenixDatabaseMetaData.LOGICAL_TABLE_NAME + " ='" + transformRecord.getLogicalTableName() + "' AND " + |
| PhoenixDatabaseMetaData.NEW_PHYS_TABLE_NAME + " ='" + transformRecord.getNewPhysicalTableName() + "' AND " + |
| PhoenixDatabaseMetaData.TRANSFORM_TYPE + " =" + transformRecord.getTransformType().getSerializedValue() |
| ).execute(); |
| } |
| |
| /** |
| * Disable caching re-design if you use Online Data Format Change since the cutover logic |
| * is currently incompatible and clients may not learn about the physical table change. |
| * See https://issues.apache.org/jira/browse/PHOENIX-6883 and |
| * https://issues.apache.org/jira/browse/PHOENIX-7284. |
| */ |
| public static void doCutover(PhoenixConnection connection, SystemTransformRecord systemTransformRecord) throws Exception{ |
| String tenantId = systemTransformRecord.getTenantId(); |
| String schema = systemTransformRecord.getSchemaName(); |
| String tableName = systemTransformRecord.getLogicalTableName(); |
| String newTableName = SchemaUtil.getTableNameFromFullName(systemTransformRecord.getNewPhysicalTableName()); |
| |
| // Calculate changed metadata |
| List<String> columnNames = new ArrayList<>(); |
| List<String> columnValues = new ArrayList<>(); |
| |
| getMetadataDifference(connection, systemTransformRecord, columnNames, columnValues); |
| // TODO In the future, we need to handle rowkey changes and column type changes as well |
| |
| String changeViewStmt = "UPSERT INTO SYSTEM.CATALOG " |
| + "(TENANT_ID, TABLE_SCHEM, TABLE_NAME %s) VALUES (?, ?, ? %s)"; |
| |
| String |
| changeTable = String.format("UPSERT INTO SYSTEM.CATALOG " |
| + "(TENANT_ID, TABLE_SCHEM, TABLE_NAME, PHYSICAL_TABLE_NAME %s ) " |
| + "VALUES(?, ?, ?, ? %s)", columnNames.size() > 0 ? "," |
| + String.join(",", columnNames) : "", columnNames.size() > 0 |
| ? "," + QueryUtil.generateInListParams(columnValues.size()) : ""); |
| |
| LOGGER.info("About to do cutover via " + changeTable); |
| TableViewFinderResult childViewsResult = ViewUtil.findChildViews(connection, tenantId, schema, tableName); |
| boolean wasCommit = connection.getAutoCommit(); |
| connection.setAutoCommit(false); |
| List<TableInfo> viewsToUpdateCache = new ArrayList<>(); |
| try { |
| try (PreparedStatement stmt = connection.prepareStatement(changeTable)) { |
| int param = 0; |
| if (tenantId == null) { |
| stmt.setNull(++param, Types.VARCHAR); |
| } else { |
| stmt.setString(++param, tenantId); |
| } |
| if (schema == null) { |
| stmt.setNull(++param, Types.VARCHAR); |
| } else { |
| stmt.setString(++param, schema); |
| } |
| stmt.setString(++param, tableName); |
| stmt.setString(++param, newTableName); |
| for (int i = 0; i < columnValues.size(); i++) { |
| stmt.setInt(++param, Integer.parseInt(columnValues.get(i))); |
| } |
| stmt.execute(); |
| } |
| // Update column qualifiers |
| PTable pNewTable = connection.getTable(systemTransformRecord.getNewPhysicalTableName()); |
| PTable pOldTable = connection.getTable(SchemaUtil.getTableName(schema, tableName)); |
| if (pOldTable.getImmutableStorageScheme() != pNewTable.getImmutableStorageScheme() || |
| pOldTable.getEncodingScheme() != pNewTable.getEncodingScheme()) { |
| MetaDataClient.mutateTransformProperties(connection, tenantId, schema, tableName, newTableName, |
| pNewTable.getImmutableStorageScheme(), pNewTable.getEncodingScheme()); |
| // We need to update the columns's qualifiers as well |
| mutateColumns(connection.unwrap(PhoenixConnection.class), pOldTable, pNewTable); |
| |
| HashMap<String, PColumn> columnMap = new HashMap<>(); |
| for (PColumn column : pNewTable.getColumns()) { |
| columnMap.put(column.getName().getString(), column); |
| } |
| |
| // Also update view column qualifiers |
| for (TableInfo view : childViewsResult.getLinks()) { |
| PTable pView = connection.getTable(view.getTenantId() == null |
| ? null : Bytes.toString(view.getTenantId()), |
| SchemaUtil.getTableName(view.getSchemaName(), view.getTableName())); |
| mutateViewColumns(connection.unwrap(PhoenixConnection.class), pView, pNewTable, columnMap); |
| } |
| } |
| connection.commit(); |
| |
| // We can have millions of views. We need to send it in batches |
| int maxBatchSize = connection.getQueryServices().getConfiguration().getInt(MUTATE_BATCH_SIZE_ATTRIB, QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE); |
| int batchSize = 0; |
| for (TableInfo view : childViewsResult.getLinks()) { |
| String changeView = String.format(changeViewStmt, |
| columnNames.size() > 0 ? "," + String.join(",", columnNames) : "", |
| columnNames.size() > 0 ? "," |
| + QueryUtil.generateInListParams(columnValues.size()) : ""); |
| LOGGER.info("Cutover changing view via " + changeView); |
| try (PreparedStatement stmt = connection.prepareStatement(changeView)) { |
| int param = 0; |
| if (view.getTenantId() == null || view.getTenantId().length == 0) { |
| stmt.setNull(++param, Types.VARCHAR); |
| } else { |
| stmt.setString(++param, Bytes.toString(view.getTenantId())); |
| } |
| if (view.getSchemaName() == null || view.getSchemaName().length == 0) { |
| stmt.setNull(++param, Types.VARCHAR); |
| } else { |
| stmt.setString(++param, Bytes.toString(view.getSchemaName())); |
| } |
| stmt.setString(++param, Bytes.toString(view.getTableName())); |
| for (int i = 0; i < columnValues.size(); i++) { |
| stmt.setInt(++param, Integer.parseInt(columnValues.get(i))); |
| } |
| stmt.execute(); |
| } |
| viewsToUpdateCache.add(view); |
| batchSize++; |
| if (batchSize >= maxBatchSize) { |
| connection.commit(); |
| batchSize = 0; |
| } |
| } |
| if (batchSize > 0) { |
| connection.commit(); |
| batchSize = 0; |
| } |
| |
| connection.unwrap(PhoenixConnection.class).getQueryServices().clearCache(); |
| UpgradeUtil.clearCacheAndGetNewTable(connection.unwrap(PhoenixConnection.class), |
| connection.getTenantId(), |
| schema, tableName, systemTransformRecord.getLogicalParentName(), MIN_TABLE_TIMESTAMP); |
| for (TableInfo view : viewsToUpdateCache) { |
| UpgradeUtil.clearCache(connection.unwrap(PhoenixConnection.class), |
| PNameFactory.newName(view.getTenantId()), |
| PNameFactory.newName(view.getSchemaName()).getString(), Bytes.toString(view.getTableName()), |
| tableName, MIN_TABLE_TIMESTAMP); |
| } |
| |
| // TODO: Cleanup syscat so that we don't have an extra index |
| } catch (Exception e) { |
| LOGGER.error("Error happened during cutover ", e); |
| connection.rollback(); |
| throw e; |
| } finally { |
| connection.setAutoCommit(wasCommit); |
| } |
| } |
| |
| private static void getMetadataDifference(PhoenixConnection connection, |
| SystemTransformRecord systemTransformRecord, |
| List<String> columnNames, List<String> columnValues) throws SQLException { |
| PTable pOldTable = connection.getTable(SchemaUtil.getQualifiedTableName( |
| systemTransformRecord.getSchemaName(), |
| systemTransformRecord.getLogicalTableName())); |
| PTable pNewTable = connection.getTable(SchemaUtil.getQualifiedTableName(SchemaUtil |
| .getSchemaNameFromFullName(systemTransformRecord.getNewPhysicalTableName()), |
| SchemaUtil.getTableNameFromFullName( |
| systemTransformRecord.getNewPhysicalTableName()))); |
| |
| Map<String, String> map = pOldTable.getPropertyValues(); |
| for(Map.Entry<String, String> entry : map.entrySet()) { |
| String oldKey = entry.getKey(); |
| String oldValue = entry.getValue(); |
| if (pNewTable.getPropertyValues().containsKey(oldKey)) { |
| if (PHYSICAL_TABLE_NAME.equals(oldKey)) { |
| // No need to add here. We will add it. |
| continue; |
| } |
| String newValue = pNewTable.getPropertyValues().get(oldKey); |
| if (!Strings.nullToEmpty(oldValue).equals(Strings.nullToEmpty(newValue))) { |
| columnNames.add(oldKey); |
| // properties value that corresponds to a number will not need single quotes around it |
| // properties value that corresponds to a boolean value will not need single quotes around it |
| if (!Strings.isNullOrEmpty(newValue)) { |
| if(!(StringUtils.isNumeric(newValue)) && |
| !(newValue.equalsIgnoreCase(Boolean.TRUE.toString()) ||newValue.equalsIgnoreCase(Boolean.FALSE.toString()))) { |
| if (ENCODING_SCHEME.equals(oldKey)) { |
| newValue = String.valueOf(PTable.QualifierEncodingScheme.valueOf(newValue).getSerializedMetadataValue()); |
| } else if (IMMUTABLE_STORAGE_SCHEME.equals(oldKey)) { |
| newValue = String.valueOf(PTable.ImmutableStorageScheme.valueOf(newValue).getSerializedMetadataValue()); |
| } |
| else { |
| newValue = "'" + newValue + "'"; |
| } |
| } |
| } |
| columnValues.add(newValue); |
| } |
| } |
| } |
| } |
| |
| public static void completeTransform(Connection connection, Configuration configuration) throws Exception{ |
| // Will be called from Reducer |
| String tenantId = configuration.get(MAPREDUCE_TENANT_ID, null); |
| String fullOldTableName = PhoenixConfigurationUtil.getInputTableName(configuration); |
| String schemaName = SchemaUtil.getSchemaNameFromFullName(fullOldTableName); |
| String oldTableLogicalName = SchemaUtil.getTableNameFromFullName(fullOldTableName); |
| String indexTableName = SchemaUtil.getTableNameFromFullName(PhoenixConfigurationUtil.getIndexToolIndexTableName(configuration)); |
| String logicaTableName = oldTableLogicalName; |
| String logicalParentName = null; |
| if (PhoenixConfigurationUtil.getTransformingTableType(configuration) == IndexScrutinyTool.SourceTable.INDEX_TABLE_SOURCE) { |
| if (!Strings.isNullOrEmpty(indexTableName)) { |
| logicaTableName = indexTableName; |
| logicalParentName = SchemaUtil.getTableName(schemaName, oldTableLogicalName); |
| } |
| } |
| |
| SystemTransformRecord transformRecord = getTransformRecord(schemaName, logicaTableName, logicalParentName, |
| tenantId, connection.unwrap(PhoenixConnection.class)); |
| |
| if (!PTable.TransformType.isPartialTransform(transformRecord.getTransformType())) { |
| updateTransformRecord(connection.unwrap(PhoenixConnection.class), transformRecord, PTable.TransformStatus.PENDING_CUTOVER); |
| connection.commit(); |
| } else { |
| updateTransformRecord(connection.unwrap(PhoenixConnection.class), transformRecord, PTable.TransformStatus.COMPLETED); |
| connection.commit(); |
| } |
| } |
| |
| public static void updateTransformRecord(PhoenixConnection connection, SystemTransformRecord transformRecord, PTable.TransformStatus newStatus) throws SQLException { |
| SystemTransformRecord.SystemTransformBuilder builder = new SystemTransformRecord.SystemTransformBuilder(transformRecord); |
| builder.setTransformStatus(newStatus.name()); |
| builder.setLastStateTs(new Timestamp(EnvironmentEdgeManager.currentTimeMillis())); |
| if (newStatus == PTable.TransformStatus.STARTED) { |
| builder.setTransformRetryCount(transformRecord.getTransformRetryCount() + 1); |
| } |
| Transform.upsertTransform(builder.build(), connection); |
| } |
| |
| private static void mutateColumns(PhoenixConnection connection, PTable pOldTable, PTable pNewTable) throws SQLException { |
| if (pOldTable.getEncodingScheme() != pNewTable.getEncodingScheme()) { |
| Short nextKeySeq = 0; |
| for (PColumn column : pNewTable.getColumns()) { |
| boolean isPk = SchemaUtil.isPKColumn(column); |
| Short keySeq = isPk ? ++nextKeySeq : null; |
| PColumn newCol = new PColumnImpl(column.getName(), column.getFamilyName(), column.getDataType(), |
| column.getMaxLength(), column.getScale(), column.isNullable(), column.getPosition(), column.getSortOrder() |
| , column.getArraySize(), |
| column.getViewConstant(), column.isViewReferenced(), column.getExpressionStr(), column.isRowTimestamp(), |
| column.isDynamic(), column.getColumnQualifierBytes(), EnvironmentEdgeManager.currentTimeMillis()); |
| addColumnMutation(connection, pOldTable.getSchemaName()==null?null:pOldTable.getSchemaName().getString() |
| , pOldTable.getTableName().getString(), newCol, |
| pNewTable.getParentTableName() == null ? null : pNewTable.getParentTableName().getString() |
| , pNewTable.getPKName() == null ? null : pNewTable.getPKName().getString(), keySeq, pNewTable.getBucketNum() != null); |
| } |
| } |
| } |
| |
| public static PTable getTransformedView(PTable pOldView, PTable pNewTable, HashMap<String, PColumn> columnMap, boolean withDerivedColumns) throws SQLException { |
| List<PColumn> newColumns = new ArrayList<>(); |
| PTable pNewView = null; |
| if (pOldView.getEncodingScheme() != pNewTable.getEncodingScheme()) { |
| Short nextKeySeq = 0; |
| PTable.EncodedCQCounter cqCounterToUse = pNewTable.getEncodedCQCounter(); |
| String defaultColumnFamily = pNewTable.getDefaultFamilyName() != null && !Strings.isNullOrEmpty(pNewTable.getDefaultFamilyName().getString()) ? |
| pNewTable.getDefaultFamilyName().getString() : DEFAULT_COLUMN_FAMILY; |
| |
| for (PColumn column : pOldView.getColumns()) { |
| boolean isPk = SchemaUtil.isPKColumn(column); |
| Short keySeq = isPk ? ++nextKeySeq : null; |
| if (isPk) { |
| continue; |
| } |
| String familyName = null; |
| if (pNewTable.getImmutableStorageScheme() == SINGLE_CELL_ARRAY_WITH_OFFSETS) { |
| familyName = column.getFamilyName() != null ? column.getFamilyName().getString() : defaultColumnFamily; |
| } else { |
| familyName = defaultColumnFamily; |
| } |
| int encodedCQ = pOldView.isAppendOnlySchema() ? Integer.valueOf(ENCODED_CQ_COUNTER_INITIAL_VALUE + keySeq) : cqCounterToUse.getNextQualifier(familyName); |
| byte[] colQualifierBytes = EncodedColumnsUtil.getColumnQualifierBytes(column.getName().getString(), |
| encodedCQ, pNewTable, isPk); |
| if (columnMap.containsKey(column.getName().getString())) { |
| colQualifierBytes = columnMap.get(column.getName().getString()).getColumnQualifierBytes(); |
| } else { |
| if (!column.isDerived()) { |
| cqCounterToUse.increment(familyName); |
| } |
| } |
| |
| if (!withDerivedColumns && column.isDerived()) { |
| // Don't need to add/change derived columns |
| continue; |
| } |
| |
| PColumn newCol = new PColumnImpl(column.getName(), PNameFactory.newName(familyName), column.getDataType(), |
| column.getMaxLength(), column.getScale(), column.isNullable(), column.getPosition(), column.getSortOrder() |
| , column.getArraySize(), |
| column.getViewConstant(), column.isViewReferenced(), column.getExpressionStr(), column.isRowTimestamp(), |
| column.isDynamic(), colQualifierBytes, EnvironmentEdgeManager.currentTimeMillis()); |
| newColumns.add(newCol); |
| if (!columnMap.containsKey(newCol.getName().getString())) { |
| columnMap.put(newCol.getName().getString(), newCol) ; |
| } |
| } |
| |
| pNewView = PTableImpl.builderWithColumns(pOldView, newColumns) |
| .setQualifierEncodingScheme(pNewTable.getEncodingScheme()) |
| .setImmutableStorageScheme(pNewTable.getImmutableStorageScheme()) |
| .setPhysicalNames( |
| Collections.singletonList(SchemaUtil.getPhysicalHBaseTableName( |
| pNewTable.getSchemaName(), pNewTable.getTableName(), pNewTable.isNamespaceMapped()))) |
| .build(); |
| } else { |
| // Have to change this per transform type |
| } |
| return pNewView; |
| } |
| |
| private static void mutateViewColumns(PhoenixConnection connection, PTable pView, PTable pNewTable, HashMap<String, PColumn> columnMap) throws SQLException { |
| if (pView.getEncodingScheme() != pNewTable.getEncodingScheme()) { |
| Short nextKeySeq = 0; |
| PTable newView = getTransformedView(pView, pNewTable, columnMap,false); |
| for (PColumn newCol : newView.getColumns()) { |
| boolean isPk = SchemaUtil.isPKColumn(newCol); |
| Short keySeq = isPk ? ++nextKeySeq : null; |
| if (isPk) { |
| continue; |
| } |
| String tenantId = pView.getTenantId() == null ? null : pView.getTenantId().getString(); |
| addColumnMutation(connection, tenantId, pView.getSchemaName() == null ? null : pView.getSchemaName().getString() |
| , pView.getTableName().getString(), newCol, |
| pView.getParentTableName() == null ? null : pView.getParentTableName().getString() |
| , pView.getPKName() == null ? null : pView.getPKName().getString(), keySeq, pView.getBucketNum() != null); |
| } |
| } |
| } |
| |
| public static void doForceCutover(Connection connection, Configuration configuration) throws Exception{ |
| PhoenixConnection phoenixConnection = connection.unwrap(PhoenixConnection.class); |
| // Will be called from Reducer |
| String tenantId = configuration.get(MAPREDUCE_TENANT_ID, null); |
| String fullOldTableName = PhoenixConfigurationUtil.getInputTableName(configuration); |
| String schemaName = SchemaUtil.getSchemaNameFromFullName(fullOldTableName); |
| String oldTableLogicalName = SchemaUtil.getTableNameFromFullName(fullOldTableName); |
| String indexTableName = SchemaUtil.getTableNameFromFullName(PhoenixConfigurationUtil.getIndexToolIndexTableName(configuration)); |
| String logicaTableName = oldTableLogicalName; |
| String logicalParentName = null; |
| if (PhoenixConfigurationUtil.getTransformingTableType(configuration) == IndexScrutinyTool.SourceTable.INDEX_TABLE_SOURCE) |
| if (!Strings.isNullOrEmpty(indexTableName)) { |
| logicaTableName = indexTableName; |
| logicalParentName = SchemaUtil.getTableName(schemaName, oldTableLogicalName); |
| } |
| |
| SystemTransformRecord transformRecord = getTransformRecord(schemaName, logicaTableName, logicalParentName, |
| tenantId, phoenixConnection); |
| Transform.doCutover(phoenixConnection, transformRecord); |
| updateTransformRecord(phoenixConnection, transformRecord, PTable.TransformStatus.COMPLETED); |
| phoenixConnection.commit(); |
| } |
| |
| } |
| |
| |
