Support CREATE TABLE LIKE with INDEXES

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Support CREATE TABLE LIKE WITH INDEXES (CASSANDRA-19965)
  * Invalidate relevant prepared statements on every change to TableMetadata (CASSANDRA-20318)
  * Add per type max size guardrails (CASSANDRA-19677)
  * Make it possible to abort all kinds of multi step operations (CASSANDRA-20217)

doc/cql3/CQL.textile

@@ -414,16 +414,19 @@ bc(syntax)..
 <copy-table-stmt> ::= CREATE ( TABLE | COLUMNFAMILY ) ( IF NOT EXISTS )? <newtablename> LIKE <oldtablename>
                     ( WITH <option> ( AND <option>)* )?
 
-<option> ::= <property>
+<option> ::= <property> | INDEXES
 
-p. 
+p.
 __Sample:__
 
 bc(sample).. 
 CREATE TABLE newtb1 LIKE oldtb;
 
 CREATE TABLE newtb2 LIKE oldtb WITH compaction = { 'class' : 'LeveledCompactionStrategy' };
-p. 
+
+CREATE TABLE newtb4 LIKE oldtb WITH INDEXES AND compaction = { 'class' : 'LeveledCompactionStrategy' };
+
+p.
 The @COPY TABLE@ statement creates a new table which is a clone of old table. The new table have the same column numbers, column names, column data types, column data mask with the old table. The new table is defined by a "name":#copyNewTableName, and the name of the old table being cloned is defined by a "name":#copyOldTableName . The table options of the new table can be defined by setting "copyoptions":#copyTableOptions. Note that the @CREATE COLUMNFAMILY LIKE@ syntax is supported as an alias for @CREATE TABLE like@ (for historical reasons).
 
 Attempting to create an already existing table will return an error unless the @IF NOT EXISTS@ option is used. If it is used, the statement will be a no-op if the table already exists.
