feat(optimizer): enforce required_field_filter_paths on VGI Tables

Adds VgiRequiredFiltersOptimizer (post-optimize OptimizerExtension)
that walks every LogicalGet and rejects scans missing any path the
Table declared in TableInfo.required_field_filter_paths. The error
fires at bind/optimize time before any worker / S3 byte is read:

    Binder Error: Table 'overture.places.place' requires WHERE
    filters on: bbox.xmin, bbox.xmax, bbox.ymin, bbox.ymax.
    Missing: bbox.xmax, bbox.ymin, bbox.ymax. ...

Prefix-aware: a filter on a parent struct (e.g. WHERE bbox = small)
satisfies all bbox.* requirements, because the wider filter is at
least as constraining as the four-corner conjunction.

Native scan delegation via ScanFunctionResult(function_name=
"read_parquet", ...): VgiTableEntry::GetScanFunctionImpl recognises
SYSTEM_CATALOG function names, eagerly binds the native function,
validates that declared catalog columns match the bind output by
name+position (catches Hive-partition column mismatches like the
Overture theme/type columns at attach time), and stows the bound
function inside a VgiNativeDelegationMarkerBindData placeholder.
VgiRequiredFiltersOptimizer then runs its filter-presence check
against the marker's catalog metadata and swaps the marker out for
the real native bind — post-pass the LogicalGet looks exactly like
what DuckDB would produce for a direct read_parquet call, so every
native callback (scan/init/statistics/dynamic_to_string/future
MultiFile additions) just works.

The marker function vgi_native_delegation_marker is registered with
a throwing deserialize callback: it names the function and explains
the optimizer-ordering invariant if it's ever serialized (a
LogicalOperator::Copy() round-trips through serialize/deserialize
and looks the function up by name in the system catalog).

WalkTableFilter (the path-collection helper) mirrors the existing
serializer at vgi_table_function_impl.cpp:714 — OPTIONAL_FILTER
subtrees containing a DynamicFilter are skipped so Top-N machinery
artifacts can't spuriously satisfy required_field_filter_paths the
user never wrote. VgiContainsDynamicFilter is promoted from
anonymous-namespace static to external linkage in duckdb::vgi.

Wire-additive on TableInfo (see vgi_catalog_api.cpp ParseTableInfo);
backward-compatible with older workers.

extension_config.cmake: hard-code haybarn-httpfs until the CMake-
side HAYBARN_VERSION_STRING propagation is sorted (the C++ -D define
is on the compile line but no set(...) reaches this scope, so the
stock httpfs was silently winning the local make release path).

Sqllogictests under test/sql/integration/table/required_field_-
filter_paths_*.test exercise basic/struct/nested/prefix/disjunction-
null/above-get/complex scenarios; comments.test + function_-
registration.test counts adjusted for the new rff_* fixtures.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rustyconover

extension_config.cmake

