Add non-filter VGI integration parity fixtures

example-worker/CMakeLists.txt

@@ -16,6 +16,7 @@ add_executable(vgi-example-worker
     scalar/secrets.cpp
     scalar/cached.cpp
     table/sequence.cpp
+    table/same_name.cpp
     table/cache.cpp
     table/more.cpp
     table/generators.cpp

example-worker/buffering/buffer_input.cpp

@@ -3,9 +3,9 @@
 
 #include <algorithm>
 #include <chrono>
-#include <thread>
 #include <memory>
 #include <string>
+#include <thread>
 #include <vector>
 
 #include <arrow/array.h>
@@ -167,8 +167,8 @@ class BufferInput : public vgi::TableBufferingFunction {
                 (void)id;
                 ordered.push_back(decode_indexed(blob));
             }
-            std::sort(ordered.begin(), ordered.end(),
-                      [](const auto& a, const auto& b) { return a.first < b.first; });
+            std::stable_sort(ordered.begin(), ordered.end(),
+                             [](const auto& a, const auto& b) { return a.first < b.first; });
             for (const auto& [index, bytes] : ordered) {
                 (void)index;
                 params.storage->append(params.execution_id, kNamespace, "", bytes);

example-worker/catalog_def.cpp

@@ -70,6 +70,15 @@ vgi::CatalogBranch sequence_branch(int64_t count,
     return branch;
 }
 
+vgi::CatalogBranch split_sequence_branch(int64_t count, int64_t splits) {
+    vgi::CatalogBranch branch;
+    branch.function_name = "split_sequence";
+    branch.scan_arguments =
+        vgi::serialize_scan_arguments({}, {{"n", int64_arg(count)}, {"splits", int64_arg(splits)}});
+    branch.schema_path = vgi::SchemaPath{"data"};
+    return branch;
+}
+
 vgi::CatalogTable multi_branch(std::string name, std::vector<vgi::CatalogBranch> branches,
                                std::string comment = {}) {
     vgi::CatalogTable table;
@@ -391,6 +400,10 @@ void declare_catalog(vgi::Worker& worker) {
     data.tables.push_back(multi_branch("multi_branch_empty", {},
                                        "Multi-branch: empty branches list — used by "
                                        "multi_branch_empty_branches.test"));
+    data.tables.push_back(
+        multi_branch("multi_branch_split", {split_sequence_branch(30, 6), sequence_branch(20)},
+                     "Multi-branch: split_sequence(30, splits=6) + sequence(20) — used by "
+                     "splits/multi_branch.test"));
 
     // Heterogeneous branches: one arm is this worker, the others are DuckDB's
     // own readers over files the test writes first. What they probe is that a
@@ -526,6 +539,12 @@ void declare_catalog(vgi::Worker& worker) {
     // and the suite names `main` for these two.
     auto& main = worker.catalog().schema("main");
     main.comment = "Example functions for testing VGI";
+    main.tables.push_back(backed_by("test_same_name_table", "test_same_name_table_scan",
+                                    columns({{"tag", arrow::utf8()}}),
+                                    "Schema-disambiguation probe; the main-schema table"));
+    data.tables.push_back(backed_by("test_same_name_table", "test_same_name_table_scan",
+                                    columns({{"tag", arrow::utf8()}}),
+                                    "Schema-disambiguation probe; the data-schema table"));
     // Macros never reach the worker at run time — the engine substitutes the
     // text — so declaring them is the whole implementation.
     main.macros.push_back({"vgi_multiply",

example-worker/main.cpp

@@ -62,6 +62,7 @@ int main(int argc, char** argv) {
     example::register_secret_fixtures(worker);
     example::register_series(worker);
     example::register_splits(worker);
+    if (composite) example::register_same_name_tables(worker);
     example::register_filter_fixtures(worker);
     example::register_logging_fixtures(worker);
     example::register_global_probes(worker);

example-worker/registry.h

@@ -36,6 +36,7 @@ void register_settings_tables(vgi::Worker& worker);
 void register_secret_fixtures(vgi::Worker& worker);
 void register_series(vgi::Worker& worker);
 void register_splits(vgi::Worker& worker);
+void register_same_name_tables(vgi::Worker& worker);
 void register_filter_fixtures(vgi::Worker& worker);
 void register_logging_fixtures(vgi::Worker& worker);
 void register_global_probes(vgi::Worker& worker);

example-worker/table/same_name.cpp

@@ -0,0 +1,70 @@
+// © Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+// Two identically-named scans in different schemas. Each backs a declarative
+// table in its own schema, so dispatch has to retain the full schema path.
+
+#include <memory>
+#include <string>
+
+#include <arrow/array.h>
+#include <arrow/array/builder_binary.h>
+#include <arrow/record_batch.h>
+
+#include <vgi/worker.h>
+
+namespace example {
+namespace {
+
+class SameNameTableScan : public vgi::TableFunction {
+public:
+    explicit SameNameTableScan(std::string schema) : schema_(std::move(schema)) {}
+
+    std::string name() const override { return "test_same_name_table_scan"; }
+
+    vgi::FunctionMetadata metadata() const override {
+        vgi::FunctionMetadata md;
+        md.description = "Schema-disambiguation probe; the " + schema_ + "-schema producer";
+        md.categories = {"generator", "testing"};
+        return md;
+    }
+
+    std::vector<vgi::ArgSpec> argument_specs() const override { return {}; }
+
+    std::shared_ptr<arrow::Schema> bind(const vgi::BindParams&) const override {
+        return arrow::schema({arrow::field("tag", arrow::utf8(), /*nullable=*/true)});
+    }
+
+    std::unique_ptr<vgi::TableProducer> init(const vgi::ProcessParams& params) const override {
+        arrow::StringBuilder builder;
+        (void)builder.Append(schema_);
+        std::shared_ptr<arrow::Array> tag;
+        (void)builder.Finish(&tag);
+        return std::make_unique<Producer>(arrow::RecordBatch::Make(params.output_schema, 1, {tag}));
+    }
+
+private:
+    class Producer : public vgi::TableProducer {
+    public:
+        explicit Producer(std::shared_ptr<arrow::RecordBatch> batch) : batch_(std::move(batch)) {}
+
+        std::shared_ptr<arrow::RecordBatch> next_batch() override {
+            auto result = batch_;
+            batch_ = nullptr;
+            return result;
+        }
+
+    private:
+        std::shared_ptr<arrow::RecordBatch> batch_;
+    };
+
+    std::string schema_;
+};
+
+}  // namespace
+
+void register_same_name_tables(vgi::Worker& worker) {
+    worker.register_table_in("example", "main", std::make_shared<SameNameTableScan>("main"));
+    worker.register_table_in("example", "data", std::make_shared<SameNameTableScan>("data"));
+}
+
+}  // namespace example