Implement Filter Encoding v2 parity

example-worker/table/splits.cpp

@@ -20,6 +20,7 @@
 #include <vector>
 
 #include <arrow/array.h>
+#include <arrow/array/builder_binary.h>
 #include <arrow/array/builder_primitive.h>
 #include <arrow/record_batch.h>
 #include <arrow/type.h>
@@ -63,6 +64,27 @@ std::shared_ptr<arrow::Schema> n_schema() {
     return arrow::schema({arrow::field("n", arrow::int64(), /*nullable=*/false)});
 }
 
+std::shared_ptr<arrow::Schema> dynamic_filter_schema() {
+    return arrow::schema({arrow::field("n", arrow::int64(), /*nullable=*/false),
+                          arrow::field("pushed_filters", arrow::utf8(), /*nullable=*/false)});
+}
+
+std::string render_filter_bounds(const vgi::PushdownFilters& filters) {
+    std::string result;
+    for (const auto& column : filters.filtered_columns()) {
+        const auto bounds = filters.column_bounds(column);
+        if (bounds.min) {
+            if (!result.empty()) result += ',';
+            result += column + ">=" + std::to_string(*bounds.min);
+        }
+        if (bounds.max) {
+            if (!result.empty()) result += ',';
+            result += column + "<=" + std::to_string(*bounds.max);
+        }
+    }
+    return result.empty() ? "(none)" : result;
+}
+
 // Emits the rows of one or more half-open ranges, in order.
 class RangeProducer : public vgi::TableProducer {
 public:
@@ -392,6 +414,141 @@ class SplitEndlessCursor : public vgi::TableFunction {
     }
 };
 
+// `split_dynamic_filter(n, splits)` — a split scan that reports and applies
+// the filter in force for every batch. A reader re-initializes between claimed
+// splits, so reporting the filter as data makes lost state observable even
+// though DuckDB also checks the predicate above the scan.
+class SplitDynamicFilter : public vgi::TableFunction {
+public:
+    std::string name() const override { return "split_dynamic_filter"; }
+
+    vgi::FunctionMetadata metadata() const override {
+        vgi::FunctionMetadata md;
+        md.description = "Echoes the dynamic filter each tick carried, per split";
+        md.categories = {"generator", "diagnostic"};
+        md.projection_pushdown = true;
+        md.filter_pushdown = true;
+        md.auto_apply_filters = true;
+        return md;
+    }
+
+    std::vector<vgi::ArgSpec> argument_specs() const override {
+        return {vgi::ArgSpec::named("n", "int64", "How many rows to generate"),
+                vgi::ArgSpec::named("splits", "int64", "How many splits")};
+    }
+
+    std::shared_ptr<arrow::Schema> bind(const vgi::BindParams&) const override {
+        return dynamic_filter_schema();
+    }
+
+    vgi::TableCardinality cardinality(const vgi::ProcessParams& params) const override {
+        const auto rows = std::max<int64_t>(0, params.arguments.named_int64("n").value_or(0));
+        return {rows, rows};
+    }
+
+    bool supports_splits() const override { return true; }
+
+    vgi::PlanResult plan(const vgi::BindParams& params, const vgi::PlanParams&) const override {
+        const int64_t rows = std::max<int64_t>(0, params.arguments.named_int64("n").value_or(0));
+        const int64_t want =
+            std::max<int64_t>(1, params.arguments.named_int64("splits").value_or(1));
+
+        vgi::PlanResult result;
+        result.estimated_total_rows = rows;
+        result.estimated_total_splits = want;
+        for (int64_t i = 0; i < want; ++i) {
+            vgi::ScanSplit split;
+            const int64_t begin = rows * i / want;
+            const int64_t end = rows * (i + 1) / want;
+            split.payload = encode_range(begin, end);
+            split.estimated_rows = end - begin;
+            split.rows_exact = true;
+            result.splits.push_back(std::move(split));
+        }
+        return result;
+    }
+
+    std::unique_ptr<vgi::TableProducer> init(const vgi::ProcessParams& params) const override {
+        if (!params.split_payloads) {
+            throw std::runtime_error(
+                "split_dynamic_filter is split-only but was initialized with no split tokens");
+        }
+        std::vector<std::pair<int64_t, int64_t>> ranges;
+        ranges.reserve(params.split_payloads->size());
+        for (const auto& payload : *params.split_payloads) {
+            auto range = decode_range(payload);
+            if (!range) {
+                throw std::runtime_error("split_dynamic_filter: unrecognized split payload");
+            }
+            ranges.push_back(*range);
+        }
+        return std::make_unique<Producer>(
+            params.output_schema ? params.output_schema : dynamic_filter_schema(),
+            std::move(ranges), render_filter_bounds(params.pushdown_filters));
+    }
+
+private:
+    class Producer : public vgi::TableProducer {
+    public:
+        Producer(std::shared_ptr<arrow::Schema> schema,
+                 std::vector<std::pair<int64_t, int64_t>> ranges, std::string rendered)
+            : schema_(std::move(schema)),
+              ranges_(std::move(ranges)),
+              rendered_(std::move(rendered)) {
+            if (!ranges_.empty()) cursor_ = ranges_.front().first;
+        }
+
+        void on_dynamic_filters(const vgi::PushdownFilters& filters) override {
+            rendered_ = render_filter_bounds(filters);
+        }
+
+        std::shared_ptr<arrow::RecordBatch> next_batch() override {
+            while (at_ < ranges_.size() && cursor_ >= ranges_[at_].second) {
+                ++at_;
+                if (at_ < ranges_.size()) cursor_ = ranges_[at_].first;
+            }
+            if (at_ >= ranges_.size()) return nullptr;
+
+            constexpr int64_t kBatchRows = 4;
+            const int64_t begin = cursor_;
+            const int64_t end = std::min(begin + kBatchRows, ranges_[at_].second);
+            cursor_ = end;
+
+            arrow::Int64Builder ns;
+            arrow::StringBuilder reports;
+            (void)ns.Reserve(end - begin);
+            (void)reports.Reserve(end - begin);
+            for (int64_t value = begin; value < end; ++value) {
+                (void)ns.Append(value);
+                (void)reports.Append(rendered_);
+            }
+            std::vector<std::shared_ptr<arrow::Array>> built(2);
+            (void)ns.Finish(&built[0]);
+            (void)reports.Finish(&built[1]);
+
+            const std::vector<std::string> names{"n", "pushed_filters"};
+            std::vector<std::shared_ptr<arrow::Array>> projected;
+            projected.reserve(static_cast<size_t>(schema_->num_fields()));
+            for (const auto& field : schema_->fields()) {
+                const auto found = std::find(names.begin(), names.end(), field->name());
+                if (found == names.end()) {
+                    throw std::runtime_error("split_dynamic_filter: unexpected column '" +
+                                             field->name() + "'");
+                }
+                projected.push_back(built[static_cast<size_t>(found - names.begin())]);
+            }
+            return arrow::RecordBatch::Make(schema_, end - begin, std::move(projected));
+        }
+
+    private:
+        std::shared_ptr<arrow::Schema> schema_;
+        std::vector<std::pair<int64_t, int64_t>> ranges_;
+        size_t at_ = 0;
+        int64_t cursor_ = 0;
+        std::string rendered_;
+    };
+};
+
 // `split_echo_filters(splits)` — reports what `plan()` was told.
 //
 // One row per split, carrying the split's ordinal and whether planning saw any