@@ -10,13 +10,22 @@ duckdb_extension_load(vgi
 # engine fork defines HAYBARN_VERSION_STRING (a top-level CMake var, set before
 # extension configs are included); stock DuckDB never does. Build against the
 # matching httpfs in each case — they have genuinely different sources.
-if(DEFINED HAYBARN_VERSION_STRING)
-    set(HTTPFS_GIT_URL https://github.com/Query-farm-haybarn/haybarn-httpfs)
-    set(HTTPFS_GIT_TAG 1f23e5bd8f7a50253c08b00bb7a88ecfa15862df)
-else()
-    set(HTTPFS_GIT_URL https://github.com/duckdb/duckdb-httpfs)
-    set(HTTPFS_GIT_TAG 52afb4204a3238d6ee132e83340f8d68c40ee91c)
-endif()
+#
+# TEMPORARILY hard-coded to the haybarn fork: the CMake-side
+# `if(DEFINED HAYBARN_VERSION_STRING)` check wasn't firing on the local
+# `make release` path here (the C++ `-DHAYBARN_VERSION_STRING=...` preprocessor
+# define is set on the compile command line but no `set(...)` CMake variable
+# propagates to this scope), so the stock-DuckDB httpfs was being selected.
+# Force the haybarn fork until the CMake propagation is sorted out.
+set(HTTPFS_GIT_URL https://github.com/Query-farm-haybarn/haybarn-httpfs)
+set(HTTPFS_GIT_TAG 1f23e5bd8f7a50253c08b00bb7a88ecfa15862df)
+# if(DEFINED HAYBARN_VERSION_STRING)
+#     set(HTTPFS_GIT_URL https://github.com/Query-farm-haybarn/haybarn-httpfs)
+#     set(HTTPFS_GIT_TAG 1f23e5bd8f7a50253c08b00bb7a88ecfa15862df)
+# else()
+#     set(HTTPFS_GIT_URL https://github.com/duckdb/duckdb-httpfs)
+#     set(HTTPFS_GIT_TAG 52afb4204a3238d6ee132e83340f8d68c40ee91c)
+# endif()
 
 # DuckDB 1.5.3's httpfs (52afb42) is curl-based. On the MinGW toolchain
 # (x64-mingw-static, rtools GCC) it fails to link against static curl with

src/generated/vgi_protocol_schemas.hpp

@@ -2,7 +2,7 @@
 // GENERATED by vgi.codegen.cpp_schemas. DO NOT EDIT BY HAND.
 //
 // Generator: vgi-gen-cpp-schemas v1
-// Content hash: b6d03efdf957
+// Content hash: 15ff8ede31b1
 //
 // To regenerate:
 //   uv run --project ~/Development/vgi-python vgi-gen-cpp-schemas \
@@ -69,6 +69,7 @@ inline const std::shared_ptr<arrow::Schema> &TableInfoSchema() {
 	    arrow::field("cardinality_max", arrow::int64(), /*nullable=*/false),
 	    arrow::field("column_statistics", arrow::binary(), /*nullable=*/false),
 	    arrow::field("bind_result", arrow::binary(), /*nullable=*/false),
+	    arrow::field("required_field_filter_paths", arrow::list(arrow::utf8()), /*nullable=*/false),
 	});
 	return schema;
 }

src/include/storage/vgi_table_entry.hpp

@@ -9,14 +9,67 @@
 
 #include "duckdb/catalog/catalog_entry/table_catalog_entry.hpp"
 #include "duckdb/catalog/entry_lookup_info.hpp"
+#include "duckdb/function/table_function.hpp"
 #include "duckdb/parser/parsed_data/create_table_info.hpp"
+#include "duckdb/planner/operator/logical_get.hpp"
 #include "duckdb/storage/statistics/base_statistics.hpp"
 #include "vgi_catalog_api.hpp"
 
 namespace duckdb {
 
 class VgiCatalog;
 class VgiSchemaEntry;
+class VgiTableEntry;
+
+// VgiNativeDelegationMarkerBindData — single-branch sibling of
+// VgiMultiBranchMarkerBindData. Carried by the placeholder TableFunction
+// returned from VgiTableEntry::GetScanFunctionImpl when the worker's
+// ScanFunctionResult names a SYSTEM_CATALOG function (read_parquet, etc.).
+// VgiRequiredFiltersOptimizer (vgi_extension.cpp) detects this marker via
+// `dynamic_cast<VgiNativeDelegationMarkerBindData *>(get.bind_data.get())`,
+// enforces `table.required_field_filter_paths`, and rewrites the LogicalGet
+// in place to point at the bound native TableFunction. See the comment
+// block at the struct definition in vgi_table_entry.cpp for full lifecycle.
+struct VgiNativeDelegationMarkerBindData : public duckdb::TableFunctionData {
+	// Owning VgiTableEntry — the rewriter reads
+	// `required_field_filter_paths` from this to know which paths must appear
+	// in `LogicalGet::table_filters`.
+	reference<VgiTableEntry> table;
+
+	// Pre-bound native TableFunction + bind_data. GetScanFunctionImpl binds
+	// eagerly (so we can populate the LogicalGet's returned_types/names
+	// faithfully), and the rewriter just transfers ownership of these into
+	// the LogicalGet during the in-place swap. No re-binding in the optimizer.
+	TableFunction native_tf;
+	mutable unique_ptr<duckdb::FunctionData> native_bind;
+	std::vector<LogicalType> native_return_types;
+	std::vector<std::string> native_return_names;
+	virtual_column_map_t native_virtual_columns;
+
+	// Originating ScanFunctionResult — kept for diagnostics + logging.
+	vgi::VgiScanFunctionResult scan_result;
+	std::string worker_path;
+
+	VgiNativeDelegationMarkerBindData(VgiTableEntry &table_p, TableFunction native_tf_p,
+	                                   unique_ptr<duckdb::FunctionData> native_bind_p,
+	                                   std::vector<LogicalType> native_return_types_p,
+	                                   std::vector<std::string> native_return_names_p,
+	                                   virtual_column_map_t native_virtual_columns_p,
+	                                   vgi::VgiScanFunctionResult scan_result_p,
+	                                   std::string worker_path_p)
+	    : table(table_p), native_tf(std::move(native_tf_p)),
+	      native_bind(std::move(native_bind_p)),
+	      native_return_types(std::move(native_return_types_p)),
+	      native_return_names(std::move(native_return_names_p)),
+	      native_virtual_columns(std::move(native_virtual_columns_p)),
+	      scan_result(std::move(scan_result_p)), worker_path(std::move(worker_path_p)) {
+	}
+};
+
+// Construct the native-delegation marker. Its execute callback throws
+// InternalException — reaching it means VgiRequiredFiltersOptimizer didn't
+// run, which IS a bug.
+TableFunction MakeNativeDelegationMarkerFunction();
 
 class VgiTableEntry : public TableCatalogEntry {
 public:

src/include/vgi_catalog_api.hpp

@@ -458,6 +458,19 @@ struct VgiTableInfo {
 	// declarative path (which is restricted to `@bind_fixed_schema`-decorated
 	// functions whose bind output is a pure function of `cls.FIXED_SCHEMA`).
 	std::optional<std::vector<uint8_t>> bind_result;
+
+	// Dotted-path column references that the optimizer extension must verify
+	// appear in any scan's WHERE expression. Top-level column names
+	// (`"country"`) or struct subfields (`"bbox.xmin"`, `"nested.outer.inner"`).
+	// Empty means no enforcement — the zero-cost fast path for every existing
+	// table.
+	//
+	// Satisfaction is prefix-based: a present filter on a shorter dotted path
+	// satisfies any required path it's a prefix of. A whole-struct filter on
+	// `bbox` therefore satisfies every required `"bbox.*"` path.
+	// `VgiRequiredFiltersOptimizer` consults this list at post-optimize time
+	// and throws `BinderException` listing any unsatisfied paths.
+	std::vector<std::string> required_field_filter_paths;
 };
 
 // View metadata from the worker

src/include/vgi_table_function_impl.hpp

@@ -584,6 +584,14 @@ SerializedFilters VgiSerializeFilters(ClientContext &context, const vector<colum
                                       const vector<string> &column_names, const string &worker_path,
                                       const string &rowid_column_name = "");
 
+//! Returns true if any descendant of ``filter`` is a DynamicFilter (Top-N
+//! tick-time bound). Consumers that walk TableFilter trees for *static*
+//! information (presence-of-filter checks, serialization of constants, etc.)
+//! should skip the entire OptionalFilter subtree when this returns true:
+//! the DynamicFilter has no value at init time, and any partial walk yields
+//! a stricter view than the OptionalFilter actually constrains.
+bool VgiContainsDynamicFilter(const TableFilter &filter);
+
 //! Expression pushdown callback: checks if the expression tree only uses functions the worker supports
 bool VgiPushdownExpression(ClientContext &context, const LogicalGet &get, Expression &expr);
 

src/storage/vgi_table_entry.cpp

@@ -8,6 +8,7 @@
 #include "duckdb/function/table_function.hpp"
 #include "duckdb/main/client_context.hpp"
 #include "duckdb/main/extension_helper.hpp"
+#include "duckdb/parser/tableref/table_function_ref.hpp"
 #include "duckdb/planner/tableref/bound_at_clause.hpp"
 #include "duckdb/storage/table_storage_info.hpp"
 
@@ -21,6 +22,36 @@
 
 namespace duckdb {
 
+// VgiNativeDelegationMarkerBindData is declared in storage/vgi_table_entry.hpp
+// so VgiRequiredFiltersOptimizer (vgi_extension.cpp) can dynamic_cast against
+// it. See the header for full docstring and lifecycle.
+
+namespace {
+
+// Marker placeholder TableFunction — never executed; the optimizer
+// extension must replace it. Mirrors MakeMultiBranchMarkerFunction in
+// vgi_multi_scan_rewriter.cpp.
+void NativeDelegationMarkerExecute(ClientContext &, TableFunctionInput &, DataChunk &) {
+	throw InternalException(
+	    "VgiRequiredFiltersOptimizer did not fire — native-delegation placeholder "
+	    "reached execution. Check that the optimizer extension is registered and "
+	    "that no other pass dropped the marker. This is a bug — please report it.");
+}
+
+}  // namespace
+
+TableFunction MakeNativeDelegationMarkerFunction() {
+	TableFunction fn("vgi_native_delegation_marker", {}, NativeDelegationMarkerExecute);
+	// No bind callback — bind_data is supplied externally by GetScanFunctionImpl.
+	// No init_global / init_local — the marker should never be executed.
+	// filter_pushdown=true so DuckDB's FilterPushdown still installs filters
+	// on this LogicalGet's table_filters; the rewriter then hands them off to
+	// the real native function on the rewritten LogicalGet.
+	fn.filter_pushdown = true;
+	fn.projection_pushdown = true;
+	return fn;
+}
+
 VgiTableEntry::VgiTableEntry(Catalog &catalog, SchemaCatalogEntry &schema, CreateTableInfo &info,
                              const vgi::VgiTableInfo &table_info)
     : TableCatalogEntry(catalog, schema, info), table_info_(table_info), catalog_(catalog) {
@@ -482,6 +513,111 @@ TableFunction VgiTableEntry::GetScanFunctionImpl(ClientContext &context, unique_
 			    context, default_schema, scan_result.function_name, OnEntryNotFound::RETURN_NULL);
 		}
 	}
+	bool from_system_catalog = false;
+	if (!func_entry) {
+		// Last-resort fallback to the system catalog for built-in DuckDB
+		// table functions like `read_parquet` or `iceberg_scan` that workers
+		// declare via ScanFunctionResult. The multi-branch rewriter
+		// (vgi_multi_scan_rewriter.cpp:203) already does this fallback —
+		// mirror it for the single-branch path so workers can delegate scans
+		// to native DuckDB functions without going through a UNION ALL.
+		EntryLookupInfo lookup(CatalogType::TABLE_FUNCTION_ENTRY, scan_result.function_name);
+		auto sys_entry = Catalog::GetEntry(context, SYSTEM_CATALOG, DEFAULT_SCHEMA, lookup,
+		                                   OnEntryNotFound::RETURN_NULL);
+		if (sys_entry) {
+			func_entry = &sys_entry->Cast<TableFunctionCatalogEntry>();
+			from_system_catalog = true;
+		}
+	}
+
+	if (from_system_catalog && func_entry) {
+		// Native delegation: bind the system function eagerly here, then return
+		// a marker carrying the bound function + bind_data + return shapes.
+		// VgiRequiredFiltersOptimizer (vgi_extension.cpp) enforces this table's
+		// `required_field_filter_paths` against the LogicalGet's table_filters,
+		// then swaps `function` / `bind_data` / `returned_types` / `names` in
+		// place to the stashed native ones. Subsequent passes see a vanilla
+		// native scan. Matches VgiMultiScanRewriter's per-arm binding shape
+		// (vgi_multi_scan_rewriter.cpp:219-247) but for the single-branch path.
+		vector<LogicalType> arg_types;
+		arg_types.reserve(scan_result.positional_arguments.size());
+		for (const auto &v : scan_result.positional_arguments) {
+			arg_types.push_back(v.type());
+		}
+		TableFunction native_tf =
+		    func_entry->functions.GetFunctionByArguments(context, arg_types);
+		vector<Value> parameters(scan_result.positional_arguments.begin(),
+		                          scan_result.positional_arguments.end());
+		named_parameter_map_t named_parameters;
+		for (auto &kv : scan_result.named_arguments) {
+			named_parameters.emplace(kv.first, kv.second);
+		}
+		vector<LogicalType> input_table_types;
+		vector<string> input_table_names;
+		TableFunctionRef ref;
+		TableFunctionBindInput bind_input(parameters, named_parameters, input_table_types,
+		                                   input_table_names, native_tf.function_info.get(),
+		                                   nullptr, native_tf, ref);
+		vector<LogicalType> return_types;
+		vector<string> return_names;
+		auto native_bind = native_tf.bind(context, bind_input, return_types, return_names);
+		virtual_column_map_t native_virtual_columns;
+		if (native_tf.get_virtual_columns) {
+			native_virtual_columns = native_tf.get_virtual_columns(context, native_bind.get());
+		}
+
+		// Validate the catalog's declared columns match the native bind's
+		// output by position+name. The LogicalGet that DuckDB constructs uses
+		// the catalog's column list for FilterPushdown's column_ids /
+		// table_filters keys; if the native function emits a different shape,
+		// those indices mis-resolve once VgiRequiredFiltersOptimizer rewrites
+		// the marker. Two common causes:
+		//   - The worker's pa.Schema source omits Hive-partition columns that
+		//     the native bind appends (read_parquet on `theme=…/type=…/*`).
+		//   - The worker introspected against a different release than what
+		//     the URL points at.
+		// Either way the right move is to fail loudly here so the worker
+		// author sees the mismatch immediately instead of silent column
+		// misrouting at scan time.
+		{
+			const auto &decl_columns = GetColumns();
+			const auto decl_count = decl_columns.LogicalColumnCount();
+			if (decl_count != return_names.size()) {
+				throw BinderException(
+				    "VGI native delegation for '%s.%s.%s' (function '%s'): catalog declares "
+				    "%llu column(s) but the native bind returned %llu. The catalog's columns "
+				    "must match exactly what the native function emits at scan time (positions "
+				    "+ names). Common cause: Hive-partition columns that read_parquet appends "
+				    "but the worker's schema source omitted.",
+				    catalog_.GetName(), ParentSchema().name, name, scan_result.function_name,
+				    static_cast<unsigned long long>(decl_count),
+				    static_cast<unsigned long long>(return_names.size()));
+			}
+			for (idx_t i = 0; i < decl_count; ++i) {
+				const auto &decl_name = decl_columns.GetColumn(LogicalIndex(i)).Name();
+				if (decl_name != return_names[i]) {
+					throw BinderException(
+					    "VGI native delegation for '%s.%s.%s' (function '%s'): catalog "
+					    "declared column %llu as '%s' but the native bind returned '%s'. "
+					    "Names must match by position.",
+					    catalog_.GetName(), ParentSchema().name, name, scan_result.function_name,
+					    static_cast<unsigned long long>(i), decl_name, return_names[i]);
+				}
+			}
+		}
+
+		auto function_name_log = scan_result.function_name;
+		bind_data = make_uniq<VgiNativeDelegationMarkerBindData>(
+		    *this, std::move(native_tf), std::move(native_bind), std::move(return_types),
+		    std::move(return_names), std::move(native_virtual_columns), std::move(scan_result),
+		    attach_params->worker_path());
+
+		VGI_LOG(context, "vgi.scan_function.native_delegation_marker",
+		        {{"schema", ParentSchema().name},
+		         {"table", name},
+		         {"function", function_name_log}});
+		return MakeNativeDelegationMarkerFunction();
+	}
 	if (func_entry) {
 		for (auto &tf : func_entry->functions.functions) {
 			has_projection_pushdown = has_projection_pushdown || tf.projection_pushdown;

src/vgi_catalog_api.cpp

@@ -1903,6 +1903,13 @@ VgiTableInfo ParseTableInfo(ClientContext &context, const std::shared_ptr<arrow:
 		    bind_bytes.empty() ? std::nullopt : std::make_optional(std::move(bind_bytes));
 	}
 
+	// Parse required_field_filter_paths (optional, backward-compatible —
+	// missing column or empty list means no enforcement). The optimizer
+	// extension VgiRequiredFiltersOptimizer reads these from the cached
+	// VgiTableInfo at bind/optimize time.
+	info.required_field_filter_paths =
+	    row["required_field_filter_paths"].value_or(std::vector<std::string>{});
+
 	// Validate: UPDATE/DELETE require a row ID column
 	if ((info.supports_update || info.supports_delete) && info.row_id_column < 0) {
 		throw InvalidInputException(