diff --git a/conf/scylla.yaml b/conf/scylla.yaml
index 9c5ede9263..1c3afc7cc3 100644
--- a/conf/scylla.yaml
+++ b/conf/scylla.yaml
@@ -573,3 +573,20 @@ force_schema_commit_log: true
 # A cluster not using Raft can be 'upgraded' to use Raft. Refer to the aforementioned
 # documentation, section 'Enabling Raft in ScyllaDB 5.2 and further', for the procedure.
 consistent_cluster_management: true
+
+# In materialized views, restrictions are allowed only on the view's primary key columns.
+# In old versions Scylla mistakenly allowed IS NOT NULL restrictions on columns which were not part
+# of the view's primary key. These invalid restrictions were ignored.
+# This option controls the behavior when someone tries to create a view with such invalid IS NOT NULL restrictions.
+#
+# Can be true, false, or warn.
+# * `true`: IS NOT NULL is allowed only on the view's primary key columns,
+#           trying to use it on other columns will cause an error, as it should.
+# * `false`: Scylla accepts IS NOT NULL restrictions on regular columns, but they're silently ignored.
+#            It's useful for backwards compatibility.
+# * `warn`: The same as false, but there's a warning about invalid view restrictions.
+#
+# To preserve backwards compatibility on old clusters, Scylla's default setting is `warn`.
+# New clusters have this option set to `true` by scylla.yaml (which overrides the default `warn`)
+# to make sure that trying to create an invalid view causes an error.
+strict_is_not_null_in_views: true
diff --git a/cql3/cql_statement.hh b/cql3/cql_statement.hh
index fc0853ae0b..10f15e7a8c 100644
--- a/cql3/cql_statement.hh
+++ b/cql3/cql_statement.hh
@@ -39,6 +39,9 @@ seastar::shared_ptr<const metadata> make_empty_metadata();
 
 class query_options;
 
+// A vector of CQL warnings generated during execution of a statement.
+using cql_warnings_vec = std::vector<sstring>;
+
 class cql_statement {
     timeout_config_selector _timeout_config_selector;
 public:
diff --git a/cql3/restrictions/statement_restrictions.cc b/cql3/restrictions/statement_restrictions.cc
index 4f93f4d57f..bf21688967 100644
--- a/cql3/restrictions/statement_restrictions.cc
+++ b/cql3/restrictions/statement_restrictions.cc
@@ -1990,5 +1990,10 @@ void statement_restrictions::validate_primary_key(const query_options& options)
     validate_primary_key_restrictions(options, _clustering_prefix_restrictions);
 }
 
+
+const std::unordered_set<const column_definition*> statement_restrictions::get_not_null_columns() const {
+    return _not_null_columns;
+}
+
 } // namespace restrictions
 } // namespace cql3
diff --git a/cql3/restrictions/statement_restrictions.hh b/cql3/restrictions/statement_restrictions.hh
index 562bc487c4..773da0d483 100644
--- a/cql3/restrictions/statement_restrictions.hh
+++ b/cql3/restrictions/statement_restrictions.hh
@@ -180,6 +180,10 @@ public:
         return _clustering_columns_restrictions;
     }
 
+    // Get a set of columns restricted by the IS NOT NULL restriction.
+    // IS NOT NULL is a special case that is handled separately from other restrictions.
+    const std::unordered_set<const column_definition*> get_not_null_columns() const;
+
     bool has_token_restrictions() const {
         return has_partition_token(_partition_key_restrictions, *_schema);
     }
diff --git a/cql3/statements/alter_keyspace_statement.cc b/cql3/statements/alter_keyspace_statement.cc
index ddf154dd74..fb250da97f 100644
--- a/cql3/statements/alter_keyspace_statement.cc
+++ b/cql3/statements/alter_keyspace_statement.cc
@@ -74,7 +74,7 @@ void cql3::statements::alter_keyspace_statement::validate(query_processor& qp, c
 #endif
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     try {
         auto old_ksm = qp.db().find_keyspace(_name).metadata();
@@ -88,9 +88,9 @@ cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_proce
                 event::schema_change::target_type::KEYSPACE,
                 keyspace());
 
-        return make_ready_future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>(std::make_pair(std::move(ret), std::move(m)));
+        return make_ready_future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>(std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>()));
     } catch (data_dictionary::no_such_keyspace& e) {
-        return make_exception_future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>(exceptions::invalid_request_exception("Unknown keyspace " + _name));
+        return make_exception_future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>(exceptions::invalid_request_exception("Unknown keyspace " + _name));
     }
 }
 
diff --git a/cql3/statements/alter_keyspace_statement.hh b/cql3/statements/alter_keyspace_statement.hh
index e30c64d523..4abddd960d 100644
--- a/cql3/statements/alter_keyspace_statement.hh
+++ b/cql3/statements/alter_keyspace_statement.hh
@@ -33,7 +33,7 @@ public:
 
     future<> check_access(query_processor& qp, const service::client_state& state) const override;
     void validate(query_processor& qp, const service::client_state& state) const override;
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
     virtual future<::shared_ptr<messages::result_message>> execute(query_processor& qp, service::query_state& state, const query_options& options) const override;
 };
diff --git a/cql3/statements/alter_table_statement.cc b/cql3/statements/alter_table_statement.cc
index f9ce6ec573..7541e0267a 100644
--- a/cql3/statements/alter_table_statement.cc
+++ b/cql3/statements/alter_table_statement.cc
@@ -385,7 +385,7 @@ std::pair<schema_builder, std::vector<view_ptr>> alter_table_statement::prepare_
     return make_pair(std::move(cfm), std::move(view_updates));
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 alter_table_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
   data_dictionary::database db = qp.db();
   auto& mm = qp.get_migration_manager();
@@ -399,7 +399,7 @@ alter_table_statement::prepare_schema_mutations(query_processor& qp, api::timest
             keyspace(),
             column_family());
 
