apache · swaroopak · Sep 21, 2020 · Sep 21, 2020
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/AlterAddCascadeIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/AlterAddCascadeIndexIT.java
diff --git a/phoenix-core/src/main/antlr3/PhoenixSQL.g b/phoenix-core/src/main/antlr3/PhoenixSQL.g
@@ -656,8 +656,8 @@ alter_session_node returns [AlterSessionStatement ret]
 // Parse an alter table statement.
 alter_table_node returns [AlterTableStatement ret]
     :   ALTER (TABLE | v=VIEW) t=from_table_name
-        ( (DROP COLUMN (IF ex=EXISTS)? c=column_names) | (ADD (IF NOT ex=EXISTS)? (d=column_defs) (p=fam_properties)?) | (SET (p=fam_properties)) )
-        { PTableType tt = v==null ? (QueryConstants.SYSTEM_SCHEMA_NAME.equals(t.getSchemaName()) ? PTableType.SYSTEM : PTableType.TABLE) : PTableType.VIEW; ret = ( c == null ? factory.addColumn(factory.namedTable(null,t), tt, d, ex!=null, p) : factory.dropColumn(factory.namedTable(null,t), tt, c, ex!=null) ); }
+        ( (DROP COLUMN (IF ex=EXISTS)? c=column_names) | (ADD (IF NOT ex=EXISTS)? (d=column_defs) (p=fam_properties)?) (cas=CASCADE INDEX (list=indexes | all=ALL))? | (SET (p=fam_properties)) )
+        { PTableType tt = v==null ? (QueryConstants.SYSTEM_SCHEMA_NAME.equals(t.getSchemaName()) ? PTableType.SYSTEM : PTableType.TABLE) : PTableType.VIEW; ret = ( c == null ? factory.addColumn(factory.namedTable(null,t), tt, d, ex!=null, p, cas!=null, (all == null ? list : null)) : factory.dropColumn(factory.namedTable(null,t), tt, c, ex!=null) ); }
     ;
 
 update_statistics_node returns [UpdateStatisticsStatement ret]
@@ -684,6 +684,11 @@ column_defs returns [List<ColumnDef> ret]
     :  v = column_def {$ret.add(v);}  (COMMA v = column_def {$ret.add(v);} )*
 ;
 