@@ -438,7 +441,7 @@ The old table name defines the already existed table.
 
 h4(#copyTableOptions). @<copyoptions>@
 
-The @COPY TABLE@ statement supports a number of options that controls the configuration of a new table. These options can be specified after the @WITH@ keyword, and all options are the same as those options when creating a table except for id .
+The @COPY TABLE@ statement supports a number of options that controls the configuration of a new table. These options can be specified after the @WITH@ keyword, and all options are the same as those options when creating a table except for id. Besides the options can also be specified with keyword INDEXES which means copy source table's indexes.
 
 h3(#alterTableStmt). ALTER TABLE
 

doc/modules/cassandra/examples/BNF/create_table_like.bnf

@@ -1,3 +1,3 @@
 create_table_statement::= CREATE TABLE [ IF NOT EXISTS ] new_table_name LIKE old_table_name
 	 [ WITH table_options ]
-table_options::= options [ AND table_options ]
+table_options::= INDEXES | options [ AND table_options ]

doc/modules/cassandra/examples/CQL/create_table_like.cql

@@ -12,3 +12,6 @@ CREATE TABLE newtb3 LIKE oldtb WITH compaction = { 'class' : 'LeveledCompactionS
                                AND compression = { 'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 32 }
                                AND cdc = true;
 
+CREATE TABLE newtb4 LIKE oldtb WITH INDEXES;
+
+CREATE TABLE newtb6 LIKE oldtb WITH INDEXES AND compaction = { 'class' : 'LeveledCompactionStrategy' };

pylib/cqlshlib/cql3handling.py

@@ -387,6 +387,9 @@ def dequote_value(cqlword):
                             ( ender="," [propmapkey]=<term> ":" [propmapval]=<term> )*
                       ender="}"
                     ;
+<propertyOrOption> ::= <property>
+                     | "INDEXES"
+                     ;
 
 '''
 
@@ -1314,7 +1317,7 @@ def create_cf_composite_primary_key_comma_completer(ctxt, cass):
 <copyTableStatement> ::= "CREATE" wat=("COLUMNFAMILY" | "TABLE" ) ("IF" "NOT" "EXISTS")?
                                 ( tks=<nonSystemKeyspaceName> dot="." )? tcf=<cfOrKsName>
                                 "LIKE" ( sks=<nonSystemKeyspaceName> dot="." )? scf=<cfOrKsName>
-                                ( "WITH" <property> ( "AND" <property> )* )?
+                                ( "WITH" <propertyOrOption> ( "AND" <propertyOrOption> )* )?
                             ;
 '''
 

pylib/cqlshlib/test/test_cqlsh_completion.py

@@ -796,9 +796,9 @@ def test_complete_in_create_table_like(self):
                             choices=['<new_table_name>'])
         self.trycompletions('CREATE TABLE ' + quoted_keyspace + '.new_table L',
                             immediate='IKE ')
-        self.trycompletions('CREATE TABLE ' + 'new_table LIKE old_table W',
+        self.trycompletions('CREATE TABLE new_table LIKE old_table W',
                             immediate='ITH ')
-        self.trycompletions('CREATE TABLE ' + 'new_table LIKE old_table WITH ',
+        self.trycompletions('CREATE TABLE new_table LIKE old_table WITH ',
                             choices=['allow_auto_snapshot',
                                      'bloom_filter_fp_chance', 'compaction',
                                      'compression',
@@ -808,23 +808,16 @@ def test_complete_in_create_table_like(self):
                                      'memtable',
                                      'memtable_flush_period_in_ms',
                                      'caching', 'comment',
-                                     'min_index_interval', 'speculative_retry', 'additional_write_policy', 'cdc', 'read_repair'])
-        self.trycompletions('CREATE TABLE ' + 'new_table LIKE old_table WITH ',
-                            choices=['allow_auto_snapshot',
-                                     'bloom_filter_fp_chance', 'compaction',
-                                     'compression',
-                                     'default_time_to_live', 'gc_grace_seconds',
-                                     'incremental_backups',
-                                     'max_index_interval',
-                                     'memtable',
-                                     'memtable_flush_period_in_ms',
-                                     'caching', 'comment',
-                                     'min_index_interval', 'speculative_retry', 'additional_write_policy', 'cdc', 'read_repair'])
+                                     'min_index_interval',
+                                     'speculative_retry', 'additional_write_policy',
+                                     'cdc', 'read_repair',
+                                     'INDEXES'])
+        self.trycompletions('CREATE TABLE new_table LIKE old_table WITH INDEXES ',
+                            choices=[';' , '=', 'AND'])
         self.trycompletions('CREATE TABLE ' + 'new_table LIKE old_table WITH bloom_filter_fp_chance ',
                             immediate='= ')
         self.trycompletions('CREATE TABLE ' + 'new_table LIKE old_table WITH bloom_filter_fp_chance = ',
                             choices=['<float_between_0_and_1>'])
-
         self.trycompletions('CREATE TABLE ' + 'new_table LIKE old_table WITH compaction ',
                             immediate="= {'class': '")
         self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
@@ -868,7 +861,10 @@ def test_complete_in_create_table_like(self):
                                      'memtable',
                                      'memtable_flush_period_in_ms',
                                      'caching', 'comment',
-                                     'min_index_interval', 'speculative_retry', 'additional_write_policy', 'cdc', 'read_repair'])
+                                     'min_index_interval',
+                                     'speculative_retry', 'additional_write_policy',
+                                     'cdc', 'read_repair',
+                                     'INDEXES'])
         self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
                             + "{'class': 'TimeWindowCompactionStrategy', '",
                             choices=['compaction_window_unit', 'compaction_window_size',

src/antlr/Lexer.g

@@ -98,6 +98,7 @@ K_TABLES:      ( C O L U M N F A M I L I E S
 K_MATERIALIZED:M A T E R I A L I Z E D;
 K_VIEW:        V I E W;
 K_INDEX:       I N D E X;
+K_INDEXES:     I N D E X E S;
 K_CUSTOM:      C U S T O M;
 K_ON:          O N;
 K_TO:          T O;

src/antlr/Parser.g

@@ -836,7 +836,16 @@ copyTableStatement returns  [CopyTableStatement.Raw stmt]
     : K_CREATE K_COLUMNFAMILY (K_IF K_NOT K_EXISTS { ifNotExists = true; } )?
       newCf=columnFamilyName K_LIKE oldCf=columnFamilyName
       { $stmt = new CopyTableStatement.Raw(newCf, oldCf, ifNotExists); }
-      ( K_WITH property[stmt.attrs] ( K_AND property[stmt.attrs] )*)?
+      ( K_WITH propertyOrOption[stmt] ( K_AND propertyOrOption[stmt] )*)?
+    ;
+
+propertyOrOption[CopyTableStatement.Raw stmt]
+    : tableLikeSingleOption[stmt]
+    | property[stmt.attrs]
+    ;
+
+tableLikeSingleOption[CopyTableStatement.Raw stmt]
+    : K_INDEXES {$stmt.withLikeOption(CopyTableStatement.CreateLikeOption.INDEXES);}
     ;
 
 /**
@@ -2084,5 +2093,6 @@ basic_unreserved_keyword returns [String str]
         | K_ANN
         | K_BETWEEN
         | K_CHECK
+        | K_INDEXES
         ) { $str = $k.text; }
     ;

src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java

@@ -19,6 +19,8 @@
 package org.apache.cassandra.cql3.statements.schema;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -35,6 +37,9 @@
 import org.apache.cassandra.db.marshal.UserType;
 import org.apache.cassandra.db.marshal.VectorType;
 import org.apache.cassandra.exceptions.AlreadyExistsException;
+import org.apache.cassandra.index.sai.StorageAttachedIndex;
+import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.Indexes;
 import org.apache.cassandra.schema.KeyspaceMetadata;
 import org.apache.cassandra.schema.Keyspaces;
@@ -46,6 +51,7 @@
 import org.apache.cassandra.schema.Triggers;
 import org.apache.cassandra.schema.UserFunctions;
 import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.service.ClientWarn;
 import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.transport.Event.SchemaChange;
@@ -61,12 +67,14 @@ public final class CopyTableStatement extends AlterSchemaStatement
     private final String targetTableName;
     private final boolean ifNotExists;
     private final TableAttributes attrs;
+    private final CreateLikeOption createLikeOption;
 
     public CopyTableStatement(String sourceKeyspace,
                               String targetKeyspace,
                               String sourceTableName,
                               String targetTableName,
                               boolean ifNotExists,
+                              CreateLikeOption createLikeOption,
                               TableAttributes attrs)
     {
         super(targetKeyspace);
@@ -75,6 +83,7 @@ public CopyTableStatement(String sourceKeyspace,
         this.sourceTableName = sourceTableName;
         this.targetTableName = targetTableName;
         this.ifNotExists = ifNotExists;
+        this.createLikeOption = createLikeOption;
         this.attrs = attrs;
     }
 
@@ -196,6 +205,7 @@ public Keyspaces apply(ClusterMetadata metadata)
 
         TableParams originalParams = targetBuilder.build().params;
         TableParams newTableParams = attrs.asAlteredTableParams(originalParams);
+        maybeCopyIndexes(targetBuilder, sourceTableMeta, targetKeyspaceMeta);
 
         TableMetadata table = targetBuilder.params(newTableParams)
                                            .id(TableId.get(metadata))
@@ -229,12 +239,59 @@ public void validate(ClientState state)
         validateDefaultTimeToLive(attrs.asNewTableParams());
     }
 
+    private void maybeCopyIndexes(TableMetadata.Builder builder, TableMetadata sourceTableMeta, KeyspaceMetadata targetKeyspaceMeta)
+    {
+        if (createLikeOption != CreateLikeOption.INDEXES || sourceTableMeta.indexes.isEmpty())
+            return;
+
+        Set<String> customIndexes = Sets.newTreeSet();
+        List<IndexMetadata> indexesToCopy = new ArrayList<>();
+        for (IndexMetadata indexMetadata : sourceTableMeta.indexes)
+        {
+            // only sai and legacy secondary index is supported
+            if (indexMetadata.isCustom() && !StorageAttachedIndex.class.getCanonicalName().equals(indexMetadata.getIndexClassName()))
+            {
+                customIndexes.add(indexMetadata.name);
+                continue;
+            }
+
+            ColumnMetadata targetColumn = sourceTableMeta.getColumn(UTF8Type.instance.decompose(indexMetadata.options.get("target")));
+            String indexName;
+            // The rules for generating the index names of the target table are:
+            // (1) If the source table's index names follow the pattern sourcetablename_columnname_idx_number, the index names are considered to be generated by the system,
+            //     then we directly replace the name of source table with the name of target table, and increment the number after idx to avoid index name conflicts.
+            // (2) Index names that do not follow the above pattern are considered user-defined, so the index names are retained and increment the number after idx to avoid conflicts.
+            if (indexMetadata.name.startsWith(sourceTableName + "_" + targetColumn.name + "_idx"))
+            {
+                String baseName = IndexMetadata.generateDefaultIndexName(targetTableName, targetColumn.name);
+                indexName = targetKeyspaceMeta.findAvailableIndexName(baseName, indexesToCopy, targetKeyspaceMeta);
+            }
+            else
+            {
+                indexName = targetKeyspaceMeta.findAvailableIndexName(indexMetadata.name, indexesToCopy, targetKeyspaceMeta);
+            }
+            indexesToCopy.add(IndexMetadata.fromSchemaMetadata(indexName, indexMetadata.kind, indexMetadata.options));
+        }
+
+        if (!indexesToCopy.isEmpty())
+            builder.indexes(Indexes.builder().add(indexesToCopy).build());
+
+        if (!customIndexes.isEmpty())
+            ClientWarn.instance.warn(String.format("Source table %s.%s to copy indexes from to %s.%s has custom indexes. These indexes were not copied: %s",
+                                                   sourceKeyspace,
+                                                   sourceTableName,
+                                                   targetKeyspace,
+                                                   targetTableName,
+                                                   customIndexes));
+    }
+
     public final static class Raw extends CQLStatement.Raw
     {
         private final QualifiedName oldName;
         private final QualifiedName newName;
         private final boolean ifNotExists;
         public final TableAttributes attrs = new TableAttributes();
+        private CreateLikeOption createLikeOption = null;
 
         public Raw(QualifiedName newName, QualifiedName oldName, boolean ifNotExists)
         {
@@ -248,7 +305,17 @@ public CQLStatement prepare(ClientState state)
         {
             String oldKeyspace = oldName.hasKeyspace() ? oldName.getKeyspace() : state.getKeyspace();
             String newKeyspace = newName.hasKeyspace() ? newName.getKeyspace() : state.getKeyspace();
-            return new CopyTableStatement(oldKeyspace, newKeyspace, oldName.getName(), newName.getName(), ifNotExists, attrs);
+            return new CopyTableStatement(oldKeyspace, newKeyspace, oldName.getName(), newName.getName(), ifNotExists, createLikeOption, attrs);
+        }
+
+        public void withLikeOption(CreateLikeOption option)
+        {
+            this.createLikeOption = option;
         }
     }
+
+    public enum CreateLikeOption
+    {
+        INDEXES;
+    }
 }

src/java/org/apache/cassandra/schema/KeyspaceMetadata.java

@@ -18,6 +18,8 @@
 package org.apache.cassandra.schema;
 
 import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
@@ -207,19 +209,49 @@ public Stream<TableMetadata> tablesUsingFunction(Function function)
 
     public String findAvailableIndexName(String baseName)
     {
-        if (!hasIndex(baseName))
+        return findAvailableIndexName(baseName, Collections.emptySet(), this);
+    }
+
+    /**
+     * find an avaiable index name based on the indexes in target keyspace and indexes collections
+     * @param baseName the base name of index
+     * @param indexes find out whether there is any conflict with baseName in the indexes
+     * @param keyspaceMetadata find out whether there is any conflict with baseName in keyspaceMetadata
+     * */
+    public String findAvailableIndexName(String baseName, Collection<IndexMetadata> indexes, KeyspaceMetadata keyspaceMetadata)
+    {
+        if (!hasIndex(baseName, indexes, keyspaceMetadata))
             return baseName;
 
-        int i = 1;
         do
         {
-            String name = baseName + '_' + i++;
-            if (!hasIndex(name))
+            String name = generateIndexName(baseName);
+            if (!hasIndex(name, indexes, keyspaceMetadata))
                 return name;
+            baseName = name;
         }
         while (true);
     }
 
+    private String generateIndexName(String baseName)
+    {
+        if (baseName.matches(".*_\\d+$"))
+        {
+            int lastUnderscoreIndex = baseName.lastIndexOf('_');
+            String numberStr = baseName.substring(lastUnderscoreIndex + 1);
+            int number = Integer.parseInt(numberStr) + 1;
+            return baseName.substring(0, lastUnderscoreIndex + 1) + number;
+        }
+
+        return baseName + "_1";
+    }
+
+    private boolean hasIndex(String baseName, Collection<IndexMetadata> indexes, KeyspaceMetadata keyspaceMetadata)
+    {
+        return any(indexes, t -> t.name.equals(baseName)) ||
+               any(keyspaceMetadata.tables, t -> t.indexes.has(baseName));
+    }
+
     public Optional<TableMetadata> findIndexedTable(String indexName)
     {
         for (TableMetadata table : tablesAndViews())

test/unit/org/apache/cassandra/cql3/AlterSchemaStatementTest.java

@@ -40,7 +40,8 @@ public class AlterSchemaStatementTest extends CQLTester
                      "CREATE TABLE ks.t1 (k int PRIMARY KEY)",
                      "ALTER MATERIALIZED VIEW ks.v1 WITH compaction = { 'class' : 'LeveledCompactionStrategy' }",
                      "ALTER TABLE ks.t1 ADD v int",
-                     "CREATE TABLE ks.tb like ks1.tb"
+                     "CREATE TABLE ks.tb like ks1.tb",
+                     "CREATE TABLE ks.tb like ks1.tb WITH indexes"
                      };
     private final ClientState clientState = ClientState.forExternalCalls(InetSocketAddress.createUnresolved("127.0.0.1", 1234));