From 46ef00a31594312403e8b42e66ae9354e7bf87ef Mon Sep 17 00:00:00 2001
From: Maxwell Guo <cclive1601@gmail.com>
Date: Fri, 11 Oct 2024 19:23:41 +0800
Subject: [PATCH] Implementation of CEP-43

patch by Maxwell Guo; reviewed by Stefan Miklosovic, Benjamin Lerer for CASSANDRA-19964
---
 CHANGES.txt                                   |   1 +
 NEWS.txt                                      |   2 +
 doc/cql3/CQL.textile                          |  33 ++
 .../examples/BNF/create_table_like.bnf        |   3 +
 .../examples/CQL/create_table_like.cql        |  14 +
 pylib/cqlshlib/cql3handling.py                |  22 +
 pylib/cqlshlib/test/test_cql_parsing.py       |   3 +
 pylib/cqlshlib/test/test_cqlsh_completion.py  | 117 +++-
 src/antlr/Parser.g                            |  12 +
 .../cassandra/audit/AuditLogEntryType.java    |   1 +
 .../statements/schema/CopyTableStatement.java | 220 +++++++
 .../schema/CreateTableStatement.java          |  23 +-
 .../test/CreateTableNonDeterministicTest.java |  25 +
 .../test/metric/TableMetricTest.java          |   6 +
 .../cassandra/audit/AuditLoggerTest.java      |  23 +
 .../cassandra/auth/GrantAndRevokeTest.java    |  92 +++
 .../cql3/AlterSchemaStatementTest.java        |   3 +-
 .../org/apache/cassandra/cql3/CQLTester.java  |  73 +++
 .../statements/DescribeStatementTest.java     |  62 ++
 .../createlike/CreateLikeCqlParseTest.java    |  51 ++
 .../schema/createlike/CreateLikeTest.java     | 557 ++++++++++++++++++
 .../createlike/CreateLikeWithSessionTest.java |  71 +++
 22 files changed, 1404 insertions(+), 10 deletions(-)
 create mode 100644 doc/modules/cassandra/examples/BNF/create_table_like.bnf
 create mode 100644 doc/modules/cassandra/examples/CQL/create_table_like.cql
 create mode 100644 src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java
 create mode 100644 test/unit/org/apache/cassandra/schema/createlike/CreateLikeCqlParseTest.java
 create mode 100644 test/unit/org/apache/cassandra/schema/createlike/CreateLikeTest.java
 create mode 100644 test/unit/org/apache/cassandra/schema/createlike/CreateLikeWithSessionTest.java

diff --git a/CHANGES.txt b/CHANGES.txt
index ab3b662f6a..02a0be0571 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 5.1
+ * Implementation of CEP-43 (CASSANDRA-19964)
  * Periodically disconnect roles that are revoked or have LOGIN=FALSE set (CASSANDRA-19385)
  * AST library for CQL-based fuzz tests (CASSANDRA-20198)
  * Support audit logging for JMX operations (CASSANDRA-20128)
diff --git a/NEWS.txt b/NEWS.txt
index 22921153da..a4cec79660 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -107,6 +107,8 @@ New features
       JMX to JVM parameters. You have to opt-in to use the configuration via cassandra.yaml by uncommenting
       the respective configuration sections and by commenting out `configure_jmx` function call in cassandra-env.sh.
       Enabling both ways of configuring JMX will result in a node failing to start.
+    - CEP-43 - it is possible to create a table by "copying" as `CREATE TABLE ks.tb_copy LIKE ks.tb;`.
+      A newly created table will have no data.
 
 Upgrading
 ---------
diff --git a/doc/cql3/CQL.textile b/doc/cql3/CQL.textile
index 4c2b09acc3..f819ec193d 100644
--- a/doc/cql3/CQL.textile
+++ b/doc/cql3/CQL.textile
@@ -406,6 +406,39 @@ h4. Other considerations:
 
 * When "inserting":#insertStmt / "updating":#updateStmt a given row, not all columns needs to be defined (except for those part of the key), and missing columns occupy no space on disk. Furthermore, adding new columns (see <a href=#alterStmt><tt>ALTER TABLE</tt></a>) is a constant time operation. There is thus no need to try to anticipate future usage (or to cry when you haven't) when creating a table.
 
+h3(#copyStmt). CREATE TABLE LIKE
+
+__Syntax:__
+
+bc(syntax).. 
+<copy-table-stmt> ::= CREATE ( TABLE | COLUMNFAMILY ) ( IF NOT EXISTS )? <newtablename> LIKE <oldtablename>
+                    ( WITH <option> ( AND <option>)* )?
+
+<option> ::= <property>
+
+p. 
+__Sample:__
+
+bc(sample).. 
+CREATE TABLE newtb1 LIKE oldtb;
+
+CREATE TABLE newtb2 LIKE oldtb WITH 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.
+
+h4(#copyNewTableName). @<newtablename>@
+
+Valid table names are the same as valid "keyspace names":#createKeyspaceStmt (up to 32 characters long alphanumerical identifiers). If the table name is provided alone, the table is created within the current keyspace (see <a href="#useStmt"><tt>USE</tt></a>), but if it is prefixed by an existing keyspace name (see "@<tablename>@":#statements grammar), it is created in the specified keyspace (but does *not* change the current keyspace).
+
+h4(#copyOldTableName). @<oldtablename>@
+
+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 .
 
 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
new file mode 100644
index 0000000000..56d209c6ef
--- /dev/null
+++ b/doc/modules/cassandra/examples/BNF/create_table_like.bnf
@@ -0,0 +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
diff --git a/doc/modules/cassandra/examples/CQL/create_table_like.cql b/doc/modules/cassandra/examples/CQL/create_table_like.cql
new file mode 100644
index 0000000000..ef65b8b050
--- /dev/null
+++ b/doc/modules/cassandra/examples/CQL/create_table_like.cql
@@ -0,0 +1,14 @@
+CREATE TABLE ks.newtb1 LIKE ks.oldtb;
+
+CREATE TABLE ks1.newtb1 LIKE ks.oldtb;
+
+USE ks;
+
+CREATE TABLE newtb1 LIKE oldtb;
+
+CREATE TABLE IF NOT EXISTS newtb2 LIKE oldtb;
+
+CREATE TABLE newtb3 LIKE oldtb WITH compaction = { 'class' : 'LeveledCompactionStrategy' }
+                               AND compression = { 'class' : 'SnappyCompressor', 'chunk_length_in_kb' : 32 }
+                               AND cdc = true;
+
diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py
index 457582f3d3..4d2f4dffcf 100644
--- a/pylib/cqlshlib/cql3handling.py
+++ b/pylib/cqlshlib/cql3handling.py
@@ -279,6 +279,7 @@ JUNK ::= /([ \t\r\f\v]+|(--|[/][/])[^\n\r]*([\n\r]|$)|[/][*].*?[*][/])/ ;
 
 <schemaChangeStatement> ::= <createKeyspaceStatement>
                           | <createColumnFamilyStatement>
+                          | <copyTableStatement>
                           | <createIndexStatement>
                           | <createMaterializedViewStatement>
                           | <createUserTypeStatement>
@@ -1302,6 +1303,27 @@ def create_cf_composite_primary_key_comma_completer(ctxt, cass):
     return [',']
 
 
+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> )* )?
+                            ;
+'''
+
+
+@completer_for('copyTableStatement', 'wat')
+def create_tb_wat_completer(ctxt, cass):
+    # would prefer to get rid of the "schema" nomenclature in cql3
+    if ctxt.get_binding('partial', '') == '':
+        return ['TABLE']
+    return ['COLUMNFAMILY', 'TABLE']
+
+
+explain_completion('copyTableStatement', 'tcf', '<new_table_name>')
+explain_completion('copyTableStatement', 'scf', '<old_table_name>')
+
+
 syntax_rules += r'''
 
 <idxName> ::= <identifier>
diff --git a/pylib/cqlshlib/test/test_cql_parsing.py b/pylib/cqlshlib/test/test_cql_parsing.py
index 00227c83e6..b9eb716a78 100644
--- a/pylib/cqlshlib/test/test_cql_parsing.py
+++ b/pylib/cqlshlib/test/test_cql_parsing.py
@@ -568,6 +568,9 @@ class TestCqlParsing(TestCase):
     def test_parse_create_table(self):
         pass
 
+    def test_parse_create_table_like(self):
+        pass
+
     def test_parse_drop_table(self):
         pass
 
diff --git a/pylib/cqlshlib/test/test_cqlsh_completion.py b/pylib/cqlshlib/test/test_cqlsh_completion.py
index 112474e7c7..2a02fcf464 100644
--- a/pylib/cqlshlib/test/test_cqlsh_completion.py
+++ b/pylib/cqlshlib/test/test_cqlsh_completion.py
@@ -397,7 +397,7 @@ class TestCqlshCompletion(CqlshCompletionCase):
                             choices=['EXISTS', '<quotedName>', '<identifier>'])
 
         self.trycompletions("UPDATE empty_table SET lonelycol = 'eggs' WHERE TOKEN(lonelykey) <= TOKEN(13) IF EXISTS ",
-                            choices=['>=', '!=', '<=', 'IN','[', ';', '=', '<', '>', '.', 'CONTAINS'])
+                            choices=['>=', '!=', '<=', 'IN', '[', ';', '=', '<', '>', '.', 'CONTAINS'])
 
         self.trycompletions("UPDATE empty_table SET lonelycol = 'eggs' WHERE TOKEN(lonelykey) <= TOKEN(13) IF lonelykey ",
                             choices=['>=', '!=', '<=', 'IN', '=', '<', '>', 'CONTAINS'])
