From 989f0414b72db77e0897042a2e5b06a601032d8c Mon Sep 17 00:00:00 2001
From: Maxwell Guo <cclive1601@gmail.com>
Date: Sat, 25 Jan 2025 00:27:51 +0800
Subject: [PATCH] Support CREATE TABLE LIKE with INDEXES

patch by Maxwell Guo; reviewed by Stefan Miklosovic, Sam Tunnicliffe for CASSANDRA-19965
---
 CHANGES.txt                                   |   1 +
 doc/cql3/CQL.textile                          |  11 +-
 .../examples/BNF/create_table_like.bnf        |   2 +-
 .../examples/CQL/create_table_like.cql        |   3 +
 pylib/cqlshlib/cql3handling.py                |   5 +-
 pylib/cqlshlib/test/test_cqlsh_completion.py  |  28 +-
 src/antlr/Lexer.g                             |   1 +
 src/antlr/Parser.g                            |  12 +-
 .../statements/schema/CopyTableStatement.java |  69 +++-
 .../cassandra/schema/KeyspaceMetadata.java    |  40 +-
 .../cql3/AlterSchemaStatementTest.java        |   3 +-
 .../org/apache/cassandra/cql3/CQLTester.java  |  32 +-
 .../schema/createlike/CreateLikeTest.java     | 390 +++++++++++-------
 13 files changed, 424 insertions(+), 173 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index b704cca5d4..269b8a8f45 100644
--- a/CHANGES.txt
+++ b/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)
diff --git a/doc/cql3/CQL.textile b/doc/cql3/CQL.textile
index f819ec193d..87812fc19b 100644
--- a/doc/cql3/CQL.textile
+++ b/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
 
diff --git a/doc/modules/cassandra/examples/BNF/create_table_like.bnf b/doc/modules/cassandra/examples/BNF/create_table_like.bnf
index 56d209c6ef..ebbd9b4658 100644
--- a/doc/modules/cassandra/examples/BNF/create_table_like.bnf
+++ b/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 ]
\ No newline at end of file
+table_options::= INDEXES | options [ AND table_options ]
\ No newline at end of file
diff --git a/doc/modules/cassandra/examples/CQL/create_table_like.cql b/doc/modules/cassandra/examples/CQL/create_table_like.cql
index ef65b8b050..f47fb191f0 100644
--- a/doc/modules/cassandra/examples/CQL/create_table_like.cql
+++ b/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' };
\ No newline at end of file
diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py
index 5c48f8e600..e231d5c8a8 100644
--- a/pylib/cqlshlib/cql3handling.py
+++ b/pylib/cqlshlib/cql3handling.py
@@ -387,6 +387,9 @@ JUNK ::= /([ \t\r\f\v]+|(--|[/][/])[^\n\r]*([\n\r]|$)|[/][*].*?[*][/])/ ;
                             ( ender="," [propmapkey]=<term> ":" [propmapval]=<term> )*
                       ender="}"
                     ;