@@ -523,6 +680,7 @@ void register_splits(vgi::Worker& worker) {
         "split_many", Shape::Many, "Integers 0..n-1, divided into many more splits than threads"));
     worker.register_table(std::make_shared<SplitFailAt>());
     worker.register_table(std::make_shared<SplitEndlessCursor>());
+    worker.register_table(std::make_shared<SplitDynamicFilter>());
     worker.register_table(std::make_shared<SplitEchoFilters>());
 }
 

include/vgi/pushdown.h

@@ -5,6 +5,7 @@
 #include <map>
 #include <memory>
 #include <optional>
+#include <set>
 #include <string>
 #include <vector>
 
@@ -36,22 +37,23 @@ struct ColumnBounds {
 
 // The predicates the engine pushed into this scan.
 //
-// The wire form is a one-row batch whose first column is a JSON filter tree
-// and whose remaining columns are the constant *values* the tree references by
-// index. Values ride as Arrow columns rather than inside the JSON so they keep
-// their exact type — a decimal or a timestamp survives, where a JSON number
-// would not.
+// Filter Encoding v2 is a one-row Arrow batch. Its non-null `filter_spec` UTF-8
+// field contains the versioned snapshot/delta document; sibling `value_N`,
+// `type_N`, and `artifact_N` fields carry typed payloads. External IN sets are
+// addressed by batch and column index in the request's `join_keys` list.
 class PushdownFilters {
 public:
     // Parse the IPC filter blob. Empty input yields no filters, which is what
     // a scan with nothing pushed into it sees.
     //
-    // `join_key_batches` are the side batches a `join_keys` filter refers to —
-    // DuckDB turns an `IN (…)` list, and a semi-join's build side, into a
-    // filter that names a column in them rather than carrying the values
-    // inline. Without them such a filter has no values at all.
+    // `output_schema` is the authoritative unprojected bind output schema used
+    // to resolve v2 column indices. It is required for non-empty snapshots.
     static PushdownFilters parse(const std::string& ipc_bytes,
-                                 const std::vector<std::string>& join_key_batches = {});
+                                 const std::vector<std::string>& join_key_batches = {},
+                                 std::shared_ptr<arrow::Schema> output_schema = nullptr);
+
+    // Validate and atomically apply a dynamic v2 delta to this scan's state.
+    void apply_delta(const std::string& ipc_bytes);
 
     bool empty() const noexcept { return filters_.empty(); }
 
@@ -83,11 +85,8 @@ class PushdownFilters {
 
     // Apply every filter to `batch`, returning the surviving rows.
     //
-    // Best effort by design: a filter this cannot evaluate is skipped rather
-    // than failing the scan, because pushdown is an optimization and the
-    // engine re-checks the predicate itself. Dropping a row it should have
-    // kept would be a wrong answer; keeping one it could have dropped is only
-    // slower.
+    // Required predicates fail closed when they cannot be evaluated. Advisory
+    // predicates may be ignored, as prescribed by Filter Encoding v2.
     std::shared_ptr<arrow::RecordBatch> apply(
         const std::shared_ptr<arrow::RecordBatch>& batch) const;
 
@@ -107,7 +106,7 @@ class PushdownFilters {
     // a count — because the tests compare the string.
     std::string format() const;
 
-    // The parsed filter tree. Public only so the implementation's free
+    // The parsed expression tree. Public only so the implementation's free
     // helpers can name it; it is not part of the SDK's surface.
     struct Spec;
 
@@ -122,8 +121,11 @@ class PushdownFilters {
 
     std::vector<Filter> filters_;
     std::vector<std::shared_ptr<Spec>> specs_;
-    std::vector<std::shared_ptr<arrow::Array>> values_;
-    std::map<std::string, std::shared_ptr<arrow::Array>> join_keys_;
+    std::map<std::string, uint64_t> revisions_;
+    std::set<std::string> required_ids_;
+    std::vector<std::shared_ptr<arrow::RecordBatch>> join_keys_;
+    std::shared_ptr<arrow::Schema> output_schema_;
+    std::string evaluation_context_;
 };
 
 }  // namespace vgi

include/vgi/types.h

@@ -1,6 +1,7 @@
 // © Copyright 2025, 2026 Query Farm LLC - https://query.farm
 #pragma once
 
+#include <cstdint>
 #include <memory>
 #include <optional>
 #include <string>
@@ -176,6 +177,31 @@ inline constexpr const char* kNoOrderGuarantee = "NO_ORDER_GUARANTEE";
 inline constexpr const char* kFixedOrder = "FIXED_ORDER";
 }  // namespace order_preservations
 
+namespace filter_semantic_profiles {
+inline constexpr const char* kDuckDBStandardV1 = "vgi.duckdb.standard.v1";
+}  // namespace filter_semantic_profiles
+
+// A versioned extension function the worker can evaluate inside a Filter-v2
+// expression. The standard profile's built-in functions do not appear here.
+struct FilterFunctionCapability {
+    std::string namespace_name;
+    std::string name;
+    uint64_t version = 0;
+};
+
+// A versioned runtime-filter artifact algorithm the worker can evaluate.
+struct RuntimeFilterAlgorithmCapability {
+    std::string namespace_name;
+    std::string name;
+    uint64_t version = 0;
+};
+
+// An evaluation-context profile the worker can apply in an isolated session.
+struct EvaluationContextCapability {
+    std::string profile;
+    std::optional<std::string> provider_fingerprint;
+};
+
 // Everything the engine shows a user about a function, plus the return type
 // when it is fixed.  A function whose return type depends on its arguments
 // leaves `return_type` empty and answers during bind instead.
@@ -226,6 +252,19 @@ struct FunctionMetadata {
     // On, the framework filters each emitted batch, which is what a fixture
     // that merely advertises the capability wants.
     bool auto_apply_filters = false;
+    // Filter Encoding v2 semantics implemented by this function. A function
+    // that enables filter_pushdown and leaves this empty advertises the C++
+    // SDK's standard-v1 evaluator.
+    std::vector<std::string> filter_semantic_profiles;
+    // Capability-gated extensions to the standard profile. The C++ SDK
+    // currently rejects non-empty lists until matching evaluators exist.
+    std::vector<FilterFunctionCapability> additional_filter_functions;
+    std::vector<RuntimeFilterAlgorithmCapability> runtime_filter_algorithms;
+    std::vector<EvaluationContextCapability> filter_evaluation_contexts;
+    // True only when the worker applies every pushed predicate exactly and
+    // the engine may therefore remove its residual. Never inferred from
+    // auto_apply_filters.
+    bool filters_exactly_applied = false;
     // Whether the engine may rewrite a scan of this function into a
     // late-materialization plan: fetch the row ids first, then fetch only the
     // surviving rows' columns.
@@ -269,6 +308,10 @@ struct FunctionMetadata {
     // engine forward their values; a setting not declared here never arrives,
     // however it was set.
     std::vector<std::string> required_settings;
+
+    // Resolve the wire advertisement, applying the SDK default and rejecting
+    // semantic profiles for which this SDK has no evaluator.
+    std::vector<std::string> resolved_filter_semantic_profiles() const;
 };
 
 }  // namespace vgi