-  co_return std::make_pair(std::move(ret), std::move(m));
+  co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/alter_table_statement.hh b/cql3/statements/alter_table_statement.hh
index 033cb439af..2f5b85434c 100644
--- a/cql3/statements/alter_table_statement.hh
+++ b/cql3/statements/alter_table_statement.hh
@@ -55,7 +55,7 @@ public:
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
     virtual future<::shared_ptr<messages::result_message>> execute(query_processor& qp, service::query_state& state, const query_options& options) const override;
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 private:
     void add_column(const schema& schema, data_dictionary::table cf, schema_builder& cfm, std::vector<view_ptr>& view_updates, const column_identifier& column_name, const cql3_type validator, const column_definition* def, bool is_static) const;
     void alter_column(const schema& schema, data_dictionary::table cf, schema_builder& cfm, std::vector<view_ptr>& view_updates, const column_identifier& column_name, const cql3_type validator, const column_definition* def, bool is_static) const;
diff --git a/cql3/statements/alter_type_statement.cc b/cql3/statements/alter_type_statement.cc
index f9af0166e2..a8252fdd1d 100644
--- a/cql3/statements/alter_type_statement.cc
+++ b/cql3/statements/alter_type_statement.cc
@@ -103,7 +103,7 @@ future<std::vector<mutation>> alter_type_statement::prepare_announcement_mutatio
     co_return m;
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 alter_type_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     try {
         auto m = co_await prepare_announcement_mutations(qp.db(), qp.get_migration_manager(), ts);
@@ -115,7 +115,7 @@ alter_type_statement::prepare_schema_mutations(query_processor& qp, api::timesta
                 keyspace(),
                 _name.get_string_type_name());
 
-        co_return std::make_pair(std::move(ret), std::move(m));
+        co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
     } catch(data_dictionary::no_such_keyspace& e) {
         auto&& ex = std::make_exception_ptr(exceptions::invalid_request_exception(format("Cannot alter type in unknown keyspace {}", keyspace())));
         co_return coroutine::exception(std::move(ex));
diff --git a/cql3/statements/alter_type_statement.hh b/cql3/statements/alter_type_statement.hh
index 194eb4bc91..74bfc67346 100644
--- a/cql3/statements/alter_type_statement.hh
+++ b/cql3/statements/alter_type_statement.hh
@@ -41,7 +41,7 @@ public:
     virtual const sstring& keyspace() const override;
 
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
     class add_or_alter;
     class renames;
diff --git a/cql3/statements/alter_view_statement.cc b/cql3/statements/alter_view_statement.cc
index b90787b541..7a8c17f77f 100644
--- a/cql3/statements/alter_view_statement.cc
+++ b/cql3/statements/alter_view_statement.cc
@@ -81,7 +81,7 @@ view_ptr alter_view_statement::prepare_view(data_dictionary::database db) const
     return view_ptr(builder.build());
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> alter_view_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> alter_view_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     auto m = co_await qp.get_migration_manager().prepare_view_update_announcement(prepare_view(qp.db()), ts);
 
     using namespace cql_transport;
@@ -91,7 +91,7 @@ future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<
             keyspace(),
             column_family());
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/alter_view_statement.hh b/cql3/statements/alter_view_statement.hh
index bdd5a45c50..fa900e516f 100644
--- a/cql3/statements/alter_view_statement.hh
+++ b/cql3/statements/alter_view_statement.hh
@@ -36,7 +36,7 @@ public:
     virtual void validate(query_processor&, const service::client_state& state) const override;
 
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
 };
