table: Add a flush RP mark to table, and shortcut if not above

Adds a second RP to table, marking where we flushed last.
If a new flush request comes in that is below this mark, we
can skip a second flush.

This is to (in future) support incremental CL flush.

database.cc

@@ -872,7 +872,7 @@ database::init_commitlog() {
                 return;
             }
             // Initiate a background flush. Waited upon in `stop()`.
-            (void)_column_families[id]->flush();
+            (void)_column_families[id]->flush(pos);
         }).release(); // we have longer life time than CL. Ignore reg anchor
     });
 }

database.hh

@@ -449,6 +449,7 @@ private:
     std::optional<int64_t> _sstable_generation = {};
 
     db::replay_position _highest_rp;
+    db::replay_position _flush_rp;
     db::replay_position _lowest_allowed_rp;
 
     // Provided by the database that owns this commitlog
@@ -751,7 +752,7 @@ public:
 
     void start();
     future<> stop();
-    future<> flush();
+    future<> flush(std::optional<db::replay_position> = {});
     future<> clear(); // discards memtable(s) without flushing them to disk.
     future<db::replay_position> discard_sstables(db_clock::time_point);
 

table.cc

@@ -1262,9 +1262,14 @@ future<std::unordered_map<sstring, table::snapshot_details>> table::get_snapshot
     });
 }
 
-future<> table::flush() {
+future<> table::flush(std::optional<db::replay_position> pos) {
+    if (pos && *pos < _flush_rp) {
+        return make_ready_future<>();
+    }
     auto op = _pending_flushes_phaser.start();
-    return _memtables->request_flush().then([op = std::move(op)] {});
+    return _memtables->request_flush().then([this, op = std::move(op), fp = _highest_rp] {
+        _flush_rp = std::max(_flush_rp, fp);
+    });
 }
 
 bool table::can_flush() const {