@@ -464,7 +464,7 @@ class TestCqlshCompletion(CqlshCompletionCase):
                             choices=['a', 'b', 'TOKEN('])
 
         self.trycompletions('DELETE FROM twenty_rows_composite_table USING TIMESTAMP 0 WHERE a ',
-                            choices=['<=', '>=', 'BETWEEN', 'CONTAINS', 'IN', 'NOT' , '[', '=', '<', '>', '!='])
+                            choices=['<=', '>=', 'BETWEEN', 'CONTAINS', 'IN', 'NOT', '[', '=', '<', '>', '!='])
 
         self.trycompletions('DELETE FROM twenty_rows_composite_table USING TIMESTAMP 0 WHERE TOKEN(',
                             immediate='a ')
@@ -614,9 +614,9 @@ class TestCqlshCompletion(CqlshCompletionCase):
         self.trycompletions(prefix + 'IF NOT EXISTS ',
                             choices=['<new_table_name>', self.cqlsh.keyspace])
         self.trycompletions(prefix + 'IF NOT EXISTS new_table ',
-                            immediate='( ')
+                            choices=['(', '.', 'LIKE'])
 
-        self.trycompletions(prefix + quoted_keyspace, choices=['.', '('])
+        self.trycompletions(prefix + quoted_keyspace, choices=['.', '(', 'LIKE'])
 
         self.trycompletions(prefix + quoted_keyspace + '( ',
                             choices=['<new_column_name>', '<identifier>',
@@ -625,7 +625,7 @@ class TestCqlshCompletion(CqlshCompletionCase):
         self.trycompletions(prefix + quoted_keyspace + '.',
                             choices=['<new_table_name>'])
         self.trycompletions(prefix + quoted_keyspace + '.new_table ',
-                            immediate='( ')
+                            choices=['(', 'LIKE'])
         self.trycompletions(prefix + quoted_keyspace + '.new_table ( ',
                             choices=['<new_column_name>', '<identifier>',
                                      '<quotedName>'])
@@ -780,6 +780,113 @@ class TestCqlshCompletion(CqlshCompletionCase):
         self.trycompletions('CREATE TA', immediate='BLE ')
         self.create_columnfamily_table_template('TABLE')
 
+    def test_complete_in_create_table_like(self):
+        self.trycompletions('CREATE T', choices=['TRIGGER', 'TABLE', 'TYPE'])
+        self.trycompletions('CREATE TA', immediate='BLE ')
+        quoted_keyspace = '"' + self.cqlsh.keyspace + '"'
+        self.trycompletions('CREATE TABLE ',
+                            choices=['IF', self.cqlsh.keyspace, '<new_table_name>'])
+        self.trycompletions('CREATE TABLE IF ',
+                            immediate='NOT EXISTS ')
+        self.trycompletions('CREATE TABLE IF NOT EXISTS ',
+                            choices=['<new_table_name>', self.cqlsh.keyspace])
+        self.trycompletions('CREATE TABLE IF NOT EXISTS new_table L',
+                            immediate='IKE ')
+        self.trycompletions('CREATE TABLE ' + quoted_keyspace + '.',
+                            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',
+                            immediate='ITH ')
+        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'])
+        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'])
+        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 = "
+                            + "{'class': '",
+                            choices=['SizeTieredCompactionStrategy',
+                                     'LeveledCompactionStrategy',
+                                     'TimeWindowCompactionStrategy',
+                                     'UnifiedCompactionStrategy'])
+        self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
+                            + "{'class': 'S",
+                            immediate="izeTieredCompactionStrategy'")
+        self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
+                            + "{'class': 'SizeTieredCompactionStrategy",
+                            immediate="'")
+        self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
+                            + "{'class': 'SizeTieredCompactionStrategy'",
+                            choices=['}', ','])
+        self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
+                            + "{'class': 'SizeTieredCompactionStrategy', ",
+                            immediate="'")
+        self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
+                            + "{'class': 'SizeTieredCompactionStrategy', '",
+                            choices=['bucket_high', 'bucket_low', 'class',
+                                     'enabled', 'max_threshold',
+                                     'min_sstable_size', 'min_threshold',
+                                     'tombstone_compaction_interval',
+                                     'tombstone_threshold',
+                                     'unchecked_tombstone_compaction',
+                                     'only_purge_repaired_tombstones',
+                                     'provide_overlapping_tombstones'])
+        self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
+                            + "{'class': 'SizeTieredCompactionStrategy'}",
+                            choices=[';', 'AND'])
+        self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
+                            + "{'class': 'SizeTieredCompactionStrategy'} AND ",
+                            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'])
+        self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
+                            + "{'class': 'TimeWindowCompactionStrategy', '",
+                            choices=['compaction_window_unit', 'compaction_window_size',
+                                     'timestamp_resolution', 'min_threshold', 'class', 'max_threshold',
+                                     'tombstone_compaction_interval', 'tombstone_threshold',
+                                     'enabled', 'unchecked_tombstone_compaction',
+                                     'only_purge_repaired_tombstones', 'provide_overlapping_tombstones'])
+        self.trycompletions('CREATE TABLE ' + "new_table LIKE old_table WITH compaction = "
+                            + "{'class': 'UnifiedCompactionStrategy', '",
+                            choices=['scaling_parameters', 'min_sstable_size',
+                                     'flush_size_override', 'base_shard_count', 'class', 'target_sstable_size',
+                                     'sstable_growth', 'max_sstables_to_compact',
+                                     'enabled', 'expired_sstable_check_frequency_seconds',
+                                     'unsafe_aggressive_sstable_expiration', 'overlap_inclusion_method',
+                                     'tombstone_threshold', 'tombstone_compaction_interval',
+                                     'unchecked_tombstone_compaction', 'provide_overlapping_tombstones',
+                                     'max_threshold', 'only_purge_repaired_tombstones'])
+
     def test_complete_in_describe(self):  # Cassandra-10733
         self.trycompletions('DES', immediate='C')
         # quoted_keyspace = '"' + self.cqlsh.keyspace + '"'
diff --git a/src/antlr/Parser.g b/src/antlr/Parser.g
index 15c2a0140b..9400395733 100644
--- a/src/antlr/Parser.g
+++ b/src/antlr/Parser.g
@@ -236,6 +236,7 @@ cqlStatement returns [CQLStatement.Raw stmt]
     | st42=addIdentityStatement            { $stmt = st42; }
     | st43=dropIdentityStatement           { $stmt = st43; }
     | st44=listSuperUsersStatement         { $stmt = st44; }
+    | st45=copyTableStatement              { $stmt = st45; }
     ;
 
 /*
@@ -813,6 +814,17 @@ tableClusteringOrder[CreateTableStatement.Raw stmt]
     : k=ident (K_ASC | K_DESC { ascending = false; } ) { $stmt.extendClusteringOrder(k, ascending); }
     ;
 
+/**
+ * CREATE TABLE [IF NOT EXISTS] <NEW_TABLE> LIKE <OLD_TABLE> WITH <property> = <value> AND ...;
+ */
+copyTableStatement returns  [CopyTableStatement.Raw stmt]
+    @init { boolean ifNotExists = false; }
+    : 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] )*)?
+    ;
+
 /**
  * CREATE TYPE foo (
  *    <name1> <type1>,
diff --git a/src/java/org/apache/cassandra/audit/AuditLogEntryType.java b/src/java/org/apache/cassandra/audit/AuditLogEntryType.java
index 484895a21b..2bbff08429 100644
--- a/src/java/org/apache/cassandra/audit/AuditLogEntryType.java
+++ b/src/java/org/apache/cassandra/audit/AuditLogEntryType.java
@@ -32,6 +32,7 @@ public enum AuditLogEntryType
     ALTER_KEYSPACE(AuditLogEntryCategory.DDL),
     DROP_KEYSPACE(AuditLogEntryCategory.DDL),
     CREATE_TABLE(AuditLogEntryCategory.DDL),
+    CREATE_TABLE_LIKE(AuditLogEntryCategory.DDL),
     DROP_TABLE(AuditLogEntryCategory.DDL),
     PREPARE_STATEMENT(AuditLogEntryCategory.PREPARE),
     DROP_TRIGGER(AuditLogEntryCategory.DDL),
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java
new file mode 100644
index 0000000000..0df8d00f45
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CopyTableStatement.java
@@ -0,0 +1,220 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.cql3.statements.schema;
+
+import org.apache.cassandra.audit.AuditLogContext;
+import org.apache.cassandra.audit.AuditLogEntryType;
+import org.apache.cassandra.auth.Permission;
+import org.apache.cassandra.cql3.CQLStatement;
+import org.apache.cassandra.cql3.QualifiedName;
+import org.apache.cassandra.db.guardrails.Guardrails;
+import org.apache.cassandra.db.marshal.VectorType;
+import org.apache.cassandra.exceptions.AlreadyExistsException;
+import org.apache.cassandra.schema.Indexes;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.Keyspaces;
+import org.apache.cassandra.schema.MemtableParams;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
+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.reads.repair.ReadRepairStrategy;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.transport.Event.SchemaChange;
+
+/**
+ * {@code CREATE TABLE [IF NOT EXISTS] <newtable> LIKE <oldtable> WITH <property> = <value>}
+ */
+public final class CopyTableStatement extends AlterSchemaStatement
+{
+    private final String sourceKeyspace;
+    private final String sourceTableName;
+    private final String targetKeyspace;
+    private final String targetTableName;
+    private final boolean ifNotExists;
+    private final TableAttributes attrs;
+
+    public CopyTableStatement(String sourceKeyspace,
+                              String targetKeyspace,
+                              String sourceTableName,
+                              String targetTableName,
+                              boolean ifNotExists,
+                              TableAttributes attrs)
+    {
+        super(targetKeyspace);
+        this.sourceKeyspace = sourceKeyspace;
+        this.targetKeyspace = targetKeyspace;
+        this.sourceTableName = sourceTableName;
+        this.targetTableName = targetTableName;
+        this.ifNotExists = ifNotExists;
+        this.attrs = attrs;
+    }
+
+    @Override
+    SchemaChange schemaChangeEvent(Keyspaces.KeyspacesDiff diff)
+    {
+        return new SchemaChange(SchemaChange.Change.CREATED, SchemaChange.Target.TABLE, targetKeyspace, targetTableName);
+    }
+
+    @Override
+    public void authorize(ClientState client)
+    {
+        client.ensureTablePermission(sourceKeyspace, sourceTableName, Permission.SELECT);
+        client.ensureAllTablesPermission(targetKeyspace, Permission.CREATE);
+    }
+
+    @Override
+    public AuditLogContext getAuditLogContext()
+    {
+        return new AuditLogContext(AuditLogEntryType.CREATE_TABLE_LIKE, targetKeyspace, targetTableName);
+    }
+
+    @Override
+    public Keyspaces apply(ClusterMetadata metadata)
+    {
+        Keyspaces schema = metadata.schema.getKeyspaces();
+        KeyspaceMetadata sourceKeyspaceMeta = schema.getNullable(sourceKeyspace);
+
+        if (null == sourceKeyspaceMeta)
+            throw ire("Source Keyspace '%s' doesn't exist", sourceKeyspace);
+
+        TableMetadata sourceTableMeta = sourceKeyspaceMeta.getTableNullable(sourceTableName);
+
+        if (null == sourceTableMeta)
+            throw ire("Souce Table '%s.%s' doesn't exist", sourceKeyspace, sourceTableName);
+
+        if (sourceTableMeta.isIndex())
+            throw ire("Cannot use CREATE TABLE LIKE on an index table '%s.%s'.", sourceKeyspace, sourceTableName);
+
+        if (sourceTableMeta.isView())
+            throw ire("Cannot use CREATE TABLE LIKE on a materialized view '%s.%s'.", sourceKeyspace, sourceTableName);
+
+        KeyspaceMetadata targetKeyspaceMeta = schema.getNullable(targetKeyspace);
+        if (null == targetKeyspaceMeta)
+            throw ire("Target Keyspace '%s' doesn't exist", targetKeyspace);
+
+        if (targetKeyspaceMeta.hasTable(targetTableName))
+        {
+            if(ifNotExists)
+                return schema;
+
+            throw new AlreadyExistsException(targetKeyspace, targetTableName);
+        }
+        // todo support udt for differenet ks latter
+        if (!sourceKeyspace.equalsIgnoreCase(targetKeyspace) && !sourceTableMeta.getReferencedUserTypes().isEmpty())
+            throw ire("Cannot use CREATE TABLE LIKE across different keyspace when source table have UDTs.");
+
+        // Guardrail on columns per table
+        Guardrails.columnsPerTable.guard(sourceTableMeta.columns().size(), targetTableName, false, state);
+
+        sourceTableMeta.columns().forEach(columnMetadata -> {
+            if (columnMetadata.type.isVector())
+            {
+                Guardrails.vectorTypeEnabled.ensureEnabled(columnMetadata.name.toString(), state);
+                int dimensions = ((VectorType)columnMetadata.type).dimension;
+                Guardrails.vectorDimensions.guard(dimensions, columnMetadata.name.toString(), false, state);
+            }
+        });
+
+        // Guardrail to check whether creation of new COMPACT STORAGE tables is allowed
+        if (sourceTableMeta.isCompactTable())
+            Guardrails.compactTablesEnabled.ensureEnabled(state);
+
+        if (sourceKeyspaceMeta.replicationStrategy.hasTransientReplicas()
+            && sourceTableMeta.params.readRepair != ReadRepairStrategy.NONE)
+        {
+            throw ire("read_repair must be set to 'NONE' for transiently replicated keyspaces");
+        }
+
+        if (!sourceTableMeta.params.compression.isEnabled())
+            Guardrails.uncompressedTablesEnabled.ensureEnabled(state);
+
+        // withInternals can be set to false as it is only used for souce table id, which is not need for target table and the table
+        // id can be set through create table like cql using WITH ID
+        String sourceCQLString = sourceTableMeta.toCqlString(false, false, false, false);
+
+        TableMetadata.Builder targetBuilder = CreateTableStatement.parse(sourceCQLString,
+                                                                         targetKeyspace,
+                                                                         targetTableName,
+                                                                         sourceKeyspaceMeta.types,
+                                                                         UserFunctions.none())
+                                                                  .indexes(Indexes.none())
+                                                                  .triggers(Triggers.none());
+
+        TableParams originalParams = targetBuilder.build().params;
+        TableParams newTableParams = attrs.asAlteredTableParams(originalParams);
+
+        TableMetadata table = targetBuilder.params(newTableParams)
+                                           .id(TableId.get(metadata))
+                                           .build();
+        table.validate();
+
+        return schema.withAddedOrUpdated(targetKeyspaceMeta.withSwapped(targetKeyspaceMeta.tables.with(table)));
+    }
+
+    @Override
+    public void validate(ClientState state)
+    {
+        super.validate(state);
+
+        // If a memtable configuration is specified, validate it against config
+        if (attrs.hasOption(TableParams.Option.MEMTABLE))
+            MemtableParams.get(attrs.getString(TableParams.Option.MEMTABLE.toString()));
+
+        // Guardrail on table properties
+        Guardrails.tableProperties.guard(attrs.updatedProperties(), attrs::removeProperty, state);
+
+        // Guardrail on number of tables
+        if (Guardrails.tables.enabled(state))
+        {
+            int totalUserTables = Schema.instance.getUserKeyspaces()
+                                                 .stream()
+                                                 .mapToInt(ksm -> ksm.tables.size())
+                                                 .sum();
+            Guardrails.tables.guard(totalUserTables + 1, targetTableName, false, state);
+        }
+        validateDefaultTimeToLive(attrs.asNewTableParams());
+    }
+
+    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();
+
+        public Raw(QualifiedName newName, QualifiedName oldName, boolean ifNotExists)
+        {
+            this.newName = newName;
+            this.oldName = oldName;
+            this.ifNotExists = ifNotExists;
+        }
+
+        @Override
+        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);
+        }
+    }
+}
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java
index 7c705f7fee..5970360f24 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java
@@ -479,12 +479,21 @@ public final class CreateTableStatement extends AlterSchemaStatement
         }
     }
 
+    public static TableMetadata.Builder parse(String cql, String keyspace, String table, Types types, UserFunctions userFunctions)
+    {
+        Raw createTable = CQLFragmentParser.parseAny(CqlParser::createTableStatement, cql, "CREATE TABLE")
+                                           .keyspace(keyspace);
+
+        if (table != null)
+            createTable.table(table);
+
+        return createTable.prepare(null) // works around a messy ClientState/QueryProcessor class init deadlock
+                .builder(types, userFunctions);
+    }
+
     public static TableMetadata.Builder parse(String cql, String keyspace)
     {
-        return CQLFragmentParser.parseAny(CqlParser::createTableStatement, cql, "CREATE TABLE")
-                                .keyspace(keyspace)
-                                .prepare(null) // works around a messy ClientState/QueryProcessor class init deadlock
-                                .builder(Types.none(), UserFunctions.none());
+        return parse(cql, keyspace, null, Types.none(), UserFunctions.none());
     }
 
     public final static class Raw extends CQLStatement.Raw
@@ -538,6 +547,12 @@ public final class CreateTableStatement extends AlterSchemaStatement
             return this;
         }
 
+        public Raw table(String table)
+        {
+            name.setName(table, true);
+            return this;
+        }
+
         public String table()
         {
             return name.getName();
diff --git a/test/distributed/org/apache/cassandra/distributed/test/CreateTableNonDeterministicTest.java b/test/distributed/org/apache/cassandra/distributed/test/CreateTableNonDeterministicTest.java
index 74c08047fc..0cf551e2e8 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/CreateTableNonDeterministicTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/CreateTableNonDeterministicTest.java
@@ -25,10 +25,13 @@ import org.junit.Test;
 
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.distributed.api.IInvokableInstance;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.tcm.ClusterMetadata;
 
+import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
+import static org.apache.cassandra.distributed.shared.AssertUtils.row;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 
@@ -77,6 +80,28 @@ public class CreateTableNonDeterministicTest extends TestBaseImpl
         }
     }
 
+    @Test
+    public void testCreateLikeTable() throws IOException
+    {
+        try (Cluster cluster = init(Cluster.build(2).start()))
+        {
+            cluster.schemaChange(withKeyspace("CREATE TABLE %s.sourcetb (k int primary key, v text)"));
+            TableId node1id = tableId(cluster.get(1), "sourcetb");
+            TableId node2id = tableId(cluster.get(2), "sourcetb");
+            assertEquals(node1id, node2id);
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".targettb LIKE " + KEYSPACE + ".sourcetb");
+            TableId node1id2 = tableId(cluster.get(1), "targettb");
+            TableId node2id2 = tableId(cluster.get(2), "targettb");
+            assertNotEquals(node1id, node1id2);
+            assertEquals(node1id2, node2id2);
+            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.sourcetb(k, v) VALUES (1, 'v1')"), ConsistencyLevel.QUORUM);
+            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.targettb(k, v) VALUES (1, 'v1')"), ConsistencyLevel.QUORUM);
+            Object[] row = row(1, "v1");
+            assertRows(cluster.coordinator(1).execute(withKeyspace("SELECT * FROM %s.sourcetb"), ConsistencyLevel.QUORUM), row);
+            assertRows(cluster.coordinator(1).execute(withKeyspace("SELECT * FROM %s.targettb "), ConsistencyLevel.QUORUM), row);
+        }
+    }
+
     long epoch(IInvokableInstance inst)
     {
         return inst.callOnInstance(() -> ClusterMetadata.current().epoch.getEpoch());
diff --git a/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java b/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
index 58a3feef40..1ea737b191 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java
@@ -105,10 +105,16 @@ public class TableMetricTest extends TestBaseImpl
             cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl DROP value"));
             cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
 
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".targettb LIKE " + KEYSPACE + ".tbl");
+            cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "targettb"));
+
             // drop and make sure table no longer exists
             cluster.schemaChange(withKeyspace("DROP TABLE %s.tbl"));
             cluster.forEach(i -> assertTableMetricsDoesNotExist(i, KEYSPACE, "tbl"));
 
+            cluster.schemaChange(withKeyspace("DROP TABLE %s.targettb"));
+            cluster.forEach(i -> assertTableMetricsDoesNotExist(i, KEYSPACE, "tbl"));
+
             cluster.schemaChange(withKeyspace("DROP KEYSPACE %s"));
             cluster.forEach(i -> assertKeyspaceMetricDoesNotExists(i, KEYSPACE));
 
diff --git a/test/unit/org/apache/cassandra/audit/AuditLoggerTest.java b/test/unit/org/apache/cassandra/audit/AuditLoggerTest.java
index 46345d88fc..f97e0bf5ce 100644
--- a/test/unit/org/apache/cassandra/audit/AuditLoggerTest.java
+++ b/test/unit/org/apache/cassandra/audit/AuditLoggerTest.java
@@ -560,6 +560,29 @@ public class AuditLoggerTest extends CQLTester
         executeAndAssert(cql, AuditLogEntryType.DROP_TRIGGER);
     }
 
+    @Test
+    public void testCqlCreateTableLikeAuditing() throws Throwable
+    {
+        String cql = "CREATE TABLE " + KEYSPACE + "." + createTableName() + " (id int primary key, v1 text, v2 text)";
+        executeAndAssert(cql, AuditLogEntryType.CREATE_TABLE);
+
+        cql = "CREATE TABLE IF NOT EXISTS " + KEYSPACE + "." + createTableName() + " (id int primary key, v1 text, v2 text)";
+        executeAndAssert(cql, AuditLogEntryType.CREATE_TABLE);
+
+        String sourceTable = currentTable();
+
+        cql = "CREATE TABLE " + KEYSPACE + "." + createTableName() + " LIKE " + KEYSPACE + "." + sourceTable;
+        executeAndAssert(cql, AuditLogEntryType.CREATE_TABLE_LIKE);
+
+        cql = "INSERT INTO " + KEYSPACE + '.' + currentTable() + "  (id, v1, v2) VALUES (?, ?, ?)";
+        executeAndAssertWithPrepare(cql, AuditLogEntryType.UPDATE, 1, "insert_audit", "test");
+
+        cql = "SELECT id, v1, v2 FROM " + KEYSPACE + '.' + currentTable() + " WHERE id = ?";
+        ResultSet rs = executeAndAssertWithPrepare(cql, AuditLogEntryType.SELECT, 1);
+
+        assertEquals(1, rs.all().size());
+    }
+
     @Test
     public void testCqlAggregateAuditing() throws Throwable
     {
diff --git a/test/unit/org/apache/cassandra/auth/GrantAndRevokeTest.java b/test/unit/org/apache/cassandra/auth/GrantAndRevokeTest.java
index 55e1135cd4..2ac0232139 100644
--- a/test/unit/org/apache/cassandra/auth/GrantAndRevokeTest.java
+++ b/test/unit/org/apache/cassandra/auth/GrantAndRevokeTest.java
@@ -386,4 +386,96 @@ public class GrantAndRevokeTest extends CQLTester
         res = executeNet("REVOKE SELECT, MODIFY ON KEYSPACE revoke_yeah FROM " + user);
         assertWarningsContain(res.getExecutionInfo().getWarnings(), "Role '" + user + "' was not granted MODIFY on <keyspace revoke_yeah>");
     }