+<propertyOrOption> ::= <property>
+                     | "INDEXES"
+                     ;
 
 '''
 
@@ -1314,7 +1317,7 @@ syntax_rules += r'''
 <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> )* )?
                             ;
 '''
 
diff --git a/pylib/cqlshlib/test/test_cqlsh_completion.py b/pylib/cqlshlib/test/test_cqlsh_completion.py
index da362c50f4..55f5511ef1 100644
--- a/pylib/cqlshlib/test/test_cqlsh_completion.py
+++ b/pylib/cqlshlib/test/test_cqlsh_completion.py
@@ -796,9 +796,9 @@ class TestCqlshCompletion(CqlshCompletionCase):
                             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 @@ class TestCqlshCompletion(CqlshCompletionCase):
                                      '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 @@ class TestCqlshCompletion(CqlshCompletionCase):
                                      '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',
diff --git a/src/antlr/Lexer.g b/src/antlr/Lexer.g
index 96a51e23c8..b192021fa3 100644
--- a/src/antlr/Lexer.g
+++ b/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;
diff --git a/src/antlr/Parser.g b/src/antlr/Parser.g
index 682ba873e4..30e07fddf9 100644
--- a/src/antlr/Parser.g
+++ b/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; }
     ;
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java
index e8ee7e2cd3..3a79311dc7 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java
+++ b/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.UTF8Type;
 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.TableParams;
 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 final class CopyTableStatement extends AlterSchemaStatement
         this.sourceTableName = sourceTableName;
         this.targetTableName = targetTableName;
         this.ifNotExists = ifNotExists;
+        this.createLikeOption = createLikeOption;
         this.attrs = attrs;
     }
 
@@ -196,6 +205,7 @@ public final class CopyTableStatement extends AlterSchemaStatement
 
         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 final class CopyTableStatement extends AlterSchemaStatement
         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 final class CopyTableStatement extends AlterSchemaStatement
         {
             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;
+    }
 }
diff --git a/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java b/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java
index aec93c9f47..8065c59290 100644
--- a/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java
+++ b/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 final class KeyspaceMetadata implements SchemaElement
 
     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())
diff --git a/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementTest.java b/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementTest.java
index 57d7d2b5a5..6d58e7bcee 100644
--- a/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementTest.java
+++ b/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));
 
diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java
index 59ab7be83f..22ab64c12f 100644
--- a/test/unit/org/apache/cassandra/cql3/CQLTester.java
+++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java
@@ -1095,7 +1095,7 @@ public abstract class CQLTester
         }
 
         String currentTable = createTableName(targetTable);
-        String fullQuery = currentTable == null ? query : String.format(query, targetKeyspace + "." + currentTable, sourceKeyspace + "." + sourceTable);;
+        String fullQuery = currentTable == null ? query : String.format(query, targetKeyspace + "." + currentTable, sourceKeyspace + "." + sourceTable);
         logger.info(fullQuery);
         schemaChange(fullQuery);
         return currentTable;
@@ -1972,21 +1972,41 @@ public abstract class CQLTester
      * Determine whether the source and target TableMetadata is equal without compare the table name and dropped columns.
      * @param source the source TableMetadata
      * @param target the target TableMetadata
-     * @param compareParams wether compare table params
+     * @param compareParams wether compare table's params
      * @param compareIndexes wether compare table's indexes
-     * @param compareTrigger wether compare table's triggers
+     * @param compareIndexWithOutName wether ignore indexes' name when doing index comparison
+     *                                if true then compare the index without name
      * */
-    protected boolean equalsWithoutTableNameAndDropCns(TableMetadata source, TableMetadata target, boolean compareParams, boolean compareIndexes, boolean compareTrigger)
+    protected boolean equalsWithoutTableNameAndDropCns(TableMetadata source, TableMetadata target, boolean compareParams, boolean compareIndexes, boolean compareIndexWithOutName)
     {
         return source.partitioner.equals(target.partitioner)
                && source.kind == target.kind
                && source.flags.equals(target.flags)
                && (!compareParams || source.params.equals(target.params))
-               && (!compareIndexes || source.indexes.equals(target.indexes))
-               && (!compareTrigger || source.triggers.equals(target.triggers))
+               && (!compareIndexes || compareIndexes(source, target, compareIndexWithOutName))
                && columnsEqualWitoutKsTb(source, target);
     }
 
+    private boolean compareIndexes(TableMetadata source, TableMetadata target, boolean compareIndexWithOutName)
+    {
+        if (compareIndexWithOutName)
+        {
+            if (source.indexes.size() != target.indexes.size())
+                return false;
+            Iterator<IndexMetadata> leftIter = source.indexes.stream().sorted(Comparator.comparing(idx -> idx.name)).iterator();
+            Iterator<IndexMetadata> rightIter = target.indexes.stream().sorted(Comparator.comparing(idx -> idx.name)).iterator();
+            boolean result = true;
+            while (leftIter.hasNext() && rightIter.hasNext())
+            {
+                IndexMetadata left = leftIter.next();
+                IndexMetadata right = rightIter.next();
+                result &= right.equalsWithoutName(left);
+            }
+            return result;
+        }
+        return source.indexes.equals(target.indexes);
+    }
+
     // only compare columns
     private boolean columnsEqualWitoutKsTb(TableMetadata source, TableMetadata target)
     {
diff --git a/test/unit/org/apache/cassandra/schema/createlike/CreateLikeTest.java b/test/unit/org/apache/cassandra/schema/createlike/CreateLikeTest.java
index 1fe72121a6..291a050464 100644
--- a/test/unit/org/apache/cassandra/schema/createlike/CreateLikeTest.java
+++ b/test/unit/org/apache/cassandra/schema/createlike/CreateLikeTest.java
@@ -20,11 +20,13 @@ package org.apache.cassandra.schema.createlike;
 
 import java.math.BigDecimal;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
@@ -34,14 +36,16 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import com.datastax.driver.core.ResultSet;
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.cql3.Duration;
 import org.apache.cassandra.cql3.validation.operations.CreateTest;
-import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.compaction.LeveledCompactionStrategy;
 import org.apache.cassandra.exceptions.AlreadyExistsException;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.index.StubIndex;
 import org.apache.cassandra.schema.CachingParams;
 import org.apache.cassandra.schema.CompactionParams;
 import org.apache.cassandra.schema.CompressionParams;
@@ -54,6 +58,7 @@ import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
 import org.apache.cassandra.utils.TimeUUID;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
@@ -68,28 +73,27 @@ public class CreateLikeTest extends CQLTester
     public static Collection<Object[]> data()
     {
         List<Object[]> result = new ArrayList<>();
-        result.add(new Object[]{false});
-        result.add(new Object[]{true});
+        result.add(new Object[]{ false });
+        result.add(new Object[]{ true });
         return result;
     }
 
-    private UUID uuid1 = UUID.fromString("62c3e96f-55cd-493b-8c8e-5a18883a1698");
-    private UUID uuid2 = UUID.fromString("52c3e96f-55cd-493b-8c8e-5a18883a1698");
-    private TimeUUID timeUuid1 = TimeUUID.fromString("00346642-2d2f-11ed-a261-0242ac120002");
-    private TimeUUID timeUuid2 = TimeUUID.fromString("10346642-2d2f-11ed-a261-0242ac120002");
-    private Duration duration1 = Duration.newInstance(1, 2, 3);
-    private Duration duration2 = Duration.newInstance(1, 2, 4);
-    private Date date1 = new Date();
-    private Double d1 = Double.valueOf("1.1");
-    private Double d2 = Double.valueOf("2.2");
-    private Float f1 = Float.valueOf("3.33");
-    private Float f2 = Float.valueOf("4.44");
-    private BigDecimal decimal1 =  BigDecimal.valueOf(1.1);
-    private BigDecimal decimal2 =  BigDecimal.valueOf(2.2);
-    private Vector<Integer> vector1  = vector(1, 2);
-    private Vector<Integer> vector2 = vector(3, 4);
-    private String keyspace1 = "keyspace1";
-    private String keyspace2 = "keyspace2";
+    private final UUID uuid1 = UUID.fromString("62c3e96f-55cd-493b-8c8e-5a18883a1698");
+    private final UUID uuid2 = UUID.fromString("52c3e96f-55cd-493b-8c8e-5a18883a1698");
+    private final TimeUUID timeUuid1 = TimeUUID.fromString("00346642-2d2f-11ed-a261-0242ac120002");
+    private final TimeUUID timeUuid2 = TimeUUID.fromString("10346642-2d2f-11ed-a261-0242ac120002");
+    private final Duration duration1 = Duration.newInstance(1, 2, 3);
+    private final Duration duration2 = Duration.newInstance(1, 2, 4);
+    private final Double d1 = Double.valueOf("1.1");
+    private final Double d2 = Double.valueOf("2.2");
+    private final Float f1 = Float.valueOf("3.33");
+    private final Float f2 = Float.valueOf("4.44");
+    private final BigDecimal decimal1 = BigDecimal.valueOf(1.1);
+    private final BigDecimal decimal2 = BigDecimal.valueOf(2.2);
+    private final Vector<Integer> vector1 = vector(1, 2);
+    private final Vector<Integer> vector2 = vector(3, 4);
+    private final String keyspace1 = "keyspace1";
+    private final String keyspace2 = "keyspace2";
     private String sourceKs;
     private String targetKs;
 
@@ -139,38 +143,38 @@ public class CreateLikeTest extends CQLTester
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
         execute("INSERT INTO " + sourceKs + "." + sourceTb + " (a, b, c, d, e, f) VALUES (?, ?, ?, ?, ?, ?)",
-                1, set(list("1", "2"), list("3", "4")), map("k", 1), (short)2, duration1, (byte)4);
+                1, set(list("1", "2"), list("3", "4")), map("k", 1), (short) 2, duration1, (byte) 4);
         execute("INSERT INTO " + targetKs + "." + targetTb + " (a, b, c, d, e, f) VALUES (?, ?, ?, ?, ?, ?)",
-                2, set(list("5", "6"), list("7", "8")), map("nk", 2), (short)3, duration2, (byte)5);
+                2, set(list("5", "6"), list("7", "8")), map("nk", 2), (short) 3, duration2, (byte) 5);
         assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
-                   row(1, set(list("1", "2"), list("3", "4")), map("k", 1), (short)2, duration1, (byte)4));
+                   row(1, set(list("1", "2"), list("3", "4")), map("k", 1), (short) 2, duration1, (byte) 4));
         assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
-                   row(2, set(list("5", "6"), list("7", "8")), map("nk", 2), (short)3, duration2, (byte)5));
+                   row(2, set(list("5", "6"), list("7", "8")), map("nk", 2), (short) 3, duration2, (byte) 5));
 
         sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int , b double, c tinyint, d float, e list<text>, f map<text, int>, g duration, PRIMARY KEY((a, b, c), d));");
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
         execute("INSERT INTO " + sourceKs + "." + sourceTb + " (a, b, c, d, e, f, g) VALUES (?, ?, ?, ?, ?, ?, ?) ",
-                1, d1, (byte)4, f1, list("a", "b"), map("k", 1), duration1);
+                1, d1, (byte) 4, f1, list("a", "b"), map("k", 1), duration1);
         execute("INSERT INTO " + targetKs + "." + targetTb + " (a, b, c, d, e, f, g) VALUES (?, ?, ?, ?, ?, ?, ?) ",
-                2, d2, (byte)5, f2, list("c", "d"), map("nk", 2), duration2);
+                2, d2, (byte) 5, f2, list("c", "d"), map("nk", 2), duration2);
         assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
-                   row(1, d1, (byte)4, f1, list("a", "b"), map("k", 1), duration1));
+                   row(1, d1, (byte) 4, f1, list("a", "b"), map("k", 1), duration1));
         assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
-                   row(2, d2, (byte)5, f2, list("c", "d"), map("nk", 2), duration2));
+                   row(2, d2, (byte) 5, f2, list("c", "d"), map("nk", 2), duration2));
 
         sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int , " +
-                                                                "b text, " +
-                                                                "c bigint, " +
-                                                                "d decimal, " +
-                                                                "e set<text>, " +
-                                                                "f uuid, " +
-                                                                "g vector<int, 2>, " +
-                                                                "h list<float>, " +
-                                                                "i timeuuid, " +
-                                                                "j map<text, frozen<set<int>>>, " +
-                                                                "PRIMARY KEY((a, b), c, d)) " +
-                                                                "WITH CLUSTERING ORDER BY (c DESC, d ASC);");
+                                         "b text, " +
+                                         "c bigint, " +
+                                         "d decimal, " +
+                                         "e set<text>, " +
+                                         "f uuid, " +
+                                         "g vector<int, 2>, " +
+                                         "h list<float>, " +
+                                         "i timeuuid, " +
+                                         "j map<text, frozen<set<int>>>, " +
+                                         "PRIMARY KEY((a, b), c, d)) " +
+                                         "WITH CLUSTERING ORDER BY (c DESC, d ASC);");
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
         execute("INSERT INTO " + sourceKs + "." + sourceTb + " (a, b, c, d, e, f, g, h, i, j)  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
@@ -207,15 +211,15 @@ public class CreateLikeTest extends CQLTester
         String targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
 
-        alterTable("ALTER TABLE  " + sourceKs + " ." + sourceTb + " DROP d");
+        alterTable("ALTER TABLE " + sourceKs + "." + sourceTb + " DROP d");
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
 
-        alterTable("ALTER TABLE " + sourceKs + " ." + sourceTb + " ADD e uuid");
+        alterTable("ALTER TABLE " + sourceKs + "." + sourceTb + " ADD e uuid");
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
 
-        alterTable("ALTER TABLE " + sourceKs + " ." + sourceTb + " ADD f float");
+        alterTable("ALTER TABLE " + sourceKs + "." + sourceTb + " ADD f float");
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
 
@@ -226,15 +230,15 @@ public class CreateLikeTest extends CQLTester
         assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
                    row(2, "2", duration2, uuid2, f2));
 
-        alterTable("ALTER TABLE " + sourceKs + " ." + sourceTb + " DROP f USING TIMESTAMP 20000");
+        alterTable("ALTER TABLE " + sourceKs + "." + sourceTb + " DROP f USING TIMESTAMP 20000");
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
 
-        alterTable("ALTER TABLE " + sourceKs + " ." + sourceTb + " RENAME b TO bb ");
+        alterTable("ALTER TABLE " + sourceKs + "." + sourceTb + " RENAME b TO bb ");
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
 
-        alterTable("ALTER TABLE " + sourceKs + " ." + sourceTb + " WITH compaction = {'class':'LeveledCompactionStrategy', 'sstable_size_in_mb':10, 'fanout_size':16} ");
+        alterTable("ALTER TABLE " + sourceKs + "." + sourceTb + " WITH compaction = {'class':'LeveledCompactionStrategy', 'sstable_size_in_mb' : 10, 'fanout_size' : 16} ");
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
 
@@ -244,7 +248,6 @@ public class CreateLikeTest extends CQLTester
                    row(1, "1", duration1, uuid1));
         assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
                    row(2, "2", duration2, uuid2));
-
     }
 
     @Test
@@ -252,52 +255,52 @@ public class CreateLikeTest extends CQLTester
     {
         // compression
         String tbCompressionDefault1 = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))");
-        String tbCompressionDefault2 =createTable(sourceKs,"CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
-                                                           " WITH compression = { 'enabled' : 'false'};");
+        String tbCompressionDefault2 = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
+                                                             " WITH compression = { 'enabled' : 'false'};");
         String tbCompressionSnappy1 = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
-                                                           " WITH compression = { 'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 32 };");
-        String tbCompressionSnappy2 =createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
-                                                           " WITH compression = { 'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 32, 'enabled' : true };");
-        String tbCompressionSnappy3 = createTable(sourceKs,"CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
-                                                           " WITH compression = { 'class' : 'SnappyCompressor', 'min_compress_ratio' : 2 };");
-        String tbCompressionSnappy4 = createTable(sourceKs,"CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
-                                                           " WITH compression = { 'class' : 'SnappyCompressor', 'min_compress_ratio' : 1 };");
-        String tbCompressionSnappy5 = createTable(sourceKs,"CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
-                                                           " WITH compression = { 'class' : 'SnappyCompressor', 'min_compress_ratio' : 0 };");
+                                                            " WITH compression = { 'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 32 };");
+        String tbCompressionSnappy2 = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
+                                                            " WITH compression = { 'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 32, 'enabled' : true };");
+        String tbCompressionSnappy3 = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
+                                                            " WITH compression = { 'class' : 'SnappyCompressor', 'min_compress_ratio' : 2 };");
+        String tbCompressionSnappy4 = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
+                                                            " WITH compression = { 'class' : 'SnappyCompressor', 'min_compress_ratio' : 1 };");
+        String tbCompressionSnappy5 = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
+                                                            " WITH compression = { 'class' : 'SnappyCompressor', 'min_compress_ratio' : 0 };");
 
         // memtable
-        String tableMemtableSkipList = createTable(sourceKs,"CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
-                                                            " WITH memtable = 'skiplist';");
-        String tableMemtableTrie = createTable(sourceKs,"CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
-                                                        " WITH memtable = 'trie';");
-        String tableMemtableDefault = createTable(sourceKs,"CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
-                                                           " WITH memtable = 'default';");
+        String tableMemtableSkipList = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
+                                                             " WITH memtable = 'skiplist';");
+        String tableMemtableTrie = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
+                                                         " WITH memtable = 'trie';");
+        String tableMemtableDefault = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))" +
+                                                            " WITH memtable = 'default';");
 
         // compaction
-        String tableCompactionStcs = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))  WITH compaction = {'class':'SizeTieredCompactionStrategy', 'min_threshold':2, 'enabled':false};");
-        String tableCompactionLcs =createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b)) WITH compaction = {'class':'LeveledCompactionStrategy', 'sstable_size_in_mb':1, 'fanout_size':5};");
-        String tableCompactionTwcs = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b)) WITH compaction = {'class':'TimeWindowCompactionStrategy', 'min_threshold':2};");
-        String tableCompactionUcs =createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b)) WITH compaction = {'class':'UnifiedCompactionStrategy'};");
+        String tableCompactionStcs = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b)) WITH compaction = {'class' : 'SizeTieredCompactionStrategy', 'min_threshold' : 2, 'enabled' : false};");
+        String tableCompactionLcs = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b)) WITH compaction = {'class' : 'LeveledCompactionStrategy', 'sstable_size_in_mb' : 1, 'fanout_size' : 5};");
+        String tableCompactionTwcs = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b)) WITH compaction = {'class' : 'TimeWindowCompactionStrategy', 'min_threshold' : 2};");
+        String tableCompactionUcs = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b)) WITH compaction = {'class' : 'UnifiedCompactionStrategy'};");
 
         // other options are all different from default
         String tableOtherOptions = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b)) WITH" +
-                                                        " additional_write_policy = '95p' " +
-                                                        " AND bloom_filter_fp_chance = 0.1 " +
-                                                        " AND caching = {'keys': 'ALL', 'rows_per_partition': '100'}" +
-                                                        " AND cdc = true " +
-                                                        " AND comment = 'test for create like'" +
-                                                        " AND crc_check_chance = 0.1" +
-                                                        " AND default_time_to_live = 10" +
-                                                        " AND compaction = {'class':'UnifiedCompactionStrategy'} " +
-                                                        " AND compression = { 'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 32 }" +
-                                                        " AND gc_grace_seconds = 100" +
-                                                        " AND incremental_backups = false" +
-                                                        " AND max_index_interval = 1024" +
-                                                        " AND min_index_interval = 64" +
-                                                        " AND speculative_retry = '95p'" +
-                                                        " AND read_repair = 'NONE'" +
-                                                        " AND memtable_flush_period_in_ms = 360000" +
-                                                        " AND memtable = 'default';" );
+                                                         " additional_write_policy = '95p' " +
+                                                         " AND bloom_filter_fp_chance = 0.1 " +
+                                                         " AND caching = {'keys' : 'ALL', 'rows_per_partition' : '100'}" +
+                                                         " AND cdc = true " +
+                                                         " AND comment = 'test for create like'" +
+                                                         " AND crc_check_chance = 0.1" +
+                                                         " AND default_time_to_live = 10" +
+                                                         " AND compaction = {'class' : 'UnifiedCompactionStrategy'} " +
+                                                         " AND compression = {'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 32 }" +
+                                                         " AND gc_grace_seconds = 100" +
+                                                         " AND incremental_backups = false" +
+                                                         " AND max_index_interval = 1024" +
+                                                         " AND min_index_interval = 64" +
+                                                         " AND speculative_retry = '95p'" +
+                                                         " AND read_repair = 'NONE'" +
+                                                         " AND memtable_flush_period_in_ms = 360000" +
+                                                         " AND memtable = 'default';");
 
         String tbLikeCompressionDefault1 = createTableLike("CREATE TABLE %s LIKE %s", tbCompressionDefault1, sourceKs, targetKs);
         String tbLikeCompressionDefault2 = createTableLike("CREATE TABLE %s LIKE %s", tbCompressionDefault2, sourceKs, targetKs);
@@ -313,7 +316,7 @@ public class CreateLikeTest extends CQLTester
         String tbLikeCompactionLcs = createTableLike("CREATE TABLE %s LIKE %s", tableCompactionLcs, sourceKs, targetKs);
         String tbLikeCompactionTwcs = createTableLike("CREATE TABLE %s LIKE %s", tableCompactionTwcs, sourceKs, targetKs);
         String tbLikeCompactionUcs = createTableLike("CREATE TABLE %s LIKE %s", tableCompactionUcs, sourceKs, targetKs);
-        String tbLikeCompactionOthers= createTableLike("CREATE TABLE %s LIKE %s", tableOtherOptions, sourceKs, targetKs);
+        String tbLikeCompactionOthers = createTableLike("CREATE TABLE %s LIKE %s", tableOtherOptions, sourceKs, targetKs);
 
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, tbCompressionDefault1, tbLikeCompressionDefault1);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, tbCompressionDefault2, tbLikeCompressionDefault2);
@@ -332,27 +335,28 @@ public class CreateLikeTest extends CQLTester
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, tableOtherOptions, tbLikeCompactionOthers);
 
         // a copy of the table with the table parameters set
-        String tableCopyAndSetCompression = createTableLike("CREATE TABLE %s LIKE %s WITH compression = { 'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 64 };",
+        String tableCopyAndSetCompression = createTableLike("CREATE TABLE %s LIKE %s WITH compression = {'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 64 };",
                                                             tbCompressionSnappy1, sourceKs, targetKs);
-        String tableCopyAndSetLCSCompaction = createTableLike("CREATE TABLE %s LIKE %s WITH compaction = {'class':'LeveledCompactionStrategy', 'sstable_size_in_mb':10, 'fanout_size':16};",
-                                                           tableCompactionLcs, sourceKs, targetKs);
+        String tableCopyAndSetLCSCompaction = createTableLike("CREATE TABLE %s LIKE %s WITH compaction = {'class' : 'LeveledCompactionStrategy', 'sstable_size_in_mb' : 10, 'fanout_size' : 16};",
+                                                              tableCompactionLcs, sourceKs, targetKs);
         String tableCopyAndSetAllParams = createTableLike("CREATE TABLE %s (a text, b int, c int, primary key (a, b)) WITH" +
-                                                           " bloom_filter_fp_chance = 0.75 " +
-                                                           " AND caching = {'keys': 'NONE', 'rows_per_partition': '10'}" +
-                                                           " AND cdc = true " +
-                                                           " AND comment = 'test for create like and set params'" +
-                                                           " AND crc_check_chance = 0.8" +
-                                                           " AND default_time_to_live = 100" +
-                                                           " AND compaction = {'class':'SizeTieredCompactionStrategy'} " +
-                                                           " AND compression = { 'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 64 }" +
-                                                           " AND gc_grace_seconds = 1000" +
-                                                           " AND incremental_backups = true" +
-                                                           " AND max_index_interval = 128" +
-                                                           " AND min_index_interval = 16" +
-                                                           " AND speculative_retry = '96p'" +
-                                                           " AND read_repair = 'NONE'" +
-                                                           " AND memtable_flush_period_in_ms = 3600;",
-                                                           tableOtherOptions, sourceKs, targetKs);
+                                                          " bloom_filter_fp_chance = 0.75 " +
+                                                          " AND caching = {'keys' : 'NONE', 'rows_per_partition' : '10'}" +
+                                                          " AND cdc = true " +
+                                                          " AND comment = 'test for create like and set params'" +
+                                                          " AND crc_check_chance = 0.8" +
+                                                          " AND default_time_to_live = 100" +
+                                                          " AND compaction = {'class' : 'SizeTieredCompactionStrategy'} " +
+                                                          " AND compression = {'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 64}" +
+                                                          " AND gc_grace_seconds = 1000" +
+                                                          " AND incremental_backups = true" +
+                                                          " AND max_index_interval = 128" +
+                                                          " AND min_index_interval = 16" +
+                                                          " AND speculative_retry = '96p'" +
+                                                          " AND read_repair = 'NONE'" +
+                                                          " AND memtable_flush_period_in_ms = 3600;",
+                                                          tableOtherOptions, sourceKs, targetKs);
+
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, tbCompressionDefault1, tableCopyAndSetCompression, false, false, false);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, tableCompactionLcs, tableCopyAndSetLCSCompaction, false, false, false);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, tableOtherOptions, tableCopyAndSetAllParams, false, false, false);
@@ -364,29 +368,29 @@ public class CreateLikeTest extends CQLTester
         assertEquals(paramsSetLCSCompaction, TableParams.builder().compaction(CompactionParams.create(LeveledCompactionStrategy.class,
                                                                                                       Map.of("sstable_size_in_mb", "10",
                                                                                                              "fanout_size", "16")))
-                                                                  .build());
+                                                        .build());
         assertEquals(paramsSetAllParams, TableParams.builder().bloomFilterFpChance(0.75)
-                                                              .caching(new CachingParams(false, 10))
-                                                              .cdc(true)
-                                                              .comment("test for create like and set params")
-                                                              .crcCheckChance(0.8)
-                                                              .defaultTimeToLive(100)
-                                                              .compaction(CompactionParams.stcs(Collections.emptyMap()))
-                                                              .compression(CompressionParams.snappy(64 * 1024, 0.0))
-                                                              .gcGraceSeconds(1000)
-                                                              .incrementalBackups(true)
-                                                              .maxIndexInterval(128)
-                                                              .minIndexInterval(16)
-                                                              .speculativeRetry(SpeculativeRetryPolicy.fromString("96PERCENTILE"))
-                                                              .readRepair(ReadRepairStrategy.NONE)
-                                                              .memtableFlushPeriodInMs(3600)
-                                                              .build());
+                                                    .caching(new CachingParams(false, 10))
+                                                    .cdc(true)
+                                                    .comment("test for create like and set params")
+                                                    .crcCheckChance(0.8)
+                                                    .defaultTimeToLive(100)
+                                                    .compaction(CompactionParams.stcs(Collections.emptyMap()))
+                                                    .compression(CompressionParams.snappy(64 * 1024, 0.0))
+                                                    .gcGraceSeconds(1000)
+                                                    .incrementalBackups(true)
+                                                    .maxIndexInterval(128)
+                                                    .minIndexInterval(16)
+                                                    .speculativeRetry(SpeculativeRetryPolicy.fromString("96PERCENTILE"))
+                                                    .readRepair(ReadRepairStrategy.NONE)
+                                                    .memtableFlushPeriodInMs(3600)
+                                                    .build());
 
         // table id
         TableId id = TableId.generate();
         String tbNormal = createTable(sourceKs, "CREATE TABLE %s (a text, b int, c int, primary key (a, b))");
         assertInvalidThrowMessage("Cannot alter table id.", ConfigurationException.class,
-                                  "CREATE TABLE " + targetKs + ".targetnormal LIKE " +  sourceKs + "." + tbNormal + " WITH ID = " + id);
+                                  "CREATE TABLE " + targetKs + ".targetnormal LIKE " + sourceKs + "." + tbNormal + " WITH ID = " + id);
     }
 
     @Test
@@ -401,7 +405,7 @@ public class CreateLikeTest extends CQLTester
 
         // add static column
         sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int, b int, c text, PRIMARY KEY (a, b))");
-        alterTable("ALTER TABLE " + sourceKs + "." + sourceTb + " ADD d  int static");
+        alterTable("ALTER TABLE " + sourceKs + "." + sourceTb + " ADD d int static");
         targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb);
     }
@@ -486,35 +490,35 @@ public class CreateLikeTest extends CQLTester
         // source table's column's data type is udt, and its subtypes are native type and udt
         String sourceTbUdtFrozen = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c " + udtFrozen + ");");
         // source table's column's data type is udt, and its subtypes are native type and  more than one udt
-        String sourceTbUdtComb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c " + udtFrozen + ", d " + udt+ ");");
+        String sourceTbUdtComb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c " + udtFrozen + ", d " + udt + ");");
         // source table's column's data type is udt, and its subtypes are native type and  more than one udt
-        String sourceTbUdtCombNotExist = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c " + udtFrozen + ", d " + udtFrozenNotExist+ ");");
+        String sourceTbUdtCombNotExist = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c " + udtFrozen + ", d " + udtFrozenNotExist + ");");
 
         if (differentKs)
         {
-            assertInvalidThrowMessage("UDTs " + udt + " do not exist in target keyspace '" + targetKs +"'.",
+            assertInvalidThrowMessage("UDTs " + udt + " do not exist in target keyspace '" + targetKs + "'.",
                                       InvalidRequestException.class,
                                       "CREATE TABLE " + targetKs + ".tbudt LIKE " + sourceKs + "." + sourceTbUdt);
-            assertInvalidThrowMessage("UDTs " + udtSet + " do not exist in target keyspace '" + targetKs +"'.",
+            assertInvalidThrowMessage("UDTs " + udtSet + " do not exist in target keyspace '" + targetKs + "'.",
                                       InvalidRequestException.class,
                                       "CREATE TABLE " + targetKs + ".tbdtset LIKE " + sourceKs + "." + sourceTbUdtSet);
             assertInvalidThrowMessage(String.format("UDTs %s do not exist in target keyspace '%s'.", Sets.newHashSet(udt, udtFrozen).stream().sorted().collect(Collectors.joining(", ")), targetKs),
                                       InvalidRequestException.class,
-                                "CREATE TABLE " + targetKs + ".tbudtfrozen LIKE " + sourceKs + "." + sourceTbUdtFrozen);
+                                      "CREATE TABLE " + targetKs + ".tbudtfrozen LIKE " + sourceKs + "." + sourceTbUdtFrozen);
             assertInvalidThrowMessage(String.format("UDTs %s do not exist in target keyspace '%s'.", Sets.newHashSet(udt, udtFrozen).stream().sorted().collect(Collectors.joining(", ")), targetKs),
                                       InvalidRequestException.class,
                                       "CREATE TABLE " + targetKs + ".tbudtfrozen LIKE " + sourceKs + "." + sourceTbUdtFrozen);
             assertInvalidThrowMessage(String.format("UDTs %s do not exist in target keyspace '%s'.", Sets.newHashSet(udt, udtFrozen).stream().sorted().collect(Collectors.joining(", ")), targetKs),
                                       InvalidRequestException.class,
                                       "CREATE TABLE " + targetKs + ".tbudtcomb LIKE " + sourceKs + "." + sourceTbUdtComb);
-            assertInvalidThrowMessage(String.format("UDTs %s do not exist in target keyspace '%s'.", Sets.newHashSet(udtNew, udt,udtFrozenNotExist, udtFrozen).stream().sorted().collect(Collectors.joining(", ")), targetKs),
-                    InvalidRequestException.class,
-                    "CREATE TABLE " + targetKs + ".tbudtcomb LIKE " + sourceKs + "." + sourceTbUdtCombNotExist);
+            assertInvalidThrowMessage(String.format("UDTs %s do not exist in target keyspace '%s'.", Sets.newHashSet(udtNew, udt, udtFrozenNotExist, udtFrozen).stream().sorted().collect(Collectors.joining(", ")), targetKs),
+                                      InvalidRequestException.class,
+                                      "CREATE TABLE " + targetKs + ".tbudtcomb LIKE " + sourceKs + "." + sourceTbUdtCombNotExist);
             // different keyspaces with udts that have same udt name, different fields
             String udtWithDifferentField = createType(sourceKs, "CREATE TYPE %s (aa int, bb text)");
             createType("CREATE TYPE IF NOT EXISTS " + targetKs + "." + udtWithDifferentField + " (aa int, cc text)");
             String sourceTbDiffUdt = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c " + udtWithDifferentField + ");");
-            assertInvalidThrowMessage("Target keyspace '" + targetKs + "' has same UDT name '"+ udtWithDifferentField  +"' as source keyspace '" + sourceKs + "' but with different structure.",
+            assertInvalidThrowMessage("Target keyspace '" + targetKs + "' has same UDT name '" + udtWithDifferentField + "' as source keyspace '" + sourceKs + "' but with different structure.",
                                       InvalidRequestException.class,
                                       "CREATE TABLE " + targetKs + ".tbdiffudt LIKE " + sourceKs + "." + sourceTbDiffUdt);
         }
@@ -576,28 +580,138 @@ public class CreateLikeTest extends CQLTester
     public void testUnSupportedSchema() throws Throwable
     {
         createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b int, c text)", "tb");
-        String index = createIndex( "CREATE INDEX ON " + sourceKs + ".tb (c)");
+        String index = createIndex("CREATE INDEX ON " + sourceKs + ".tb (c)");
         assertInvalidThrowMessage("Souce Table '" + targetKs + "." + index + "' doesn't exist", InvalidRequestException.class,
-                            "CREATE TABLE " + sourceKs + ".newtb LIKE  " + targetKs + "." + index + ";");
-
+                                  "CREATE TABLE " + sourceKs + ".newtb LIKE  " + targetKs + "." + index + ";");
         assertInvalidThrowMessage("System keyspace 'system' is not user-modifiable", InvalidRequestException.class,
                                   "CREATE TABLE system.local_clone LIKE system.local ;");
         assertInvalidThrowMessage("System keyspace 'system_views' is not user-modifiable", InvalidRequestException.class,
                                   "CREATE TABLE system_views.newtb LIKE system_views.snapshots ;");
     }
 
-    private void assertTableMetaEqualsWithoutKs(String sourceKs, String targetKs, String sourceTb, String targetTb)
+    @Test
+    public void testTableCopyWithIndexes()
     {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b int, c text, d int, e text, f int, g text)", "sourcetb");
+        createIndex(sourceKs, "CREATE INDEX ON %s (d)");
+        createIndex(sourceKs, "CREATE INDEX ON %s (c)");
+        createIndex(sourceKs, "CREATE INDEX ON %s (b) USING 'sai'");
+        createIndex(sourceKs, "CREATE CUSTOM INDEX ON %s (e) USING 'storageattachedindex'");
+        createIndex(sourceKs, "CREATE CUSTOM INDEX ON %s (f) USING 'org.apache.cassandra.index.sai.StorageAttachedIndex'");
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s WITH INDEXES", sourceTb, sourceKs, targetKs);
         assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb, true, true, true);
     }
 
-    private void assertTableMetaEqualsWithoutKs(String sourceKs, String targetKs, String sourceTb, String targetTb, boolean compareParams, boolean compareIndexes, boolean compareTrigger)
+    @Test
+    public void testTableCopyWithMultiIndexOnSameColumn()
+    {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b int, c text, d int, e text, f int, g text)", "sourcetb");
+        createIndex(sourceKs, "CREATE INDEX " + sourceTb + "_b_idx1 ON %s (b) USING 'legacy_local_table'");
+        createIndex(sourceKs, "CREATE INDEX " + sourceTb + "_b_idx2 ON %s (b) USING 'sai'");
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s WITH INDEXES", sourceTb, sourceKs, targetKs);
+        assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb, true, true, true);
+    }
+
+    @Test
+    public void testTableCopyWithOutIndexes()
+    {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b int)", "sourcetb");
+        // not copied
+        createIndex(sourceKs, "CREATE CUSTOM INDEX ON %s (b) USING 'org.apache.cassandra.index.sasi.SASIIndex'");
+        createIndex(sourceKs, "CREATE CUSTOM INDEX testidx ON %s (b) USING '" + StubIndex.class.getName() + "'");
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s WITH indexes", sourceTb, sourceKs, targetKs);
+        assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb, true, false, false);
+        assertEquals(2, getTableMetadata(sourceKs, sourceTb).indexes.size());
+        assertEquals(0, getTableMetadata(targetKs, targetTb).indexes.size());
+    }
+
+    @Test
+    public void testManyTableCopyWithIndex()
+    {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b int, c int)", "sourcetb");
+        createIndex(sourceKs, "CREATE INDEX myindex ON %s (b) USING 'legacy_local_table'");
+        createIndex(sourceKs, "CREATE INDEX myindex_1 ON %s (b) USING 'sai'");
+        createIndex(sourceKs, "CREATE INDEX myindex_1_1 ON %s (c) USING 'sai'");
+        createIndex(sourceKs, "CREATE INDEX myindex__1 ON %s (c) USING 'legacy_local_table'");
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s WITH indexes", sourceTb, sourceKs, targetKs);
+        assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb, true, false, false);
+        Set<String> resultIndexNames = getTableMetadata(targetKs, targetTb).indexes.stream().map(meta -> meta.name).collect(Collectors.toSet());
+        Set<String> expectedIndexNames = differentKs ? Sets.newHashSet("myindex", "myindex_1", "myindex_1_1", "myindex__1") :
+                                                       Sets.newHashSet("myindex_2", "myindex_3", "myindex_1_2", "myindex__2");
+        assertEquals(0, Sets.difference(resultIndexNames, expectedIndexNames).size());
+    }
+
+    @Test
+    public void testTableCopyWithIndexesAndTableProperty()
+    {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b int, c text, d int)", "sourcetb");
+        createIndex(sourceKs, "CREATE INDEX ON %s (b)");
+
+        String targetTbWithAll = createTableLike("CREATE TABLE %s LIKE %s WITH INDEXES AND crc_check_chance = 0.8 AND cdc = true", sourceTb, sourceKs, targetKs);
+        assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTbWithAll, false, true, true);
+
+        TableMetadata source = getTableMetadata(sourceKs, sourceTb);
+        TableMetadata target = getTableMetadata(targetKs, targetTbWithAll);
+        assertNotEquals(source.params, target.params);
+        assertTrue(target.params.cdc);
+        assertFalse(source.params.cdc);
+        assertEquals(0.8, target.params.crcCheckChance, 0.0);
+    }
+
+    @Test
+    public void testIndexesNameForCopiedTable()
+    {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b int, c text, d int, e text)", "sourcetb");
+        String idx1 = createIndex(sourceKs, "CREATE INDEX ON %s (d)");
+        String idx2 = createIndex(sourceKs, "CREATE INDEX ON %s (c)");
+        String idx3 = createIndex(sourceKs, "CREATE INDEX ON %s (b) USING 'sai'");
+        String idx4 = createIndex(sourceKs, "CREATE INDEX idx_for_e_column ON %s (e) USING 'sai'");
+        assertEquals(sourceTb + "_d_idx" , idx1);
+        assertEquals(sourceTb + "_c_idx" , idx2);
+        assertEquals(sourceTb + "_b_idx" , idx3);
+        assertEquals("idx_for_e_column", idx4);
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s WITH INDEXES", sourceTb, sourceKs, targetKs);
+        assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb, true, true, true);
+        TableMetadata target = getTableMetadata(targetKs, targetTb);
+        assertTrue(target.indexes.has(targetTb + "_d_idx"));
+        assertTrue(target.indexes.has(targetTb + "_c_idx"));
+        assertTrue(target.indexes.has(targetTb + "_b_idx"));
+
+        if (differentKs)
+        {
+            assertTrue(target.indexes.has("idx_for_e_column"));
+        }
+        else
+        {
+            // if within the same keyspace, the target table will be created with a new
+            // index name in the format oldindexname_number, where the number starts from 1.
+            assertTrue(target.indexes.has("idx_for_e_column_1"));
+        }
+    }
+
+    @Test
+    public void testTableCopyWithCustomIndexes()
+    {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b int, c text, d int)", "sourcetb");
+        String idx = createIndex(sourceKs, "CREATE CUSTOM INDEX testidx ON %s (b) USING '" + StubIndex.class.getName() + "'");
+        Set<String> customIndexes = new TreeSet<>(Arrays.asList(idx));
+        ResultSet res = executeNet("CREATE TABLE " + targetKs + ".targettbb LIKE " + sourceKs + "." + sourceTb + " WITH INDEXES");
+        assertWarningsContain(res.getExecutionInfo().getWarnings(),
+                              "Source table " + sourceKs + "." + sourceTb + " to copy indexes from to " + targetKs + ".targettbb has custom indexes. These indexes were not copied: " + customIndexes);
+    }
+
+    private void assertTableMetaEqualsWithoutKs(String sourceKs, String targetKs, String sourceTb, String targetTb)
+    {
+        assertTableMetaEqualsWithoutKs(sourceKs, targetKs, sourceTb, targetTb, true, true, false);
+    }
+
+    private void assertTableMetaEqualsWithoutKs(String sourceKs, String targetKs, String sourceTb, String targetTb, boolean compareParams, boolean compareIndexes, boolean compareIndexWithOutName)
     {
         TableMetadata left = getTableMetadata(sourceKs, sourceTb);
         TableMetadata right = getTableMetadata(targetKs, targetTb);
         assertNotNull(left);
         assertNotNull(right);
-        assertTrue(equalsWithoutTableNameAndDropCns(left, right, compareParams, compareIndexes, compareTrigger));
+        assertTrue(equalsWithoutTableNameAndDropCns(left, right, compareParams, compareIndexes, compareIndexWithOutName));
         assertNotEquals(left.id, right.id);
         assertNotEquals(left.name, right.name);
     }