diff --git a/db/large_data_handler.cc b/db/large_data_handler.cc
index 1e13c3f299..1dacc91610 100644
--- a/db/large_data_handler.cc
+++ b/db/large_data_handler.cc
@@ -18,19 +18,20 @@ namespace db {
 
 nop_large_data_handler::nop_large_data_handler()
     : large_data_handler(std::numeric_limits<uint64_t>::max(), std::numeric_limits<uint64_t>::max(),
-          std::numeric_limits<uint64_t>::max(), std::numeric_limits<uint64_t>::max()) {
+          std::numeric_limits<uint64_t>::max(), std::numeric_limits<uint64_t>::max(), std::numeric_limits<uint64_t>::max()) {
     // Don't require start() to be called on nop large_data_handler.
     start();
 }
 
-large_data_handler::large_data_handler(uint64_t partition_threshold_bytes, uint64_t row_threshold_bytes, uint64_t cell_threshold_bytes, uint64_t rows_count_threshold)
+large_data_handler::large_data_handler(uint64_t partition_threshold_bytes, uint64_t row_threshold_bytes, uint64_t cell_threshold_bytes, uint64_t rows_count_threshold, uint64_t collection_elements_count_threshold)
         : _partition_threshold_bytes(partition_threshold_bytes)
         , _row_threshold_bytes(row_threshold_bytes)
         , _cell_threshold_bytes(cell_threshold_bytes)
         , _rows_count_threshold(rows_count_threshold)
+        , _collection_elements_count_threshold(collection_elements_count_threshold)
 {
-    large_data_logger.debug("partition_threshold_bytes={} row_threshold_bytes={} cell_threshold_bytes={} rows_count_threshold={}",
-        partition_threshold_bytes, row_threshold_bytes, cell_threshold_bytes, rows_count_threshold);
+    large_data_logger.debug("partition_threshold_bytes={} row_threshold_bytes={} cell_threshold_bytes={} rows_count_threshold={} collection_elements_count_threshold={}",
+        partition_threshold_bytes, row_threshold_bytes, cell_threshold_bytes, rows_count_threshold, _collection_elements_count_threshold);
 }
 
 future<large_data_handler::partition_above_threshold> large_data_handler::maybe_record_large_partitions(const sstables::sstable& sst, const sstables::key& key, uint64_t partition_size, uint64_t rows) {
diff --git a/db/large_data_handler.hh b/db/large_data_handler.hh
index 9db24b1616..6c6cee08f7 100644
--- a/db/large_data_handler.hh
+++ b/db/large_data_handler.hh
@@ -54,10 +54,11 @@ private:
     uint64_t _row_threshold_bytes;
     uint64_t _cell_threshold_bytes;
     uint64_t _rows_count_threshold;
+    uint64_t _collection_elements_count_threshold;
     mutable large_data_handler::stats _stats;
 
 public:
-    explicit large_data_handler(uint64_t partition_threshold_bytes, uint64_t row_threshold_bytes, uint64_t cell_threshold_bytes, uint64_t rows_count_threshold);
+    explicit large_data_handler(uint64_t partition_threshold_bytes, uint64_t row_threshold_bytes, uint64_t cell_threshold_bytes, uint64_t rows_count_threshold, uint64_t collection_elements_count_threshold);
     virtual ~large_data_handler() {}
 
     // Once large_data_handler is stopped no further updates will be accepted.
@@ -113,6 +114,9 @@ public:
     uint64_t get_rows_count_threshold() const noexcept {
         return _rows_count_threshold;
     }
+    uint64_t get_collection_elements_count_threshold() const noexcept {
+        return _collection_elements_count_threshold;
+    }
 
     static sstring sst_filename(const sstables::sstable& sst);
 
@@ -126,8 +130,8 @@ protected:
 
 class cql_table_large_data_handler : public large_data_handler {
 public:
-    explicit cql_table_large_data_handler(uint64_t partition_threshold_bytes, uint64_t row_threshold_bytes, uint64_t cell_threshold_bytes, uint64_t rows_count_threshold)
-        : large_data_handler(partition_threshold_bytes, row_threshold_bytes, cell_threshold_bytes, rows_count_threshold) {}
+    explicit cql_table_large_data_handler(uint64_t partition_threshold_bytes, uint64_t row_threshold_bytes, uint64_t cell_threshold_bytes, uint64_t rows_count_threshold, uint64_t collection_elements_count_threshold)
+        : large_data_handler(partition_threshold_bytes, row_threshold_bytes, cell_threshold_bytes, rows_count_threshold, collection_elements_count_threshold) {}
 
 protected:
     virtual future<> record_large_partitions(const sstables::sstable& sst, const sstables::key& partition_key, uint64_t partition_size, uint64_t rows) const override;
diff --git a/replica/database.cc b/replica/database.cc
index 0f2d4658cf..be8854b844 100644
--- a/replica/database.cc
+++ b/replica/database.cc
@@ -351,7 +351,8 @@ database::database(const db::config& cfg, database_config dbcfg, service::migrat
     , _large_data_handler(std::make_unique<db::cql_table_large_data_handler>(_cfg.compaction_large_partition_warning_threshold_mb()*1024*1024,
               _cfg.compaction_large_row_warning_threshold_mb()*1024*1024,
               _cfg.compaction_large_cell_warning_threshold_mb()*1024*1024,
-              _cfg.compaction_rows_count_warning_threshold()))
+              _cfg.compaction_rows_count_warning_threshold(),
+              _cfg.compaction_collection_elements_count_warning_threshold()))
     , _nop_large_data_handler(std::make_unique<db::nop_large_data_handler>())
     , _user_sstables_manager(std::make_unique<sstables::sstables_manager>(*_large_data_handler, _cfg, feat, _row_cache_tracker, dbcfg.available_memory))
     , _system_sstables_manager(std::make_unique<sstables::sstables_manager>(*_nop_large_data_handler, _cfg, feat, _row_cache_tracker, dbcfg.available_memory))
diff --git a/test/boost/sstable_3_x_test.cc b/test/boost/sstable_3_x_test.cc
index ce222551fd..c3953330ca 100644
--- a/test/boost/sstable_3_x_test.cc
+++ b/test/boost/sstable_3_x_test.cc
@@ -5142,7 +5142,7 @@ struct large_row_handler : public db::large_data_handler {
 
     large_row_handler(uint64_t large_rows_threshold, uint64_t rows_count_threshold, uint64_t cell_threshold_bytes, callback_t callback)
         : large_data_handler(std::numeric_limits<uint64_t>::max(), large_rows_threshold, cell_threshold_bytes,
-            rows_count_threshold)
+            rows_count_threshold, std::numeric_limits<uint64_t>::max())
         , callback(std::move(callback)) {
         start();
     }