diff --git a/cql3/statements/create_aggregate_statement.cc b/cql3/statements/create_aggregate_statement.cc
index 9a97594f8f..0073626c7a 100644
--- a/cql3/statements/create_aggregate_statement.cc
+++ b/cql3/statements/create_aggregate_statement.cc
@@ -76,7 +76,7 @@ std::unique_ptr<prepared_statement> create_aggregate_statement::prepare(data_dic
     return std::make_unique<prepared_statement>(make_shared<create_aggregate_statement>(*this));
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 create_aggregate_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
@@ -87,7 +87,7 @@ create_aggregate_statement::prepare_schema_mutations(query_processor& qp, api::t
         ret = create_schema_change(*aggregate, true);
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 seastar::future<> create_aggregate_statement::check_access(query_processor &qp, const service::client_state &state) const {
diff --git a/cql3/statements/create_aggregate_statement.hh b/cql3/statements/create_aggregate_statement.hh
index 29ee67751c..1a0419903d 100644
--- a/cql3/statements/create_aggregate_statement.hh
+++ b/cql3/statements/create_aggregate_statement.hh
@@ -25,7 +25,7 @@ namespace statements {
 
 class create_aggregate_statement final : public create_function_statement_base {
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
     virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
 
     virtual seastar::future<shared_ptr<db::functions::function>> create(query_processor& qp, db::functions::function* old) const override;
diff --git a/cql3/statements/create_function_statement.cc b/cql3/statements/create_function_statement.cc
index d4b0d14b86..998a19727a 100644
--- a/cql3/statements/create_function_statement.cc
+++ b/cql3/statements/create_function_statement.cc
@@ -65,7 +65,7 @@ std::unique_ptr<prepared_statement> create_function_statement::prepare(data_dict
     return std::make_unique<prepared_statement>(make_shared<create_function_statement>(*this));
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 create_function_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
@@ -77,7 +77,7 @@ create_function_statement::prepare_schema_mutations(query_processor& qp, api::ti
         ret = create_schema_change(*func, true);
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 create_function_statement::create_function_statement(functions::function_name name, sstring language, sstring body,
diff --git a/cql3/statements/create_function_statement.hh b/cql3/statements/create_function_statement.hh
index f020698829..3571ad980e 100644
--- a/cql3/statements/create_function_statement.hh
+++ b/cql3/statements/create_function_statement.hh
@@ -23,7 +23,7 @@ namespace statements {
 
 class create_function_statement final : public create_function_statement_base {
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
     virtual seastar::future<shared_ptr<db::functions::function>> create(query_processor& qp, db::functions::function* old) const override;
     sstring _language;
diff --git a/cql3/statements/create_index_statement.cc b/cql3/statements/create_index_statement.cc
index 5299368843..dc224ee538 100644
--- a/cql3/statements/create_index_statement.cc
+++ b/cql3/statements/create_index_statement.cc
@@ -375,7 +375,7 @@ schema_ptr create_index_statement::build_index_schema(query_processor& qp) const
     return builder.build();
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 create_index_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     using namespace cql_transport;
     auto schema = build_index_schema(qp);
@@ -393,7 +393,7 @@ create_index_statement::prepare_schema_mutations(query_processor& qp, api::times
                 column_family());
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/create_index_statement.hh b/cql3/statements/create_index_statement.hh
index 806d2c28ee..66cb468880 100644
--- a/cql3/statements/create_index_statement.hh
+++ b/cql3/statements/create_index_statement.hh
@@ -47,7 +47,7 @@ public:
 
     future<> check_access(query_processor& qp, const service::client_state& state) const override;
     void validate(query_processor&, const service::client_state& state) const override;
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
diff --git a/cql3/statements/create_keyspace_statement.cc b/cql3/statements/create_keyspace_statement.cc
index 8a46602016..63dec6b8dd 100644
--- a/cql3/statements/create_keyspace_statement.cc
+++ b/cql3/statements/create_keyspace_statement.cc
@@ -93,7 +93,7 @@ void create_keyspace_statement::validate(query_processor& qp, const service::cli
 #endif
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> create_keyspace_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> create_keyspace_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     using namespace cql_transport;
     const auto& tm = *qp.proxy().get_token_metadata_ptr();
     ::shared_ptr<event::schema_change> ret;
@@ -112,7 +112,7 @@ future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<
         }
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/create_keyspace_statement.hh b/cql3/statements/create_keyspace_statement.hh
index 891b6f2c62..55d14ae361 100644
--- a/cql3/statements/create_keyspace_statement.hh
+++ b/cql3/statements/create_keyspace_statement.hh
@@ -64,7 +64,7 @@ public:
     virtual void validate(query_processor&, const service::client_state& state) const override;
 
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
 
diff --git a/cql3/statements/create_table_statement.cc b/cql3/statements/create_table_statement.cc
index 25dd412da4..8c8ecdb67e 100644
--- a/cql3/statements/create_table_statement.cc
+++ b/cql3/statements/create_table_statement.cc
@@ -74,7 +74,7 @@ std::vector<column_definition> create_table_statement::get_columns() const
     return column_defs;
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 create_table_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
@@ -94,7 +94,7 @@ create_table_statement::prepare_schema_mutations(query_processor& qp, api::times
         }
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 /**
diff --git a/cql3/statements/create_table_statement.hh b/cql3/statements/create_table_statement.hh
index 5299d1653f..a4214aad69 100644
--- a/cql3/statements/create_table_statement.hh
+++ b/cql3/statements/create_table_statement.hh
@@ -72,7 +72,7 @@ public:
 
     virtual void validate(query_processor&, const service::client_state& state) const override;
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
diff --git a/cql3/statements/create_type_statement.cc b/cql3/statements/create_type_statement.cc
index e92e0f7459..b6a95f3751 100644
--- a/cql3/statements/create_type_statement.cc
+++ b/cql3/statements/create_type_statement.cc
@@ -118,7 +118,7 @@ std::optional<user_type> create_type_statement::make_type(query_processor& qp) c
     return type;
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> create_type_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> create_type_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
     try {
@@ -141,7 +141,7 @@ future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<
         co_return coroutine::exception(std::current_exception());
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/create_type_statement.hh b/cql3/statements/create_type_statement.hh
index d6fdf4106f..fe3252a45b 100644
--- a/cql3/statements/create_type_statement.hh
+++ b/cql3/statements/create_type_statement.hh
@@ -37,7 +37,7 @@ public:
 
     virtual const sstring& keyspace() const override;
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
 
diff --git a/cql3/statements/create_view_statement.cc b/cql3/statements/create_view_statement.cc
index fdf530e031..8349a2237e 100644
--- a/cql3/statements/create_view_statement.cc
+++ b/cql3/statements/create_view_statement.cc
@@ -112,7 +112,7 @@ static bool validate_primary_key(
     return new_non_pk_column;
 }
 
-view_ptr create_view_statement::prepare_view(data_dictionary::database db) const {
+std::pair<view_ptr, cql3::cql_warnings_vec> create_view_statement::prepare_view(data_dictionary::database db) const {
     // We need to make sure that:
     //  - primary key includes all columns in base table's primary key
     //  - make sure that the select statement does not have anything other than columns
@@ -122,6 +122,8 @@ view_ptr create_view_statement::prepare_view(data_dictionary::database db) const
     //  - make sure there is not currently a table or view
     //  - make sure base_table gc_grace_seconds > 0
 
+    cql3::cql_warnings_vec warnings;
+
     auto schema_extensions = _properties.properties()->make_schema_extensions(db.extensions());
     _properties.validate(db, keyspace(), schema_extensions);
 
@@ -299,6 +301,35 @@ view_ptr create_view_statement::prepare_view(data_dictionary::database db) const
                 column_family(), column_names));
     }
 
+    // IS NOT NULL restrictions are handled separately from other restrictions.
+    // They need a separate check as they won't be included in non_pk_restrictions.
+    std::vector<std::string_view> invalid_not_null_column_names;
+    for (const column_definition* not_null_cdef : restrictions->get_not_null_columns()) {
+        if (!target_primary_keys.contains(not_null_cdef)) {
+            invalid_not_null_column_names.push_back(not_null_cdef->name_as_text());
+        }
+    }
+
+    if (!invalid_not_null_column_names.empty() &&
+        db.get_config().strict_is_not_null_in_views() == db::tri_mode_restriction_t::mode::TRUE) {
+        throw exceptions::invalid_request_exception(
+            fmt::format("The IS NOT NULL restriction is allowed only columns which are part of the view's primary key,"
+                        " found columns: {}. The flag strict_is_not_null_in_views can be used to turn this error "
+                        "into a warning, or to silence it. (true - error, warn - warning, false - silent)",
+                        fmt::join(invalid_not_null_column_names, ", ")));
+    }
+
+    if (!invalid_not_null_column_names.empty() &&
+        db.get_config().strict_is_not_null_in_views() == db::tri_mode_restriction_t::mode::WARN) {
+        sstring warning_text = fmt::format(
+            "The IS NOT NULL restriction is allowed only columns which are part of the view's primary key,"
+            " found columns: {}. Restrictions on these columns will be silently ignored. "
+            "The flag strict_is_not_null_in_views can be used to turn this warning into an error, or to silence it. "
+            "(true - error, warn - warning, false - silent)",
+            fmt::join(invalid_not_null_column_names, ", "));
+        warnings.emplace_back(std::move(warning_text));
+    }
+
     schema_builder builder{keyspace(), column_family()};
     auto add_columns = [this, &builder] (std::vector<const column_definition*>& defs, column_kind kind) mutable {
         for (auto* def : defs) {
@@ -334,14 +365,14 @@ view_ptr create_view_statement::prepare_view(data_dictionary::database db) const
     auto where_clause_text = util::relations_to_where_clause(_where_clause);
     builder.with_view_info(schema->id(), schema->cf_name(), included.empty(), std::move(where_clause_text));
 
-    return view_ptr(builder.build());
+    return std::make_pair(view_ptr(builder.build()), std::move(warnings));
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 create_view_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
-    auto definition = prepare_view(qp.db());
+    auto [definition, warnings] = prepare_view(qp.db());
     try {
         m = co_await qp.get_migration_manager().prepare_new_view_announcement(std::move(definition), ts);
         using namespace cql_transport;
@@ -356,7 +387,7 @@ create_view_statement::prepare_schema_mutations(query_processor& qp, api::timest
         }
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::move(warnings));
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/create_view_statement.hh b/cql3/statements/create_view_statement.hh
index 27a54742de..3a98fb9368 100644
--- a/cql3/statements/create_view_statement.hh
+++ b/cql3/statements/create_view_statement.hh
@@ -39,7 +39,7 @@ private:
     cf_properties _properties;
     bool _if_not_exists;
 
-    view_ptr prepare_view(data_dictionary::database db) const;
+    std::pair<view_ptr, cql3::cql_warnings_vec> prepare_view(data_dictionary::database db) const;
 
 public:
     create_view_statement(
@@ -58,7 +58,7 @@ public:
     // Functions we need to override to subclass schema_altering_statement
     virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
     virtual void validate(query_processor&, const service::client_state& state) const override;
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
 
diff --git a/cql3/statements/drop_aggregate_statement.cc b/cql3/statements/drop_aggregate_statement.cc
index 706132405c..007a27482a 100644
--- a/cql3/statements/drop_aggregate_statement.cc
+++ b/cql3/statements/drop_aggregate_statement.cc
@@ -23,7 +23,7 @@ std::unique_ptr<prepared_statement> drop_aggregate_statement::prepare(data_dicti
     return std::make_unique<prepared_statement>(make_shared<drop_aggregate_statement>(*this));
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 drop_aggregate_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
@@ -38,7 +38,7 @@ drop_aggregate_statement::prepare_schema_mutations(query_processor& qp, api::tim
         ret = create_schema_change(*func, false);
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 drop_aggregate_statement::drop_aggregate_statement(functions::function_name name,
diff --git a/cql3/statements/drop_aggregate_statement.hh b/cql3/statements/drop_aggregate_statement.hh
index c000b2cab5..7cad55ffef 100644
--- a/cql3/statements/drop_aggregate_statement.hh
+++ b/cql3/statements/drop_aggregate_statement.hh
@@ -15,7 +15,7 @@ class query_processor;
 namespace statements {
 class drop_aggregate_statement final : public drop_function_statement_base {
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
 public:
     drop_aggregate_statement(functions::function_name name, std::vector<shared_ptr<cql3_type::raw>> arg_types,
diff --git a/cql3/statements/drop_function_statement.cc b/cql3/statements/drop_function_statement.cc
index f531a20245..f3f1535582 100644
--- a/cql3/statements/drop_function_statement.cc
+++ b/cql3/statements/drop_function_statement.cc
@@ -23,7 +23,7 @@ std::unique_ptr<prepared_statement> drop_function_statement::prepare(data_dictio
     return std::make_unique<prepared_statement>(make_shared<drop_function_statement>(*this));
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 drop_function_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
@@ -42,7 +42,7 @@ drop_function_statement::prepare_schema_mutations(query_processor& qp, api::time
         ret = create_schema_change(*func, false);
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 drop_function_statement::drop_function_statement(functions::function_name name,
diff --git a/cql3/statements/drop_function_statement.hh b/cql3/statements/drop_function_statement.hh
index 52c47f2aa9..63db8d72ae 100644
--- a/cql3/statements/drop_function_statement.hh
+++ b/cql3/statements/drop_function_statement.hh
@@ -15,7 +15,7 @@ class query_processor;
 namespace statements {
 class drop_function_statement final : public drop_function_statement_base {
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
 public:
     drop_function_statement(functions::function_name name, std::vector<shared_ptr<cql3_type::raw>> arg_types,
diff --git a/cql3/statements/drop_index_statement.cc b/cql3/statements/drop_index_statement.cc
index be60fb8385..9500b8b335 100644
--- a/cql3/statements/drop_index_statement.cc
+++ b/cql3/statements/drop_index_statement.cc
@@ -72,7 +72,7 @@ schema_ptr drop_index_statement::make_drop_idex_schema(query_processor& qp) cons
     return builder.build();
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 drop_index_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
@@ -88,7 +88,7 @@ drop_index_statement::prepare_schema_mutations(query_processor& qp, api::timesta
                                                  cfm->cf_name());
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/drop_index_statement.hh b/cql3/statements/drop_index_statement.hh
index 950e4845c5..e25b93500c 100644
--- a/cql3/statements/drop_index_statement.hh
+++ b/cql3/statements/drop_index_statement.hh
@@ -44,7 +44,7 @@ public:
 
     virtual void validate(query_processor&, const service::client_state& state) const override;
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
 private:
diff --git a/cql3/statements/drop_keyspace_statement.cc b/cql3/statements/drop_keyspace_statement.cc
index d727a9cd25..969efed70c 100644
--- a/cql3/statements/drop_keyspace_statement.cc
+++ b/cql3/statements/drop_keyspace_statement.cc
@@ -46,7 +46,7 @@ const sstring& drop_keyspace_statement::keyspace() const
     return _keyspace;
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 drop_keyspace_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     std::vector<mutation> m;
     ::shared_ptr<cql_transport::event::schema_change> ret;
@@ -65,7 +65,7 @@ drop_keyspace_statement::prepare_schema_mutations(query_processor& qp, api::time
         }
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/drop_keyspace_statement.hh b/cql3/statements/drop_keyspace_statement.hh
index 7fe802f4f0..9f2a6b4603 100644
--- a/cql3/statements/drop_keyspace_statement.hh
+++ b/cql3/statements/drop_keyspace_statement.hh
@@ -30,7 +30,7 @@ public:
 
     virtual const sstring& keyspace() const override;
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
 };
diff --git a/cql3/statements/drop_table_statement.cc b/cql3/statements/drop_table_statement.cc
index dd2fe82241..08db18ba86 100644
--- a/cql3/statements/drop_table_statement.cc
+++ b/cql3/statements/drop_table_statement.cc
@@ -44,7 +44,7 @@ void drop_table_statement::validate(query_processor&, const service::client_stat
     // validated in prepare_schema_mutations()
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 drop_table_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
@@ -64,7 +64,7 @@ drop_table_statement::prepare_schema_mutations(query_processor& qp, api::timesta
         }
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/drop_table_statement.hh b/cql3/statements/drop_table_statement.hh
index 4ef39e3cad..08bb156615 100644
--- a/cql3/statements/drop_table_statement.hh
+++ b/cql3/statements/drop_table_statement.hh
@@ -28,7 +28,7 @@ public:
 
     virtual void validate(query_processor&, const service::client_state& state) const override;
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
diff --git a/cql3/statements/drop_type_statement.cc b/cql3/statements/drop_type_statement.cc
index a76347a8c9..6f3536845e 100644
--- a/cql3/statements/drop_type_statement.cc
+++ b/cql3/statements/drop_type_statement.cc
@@ -124,7 +124,7 @@ const sstring& drop_type_statement::keyspace() const
     return _name.get_keyspace();
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 drop_type_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     validate_while_executing(qp);
 
@@ -151,7 +151,7 @@ drop_type_statement::prepare_schema_mutations(query_processor& qp, api::timestam
                 _name.get_string_type_name());
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/drop_type_statement.hh b/cql3/statements/drop_type_statement.hh
index f3e84e0141..66abbdf37b 100644
--- a/cql3/statements/drop_type_statement.hh
+++ b/cql3/statements/drop_type_statement.hh
@@ -31,7 +31,7 @@ public:
 
     virtual const sstring& keyspace() const override;
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
diff --git a/cql3/statements/drop_view_statement.cc b/cql3/statements/drop_view_statement.cc
index 56bf9137ea..1ab67012e2 100644
--- a/cql3/statements/drop_view_statement.cc
+++ b/cql3/statements/drop_view_statement.cc
@@ -46,7 +46,7 @@ void drop_view_statement::validate(query_processor&, const service::client_state
     // validated in migration_manager::announce_view_drop()
 }
 
-future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
+future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>>
 drop_view_statement::prepare_schema_mutations(query_processor& qp, api::timestamp_type ts) const {
     ::shared_ptr<cql_transport::event::schema_change> ret;
     std::vector<mutation> m;
@@ -66,7 +66,7 @@ drop_view_statement::prepare_schema_mutations(query_processor& qp, api::timestam
         }
     }
 
-    co_return std::make_pair(std::move(ret), std::move(m));
+    co_return std::make_tuple(std::move(ret), std::move(m), std::vector<sstring>());
 }
 
 std::unique_ptr<cql3::statements::prepared_statement>
diff --git a/cql3/statements/drop_view_statement.hh b/cql3/statements/drop_view_statement.hh
index 0e8cc5e87b..9aa5e54293 100644
--- a/cql3/statements/drop_view_statement.hh
+++ b/cql3/statements/drop_view_statement.hh
@@ -34,7 +34,7 @@ public:
 
     virtual void validate(query_processor&, const service::client_state& state) const override;
 
-    future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
+    future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;
 
 
     virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
diff --git a/cql3/statements/schema_altering_statement.cc b/cql3/statements/schema_altering_statement.cc
index fd54e14f3e..1d0588430f 100644
--- a/cql3/statements/schema_altering_statement.cc
+++ b/cql3/statements/schema_altering_statement.cc
@@ -70,12 +70,15 @@ schema_altering_statement::execute0(query_processor& qp, service::query_state& s
                     std::move(const_cast<cql3::query_options&>(options).take_cached_pk_function_calls()));
     }
 
+    cql3::cql_warnings_vec warnings;
+
     auto retries = mm.get_concurrent_ddl_retries();
     while (true) {
         try {
             auto group0_guard = co_await mm.start_group0_operation();
 
-            auto [ret, m] = co_await prepare_schema_mutations(qp, group0_guard.write_timestamp());
+            auto [ret, m, cql_warnings] = co_await prepare_schema_mutations(qp, group0_guard.write_timestamp());
+            warnings = std::move(cql_warnings);
 
             if (!m.empty()) {
                 auto description = format("CQL DDL statement: \"{}\"", raw_cql_statement);
@@ -96,11 +99,18 @@ schema_altering_statement::execute0(query_processor& qp, service::query_state& s
 
     // If an IF [NOT] EXISTS clause was used, this may not result in an actual schema change.  To avoid doing
     // extra work in the drivers to handle schema changes, we return an empty message in this case. (CASSANDRA-7600)
+    ::shared_ptr<messages::result_message> result;
     if (!ce) {
-        co_return ::make_shared<messages::result_message::void_message>();
+        result = ::make_shared<messages::result_message::void_message>();
     } else {
-        co_return ::make_shared<messages::result_message::schema_change>(ce);
+        result = ::make_shared<messages::result_message::schema_change>(ce);
     }
+
+    for (const sstring& warning : warnings) {
+        result->add_warning(warning);
+    }
+
+    co_return result;
 }
 
 future<::shared_ptr<messages::result_message>>
diff --git a/cql3/statements/schema_altering_statement.hh b/cql3/statements/schema_altering_statement.hh
index 327ed17a79..3707e1088d 100644
--- a/cql3/statements/schema_altering_statement.hh
+++ b/cql3/statements/schema_altering_statement.hh
@@ -56,7 +56,7 @@ protected:
 
     virtual void prepare_keyspace(const service::client_state& state) override;
 
-    virtual future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const = 0;
+    virtual future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const = 0;
 
     virtual future<::shared_ptr<messages::result_message>>
     execute(query_processor& qp, service::query_state& state, const query_options& options) const override;
diff --git a/db/config.cc b/db/config.cc
index d1f6c55355..fdef02cb93 100644
--- a/db/config.cc
+++ b/db/config.cc
@@ -911,6 +911,20 @@ db::config::config(std::shared_ptr<db::extensions> exts)
     , cdc_dont_rewrite_streams(this, "cdc_dont_rewrite_streams", value_status::Used, false,
             "Disable rewriting streams from cdc_streams_descriptions to cdc_streams_descriptions_v2. Should not be necessary, but the procedure is expensive and prone to failures; this config option is left as a backdoor in case some user requires manual intervention.")
     , strict_allow_filtering(this, "strict_allow_filtering", liveness::LiveUpdate, value_status::Used, strict_allow_filtering_default(), "Match Cassandra in requiring ALLOW FILTERING on slow queries. Can be true, false, or warn. When false, Scylla accepts some slow queries even without ALLOW FILTERING that Cassandra rejects. Warn is same as false, but with warning.")
+    , strict_is_not_null_in_views(this, "strict_is_not_null_in_views", liveness::LiveUpdate, value_status::Used,db::tri_mode_restriction_t::mode::WARN, 
+        "In materialized views, restrictions are allowed only on the view's primary key columns.\n"
+        "In old versions Scylla mistakenly allowed IS NOT NULL restrictions on columns which were not part of the view's"
+        " primary key. These invalid restrictions were ignored.\n"
+        "This option controls the behavior when someone tries to create a view with such invalid IS NOT NULL restrictions.\n\n"
+        "Can be true, false, or warn:\n"
+        " * `true`: IS NOT NULL is allowed only on the view's primary key columns, "
+        "trying to use it on other columns will cause an error, as it should.\n"
+        " * `false`: Scylla accepts IS NOT NULL restrictions on regular columns, but they're silently ignored. "
+        "It's useful for backwards compatibility.\n"
+        " * `warn`: The same as false, but there's a warning about invalid view restrictions.\n\n"
+        "To preserve backwards compatibility on old clusters, Scylla's default setting is `warn`. "
+        "New clusters have this option set to `true` by scylla.yaml (which overrides the default `warn`), "
+        "to make sure that trying to create an invalid view causes an error.")
     , reversed_reads_auto_bypass_cache(this, "reversed_reads_auto_bypass_cache", liveness::LiveUpdate, value_status::Used, false,
             "Bypass in-memory data cache (the row cache) when performing reversed queries.")
     , enable_optimized_reversed_reads(this, "enable_optimized_reversed_reads", liveness::LiveUpdate, value_status::Used, true,
diff --git a/db/config.hh b/db/config.hh
index 6028af8406..2c0774b55b 100644
--- a/db/config.hh
+++ b/db/config.hh
@@ -377,6 +377,7 @@ public:
     named_value<uint32_t> max_concurrent_requests_per_shard;
     named_value<bool> cdc_dont_rewrite_streams;
     named_value<tri_mode_restriction> strict_allow_filtering;
+    named_value<tri_mode_restriction> strict_is_not_null_in_views;
     named_value<bool> reversed_reads_auto_bypass_cache;
     named_value<bool> enable_optimized_reversed_reads;
     named_value<bool> enable_cql_config_updates;
diff --git a/docs/cql/mv.rst b/docs/cql/mv.rst
index fad3274a8b..3ec5a5159c 100644
--- a/docs/cql/mv.rst
+++ b/docs/cql/mv.rst
@@ -82,8 +82,7 @@ statement is limited in a number of ways:
 - The ``WHERE`` clause has the following restrictions:
 
   - It cannot include any :token:`bind_marker`.
-  - The columns that are not part of the *base table* primary key can only be restricted by an ``IS NOT NULL``
-    restriction. No other restriction is allowed.
+  - The columns that are not part of the *view table* primary key can't be restricted.
   - As the columns that are part of the *view* primary key cannot be null, they must always be at least restricted by a
     ``IS NOT NULL`` restriction (or any other restriction, but they must have one).
   - They can also be restricted by relational operations (=, >, <).
diff --git a/test/boost/restrictions_test.cc b/test/boost/restrictions_test.cc
index 1d39fd578c..e6cc552c2b 100644
--- a/test/boost/restrictions_test.cc
+++ b/test/boost/restrictions_test.cc
@@ -952,3 +952,99 @@ SEASTAR_THREAD_TEST_CASE(strict_allow_filtering_live_update) {
                 exception_predicate::message_contains("use ALLOW FILTERING"));
     }, cfg).get();
 }
+
+// Ok - only view primary key columns are restricted by IS NOT NULL
+static const char* good_is_not_null_in_views_query =
+    "CREATE MATERIALIZED VIEW t_view AS SELECT p, c, a, b FROM t "
+    "WHERE p IS NOT NULL and c IS NOT NULL and a IS NOT NULL PRIMARY KEY (a, c, p)";
+
+// Bad - b IS NOT NULL is an invalid restriction, as b isn't a part of the view's primary key
+sstring bad_is_not_null_in_views_query(int bad_view_index = 0) {
+    return fmt::format(
+        "CREATE MATERIALIZED VIEW t_badview_{} AS SELECT p, c, a, b FROM t "
+        "WHERE p IS NOT NULL and c IS NOT NULL and a IS NOT NULL AND b IS NOT NULL PRIMARY KEY (a, c, p)",
+        bad_view_index);
+}
+
+static bool contains_is_not_null_warning(const cql_transport::messages::result_message& result_msg) {
+    const std::vector<sstring>& warnings = result_msg.warnings();
+
+    auto found_warning = std::find_if(warnings.begin(), warnings.end(), [&](const sstring& warning) -> bool {
+        return warning.find("IS NOT NULL") != std::string::npos;
+    });
+
+    return found_warning != warnings.end();
+}
+
+// Test that setting strict_is_not_null_in_views=true doesn't allow invalid IS NOT NULL in views.
+SEASTAR_THREAD_TEST_CASE(strict_is_not_null_in_views_true) {
+    auto cfg = make_shared<db::config>();
+    cfg->strict_is_not_null_in_views(db::tri_mode_restriction_t::mode::TRUE);
+    do_with_cql_env_thread([&](cql_test_env& e) {
+            cquery_nofail(e, "CREATE TABLE t (p int, c int, a int, b int, PRIMARY KEY (p, c))");
+            cquery_nofail(e, good_is_not_null_in_views_query);
+
+            BOOST_REQUIRE_EXCEPTION(e.execute_cql(bad_is_not_null_in_views_query()).get(),
+                                    exceptions::invalid_request_exception,
+                                    exception_predicate::message_contains("IS NOT NULL"));
+    }, cfg).get();
+}
+
+// Test that setting strict_is_not_null_in_views=warn allows invalid IS NOT NULL in views, but throws a warning.
+SEASTAR_THREAD_TEST_CASE(strict_is_not_null_in_views_warn) {
+    auto cfg = make_shared<db::config>();
+    cfg->strict_is_not_null_in_views(db::tri_mode_restriction_t::mode::WARN);
+    do_with_cql_env_thread([&](cql_test_env& e) {
+            cquery_nofail(e, "CREATE TABLE t (p int, c int, a int, b int, PRIMARY KEY (p, c))");
+            cquery_nofail(e, good_is_not_null_in_views_query);
+
+            shared_ptr<cql_transport::messages::result_message> bad_res =
+                e.execute_cql(bad_is_not_null_in_views_query()).get();
+            BOOST_REQUIRE(contains_is_not_null_warning(*bad_res));
+    }, cfg).get();
+}
+
+// Test that setting strict_is_not_null_in_views=false allows invalid IS NOT NULL in views without any warnings.
+SEASTAR_THREAD_TEST_CASE(strict_is_not_null_in_views_false) {
+    auto cfg = make_shared<db::config>();
+    cfg->strict_is_not_null_in_views(db::tri_mode_restriction_t::mode::FALSE);
+    do_with_cql_env_thread([&](cql_test_env& e) {
+            cquery_nofail(e, "CREATE TABLE t (p int, c int, a int, b int, PRIMARY KEY (p, c))");
+            cquery_nofail(e, good_is_not_null_in_views_query);
+
+            shared_ptr<cql_transport::messages::result_message> bad_res =
+                e.execute_cql(bad_is_not_null_in_views_query()).get();
+            BOOST_REQUIRE(!contains_is_not_null_warning(*bad_res));
+    }, cfg).get();
+}
+
+// Test that the strict_is_not_null_in_views flag handles live updates properly.
+SEASTAR_THREAD_TEST_CASE(strict_is_not_null_in_views_live_update) {
+    auto cfg = make_shared<db::config>();
+    cfg->strict_is_not_null_in_views(db::tri_mode_restriction_t::mode::FALSE);
+    do_with_cql_env_thread([&](cql_test_env& e) {
+            cquery_nofail(e, "CREATE TABLE t (p int, c int, a int, b int, PRIMARY KEY (p, c))");
+            cquery_nofail(e, good_is_not_null_in_views_query);
+
+            shared_ptr<cql_transport::messages::result_message> bad_res_with_false =
+                e.execute_cql(bad_is_not_null_in_views_query(0)).get();
+            BOOST_REQUIRE(!contains_is_not_null_warning(*bad_res_with_false));
+
+            cfg->strict_is_not_null_in_views(db::tri_mode_restriction_t::mode::WARN);
+
+            shared_ptr<cql_transport::messages::result_message> bad_res_with_warn =
+                e.execute_cql(bad_is_not_null_in_views_query(1)).get();
+            BOOST_REQUIRE(contains_is_not_null_warning(*bad_res_with_warn));
+
+            cfg->strict_is_not_null_in_views(db::tri_mode_restriction_t::mode::TRUE);
+            BOOST_REQUIRE_EXCEPTION(e.execute_cql(bad_is_not_null_in_views_query(2)).get(),
+                                    exceptions::invalid_request_exception,
+                                    exception_predicate::message_contains("IS NOT NULL"));
+    }, cfg).get();
+}
+
+// Test that the default value for the strict_is_not_null_in_views flag is `warn`.
+SEASTAR_THREAD_TEST_CASE(strict_is_not_null_in_views_default_value) {
+    auto cfg = make_shared<db::config>();
+    BOOST_REQUIRE(cfg->strict_is_not_null_in_views() == db::tri_mode_restriction_t::mode::WARN);
+}
diff --git a/test/boost/view_schema_test.cc b/test/boost/view_schema_test.cc
index dd12259daf..8120cf0ffb 100644
--- a/test/boost/view_schema_test.cc
+++ b/test/boost/view_schema_test.cc
@@ -39,10 +39,10 @@ SEASTAR_TEST_CASE(test_case_sensitivity) {
     return do_with_cql_env_thread([] (auto& e) {
         e.execute_cql("create table cf (\"theKey\" int, \"theClustering\" int, \"theValue\" int, primary key (\"theKey\", \"theClustering\"));").get();
         e.execute_cql("create materialized view mv_test as select * from cf "
-                       "where \"theKey\" is not null and \"theClustering\" is not null and \"theValue\" is not null "
+                       "where \"theKey\" is not null and \"theClustering\" is not null "
                        "primary key (\"theKey\",\"theClustering\")").get();
         e.execute_cql("create materialized view mv_test2 as select \"theKey\", \"theClustering\", \"theValue\" from cf "
-                       "where \"theKey\" is not null and \"theClustering\" is not null and \"theValue\" is not null "
+                       "where \"theKey\" is not null and \"theClustering\" is not null "
                        "primary key (\"theKey\",\"theClustering\")").get();
         e.execute_cql("insert into cf (\"theKey\", \"theClustering\", \"theValue\") values (0 ,0, 0);").get();
 
@@ -2474,7 +2474,7 @@ SEASTAR_TEST_CASE(test_alter_table_with_updates) {
     return do_with_cql_env_thread([] (auto& e) {
         e.execute_cql("create table cf (p int, c int, v1 int, v2 int, primary key (p, c));").get();
         e.execute_cql("create materialized view vcf as select p, c, v1, v2 from cf "
-                      "where p is not null and c is not null and v1 is not null and v2 is not null "
+                      "where p is not null and c is not null and v1 is not null "
                       "primary key (v1, p, c)").get();
         e.execute_cql("update cf set v1 = 4, v2 = 5 where p = 1 and c = 1").get();
         e.execute_cql("alter table cf add f int;").get();
diff --git a/test/cql-pytest/cassandra_tests/validation/entities/user_types_test.py b/test/cql-pytest/cassandra_tests/validation/entities/user_types_test.py
index 66eeba7b9b..7c2196ccc6 100644
--- a/test/cql-pytest/cassandra_tests/validation/entities/user_types_test.py
+++ b/test/cql-pytest/cassandra_tests/validation/entities/user_types_test.py
@@ -570,7 +570,7 @@ def testReadAfterAlteringUserTypeNestedWithinList(cql, test_keyspace):
 def testAlteringUserTypeNestedWithinSetWithView(cql, test_keyspace):
     with create_type(cql, test_keyspace, "(a int)") as columnType:
         with create_table(cql, test_keyspace, f"(pk int, c int, v int, s set<frozen<{columnType}>>, PRIMARY KEY (pk, c))") as table:
-            with create_materialized_view(cql, test_keyspace, f"AS SELECT c, pk, v FROM {table} WHERE pk IS NOT NULL AND c IS NOT NULL AND v IS NOT NULL PRIMARY KEY (c, pk)") as mv:
+            with create_materialized_view(cql, test_keyspace, f"AS SELECT c, pk, v FROM {table} WHERE pk IS NOT NULL AND c IS NOT NULL PRIMARY KEY (c, pk)") as mv:
                 execute(cql, table, "INSERT INTO %s (pk, c, v, s) VALUES(?, ?, ?, ?)", 1, 1, 1, {user_type("a", 1), user_type("a", 2)})
                 flush(cql, table)
                 execute(cql, table, "ALTER TYPE " + columnType + " ADD b int")
diff --git a/test/cql-pytest/run.py b/test/cql-pytest/run.py
index dde725382b..e69f1b09aa 100755
--- a/test/cql-pytest/run.py
+++ b/test/cql-pytest/run.py
@@ -271,6 +271,7 @@ def run_scylla_cmd(pid, dir):
         '--authenticator', 'PasswordAuthenticator',
         '--authorizer', 'CassandraAuthorizer',
         '--strict-allow-filtering', 'true',
+        '--strict-is-not-null-in-views', 'true',
         '--permissions-update-interval-in-ms', '100',
         '--permissions-validity-in-ms', '100',
         ], env)
diff --git a/test/cql-pytest/test_materialized_view.py b/test/cql-pytest/test_materialized_view.py
index 8aea771b01..14d224c173 100644
--- a/test/cql-pytest/test_materialized_view.py
+++ b/test/cql-pytest/test_materialized_view.py
@@ -223,7 +223,7 @@ def test_static_columns_are_disallowed(cql, test_keyspace):
         mv = unique_name()
         try:
             with pytest.raises(InvalidRequest, match="[Ss]tatic column"):
-                cql.execute(f"CREATE MATERIALIZED VIEW {test_keyspace}.{mv} AS SELECT p, s FROM {table} WHERE s IS NOT NULL PRIMARY KEY (p)")
+                cql.execute(f"CREATE MATERIALIZED VIEW {test_keyspace}.{mv} AS SELECT p, s FROM {table} PRIMARY KEY (p)")
         finally:
             cql.execute(f"DROP MATERIALIZED VIEW IF EXISTS {test_keyspace}.{mv}")
 
@@ -258,7 +258,6 @@ def test_is_not_operator_must_be_null(cql, test_keyspace):
 # NOTE: if issue #8517 (IS NOT NULL in filters) is implemented, we will need to
 # replace this test by a test that checks that the filter works as expected,
 # both in ordinary base-table SELECT and in materialized-view definition.
-@pytest.mark.xfail(reason="issue #10365")
 def test_is_not_null_forbidden_in_filter(cql, test_keyspace, cassandra_bug):
     with new_test_table(cql, test_keyspace, 'p int primary key, xyz int') as table:
         # Check that "IS NOT NULL" is not supported in a regular (base table)
diff --git a/test/pylib/scylla_cluster.py b/test/pylib/scylla_cluster.py
index 91e8cd5185..c867919992 100644
--- a/test/pylib/scylla_cluster.py
+++ b/test/pylib/scylla_cluster.py
@@ -96,6 +96,7 @@ def make_scylla_conf(workdir: pathlib.Path, host_addr: str, seed_addrs: List[str
         'request_timeout_in_ms': 300000,
 
         'strict_allow_filtering': True,
+        'strict_is_not_null_in_views': True,
 
         'permissions_update_interval_in_ms': 100,
         'permissions_validity_in_ms': 100,