feat: query-time MVCC DELETE masks for the duckdb-native pinned cache (#819 PR3)

CMakeLists.txt

@@ -268,10 +268,12 @@ set(EXTENSION_SOURCES
     src/op/partition/gpu_partition_impl.cpp
     src/op/result/host_table_chunk_reader.cpp
     src/op/scan/cached_ranges.cpp
+    src/op/scan/duckdb_mvcc_visibility.cpp
     src/op/scan/duckdb_native_metadata.cpp
     src/op/scan/duckdb_native_decoder.cpp
     src/op/scan/dynamic_filter_merge.cpp
     src/op/scan/sirius_physical_dynamic_filter.cpp
+    src/op/scan/host_keep_mask.cpp
     src/op/scan/owning_table_view.cpp
     src/op/scan/parquet_schema_mapping.cpp
     src/op/scan/scan_plan.cpp
@@ -343,6 +345,7 @@ set(EXTENSION_SOURCES
     src/planner/sirius_plan_top_n.cpp
     src/pin_table.cpp
     src/scan_manager/load_balancing_scan_batch_coalescer.cpp
+    src/scan_manager/mvcc_mask_job.cpp
     src/scan_manager/pinned_chunk_stats.cpp
     src/scan_manager/round_robin_strategy.cpp
     src/scan_manager/sirius_scan_manager.cpp
@@ -602,6 +605,7 @@ set(TEST_SOURCES
     test/cpp/integration/test_pin_table_host_streaming.cpp
     test/cpp/integration/test_pin_table_merge_columns.cpp
     test/cpp/integration/test_pin_table_column_order.cpp
+    test/cpp/integration/test_pin_table_mvcc_delete.cpp
     test/cpp/integration/test_pin_table_mvcc_foundation.cpp
     test/cpp/integration/test_pin_table_zone_map_pruning.cpp
     test/cpp/integration/test_table_gpu_cache_warm_mgpu.cpp
@@ -618,8 +622,8 @@ set(TEST_SOURCES
     test/cpp/log/test_logging.cpp
     test/cpp/scan_manager/test_s3_routing_cutover.cpp
     test/cpp/scan_manager/test_pin_table_multi_gpu.cpp
-    test/cpp/scan_manager/test_pin_table_duckdb_native_cache.cpp
     test/cpp/scan_manager/test_cached_serving_hardening.cpp
+    test/cpp/scan_manager/test_mvcc_mask_job.cpp
     test/cpp/scan_manager/test_pinned_chunk_stats.cpp
     test/cpp/operator/test_physical_grouped_aggregate_merge_mgpu.cpp
     test/cpp/operator/test_physical_hash_join_mgpu.cpp
@@ -676,6 +680,7 @@ set(TEST_SOURCES
     test/cpp/scan/bench_decode_codecs.cpp
     test/cpp/scan/test_can_serve_with_columns.cpp
     test/cpp/scan/test_duckdb_block_layout.cpp
+    test/cpp/scan/test_duckdb_mvcc_visibility.cpp
     test/cpp/scan/test_duckdb_native_batch_coalescer.cpp
     test/cpp/scan/test_duckdb_native_dynamic_filter_remap.cpp
     test/cpp/scan/test_duckdb_native_walker.cpp
@@ -684,6 +689,7 @@ set(TEST_SOURCES
     test/cpp/scan/test_gpu_decode_bitpacking.cpp
     test/cpp/scan/test_gpu_decode_rle.cpp
     test/cpp/scan/test_gpu_decode_strings.cpp
+    test/cpp/scan/test_host_keep_mask.cpp
     test/cpp/scan/test_gpu_native_decode.cpp
     test/cpp/scan/test_owning_table_view.cpp
     test/cpp/scan/test_parquet_schema_mapping.cpp

src/include/op/scan/duckdb_mvcc_visibility.hpp

@@ -0,0 +1,158 @@
+/*
+ * Copyright 2026, Sirius Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <duckdb/common/typedefs.hpp>
+#include <duckdb/transaction/transaction_data.hpp>
+#include <scan_manager/duckdb_mvcc_metadata.hpp>
+
+#include <cstddef>
+#include <cstdint>
+#include <span>
+#include <vector>
+
+namespace duckdb {
+class ClientContext;
+class DataTable;
+class RowGroup;
+}  // namespace duckdb
+
+namespace sirius::op::scan {
+
+/**
+ * @brief One row group's contribution to one cached chunk's keep-mask.
+ *
+ * The pointed-to RowGroup is owned by the table's segment tree and only
+ * removed by checkpoint compaction (suppressed while a duckdb table is
+ * pinned), so the raw pointer is stable for the query's prepare/serve window.
+ */
+struct mvcc_row_group_slice {
+  duckdb::RowGroup* row_group{nullptr};
+  duckdb::idx_t row_group_start{0};  ///< absolute rowid of the RG's first row
+  std::size_t chunk_offset{0};       ///< where its rows land in the chunk's mask
+  std::size_t row_count{0};          ///< rows covered by the pin (last RG clamped)
+  bool has_version_state{false};     ///< capture's non-loading dirty probe (skip signal)
+};
+
+/**
+ * @brief Per-query visibility work plan over a pinned entry's cached prefix.
+ *
+ * Slot i of @ref mvcc_row_groups holds the slices for cached chunk i,
+ * paralleling base_row_count_per_chunk[i]. Holds copies
+ * and tree-owned pointers only — no pinned_entry references escape into it.
+ */
+struct mvcc_visibility_plan {
+  explicit mvcc_visibility_plan(duckdb::TransactionData transaction) : transaction(transaction) {}
+
+  duckdb::TransactionData transaction;  ///< query txn on the pinned table's own database
+  std::vector<std::vector<mvcc_row_group_slice>> mvcc_row_groups;
+  std::vector<bool> chunk_has_version_state;  ///< OR of each chunk's slice probes
+
+  /// True when any chunk has version state — the mask job's do-anything-at-all
+  /// signal. False = every mask slot stays null; no allocation, no tasks.
+  [[nodiscard]] bool any_version_state() const;
+};
+
+/**
+ * @brief Build the visibility plan for one pinned duckdb table. SERIAL —
+ *        prepare/query thread only (the ClientContext discipline of
+ *        prepare_duckdb_native_walk).
+ *
+ * Captures the query transaction with BOTH ids off the table's own database
+ * (so the query's own uncommitted deletes are honored) and walks the row
+ * groups covering the pinned prefix [0, metadata.n_cache()), clamping the
+ * last row group to pin coverage (its live count grows under appends). The
+ * same pass records each row group's non-loading dirty probe via
+ * RowGroup::GetPartitionStats (count_type APPROXIMATE <=> version state):
+ * safe because a delete visible to any active snapshot always has version
+ * state present — delete markers persist until checkpoint compaction
+ * (suppressed while pinned), and checkpointed tombstones show as unloaded
+ * deletes. Conservative in two ways that cost a walk but never a wrong mask:
+ * append-only version info probes dirty, and a RowVersionManager created by
+ * this session's own writes stays attached forever — only freshly-loaded
+ * (ATTACH-then-pin) tables take the zero-cost clean path.
+ *
+ * Throws std::runtime_error on invariants validated at pin time
+ * (validate_duckdb_pin_chunk): start_time < v_base (a re-pin raced
+ * plan->prepare), non-contiguous or short row-group coverage, a chunk
+ * boundary off a row-group start, or a slice offset not 32-row aligned (the
+ * lock-free bit fill requires task ranges to never share a mask word).
+ */
+mvcc_visibility_plan capture_mvcc_visibility_plan(
+  duckdb::DataTable& storage,
+  duckdb::ClientContext& context,
+  scan_manager::duckdb_mvcc_metadata const& metadata);
+
+/**
+ * @brief Fill the keep-mask bits for @p slices (all belonging to ONE chunk).
+ *
+ * Worker-safe with plain stores: concurrent calls over disjoint slice lists
+ * write disjoint words (slice offsets are 32-row aligned, asserted at
+ * capture). Writes exactly bits [chunk_offset, chunk_offset + row_count) per
+ * slice, 1 = keep; a slice without version state bulk-sets ones with no MVCC
+ * calls or locks, otherwise each 2048-row vector goes through
+ * RowGroup::GetSelVector. Whole-word writes never collide with a later
+ * writer (a partial vector ends its slice; a partial slice ends the table),
+ * and padding bits past the covered range are don't-care.
+ *
+ * @return true when any row in the covered range was dropped.
+ */
+bool fill_keep_mask_for_row_groups(std::span<mvcc_row_group_slice const> slices,
+                                   duckdb::TransactionData transaction,
+                                   std::span<std::uint32_t> chunk_mask_words);
+
+/**
+ * @brief True when any row group intersecting rowids [0, row_prefix) carries
+ *        an in-memory update chain (UpdateSegment) on any of
+ *        @p storage_column_indices.
+ *
+ * Update chains version VALUES in place, invisibly to GetSelVector, so the
+ * cached base (and the on-disk image) would serve stale data. SERIAL — may
+ * lazily load column data.
+ */
+bool any_update_chains(duckdb::DataTable& storage,
+                       std::span<duckdb::storage_t const> storage_column_indices,
+                       std::size_t row_prefix);
+
+/// Whether the MVCC-blind disk-native read of a table is exact for a given
+/// snapshot, and if not, why.
+enum class native_read_mvcc_state : std::uint8_t {
+  exact,               ///< raw bytes match this snapshot's visibility
+  has_update_chains,   ///< a scanned column has in-memory UpdateSegment values
+  has_invisible_rows,  ///< tombstones or in-flight inserts at this snapshot
+};
+
+/**
+ * @brief Single row-group walk checking update chains and row visibility
+ *        together, early-out on the first finding.
+ *
+ * Per row group, cheapest first: update-chain checks on the scanned columns,
+ * the non-loading version-state probe, and only probe-dirty row groups pay
+ * RowGroup::GetVisibleRowCount(transaction) == count. The last tier keeps
+ * session-written tables on the GPU: their RowVersionManager stays attached
+ * forever, so the probe alone would report them dirty long after every row
+ * became visible. SERIAL — prepare/query thread only.
+ *
+ * Transaction-local appends live in LocalStorage, outside the segment tree —
+ * callers check those separately.
+ */
+native_read_mvcc_state check_native_read_mvcc_state(
+  duckdb::DataTable& storage,
+  std::span<duckdb::storage_t const> scanned_columns,
+  duckdb::TransactionData transaction);
+
+}  // namespace sirius::op::scan

src/include/op/scan/duckdb_native_gpu_ingestible.hpp

@@ -168,6 +168,11 @@ class duckdb_native_gpu_ingestible : public op::scan::gpu_ingestible {
 
   [[nodiscard]] std::vector<std::size_t> materialized_column_order() const override;
 
+  [[nodiscard]] bool has_row_filter() const noexcept override
+  {
+    return _filter_expression != nullptr;
+  }
+
  private:
   std::unique_ptr<op::scan::duckdb_native_ingestible_table_info> _info;
   duckdb_native_walk_plan _plan;

src/include/op/scan/duckdb_native_metadata.hpp

@@ -198,4 +198,9 @@ struct duckdb_native_row_group_range {
 duckdb_native_row_group_range walk_duckdb_native_row_group_range(
   const duckdb_native_walk_plan& plan, std::size_t rg_begin, std::size_t rg_end);
 
+/// @brief Row groups per parallel-walk task unit (env `SIRIUS_METADATA_PARSE_CHUNK`,
+/// default 8). Shared by the metadata parse fan-out and the MVCC mask job so both
+/// slice row-group work into the same task granularity.
+std::size_t metadata_parse_chunk();
+
 }  // namespace sirius::op::scan

src/include/op/scan/gpu_ingestible.hpp

@@ -137,6 +137,14 @@ class gpu_ingestible : public std::enable_shared_from_this<gpu_ingestible> {
     const cucascade::memory::memory_space& mem_space,
     rmm::cuda_stream_view stream) = 0;
 
+  /**
+   * @brief Whether this ingestible holds a row-filter expression that
+   *        @ref post_filter_and_project applies to splits not already
+   *        row-filtered. Drives the working-set estimate of resident
+   *        (pinned-cache) splits, which always reach post-filter unfiltered.
+   */
+  [[nodiscard]] virtual bool has_row_filter() const noexcept { return false; }
+
   [[nodiscard]] virtual const ingestible_table_info& table_info() const noexcept = 0;
 
   /// Column primary (storage) indices in the exact order @ref materialize_table emits

src/include/op/scan/host_keep_mask.hpp

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2026, Sirius Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cudf/table/table.hpp>
+#include <cudf/table/table_view.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/resource_ref.hpp>
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <span>
+
+namespace sirius::op::scan {
+
+/**
+ * @brief Filter @p view by a host-computed byte keep-mask.
+ *
+ * Uploads @p keep (one byte per row, 1 = keep) as a BOOL8 column and returns
+ * @c cudf::apply_boolean_mask(view), so every host-computed keep-mask shares
+ * one implementation.
+ *
+ * The upload is a @c cudaMemcpyAsync from pageable memory, which stages the
+ * bytes out of @p keep before returning — the caller may free @p keep as soon
+ * as this returns.
+ *
+ * @throws std::invalid_argument when keep.size() != view.num_rows().
+ */
+std::unique_ptr<cudf::table> apply_host_keep_mask(cudf::table_view const& view,
+                                                  std::span<std::uint8_t const> keep,
+                                                  rmm::cuda_stream_view stream,
+                                                  rmm::device_async_resource_ref mr);
+
+/**
+ * @brief Filter @p view by a host-computed bit-packed keep-mask.
+ *
+ * @p keep_words is a cuDF-convention bitmask (bit `row % 32` of uint32 word
+ * `row / 32`, 1 = keep; padding bits past @p row_count ignored). Uploads the
+ * words, expands them with @c cudf::mask_to_bools, and returns
+ * @c cudf::apply_boolean_mask(view). Used by the MVCC keep-mask path.
+ *
+ * ⚠ The upload from pinned host memory is a true-async DMA: @p keep_words
+ * must stay alive until the work enqueued on @p stream completes.
+ *
+ * @throws std::invalid_argument on row_count != view.num_rows() or
+ *         keep_words.size() != ceil(row_count / 32).
+ */
+std::unique_ptr<cudf::table> apply_host_keep_bitmask(cudf::table_view const& view,
+                                                     std::span<std::uint32_t const> keep_words,
+                                                     std::size_t row_count,
+                                                     rmm::cuda_stream_view stream,
+                                                     rmm::device_async_resource_ref mr);
+
+}  // namespace sirius::op::scan

src/include/op/scan/parquet_gpu_ingestible.hpp

@@ -284,6 +284,11 @@ class parquet_gpu_ingestible : public gpu_ingestible {
 
   [[nodiscard]] std::vector<std::size_t> materialized_column_order() const override;
 
+  [[nodiscard]] bool has_row_filter() const noexcept override
+  {
+    return _duckdb_filter_expression != nullptr;
+  }
+
  private:
   /// Read one file's footer, prune its row groups against the filter, and record
   /// per-row-group byte accounting. Returns a single @c parquet_file_scan_info.

src/include/op/scan/sirius_gpu_scan_operator_data.hpp

@@ -19,6 +19,7 @@
 // sirius
 #include <op/scan/gpu_ingestible.hpp>
 #include <op/sirius_physical_operator.hpp>
+#include <scan_manager/mvcc_chunk_mask.hpp>
 // cucascade
 #include <cucascade/cudf/gpu_data_representation.hpp>
 #include <cucascade/data/data_batch.hpp>
@@ -134,6 +135,17 @@ class scan_operator_input : public op::operator_data {
   cucascade::memory::memory_space* gpu_memory_space = nullptr;
   std::variant<std::monostate, std::unique_ptr<scan_info>, std::shared_ptr<::cucascade::data_batch>>
     materialization_info;
+  /// Per-query MVCC keep-mask for a resident cached chunk; a default mask =
+  /// every row visible (no upload, no kernel). Attached by
+  /// drain_cached_provider, applied in gpu_ingestible::materialize_table's
+  /// resident branch. Self-contained: the mask's owning words pointer keeps
+  /// the storage alive for the split's lifetime.
+  scan_manager::mvcc_chunk_mask mvcc_keep_mask;
+  /// True when the op's ingestible will run a row-filter expression against
+  /// this split's materialized table (post_filter_and_project filters by
+  /// copy). Stamped by drain_cached_provider on resident splits; scan_info
+  /// splits fold filter costs into their own estimates instead.
+  bool row_filter_pending{false};
 };
 
 }  // namespace sirius::op::scan