+
+    @Test
+    public void testCreateTableLikeAuthorize() throws Throwable
+    {
+        useSuperUser();
+
+        // two keyspaces
+        executeNet("CREATE KEYSPACE ks1 WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}");
+        executeNet("CREATE KEYSPACE ks2 WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}");
+        executeNet("CREATE TABLE ks1.sourcetb (id int PRIMARY KEY, val text)");
+        executeNet("CREATE USER '" + user + "' WITH PASSWORD '" + pass + "'");
+
+        // same keyspace
+        // have no select permission on source table
+        ResultSet res = executeNet("REVOKE SELECT ON TABLE ks1.sourcetb FROM " + user);
+        assertWarningsContain(res.getExecutionInfo().getWarnings(), "Role '" + user + "' was not granted SELECT on <table ks1.sourcetb>");
+
+        useUser(user, pass);
+        // Spin assert for effective auth changes.
+        Util.spinAssertEquals(false, () -> {
+            try
+            {
+                assertUnauthorizedQuery("User user has no SELECT permission on <table ks1.sourcetb> or any of its parents",
+                        formatQuery("SELECT * FROM ks1.sourcetb LIMIT 1"));
+            }
+            catch(Throwable e)
+            {
+                return true;
+            }
+            return false;
+        }, 10);
+
+        assertUnauthorizedQuery("User user has no SELECT permission on <table ks1.sourcetb> or any of its parents",
+                                "CREATE TABLE ks1.targetTb LIKE ks1.sourcetb");
+
+        // have select permission on source table and do not have create permission on target keyspace
+        useSuperUser();
+        executeNet("GRANT SELECT ON TABLE ks1.sourcetb TO " + user);
+        res = executeNet("REVOKE CREATE ON KEYSPACE ks1 FROM " + user);
+        assertWarningsContain(res.getExecutionInfo().getWarnings(), "Role '" + user + "' was not granted CREATE on <keyspace ks1>");
+
+        useUser(user, pass);
+        Util.spinAssertEquals(false, () -> {
+            try
+            {
+                assertUnauthorizedQuery("User user has no CREATE permission on <all tables in ks1> or any of its parents",
+                        formatQuery("CREATE TABLE ks1.targetTb LIKE ks1.sourcetb"));
+            }
+            catch(Throwable e)
+            {
+                return true;
+            }
+            return false;
+        }, 10);
+
+        assertUnauthorizedQuery("User user has no CREATE permission on <all tables in ks1> or any of its parents",
+                                "CREATE TABLE ks1.targetTb LIKE ks1.sourcetb");
+
+        // different keyspaces
+        // have select permission on source table and do not have create permission on target keyspace
+        useSuperUser();
+        executeNet("GRANT SELECT ON TABLE ks1.sourcetb TO " + user);
+        res = executeNet("REVOKE CREATE ON KEYSPACE ks2 FROM " + user);
+        assertWarningsContain(res.getExecutionInfo().getWarnings(), "Role '" + user + "' was not granted CREATE on <keyspace ks2>");
+
+        useUser(user, pass);
+        Util.spinAssertEquals(false, () -> {
+            try
+            {
+                assertUnauthorizedQuery("User user has no CREATE permission on <all tables in ks2> or any of its parents",
+                        formatQuery("CREATE TABLE ks2.targetTb LIKE ks1.sourcetb"));
+            }
+            catch(Throwable e)
+            {
+                return true;
+            }
+            return false;
+        }, 10);
+
+        assertUnauthorizedQuery("User user has no CREATE permission on <all tables in ks2> or any of its parents",
+                                "CREATE TABLE ks2.targetTb LIKE ks1.sourcetb");
+
+        // source keyspace and table do not exist
+        assertUnauthorizedQuery("User user has no SELECT permission on <table ks1.tbnotexist> or any of its parents",
+                                "CREATE TABLE ks2.targetTb LIKE ks1.tbnotexist");
+        assertUnauthorizedQuery("User user has no SELECT permission on <table ksnotexists.sourcetb> or any of its parents",
+                                "CREATE TABLE ks2.targetTb LIKE ksnotexists.sourcetb");
+        // target keyspace does not exist
+        assertUnauthorizedQuery("User user has no CREATE permission on <all tables in ksnotexists> or any of its parents",
+                                "CREATE TABLE ksnotexists.targetTb LIKE ks1.sourcetb");
+
+    }
 }
diff --git a/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementTest.java b/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementTest.java
index 4c0f833504..57d7d2b5a5 100644
--- a/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementTest.java
+++ b/test/unit/org/apache/cassandra/cql3/AlterSchemaStatementTest.java
@@ -39,7 +39,8 @@ public class AlterSchemaStatementTest extends CQLTester
                      "CREATE KEYSPACE ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}",
                      "CREATE TABLE ks.t1 (k int PRIMARY KEY)",
                      "ALTER MATERIALIZED VIEW ks.v1 WITH compaction = { 'class' : 'LeveledCompactionStrategy' }",
-                     "ALTER TABLE ks.t1 ADD v int"
+                     "ALTER TABLE ks.t1 ADD v int",
+                     "CREATE TABLE ks.tb like ks1.tb"
                      };
     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 9410564b83..92f865b202 100644
--- a/test/unit/org/apache/cassandra/cql3/CQLTester.java
+++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java
@@ -163,6 +163,7 @@ import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.metrics.CassandraMetricsRegistry;
 import org.apache.cassandra.metrics.ClientMetrics;
 import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.KeyspaceMetadata;
 import org.apache.cassandra.schema.Schema;
@@ -1080,6 +1081,25 @@ public abstract class CQLTester
         return currentTable;
     }
 
+    protected String createTableLike(String query, String sourceTable, String sourceKeyspace, String targetKeyspace)
+    {
+        return createTableLike(query, sourceTable, sourceKeyspace, null, targetKeyspace);
+    }
+
+    protected String createTableLike(String query, String sourceTable, String sourceKeyspace, String targetTable, String targetKeyspace)
+    {
+        if (!tables.contains(sourceTable))
+        {
+            throw new IllegalArgumentException("Source table " + sourceTable + " does not exist");
+        }
+
+        String currentTable = createTableName(targetTable);
+        String fullQuery = currentTable == null ? query : String.format(query, targetKeyspace + "." + currentTable, sourceKeyspace + "." + sourceTable);;
+        logger.info(fullQuery);
+        schemaChange(fullQuery);
+        return currentTable;
+    }
+
     protected String createTableName()
     {
         return createTableName(null);
@@ -1545,6 +1565,11 @@ public abstract class CQLTester
         }
     }
 
+    protected TableMetadata getTableMetadata(String keyspace, String table)
+    {
+        return Schema.instance.getTableMetadata(keyspace, table);
+    }
+
     protected TableMetadata currentTableMetadata()
     {
         return Schema.instance.getTableMetadata(KEYSPACE, currentTable());
@@ -1935,6 +1960,54 @@ public abstract class CQLTester
         return false;
     }
 
+    /**
+     * 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 compareIndexes wether compare table's indexes
+     * @param compareTrigger wether compare table's triggers
+     * */
+    protected boolean equalsWithoutTableNameAndDropCns(TableMetadata source, TableMetadata target, boolean compareParams, boolean compareIndexes, boolean compareTrigger)
+    {
+        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))
+               && columnsEqualWitoutKsTb(source, target);
+    }
+
+    // only compare columns
+    private boolean columnsEqualWitoutKsTb(TableMetadata source, TableMetadata target)
+    {
+        if (target.columns().size() != source.columns().size())
+            return false;
+
+        Iterator<ColumnMetadata> leftIterator = source.allColumnsInCreateOrder();
+        Iterator<ColumnMetadata> rightIterator = target.allColumnsInCreateOrder();
+
+        while (leftIterator.hasNext() && rightIterator.hasNext())
+        {
+            ColumnMetadata leftCn = leftIterator.next();
+            ColumnMetadata rightCn = rightIterator.next();
+            if (!equalsWithoutKsTb(leftCn, rightCn))
+                return false;
+        }
+
+        return true;
+    }
+
+    private boolean equalsWithoutKsTb(ColumnMetadata left, ColumnMetadata right)
+    {
+        return left.name.equals(right.name)
+               && left.kind == right.kind
+               && left.position() == right.position()
+               && java.util.Objects.equals(left.getMask(), right.getMask())
+               && left.type.equals(right.type);
+    }
+
     protected void assertRowCountNet(ResultSet r1, int expectedCount)
     {
         Assert.assertFalse("Received a null resultset when expected count was > 0", expectedCount > 0 && r1 == null);
diff --git a/test/unit/org/apache/cassandra/cql3/statements/DescribeStatementTest.java b/test/unit/org/apache/cassandra/cql3/statements/DescribeStatementTest.java
index cd19414166..f3141a7640 100644
--- a/test/unit/org/apache/cassandra/cql3/statements/DescribeStatementTest.java
+++ b/test/unit/org/apache/cassandra/cql3/statements/DescribeStatementTest.java
@@ -53,6 +53,7 @@ import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME;
 import static org.apache.cassandra.schema.SchemaConstants.TRACE_KEYSPACE_NAME;
 import static org.apache.cassandra.schema.SchemaConstants.VIRTUAL_SCHEMA;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -822,6 +823,67 @@ public class DescribeStatementTest extends CQLTester
                       row(KEYSPACE_PER_TEST, "index", indexWithOptions, expectedIndexStmtWithOptions));
     }
 
