diff --git a/CHANGES.txt b/CHANGES.txt
index 8f718e4e89e8..9ce6a2d4ba0d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 4.1.5
+ * Guardrail to block DDL/DCL queries
  * Fix hints delivery for a node going down repeatedly (CASSANDRA-19495)
  * Do not go to disk for reading hints file sizes (CASSANDRA-19477)
  * Fix system_views.settings to handle array types (CASSANDRA-19475)
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index be1439c80ef5..1ddffdd47aa7 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -1750,6 +1750,11 @@ drop_compact_storage_enabled: false
 # group_by_enabled: true
 # Guardrail to allow/disallow TRUNCATE and DROP TABLE statements
 # drop_truncate_table_enabled: true
+#
+# Guardrail to allow/disallow DDL/DCL statements
+# ddl_enabled: true
+# dcl_enabled: true
+#
 # Guardrail to warn or fail when using a page size greater than threshold.
 # The two thresholds default to -1 to disable.
 # page_size_warn_threshold: -1
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index 3a62aae7c773..66a8c29c0eef 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -834,6 +834,8 @@ public static void setClientMode(boolean clientMode)
     public volatile Set<ConsistencyLevel> write_consistency_levels_disallowed = Collections.emptySet();
     public volatile boolean user_timestamps_enabled = true;
     public volatile boolean group_by_enabled = true;
+    public volatile boolean ddl_enabled = true;
+    public volatile boolean dcl_enabled = true;
     public volatile boolean drop_truncate_table_enabled = true;
     public volatile boolean secondary_indexes_enabled = true;
     public volatile boolean uncompressed_tables_enabled = true;
diff --git a/src/java/org/apache/cassandra/config/GuardrailsOptions.java b/src/java/org/apache/cassandra/config/GuardrailsOptions.java
index e4694b9c307c..2b929f51793f 100644
--- a/src/java/org/apache/cassandra/config/GuardrailsOptions.java
+++ b/src/java/org/apache/cassandra/config/GuardrailsOptions.java
@@ -343,6 +343,34 @@ public void setDropTruncateTableEnabled(boolean enabled)
                                   x -> config.drop_truncate_table_enabled = x);
     }
 
+    @Override
+    public boolean getDDLEnabled()
+    {
+        return config.ddl_enabled;
+    }
+
+    public void setDDLEnabled(boolean enabled)
+    {
+        updatePropertyWithLogging("ddl_enabled",
+                                  enabled,
+                                  () -> config.ddl_enabled,
+                                  x -> config.ddl_enabled = x);
+    }
+
+    @Override
+    public boolean getDCLEnabled()
+    {
+        return config.dcl_enabled;
+    }
+
+    public void setDCLEnabled(boolean enabled)
+    {
+        updatePropertyWithLogging("dcl_enabled",
+                                  enabled,
+                                  () -> config.dcl_enabled,
+                                  x -> config.dcl_enabled = x);
+    }
+
     @Override
     public boolean getSecondaryIndexesEnabled()
     {
diff --git a/src/java/org/apache/cassandra/cql3/statements/AlterRoleStatement.java b/src/java/org/apache/cassandra/cql3/statements/AlterRoleStatement.java
index eb0e3e01f23c..33aa19f28972 100644
--- a/src/java/org/apache/cassandra/cql3/statements/AlterRoleStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/AlterRoleStatement.java
@@ -24,6 +24,7 @@
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.PasswordObfuscator;
 import org.apache.cassandra.cql3.RoleName;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.transport.messages.ResultMessage;
@@ -69,6 +70,8 @@ public void validate(ClientState state) throws RequestValidationException
         {
             checkTrue(ifExists, "Role %s doesn't exist", role.getRoleName());
         }
+
+        Guardrails.dclEnabled.ensureEnabled(state);
     }
 
     public void authorize(ClientState state) throws UnauthorizedException
diff --git a/src/java/org/apache/cassandra/cql3/statements/CreateRoleStatement.java b/src/java/org/apache/cassandra/cql3/statements/CreateRoleStatement.java
index 04f183d7bb03..6330b568a196 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CreateRoleStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CreateRoleStatement.java
@@ -23,6 +23,7 @@
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.PasswordObfuscator;
 import org.apache.cassandra.cql3.RoleName;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.transport.messages.ResultMessage;
@@ -71,6 +72,8 @@ public void validate(ClientState state) throws RequestValidationException
 
         if (!ifNotExists && DatabaseDescriptor.getRoleManager().isExistingRole(role))
             throw new InvalidRequestException(String.format("%s already exists", role.getRoleName()));
+
+        Guardrails.dclEnabled.ensureEnabled(state);
     }
 
     public ResultMessage execute(ClientState state) throws RequestExecutionException, RequestValidationException
diff --git a/src/java/org/apache/cassandra/cql3/statements/DropRoleStatement.java b/src/java/org/apache/cassandra/cql3/statements/DropRoleStatement.java
index 058ab01c04ba..87d5f2c5b3d4 100644
--- a/src/java/org/apache/cassandra/cql3/statements/DropRoleStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/DropRoleStatement.java
@@ -22,6 +22,7 @@
 import org.apache.cassandra.auth.*;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.RoleName;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.transport.messages.ResultMessage;
@@ -62,6 +63,8 @@ public void validate(ClientState state) throws RequestValidationException
         AuthenticatedUser user = state.getUser();
         if (user != null && user.getName().equals(role.getRoleName()))
             throw new InvalidRequestException("Cannot DROP primary role for current login");