+indexes returns [List<NamedNode> ret]
+@init{ret = new ArrayList<NamedNode>(); }
+    :  v = index_name {$ret.add(v);}  (COMMA v = index_name {$ret.add(v);} )*
+;
+
 column_def returns [ColumnDef ret]
     :   c=column_name dt=identifier (LPAREN l=NUMBER (COMMA s=NUMBER)? RPAREN)? ar=ARRAY? (lsq=LSQUARE (a=NUMBER)? RSQUARE)? (nn=NOT? n=NULL)? (DEFAULT df=expression)? (pk=PRIMARY KEY (order=ASC|order=DESC)? rr=ROW_TIMESTAMP?)?
         { $ret = factory.columnDef(c, dt, ar != null || lsq != null, a == null ? null :  Integer.parseInt( a.getText() ), nn!=null ? Boolean.FALSE : n!=null ? Boolean.TRUE : null, 

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
@@ -528,7 +528,13 @@ public SQLException newException(SQLExceptionInfo info) {
             "Cannot use a connection with SCN set to upsert data for a table with indexes."),
 
     CANNOT_PERFORM_DDL_WITH_PENDING_MUTATIONS(904, "43M15",
-            "Cannot perform DDL with pending mutations. Commit or rollback mutations before performing DDL");
+            "Cannot perform DDL with pending mutations. Commit or rollback mutations before performing DDL"),
+
+    NOT_SUPPORTED_CASCADE_FEATURE_PK(905, "43M16", "CASCADE INDEX feature is not supported to add new PK column in INDEX"),
+
+    INCORRECT_INDEX_NAME(906, "43M17", "The list contains one or more incorrect index name(s)"),
+
+    NOT_SUPPORTED_CASCADE_FEATURE_LOCAL_INDEX(907, "43M18", "CASCADE INDEX feature is not supported for local index");
 
     private final int errorCode;
     private final String sqlState;

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
@@ -1467,8 +1467,8 @@ public Long getEstimateInfoTimestamp() throws SQLException {
 
     private static class ExecutableAddColumnStatement extends AddColumnStatement implements CompilableStatement {
 
-        ExecutableAddColumnStatement(NamedTableNode table, PTableType tableType, List<ColumnDef> columnDefs, boolean ifNotExists, ListMultimap<String,Pair<String,Object>> props) {
-            super(table, tableType, columnDefs, ifNotExists, props);
+        ExecutableAddColumnStatement(NamedTableNode table, PTableType tableType, List<ColumnDef> columnDefs, boolean ifNotExists, ListMultimap<String,Pair<String,Object>> props, boolean cascade, List<NamedNode> indexes) {
+            super(table, tableType, columnDefs, ifNotExists, props, cascade, indexes);
         }
 
         @SuppressWarnings("unchecked")
@@ -1609,8 +1609,8 @@ public CreateIndexStatement createIndex(NamedNode indexName, NamedTableNode data
         }
 
         @Override
-        public AddColumnStatement addColumn(NamedTableNode table,  PTableType tableType, List<ColumnDef> columnDefs, boolean ifNotExists, ListMultimap<String,Pair<String,Object>> props) {
-            return new ExecutableAddColumnStatement(table, tableType, columnDefs, ifNotExists, props);
+        public AddColumnStatement addColumn(NamedTableNode table,  PTableType tableType, List<ColumnDef> columnDefs, boolean ifNotExists, ListMultimap<String,Pair<String,Object>> props, boolean cascade, List<NamedNode> indexes) {
+            return new ExecutableAddColumnStatement(table, tableType, columnDefs, ifNotExists, props, cascade, indexes);
         }
 
         @Override

diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/AddColumnStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/AddColumnStatement.java
@@ -29,12 +29,16 @@ public class AddColumnStatement extends AlterTableStatement {
     private final List<ColumnDef> columnDefs;
     private final boolean ifNotExists;
     private final ListMultimap<String,Pair<String,Object>> props;
-
-    protected AddColumnStatement(NamedTableNode table, PTableType tableType, List<ColumnDef> columnDefs, boolean ifNotExists, ListMultimap<String,Pair<String,Object>> props) {
+    private final boolean cascade;
+    private final List<NamedNode> indexes;
+
+    protected AddColumnStatement(NamedTableNode table, PTableType tableType, List<ColumnDef> columnDefs, boolean ifNotExists, ListMultimap<String,Pair<String,Object>> props, boolean cascade, List<NamedNode> indexes) {
         super(table, tableType);
         this.columnDefs = columnDefs;
         this.props = props == null ? ImmutableListMultimap.<String,Pair<String,Object>>of()  : props;
         this.ifNotExists = ifNotExists;
+        this.cascade = cascade;
+        this.indexes = indexes;
     }
 
     public List<ColumnDef> getColumnDefs() {
@@ -48,4 +52,8 @@ public boolean ifNotExists() {
     public ListMultimap<String,Pair<String,Object>> getProps() {
         return props;
     }
+
+    public boolean isCascade() { return cascade; }
+
+    public List<NamedNode> getIndexes() { return indexes; }
 }
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
@@ -357,8 +357,8 @@ public SequenceValueParseNode nextValueFor(TableName tableName, ParseNode numToA
         return new SequenceValueParseNode(tableName, SequenceValueParseNode.Op.NEXT_VALUE, numToAllocateNode);
     }
 
-    public AddColumnStatement addColumn(NamedTableNode table,  PTableType tableType, List<ColumnDef> columnDefs, boolean ifNotExists, ListMultimap<String,Pair<String,Object>> props) {
-        return new AddColumnStatement(table, tableType, columnDefs, ifNotExists, props);
+    public AddColumnStatement addColumn(NamedTableNode table,  PTableType tableType, List<ColumnDef> columnDefs, boolean ifNotExists, ListMultimap<String,Pair<String,Object>> props, boolean cascade, List<NamedNode>indexes) {
+        return new AddColumnStatement(table, tableType, columnDefs, ifNotExists, props, cascade, indexes);
     }
 
     public DropColumnStatement dropColumn(NamedTableNode table,  PTableType tableType, List<ColumnName> columnNodes, boolean ifExists) {

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
@@ -156,6 +156,7 @@
 import org.apache.hadoop.hbase.security.access.Permission;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.util.StringUtils;
 import org.apache.phoenix.compile.ColumnResolver;
 import org.apache.phoenix.compile.FromCompiler;
 import org.apache.phoenix.compile.IndexExpressionCompiler;
@@ -206,6 +207,7 @@
 import org.apache.phoenix.parse.DropSequenceStatement;
 import org.apache.phoenix.parse.DropTableStatement;
 import org.apache.phoenix.parse.IndexKeyConstraint;
+import org.apache.phoenix.parse.NamedNode;
 import org.apache.phoenix.parse.NamedTableNode;
 import org.apache.phoenix.parse.OpenStatement;
 import org.apache.phoenix.parse.PFunction;
@@ -505,7 +507,6 @@ public class MetaDataClient {
 
     public static final String EMPTY_TABLE = " ";
 
-
     private final PhoenixConnection connection;
 
     public MetaDataClient(PhoenixConnection connection) {
@@ -1095,7 +1096,7 @@ public MutationState createTable(CreateTableStatement statement, byte[][] splits
                 }
                 // if there are new columns to add
                 return addColumn(table, columnDefs, statement.getProps(), statement.ifNotExists(),
-                        true, NamedTableNode.create(statement.getTableName()), statement.getTableType());
+                        true, NamedTableNode.create(statement.getTableName()), statement.getTableType(), false, null);
             }
         }
         table = createTableInternal(statement, splits, parent, viewStatement, viewType, viewIndexIdType, viewColumnConstants, isViewColumnReferenced, false, null, null, tableProps, commonFamilyProps);
@@ -3629,14 +3630,25 @@ private void mutateStringProperty(String tenantId, String schemaName, String tab
 
     public MutationState addColumn(AddColumnStatement statement) throws SQLException {
         PTable table = FromCompiler.getResolver(statement, connection).getTables().get(0).getTable();
-        return addColumn(table, statement.getColumnDefs(), statement.getProps(), statement.ifNotExists(), false, statement.getTable(), statement.getTableType());
+        return addColumn(table, statement.getColumnDefs(), statement.getProps(), statement.ifNotExists(), false, statement.getTable(), statement.getTableType(), statement.isCascade(), statement.getIndexes());
     }
 
     public MutationState addColumn(PTable table, List<ColumnDef> origColumnDefs,
             ListMultimap<String, Pair<String, Object>> stmtProperties, boolean ifNotExists,
-            boolean removeTableProps, NamedTableNode namedTableNode, PTableType tableType)
+            boolean removeTableProps, NamedTableNode namedTableNode, PTableType tableType, boolean cascade, List<NamedNode> indexes)
                     throws SQLException {
         connection.rollback();
+        List<PTable> indexesPTable = Lists.newArrayListWithExpectedSize(indexes != null ?
+                indexes.size() : table.getIndexes().size());
+        Map<PTable, Integer> indexToColumnSizeMap = new HashMap<>();
+
+        // if cascade keyword is passed and indexes are provided either implicitly or explicitly
+        if (cascade && (indexes == null || !indexes.isEmpty())) {
+            indexesPTable = getIndexesPTableForCascade(indexes, table);
+            for(PTable index : indexesPTable) {
+                indexToColumnSizeMap.put(index, index.getColumns().size());
+            }
+        }
         boolean wasAutoCommit = connection.getAutoCommit();
         List<PColumn> columns = Lists.newArrayListWithExpectedSize(origColumnDefs != null ?
             origColumnDefs.size() : 0);
@@ -3769,10 +3781,10 @@ public MutationState addColumn(PTable table, List<ColumnDef> origColumnDefs,
                             if (!colDef.validateDefault(context, null)) {
                                 colDef = new ColumnDef(colDef, null); // Remove DEFAULT as it's not necessary
                             }
+                            String familyName = null;
                             Integer encodedCQ = null;
                             if (!colDef.isPK()) {
                                 String colDefFamily = colDef.getColumnDefName().getFamilyName();
-                                String familyName = null;
                                 ImmutableStorageScheme storageScheme = table.getImmutableStorageScheme();
                                 String defaultColumnFamily = tableForCQCounters.getDefaultFamilyName() != null && !Strings.isNullOrEmpty(tableForCQCounters.getDefaultFamilyName().getString()) ? 
                                         tableForCQCounters.getDefaultFamilyName().getString() : DEFAULT_COLUMN_FAMILY;
@@ -3800,6 +3812,12 @@ public MutationState addColumn(PTable table, List<ColumnDef> origColumnDefs,
                                 .setTableName(tableName).build().buildException();
                             }
                             PColumn column = newColumn(position++, colDef, PrimaryKeyConstraint.EMPTY, table.getDefaultFamilyName() == null ? null : table.getDefaultFamilyName().getString(), true, columnQualifierBytes, willBeImmutableRows);
+                            HashMap<PTable, PColumn> indexToIndexColumnMap = null;
+                            if (cascade) {
+                                indexToIndexColumnMap = getPTablePColumnHashMapForCascade(indexesPTable, willBeImmutableRows,
+                                                colDef, familyName, indexToColumnSizeMap);
+                            }
+
                             columns.add(column);
                             String pkName = null;
                             Short keySeq = null;
@@ -3814,6 +3832,13 @@ public MutationState addColumn(PTable table, List<ColumnDef> origColumnDefs,
                             }
                             colFamiliesForPColumnsToBeAdded.add(column.getFamilyName() == null ? null : column.getFamilyName().getString());
                             addColumnMutation(schemaName, tableName, column, colUpsert, null, pkName, keySeq, table.getBucketNum() != null);
+                            // add new columns for given indexes one by one
+                            if (cascade) {
+                                for (PTable index: indexesPTable) {
+                                    LOGGER.info("Adding column "+column.getName().getString()+" to "+index.getTableName().toString());
+                                    addColumnMutation(schemaName, index.getTableName().getString(), indexToIndexColumnMap.get(index), colUpsert, null, "", keySeq, index.getBucketNum() != null);
+                                }
+                            }
                         }
 
                         // Add any new PK columns to end of index PK
@@ -3879,7 +3904,18 @@ public MutationState addColumn(PTable table, List<ColumnDef> origColumnDefs,
                     tableMetaData.addAll(connection.getMutationState().toMutations(timeStamp).next().getSecond());
                     connection.rollback();
                 }
-
+
+                if (cascade) {
+                    for (PTable index : indexesPTable) {
+                        incrementTableSeqNum(index, index.getType(), columnDefs.size(),
+                                Boolean.FALSE,
+                                metaPropertiesEvaluated.getUpdateCacheFrequency(),
+                                metaPropertiesEvaluated.getPhoenixTTL());
+                    }
+                    tableMetaData.addAll(connection.getMutationState().toMutations(timeStamp).next().getSecond());
+                    connection.rollback();
+                }
+
                 if (changingPhoenixTableProperty || columnDefs.size() > 0) {
                     incrementTableSeqNum(table, tableType, columnDefs.size(), metaPropertiesEvaluated);
 
@@ -4056,6 +4092,80 @@ public MutationState addColumn(PTable table, List<ColumnDef> origColumnDefs,
         }
     }
 
+    private List<PTable> getIndexesPTableForCascade(List<NamedNode> indexes, PTable table) throws SQLException {
+        boolean isView = table.getType().equals(PTableType.VIEW);
+        List<PTable> indexesPTable = new ArrayList<>();
+
+        // when indexes is null, that means ALL keyword is passed and
+        // we ll collect all global indexes for cascading
+        if(indexes == null) {
+            indexesPTable.addAll(table.getIndexes());
+            for (PTable index : table.getIndexes()) {
+                // a child view has access to its parents indexes,
+                // this if clause ensures we only get the indexes that
+                // are only created on the view itself.
+                if (index.getIndexType().equals(IndexType.LOCAL)
+                        || (isView && index.getTableName().toString().contains("#"))) {
+                    indexesPTable.remove(index);
+                }
+            }
+        } else {
+            List<String> indexesParam = Lists.newArrayListWithExpectedSize(indexes.size());
+            for (NamedNode index : indexes) {
+                indexesParam.add(index.getName());
+            }
+            // gets the PTable for list of indexes passed in the function
+            // if all the names in parameter list are correct, indexesParam list should be empty
+            // by end of the loop
+            for (PTable index : table.getIndexes()) {
+                if(index.getIndexType().equals(IndexType.LOCAL)) {
+                    throw new SQLExceptionInfo
+                            .Builder(SQLExceptionCode.NOT_SUPPORTED_CASCADE_FEATURE_LOCAL_INDEX)
+                            .setTableName(index.getName().getString())
+                            .build()
+                            .buildException();
+                }
+                if (indexesParam.remove(index.getTableName().getString())) {
+                    indexesPTable.add(index);
+                }
+            }
+            // indexesParam has index names that are not correct
+            if (!indexesParam.isEmpty()) {
+                throw new SQLExceptionInfo
+                        .Builder(SQLExceptionCode.INCORRECT_INDEX_NAME)
+                        .setTableName(StringUtils.join(",", indexesParam))
+                        .build()
+                        .buildException();
+            }
+        }
+        return indexesPTable;
+    }
+
+    private HashMap<PTable, PColumn> getPTablePColumnHashMapForCascade(List<PTable> indexesPTable,
+            boolean willBeImmutableRows, ColumnDef colDef, String familyName, Map<PTable, Integer> indexToColumnSizeMap) throws SQLException {
+        HashMap<PTable, PColumn> indexColumn;
+        if (colDef.isPK()) {
+            //only supported for non pk column
+            throw new SQLExceptionInfo.Builder(
+                    SQLExceptionCode.NOT_SUPPORTED_CASCADE_FEATURE_PK)
+                    .build()
+                    .buildException();
+        }
+        indexColumn = new HashMap(indexesPTable.size());
+        PDataType indexColDataType = IndexUtil.getIndexColumnDataType(colDef.isNull(), colDef.getDataType());
+        ColumnName
+                indexColName = ColumnName.caseSensitiveColumnName(IndexUtil.getIndexColumnName(familyName, colDef.getColumnDefName().getColumnName()));
+        ColumnDef indexColDef = FACTORY.columnDef(indexColName, indexColDataType.getSqlTypeName(), colDef.isNull(), colDef.getMaxLength(), colDef.getScale(), false, colDef.getSortOrder(), colDef.getExpression(), colDef.isRowTimestamp());
+        // TODO: add support to specify tenant owned indexes in the DDL statement with CASCADE executed with Global connection
+        for (PTable index : indexesPTable) {
+            int iPos = indexToColumnSizeMap.get(index);
+            PColumn iColumn = newColumn(iPos, indexColDef, null, "", false, null, willBeImmutableRows);
+            indexColumn.put(index, iColumn);
+            indexToColumnSizeMap.put(index, iPos+1);
+        }
+        return indexColumn;
+    }
+
     private void deleteMutexCells(String physicalSchemaName, String physicalTableName, Set<String> acquiredColumnMutexSet) throws SQLException {
         if (!acquiredColumnMutexSet.isEmpty()) {
             for (String columnName : acquiredColumnMutexSet) {