+    @Test
+    public void testDescribeCreateLikeTable() throws Throwable
+    {
+        requireNetwork();
+        DatabaseDescriptor.setDynamicDataMaskingEnabled(true);
+        String souceTable = createTable(KEYSPACE_PER_TEST,
+                                  "CREATE TABLE %s (" +
+                                        "  pk1 text, " +
+                                        "  pk2 int MASKED WITH DEFAULT, " +
+                                        "  ck1 int, " +
+                                        "  ck2 double," +
+                                        "  s1 float static, " +
+                                        "  v1 int, " +
+                                        "  v2 int, " +
+                                        "PRIMARY KEY ((pk1, pk2), ck1, ck2 ))");
+        TableMetadata source = getTableMetadata(KEYSPACE_PER_TEST, currentTable());
+        assertNotNull(source);
+        String targetTable = createTableLike("CREATE TABLE %s LIKE %s", souceTable, KEYSPACE_PER_TEST, KEYSPACE_PER_TEST);
+        TableMetadata target = getTableMetadata(KEYSPACE_PER_TEST, currentTable());
+        assertNotNull(target);
+        assertTrue(equalsWithoutTableNameAndDropCns(source, target, true, true, true));
+        assertNotEquals(source.id, target.id);
+        assertNotEquals(source.name, target.name);
+
+        String sourceTableCreateStatement = "CREATE TABLE " + KEYSPACE_PER_TEST + "." + souceTable + " (\n" +
+                                            "    pk1 text,\n" +
+                                            "    pk2 int MASKED WITH system.mask_default(),\n" +
+                                            "    ck1 int,\n" +
+                                            "    ck2 double,\n" +
+                                            "    s1 float static,\n" +
+                                            "    v1 int,\n" +
+                                            "    v2 int,\n" +
+                                            "    PRIMARY KEY ((pk1, pk2), ck1, ck2)\n" +
+                                            ") WITH ID = " + source.id + "\n" +
+                                            "    AND CLUSTERING ORDER BY (ck1 ASC, ck2 ASC)\n" +
+                                            "    AND " + tableParametersCql();
+        String targetTableCreateStatement = "CREATE TABLE " + KEYSPACE_PER_TEST + "." + targetTable + " (\n" +
+                                            "    pk1 text,\n" +
+                                            "    pk2 int MASKED WITH system.mask_default(),\n" +
+                                            "    ck1 int,\n" +
+                                            "    ck2 double,\n" +
+                                            "    s1 float static,\n" +
+                                            "    v1 int,\n" +
+                                            "    v2 int,\n" +
+                                            "    PRIMARY KEY ((pk1, pk2), ck1, ck2)\n" +
+                                            ") WITH ID = " + target.id + "\n" +
+                                            "    AND CLUSTERING ORDER BY (ck1 ASC, ck2 ASC)\n" +
+                                            "    AND " + tableParametersCql();
+
+        assertRowsNet(executeDescribeNet("DESCRIBE TABLE " + KEYSPACE_PER_TEST + "." + souceTable + " WITH INTERNALS"),
+                      row(KEYSPACE_PER_TEST,
+                          "table",
+                          souceTable,
+                          sourceTableCreateStatement));
+        assertRowsNet(executeDescribeNet("DESCRIBE TABLE " + KEYSPACE_PER_TEST + "." + targetTable + " WITH INTERNALS"),
+                      row(KEYSPACE_PER_TEST,
+                          "table",
+                          targetTable,
+                          targetTableCreateStatement));
+    }
+
     @Test
     public void testDescribeTableWithColumnMasks() throws Throwable
     {
diff --git a/test/unit/org/apache/cassandra/schema/createlike/CreateLikeCqlParseTest.java b/test/unit/org/apache/cassandra/schema/createlike/CreateLikeCqlParseTest.java
new file mode 100644
index 0000000000..67a79870d0
--- /dev/null
+++ b/test/unit/org/apache/cassandra/schema/createlike/CreateLikeCqlParseTest.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.schema.createlike;
+
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.exceptions.SyntaxException;
+
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
+public class CreateLikeCqlParseTest extends CQLTester
+{
+    // Incorrect use of create table like cql
+    private static final String[] unSupportedCqls = new String[]
+            {
+                    "CREATE TABLE ta (a int primary key, b int) LIKE tb", // useless column information
+                    "CREATE TABLE ta (a int primary key, b int MASKED WITH DEFAULT) LIKE tb",
+                    "CREATE TABLE IF NOT EXISTS LIKE tb", // missing target table
+                    "CREATE TABLE IF NOT EXISTS LIKE tb WITH compression = { 'enabled' : 'false'}",
+                    "CREATE TABLE ta IF NOT EXISTS LIKE ", // missing source table
+                    "CREATE TABLE ta LIKE WITH id = '123-111'" // id is not supported
+            };
+
+    @Test
+    public void testUnsupportedCqlParse()
+    {
+        for (String cql : unSupportedCqls)
+        {
+            assertThatExceptionOfType(SyntaxException.class)
+                    .isThrownBy(() -> QueryProcessor.parseStatement(cql));
+        }
+    }
+}
diff --git a/test/unit/org/apache/cassandra/schema/createlike/CreateLikeTest.java b/test/unit/org/apache/cassandra/schema/createlike/CreateLikeTest.java
new file mode 100644
index 0000000000..02406acdfe
--- /dev/null
+++ b/test/unit/org/apache/cassandra/schema/createlike/CreateLikeTest.java
@@ -0,0 +1,557 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.schema.createlike;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+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.schema.CachingParams;
+import org.apache.cassandra.schema.CompactionParams;
+import org.apache.cassandra.schema.CompressionParams;
+import org.apache.cassandra.schema.Indexes;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.schema.TableParams;
+import org.apache.cassandra.service.reads.SpeculativeRetryPolicy;
+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.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public class CreateLikeTest extends CQLTester
+{
+    @Parameterized.Parameter
+    public boolean differentKs;
+
+    @Parameterized.Parameters(name = "{index}: differentKs={0}")
+    public static Collection<Object[]> data()
+    {
+        List<Object[]> result = new ArrayList<>();
+        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 String sourceKs;
+    private String targetKs;
+
+    @Before
+    public void before()
+    {
+        createKeyspace("CREATE KEYSPACE " + keyspace1 + " WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }");
+        createKeyspace("CREATE KEYSPACE " + keyspace2 + " WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }");
+        sourceKs = keyspace1;
+        targetKs = differentKs ? keyspace2 : keyspace1;
+    }
+
+    @Test
+    public void testTableSchemaCopy()
+    {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c text);");
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+        execute("INSERT INTO " + sourceKs + "." + sourceTb + " (a, b, c) VALUES (?, ?, ?)", 1, duration1, "1");
+        execute("INSERT INTO " + targetKs + "." + targetTb + " (a, b, c) VALUES (?, ?, ?)", 2, duration2, "2");
+        assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
+                   row(1, duration1, "1"));
+        assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
+                   row(2, duration2, "2"));
+
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY);");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+        execute("INSERT INTO " + sourceKs + "." + sourceTb + " (a) VALUES (1)");
+        execute("INSERT INTO " + targetKs + "." + targetTb + " (a) VALUES (2)");
+        assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
+                   row(1));
+        assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
+                   row(2));
+
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (a frozen<map<text, text>> PRIMARY KEY);");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+        execute("INSERT INTO " + sourceKs + "." + sourceTb + " (a) VALUES (?)", map("k", "v"));
+        execute("INSERT INTO " + targetKs + "." + targetTb + " (a) VALUES (?)", map("nk", "nv"));
+        assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
+                   row(map("k", "v")));
+        assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
+                   row(map("nk", "nv")));
+
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b set<frozen<list<text>>>, c map<text, int>, d smallint, e duration, f tinyint);");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(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);
+        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);
+        assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
+                   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));
+
+        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);
+        assertTableMetaEquals(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);
+        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);
+        assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
+                   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));
+
+        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);");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+        execute("INSERT INTO " + sourceKs + "." + sourceTb + " (a, b, c, d, e, f, g, h, i, j)  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                1, "b", 100L, decimal1, set("1", "2"), uuid1, vector1, list(1.1F, 2.2F), timeUuid1, map("k", set(1, 2)));
+        execute("INSERT INTO " + targetKs + "." + targetTb + " (a, b, c, d, e, f, g, h, i, j) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                2, "nb", 200L, decimal2, set("3", "4"), uuid2, vector2, list(3.3F, 4.4F), timeUuid2, map("nk", set(3, 4)));
+        assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
+                   row(1, "b", 100L, decimal1, set("1", "2"), uuid1, vector1, list(1.1F, 2.2F), timeUuid1, map("k", set(1, 2))));
+        assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
+                   row(2, "nb", 200L, decimal2, set("3", "4"), uuid2, vector2, list(3.3F, 4.4F), timeUuid2, map("nk", set(3, 4))));
+
+        // test that can create a copy of a copied table
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c text);");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        createTableLike("CREATE TABLE %s LIKE %s", targetTb, targetKs, "newtargettb", sourceKs);
+    }
+
+    @Test
+    public void testIfNotExists() throws Throwable
+    {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int, b text, c duration, d float, PRIMARY KEY(a, b));");
+        String targetTb = createTableLike("CREATE TABLE IF NOT EXISTS %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        createTableLike("CREATE TABLE IF NOT EXISTS %s LIKE %s", sourceTb, sourceKs, targetTb, targetKs);
+        assertInvalidThrowMessage("Cannot add already existing table \"" + targetTb + "\" to keyspace \"" + targetKs + "\"", AlreadyExistsException.class,
+                                  "CREATE TABLE " + targetKs + "." + targetTb + " LIKE " + sourceKs + "." + sourceTb);
+    }
+
+    @Test
+    public void testCopyAfterAlterTable()
+    {
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int, b text, c duration, d float, PRIMARY KEY(a, b));");
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        alterTable("ALTER TABLE  " + sourceKs + " ." + sourceTb + " DROP d");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        alterTable("ALTER TABLE " + sourceKs + " ." + sourceTb + " ADD e uuid");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        alterTable("ALTER TABLE " + sourceKs + " ." + sourceTb + " ADD f float");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        execute("INSERT INTO " + sourceKs + "." + sourceTb + " (a, b, c, e, f) VALUES (?, ?, ?, ?, ?)", 1, "1", duration1, uuid1, f1);
+        execute("INSERT INTO " + targetKs + "." + targetTb + " (a, b, c, e, f) VALUES (?, ?, ?, ?, ?)", 2, "2", duration2, uuid2, f2);
+        assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
+                   row(1, "1", duration1, uuid1, f1));
+        assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
+                   row(2, "2", duration2, uuid2, f2));
+
+        alterTable("ALTER TABLE " + sourceKs + " ." + sourceTb + " DROP f USING TIMESTAMP 20000");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        alterTable("ALTER TABLE " + sourceKs + " ." + sourceTb + " RENAME b TO bb ");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        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);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        execute("INSERT INTO " + sourceKs + "." + sourceTb + " (a, bb, c, e) VALUES (?, ?, ?, ?)", 1, "1", duration1, uuid1);
+        execute("INSERT INTO " + targetKs + "." + targetTb + " (a, bb, c, e) VALUES (?, ?, ?, ?)", 2, "2", duration2, uuid2);
+        assertRows(execute("SELECT * FROM " + sourceKs + "." + sourceTb),
+                   row(1, "1", duration1, uuid1));
+        assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb),
+                   row(2, "2", duration2, uuid2));
+
+    }
+
+    @Test
+    public void testTableOptionsCopy() throws Throwable
+    {
+        // 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 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 };");
+
+        // 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';");
+
+        // 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'};");
+
+        // 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';" );
+
+        String tbLikeCompressionDefault1 = createTableLike("CREATE TABLE %s LIKE %s", tbCompressionDefault1, sourceKs, targetKs);
+        String tbLikeCompressionDefault2 = createTableLike("CREATE TABLE %s LIKE %s", tbCompressionDefault2, sourceKs, targetKs);
+        String tbLikeCompressionSp1 = createTableLike("CREATE TABLE %s LIKE %s", tbCompressionSnappy1, sourceKs, targetKs);
+        String tbLikeCompressionSp2 = createTableLike("CREATE TABLE %s LIKE %s", tbCompressionSnappy2, sourceKs, targetKs);
+        String tbLikeCompressionSp3 = createTableLike("CREATE TABLE %s LIKE %s", tbCompressionSnappy3, sourceKs, targetKs);
+        String tbLikeCompressionSp4 = createTableLike("CREATE TABLE %s LIKE %s", tbCompressionSnappy4, sourceKs, targetKs);
+        String tbLikeCompressionSp5 = createTableLike("CREATE TABLE %s LIKE %s", tbCompressionSnappy5, sourceKs, targetKs);
+        String tbLikeMemtableSkipList = createTableLike("CREATE TABLE %s LIKE %s", tableMemtableSkipList, sourceKs, targetKs);
+        String tbLikeMemtableTrie = createTableLike("CREATE TABLE %s LIKE %s", tableMemtableTrie, sourceKs, targetKs);
+        String tbLikeMemtableDefault = createTableLike("CREATE TABLE %s LIKE %s", tableMemtableDefault, sourceKs, targetKs);
+        String tbLikeCompactionStcs = createTableLike("CREATE TABLE %s LIKE %s", tableCompactionStcs, sourceKs, targetKs);
+        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);
+
+        assertTableMetaEquals(sourceKs, targetKs, tbCompressionDefault1, tbLikeCompressionDefault1);
+        assertTableMetaEquals(sourceKs, targetKs, tbCompressionDefault2, tbLikeCompressionDefault2);
+        assertTableMetaEquals(sourceKs, targetKs, tbCompressionSnappy1, tbLikeCompressionSp1);
+        assertTableMetaEquals(sourceKs, targetKs, tbCompressionSnappy2, tbLikeCompressionSp2);
+        assertTableMetaEquals(sourceKs, targetKs, tbCompressionSnappy3, tbLikeCompressionSp3);
+        assertTableMetaEquals(sourceKs, targetKs, tbCompressionSnappy4, tbLikeCompressionSp4);
+        assertTableMetaEquals(sourceKs, targetKs, tbCompressionSnappy5, tbLikeCompressionSp5);
+        assertTableMetaEquals(sourceKs, targetKs, tableMemtableSkipList, tbLikeMemtableSkipList);
+        assertTableMetaEquals(sourceKs, targetKs, tableMemtableTrie, tbLikeMemtableTrie);
+        assertTableMetaEquals(sourceKs, targetKs, tableMemtableDefault, tbLikeMemtableDefault);
+        assertTableMetaEquals(sourceKs, targetKs, tableCompactionStcs, tbLikeCompactionStcs);
+        assertTableMetaEquals(sourceKs, targetKs, tableCompactionLcs, tbLikeCompactionLcs);
+        assertTableMetaEquals(sourceKs, targetKs, tableCompactionTwcs, tbLikeCompactionTwcs);
+        assertTableMetaEquals(sourceKs, targetKs, tableCompactionUcs, tbLikeCompactionUcs);
+        assertTableMetaEquals(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 };",
+                                                            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 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);
+        assertTableMetaEquals(sourceKs, targetKs, tbCompressionDefault1, tableCopyAndSetCompression, false, false, false);
+        assertTableMetaEquals(sourceKs, targetKs, tableCompactionLcs, tableCopyAndSetLCSCompaction, false, false, false);
+        assertTableMetaEquals(sourceKs, targetKs, tableOtherOptions, tableCopyAndSetAllParams, false, false, false);
+        TableParams paramsSetCompression = getTableMetadata(targetKs, tableCopyAndSetCompression).params;
+        TableParams paramsSetLCSCompaction = getTableMetadata(targetKs, tableCopyAndSetLCSCompaction).params;
+        TableParams paramsSetAllParams = getTableMetadata(targetKs, tableCopyAndSetAllParams).params;
+
+        assertEquals(paramsSetCompression, TableParams.builder().compression(CompressionParams.snappy(64 * 1024, 0.0)).build());
+        assertEquals(paramsSetLCSCompaction, TableParams.builder().compaction(CompactionParams.create(LeveledCompactionStrategy.class,
+                                                                                                      Map.of("sstable_size_in_mb", "10",
+                                                                                                             "fanout_size", "16")))
+                                                                  .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());
+
+        // 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);
+    }
+
+    @Test
+    public void testStaticColumnCopy()
+    {
+        // create with static column
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int , b int , c int static, d int, e list<text>, PRIMARY KEY(a, b));", "tb1");
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+        execute("INSERT INTO " + targetKs + "." + targetTb + " (a, b, c, d, e) VALUES (0, 1, 2, 3, ?)", list("1", "2", "3", "4"));
+        assertRows(execute("SELECT * FROM " + targetKs + "." + targetTb), row(0, 1, 2, 3, list("1", "2", "3", "4")));
+
+        // 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");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+    }
+
+    @Test
+    public void testColumnMaskTableCopy()
+    {
+        DatabaseDescriptor.setDynamicDataMaskingEnabled(true);
+        // masked partition key
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (k int MASKED WITH mask_default() PRIMARY KEY, r int)");
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        // masked partition key component
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (k1 int, k2 text MASKED WITH DEFAULT, r int, PRIMARY KEY(k1, k2))");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        // masked clustering key
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (k int, c int MASKED WITH mask_default(), r int, PRIMARY KEY (k, c))");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        // masked clustering key with reverse order
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (k int, c text MASKED WITH mask_default(), r int, PRIMARY KEY (k, c)) " +
+                                         "WITH CLUSTERING ORDER BY (c DESC)");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        // masked clustering key component
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (k int, c1 int, c2 text MASKED WITH DEFAULT, r int, PRIMARY KEY (k, c1, c2))");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        // masked regular column
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (k int PRIMARY KEY, r1 text MASKED WITH DEFAULT, r2 int)");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        // masked static column
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (k int, c int, r int, s int STATIC MASKED WITH DEFAULT, PRIMARY KEY (k, c))");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        // multiple masked columns
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (" +
+                                         "k1 int, k2 int MASKED WITH DEFAULT, " +
+                                         "c1 int, c2 text MASKED WITH DEFAULT, " +
+                                         "r1 int, r2 int MASKED WITH DEFAULT, " +
+                                         "s1 int static, s2 int static MASKED WITH DEFAULT, " +
+                                         "PRIMARY KEY((k1, k2), c1, c2))");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+
+        sourceTb = createTable(sourceKs, "CREATE TABLE %s (k int PRIMARY KEY, " +
+                                         "s set<int> MASKED WITH DEFAULT, " +
+                                         "l list<int> MASKED WITH DEFAULT, " +
+                                         "m map<int, int> MASKED WITH DEFAULT, " +
+                                         "fs frozen<set<int>> MASKED WITH DEFAULT, " +
+                                         "fl frozen<list<int>> MASKED WITH DEFAULT, " +
+                                         "fm frozen<map<int, int>> MASKED WITH DEFAULT)");
+        targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb);
+    }
+
+    @Test
+    public void testUDTTableCopy() throws Throwable
+    {
+        // normal udt
+        String udt = createType(sourceKs, "CREATE TYPE %s (a int, b uuid, c text)");
+        // collection udt
+        String udtSet = createType(sourceKs, "CREATE TYPE %s (a int, c frozen <set<text>>)");
+        // frozen udt
+        String udtFrozen = createType(sourceKs, "CREATE TYPE %s (a int, c frozen<" + udt + ">)");
+
+        String sourceTbUdt = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c " + udt + ");");
+        String sourceTbUdtSet = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c " + udtSet + ");");
+        String sourceTbUdtFrozen = createTable(sourceKs, "CREATE TABLE %s (a int PRIMARY KEY, b duration, c " + udtFrozen + ");");
+
+        if (differentKs)
+        {
+            assertInvalidThrowMessage("Cannot use CREATE TABLE LIKE across different keyspace when source table have UDT",
+                                      InvalidRequestException.class, "CREATE TABLE " + targetKs + ".tbudt LIKE " + sourceKs + "." + sourceTbUdt);
+            assertInvalidThrowMessage("Cannot use CREATE TABLE LIKE across different keyspace when source table have UDT",
+                                      InvalidRequestException.class, "CREATE TABLE " + targetKs + ".tbdtset LIKE " + sourceKs + "." + sourceTbUdt);
+            assertInvalidThrowMessage("Cannot use CREATE TABLE LIKE across different keyspace when source table have UDT", InvalidRequestException.class,
+                                      "CREATE TABLE " + targetKs + ".tbudtfrozen LIKE " + sourceKs + "." + sourceTbUdt);
+        }
+        else
+        {
+            String targetTbUdt = createTableLike("CREATE TABLE %s LIKE %s", sourceTbUdt, sourceKs, "tbudt", targetKs);
+            String targetTbUdtSet = createTableLike("CREATE TABLE %s LIKE %s", sourceTbUdtSet, sourceKs, "tbdtset", targetKs);
+            String targetTbUdtFrozen = createTableLike("CREATE TABLE %s LIKE %s", sourceTbUdtFrozen, sourceKs, "tbudtfrozen", targetKs);
+            assertTableMetaEquals(sourceKs, targetKs, sourceTbUdt, targetTbUdt);
+            assertTableMetaEquals(sourceKs, targetKs, sourceTbUdtSet, targetTbUdtSet);
+            assertTableMetaEquals(sourceKs, targetKs, sourceTbUdtFrozen, targetTbUdtFrozen);
+        }
+    }
+
+    @Test
+    public void testIndexOperationOnCopiedTable()
+    {
+        // copied table can do index creation
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (id text PRIMARY KEY, val text, num int);");
+        String targetTb = createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        String saiIndex = createIndex(targetKs, "CREATE INDEX ON %s(val) USING 'sai'");
+        execute("INSERT INTO " + targetKs + "." + targetTb + " (id, val, num) VALUES ('1', 'value', 1)");
+        assertEquals(1, execute("SELECT id FROM " + targetKs + "." + targetTb + " WHERE val = 'value'").size());
+        String normalIndex = createIndex(targetKs, "CREATE INDEX ON %s(num)");
+        Indexes targetIndexes = getTableMetadata(targetKs, targetTb).indexes;
+        assertEquals(targetIndexes.size(), 2);
+        assertNotNull(targetIndexes.get(saiIndex));
+        assertNotNull(targetIndexes.get(normalIndex));
+    }
+
+    @Test
+    public void testTriggerOperationOnCopiedTable()
+    {
+        String triggerName = "trigger_1";
+        String sourceTb = createTable(sourceKs, "CREATE TABLE %s (a int, b int, c int, PRIMARY KEY (a))");
+        String targetTb =  createTableLike("CREATE TABLE %s LIKE %s", sourceTb, sourceKs, targetKs);
+        execute("CREATE TRIGGER " + triggerName + " ON " + targetKs + "." + targetTb + " USING '" + CreateTest.TestTrigger.class.getName() + "'");
+        assertNotNull(getTableMetadata(targetKs, targetTb).triggers.get(triggerName));
+    }
+
+    @Test
+    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)");
+        assertInvalidThrowMessage("Souce Table '" + targetKs + "." + index + "' doesn't exist", InvalidRequestException.class,
+                            "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 assertTableMetaEquals(String sourceKs, String targetKs, String sourceTb, String targetTb)
+    {
+        assertTableMetaEquals(sourceKs, targetKs, sourceTb, targetTb, true, true, true);
+    }
+
+    private void assertTableMetaEquals(String sourceKs, String targetKs, String sourceTb, String targetTb, boolean compareParams, boolean compareIndexes, boolean compareTrigger)
+    {
+        TableMetadata left = getTableMetadata(sourceKs, sourceTb);
+        TableMetadata right = getTableMetadata(targetKs, targetTb);
+        assertNotNull(left);
+        assertNotNull(right);
+        assertTrue(equalsWithoutTableNameAndDropCns(left, right, compareParams, compareIndexes, compareTrigger));
+        assertNotEquals(left.id, right.id);
+        assertNotEquals(left.name, right.name);
+    }
+}
diff --git a/test/unit/org/apache/cassandra/schema/createlike/CreateLikeWithSessionTest.java b/test/unit/org/apache/cassandra/schema/createlike/CreateLikeWithSessionTest.java
new file mode 100644
index 0000000000..e676961d18
--- /dev/null
+++ b/test/unit/org/apache/cassandra/schema/createlike/CreateLikeWithSessionTest.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.schema.createlike;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.CQLTester;
+
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+public class CreateLikeWithSessionTest extends CQLTester
+{
+    private static String ks1 = "ks1";
+    private static String ks2 = "ks2";
+    private static String tb1 = "tb1";
+    private static String tb2 = "tb2";
+
+    @BeforeClass
+    public static void beforeClass()
+    {
+        requireNetwork();
+    }
+
+    @Test
+    public void testCreateLikeWithSession()
+    {
+        // create two keyspaces and tables
+        createKeyspace("CREATE KEYSPACE " + ks1 + " WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }");
+        createKeyspace("CREATE KEYSPACE " + ks2 + " WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }");
+        createTable("CREATE TABLE " + ks1 + "." + tb1 + " (id int PRIMARY KEY, age int);");
+        createTable("CREATE TABLE " + ks2 + "." + tb2 + " (name text PRIMARY KEY, address text);");
+
+        // use ks1
+        executeNet("use " + ks1);
+        executeNet("CREATE TABLE tb3 LIKE " + tb1);
+        executeNet("CREATE TABLE " + ks1 + ".tb4 LIKE " + tb1);
+        executeNet("CREATE TABLE tb5 like " + ks1 + "." + tb1);
+
+        executeNet("CREATE TABLE " + ks2 + ".tb6 LIKE " + tb1);
+
+
+        assertThatExceptionOfType(com.datastax.driver.core.exceptions.InvalidQueryException.class).isThrownBy(() -> executeNet("CREATE TABLE tb7 LIKE " + ks2 + "." + tb1))
+                                                                                                  .withMessage("Souce Table 'ks2.tb1' doesn't exist");
+
+        assertNotNull(getTableMetadata(ks1, tb1));
+        assertNotNull(getTableMetadata(ks1, "tb3"));
+        assertNotNull(getTableMetadata(ks1, "tb4"));
+        assertNotNull(getTableMetadata(ks1, "tb5"));
+        assertNotNull(getTableMetadata(ks2, "tb6"));
+        assertNull(getTableMetadata(ks2, "tb7"));
+    }
+}