db/view: track range tombstones in update stream during view update building

The view update builder ignored range tombstone changes from the update
stream when there all existing mutation fragments were already consumed.
The old code assumed range tombstones 'remove nothing pre-existing, so
we can ignore it', but this failed to update _update_current_tombstone.
Consequently, when a range delete and an insert within that range appeared
in the same batch, the range tombstone was not applied to the inserted row,
or was applied to a row outside the range that it covered causing it to
incorrectly survive/be deleted in the materialized view.

Fix by handling is_range_tombstone_change() fragments in the update-only
branch, updating _update_current_tombstone so subsequent clustering rows
correctly have the range tombstone applied to them.

Fixes SCYLLADB-1555

Closes scylladb/scylladb#29483

(cherry picked from commit 6011cb8a4c)

Closes scylladb/scylladb#29569

Closes scylladb/scylladb#29643

db/view/view.cc

@@ -1594,9 +1594,10 @@ future<stop_iteration> view_update_builder::on_results() {
         return should_stop_updates() ? stop() : advance_existings();
     }
 
-    // If we have updates and it's a range tombstone, it removes nothing pre-exisiting, so we can ignore it
     if (_update && !_update->is_end_of_partition()) {
-        if (_update->is_clustering_row()) {
+        if (_update->is_range_tombstone_change()) {
+            _update_current_tombstone = _update->as_range_tombstone_change().tombstone();
+        } else if (_update->is_clustering_row()) {
             _update->mutate_as_clustering_row(*_schema, [&] (clustering_row& cr) mutable {
                 cr.apply(std::max(_update_partition_tombstone, _update_current_tombstone));
             });

test/cqlpy/test_materialized_view.py

@@ -1497,6 +1497,42 @@ def test_views_with_future_tombstones(cql, test_keyspace):
             assert [] == list(cql.execute(f'select * from {table}'))
             assert [] == list(cql.execute(f'select * from {mv}'))
 
+# Test that a range delete in the same batch as an insert correctly covers
+# rows within the deleted range in the materialized view and that it doesn't
+# cover rows outside the deleted range. The view update builder must track
+# range tombstone changes from the update stream so that all range tombstones
+# are applied to the clustering rows that they cover.
+# Without this, an inserted row within the range incorrectly survives in the
+# view or is incorrectly deleted.
+# Reproduces SCYLLADB-1555.
+def test_mv_range_delete_and_insert_in_same_batch(cql, test_keyspace):
+    # Case 1: Insert within the range-deleted interval. The range tombstone
+    # should shadow the insert, leaving both base and view empty.
+    with new_test_table(cql, test_keyspace,
+            'p int, c int, v int, w int, primary key (p, c)') as table:
+        with new_materialized_view(cql, table, '*', 'v, p, c',
+                'v is not null and p is not null and c is not null') as mv:
+            cql.execute(f"BEGIN BATCH "
+                        f"DELETE FROM {table} WHERE p = 1 AND c >= 1 AND c <= 3; "
+                        f"INSERT INTO {table} (p, c, v) VALUES (1, 3, 3); "
+                        f"APPLY BATCH")
+            assert [] == list(cql.execute(f"SELECT * FROM {table}"))
+            assert [] == list(cql.execute(f"SELECT * FROM {mv}"))
+    # Case 2: A pre-existing row within the range, and an insert outside it.
+    # The range delete should remove the existing row, but the new row at c=4
+    # falls outside the range and should survive in both base and view.
+    with new_test_table(cql, test_keyspace,
+            'p int, c int, v int, w int, primary key (p, c)') as table:
+        with new_materialized_view(cql, table, '*', 'v, p, c',
+                'v is not null and p is not null and c is not null') as mv:
+            cql.execute(f"INSERT INTO {table} (p, c, v) VALUES (1, 2, 1)")
+            cql.execute(f"BEGIN BATCH "
+                        f"DELETE FROM {table} WHERE p = 1 AND c >= 1 AND c <= 3; "
+                        f"INSERT INTO {table} (p, c, v) VALUES (1, 4, 3); "
+                        f"APPLY BATCH")
+            assert [] != list(cql.execute(f"SELECT * FROM {table}"))
+            assert [] != list(cql.execute(f"SELECT * FROM {mv}"))
+
 # Test view representation in system.* tables
 def test_view_in_system_tables(cql, test_keyspace):
     with new_test_table(cql, test_keyspace, "p int PRIMARY KEY, v int") as base: