replica/mutation_dump: detect end-of-page in range-scans

The current read-loop fails to detect end-of-page and if the query
result buider cuts the page, it will just proceed to the next
partition. This will result in distorted query results, as the result
builder will request for the consumption to stop after each clustering
row.
To fix, check if the page was cut before moving on to the next
partition.
A unit test reproducing the bug was also added.

replica/mutation_dump.cc

@@ -604,7 +604,11 @@ future<foreign_ptr<lw_shared_ptr<query::result>>> dump_mutations(
             std::rethrow_exception(std::move(ex));
         }
 
-        dk_opt = co_await partition_key_generator();
+        if (compaction_state->are_limits_reached() || qs.builder.is_short_read()) {
+            dk_opt = {};
+        } else {
+            dk_opt = co_await partition_key_generator();
+        }
     }
 
     co_return make_lw_shared<query::result>(qs.builder.build(compaction_state->current_full_position()));

test/cql-pytest/test_select_from_mutation_fragments.py

@@ -422,3 +422,30 @@ def test_ck_in_query(cql, test_table):
             for col_name, expected_value in zip(columns, expected_row):
                 assert hasattr(row, col_name)
                 assert getattr(row, col_name) == expected_value
+
+
+def test_many_partitions(cql, test_keyspace, scylla_only):
+    num_partitions = 5000
+    with util.new_test_table(cql, test_keyspace, 'pk int PRIMARY KEY, v int') as table:
+        delete_id = cql.prepare(f"DELETE FROM {table} WHERE pk = ?")
+        for pk in range(num_partitions):
+            cql.execute(delete_id, (pk,))
+
+        res = list(cql.execute(f"SELECT * FROM MUTATION_FRAGMENTS({table})"))
+        pks = set()
+        partition_starts = 0
+        partition_ends = 0
+        for row in res:
+            assert row.pk >= 0 and row.pk < num_partitions
+            if row.mutation_fragment_kind == "partition start":
+                partition_starts += 1
+                pks.add(row.pk)
+            elif row.mutation_fragment_kind == "partition end":
+                partition_ends += 1
+                assert row.pk in pks
+            else:
+                pytest.fail(f"Unexpected mutation fragment kind: {row.mutation_fragment_kind}")
+
+        assert partition_starts == num_partitions
+        assert partition_ends == num_partitions
+        assert len(pks) == num_partitions