From cf51e222276f9f56b0e151d2219ace59ed8b6bd7 Mon Sep 17 00:00:00 2001 From: Rusty Conover Date: Thu, 10 Sep 2026 21:15:06 -0400 Subject: [PATCH 1/5] Implement Filter Encoding v2 consumer --- include/vgi/pushdown.h | 38 +- src/function_dispatch.cpp | 44 +- src/pushdown.cpp | 1238 ++++++++++++++++++++++++++----------- tests/function_test.cpp | 58 ++ 4 files changed, 967 insertions(+), 411 deletions(-) diff --git a/include/vgi/pushdown.h b/include/vgi/pushdown.h index 6ed34d8..2e2cc53 100644 --- a/include/vgi/pushdown.h +++ b/include/vgi/pushdown.h @@ -5,6 +5,7 @@ #include #include #include +#include #include #include @@ -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& join_key_batches = {}); + const std::vector& join_key_batches = {}, + std::shared_ptr 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 apply( const std::shared_ptr& 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 filters_; std::vector> specs_; - std::vector> values_; - std::map> join_keys_; + std::map revisions_; + std::set required_ids_; + std::vector> join_keys_; + std::shared_ptr output_schema_; + std::string evaluation_context_; }; } // namespace vgi diff --git a/src/function_dispatch.cpp b/src/function_dispatch.cpp index e01ed6f..3cbcc26 100644 --- a/src/function_dispatch.cpp +++ b/src/function_dispatch.cpp @@ -287,10 +287,12 @@ std::shared_ptr narrow_to(const std::shared_ptr producer, PushdownFilters filters = {}, - std::shared_ptr output_schema = nullptr) + std::shared_ptr output_schema = nullptr, + bool auto_apply_filters = false) : producer_(std::move(producer)), filters_(std::move(filters)), - output_schema_(std::move(output_schema)) {} + output_schema_(std::move(output_schema)), + auto_apply_filters_(auto_apply_filters) {} // Conditional-request validators from the init request, which is where // they arrive for a producer: over HTTP the first tick is folded into the @@ -319,18 +321,14 @@ class TableProduce : public vgi_rpc::ProducerState { // below runs first, and a producer holding a collector from an earlier // tick would write into one that has already been destroyed. bind_log(out); - // Re-read every tick, not only the first: the engine re-sends them as - // the join's build side fills in, and the last one is the tightest. - // - // Kept per-tick rather than folded into `filters_`: they describe this - // tick's build side, and a later tick that carries none must fall back - // to the static predicates rather than keep stale dynamic ones. - dynamic_filters_.reset(); + // Dynamic Filter v2 metadata is a revisioned delta over the init-time + // snapshot, not a standalone replacement tree. Keep the accumulated + // state when a later tick carries no update. if (auto encoded = metadata_value(input.custom_metadata, keys::kDynamicFilters)) { auto decoded = wire::base64_decode(*encoded); - dynamic_filters_ = PushdownFilters::parse(decoded); - if (producer_) producer_->on_dynamic_filters(*dynamic_filters_); + filters_.apply_delta(decoded); } + if (producer_) producer_->on_dynamic_filters(filters_); produce(out, context); } @@ -353,11 +351,8 @@ class TableProduce : public vgi_rpc::ProducerState { } // Applied here rather than in the producer so a function that only // advertises the capability gets it for free, and one that uses the - // filters itself is not filtered twice. This tick's dynamic filters - // supersede the static ones when it carried any — they are the - // tighter predicate, derived from the same scan. - const PushdownFilters& active = dynamic_filters_ ? *dynamic_filters_ : filters_; - if (!active.empty()) batch = active.apply(batch); + // filters itself is not filtered twice. + if (auto_apply_filters_ && !filters_.empty()) batch = filters_.apply(batch); // Validated before it leaves. // @@ -400,13 +395,12 @@ class TableProduce : public vgi_rpc::ProducerState { std::unique_ptr producer_; PushdownFilters filters_; - // This tick's join-side predicate, if it carried one. - std::optional dynamic_filters_; // What the engine asked for, which a producer that ignores the projection // does not emit. std::shared_ptr output_schema_; std::optional if_none_match_; std::optional if_modified_since_; + bool auto_apply_filters_ = false; // Asked once, before the first batch: a producer that answered // `not_modified` has nothing more to decide. bool asked_ = false; @@ -1058,6 +1052,8 @@ vgi_rpc::Result Dispatcher::table_function_plan(const vgi_rpc::Request& request) PlanResult result; auto table = find_table(function_name, scope_of(bind_params), &bind_params); if (table && table->supports_splits()) { + const auto output_schema = table->bind(bind_params); + if (!output_schema) throw std::runtime_error("plan: function bound to no output schema"); PlanParams plan; plan.target_split_bytes = wire::get_optional_int64(plan_request, "target_split_bytes"); plan.min_splits = wire::get_optional_int64(plan_request, "min_splits"); @@ -1072,7 +1068,7 @@ vgi_rpc::Result Dispatcher::table_function_plan(const vgi_rpc::Request& request) } plan.pushdown_filters = PushdownFilters::parse( wire::get_optional_binary(plan_request, "pushdown_filters").value_or(std::string{}), - wire::get_binary_list(plan_request, "join_keys")); + wire::get_binary_list(plan_request, "join_keys"), output_schema); result = table->plan(bind_params, plan); } else { // The whole scan as one unit. This is what a function that has not @@ -1245,6 +1241,7 @@ vgi_rpc::Stream Dispatcher::init(const vgi_rpc::Request& request) { // worker re-derive it, so a function whose bind is expensive pays once. auto output_schema = wire::get_schema(init_request, "output_schema"); if (!output_schema) throw std::runtime_error("init: request carries no output_schema"); + const auto bind_output_schema = output_schema; // Projection pushdown arrives as indices into the bound schema rather than // as a narrowed schema, so the narrowing has to happen here: a function @@ -1345,7 +1342,7 @@ vgi_rpc::Stream Dispatcher::init(const vgi_rpc::Request& request) { // Parsed once for the whole scan; the engine sends it on init. params.pushdown_filters = PushdownFilters::parse( wire::get_optional_binary(init_request, "pushdown_filters").value_or(std::string{}), - wire::get_binary_list(init_request, "join_keys")); + wire::get_binary_list(init_request, "join_keys"), bind_output_schema); int64_t max_workers = 1; // Same tie-break as `bind`: a COPY TO whose writer shares a name with a @@ -1416,10 +1413,9 @@ vgi_rpc::Stream Dispatcher::init(const vgi_rpc::Request& request) { if (auto table = find_table(function_name, scope_of(params), &bind_params)) { stream.input_schema = arrow::schema({}); - auto auto_apply = - table->metadata().auto_apply_filters ? params.pushdown_filters : PushdownFilters{}; - auto produce = std::make_shared(table->init(params), std::move(auto_apply), - output_schema); + const bool auto_apply = table->metadata().auto_apply_filters; + auto produce = std::make_shared(table->init(params), params.pushdown_filters, + output_schema, auto_apply); produce->set_validators(request_metadata(request, keys::kIfNoneMatch), request_metadata(request, keys::kIfModifiedSince)); stream.state = std::move(produce); diff --git a/src/pushdown.cpp b/src/pushdown.cpp index 2a0fb7e..3118a60 100644 --- a/src/pushdown.cpp +++ b/src/pushdown.cpp @@ -5,52 +5,407 @@ #include #include #include +#include +#include +#include +#include #include #include #include +#include +#include +#include #include #include "wire.h" namespace vgi { -// The filter tree as it arrives. Kept out of the header: the public `Filter` -// is a flattened view, and callers should not have to walk an AST to answer -// "is this column constrained". struct PushdownFilters::Spec { std::string kind; std::string column_name; + size_t column_index = 0; + size_t field_index = 0; + std::string field_name; std::string op; - std::optional value_ref; - // A `join_keys` filter names a column in the side batches rather than - // carrying its values inline. - std::string keys_column; + std::string function; + std::shared_ptr data_type; + std::shared_ptr value; + bool negated = false; + bool advisory = false; + std::string id; + uint64_t revision = 0; std::vector> children; }; using Spec = PushdownFilters::Spec; +using json = nlohmann::json; namespace { -using nlohmann::json; +constexpr size_t kMaxPayloadBytes = 16U << 20U; +constexpr size_t kMaxJsonBytes = 1U << 20U; +constexpr size_t kMaxDepth = 64; +constexpr size_t kMaxNodes = 10000; +constexpr size_t kMaxPredicates = 1024; +constexpr size_t kMaxPredicateIds = 4096; -std::shared_ptr parse_spec(const json& node) { - auto spec = std::make_shared(); - spec->kind = node.value("type", ""); - spec->column_name = node.value("column_name", ""); - if (node.contains("op") && node["op"].is_string()) spec->op = node["op"].get(); - if (node.contains("value_ref") && node["value_ref"].is_number_unsigned()) { - spec->value_ref = node["value_ref"].get(); +[[noreturn]] void invalid(const std::string& message) { + throw std::invalid_argument("Filter Encoding v2: " + message); +} + +template +T value_or_throw(arrow::Result result, const std::string& context) { + if (!result.ok()) throw std::runtime_error(context + ": " + result.status().ToString()); + return std::move(result).ValueUnsafe(); +} + +void require_keys(const json& object, std::initializer_list required, + std::initializer_list optional, const std::string& where) { + if (!object.is_object()) invalid(where + " must be an object"); + std::unordered_set allowed; + for (const auto* key : required) { + allowed.insert(key); + if (!object.contains(key)) invalid(where + " is missing '" + key + "'"); + } + for (const auto* key : optional) allowed.insert(key); + for (auto it = object.begin(); it != object.end(); ++it) { + if (!allowed.count(it.key())) invalid(where + " has unknown property '" + it.key() + "'"); + } +} + +std::string required_string(const json& object, const char* key, const std::string& where) { + const auto& value = object.at(key); + if (!value.is_string() || value.get_ref().empty()) { + invalid(where + "." + key + " must be a nonempty string"); + } + return value.get(); +} + +uint64_t required_uint(const json& object, const char* key, const std::string& where) { + const auto& value = object.at(key); + if (!value.is_number_unsigned()) invalid(where + "." + key + " must be uint64"); + return value.get(); +} + +bool canonical_payload_name(const std::string& name) { + for (const auto* prefix : {"value_", "type_", "artifact_"}) { + if (name.rfind(prefix, 0) != 0) continue; + const auto suffix = name.substr(std::char_traits::length(prefix)); + if (suffix.empty() || (suffix.size() > 1 && suffix.front() == '0')) return false; + return std::all_of(suffix.begin(), suffix.end(), + [](char c) { return c >= '0' && c <= '9'; }); + } + return false; +} + +std::string metadata_value(const std::shared_ptr& schema, const std::string& key) { + if (!schema->metadata()) invalid("missing schema metadata '" + key + "'"); + auto value = schema->metadata()->Get(key); + if (!value.ok()) invalid("missing schema metadata '" + key + "'"); + return value.ValueUnsafe(); +} + +std::string validate_batch(const std::shared_ptr& batch) { + if (!batch || batch->num_rows() != 1) + invalid("filter RecordBatch must contain exactly one row"); + if (batch->num_columns() == 0) invalid("filter RecordBatch has no filter_spec field"); + const auto& first = batch->schema()->field(0); + if (first->name() != "filter_spec" || first->type()->id() != arrow::Type::STRING || + first->nullable()) { + invalid("first field must be filter_spec: utf8 not null"); + } + auto text = std::dynamic_pointer_cast(batch->column(0)); + if (!text || text->IsNull(0)) invalid("filter_spec value must not be NULL"); + if (text->value_length(0) > static_cast(kMaxJsonBytes)) + invalid("filter JSON exceeds 1 MiB"); + std::unordered_set names{"filter_spec"}; + for (int i = 1; i < batch->num_columns(); ++i) { + const auto& field = batch->schema()->field(i); + if (!names.insert(field->name()).second || !canonical_payload_name(field->name())) { + invalid("noncanonical or duplicate payload field '" + field->name() + "'"); + } + if (field->name().rfind("type_", 0) == 0 && !batch->column(i)->IsNull(0)) { + invalid("type payload must contain NULL"); + } + } + if (metadata_value(batch->schema(), "vgi_filter_encoding") != "vgi.filters.v2" || + metadata_value(batch->schema(), "vgi_filter_version") != "2") { + invalid("unsupported filter encoding/version"); + } + const auto context = metadata_value(batch->schema(), "vgi_evaluation_context"); + if (context != "vgi.none.v1") + invalid("evaluation context '" + context + "' was not advertised"); + for (const auto* key : + {"vgi_time_zone", "vgi_calendar", "vgi_default_collation", "vgi_ieee_floating_point_ops", + "vgi_integer_division", "vgi_context_provider_fingerprint"}) { + if (batch->schema()->metadata() && batch->schema()->metadata()->Contains(key)) { + invalid("vgi.none.v1 forbids session-context metadata"); + } + } + return context; +} + +std::shared_ptr payload(const std::shared_ptr& batch, + const std::string& prefix, uint64_t reference) { + const auto name = prefix + "_" + std::to_string(reference); + const auto indices = batch->schema()->GetAllFieldIndices(name); + if (indices.size() != 1) invalid("missing or duplicate payload field '" + name + "'"); + return batch->column(indices.front()); +} + +std::string root_column(const std::shared_ptr& expression) { + if (!expression) return {}; + if (expression->kind == "column") return expression->column_name; + if (expression->kind == "field" && !expression->children.empty()) + return root_column(expression->children[0]); + return {}; +} + +struct Parser { + std::shared_ptr batch; + const std::vector>& join_keys; + std::shared_ptr output_schema; + size_t nodes = 0; + + std::shared_ptr expression(const json& node, size_t depth, bool root = false) { + if (depth > kMaxDepth) invalid("expression exceeds nesting-depth limit"); + if (++nodes > kMaxNodes) invalid("document exceeds expression-node limit"); + if (!node.is_object() || !node.contains("node") || !node.at("node").is_string()) { + invalid("expression must have a string node"); + } + const auto kind = node.at("node").get(); + auto result = std::make_shared(); + if (kind == "column_ref") { + require_keys(node, {"node", "column_index", "column_name"}, {}, "column_ref"); + result->kind = "column"; + result->column_index = + static_cast(required_uint(node, "column_index", "column_ref")); + result->column_name = required_string(node, "column_name", "column_ref"); + if (!output_schema || + result->column_index >= static_cast(output_schema->num_fields())) { + invalid("column_ref index is outside the authoritative output schema"); + } + const auto& field = output_schema->field(static_cast(result->column_index)); + if (field->name() != result->column_name) + invalid("column_ref name does not match authoritative index"); + result->data_type = field->type(); + return result; + } + if (kind == "field_ref") { + require_keys(node, {"node", "expression", "field_index", "field_name"}, {}, + "field_ref"); + result->kind = "field"; + result->children.push_back(expression(node.at("expression"), depth + 1)); + result->field_index = + static_cast(required_uint(node, "field_index", "field_ref")); + result->field_name = required_string(node, "field_name", "field_ref"); + const auto parent = result->children[0]->data_type; + if (!parent || parent->id() != arrow::Type::STRUCT) + invalid("field_ref input must be STRUCT"); + const auto& fields = static_cast(*parent).fields(); + if (result->field_index >= fields.size() || + fields[result->field_index]->name() != result->field_name) { + invalid("field_ref name/index does not match authoritative struct"); + } + result->column_name = root_column(result->children[0]); + result->data_type = fields[result->field_index]->type(); + return result; + } + if (kind == "literal") { + require_keys(node, {"node", "value_ref"}, {}, "literal"); + result->kind = "literal"; + result->value = payload(batch, "value", required_uint(node, "value_ref", "literal")); + result->data_type = result->value->type(); + return result; + } + if (kind == "comparison") { + require_keys(node, {"node", "op", "left", "right"}, {}, "comparison"); + result->kind = "constant"; + result->op = required_string(node, "op", "comparison"); + if (result->op != "eq" && result->op != "ne" && result->op != "lt" && + result->op != "le" && result->op != "gt" && result->op != "ge" && + result->op != "distinct_from" && result->op != "not_distinct_from") + invalid("unknown comparison operator '" + result->op + "'"); + result->children = {expression(node.at("left"), depth + 1), + expression(node.at("right"), depth + 1)}; + result->column_name = root_column(result->children[0]); + if (result->column_name.empty()) result->column_name = root_column(result->children[1]); + if (result->children[1]->kind == "literal") result->value = result->children[1]->value; + result->data_type = arrow::boolean(); + return result; + } + if (kind == "and" || kind == "or") { + require_keys(node, {"node", "children"}, {}, kind); + if (!node.at("children").is_array() || node.at("children").size() < 2) + invalid(kind + " requires at least two children"); + result->kind = kind; + for (const auto& child : node.at("children")) + result->children.push_back(expression(child, depth + 1)); + result->data_type = arrow::boolean(); + return result; + } + if (kind == "not") { + require_keys(node, {"node", "expression"}, {}, "not"); + result->kind = "not"; + result->children.push_back(expression(node.at("expression"), depth + 1)); + result->data_type = arrow::boolean(); + return result; + } + if (kind == "is_null") { + require_keys(node, {"node", "expression", "negated"}, {}, "is_null"); + if (!node.at("negated").is_boolean()) invalid("is_null.negated must be Boolean"); + result->negated = node.at("negated").get(); + result->kind = result->negated ? "is_not_null" : "is_null"; + result->children.push_back(expression(node.at("expression"), depth + 1)); + result->column_name = root_column(result->children[0]); + result->data_type = arrow::boolean(); + return result; + } + if (kind == "in") { + require_keys(node, {"node", "expression", "set", "negated"}, {}, "in"); + if (!node.at("negated").is_boolean()) invalid("in.negated must be Boolean"); + result->kind = "in"; + result->negated = node.at("negated").get(); + result->children.push_back(expression(node.at("expression"), depth + 1)); + result->column_name = root_column(result->children[0]); + const auto& set = node.at("set"); + if (!set.is_object() || !set.contains("kind") || !set.at("kind").is_string()) + invalid("in.set must have a string kind"); + if (set.at("kind") == "literal") { + require_keys(set, {"kind", "value_ref"}, {}, "in.set"); + auto list = payload(batch, "value", required_uint(set, "value_ref", "in.set")); + if (list->type_id() == arrow::Type::LIST) { + const auto& values = static_cast(*list); + if (values.IsNull(0)) invalid("literal IN list must not be NULL"); + result->value = values.value_slice(0); + } else if (list->type_id() == arrow::Type::LARGE_LIST) { + const auto& values = static_cast(*list); + if (values.IsNull(0)) invalid("literal IN list must not be NULL"); + result->value = values.value_slice(0); + } else + invalid("literal IN payload must be a list scalar"); + } else if (set.at("kind") == "external") { + require_keys(set, {"kind", "batch_index", "column_index", "column_name"}, {}, + "in.set"); + const auto bi = static_cast(required_uint(set, "batch_index", "in.set")); + const auto ci = static_cast(required_uint(set, "column_index", "in.set")); + const auto name = required_string(set, "column_name", "in.set"); + if (bi >= join_keys.size() || + ci >= static_cast(join_keys[bi]->num_columns())) + invalid("external IN batch/column index is unavailable"); + if (join_keys[bi]->schema()->field(static_cast(ci))->name() != name) + invalid("external IN column name does not match authoritative index"); + result->value = join_keys[bi]->column(static_cast(ci)); + } else + invalid("in.set has an unknown kind"); + result->data_type = arrow::boolean(); + return result; + } + if (kind == "cast") { + require_keys(node, {"node", "expression", "type_ref"}, {}, "cast"); + result->kind = "cast"; + result->children.push_back(expression(node.at("expression"), depth + 1)); + result->value = payload(batch, "type", required_uint(node, "type_ref", "cast")); + if (!result->value->IsNull(0)) invalid("cast type payload must contain NULL"); + result->data_type = result->value->type(); + return result; + } + if (kind == "arithmetic") { + require_keys(node, {"node", "op", "left", "right"}, {}, "arithmetic"); + result->kind = "arithmetic"; + result->op = required_string(node, "op", "arithmetic"); + if (result->op != "add" && result->op != "subtract" && result->op != "multiply" && + result->op != "divide" && result->op != "modulo") + invalid("unknown arithmetic operator"); + if (result->op == "divide" || result->op == "modulo") + invalid("context-dependent arithmetic requires vgi.duckdb.session.v1"); + result->children = {expression(node.at("left"), depth + 1), + expression(node.at("right"), depth + 1)}; + result->data_type = result->children[0]->data_type; + return result; + } + if (kind == "negate") { + require_keys(node, {"node", "expression"}, {}, "negate"); + result->kind = "negate"; + result->children.push_back(expression(node.at("expression"), depth + 1)); + result->data_type = result->children[0]->data_type; + return result; + } + if (kind == "call") { + require_keys(node, {"node", "function", "arguments"}, {"options"}, "call"); + if (!node.at("function").is_string()) + invalid("extension filter functions were not advertised"); + result->function = node.at("function").get(); + if (result->function != "starts_with" && result->function != "ends_with" && + result->function != "contains" && result->function != "list_contains") + invalid("unknown standard filter function"); + if (!node.at("arguments").is_array() || node.at("arguments").size() != 2) + invalid("standard filter functions require exactly two arguments"); + if (node.contains("options") && !node.at("options").empty()) + invalid("standard filter functions do not accept options"); + result->kind = "call"; + for (const auto& argument : node.at("arguments")) + result->children.push_back(expression(argument, depth + 1)); + result->data_type = arrow::boolean(); + return result; + } + if (kind == "runtime_filter") { + require_keys(node, {"node", "algorithm", "input", "artifact_ref", "null_handling"}, {}, + "runtime_filter"); + if (!root) invalid("runtime_filter may appear only at a predicate root"); + result->kind = "runtime_filter"; + result->children.push_back(expression(node.at("input"), depth + 1)); + (void)payload(batch, "artifact", required_uint(node, "artifact_ref", "runtime_filter")); + return result; + } + invalid("unknown expression node '" + kind + "'"); } - spec->keys_column = node.value("keys_column", ""); - if (node.contains("children") && node["children"].is_array()) { - for (const auto& child : node["children"]) spec->children.push_back(parse_spec(child)); +}; + +json document_for(const std::shared_ptr& batch) { + const auto text = std::static_pointer_cast(batch->column(0))->GetString(0); + if (std::getenv("VGI_FILTER_DEBUG")) std::fprintf(stderr, "[vgi-filter] %s\n", text.c_str()); + try { + return json::parse(text); + } catch (const json::exception& error) { + invalid(std::string("invalid filter JSON: ") + error.what()); } - if (node.contains("child_filter") && node["child_filter"].is_object()) { - spec->children.push_back(parse_spec(node["child_filter"])); +} + +void validate_header(const json& document, const char* kind, const char* member) { + require_keys(document, {"encoding", "semantics", "kind", member}, {}, "filter document"); + if (document.at("encoding") != "vgi.filters.v2") + invalid("document encoding must be vgi.filters.v2"); + if (document.at("semantics") != "vgi.duckdb.standard.v1") + invalid("unsupported filter semantics"); + if (document.at("kind") != kind) + invalid(std::string("expected a ") + kind + " filter document"); + if (!document.at(member).is_array()) invalid(std::string(member) + " must be an array"); +} + +std::shared_ptr parse_predicate(Parser& parser, const json& item, bool delta) { + require_keys(item, {"id", "revision", "mode", "source", "expression"}, + delta ? std::initializer_list{"operation"} + : std::initializer_list{}, + "predicate"); + auto expression = parser.expression(item.at("expression"), 1, true); + expression->id = required_string(item, "id", "predicate"); + if (expression->id.size() > 128) invalid("predicate ID exceeds 128 bytes"); + expression->revision = required_uint(item, "revision", "predicate"); + const auto mode = required_string(item, "mode", "predicate"); + if (mode != "required" && mode != "advisory") invalid("unknown predicate mode"); + if (delta && mode != "advisory") invalid("delta upserts must be advisory"); + expression->advisory = mode == "advisory"; + const auto source = required_string(item, "source", "predicate"); + if (source != "query" && source != "join" && source != "top_n" && + source != "split_refinement" && source != "other") + invalid("unknown predicate source"); + if (expression->kind == "runtime_filter" && !expression->advisory) { + invalid("runtime_filter predicates must be advisory"); } - return spec; + return expression; } Filter::Kind kind_of(const std::string& kind) { @@ -58,425 +413,570 @@ Filter::Kind kind_of(const std::string& kind) { if (kind == "in") return Filter::Kind::In; if (kind == "is_null") return Filter::Kind::IsNull; if (kind == "is_not_null") return Filter::Kind::IsNotNull; - if (kind == "join_keys") return Filter::Kind::JoinKeys; return Filter::Kind::Other; } +void collect_columns(const std::shared_ptr& spec, std::vector& columns) { + if (spec->kind == "column") columns.push_back(spec->column_name); + for (const auto& child : spec->children) collect_columns(child, columns); +} + void flatten(const std::shared_ptr& spec, std::vector& out) { if (spec->kind == "and" || spec->kind == "or") { for (const auto& child : spec->children) flatten(child, out); return; } - out.push_back({kind_of(spec->kind), spec->column_name, spec->op}); + std::vector columns; + collect_columns(spec, columns); + if (columns.empty()) out.push_back({kind_of(spec->kind), spec->column_name, spec->op}); + for (const auto& column : columns) out.push_back({kind_of(spec->kind), column, spec->op}); } -// The comparison this op denotes, in Arrow's vocabulary. The protocol has used -// several spellings for the same operator over time, so all are accepted. -const char* comparison_kernel(const std::string& op) { - if (op == "eq" || op == "=" || op == "==") return "equal"; - if (op == "ne" || op == "!=" || op == "<>") return "not_equal"; - if (op == "lt" || op == "<") return "less"; - if (op == "le" || op == "lteq" || op == "<=") return "less_equal"; - if (op == "gt" || op == ">") return "greater"; - if (op == "ge" || op == "gteq" || op == ">=") return "greater_equal"; - return nullptr; +std::shared_ptr array_from_datum(const arrow::Datum& datum, int64_t length) { + if (datum.is_array()) { + auto array = datum.make_array(); + if (array->length() != length) + throw std::runtime_error("filter expression length mismatch"); + return array; + } + if (!datum.is_scalar()) + throw std::runtime_error("filter expression did not yield an array or scalar"); + return value_or_throw(arrow::MakeArrayFromScalar(*datum.scalar(), length), + "broadcast filter scalar"); +} + +arrow::Datum decoded_dictionary(arrow::Datum value) { + if (!value.is_array() || value.type()->id() != arrow::Type::DICTIONARY) return value; + const auto& dictionary = static_cast(*value.type()); + return value_or_throw(arrow::compute::Cast(value, dictionary.value_type()), + "decode dictionary filter input"); +} + +arrow::Datum call(const std::string& name, std::vector arguments) { + static const auto initialized = arrow::compute::Initialize(); + if (!initialized.ok()) throw std::runtime_error(initialized.ToString()); + for (auto& argument : arguments) argument = decoded_dictionary(std::move(argument)); + return value_or_throw(arrow::compute::CallFunction(name, std::move(arguments)), + "evaluate " + name); +} + +std::optional string_at(const std::shared_ptr& values, int64_t index) { + if (values->IsNull(index)) return std::nullopt; + if (values->type_id() == arrow::Type::STRING) { + return static_cast(*values).GetString(index); + } + if (values->type_id() == arrow::Type::LARGE_STRING) { + return static_cast(*values).GetString(index); + } + throw std::runtime_error("standard string filter requires UTF8 arguments"); +} + +arrow::Datum evaluate(const std::shared_ptr& spec, + const std::shared_ptr& batch); + +arrow::Datum evaluate_standard_call(const Spec& spec, + const std::shared_ptr& batch) { + auto left = array_from_datum(evaluate(spec.children[0], batch), batch->num_rows()); + auto right = array_from_datum(evaluate(spec.children[1], batch), batch->num_rows()); + arrow::BooleanBuilder output; + auto status = output.Reserve(batch->num_rows()); + if (!status.ok()) throw std::runtime_error(status.ToString()); + for (int64_t row = 0; row < batch->num_rows(); ++row) { + if (left->IsNull(row) || right->IsNull(row)) { + (void)output.AppendNull(); + continue; + } + bool matched = false; + if (spec.function == "list_contains") { + std::shared_ptr values; + if (left->type_id() == arrow::Type::LIST) { + values = static_cast(*left).value_slice(row); + } else if (left->type_id() == arrow::Type::LARGE_LIST) { + values = static_cast(*left).value_slice(row); + } else { + throw std::runtime_error("list_contains requires a list argument"); + } + const auto needle = value_or_throw(right->GetScalar(row), "read list_contains needle"); + for (int64_t i = 0; i < values->length(); ++i) { + if (values->IsNull(i)) continue; + const auto candidate = + value_or_throw(values->GetScalar(i), "read list_contains value"); + if (candidate->Equals(*needle)) { + matched = true; + break; + } + } + } else { + const auto haystack = *string_at(left, row); + const auto needle = *string_at(right, row); + if (spec.function == "starts_with") matched = haystack.rfind(needle, 0) == 0; + if (spec.function == "ends_with") { + matched = + haystack.size() >= needle.size() && + haystack.compare(haystack.size() - needle.size(), needle.size(), needle) == 0; + } + if (spec.function == "contains") matched = haystack.find(needle) != std::string::npos; + } + (void)output.Append(matched); + } + std::shared_ptr result; + status = output.Finish(&result); + if (!status.ok()) throw std::runtime_error(status.ToString()); + return arrow::Datum(result); +} + +arrow::Datum evaluate(const std::shared_ptr& spec, + const std::shared_ptr& batch) { + if (spec->kind == "column") { + int index = -1; + if (spec->column_index < static_cast(batch->num_columns()) && + batch->schema()->field(static_cast(spec->column_index))->name() == + spec->column_name) { + index = static_cast(spec->column_index); + } else { + const auto indices = batch->schema()->GetAllFieldIndices(spec->column_name); + if (indices.size() == 1) index = indices.front(); + } + if (index < 0) + throw std::runtime_error("filter column '" + spec->column_name + "' is absent"); + return arrow::Datum(batch->column(index)); + } + if (spec->kind == "field") { + auto parent = array_from_datum(evaluate(spec->children[0], batch), batch->num_rows()); + auto values = std::dynamic_pointer_cast(parent); + if (!values || spec->field_index >= static_cast(values->num_fields())) { + throw std::runtime_error("filter field_ref input is not the authoritative struct"); + } + return arrow::Datum(values->field(static_cast(spec->field_index))); + } + if (spec->kind == "literal") { + return arrow::Datum(value_or_throw(spec->value->GetScalar(0), "read filter literal")); + } + if (spec->kind == "constant") { + static const std::map kernels = { + {"eq", "equal"}, + {"ne", "not_equal"}, + {"lt", "less"}, + {"le", "less_equal"}, + {"gt", "greater"}, + {"ge", "greater_equal"}, + {"distinct_from", "is_distinct_from"}, + {"not_distinct_from", "is_not_distinct_from"}, + }; + const auto found = kernels.find(spec->op); + if (found == kernels.end()) throw std::runtime_error("unsupported comparison operator"); + return call(found->second, + {evaluate(spec->children[0], batch), evaluate(spec->children[1], batch)}); + } + if (spec->kind == "and" || spec->kind == "or") { + auto result = evaluate(spec->children[0], batch); + for (size_t i = 1; i < spec->children.size(); ++i) { + result = call(spec->kind == "and" ? "and_kleene" : "or_kleene", + {std::move(result), evaluate(spec->children[i], batch)}); + } + return result; + } + if (spec->kind == "not") return call("invert", {evaluate(spec->children[0], batch)}); + if (spec->kind == "is_null" || spec->kind == "is_not_null") { + return call(spec->kind == "is_null" ? "is_null" : "is_valid", + {evaluate(spec->children[0], batch)}); + } + if (spec->kind == "in") { + static const auto initialized = arrow::compute::Initialize(); + if (!initialized.ok()) throw std::runtime_error(initialized.ToString()); + arrow::compute::SetLookupOptions options(spec->value); + auto result = value_or_throw( + arrow::compute::CallFunction( + "is_in", {decoded_dictionary(evaluate(spec->children[0], batch))}, &options), + "evaluate IN"); + return spec->negated ? call("invert", {std::move(result)}) : result; + } + if (spec->kind == "cast") { + return value_or_throw( + arrow::compute::Cast(evaluate(spec->children[0], batch), spec->data_type), + "evaluate cast"); + } + if (spec->kind == "arithmetic") { + return call(spec->op, + {evaluate(spec->children[0], batch), evaluate(spec->children[1], batch)}); + } + if (spec->kind == "negate") return call("negate", {evaluate(spec->children[0], batch)}); + if (spec->kind == "call") return evaluate_standard_call(*spec, batch); + throw std::runtime_error("runtime filter has no negotiated evaluator"); +} + +std::string format_scalar(const std::shared_ptr& array, int64_t index) { + if (!array || index >= array->length() || array->IsNull(index)) return "NULL"; + if (array->type_id() == arrow::Type::STRING) { + return "'" + static_cast(*array).GetString(index) + "'"; + } + if (array->type_id() == arrow::Type::LARGE_STRING) { + return "'" + static_cast(*array).GetString(index) + "'"; + } + if (array->type_id() == arrow::Type::BOOL) { + return static_cast(*array).Value(index) ? "True" : "False"; + } + auto casted = arrow::compute::Cast(*array->Slice(index, 1), arrow::utf8()); + if (!casted.ok()) return {}; + return std::static_pointer_cast(casted.MoveValueUnsafe())->GetString(0); +} + +const char* op_symbol(const std::string& op) { + if (op == "eq") return "="; + if (op == "ne") return "!="; + if (op == "lt") return "<"; + if (op == "le") return "<="; + if (op == "gt") return ">"; + if (op == "ge") return ">="; + if (op == "distinct_from") return "IS DISTINCT FROM"; + if (op == "not_distinct_from") return "IS NOT DISTINCT FROM"; + return "?"; +} + +std::string render(const std::shared_ptr& spec) { + if (spec->kind == "column") return spec->column_name; + if (spec->kind == "field") return render(spec->children[0]) + "." + spec->field_name; + if (spec->kind == "literal") return format_scalar(spec->value, 0); + if (spec->kind == "constant") { + return render(spec->children[0]) + " " + op_symbol(spec->op) + " " + + render(spec->children[1]); + } + if (spec->kind == "and" || spec->kind == "or") { + std::string result = "("; + for (size_t i = 0; i < spec->children.size(); ++i) { + if (i) result += spec->kind == "and" ? " AND " : " OR "; + result += render(spec->children[i]); + } + return result + ")"; + } + if (spec->kind == "not") return "NOT (" + render(spec->children[0]) + ")"; + if (spec->kind == "is_null" || spec->kind == "is_not_null") { + return render(spec->children[0]) + (spec->kind == "is_null" ? " IS NULL" : " IS NOT NULL"); + } + if (spec->kind == "in") { + std::string values; + if (spec->value && spec->value->length() > 20) { + values = std::to_string(spec->value->length()) + " values"; + } else { + for (int64_t i = 0; spec->value && i < spec->value->length(); ++i) { + if (i) values += ", "; + values += format_scalar(spec->value, i); + } + } + return render(spec->children[0]) + (spec->negated ? " NOT IN (" : " IN (") + values + ")"; + } + return "(expression)"; +} + +std::string render_repr(const std::shared_ptr& spec) { + if (spec->kind == "constant") return "ConstantFilter(" + render(spec) + ")"; + if (spec->kind == "and" || spec->kind == "or") { + std::string result = spec->kind == "and" ? "AndFilter([" : "OrFilter(["; + for (size_t i = 0; i < spec->children.size(); ++i) { + if (i) result += ", "; + result += render_repr(spec->children[i]); + } + return result + "])"; + } + return render(spec); } -bool is_equality(const std::string& op) { - return op.empty() || op == "eq" || op == "=" || op == "=="; +bool mentions(const std::shared_ptr& spec, const std::string& column) { + std::vector columns; + collect_columns(spec, columns); + return std::find(columns.begin(), columns.end(), column) != columns.end(); +} + +std::shared_ptr discrete_values(const std::shared_ptr& spec, + const std::string& column) { + if (spec->kind == "constant" && spec->op == "eq" && spec->children.size() == 2 && + root_column(spec->children[0]) == column && spec->children[1]->kind == "literal") { + return spec->children[1]->value; + } + if (spec->kind == "in" && !spec->negated && !spec->children.empty() && + root_column(spec->children[0]) == column) + return spec->value; + if (spec->kind == "and") { + for (const auto& child : spec->children) { + if (auto found = discrete_values(child, column)) return found; + } + return nullptr; + } + if (spec->kind == "or") { + arrow::ArrayVector arrays; + for (const auto& child : spec->children) { + auto found = discrete_values(child, column); + if (!found) return nullptr; + arrays.push_back(std::move(found)); + } + if (arrays.empty()) return nullptr; + auto combined = arrow::Concatenate(arrays); + if (!combined.ok()) return nullptr; + auto unique = arrow::compute::Unique(combined.MoveValueUnsafe()); + return unique.ok() ? unique.MoveValueUnsafe() : nullptr; + } + return nullptr; } -std::optional as_int64(const std::shared_ptr& array) { - if (!array || array->length() == 0 || array->IsNull(0)) return std::nullopt; - auto casted = arrow::compute::Cast(*array, arrow::int64()); +struct Bounds { + std::optional min; + std::optional max; +}; + +std::optional scalar_int64(const std::shared_ptr& value) { + if (!value || value->length() == 0 || value->IsNull(0)) return std::nullopt; + auto casted = arrow::compute::Cast(*value->Slice(0, 1), arrow::int64()); if (!casted.ok()) return std::nullopt; return std::static_pointer_cast(casted.MoveValueUnsafe())->Value(0); } +Bounds intersect(Bounds left, const Bounds& right) { + if (right.min) left.min = left.min ? std::max(*left.min, *right.min) : right.min; + if (right.max) left.max = left.max ? std::min(*left.max, *right.max) : right.max; + return left; +} + +Bounds unite(Bounds left, const Bounds& right) { + if (left.min && right.min) + left.min = std::min(*left.min, *right.min); + else + left.min.reset(); + if (left.max && right.max) + left.max = std::max(*left.max, *right.max); + else + left.max.reset(); + return left; +} + +std::optional bounds_for(const std::shared_ptr& spec, const std::string& column) { + if (spec->kind == "constant" && spec->children.size() == 2) { + auto op = spec->op; + std::shared_ptr literal; + if (root_column(spec->children[0]) == column && spec->children[1]->kind == "literal") { + literal = spec->children[1]; + } else if (root_column(spec->children[1]) == column && + spec->children[0]->kind == "literal") { + literal = spec->children[0]; + if (op == "gt") + op = "lt"; + else if (op == "ge") + op = "le"; + else if (op == "lt") + op = "gt"; + else if (op == "le") + op = "ge"; + } + auto value = literal ? scalar_int64(literal->value) : std::nullopt; + if (!value) return std::nullopt; + Bounds result; + if (op == "eq") result.min = result.max = *value; + if (op == "gt") + result.min = *value == std::numeric_limits::max() ? *value : *value + 1; + if (op == "ge") result.min = *value; + if (op == "lt") + result.max = *value == std::numeric_limits::min() ? *value : *value - 1; + if (op == "le") result.max = *value; + if (!result.min && !result.max) return std::nullopt; + return result; + } + if (spec->kind == "in" && !spec->negated && root_column(spec->children[0]) == column) { + auto casted = arrow::compute::Cast(*spec->value, arrow::int64()); + if (!casted.ok()) return std::nullopt; + const auto values = std::static_pointer_cast(casted.MoveValueUnsafe()); + Bounds result; + for (int64_t i = 0; i < values->length(); ++i) { + if (values->IsNull(i)) continue; + result.min = result.min ? std::min(*result.min, values->Value(i)) : values->Value(i); + result.max = result.max ? std::max(*result.max, values->Value(i)) : values->Value(i); + } + return result.min ? std::optional(result) : std::nullopt; + } + if (spec->kind == "and" || spec->kind == "or") { + std::optional result; + for (const auto& child : spec->children) { + auto current = bounds_for(child, column); + if (!current) { + if (spec->kind == "or") return std::nullopt; + continue; + } + result = result ? (spec->kind == "and" ? intersect(*result, *current) + : unite(*result, *current)) + : current; + } + return result; + } + return std::nullopt; +} + } // namespace PushdownFilters PushdownFilters::parse(const std::string& ipc_bytes, - const std::vector& join_key_batches) { + const std::vector& join_key_batches, + std::shared_ptr output_schema) { PushdownFilters filters; - for (const auto& blob : join_key_batches) { - auto batch = wire::decode_ipc(blob); - if (!batch) continue; - for (int i = 0; i < batch->num_columns(); ++i) { - // Last batch wins for a repeated column name. Safe under the - // contract that extra rows are only slower — a scan that misses a - // key emits more than it had to, and the engine still filters. - filters.join_keys_[batch->schema()->field(i)->name()] = batch->column(i); - } + filters.output_schema_ = std::move(output_schema); + for (const auto& bytes : join_key_batches) { + auto batch = wire::decode_ipc(bytes); + if (!batch) invalid("join-key IPC stream contains no batch"); + filters.join_keys_.push_back(std::move(batch)); } if (ipc_bytes.empty()) return filters; - + if (ipc_bytes.size() > kMaxPayloadBytes) invalid("filter payload exceeds 16 MiB"); auto batch = wire::decode_ipc(ipc_bytes); - if (!batch || batch->num_columns() == 0) return filters; - - auto encoded = std::dynamic_pointer_cast(batch->column(0)); - if (!encoded || encoded->length() == 0 || encoded->IsNull(0)) return filters; - - json tree; - try { - tree = json::parse(encoded->GetString(0)); - } catch (const json::exception&) { - // A filter blob we cannot read means "no filters", not a failed scan. - // The engine re-checks every predicate, so ignoring one costs speed - // and never correctness. - return filters; - } - // VGI_FILTER_DEBUG=1 prints the filter tree. Which predicates DuckDB - // actually pushes is not obvious — an `IN` list arrives as a `join_keys` - // filter with no inline values at all — and this is the quickest way to - // find out. - if (std::getenv("VGI_FILTER_DEBUG")) { - std::fprintf(stderr, "[vgi-filter] %s\n", encoded->GetString(0).c_str()); - } - if (!tree.is_array()) return filters; - - for (const auto& node : tree) filters.specs_.push_back(parse_spec(node)); - // value_ref N is column N + 1; column 0 held the tree. - for (int i = 1; i < batch->num_columns(); ++i) { - filters.values_.push_back(batch->column(i)); + filters.evaluation_context_ = validate_batch(batch); + auto document = document_for(batch); + validate_header(document, "snapshot", "predicates"); + if (document.at("predicates").size() > kMaxPredicates) + invalid("snapshot exceeds predicate limit"); + if (!document.at("predicates").empty() && !filters.output_schema_) { + invalid("non-empty snapshot requires the authoritative unprojected bind output schema"); + } + Parser parser{batch, filters.join_keys_, filters.output_schema_}; + for (const auto& item : document.at("predicates")) { + auto predicate = parse_predicate(parser, item, false); + if (predicate->revision != 0) invalid("snapshot predicate revisions must be zero"); + if (!filters.revisions_.emplace(predicate->id, 0).second) invalid("duplicate predicate ID"); + if (!predicate->advisory) filters.required_ids_.insert(predicate->id); + filters.specs_.push_back(std::move(predicate)); } for (const auto& spec : filters.specs_) flatten(spec, filters.filters_); return filters; } -// The values a spec refers to: inline for most kinds, from the side batches -// for `join_keys`. -std::shared_ptr PushdownFilters::values_for(const Spec& spec) const { - if (spec.kind == "join_keys") { - auto it = join_keys_.find(spec.keys_column.empty() ? spec.column_name : spec.keys_column); - return it == join_keys_.end() ? nullptr : it->second; +void PushdownFilters::apply_delta(const std::string& ipc_bytes) { + if (ipc_bytes.empty()) invalid("dynamic filter delta is empty"); + if (ipc_bytes.size() > kMaxPayloadBytes) invalid("filter payload exceeds 16 MiB"); + auto batch = wire::decode_ipc(ipc_bytes); + const auto context = validate_batch(batch); + if (context != evaluation_context_) invalid("evaluation context changed within one scan"); + auto document = document_for(batch); + validate_header(document, "delta", "updates"); + PushdownFilters next = *this; + Parser parser{batch, next.join_keys_, next.output_schema_}; + std::unordered_set seen; + for (const auto& update : document.at("updates")) { + if (!update.is_object()) invalid("delta update must be an object"); + if (!update.contains("operation") || !update.contains("id") || + !update.contains("revision")) { + invalid("delta update is missing required properties"); + } + const auto operation = required_string(update, "operation", "update"); + const auto id = required_string(update, "id", "update"); + const auto revision = required_uint(update, "revision", "update"); + if (id.size() > 128) invalid("predicate ID exceeds 128 bytes"); + if (!seen.insert(id).second) invalid("duplicate delta predicate ID"); + if (required_ids_.count(id)) invalid("delta targets required predicate"); + if (operation == "remove") { + require_keys(update, {"operation", "id", "revision"}, {}, "remove update"); + } else if (operation == "upsert") { + require_keys(update, {"operation", "id", "revision", "mode", "source", "expression"}, + {}, "upsert update"); + } else + invalid("delta operation must be remove or upsert"); + const auto old = next.revisions_.find(id); + if (old != next.revisions_.end() && revision <= old->second) continue; + next.specs_.erase(std::remove_if(next.specs_.begin(), next.specs_.end(), + [&](const auto& spec) { return spec->id == id; }), + next.specs_.end()); + if (operation == "upsert") next.specs_.push_back(parse_predicate(parser, update, true)); + next.revisions_[id] = revision; } - if (!spec.value_ref || *spec.value_ref >= values_.size()) return nullptr; - return values_[*spec.value_ref]; + if (next.revisions_.size() > kMaxPredicateIds) invalid("delta exceeds predicate-ID limit"); + next.filters_.clear(); + for (const auto& spec : next.specs_) flatten(spec, next.filters_); + *this = std::move(next); +} + +std::shared_ptr PushdownFilters::values_for(const Spec& spec) const { + return spec.value; } std::vector PushdownFilters::column_filters(const std::string& column) const { - std::vector found; - for (const auto& filter : filters_) { - if (filter.column_name == column) found.push_back(filter); - } - return found; + std::vector result; + for (const auto& filter : filters_) + if (filter.column_name == column) result.push_back(filter); + return result; } bool PushdownFilters::has_filter_for_column(const std::string& column) const { - return std::any_of(filters_.begin(), filters_.end(), - [&](const Filter& filter) { return filter.column_name == column; }); + return std::any_of(specs_.begin(), specs_.end(), + [&](const auto& spec) { return mentions(spec, column); }); } std::vector PushdownFilters::filtered_columns() const { - std::vector columns; - for (const auto& filter : filters_) { - // An `and`/`or` node is flattened away, but a leaf with no column — - // there is no such thing on the wire today — would name the empty - // string and read as a column. - if (!filter.column_name.empty()) columns.push_back(filter.column_name); - } - std::sort(columns.begin(), columns.end()); - columns.erase(std::unique(columns.begin(), columns.end()), columns.end()); - return columns; -} - -// The widest interval every predicate on `column` agrees a value could be in. -// -// Deliberately loosening rather than tightening: bounds from separate branches -// are combined with min/max, so an OR widens and an AND does not narrow. A -// scan that trusts these emits a superset, and the engine re-checks every -// predicate — where a tightened bound would drop rows the query asked for. -ColumnBounds PushdownFilters::column_bounds(const std::string& column) const { - ColumnBounds bounds; - std::vector> stack = specs_; - while (!stack.empty()) { - auto spec = stack.back(); - stack.pop_back(); - if (spec->kind == "and" || spec->kind == "or") { - stack.insert(stack.end(), spec->children.begin(), spec->children.end()); - continue; - } - if (spec->kind != "constant" || spec->column_name != column) continue; - auto value = as_int64(values_for(*spec)); - if (!value) continue; + std::vector result; + for (const auto& spec : specs_) collect_columns(spec, result); + std::sort(result.begin(), result.end()); + result.erase(std::unique(result.begin(), result.end()), result.end()); + return result; +} - const auto& op = spec->op; - if (op == "gt" || op == "ge" || op == "gteq" || op == ">" || op == ">=") { - bounds.min = bounds.min ? std::min(*bounds.min, *value) : *value; - } else if (op == "lt" || op == "le" || op == "lteq" || op == "<" || op == "<=") { - bounds.max = bounds.max ? std::max(*bounds.max, *value) : *value; - } else { - // Equality pins both ends — but only ever outward, or `n = 5 OR - // n = 9` would come back as the single point whichever branch the - // stack happened to pop last. - bounds.min = bounds.min ? std::min(*bounds.min, *value) : *value; - bounds.max = bounds.max ? std::max(*bounds.max, *value) : *value; - } +ColumnBounds PushdownFilters::column_bounds(const std::string& column) const { + std::optional result; + for (const auto& spec : specs_) { + auto current = bounds_for(spec, column); + if (current) result = result ? intersect(*result, *current) : current; } - return bounds; + return result ? ColumnBounds{result->min, result->max} : ColumnBounds{}; } std::vector> PushdownFilters::column_specs(const std::string& column) const { - std::vector> found; - for (const auto& spec : specs_) { - if (spec->column_name != column) continue; - if (spec->kind == "and") { - for (const auto& child : spec->children) { - if (child->column_name == column) found.push_back(child); - } - } else { - found.push_back(spec); - } - } - return found; + std::vector> result; + for (const auto& spec : specs_) + if (mentions(spec, column)) result.push_back(spec); + return result; } std::shared_ptr PushdownFilters::or_column_values(const Spec& spec, const std::string& column) const { - arrow::ArrayVector branches; - for (const auto& child : spec.children) { - // A branch constraining a different column — or none — leaves this - // column free to take any value within that branch. - if (child->column_name != column) return nullptr; - const bool discrete = child->kind == "in" || child->kind == "join_keys" || - (child->kind == "constant" && is_equality(child->op)); - if (!discrete) return nullptr; - auto values = values_for(*child); - if (!values || values->length() == 0) return nullptr; - branches.push_back(child->kind == "constant" ? values->Slice(0, 1) : values); - } - if (branches.empty()) return nullptr; - - auto combined = arrow::Concatenate(branches); - if (!combined.ok()) return nullptr; - // Arrow's `unique` keeps first-appearance order, which is what makes the - // rendered set stable enough for a test to compare. - auto deduped = arrow::compute::Unique(combined.MoveValueUnsafe()); - if (!deduped.ok()) return nullptr; - return deduped.MoveValueUnsafe(); + return discrete_values(std::make_shared(spec), column); } std::shared_ptr PushdownFilters::column_values(const std::string& column) const { - for (const auto& spec : column_specs(column)) { - if (spec->kind == "constant" && is_equality(spec->op)) { - auto values = values_for(*spec); - // Sliced rather than returned whole: the values column holds one - // entry per referenced constant, not one per filter. - if (values && values->length() > 0) return values->Slice(0, 1); - } else if (spec->kind == "in" || spec->kind == "join_keys") { - if (auto values = values_for(*spec)) return values; - } else if (spec->kind == "or") { - if (auto values = or_column_values(*spec, column)) return values; - } - } + for (const auto& spec : specs_) + if (auto values = discrete_values(spec, column)) return values; return nullptr; } -namespace { - -const char* op_symbol(const std::string& op) { - if (op == "eq") return "="; - if (op == "ne") return "!="; - if (op == "lt") return "<"; - if (op == "le") return "<="; - if (op == "gt") return ">"; - if (op == "ge") return ">="; - return "?"; -} - -// Rendered as the Python fixtures render it: strings single-quoted, booleans -// `True`/`False`, nulls `NULL`, everything else via its own display. The tests -// compare this text, so the spelling is the contract. -std::string format_scalar(const std::shared_ptr& array, int64_t i) { - if (!array || i >= array->length() || array->IsNull(i)) return "NULL"; - switch (array->type()->id()) { - case arrow::Type::STRING: - return "'" + static_cast(*array).GetString(i) + "'"; - case arrow::Type::LARGE_STRING: - return "'" + static_cast(*array).GetString(i) + "'"; - case arrow::Type::BOOL: - return static_cast(*array).Value(i) ? "True" : "False"; - default: break; - } - auto casted = arrow::compute::Cast(*array->Slice(i, 1), arrow::utf8()); - if (!casted.ok()) return {}; - return std::static_pointer_cast(casted.MoveValueUnsafe())->GetString(0); -} - -} // namespace - std::string PushdownFilters::format() const { if (specs_.empty()) return "(none)"; - - // Bound to the member so the recursion can reach `values_`; a free - // function would have to take them as a parameter at every level. - std::function&, const std::string&)> render = - [&](const std::shared_ptr& spec, const std::string& column) -> std::string { - const std::string& name = column.empty() ? spec->column_name : column; - auto value = [&]() -> std::shared_ptr { return values_for(*spec); }; - - if (spec->kind == "is_null") return name + " IS NULL"; - if (spec->kind == "is_not_null") return name + " IS NOT NULL"; - if (spec->kind == "constant") { - return name + " " + op_symbol(spec->op.empty() ? "eq" : spec->op) + " " + - format_scalar(value(), 0); - } - if (spec->kind == "in" || spec->kind == "join_keys") { - auto values = value(); - if (!values) return name + " IN ()"; - // A long IN list is collapsed: the point is that a filter arrived, - // and printing thousands of values makes the output unreadable and - // the test unwriteable. - if (values->length() > 20) { - return name + " IN (" + std::to_string(values->length()) + " values)"; - } - std::string items; - for (int64_t i = 0; i < values->length(); ++i) { - if (i) items += ", "; - items += format_scalar(values, i); - } - return name + " IN (" + items + ")"; - } - if (spec->kind == "and" || spec->kind == "or") { - const std::string joiner = spec->kind == "and" ? " AND " : " OR "; - std::string parts; - for (size_t i = 0; i < spec->children.size(); ++i) { - if (i) parts += joiner; - parts += render(spec->children[i], ""); - } - // Parenthesized, so a nested group reads unambiguously and the - // text matches what the reference implementations emit. - return "(" + parts + ")"; - } - return spec->kind; - }; - - std::string out; - for (size_t i = 0; i < specs_.size(); ++i) { - if (i) out += " AND "; - out += render(specs_[i], ""); + std::string result; + for (const auto& spec : specs_) { + if (!result.empty()) result += " AND "; + result += render(spec); } - return out.empty() ? "(none)" : out; + return result; } std::string PushdownFilters::format_repr() const { if (specs_.empty()) return "(none)"; - - // Bound to the member for the same reason `format` is: the recursion has - // to reach `values_`. - std::function&, const std::string&)> render = - [&](const std::shared_ptr& spec, const std::string& column) -> std::string { - const std::string& name = column.empty() ? spec->column_name : column; - - if (spec->kind == "is_null") return "IsNullFilter(" + name + " IS NULL)"; - if (spec->kind == "is_not_null") return "IsNotNullFilter(" + name + " IS NOT NULL)"; - if (spec->kind == "constant") { - return "ConstantFilter(" + name + " " + op_symbol(spec->op.empty() ? "eq" : spec->op) + - " " + format_scalar(values_for(*spec), 0) + ")"; - } - if (spec->kind == "in" || spec->kind == "join_keys") { - // Every value, unlike `format`: this rendering names the kinds, - // and a fixture asserting on it wants the whole set. - auto values = values_for(*spec); - std::string items; - for (int64_t i = 0; values && i < values->length(); ++i) { - if (i) items += ", "; - items += format_scalar(values, i); - } - return "InFilter(" + name + " IN [" + items + "])"; - } - if (spec->kind == "and" || spec->kind == "or") { - const std::string label = spec->kind == "and" ? "AndFilter(" : "OrFilter("; - const std::string joiner = spec->kind == "and" ? " AND " : " OR "; - std::string parts; - for (size_t i = 0; i < spec->children.size(); ++i) { - if (i) parts += joiner; - parts += render(spec->children[i], ""); - } - return label + parts + ")"; - } - return spec->kind; - }; - - std::string parts; + std::string result = "PushdownFilters(["; for (size_t i = 0; i < specs_.size(); ++i) { - if (i) parts += ", "; - parts += render(specs_[i], ""); + if (i) result += ", "; + result += render_repr(specs_[i]); } - return "PushdownFilters([" + parts + "])"; + return result + "])"; } std::shared_ptr PushdownFilters::apply( const std::shared_ptr& batch) const { if (!batch || specs_.empty()) return batch; - auto surviving = batch; - std::vector> stack = specs_; - while (!stack.empty()) { - auto spec = stack.back(); - stack.pop_back(); - - // Only conjunctions are decomposed. An `or` cannot be applied one - // branch at a time without dropping rows the whole disjunction keeps, - // so it is left to the engine. - if (spec->kind == "and") { - stack.insert(stack.end(), spec->children.begin(), spec->children.end()); - continue; - } - - auto column = surviving->GetColumnByName(spec->column_name); - if (!column) continue; - - // A dictionary column is decoded before comparing. DuckDB types a - // dictionary<*, utf8> column without ENUM metadata as plain VARCHAR - // and pushes a string literal, and every comparison kernel rejects the - // (dictionary, string) pair; casting the literal *up* to the - // dictionary type is what throws. - if (column->type_id() == arrow::Type::DICTIONARY) { - const auto& value_type = - static_cast(*column->type()).value_type(); - auto decoded = arrow::compute::Cast(*column, value_type); - if (!decoded.ok()) continue; - column = *decoded; - } - - arrow::Result mask; - if (spec->kind == "is_null") { - mask = arrow::compute::IsNull(column); - } else if (spec->kind == "is_not_null") { - mask = arrow::compute::IsValid(column); - } else if (spec->kind == "constant" || spec->kind == "in" || spec->kind == "join_keys") { - auto value = values_for(*spec); - if (!value) continue; - if (spec->kind != "constant") { - arrow::compute::SetLookupOptions options(value); - mask = arrow::compute::CallFunction("is_in", {column}, &options); - } else { - const char* kernel = comparison_kernel(spec->op.empty() ? "eq" : spec->op); - if (!kernel) continue; - // The constant is one element; comparing needs it as a scalar - // so Arrow broadcasts rather than requiring equal lengths. - if (value->length() == 0) continue; - auto scalar = value->GetScalar(0); - if (!scalar.ok()) continue; - mask = arrow::compute::CallFunction(kernel, {column, scalar.MoveValueUnsafe()}); - } - } else { - continue; + for (const auto& predicate : specs_) { + if (predicate->kind == "runtime_filter") continue; + try { + auto mask = array_from_datum(evaluate(predicate, surviving), surviving->num_rows()); + if (mask->type_id() != arrow::Type::BOOL) + throw std::runtime_error("predicate did not evaluate to BOOLEAN"); + arrow::compute::FilterOptions options( + arrow::compute::FilterOptions::NullSelectionBehavior::DROP); + auto filtered = arrow::compute::Filter(surviving, mask, options); + if (!filtered.ok()) throw std::runtime_error(filtered.status().ToString()); + surviving = filtered.MoveValueUnsafe().record_batch(); + } catch (const std::exception&) { + if (predicate->advisory) continue; + throw; } - if (!mask.ok()) continue; - - // A null result is neither true nor false; SQL drops those rows, and - // Filter's default emits them, so nulls are made explicit false first. - // DROP, explicitly. A null comparison result is neither true nor - // false and SQL excludes those rows; `EMIT_NULL` would keep them, - // which is a wrong answer rather than a slow one. Named rather than - // left to the default so the choice survives an Arrow upgrade. - arrow::compute::FilterOptions options( - arrow::compute::FilterOptions::NullSelectionBehavior::DROP); - auto filtered = arrow::compute::Filter(surviving, mask.MoveValueUnsafe(), options); - if (!filtered.ok()) continue; - if (auto next = filtered.MoveValueUnsafe().record_batch()) surviving = next; } return surviving; } diff --git a/tests/function_test.cpp b/tests/function_test.cpp index ee603b9..7e1e3f4 100644 --- a/tests/function_test.cpp +++ b/tests/function_test.cpp @@ -3,10 +3,12 @@ #include #include +#include #include "vgi/function.h" #include "vgi/catalog.h" #include "vgi/generated/vgi_protocol_schemas.hpp" +#include "vgi/pushdown.h" #include "wire.h" namespace { @@ -32,6 +34,33 @@ class Dynamic : public Fixed { vgi::FunctionMetadata metadata() const override { return {}; } }; +std::string filter_batch(const std::string& document, + const std::shared_ptr& payload = nullptr) { + arrow::StringBuilder spec_builder; + REQUIRE(spec_builder.Append(document).ok()); + std::shared_ptr spec; + REQUIRE(spec_builder.Finish(&spec).ok()); + arrow::FieldVector fields{arrow::field("filter_spec", arrow::utf8(), false)}; + arrow::ArrayVector arrays{spec}; + if (payload) { + fields.push_back(arrow::field("value_0", payload->type(), true)); + arrays.push_back(payload); + } + auto metadata = arrow::key_value_metadata( + {"vgi_filter_encoding", "vgi_filter_version", "vgi_evaluation_context"}, + {"vgi.filters.v2", "2", "vgi.none.v1"}); + return vgi::wire::encode_ipc( + arrow::RecordBatch::Make(arrow::schema(fields, metadata), 1, arrays)); +} + +std::shared_ptr int64_values(std::initializer_list values) { + arrow::Int64Builder builder; + for (const auto value : values) REQUIRE(builder.Append(value).ok()); + std::shared_ptr result; + REQUIRE(builder.Finish(&result).ok()); + return result; +} + } // namespace TEST_CASE("a fixed return type binds without an override", "[function]") { @@ -127,3 +156,32 @@ TEST_CASE("wire argument names preserve unnamed varargs", "[wire]") { REQUIRE_FALSE(decoded->at(1).has_value()); REQUIRE(decoded->at(2) == "scale"); } + +TEST_CASE("Filter v2 snapshot binds and applies typed comparison payloads", "[filter-v2]") { + const auto document = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[{"id":"p","revision":0,"mode":"required","source":"query","expression":{"node":"comparison","op":"ge","left":{"node":"column_ref","column_index":0,"column_name":"n"},"right":{"node":"literal","value_ref":0}}}]})"; + auto filters = vgi::PushdownFilters::parse(filter_batch(document, int64_values({3})), {}, + arrow::schema({arrow::field("n", arrow::int64())})); + REQUIRE(filters.format() == "n >= 3"); + REQUIRE(filters.filtered_columns() == std::vector{"n"}); + REQUIRE(filters.column_bounds("n").min == 3); + + auto input = arrow::RecordBatch::Make(arrow::schema({arrow::field("n", arrow::int64())}), 5, + {int64_values({1, 2, 3, 4, 5})}); + const auto output = filters.apply(input); + REQUIRE(output->num_rows() == 3); +} + +TEST_CASE("Filter v2 dynamic deltas update snapshot state atomically", "[filter-v2]") { + const auto snapshot = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[]})"; + auto filters = vgi::PushdownFilters::parse(filter_batch(snapshot), {}, + arrow::schema({arrow::field("n", arrow::int64())})); + const auto delta = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"delta","updates":[{"operation":"upsert","id":"topn","revision":1,"mode":"advisory","source":"top_n","expression":{"node":"comparison","op":"lt","left":{"node":"column_ref","column_index":0,"column_name":"n"},"right":{"node":"literal","value_ref":0}}}]})"; + filters.apply_delta(filter_batch(delta, int64_values({4}))); + REQUIRE(filters.format_repr() == "PushdownFilters([ConstantFilter(n < 4)])"); + auto input = arrow::RecordBatch::Make(arrow::schema({arrow::field("n", arrow::int64())}), 5, + {int64_values({1, 2, 3, 4, 5})}); + REQUIRE(filters.apply(input)->num_rows() == 3); +} From 13e4fb41f06f7b29819a536d62e4486504f4f444 Mon Sep 17 00:00:00 2001 From: Rusty Conover Date: Thu, 10 Sep 2026 21:06:58 -0400 Subject: [PATCH 2/5] Add Filter v2 capability metadata --- include/vgi/types.h | 43 +++++++++++++++++++++++++++ src/catalog.cpp | 57 ++++++++++++++++++++++++++++++++++++ src/function.cpp | 15 ++++++++++ src/wire.cpp | 65 +++++++++++++++++++++++++++++++++++++++++ src/wire.h | 8 +++++ tests/function_test.cpp | 62 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 250 insertions(+) diff --git a/include/vgi/types.h b/include/vgi/types.h index 6186fb0..34a0fc7 100644 --- a/include/vgi/types.h +++ b/include/vgi/types.h @@ -1,6 +1,7 @@ // © Copyright 2025, 2026 Query Farm LLC - https://query.farm #pragma once +#include #include #include #include @@ -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 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 filter_semantic_profiles; + // Capability-gated extensions to the standard profile. The C++ SDK + // currently rejects non-empty lists until matching evaluators exist. + std::vector additional_filter_functions; + std::vector runtime_filter_algorithms; + std::vector 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 required_settings; + + // Resolve the wire advertisement, applying the SDK default and rejecting + // semantic profiles for which this SDK has no evaluator. + std::vector resolved_filter_semantic_profiles() const; }; } // namespace vgi diff --git a/src/catalog.cpp b/src/catalog.cpp index a16e23f..9522b86 100644 --- a/src/catalog.cpp +++ b/src/catalog.cpp @@ -220,6 +220,54 @@ const char* partition_kind_wire_value(const FunctionMetadata& metadata) { : metadata.partition_kind.c_str(); } +std::vector> filter_function_identities_of( + const FunctionMetadata& metadata) { + std::vector> identities; + identities.reserve(metadata.additional_filter_functions.size()); + for (const auto& capability : metadata.additional_filter_functions) { + identities.emplace_back(capability.namespace_name, capability.name, capability.version); + } + return identities; +} + +std::vector> runtime_filter_identities_of( + const FunctionMetadata& metadata) { + std::vector> identities; + identities.reserve(metadata.runtime_filter_algorithms.size()); + for (const auto& capability : metadata.runtime_filter_algorithms) { + identities.emplace_back(capability.namespace_name, capability.name, capability.version); + } + return identities; +} + +std::vector>> evaluation_contexts_of( + const FunctionMetadata& metadata) { + std::vector>> contexts; + contexts.reserve(metadata.filter_evaluation_contexts.size()); + for (const auto& capability : metadata.filter_evaluation_contexts) { + contexts.emplace_back(capability.profile, capability.provider_fingerprint); + } + return contexts; +} + +void validate_filter_capabilities(const FunctionMetadata& metadata) { + // Advertising a capability is a correctness claim: the producer may emit + // it in a required predicate and remove its local copy. Keep these lists + // empty until this SDK registers the corresponding evaluator. + if (!metadata.additional_filter_functions.empty()) { + throw std::invalid_argument( + "C++ SDK cannot advertise an extension filter function without an evaluator"); + } + if (!metadata.runtime_filter_algorithms.empty()) { + throw std::invalid_argument( + "C++ SDK has no registered runtime-filter artifact evaluator to advertise"); + } + if (!metadata.filter_evaluation_contexts.empty()) { + throw std::invalid_argument( + "C++ SDK has no isolated DuckDB session-context evaluator to advertise"); + } +} + // Wrap a payload batch in the `{result: binary}` envelope every non-void // method answers with. The engine unwraps it and validates the inner schema // against its own generated copy, so a drifted payload is caught there rather @@ -1070,6 +1118,8 @@ wire::ResultBuilder Dispatcher::common_function_info( const std::string& name, const SchemaPath& schema_path, const char* function_type, const std::vector& specs, const std::shared_ptr& output_schema, const FunctionMetadata& metadata) { + validate_filter_capabilities(metadata); + const auto filter_semantic_profiles = metadata.resolved_filter_semantic_profiles(); auto builder = wire::ResultBuilder(gen::FunctionInfoSchema()) .set_string("name", name) @@ -1089,6 +1139,13 @@ wire::ResultBuilder Dispatcher::common_function_info( .set_secret_lookups("required_secrets", secret_entries(metadata)) .set_bool("projection_pushdown", metadata.projection_pushdown) .set_bool("filter_pushdown", metadata.filter_pushdown) + .set_string_list("filter_semantic_profiles", filter_semantic_profiles) + .set_filter_identities("additional_filter_functions", + filter_function_identities_of(metadata)) + .set_filter_identities("runtime_filter_algorithms", + runtime_filter_identities_of(metadata)) + .set_evaluation_contexts("filter_evaluation_contexts", evaluation_contexts_of(metadata)) + .set_bool("filters_exactly_applied", metadata.filters_exactly_applied) .set_bool("sampling_pushdown", metadata.sampling_pushdown) .set_bool("input_from_args", metadata.input_from_args) .set_enum("partition_kind", partition_kind_wire_value(metadata)) diff --git a/src/function.cpp b/src/function.cpp index 132d990..b07217e 100644 --- a/src/function.cpp +++ b/src/function.cpp @@ -7,6 +7,21 @@ namespace vgi { +std::vector FunctionMetadata::resolved_filter_semantic_profiles() const { + auto profiles = filter_semantic_profiles; + if (filter_pushdown && profiles.empty()) { + profiles.emplace_back(filter_semantic_profiles::kDuckDBStandardV1); + } + for (const auto& profile : profiles) { + if (profile != filter_semantic_profiles::kDuckDBStandardV1) { + throw std::invalid_argument("C++ SDK supports only " + + std::string(filter_semantic_profiles::kDuckDBStandardV1) + + " filter semantics"); + } + } + return profiles; +} + ArgSpec ArgSpec::column(std::string name, int index, std::string type, std::string description) { ArgSpec s; s.name = std::move(name); diff --git a/src/wire.cpp b/src/wire.cpp index 8a3e576..9628f0c 100644 --- a/src/wire.cpp +++ b/src/wire.cpp @@ -544,6 +544,15 @@ arrow::StringBuilder* string_child(arrow::StructBuilder& entry, const char* name return builder; } +arrow::UInt64Builder* uint64_child(arrow::StructBuilder& entry, const char* name) { + const auto type = entry.type(); + const int index = static_cast(*type).GetFieldIndex(name); + if (index < 0) fail(std::string("struct has no '") + name + "' field"); + auto* builder = dynamic_cast(entry.field_builder(index)); + if (!builder) fail(std::string("struct field '") + name + "' is not uint64"); + return builder; +} + ResultBuilder& ResultBuilder::set_secret_lookups( const std::string& field, const std::vector>& lookups) { @@ -607,6 +616,62 @@ ResultBuilder& ResultBuilder::set_examples(const std::string& field, return *this; } +ResultBuilder& ResultBuilder::set_filter_identities( + const std::string& field, + const std::vector>& identities) { + const int index = field_index(field); + std::unique_ptr raw; + check_ok(arrow::MakeBuilder(arrow::default_memory_pool(), schema_->field(index)->type(), &raw), + "building filter capability field '" + field + "'"); + auto* list = dynamic_cast(raw.get()); + if (!list) fail("result field '" + field + "' is not a list"); + auto* entry = dynamic_cast(list->value_builder()); + if (!entry) fail("result field '" + field + "' is not a list of structs"); + + auto* namespace_builder = string_child(*entry, "namespace"); + auto* name_builder = string_child(*entry, "name"); + auto* version_builder = uint64_child(*entry, "version"); + + check_ok(list->Append(), "opening filter capability field '" + field + "'"); + for (const auto& [namespace_name, name, version] : identities) { + check_ok(entry->Append(), "opening a filter capability"); + check_ok(namespace_builder->Append(namespace_name), "appending filter namespace"); + check_ok(name_builder->Append(name), "appending filter name"); + check_ok(version_builder->Append(version), "appending filter version"); + } + arrays_[static_cast(index)] = + unwrap(list->Finish(), "finishing filter capability field '" + field + "'"); + return *this; +} + +ResultBuilder& ResultBuilder::set_evaluation_contexts( + const std::string& field, + const std::vector>>& contexts) { + const int index = field_index(field); + std::unique_ptr raw; + check_ok(arrow::MakeBuilder(arrow::default_memory_pool(), schema_->field(index)->type(), &raw), + "building evaluation-context field '" + field + "'"); + auto* list = dynamic_cast(raw.get()); + if (!list) fail("result field '" + field + "' is not a list"); + auto* entry = dynamic_cast(list->value_builder()); + if (!entry) fail("result field '" + field + "' is not a list of structs"); + + auto* profile_builder = string_child(*entry, "profile"); + auto* fingerprint_builder = string_child(*entry, "provider_fingerprint"); + + check_ok(list->Append(), "opening evaluation-context field '" + field + "'"); + for (const auto& [profile, fingerprint] : contexts) { + check_ok(entry->Append(), "opening an evaluation context"); + check_ok(profile_builder->Append(profile), "appending evaluation-context profile"); + check_ok(fingerprint ? fingerprint_builder->Append(*fingerprint) + : fingerprint_builder->AppendNull(), + "appending evaluation-context fingerprint"); + } + arrays_[static_cast(index)] = + unwrap(list->Finish(), "finishing evaluation-context field '" + field + "'"); + return *this; +} + ResultBuilder& ResultBuilder::set_int64_map( const std::string& field, const std::vector>& entries) { auto key_builder = std::make_shared(); diff --git a/src/wire.h b/src/wire.h index d1ef391..c6772f1 100644 --- a/src/wire.h +++ b/src/wire.h @@ -202,6 +202,14 @@ class ResultBuilder { // The `examples` column: a list of {sql, description, expected_output}. ResultBuilder& set_examples(const std::string& field, const std::vector& examples); + // A list<{namespace: utf8, name: utf8, version: uint64}> capability field. + ResultBuilder& set_filter_identities( + const std::string& field, + const std::vector>& identities); + // A list<{profile: utf8, provider_fingerprint: utf8?}> capability field. + ResultBuilder& set_evaluation_contexts( + const std::string& field, + const std::vector>>& contexts); // A map column. Arrow spells map entries key/value (not // keys/values), which is what the canonical Python protocol emits. ResultBuilder& set_int64_map(const std::string& field, diff --git a/tests/function_test.cpp b/tests/function_test.cpp index 7e1e3f4..410924b 100644 --- a/tests/function_test.cpp +++ b/tests/function_test.cpp @@ -122,11 +122,73 @@ TEST_CASE("protocol v2 named schemas use path lists", "[protocol]") { const auto function_info = vgi::generated::FunctionInfoSchema(); REQUIRE(function_info->GetFieldByName("parameter_default_values")->nullable()); + REQUIRE(function_info->GetFieldByName("filter_semantic_profiles")->type()->id() == + arrow::Type::LIST); + REQUIRE(function_info->GetFieldByName("additional_filter_functions")->type()->id() == + arrow::Type::LIST); + REQUIRE(function_info->GetFieldByName("runtime_filter_algorithms")->type()->id() == + arrow::Type::LIST); + REQUIRE(function_info->GetFieldByName("filter_evaluation_contexts")->type()->id() == + arrow::Type::LIST); + REQUIRE(function_info->GetFieldByName("filters_exactly_applied")->type()->id() == + arrow::Type::BOOL); REQUIRE(vgi::generated::BindRequestSchema()->GetFieldByName("argument_names")->nullable()); REQUIRE( vgi::generated::AggregateBindRequestSchema()->GetFieldByName("argument_names")->nullable()); } +TEST_CASE("filter pushdown advertises only the implemented semantic profile", "[protocol]") { + vgi::FunctionMetadata plain; + REQUIRE(plain.resolved_filter_semantic_profiles().empty()); + + vgi::FunctionMetadata filtering; + filtering.filter_pushdown = true; + REQUIRE(filtering.resolved_filter_semantic_profiles() == + std::vector{vgi::filter_semantic_profiles::kDuckDBStandardV1}); + filtering.auto_apply_filters = true; + REQUIRE_FALSE(filtering.filters_exactly_applied); + + filtering.filter_semantic_profiles = {"vgi.duckdb.standard.v2"}; + REQUIRE_THROWS(filtering.resolved_filter_semantic_profiles()); +} + +TEST_CASE("filter capability structs preserve generated FunctionInfo shape", "[protocol]") { + const auto batch = vgi::wire::ResultBuilder(vgi::generated::FunctionInfoSchema()) + .set_string_list("filter_semantic_profiles", {"vgi.duckdb.standard.v1"}) + .set_filter_identities("additional_filter_functions", + {{"duckdb.spatial", "intersects_extent", 1}}) + .set_filter_identities("runtime_filter_algorithms", + {{"duckdb.runtime_filter", "bloom", 2}}) + .set_evaluation_contexts("filter_evaluation_contexts", + {{"vgi.duckdb.session.v1", "duckdb-icu:test"}, + {"vgi.duckdb.session.v1", std::nullopt}}) + .fill_defaults() + .finish(); + + const auto identities = std::static_pointer_cast( + batch->GetColumnByName("additional_filter_functions")); + REQUIRE(identities->value_length(0) == 1); + const auto identity = std::static_pointer_cast(identities->values()); + const auto namespaces = + std::static_pointer_cast(identity->GetFieldByName("namespace")); + const auto names = + std::static_pointer_cast(identity->GetFieldByName("name")); + const auto versions = + std::static_pointer_cast(identity->GetFieldByName("version")); + REQUIRE(namespaces->GetString(0) == "duckdb.spatial"); + REQUIRE(names->GetString(0) == "intersects_extent"); + REQUIRE(versions->Value(0) == 1); + + const auto contexts = std::static_pointer_cast( + batch->GetColumnByName("filter_evaluation_contexts")); + REQUIRE(contexts->value_length(0) == 2); + const auto context = std::static_pointer_cast(contexts->values()); + const auto fingerprints = std::static_pointer_cast( + context->GetFieldByName("provider_fingerprint")); + REQUIRE(fingerprints->GetString(0) == "duckdb-icu:test"); + REQUIRE(fingerprints->IsNull(1)); +} + TEST_CASE("wire schema paths round trip every component", "[wire]") { const auto schema = arrow::schema( {arrow::field("schema_path", arrow::list(arrow::utf8()), /*nullable=*/false)}); From 7418e8c417e96c64291b423ab49e76569480b466 Mon Sep 17 00:00:00 2001 From: Rusty Conover Date: Thu, 10 Sep 2026 21:38:26 -0400 Subject: [PATCH 3/5] Harden Filter v2 evaluation semantics --- src/pushdown.cpp | 394 ++++++++++++++++++++++++++++++++++++---- tests/function_test.cpp | 180 +++++++++++++++++- 2 files changed, 536 insertions(+), 38 deletions(-) diff --git a/src/pushdown.cpp b/src/pushdown.cpp index 3118a60..a6eb342 100644 --- a/src/pushdown.cpp +++ b/src/pushdown.cpp @@ -2,10 +2,12 @@ #include "vgi/pushdown.h" #include +#include #include #include #include #include +#include #include #include @@ -147,12 +149,53 @@ std::string validate_batch(const std::shared_ptr& batch) { return context; } +void validate_arrow_extensions(const std::shared_ptr& field) { + static const std::unordered_set known = { + "arrow.bool8", + "arrow.json", + "arrow.uuid", + "geoarrow.linestring", + "geoarrow.multilinestring", + "geoarrow.multipoint", + "geoarrow.multipolygon", + "geoarrow.point", + "geoarrow.polygon", + "geoarrow.wkb", + }; + if (field->metadata() && field->metadata()->Contains("ARROW:extension:name")) { + const auto name = field->metadata()->Get("ARROW:extension:name"); + if (!name.ok() || !known.count(name.ValueUnsafe())) + invalid("unknown Arrow extension type on field '" + field->name() + "'"); + } + if (field->type()->id() == arrow::Type::EXTENSION) { + const auto& extension = static_cast(*field->type()); + if (!known.count(extension.extension_name())) + invalid("unknown Arrow extension type '" + extension.extension_name() + "'"); + } + if (field->type()->id() == arrow::Type::STRUCT) { + for (const auto& child : static_cast(*field->type()).fields()) + validate_arrow_extensions(child); + } else if (field->type()->id() == arrow::Type::LIST) { + validate_arrow_extensions( + static_cast(*field->type()).value_field()); + } else if (field->type()->id() == arrow::Type::LARGE_LIST) { + validate_arrow_extensions( + static_cast(*field->type()).value_field()); + } else if (field->type()->id() == arrow::Type::FIXED_SIZE_LIST) { + validate_arrow_extensions( + static_cast(*field->type()).value_field()); + } +} + std::shared_ptr payload(const std::shared_ptr& batch, const std::string& prefix, uint64_t reference) { const auto name = prefix + "_" + std::to_string(reference); const auto indices = batch->schema()->GetAllFieldIndices(name); if (indices.size() != 1) invalid("missing or duplicate payload field '" + name + "'"); - return batch->column(indices.front()); + const auto position = indices.front(); + const auto& field = batch->schema()->field(position); + if (prefix != "artifact") validate_arrow_extensions(field); + return batch->column(position); } std::string root_column(const std::shared_ptr& expression) { @@ -163,6 +206,86 @@ std::string root_column(const std::shared_ptr& expression) { return {}; } +bool is_direct_column(const std::shared_ptr& expression, const std::string& column) { + return expression && expression->kind == "column" && expression->column_name == column; +} + +std::shared_ptr logical_type(std::shared_ptr type) { + while (type && type->id() == arrow::Type::DICTIONARY) { + type = static_cast(*type).value_type(); + } + return type; +} + +bool same_logical_type(const std::shared_ptr& left, + const std::shared_ptr& right) { + const auto left_type = logical_type(left); + const auto right_type = logical_type(right); + return left_type && right_type && left_type->Equals(right_type); +} + +bool boolean_type(const std::shared_ptr& type) { + const auto logical = logical_type(type); + return logical && logical->id() == arrow::Type::BOOL; +} + +bool numeric_type(const std::shared_ptr& type) { + const auto logical = logical_type(type); + if (!logical) return false; + switch (logical->id()) { + case arrow::Type::UINT8: + case arrow::Type::INT8: + case arrow::Type::UINT16: + case arrow::Type::INT16: + case arrow::Type::UINT32: + case arrow::Type::INT32: + case arrow::Type::UINT64: + case arrow::Type::INT64: + case arrow::Type::HALF_FLOAT: + case arrow::Type::FLOAT: + case arrow::Type::DOUBLE: + case arrow::Type::DECIMAL128: + case arrow::Type::DECIMAL256: return true; + default: return false; + } +} + +bool exact_context_free_cast(const std::shared_ptr& source, + const std::shared_ptr& target) { + if (!source || !target) return false; + if (source->Equals(target)) return true; + if (arrow::is_integer(source->id()) && arrow::is_integer(target->id())) return true; + return source->id() == arrow::Type::FLOAT && target->id() == arrow::Type::DOUBLE; +} + +bool contextual_type(const std::shared_ptr& type) { + const auto logical = logical_type(type); + if (!logical) return false; + switch (logical->id()) { + case arrow::Type::STRING: + case arrow::Type::LARGE_STRING: + case arrow::Type::DATE32: + case arrow::Type::DATE64: + case arrow::Type::TIME32: + case arrow::Type::TIME64: + case arrow::Type::TIMESTAMP: return true; + default: return false; + } +} + +void validate_identity(const json& identity, const std::string& where) { + require_keys(identity, {"namespace", "name", "version"}, {}, where); + const auto identity_namespace = required_string(identity, "namespace", where); + const auto identity_name = required_string(identity, "name", where); + const auto identity_version = required_uint(identity, "version", where); + static const std::regex namespace_pattern(R"([a-z][a-z0-9]*(?:\.[a-z][a-z0-9_]*)*)"); + static const std::regex name_pattern(R"([a-z][a-z0-9_]*)"); + if (!std::regex_match(identity_namespace, namespace_pattern) || + !std::regex_match(identity_name, name_pattern) || identity_version == 0) { + invalid(where + " has a noncanonical identity"); + } +} + struct Parser { std::shared_ptr batch; const std::vector>& join_keys; @@ -188,6 +311,7 @@ struct Parser { invalid("column_ref index is outside the authoritative output schema"); } const auto& field = output_schema->field(static_cast(result->column_index)); + validate_arrow_extensions(field); if (field->name() != result->column_name) invalid("column_ref name does not match authoritative index"); result->data_type = field->type(); @@ -209,6 +333,7 @@ struct Parser { fields[result->field_index]->name() != result->field_name) { invalid("field_ref name/index does not match authoritative struct"); } + validate_arrow_extensions(fields[result->field_index]); result->column_name = root_column(result->children[0]); result->data_type = fields[result->field_index]->type(); return result; @@ -230,6 +355,10 @@ struct Parser { invalid("unknown comparison operator '" + result->op + "'"); result->children = {expression(node.at("left"), depth + 1), expression(node.at("right"), depth + 1)}; + if (!same_logical_type(result->children[0]->data_type, + result->children[1]->data_type)) { + invalid("comparison operands have incompatible types"); + } result->column_name = root_column(result->children[0]); if (result->column_name.empty()) result->column_name = root_column(result->children[1]); if (result->children[1]->kind == "literal") result->value = result->children[1]->value; @@ -243,6 +372,9 @@ struct Parser { result->kind = kind; for (const auto& child : node.at("children")) result->children.push_back(expression(child, depth + 1)); + for (const auto& child : result->children) { + if (!boolean_type(child->data_type)) invalid(kind + " children must be BOOLEAN"); + } result->data_type = arrow::boolean(); return result; } @@ -250,6 +382,8 @@ struct Parser { require_keys(node, {"node", "expression"}, {}, "not"); result->kind = "not"; result->children.push_back(expression(node.at("expression"), depth + 1)); + if (!boolean_type(result->children[0]->data_type)) + invalid("not operand must be BOOLEAN"); result->data_type = arrow::boolean(); return result; } @@ -297,9 +431,12 @@ struct Parser { invalid("external IN batch/column index is unavailable"); if (join_keys[bi]->schema()->field(static_cast(ci))->name() != name) invalid("external IN column name does not match authoritative index"); + validate_arrow_extensions(join_keys[bi]->schema()->field(static_cast(ci))); result->value = join_keys[bi]->column(static_cast(ci)); } else invalid("in.set has an unknown kind"); + if (!same_logical_type(result->children[0]->data_type, result->value->type())) + invalid("IN expression and set have incompatible types"); result->data_type = arrow::boolean(); return result; } @@ -310,6 +447,12 @@ struct Parser { result->value = payload(batch, "type", required_uint(node, "type_ref", "cast")); if (!result->value->IsNull(0)) invalid("cast type payload must contain NULL"); result->data_type = result->value->type(); + if (contextual_type(result->children[0]->data_type) && + contextual_type(result->data_type)) { + invalid("context-dependent cast requires vgi.duckdb.session.v1"); + } + if (!exact_context_free_cast(result->children[0]->data_type, result->data_type)) + invalid("cast has no exact context-free standard-v1 evaluator"); return result; } if (kind == "arithmetic") { @@ -323,6 +466,11 @@ struct Parser { invalid("context-dependent arithmetic requires vgi.duckdb.session.v1"); result->children = {expression(node.at("left"), depth + 1), expression(node.at("right"), depth + 1)}; + if (!numeric_type(result->children[0]->data_type) || + !same_logical_type(result->children[0]->data_type, + result->children[1]->data_type)) { + invalid("arithmetic operands require one exact numeric type"); + } result->data_type = result->children[0]->data_type; return result; } @@ -330,13 +478,17 @@ struct Parser { require_keys(node, {"node", "expression"}, {}, "negate"); result->kind = "negate"; result->children.push_back(expression(node.at("expression"), depth + 1)); + if (!numeric_type(result->children[0]->data_type)) + invalid("negate operand must be numeric"); result->data_type = result->children[0]->data_type; return result; } if (kind == "call") { require_keys(node, {"node", "function", "arguments"}, {"options"}, "call"); - if (!node.at("function").is_string()) + if (!node.at("function").is_string()) { + validate_identity(node.at("function"), "call.function"); invalid("extension filter functions were not advertised"); + } result->function = node.at("function").get(); if (result->function != "starts_with" && result->function != "ends_with" && result->function != "contains" && result->function != "list_contains") @@ -348,6 +500,21 @@ struct Parser { result->kind = "call"; for (const auto& argument : node.at("arguments")) result->children.push_back(expression(argument, depth + 1)); + if (result->function == "starts_with" || result->function == "ends_with" || + result->function == "contains") { + for (const auto& argument : result->children) { + if (!argument->data_type || argument->data_type->id() != arrow::Type::STRING) + invalid("string function arguments must be UTF8"); + } + } else { + const auto list_type = result->children[0]->data_type; + if (!list_type || list_type->id() != arrow::Type::LIST) + invalid("list_contains first argument must be LIST"); + const auto element_type = + static_cast(*list_type).value_type(); + if (!same_logical_type(element_type, result->children[1]->data_type)) + invalid("list_contains element type mismatch"); + } result->data_type = arrow::boolean(); return result; } @@ -355,6 +522,22 @@ struct Parser { require_keys(node, {"node", "algorithm", "input", "artifact_ref", "null_handling"}, {}, "runtime_filter"); if (!root) invalid("runtime_filter may appear only at a predicate root"); + const auto& algorithm = node.at("algorithm"); + validate_identity(algorithm, "runtime_filter.algorithm"); + const auto algorithm_namespace = + required_string(algorithm, "namespace", "runtime_filter.algorithm"); + const auto algorithm_name = + required_string(algorithm, "name", "runtime_filter.algorithm"); + const auto algorithm_version = + required_uint(algorithm, "version", "runtime_filter.algorithm"); + if (algorithm_namespace != "duckdb.runtime_filter" || + (algorithm_name != "bloom" && algorithm_name != "prefix_range") || + algorithm_version != 1) { + invalid("unknown runtime-filter algorithm"); + } + const auto null_handling = required_string(node, "null_handling", "runtime_filter"); + if (null_handling != "pass" && null_handling != "reject") + invalid("runtime_filter.null_handling must be pass or reject"); result->kind = "runtime_filter"; result->children.push_back(expression(node.at("input"), depth + 1)); (void)payload(batch, "artifact", required_uint(node, "artifact_ref", "runtime_filter")); @@ -368,7 +551,22 @@ json document_for(const std::shared_ptr& batch) { const auto text = std::static_pointer_cast(batch->column(0))->GetString(0); if (std::getenv("VGI_FILTER_DEBUG")) std::fprintf(stderr, "[vgi-filter] %s\n", text.c_str()); try { - return json::parse(text); + bool duplicate = false; + std::vector> object_keys; + auto callback = [&](int depth, json::parse_event_t event, json& parsed) { + if (event == json::parse_event_t::object_start) { + if (object_keys.size() <= static_cast(depth)) + object_keys.resize(static_cast(depth) + 1); + object_keys[static_cast(depth)].clear(); + } else if (event == json::parse_event_t::key && depth > 0) { + auto& keys = object_keys[static_cast(depth - 1)]; + if (!keys.insert(parsed.get()).second) duplicate = true; + } + return true; + }; + auto document = json::parse(text, callback); + if (duplicate) invalid("invalid filter JSON: duplicate object key"); + return document; } catch (const json::exception& error) { invalid(std::string("invalid filter JSON: ") + error.what()); } @@ -405,6 +603,8 @@ std::shared_ptr parse_predicate(Parser& parser, const json& item, bool del if (expression->kind == "runtime_filter" && !expression->advisory) { invalid("runtime_filter predicates must be advisory"); } + if (expression->kind != "runtime_filter" && !boolean_type(expression->data_type)) + invalid("predicate root must resolve to BOOLEAN"); return expression; } @@ -460,6 +660,123 @@ arrow::Datum call(const std::string& name, std::vector arguments) "evaluate " + name); } +bool floating_type(const std::shared_ptr& type) { + return type->id() == arrow::Type::HALF_FLOAT || type->id() == arrow::Type::FLOAT || + type->id() == arrow::Type::DOUBLE; +} + +bool comparison_type_supported(const std::shared_ptr& type) { + switch (type->id()) { + case arrow::Type::NA: + case arrow::Type::LIST: + case arrow::Type::LARGE_LIST: + case arrow::Type::FIXED_SIZE_LIST: + case arrow::Type::LIST_VIEW: + case arrow::Type::LARGE_LIST_VIEW: + case arrow::Type::STRUCT: + case arrow::Type::MAP: + case arrow::Type::SPARSE_UNION: + case arrow::Type::DENSE_UNION: + case arrow::Type::RUN_END_ENCODED: + case arrow::Type::EXTENSION: return false; + default: return true; + } +} + +bool floating_compare(double left, double right, const std::string& op) { + int ordering = 0; + if (std::isnan(left)) { + ordering = std::isnan(right) ? 0 : 1; + } else if (std::isnan(right)) { + ordering = -1; + } else if (left < right) { + ordering = -1; + } else if (left > right) { + ordering = 1; + } + if (op == "eq" || op == "not_distinct_from") return ordering == 0; + if (op == "ne" || op == "distinct_from") return ordering != 0; + if (op == "lt") return ordering < 0; + if (op == "le") return ordering <= 0; + if (op == "gt") return ordering > 0; + if (op == "ge") return ordering >= 0; + throw std::runtime_error("unsupported comparison operator"); +} + +arrow::Datum compare(arrow::Datum left, arrow::Datum right, const std::string& op, int64_t length) { + left = decoded_dictionary(std::move(left)); + right = decoded_dictionary(std::move(right)); + auto left_values = array_from_datum(left, length); + auto right_values = array_from_datum(right, length); + if (!left_values->type()->Equals(right_values->type()) || + !comparison_type_supported(left_values->type())) { + throw std::runtime_error("comparison type has no exact standard-v1 evaluator"); + } + + const bool distinct = op == "distinct_from" || op == "not_distinct_from"; + if (floating_type(left_values->type()) || floating_type(right_values->type())) { + left_values = + array_from_datum(value_or_throw(arrow::compute::Cast(left_values, arrow::float64()), + "cast floating comparison input"), + length); + right_values = + array_from_datum(value_or_throw(arrow::compute::Cast(right_values, arrow::float64()), + "cast floating comparison input"), + length); + const auto& left_floats = static_cast(*left_values); + const auto& right_floats = static_cast(*right_values); + arrow::BooleanBuilder output; + auto status = output.Reserve(length); + if (!status.ok()) throw std::runtime_error(status.ToString()); + for (int64_t row = 0; row < length; ++row) { + const bool left_null = left_floats.IsNull(row); + const bool right_null = right_floats.IsNull(row); + if (left_null || right_null) { + if (distinct) { + const bool value = left_null != right_null; + (void)output.Append(op == "distinct_from" ? value : !value); + } else { + (void)output.AppendNull(); + } + continue; + } + (void)output.Append( + floating_compare(left_floats.Value(row), right_floats.Value(row), op)); + } + std::shared_ptr result; + status = output.Finish(&result); + if (!status.ok()) throw std::runtime_error(status.ToString()); + return arrow::Datum(result); + } + + if (distinct) { + auto equals = array_from_datum(call("equal", {left_values, right_values}), length); + const auto& equal_values = static_cast(*equals); + arrow::BooleanBuilder output; + auto status = output.Reserve(length); + if (!status.ok()) throw std::runtime_error(status.ToString()); + for (int64_t row = 0; row < length; ++row) { + const bool left_null = left_values->IsNull(row); + const bool right_null = right_values->IsNull(row); + const bool value = + left_null || right_null ? left_null != right_null : !equal_values.Value(row); + (void)output.Append(op == "distinct_from" ? value : !value); + } + std::shared_ptr result; + status = output.Finish(&result); + if (!status.ok()) throw std::runtime_error(status.ToString()); + return arrow::Datum(result); + } + + static const std::map kernels = { + {"eq", "equal"}, {"ne", "not_equal"}, {"lt", "less"}, + {"le", "less_equal"}, {"gt", "greater"}, {"ge", "greater_equal"}, + }; + const auto found = kernels.find(op); + if (found == kernels.end()) throw std::runtime_error("unsupported comparison operator"); + return call(found->second, {left_values, right_values}); +} + std::optional string_at(const std::shared_ptr& values, int64_t index) { if (values->IsNull(index)) return std::nullopt; if (values->type_id() == arrow::Type::STRING) { @@ -539,6 +856,8 @@ arrow::Datum evaluate(const std::shared_ptr& spec, } if (index < 0) throw std::runtime_error("filter column '" + spec->column_name + "' is absent"); + if (!batch->schema()->field(index)->type()->Equals(spec->data_type)) + throw std::runtime_error("filter column '" + spec->column_name + "' changed type"); return arrow::Datum(batch->column(index)); } if (spec->kind == "field") { @@ -547,26 +866,18 @@ arrow::Datum evaluate(const std::shared_ptr& spec, if (!values || spec->field_index >= static_cast(values->num_fields())) { throw std::runtime_error("filter field_ref input is not the authoritative struct"); } - return arrow::Datum(values->field(static_cast(spec->field_index))); + auto child = + decoded_dictionary(arrow::Datum(values->field(static_cast(spec->field_index)))); + if (values->null_count() == 0) return child; + return call("if_else", {call("is_valid", {parent}), child, + arrow::Datum(arrow::MakeNullScalar(child.type()))}); } if (spec->kind == "literal") { return arrow::Datum(value_or_throw(spec->value->GetScalar(0), "read filter literal")); } if (spec->kind == "constant") { - static const std::map kernels = { - {"eq", "equal"}, - {"ne", "not_equal"}, - {"lt", "less"}, - {"le", "less_equal"}, - {"gt", "greater"}, - {"ge", "greater_equal"}, - {"distinct_from", "is_distinct_from"}, - {"not_distinct_from", "is_not_distinct_from"}, - }; - const auto found = kernels.find(spec->op); - if (found == kernels.end()) throw std::runtime_error("unsupported comparison operator"); - return call(found->second, - {evaluate(spec->children[0], batch), evaluate(spec->children[1], batch)}); + return compare(evaluate(spec->children[0], batch), evaluate(spec->children[1], batch), + spec->op, batch->num_rows()); } if (spec->kind == "and" || spec->kind == "or") { auto result = evaluate(spec->children[0], batch); @@ -582,26 +893,37 @@ arrow::Datum evaluate(const std::shared_ptr& spec, {evaluate(spec->children[0], batch)}); } if (spec->kind == "in") { - static const auto initialized = arrow::compute::Initialize(); - if (!initialized.ok()) throw std::runtime_error(initialized.ToString()); - arrow::compute::SetLookupOptions options(spec->value); - auto result = value_or_throw( - arrow::compute::CallFunction( - "is_in", {decoded_dictionary(evaluate(spec->children[0], batch))}, &options), - "evaluate IN"); + auto input = decoded_dictionary(evaluate(spec->children[0], batch)); + auto values = decoded_dictionary(arrow::Datum(spec->value)).make_array(); + auto result = arrow::Datum(value_or_throw( + arrow::MakeArrayFromScalar(arrow::BooleanScalar(false), batch->num_rows()), + "initialize IN result")); + for (int64_t index = 0; index < values->length(); ++index) { + auto candidate = value_or_throw(values->GetScalar(index), "read IN value"); + result = call("or_kleene", {std::move(result), compare(input, arrow::Datum(candidate), + "eq", batch->num_rows())}); + } return spec->negated ? call("invert", {std::move(result)}) : result; } if (spec->kind == "cast") { + const auto source = spec->children[0]->data_type; + const auto target = spec->data_type; + if (!exact_context_free_cast(source, target)) + throw std::runtime_error("cast has no exact context-free standard-v1 evaluator"); return value_or_throw( arrow::compute::Cast(evaluate(spec->children[0], batch), spec->data_type), "evaluate cast"); } if (spec->kind == "arithmetic") { - return call(spec->op, + return call(spec->op + "_checked", {evaluate(spec->children[0], batch), evaluate(spec->children[1], batch)}); } - if (spec->kind == "negate") return call("negate", {evaluate(spec->children[0], batch)}); - if (spec->kind == "call") return evaluate_standard_call(*spec, batch); + if (spec->kind == "negate") return call("negate_checked", {evaluate(spec->children[0], batch)}); + if (spec->kind == "call") { + if (spec->function == "list_contains") + throw std::runtime_error("list_contains has no exact standard-v1 evaluator"); + return evaluate_standard_call(*spec, batch); + } throw std::runtime_error("runtime filter has no negotiated evaluator"); } @@ -690,11 +1012,11 @@ bool mentions(const std::shared_ptr& spec, const std::string& column) { std::shared_ptr discrete_values(const std::shared_ptr& spec, const std::string& column) { if (spec->kind == "constant" && spec->op == "eq" && spec->children.size() == 2 && - root_column(spec->children[0]) == column && spec->children[1]->kind == "literal") { + is_direct_column(spec->children[0], column) && spec->children[1]->kind == "literal") { return spec->children[1]->value; } if (spec->kind == "in" && !spec->negated && !spec->children.empty() && - root_column(spec->children[0]) == column) + is_direct_column(spec->children[0], column)) return spec->value; if (spec->kind == "and") { for (const auto& child : spec->children) { @@ -725,6 +1047,7 @@ struct Bounds { std::optional scalar_int64(const std::shared_ptr& value) { if (!value || value->length() == 0 || value->IsNull(0)) return std::nullopt; + if (!arrow::is_integer(logical_type(value->type())->id())) return std::nullopt; auto casted = arrow::compute::Cast(*value->Slice(0, 1), arrow::int64()); if (!casted.ok()) return std::nullopt; return std::static_pointer_cast(casted.MoveValueUnsafe())->Value(0); @@ -752,9 +1075,9 @@ std::optional bounds_for(const std::shared_ptr& spec, const std::s if (spec->kind == "constant" && spec->children.size() == 2) { auto op = spec->op; std::shared_ptr literal; - if (root_column(spec->children[0]) == column && spec->children[1]->kind == "literal") { + if (is_direct_column(spec->children[0], column) && spec->children[1]->kind == "literal") { literal = spec->children[1]; - } else if (root_column(spec->children[1]) == column && + } else if (is_direct_column(spec->children[1], column) && spec->children[0]->kind == "literal") { literal = spec->children[0]; if (op == "gt") @@ -779,7 +1102,8 @@ std::optional bounds_for(const std::shared_ptr& spec, const std::s if (!result.min && !result.max) return std::nullopt; return result; } - if (spec->kind == "in" && !spec->negated && root_column(spec->children[0]) == column) { + if (spec->kind == "in" && !spec->negated && is_direct_column(spec->children[0], column)) { + if (!arrow::is_integer(logical_type(spec->value->type())->id())) return std::nullopt; auto casted = arrow::compute::Cast(*spec->value, arrow::int64()); if (!casted.ok()) return std::nullopt; const auto values = std::static_pointer_cast(casted.MoveValueUnsafe()); @@ -866,11 +1190,13 @@ void PushdownFilters::apply_delta(const std::string& ipc_bytes) { if (id.size() > 128) invalid("predicate ID exceeds 128 bytes"); if (!seen.insert(id).second) invalid("duplicate delta predicate ID"); if (required_ids_.count(id)) invalid("delta targets required predicate"); + std::shared_ptr parsed; if (operation == "remove") { require_keys(update, {"operation", "id", "revision"}, {}, "remove update"); } else if (operation == "upsert") { require_keys(update, {"operation", "id", "revision", "mode", "source", "expression"}, {}, "upsert update"); + parsed = parse_predicate(parser, update, true); } else invalid("delta operation must be remove or upsert"); const auto old = next.revisions_.find(id); @@ -878,7 +1204,7 @@ void PushdownFilters::apply_delta(const std::string& ipc_bytes) { next.specs_.erase(std::remove_if(next.specs_.begin(), next.specs_.end(), [&](const auto& spec) { return spec->id == id; }), next.specs_.end()); - if (operation == "upsert") next.specs_.push_back(parse_predicate(parser, update, true)); + if (parsed) next.specs_.push_back(std::move(parsed)); next.revisions_[id] = revision; } if (next.revisions_.size() > kMaxPredicateIds) invalid("delta exceeds predicate-ID limit"); diff --git a/tests/function_test.cpp b/tests/function_test.cpp index 410924b..7da3a4e 100644 --- a/tests/function_test.cpp +++ b/tests/function_test.cpp @@ -1,6 +1,10 @@ // © Copyright 2025, 2026 Query Farm LLC - https://query.farm #include +#include +#include +#include + #include #include #include @@ -34,16 +38,17 @@ class Dynamic : public Fixed { vgi::FunctionMetadata metadata() const override { return {}; } }; -std::string filter_batch(const std::string& document, - const std::shared_ptr& payload = nullptr) { +std::string filter_batch_with_payloads( + const std::string& document, + const std::vector>>& payloads) { arrow::StringBuilder spec_builder; REQUIRE(spec_builder.Append(document).ok()); std::shared_ptr spec; REQUIRE(spec_builder.Finish(&spec).ok()); arrow::FieldVector fields{arrow::field("filter_spec", arrow::utf8(), false)}; arrow::ArrayVector arrays{spec}; - if (payload) { - fields.push_back(arrow::field("value_0", payload->type(), true)); + for (const auto& [name, payload] : payloads) { + fields.push_back(arrow::field(name, payload->type(), true)); arrays.push_back(payload); } auto metadata = arrow::key_value_metadata( @@ -53,6 +58,12 @@ std::string filter_batch(const std::string& document, arrow::RecordBatch::Make(arrow::schema(fields, metadata), 1, arrays)); } +std::string filter_batch(const std::string& document, + const std::shared_ptr& payload = nullptr) { + if (payload) return filter_batch_with_payloads(document, {{"value_0", payload}}); + return filter_batch_with_payloads(document, {}); +} + std::shared_ptr int64_values(std::initializer_list values) { arrow::Int64Builder builder; for (const auto value : values) REQUIRE(builder.Append(value).ok()); @@ -61,6 +72,49 @@ std::shared_ptr int64_values(std::initializer_list values return result; } +std::shared_ptr nullable_int64_values( + std::initializer_list> values) { + arrow::Int64Builder builder; + for (const auto value : values) { + if (value) + REQUIRE(builder.Append(*value).ok()); + else + REQUIRE(builder.AppendNull().ok()); + } + std::shared_ptr result; + REQUIRE(builder.Finish(&result).ok()); + return result; +} + +std::shared_ptr int64_list(std::initializer_list> values) { + auto child = std::make_shared(); + arrow::ListBuilder builder(arrow::default_memory_pool(), child); + REQUIRE(builder.Append().ok()); + for (const auto value : values) { + if (value) + REQUIRE(child->Append(*value).ok()); + else + REQUIRE(child->AppendNull().ok()); + } + std::shared_ptr result; + REQUIRE(builder.Finish(&result).ok()); + return result; +} + +std::shared_ptr double_values(std::initializer_list values) { + arrow::DoubleBuilder builder; + for (const auto value : values) REQUIRE(builder.Append(value).ok()); + std::shared_ptr result; + REQUIRE(builder.Finish(&result).ok()); + return result; +} + +std::shared_ptr null_value(const std::shared_ptr& type) { + auto result = arrow::MakeArrayOfNull(type, 1); + REQUIRE(result.ok()); + return result.MoveValueUnsafe(); +} + } // namespace TEST_CASE("a fixed return type binds without an override", "[function]") { @@ -247,3 +301,121 @@ TEST_CASE("Filter v2 dynamic deltas update snapshot state atomically", "[filter- {int64_values({1, 2, 3, 4, 5})}); REQUIRE(filters.apply(input)->num_rows() == 3); } + +TEST_CASE("Filter v2 IN follows SQL three-valued NULL semantics", "[filter-v2]") { + const auto in_document = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[{"id":"p","revision":0,"mode":"required","source":"query","expression":{"node":"in","expression":{"node":"column_ref","column_index":0,"column_name":"n"},"set":{"kind":"literal","value_ref":0},"negated":false}}]})"; + auto in_filter = + vgi::PushdownFilters::parse(filter_batch(in_document, int64_list({1, std::nullopt})), {}, + arrow::schema({arrow::field("n", arrow::int64())})); + auto input = arrow::RecordBatch::Make(arrow::schema({arrow::field("n", arrow::int64())}), 3, + {nullable_int64_values({1, 2, std::nullopt})}); + REQUIRE(in_filter.apply(input)->num_rows() == 1); + + const auto not_in_document = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[{"id":"p","revision":0,"mode":"required","source":"query","expression":{"node":"in","expression":{"node":"column_ref","column_index":0,"column_name":"n"},"set":{"kind":"literal","value_ref":0},"negated":true}}]})"; + auto not_in_with_null = + vgi::PushdownFilters::parse(filter_batch(not_in_document, int64_list({1, std::nullopt})), + {}, arrow::schema({arrow::field("n", arrow::int64())})); + REQUIRE(not_in_with_null.apply(input)->num_rows() == 0); + + auto not_in_without_null = + vgi::PushdownFilters::parse(filter_batch(not_in_document, int64_list({1})), {}, + arrow::schema({arrow::field("n", arrow::int64())})); + auto null_input = arrow::RecordBatch::Make(arrow::schema({arrow::field("n", arrow::int64())}), + 1, {nullable_int64_values({std::nullopt})}); + REQUIRE(not_in_without_null.apply(null_input)->num_rows() == 0); +} + +TEST_CASE("Filter v2 field references propagate parent struct validity", "[filter-v2]") { + const auto document = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[{"id":"p","revision":0,"mode":"required","source":"query","expression":{"node":"comparison","op":"eq","left":{"node":"field_ref","expression":{"node":"column_ref","column_index":0,"column_name":"s"},"field_index":0,"field_name":"n"},"right":{"node":"literal","value_ref":0}}}]})"; + const auto struct_type = arrow::struct_({arrow::field("n", arrow::int64())}); + auto filters = vgi::PushdownFilters::parse(filter_batch(document, int64_values({5})), {}, + arrow::schema({arrow::field("s", struct_type)})); + + arrow::BooleanBuilder validity_builder; + REQUIRE(validity_builder.Append(false).ok()); + REQUIRE(validity_builder.Append(true).ok()); + std::shared_ptr validity; + REQUIRE(validity_builder.Finish(&validity).ok()); + auto maybe_struct = arrow::StructArray::Make( + {int64_values({5, 5})}, {arrow::field("n", arrow::int64())}, + std::static_pointer_cast(validity)->values(), 1); + REQUIRE(maybe_struct.ok()); + auto input = arrow::RecordBatch::Make(arrow::schema({arrow::field("s", struct_type)}), 2, + {maybe_struct.MoveValueUnsafe()}); + REQUIRE(filters.apply(input)->num_rows() == 1); + REQUIRE_FALSE(filters.column_values("s")); + REQUIRE_FALSE(filters.column_bounds("s").min); + REQUIRE_FALSE(filters.column_bounds("s").max); +} + +TEST_CASE("Filter v2 comparisons use DuckDB NaN and signed-zero ordering", "[filter-v2]") { + auto input = arrow::RecordBatch::Make(arrow::schema({arrow::field("n", arrow::float64())}), 4, + {double_values({NAN, -0.0, 0.0, 1.0})}); + const auto rows_for = [&](const std::string& op, double literal) { + const auto document = + std::string( + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[{"id":"p","revision":0,"mode":"required","source":"query","expression":{"node":"comparison","op":")") + + op + + R"(","left":{"node":"column_ref","column_index":0,"column_name":"n"},"right":{"node":"literal","value_ref":0}}}]})"; + auto filters = vgi::PushdownFilters::parse(filter_batch(document, double_values({literal})), + {}, input->schema()); + return filters.apply(input)->num_rows(); + }; + REQUIRE(rows_for("eq", NAN) == 1); + REQUIRE(rows_for("lt", NAN) == 3); + REQUIRE(rows_for("distinct_from", NAN) == 3); + REQUIRE(rows_for("not_distinct_from", NAN) == 1); + REQUIRE(rows_for("eq", 0.0) == 2); +} + +TEST_CASE("Filter v2 rejects duplicate JSON keys and malformed stale upserts", "[filter-v2]") { + const auto duplicate = + R"({"encoding":"vgi.filters.v2","encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[]})"; + REQUIRE_THROWS(vgi::PushdownFilters::parse(filter_batch(duplicate), {}, arrow::schema({}))); + + const auto snapshot = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[]})"; + auto filters = vgi::PushdownFilters::parse(filter_batch(snapshot), {}, arrow::schema({})); + const auto fresh = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"delta","updates":[{"operation":"remove","id":"p","revision":2}]})"; + filters.apply_delta(filter_batch(fresh)); + const auto stale_malformed = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"delta","updates":[{"operation":"upsert","id":"p","revision":1,"mode":"advisory","source":"top_n","expression":{"node":"bogus"}}]})"; + REQUIRE_THROWS(filters.apply_delta(filter_batch(stale_malformed))); +} + +TEST_CASE("Filter v2 arithmetic is checked and unsafe casts are declined", "[filter-v2]") { + const auto arithmetic = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[{"id":"p","revision":0,"mode":"required","source":"query","expression":{"node":"comparison","op":"eq","left":{"node":"arithmetic","op":"add","left":{"node":"column_ref","column_index":0,"column_name":"n"},"right":{"node":"literal","value_ref":0}},"right":{"node":"literal","value_ref":1}}}]})"; + auto filters = vgi::PushdownFilters::parse( + filter_batch_with_payloads( + arithmetic, {{"value_0", int64_values({1})}, {"value_1", int64_values({0})}}), + {}, arrow::schema({arrow::field("n", arrow::int64())})); + auto overflow = arrow::RecordBatch::Make(arrow::schema({arrow::field("n", arrow::int64())}), 1, + {int64_values({std::numeric_limits::max()})}); + REQUIRE_THROWS(filters.apply(overflow)); + + const auto unsafe_cast = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[{"id":"p","revision":0,"mode":"required","source":"query","expression":{"node":"comparison","op":"eq","left":{"node":"cast","expression":{"node":"column_ref","column_index":0,"column_name":"n"},"type_ref":0},"right":{"node":"literal","value_ref":0}}}]})"; + REQUIRE_THROWS(vgi::PushdownFilters::parse( + filter_batch_with_payloads( + unsafe_cast, {{"type_0", null_value(arrow::int64())}, {"value_0", int64_values({1})}}), + {}, arrow::schema({arrow::field("n", arrow::float64())}))); +} + +TEST_CASE("Filter v2 fails closed when a runtime batch changes column type", "[filter-v2]") { + const auto document = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[{"id":"p","revision":0,"mode":"required","source":"query","expression":{"node":"comparison","op":"eq","left":{"node":"column_ref","column_index":0,"column_name":"n"},"right":{"node":"literal","value_ref":0}}}]})"; + auto filters = vgi::PushdownFilters::parse(filter_batch(document, int64_values({1})), {}, + arrow::schema({arrow::field("n", arrow::int64())})); + arrow::StringBuilder builder; + REQUIRE(builder.Append("1").ok()); + std::shared_ptr strings; + REQUIRE(builder.Finish(&strings).ok()); + auto changed = + arrow::RecordBatch::Make(arrow::schema({arrow::field("n", arrow::utf8())}), 1, {strings}); + REQUIRE_THROWS(filters.apply(changed)); +} From 1273ca42dc5f1f56dc453e6e4d35179d6df6bd3c Mon Sep 17 00:00:00 2001 From: Rusty Conover Date: Thu, 10 Sep 2026 21:47:56 -0400 Subject: [PATCH 4/5] Implement exact list_contains filtering --- src/pushdown.cpp | 65 +++++++++++++++++++++++++++++++++++++++-- tests/function_test.cpp | 44 ++++++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 3 deletions(-) diff --git a/src/pushdown.cpp b/src/pushdown.cpp index a6eb342..3d313d5 100644 --- a/src/pushdown.cpp +++ b/src/pushdown.cpp @@ -788,6 +788,67 @@ std::optional string_at(const std::shared_ptr& values throw std::runtime_error("standard string filter requires UTF8 arguments"); } +double floating_scalar_value(const std::shared_ptr& value) { + auto array = + value_or_throw(arrow::MakeArrayFromScalar(*value, 1), "materialize floating scalar"); + auto casted = + value_or_throw(arrow::compute::Cast(array, arrow::float64()), "cast floating scalar"); + return std::static_pointer_cast(casted.make_array())->Value(0); +} + +bool nested_scalar_equal(const std::shared_ptr& left, + const std::shared_ptr& right) { + if (!left || !right) throw std::runtime_error("nested equality received no scalar"); + if (!left->is_valid || !right->is_valid) return left->is_valid == right->is_valid; + if (!same_logical_type(left->type, right->type)) return false; + const auto decode_dictionary = [](std::shared_ptr value) { + if (value->type->id() != arrow::Type::DICTIONARY) return value; + auto array = + value_or_throw(arrow::MakeArrayFromScalar(*value, 1), "materialize dictionary scalar"); + const auto target = logical_type(value->type); + auto decoded = + value_or_throw(arrow::compute::Cast(array, target), "decode dictionary scalar"); + return value_or_throw(decoded.make_array()->GetScalar(0), "read dictionary scalar"); + }; + const auto left_value = decode_dictionary(left); + const auto right_value = decode_dictionary(right); + const auto type = left_value->type; + if (floating_type(type)) { + return floating_compare(floating_scalar_value(left_value), + floating_scalar_value(right_value), "eq"); + } + if (type->id() == arrow::Type::STRUCT) { + const auto& left_struct = static_cast(*left_value); + const auto& right_struct = static_cast(*right_value); + if (left_struct.value.size() != right_struct.value.size()) return false; + for (size_t i = 0; i < left_struct.value.size(); ++i) { + if (!nested_scalar_equal(left_struct.value[i], right_struct.value[i])) return false; + } + return true; + } + switch (type->id()) { + case arrow::Type::LIST: + case arrow::Type::LARGE_LIST: + case arrow::Type::FIXED_SIZE_LIST: + case arrow::Type::LIST_VIEW: + case arrow::Type::LARGE_LIST_VIEW: + case arrow::Type::MAP: { + const auto& left_list = static_cast(*left_value); + const auto& right_list = static_cast(*right_value); + if (left_list.value->length() != right_list.value->length()) return false; + for (int64_t i = 0; i < left_list.value->length(); ++i) { + const auto left_child = + value_or_throw(left_list.value->GetScalar(i), "read nested list value"); + const auto right_child = + value_or_throw(right_list.value->GetScalar(i), "read nested list value"); + if (!nested_scalar_equal(left_child, right_child)) return false; + } + return true; + } + default: return left_value->Equals(*right_value); + } +} + arrow::Datum evaluate(const std::shared_ptr& spec, const std::shared_ptr& batch); @@ -818,7 +879,7 @@ arrow::Datum evaluate_standard_call(const Spec& spec, if (values->IsNull(i)) continue; const auto candidate = value_or_throw(values->GetScalar(i), "read list_contains value"); - if (candidate->Equals(*needle)) { + if (nested_scalar_equal(candidate, needle)) { matched = true; break; } @@ -920,8 +981,6 @@ arrow::Datum evaluate(const std::shared_ptr& spec, } if (spec->kind == "negate") return call("negate_checked", {evaluate(spec->children[0], batch)}); if (spec->kind == "call") { - if (spec->function == "list_contains") - throw std::runtime_error("list_contains has no exact standard-v1 evaluator"); return evaluate_standard_call(*spec, batch); } throw std::runtime_error("runtime filter has no negotiated evaluator"); diff --git a/tests/function_test.cpp b/tests/function_test.cpp index 7da3a4e..02bf2f7 100644 --- a/tests/function_test.cpp +++ b/tests/function_test.cpp @@ -419,3 +419,47 @@ TEST_CASE("Filter v2 fails closed when a runtime batch changes column type", "[f arrow::RecordBatch::Make(arrow::schema({arrow::field("n", arrow::utf8())}), 1, {strings}); REQUIRE_THROWS(filters.apply(changed)); } + +TEST_CASE("Filter v2 list_contains uses DuckDB nested and NaN equality", "[filter-v2]") { + const auto document = + R"({"encoding":"vgi.filters.v2","semantics":"vgi.duckdb.standard.v1","kind":"snapshot","predicates":[{"id":"p","revision":0,"mode":"required","source":"query","expression":{"node":"call","function":"list_contains","arguments":[{"node":"column_ref","column_index":0,"column_name":"values"},{"node":"literal","value_ref":0}]}}]})"; + + auto doubles = std::make_shared(); + arrow::ListBuilder list_builder(arrow::default_memory_pool(), doubles); + REQUIRE(list_builder.Append().ok()); + REQUIRE(doubles->Append(NAN).ok()); + REQUIRE(list_builder.Append().ok()); + REQUIRE(doubles->Append(1.0).ok()); + REQUIRE(list_builder.Append().ok()); + REQUIRE(doubles->AppendNull().ok()); + REQUIRE(list_builder.AppendNull().ok()); + std::shared_ptr lists; + REQUIRE(list_builder.Finish(&lists).ok()); + auto input = arrow::RecordBatch::Make(arrow::schema({arrow::field("values", lists->type())}), 4, + {lists}); + auto nan_filter = vgi::PushdownFilters::parse(filter_batch(document, double_values({NAN})), {}, + input->schema()); + REQUIRE(nan_filter.apply(input)->num_rows() == 1); + auto null_filter = vgi::PushdownFilters::parse( + filter_batch(document, null_value(arrow::float64())), {}, input->schema()); + REQUIRE(null_filter.apply(input)->num_rows() == 0); + + auto integers = std::make_shared(); + auto inner_lists = std::make_shared(arrow::default_memory_pool(), integers); + arrow::ListBuilder outer_lists(arrow::default_memory_pool(), inner_lists); + REQUIRE(outer_lists.Append().ok()); + REQUIRE(inner_lists->Append().ok()); + REQUIRE(integers->Append(1).ok()); + REQUIRE(integers->AppendNull().ok()); + REQUIRE(outer_lists.Append().ok()); + REQUIRE(inner_lists->Append().ok()); + REQUIRE(integers->Append(1).ok()); + REQUIRE(integers->Append(2).ok()); + std::shared_ptr nested_lists; + REQUIRE(outer_lists.Finish(&nested_lists).ok()); + auto nested_input = arrow::RecordBatch::Make( + arrow::schema({arrow::field("values", nested_lists->type())}), 2, {nested_lists}); + auto nested_filter = vgi::PushdownFilters::parse( + filter_batch(document, int64_list({1, std::nullopt})), {}, nested_input->schema()); + REQUIRE(nested_filter.apply(nested_input)->num_rows() == 1); +} From ee20ca82f3fa3736bb33477042461e559a71ad0a Mon Sep 17 00:00:00 2001 From: Rusty Conover Date: Thu, 10 Sep 2026 21:53:10 -0400 Subject: [PATCH 5/5] Add split dynamic filter fixture --- example-worker/table/splits.cpp | 158 ++++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/example-worker/table/splits.cpp b/example-worker/table/splits.cpp index 015e2a7..db05478 100644 --- a/example-worker/table/splits.cpp +++ b/example-worker/table/splits.cpp @@ -20,6 +20,7 @@ #include #include +#include #include #include #include @@ -63,6 +64,27 @@ std::shared_ptr n_schema() { return arrow::schema({arrow::field("n", arrow::int64(), /*nullable=*/false)}); } +std::shared_ptr 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 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 bind(const vgi::BindParams&) const override { + return dynamic_filter_schema(); + } + + vgi::TableCardinality cardinality(const vgi::ProcessParams& params) const override { + const auto rows = std::max(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(0, params.arguments.named_int64("n").value_or(0)); + const int64_t want = + std::max(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 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> 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( + 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 schema, + std::vector> 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 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> built(2); + (void)ns.Finish(&built[0]); + (void)reports.Finish(&built[1]); + + const std::vector names{"n", "pushed_filters"}; + std::vector> projected; + projected.reserve(static_cast(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(found - names.begin())]); + } + return arrow::RecordBatch::Make(schema_, end - begin, std::move(projected)); + } + + private: + std::shared_ptr schema_; + std::vector> 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()); worker.register_table(std::make_shared()); + worker.register_table(std::make_shared()); worker.register_table(std::make_shared()); }