diff --git a/replica/database.hh b/replica/database.hh
index ac493027e9..2ac31fe336 100644
--- a/replica/database.hh
+++ b/replica/database.hh
@@ -92,6 +92,7 @@ class feature_service;
 
 namespace sstables {
 
+enum class sstable_state;
 class sstable;
 class compaction_descriptor;
 class compaction_completion_desc;
@@ -499,7 +500,7 @@ private:
     db_clock::time_point _truncated_at = db_clock::time_point::min();
 
     bool _is_bootstrap_or_replace = false;
-    sstables::shared_sstable make_sstable(sstring dir);
+    sstables::shared_sstable make_sstable(sstables::sstable_state state);
 
 public:
     void deregister_metrics();
diff --git a/replica/table.cc b/replica/table.cc
index 8a87761f99..ff08b58df7 100644
--- a/replica/table.cc
+++ b/replica/table.cc
@@ -279,13 +279,13 @@ table::make_reader_v2(schema_ptr s,
 }
 
 sstables::shared_sstable table::make_streaming_sstable_for_write() {
-    auto newtab = make_sstable(_config.datadir);
+    auto newtab = make_sstable(sstables::sstable_state::normal);
     tlogger.debug("Created sstable for streaming: ks={}, cf={}", schema()->ks_name(), schema()->cf_name());
     return newtab;
 }
 
 sstables::shared_sstable table::make_streaming_staging_sstable() {
-    auto newtab = make_sstable(_config.datadir + "/" + sstables::staging_dir);
+    auto newtab = make_sstable(sstables::sstable_state::staging);
     tlogger.debug("Created staging sstable for streaming: ks={}, cf={}", schema()->ks_name(), schema()->cf_name());
     return newtab;
 }
@@ -434,13 +434,13 @@ static bool belongs_to_other_shard(const std::vector<shard_id>& shards) {
     return shards.size() != size_t(belongs_to_current_shard(shards));
 }
 
-sstables::shared_sstable table::make_sstable(sstring dir) {
+sstables::shared_sstable table::make_sstable(sstables::sstable_state state) {
     auto& sstm = get_sstables_manager();
-    return sstm.make_sstable(_schema, *_storage_opts, dir, calculate_generation_for_new_table(), sstm.get_highest_supported_format(), sstables::sstable::format_types::big);
+    return sstm.make_sstable(_schema, _config.datadir, *_storage_opts, calculate_generation_for_new_table(), state, sstm.get_highest_supported_format(), sstables::sstable::format_types::big);
 }
 
 sstables::shared_sstable table::make_sstable() {
-    return make_sstable(_config.datadir);
+    return make_sstable(sstables::sstable_state::normal);
 }
 
 void table::notify_bootstrap_or_replace_start() {