+
+        Guardrails.dclEnabled.ensureEnabled(state);
     }
 
     public ResultMessage execute(ClientState state) throws RequestValidationException, RequestExecutionException
diff --git a/src/java/org/apache/cassandra/cql3/statements/PermissionsManagementStatement.java b/src/java/org/apache/cassandra/cql3/statements/PermissionsManagementStatement.java
index aa7e85ba7307..a105f923e034 100644
--- a/src/java/org/apache/cassandra/cql3/statements/PermissionsManagementStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/PermissionsManagementStatement.java
@@ -21,6 +21,7 @@
 
 import org.apache.cassandra.auth.*;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.cql3.RoleName;
 import org.apache.cassandra.exceptions.InvalidRequestException;
@@ -64,6 +65,8 @@ public void validate(ClientState state) throws RequestValidationException
 
         if (!resource.exists())
             throw new InvalidRequestException(String.format("Resource %s doesn't exist", resource));
+
+        Guardrails.dclEnabled.ensureEnabled(state);
     }
 
     public void authorize(ClientState state) throws UnauthorizedException
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
index 87377d70ec37..b85c64945e7d 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
@@ -31,6 +31,7 @@
 import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.gms.Gossiper;
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
@@ -73,6 +74,8 @@ public Keyspaces apply(Keyspaces schema)
                 throw ire("Keyspace '%s' doesn't exist", keyspaceName);
             return schema;
         }
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
 
         KeyspaceMetadata newKeyspace = keyspace.withSwapped(attrs.asAlteredKeyspaceParams(keyspace.params));
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
index 08c5f0412391..0f3e98e5578f 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
@@ -186,6 +186,9 @@ public void validate(ClientState state)
 
         public KeyspaceMetadata apply(KeyspaceMetadata keyspace, TableMetadata table)
         {
+            // if apply is not no-op then we check guardrail for this ddl op
+            Guardrails.ddlEnabled.ensureEnabled(state);
+
             TableMetadata.Builder tableBuilder = table.unbuild();
             Views.Builder viewsBuilder = keyspace.views.unbuild();
             newColumns.forEach(c -> addColumn(keyspace, table, c, ifColumnNotExists, tableBuilder, viewsBuilder));
@@ -216,6 +219,9 @@ private void addColumn(KeyspaceMetadata keyspace,
                 return;
             }
 
+            // if apply is not no-op then we check guardrail for this ddl op
+            Guardrails.ddlEnabled.ensureEnabled(state);
+
             if (table.isCompactTable())
                 throw ire("Cannot add new column to a COMPACT STORAGE table");
 
@@ -302,6 +308,9 @@ private void dropColumn(KeyspaceMetadata keyspace, TableMetadata table, ColumnId
                 return;
             }
 
+            // if apply is not no-op then we check guardrail for this ddl op
+            Guardrails.ddlEnabled.ensureEnabled(state);
+
             if (currentColumn.isPrimaryKeyColumn())
                 throw ire("Cannot drop PRIMARY KEY column %s", column);
 
@@ -379,6 +388,9 @@ private void renameColumn(KeyspaceMetadata keyspace,
                 return;
             }
 
+            // if apply is not no-op then we check guardrail for this ddl op
+            Guardrails.ddlEnabled.ensureEnabled(state);
+
             if (!column.isPrimaryKeyColumn())
                 throw ire("Cannot rename non PRIMARY KEY column %s", oldName);
 
@@ -434,6 +446,9 @@ public void validate(ClientState state)
 
         public KeyspaceMetadata apply(KeyspaceMetadata keyspace, TableMetadata table)
         {
+            // if apply is not no-op then we check guardrail for this ddl op
+            Guardrails.ddlEnabled.ensureEnabled(state);
+
             attrs.validate();
 
             TableParams params = attrs.asAlteredTableParams(table.params);
@@ -478,6 +493,9 @@ private DropCompactStorage(String keyspaceName, String tableName, boolean ifTabl
 
         public KeyspaceMetadata apply(KeyspaceMetadata keyspace, TableMetadata table)
         {
+            // if apply is not no-op then we check guardrail for this ddl op
+            Guardrails.ddlEnabled.ensureEnabled(state);
+
             if (!DatabaseDescriptor.enableDropCompactStorage())
                 throw new InvalidRequestException("DROP COMPACT STORAGE is disabled. Enable in cassandra.yaml to use.");
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTypeStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTypeStatement.java
index d8b19467ab21..dd108ef2b56d 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTypeStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTypeStatement.java
@@ -139,6 +139,8 @@ UserType apply(KeyspaceMetadata keyspace, UserType userType)
                     throw ire("Cannot add field %s to type %s: a field with name %s already exists", fieldName, userType.getCqlTypeName(), fieldName);
                 return userType;
             }
+            // if apply is not no-op then we check guardrail for this ddl op
+            Guardrails.ddlEnabled.ensureEnabled(state);
 
             AbstractType<?> fieldType = type.prepare(keyspaceName, keyspace.types).getType();
             if (fieldType.referencesUserType(userType.name))
@@ -212,6 +214,9 @@ UserType apply(KeyspaceMetadata keyspace, UserType userType)
                 fieldNames.set(idx, newName);
             });
 
+            // if apply is not no-op then we check guardrail for this ddl op
+            Guardrails.ddlEnabled.ensureEnabled(state);
+
             fieldNames.forEach(name ->
             {
                 if (fieldNames.stream().filter(isEqual(name)).count() > 1)
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterViewStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterViewStatement.java
index 7e707f476bed..693f015e3538 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterViewStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterViewStatement.java
@@ -68,6 +68,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Materialized view '%s.%s' doesn't exist", keyspaceName, viewName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         attrs.validate();
 
         // Guardrails on table properties
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateAggregateStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateAggregateStatement.java
index 055051568fe3..f7c41f7e4517 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateAggregateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateAggregateStatement.java
@@ -32,6 +32,7 @@
 import org.apache.cassandra.auth.Permission;
 import org.apache.cassandra.cql3.*;
 import org.apache.cassandra.cql3.functions.*;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.schema.Functions.FunctionsDiff;
 import org.apache.cassandra.schema.KeyspaceMetadata;
@@ -220,6 +221,9 @@ public Keyspaces apply(Keyspaces schema)
             }
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         return schema.withAddedOrUpdated(keyspace.withSwapped(keyspace.functions.withAddedOrUpdated(aggregate)));
     }
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateFunctionStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateFunctionStatement.java
index adb8f40eb3dc..8fecd439fc92 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateFunctionStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateFunctionStatement.java
@@ -33,6 +33,7 @@
 import org.apache.cassandra.cql3.functions.Function;
 import org.apache.cassandra.cql3.functions.FunctionName;
 import org.apache.cassandra.cql3.functions.UDFunction;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.schema.Functions.FunctionsDiff;
 import org.apache.cassandra.schema.KeyspaceMetadata;
@@ -152,6 +153,9 @@ public Keyspaces apply(Keyspaces schema)
             // TODO: update dependent aggregates
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         return schema.withAddedOrUpdated(keyspace.withSwapped(keyspace.functions.withAddedOrUpdated(function)));
     }
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
index 7a3a41e61868..dd2f7e769474 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateIndexStatement.java
@@ -158,6 +158,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Index %s is a duplicate of existing index %s", index.name, equalIndex.name);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         TableMetadata newTable = table.withSwapped(table.indexes.with(index));
         newTable.validate();
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java
index dc82f93a1095..746576d5431c 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java
@@ -75,6 +75,9 @@ public Keyspaces apply(Keyspaces schema)
             throw new AlreadyExistsException(keyspaceName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         KeyspaceMetadata keyspace = KeyspaceMetadata.create(keyspaceName, attrs.asNewKeyspaceParams());
 
         if (keyspace.params.replication.klass.equals(LocalStrategy.class))
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 f204b8e023ea..b5a8cdac0349 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java
@@ -109,6 +109,9 @@ public Keyspaces apply(Keyspaces schema)
             throw new AlreadyExistsException(keyspaceName, tableName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         TableMetadata table = builder(keyspace.types).build();
         table.validate();
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTriggerStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTriggerStatement.java
index e85ffd80aecd..b7b345f8985e 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTriggerStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTriggerStatement.java
@@ -21,6 +21,7 @@
 import org.apache.cassandra.audit.AuditLogEntryType;
 import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.QualifiedName;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.schema.*;
 import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
 import org.apache.cassandra.service.ClientState;
@@ -67,6 +68,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Trigger '%s' already exists", triggerName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         try
         {
             TriggerExecutor.instance.loadTriggerInstance(triggerClass);
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTypeStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTypeStatement.java
index e015c34ede9e..64e576a684a3 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateTypeStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateTypeStatement.java
@@ -85,6 +85,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("A user type with name '%s' already exists", typeName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         Set<FieldIdentifier> usedNames = new HashSet<>();
         for (FieldIdentifier name : fieldNames)
             if (!usedNames.add(name))
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java
index 145c8fc838ef..b8988ee72b1b 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateViewStatement.java
@@ -140,6 +140,9 @@ public Keyspaces apply(Keyspaces schema)
             throw new AlreadyExistsException(keyspaceName, viewName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         /*
          * Base table validation
          */
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropAggregateStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropAggregateStatement.java
index 0cb1cbeb3619..9c43d3526faa 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropAggregateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropAggregateStatement.java
@@ -31,6 +31,7 @@
 import org.apache.cassandra.cql3.functions.Function;
 import org.apache.cassandra.cql3.functions.FunctionName;
 import org.apache.cassandra.cql3.functions.UDAggregate;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.schema.*;
 import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
@@ -110,6 +111,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Aggregate '%s' doesn't exist", name);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         return schema.withAddedOrUpdated(keyspace.withSwapped(keyspace.functions.without(aggregate)));
     }
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropFunctionStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropFunctionStatement.java
index d9d637de875e..658ddba61b7e 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropFunctionStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropFunctionStatement.java
@@ -29,6 +29,7 @@
 import org.apache.cassandra.cql3.CQL3Type;
 import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.functions.*;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.schema.*;
 import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
@@ -109,6 +110,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Function '%s' doesn't exist", name);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         String dependentAggregates =
             keyspace.functions
                     .aggregatesUsingFunction(function)
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropIndexStatement.java
index 24b372d8c3c4..9842a409b876 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropIndexStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropIndexStatement.java
@@ -22,6 +22,7 @@
 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.schema.*;
 import org.apache.cassandra.schema.KeyspaceMetadata.KeyspaceDiff;
 import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
@@ -58,6 +59,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Index '%s.%s' doesn't exist'", keyspaceName, indexName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         TableMetadata newTable = table.withSwapped(table.indexes.without(indexName));
         return schema.withAddedOrUpdated(keyspace.withSwapped(keyspace.tables.withSwapped(newTable)));
     }
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java
index f2bd30b249df..e9b09a3442bb 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java
@@ -21,6 +21,7 @@
 import org.apache.cassandra.audit.AuditLogEntryType;
 import org.apache.cassandra.auth.Permission;
 import org.apache.cassandra.cql3.CQLStatement;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.schema.Keyspaces;
 import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
 import org.apache.cassandra.service.ClientState;
@@ -39,8 +40,11 @@ public DropKeyspaceStatement(String keyspaceName, boolean ifExists)
 
     public Keyspaces apply(Keyspaces schema)
     {
-        if (schema.containsKeyspace(keyspaceName))
+        if (schema.containsKeyspace(keyspaceName)) {
+            // if apply is not no-op then we check guardrail for this ddl op
+            Guardrails.ddlEnabled.ensureEnabled(state);
             return schema.without(keyspaceName);
+        }
 
         if (ifExists)
             return schema;
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java
index 78c98be3a70c..6d430b619926 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java
@@ -65,6 +65,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Table '%s.%s' doesn't exist", keyspaceName, tableName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         if (table.isView())
             throw ire("Cannot use DROP TABLE on a materialized view. Please use DROP MATERIALIZED VIEW instead.");
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropTriggerStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropTriggerStatement.java
index 967e56834f09..93ae3decd7a4 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropTriggerStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropTriggerStatement.java
@@ -21,6 +21,7 @@
 import org.apache.cassandra.audit.AuditLogEntryType;
 import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.QualifiedName;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.schema.*;
 import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
 import org.apache.cassandra.service.ClientState;
@@ -62,6 +63,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Trigger '%s' on '%s.%s' doesn't exist", triggerName, keyspaceName, tableName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         TableMetadata newTable = table.withSwapped(table.triggers.without(triggerName));
         return schema.withAddedOrUpdated(keyspace.withSwapped(keyspace.tables.withSwapped(newTable)));
     }
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropTypeStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropTypeStatement.java
index d188bdb792e7..d19a9c43230c 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropTypeStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropTypeStatement.java
@@ -25,6 +25,7 @@
 import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.UTName;
 import org.apache.cassandra.cql3.functions.Function;
+import org.apache.cassandra.db.guardrails.Guardrails;
 import org.apache.cassandra.db.marshal.UserType;
 import org.apache.cassandra.schema.KeyspaceMetadata;
 import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
@@ -73,6 +74,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Type '%s.%s' doesn't exist", keyspaceName, typeName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         /*
          * We don't want to drop a type unless it's not used anymore (mainly because
          * if someone drops a type and recreates one with the same name but different
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropViewStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropViewStatement.java
index 2c73717546c7..a1892ebd508e 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropViewStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropViewStatement.java
@@ -22,6 +22,7 @@
 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.schema.*;
 import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
 import org.apache.cassandra.service.ClientState;
@@ -57,6 +58,9 @@ public Keyspaces apply(Keyspaces schema)
             throw ire("Materialized view '%s.%s' doesn't exist", keyspaceName, viewName);
         }
 
+        // if apply is not no-op then we check guardrail for this ddl op
+        Guardrails.ddlEnabled.ensureEnabled(state);
+
         return schema.withAddedOrUpdated(keyspace.withSwapped(keyspace.views.without(viewName)));
     }
 
diff --git a/src/java/org/apache/cassandra/db/guardrails/Guardrails.java b/src/java/org/apache/cassandra/db/guardrails/Guardrails.java
index 9d08ab039b80..9389fa9cd724 100644
--- a/src/java/org/apache/cassandra/db/guardrails/Guardrails.java
+++ b/src/java/org/apache/cassandra/db/guardrails/Guardrails.java
@@ -150,6 +150,22 @@ public final class Guardrails implements GuardrailsMBean
                     state -> !CONFIG_PROVIDER.getOrCreate(state).getDropTruncateTableEnabled(),
                     "DROP and TRUNCATE TABLE functionality");
 
+    /**
+     * Guardrail disabling DDL statements
+     */
+    public static final DisableFlag ddlEnabled =
+    new DisableFlag("ddl_enabled",
+                    state -> !CONFIG_PROVIDER.getOrCreate(state).getDDLEnabled(),
+                    "DDL statement");
+
+    /**
+     * Guardrail disabling DCL statements
+     */
+    public static final DisableFlag dclEnabled =
+    new DisableFlag("dcl_enabled",
+                    state -> !CONFIG_PROVIDER.getOrCreate(state).getDCLEnabled(),
+                    "DCL statement");
+
     /**
      * Guardrail disabling user's ability to turn off compression
      */
@@ -599,6 +615,30 @@ public void setDropTruncateTableEnabled(boolean enabled)
         DEFAULT_CONFIG.setDropTruncateTableEnabled(enabled);
     }
 
+    @Override
+    public boolean getDDLEnabled()
+    {
+        return DEFAULT_CONFIG.getDDLEnabled();
+    }
+
+    @Override
+    public void setDDLEnabled(boolean enabled)
+    {
+        DEFAULT_CONFIG.setDDLEnabled(enabled);
+    }
+
+    @Override
+    public boolean getDCLEnabled()
+    {
+        return DEFAULT_CONFIG.getDCLEnabled();
+    }
+
+    @Override
+    public void setDCLEnabled(boolean enabled)
+    {
+        DEFAULT_CONFIG.setDCLEnabled(enabled);
+    }
+
     @Override
     public int getPageSizeWarnThreshold()
     {
diff --git a/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java b/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java
index a52eeb0b2536..f2ba15c7c91a 100644
--- a/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java
+++ b/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java
@@ -160,6 +160,20 @@ public interface GuardrailsConfig
      */
     boolean getDropTruncateTableEnabled();
 
+    /**
+     * Returns whether DDL statement is allowed
+     *
+     * @return {@code true} if allowed, {@code false} otherwise.
+     */
+    boolean getDDLEnabled();
+
+    /**
+     * Returns whether DCL statement is allowed
+     *
+     * @return {@code true} if allowed, {@code false} otherwise.
+     */
+    boolean getDCLEnabled();
+
     /**
      * @return The threshold to warn when page size exceeds given size.
      */
diff --git a/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java b/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java
index ad2eddaec023..7d5ee2b322aa 100644
--- a/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java
+++ b/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java
@@ -276,6 +276,30 @@ public interface GuardrailsMBean
      */
     void setDropTruncateTableEnabled(boolean enabled);
 
+    /**
+     * Returns whether DDL statement is allowed
+     *
+     * @return {@code true} if allowed, {@code false} otherwise.
+     */
+    boolean getDDLEnabled();
+
+    /**
+     * Sets whether DDL statement is allowed
+     */
+    void setDDLEnabled(boolean enabled);
+
+    /**
+     * Returns whether DCL statement is allowed
+     *
+     * @return {@code true} if allowed, {@code false} otherwise.
+     */
+    boolean getDCLEnabled();
+
+    /**
+     * Sets whether DCL statement is allowed
+     */
+    void setDCLEnabled(boolean enabled);
+
     /**
      * @return The threshold to warn when requested page size greater than threshold.
      * -1 means disabled.
diff --git a/test/unit/org/apache/cassandra/db/guardrails/GuardrailDCLEnabledTest.java b/test/unit/org/apache/cassandra/db/guardrails/GuardrailDCLEnabledTest.java
new file mode 100644
index 000000000000..e0f0dc6dc523
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/guardrails/GuardrailDCLEnabledTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.db.guardrails;
+
+import java.net.InetSocketAddress;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.auth.AuthenticatedUser;
+import org.apache.cassandra.service.ClientState;
+
+import static java.lang.String.format;
+
+public class GuardrailDCLEnabledTest extends GuardrailTester
+{
+    private static final String TEST_USER = "testuser";
+    private static final String TEST_PW = "testpassword";
+    private static final String TEST_USER1 = "testuser1";
+    private static final String TEST_PW1 = "testpassword1";
+    private static final String TEST_KS = "dclks";
+    private static final String TEST_TABLE = "dcltbl";
+    private static final String DCL_ERROR_MSG = "DCL statement is not allowed";
+    private ClientState loginUserClientState;
+
+    private void setGuardrail(boolean enabled)
+    {
+        Guardrails.instance.setDCLEnabled(enabled);
+    }
+
+    @Before
+    public void beforeGuardrailTest() throws Throwable
+    {
+        super.beforeGuardrailTest();
+        // create user in login state
+        useSuperUser();
+        executeNet(getCreateRoleCQL(TEST_USER, true, false, TEST_PW));
+        executeNet(format("GRANT ALL ON KEYSPACE %s TO %s", KEYSPACE, TEST_USER));
+        useUser(TEST_USER, TEST_PW);
+
+        loginUserClientState = ClientState.forExternalCalls(InetSocketAddress.createUnresolved("127.0.0.1", 1234));
+        loginUserClientState.login(new AuthenticatedUser(TEST_USER));
+        execute(loginUserClientState, "USE " + keyspace());
+
+        execute(superClientState, String.format("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}", TEST_KS));
+        execute(superClientState, String.format("CREATE TABLE IF NOT EXISTS %s.%s (key text PRIMARY KEY, col1 int, col2 int)", TEST_KS, TEST_TABLE));
+    }
+
+    @After
+    public void afterTest()
+    {
+        setGuardrail(true);
+    }
+
+    @Test
+    public void testCannotCreateRoleWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        assertFails(() -> execute(loginUserClientState,
+                                  getCreateRoleCQL(TEST_USER1, true, false, TEST_PW1)),
+                    DCL_ERROR_MSG);
+        // no role is created
+        assertEmpty(execute(String.format("SELECT * FROM system_auth.roles WHERE role='%s'", TEST_USER1)));
+    }
+
+    @Test
+    public void testCannotGrantPermissionWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        assertFails(() -> execute(loginUserClientState,
+                                  getGrantPermissionCQL(TEST_USER, TEST_KS, TEST_TABLE)),
+                    DCL_ERROR_MSG);
+        // TEST_USER don't get permission on TEST_KS.TEST_TABLE
+        assertEmpty(execute(String.format("SELECT * FROM system_auth.role_permissions WHERE role='%s' AND resource='data/%s/%s'",
+                                          TEST_USER, TEST_KS, TEST_TABLE)));
+    }
+
+    @Test
+    public void testCannotRevokePermissionWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        assertFails(() -> execute(loginUserClientState,
+                                  String.format("REVOKE ALL ON KEYSPACE %s FROM %s", KEYSPACE, TEST_USER)),
+                    DCL_ERROR_MSG);
+        // TEST_USER permission wasn't revoked on KEYSPACE
+        assertRowCount(execute(String.format("SELECT * FROM system_auth.role_permissions WHERE role='%s' AND resource='data/%s'", TEST_USER, KEYSPACE)),
+                       1);
+    }
+
+    private static String getCreateRoleCQL(String role, boolean login, boolean superUser, String password)
+    {
+        return String.format("CREATE ROLE IF NOT EXISTS %s WITH LOGIN = %s AND SUPERUSER = %s AND PASSWORD = '%s'",
+                             role, login, superUser, password);
+    }
+
+    private static String getGrantPermissionCQL(String role, String ks, String tbl)
+    {
+        return String.format("GRANT ALL PERMISSIONS ON %s.%s TO %s;", ks, tbl, role);
+    }
+}
diff --git a/test/unit/org/apache/cassandra/db/guardrails/GuardrailDDLEnabledTest.java b/test/unit/org/apache/cassandra/db/guardrails/GuardrailDDLEnabledTest.java
new file mode 100644
index 000000000000..40f22d3d52dc
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/guardrails/GuardrailDDLEnabledTest.java
@@ -0,0 +1,420 @@
+/*
+ * 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.db.guardrails;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.exceptions.AlreadyExistsException;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.schema.SchemaKeyspaceTables;
+
+import static org.junit.Assert.fail;
+
+public class GuardrailDDLEnabledTest extends GuardrailTester
+{
+    private static final String TEST_KS = "ddlks";
+    private static final String TEST_TABLE = "ddltbl";
+    private static final String TEST_VIEW = "ddlview";
+    private static final String TEST_KS_NEW = "ddlks2";
+    private static final String TEST_TABLE_NEW = "ddltbl2";
+    private static final String TEST_VIEW_NEW = "ddlview2";
+    private static final String DDL_ERROR_MSG = "DDL statement is not allowed";
+
+    private void setGuardrail(boolean enabled)
+    {
+        Guardrails.instance.setDDLEnabled(enabled);
+    }
+
+    @Before
+    public void beforeGuardrailTest() throws Throwable
+    {
+        super.beforeGuardrailTest();
+        // create keyspace ddlks and table ddltbl
+        execute(superClientState, getCreateKeyspaceCQL(TEST_KS, true));
+        execute(superClientState, getCreateTableCQL(TEST_KS, TEST_TABLE, true));
+        execute(superClientState, getCreateViewCQL(TEST_KS, TEST_VIEW, true));
+        assertRowCount(execute(getSystemSchemaKeyspaceCQL(TEST_KS)), 1);
+        assertRowCount(execute(getSystemSchemaTableCQL(TEST_KS, TEST_TABLE)), 1);
+        assertRowCount(execute(getSystemSchemaViewCQL(TEST_KS, TEST_VIEW)), 1);
+    }
+
+    @After
+    public void afterTest()
+    {
+        setGuardrail(true);
+    }
+
+    @Test
+    public void testCannotCreateKeyspaceWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // CREATE should not fail if: keyspace exists and IF NOT EXISTS specified (no-op)
+        execute(userClientState, getCreateKeyspaceCQL(TEST_KS, true));
+
+        // CREATE will fail with guardrail exception if user tries to create a keyspace
+        assertFails(() -> execute(userClientState,
+                                  getCreateKeyspaceCQL(TEST_KS_NEW, false)),
+                    DDL_ERROR_MSG);
+
+        // CREATE will also fail if user doesn't specify IF NOT EXISTS but create an already existing keyspace
+        try
+        {
+            execute(userClientState, getCreateKeyspaceCQL(TEST_KS, false));
+        }
+        catch (AlreadyExistsException e)
+        {
+            // expected
+        }
+        catch (Exception e)
+        {
+            fail(String.format("failed with unexpected error: %s", e.getMessage()));
+        }
+
+        // No new keyspace should be created
+        assertEmpty(execute(getSystemSchemaKeyspaceCQL(TEST_KS_NEW)));
+    }
+
+    @Test
+    public void testCannotCreateTableWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // CREATE should not fail if: table exists and IF NOT EXISTS specified (no-op)
+        execute(userClientState, getCreateTableCQL(TEST_KS, TEST_TABLE, true));
+
+        // CREATE will fail with guardrail exception if user tries to create a table
+        assertFails(() -> execute(userClientState,
+                                  getCreateTableCQL(TEST_KS, TEST_TABLE_NEW, false)),
+                    DDL_ERROR_MSG);
+
+        // CREATE will also fail if user doesn't specify IF NOT EXISTS but create an already existing table
+        try
+        {
+            execute(userClientState, getCreateTableCQL(TEST_KS, TEST_TABLE, false));
+        }
+        catch (AlreadyExistsException e)
+        {
+            // expected
+        }
+        catch (Exception e)
+        {
+            fail(String.format("failed with unexpected error: %s", e.getMessage()));
+        }
+
+        // No new table should be created
+        assertEmpty(execute(getSystemSchemaTableCQL(TEST_KS, TEST_TABLE_NEW)));
+    }
+
+    @Test
+    public void testCannotCreateViewWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // CREATE should not fail if: view exists and IF NOT EXISTS specified (no-op)
+        execute(userClientState, getCreateViewCQL(TEST_KS, TEST_VIEW, true));
+
+        // CREATE will fail with guardrail exception if user tries to create a view
+        assertFails(() -> execute(userClientState,
+                                  getCreateViewCQL(TEST_KS, TEST_VIEW_NEW, false)),
+                    DDL_ERROR_MSG);
+
+        // CREATE will also fail if user doesn't specify IF NOT EXISTS but create an already existing view.
+        try
+        {
+            execute(userClientState, getCreateViewCQL(TEST_KS, TEST_VIEW, false));
+        }
+        catch (AlreadyExistsException e)
+        {
+            // expected
+        }
+        catch (Exception e)
+        {
+            fail(String.format("failed with unexpected error: %s", e.getMessage()));
+        }
+
+        // No new view should be created
+        assertEmpty(execute(getSystemSchemaViewCQL(TEST_KS, TEST_TABLE_NEW)));
+    }
+
+    @Test
+    public void testCannotDropKeyspaceWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // DROP should not fail if: keyspace not exists and IF EXISTS specified (no-op)
+        execute(userClientState, getDropKeyspaceCQL(TEST_KS_NEW, true));
+
+        // DROP will fail with guardrail exception if user tries to drop an existing keyspace
+        assertFails(() -> execute(userClientState,
+                                  getDropKeyspaceCQL(TEST_KS, false)),
+                    DDL_ERROR_MSG);
+
+        // DROP will also fail if user doesn't specify IF EXISTS but keyspace doesn't exist
+        try
+        {
+            execute(userClientState, getDropKeyspaceCQL(TEST_KS_NEW, false));
+        }
+        catch (InvalidRequestException e)
+        {
+            // expected
+        }
+        catch (Exception e)
+        {
+            fail(String.format("failed with unexpected error: %s", e.getMessage()));
+        }
+
+        // TEST_KS should not be dropped
+        assertRowCount(execute(getSystemSchemaKeyspaceCQL(TEST_KS)), 1);
+    }
+
+    @Test
+    public void testCannotDropTableWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // DROP should not fail if: table not exists and IF EXISTS specified (no-op)
+        execute(userClientState, getDropTableCQL(TEST_KS, TEST_TABLE_NEW, true));
+
+        // DROP will fail with guardrail exception if user tries to drop an existing table
+        assertFails(() -> execute(userClientState,
+                                  getDropTableCQL(TEST_KS, TEST_TABLE, false)),
+                    DDL_ERROR_MSG);
+
+        // DROP will also fail if user doesn't specify IF EXISTS but table doesn't exist
+        try
+        {
+            execute(userClientState, getDropTableCQL(TEST_KS, TEST_TABLE_NEW, false));
+        }
+        catch (InvalidRequestException e)
+        {
+            // expected
+        }
+        catch (Exception e)
+        {
+            fail(String.format("failed with unexpected error: %s", e.getMessage()));
+        }
+
+        // TEST_TABLE should not be dropped
+        assertRowCount(execute(getSystemSchemaTableCQL(TEST_KS, TEST_TABLE)), 1);
+    }
+
+    @Test
+    public void testCannotDropViewWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // DROP should not fail if: view not exists and IF EXISTS specified (no-op)
+        execute(userClientState, getDropViewCQL(TEST_KS, TEST_VIEW_NEW, true));
+
+        // DROP will fail with guardrail exception if user tries to drop an existing view
+        assertFails(() -> execute(userClientState,
+                                  getDropViewCQL(TEST_KS, TEST_VIEW, false)),
+                    DDL_ERROR_MSG);
+
+        // DROP will also fail if user doesn't specify IF EXISTS but view doesn't exist
+        try
+        {
+            execute(userClientState, getDropViewCQL(TEST_KS, TEST_VIEW_NEW, false));
+        }
+        catch (InvalidRequestException e)
+        {
+            // expected
+        }
+        catch (Exception e)
+        {
+            fail(String.format("failed with unexpected error: %s", e.getMessage()));
+        }
+
+        // TEST_VIEW should not be dropped
+        assertRowCount(execute(getSystemSchemaViewCQL(TEST_KS, TEST_VIEW)), 1);
+    }
+
+    @Test
+    public void testCannotDropColumnWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // table have column col1
+        assertRowCount(execute(String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND table_name='%s' AND column_name='col1'",
+                                             SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                                             SchemaKeyspaceTables.COLUMNS,
+                                             TEST_KS, TEST_TABLE)),
+                       1);
+        // ALTER TABLE drop column should not fail if: table exists and the column doesn't exist and user specify IF EXISTS (no-op)
+        execute(userClientState, String.format("ALTER TABLE %s.%s DROP IF EXISTS col3", TEST_KS, TEST_TABLE));
+
+        // ALTER TABLE will fail with guardrail excepetion if user tries to add new column to this table
+        assertFails(() -> execute(userClientState,
+                                  String.format("ALTER TABLE %s.%s DROP col1", TEST_KS, TEST_TABLE)),
+                    DDL_ERROR_MSG);
+
+        // column col1 should not be dropped
+        assertRowCount(execute(String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND table_name='%s' AND column_name='col1'",
+                                             SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                                             SchemaKeyspaceTables.COLUMNS,
+                                             TEST_KS, TEST_TABLE)),
+                       1);
+    }
+
+    @Test
+    public void testCannotAlterKeyspaceWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // keyspace should have durable_write=true by default
+        assertEmpty(execute(String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND durable_writes=false ALLOW FILTERING",
+                                          SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                                          SchemaKeyspaceTables.KEYSPACES,
+                                          TEST_KS)));
+        // ALTER KEYSPACE should not fail if: keyspace not exists and IF EXISTS specified (no-op)
+        execute(userClientState, String.format("ALTER KEYSPACE IF EXISTS %s WITH durable_writes=false", TEST_KS_NEW));
+        // ALTER TABLE will fail with guardrail excepetion if user tries to alter anything related to this table
+        assertFails(() -> execute(userClientState,
+                                  String.format("ALTER KEYSPACE %s WITH durable_writes=false", TEST_KS)),
+                    DDL_ERROR_MSG);
+
+        // keyspace should still have durable_write=true
+        assertEmpty(execute(String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND durable_writes=false ALLOW FILTERING",
+                                          SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                                          SchemaKeyspaceTables.KEYSPACES,
+                                          TEST_KS)));
+    }
+
+    @Test
+    public void testCannotAlterTableWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // table doesn't have comment
+        assertEmpty(execute(String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND table_name='%s' AND comment='test' ALLOW FILTERING",
+                                          SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                                          SchemaKeyspaceTables.TABLES,
+                                          TEST_KS, TEST_TABLE)));
+        // ALTER TABLE should not fail if: table not exists and IF EXISTS specified (no-op)
+        execute(userClientState, String.format("ALTER TABLE IF EXISTS %s.%s WITH comment='test'", TEST_KS, TEST_TABLE_NEW));
+        // ALTER TABLE will fail with guardrail excepetion if user tries to alter anything related to this table
+        assertFails(() -> execute(userClientState,
+                                  String.format("ALTER TABLE %s.%s WITH comment='test'", TEST_KS, TEST_TABLE)),
+                    DDL_ERROR_MSG);
+
+        // table should not have comment
+        assertEmpty(execute(String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND table_name='%s' AND comment='test' ALLOW FILTERING",
+                                          SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                                          SchemaKeyspaceTables.TABLES,
+                                          TEST_KS, TEST_TABLE)));
+    }
+
+    @Test
+    public void testCannotAddColumnWhileFeatureDisabled() throws Throwable
+    {
+        setGuardrail(false);
+        // table doesn't have new column col3
+        assertEmpty(execute(String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND table_name='%s' AND column_name='col3'",
+                                          SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                                          SchemaKeyspaceTables.COLUMNS,
+                                          TEST_KS, TEST_TABLE)));
+        // ALTER TABLE add column should not fail if: table not exists and IF EXISTS specified (no-op)
+        execute(userClientState, String.format("ALTER TABLE IF EXISTS %s.%s ADD col3 text", TEST_KS, TEST_TABLE_NEW));
+        // ALTER TABLE will fail with guardrail excepetion if user tries to add new column to this table
+        assertFails(() -> execute(userClientState,
+                                  String.format("ALTER TABLE %s.%s ADD col3 text", TEST_KS, TEST_TABLE)),
+                    DDL_ERROR_MSG);
+
+        // table should not have new column col3
+        assertEmpty(execute(String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND table_name='%s' AND column_name='col3'",
+                                          SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                                          SchemaKeyspaceTables.COLUMNS,
+                                          TEST_KS, TEST_TABLE)));
+    }
+
+    private String getCreateKeyspaceCQL(String ks, boolean ifNotExists)
+    {
+        if (ifNotExists)
+        {
+            return String.format("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}", ks);
+        }
+        return String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}", ks);
+    }
+
+    private String getDropKeyspaceCQL(String ks, boolean ifExists)
+    {
+        if (ifExists)
+        {
+            return String.format("DROP KEYSPACE IF EXISTS %s", ks);
+        }
+        return String.format("DROP KEYSPACE %s", ks);
+    }
+
+    private String getCreateTableCQL(String ks, String table, boolean ifNotExists)
+    {
+        if (ifNotExists)
+        {
+            return String.format("CREATE TABLE IF NOT EXISTS %s.%s (key text PRIMARY KEY, col1 text, col2 text)", ks, table);
+        }
+        return String.format("CREATE TABLE %s.%s (key text PRIMARY KEY, col1 text, col2 text)", ks, table);
+    }
+
+    private String getDropTableCQL(String ks, String table, boolean ifExists)
+    {
+        if (ifExists)
+        {
+            return String.format("DROP TABLE IF EXISTS %s.%s", ks, table);
+        }
+        return String.format("DROP TABLE %s.%s", ks, table);
+    }
+
+    private String getCreateViewCQL(String ks, String table, boolean ifNotExists)
+    {
+        if (ifNotExists)
+        {
+            return String.format("CREATE MATERIALIZED VIEW IF NOT EXISTS %s.%s AS SELECT key FROM %s.%s WHERE key IS NOT NULL PRIMARY KEY (key)",
+                                 ks, table, TEST_KS, TEST_TABLE);
+        }
+        return String.format("CREATE MATERIALIZED VIEW %s.%s AS SELECT key FROM %s.%s WHERE key IS NOT NULL PRIMARY KEY (key)",
+                             ks, table, TEST_KS, TEST_TABLE);
+    }
+
+    private String getDropViewCQL(String ks, String view, boolean ifExists)
+    {
+        if (ifExists)
+        {
+            return String.format("DROP MATERIALIZED VIEW IF EXISTS %s.%s", ks, view);
+        }
+        return String.format("DROP MATERIALIZED VIEW %s.%s", ks, view);
+    }
+
+    private String getSystemSchemaKeyspaceCQL(String ks)
+    {
+        return String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s'",
+                             SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                             SchemaKeyspaceTables.KEYSPACES,
+                             ks);
+    }
+
+    private String getSystemSchemaTableCQL(String ks, String table)
+    {
+        return String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND table_name='%s'",
+                             SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                             SchemaKeyspaceTables.TABLES,
+                             ks, table);
+    }
+
+    private String getSystemSchemaViewCQL(String ks, String view)
+    {
+        return String.format("SELECT * FROM %s.%s WHERE keyspace_name='%s' AND view_name='%s'",
+                             SchemaConstants.SCHEMA_KEYSPACE_NAME,
+                             SchemaKeyspaceTables.VIEWS,
+                             ks, view);
+    }
+}