Skip to content

Commit 43974b5

Browse files
rustyconoverclaude
andcommitted
fix(catalog): forward Table.arguments to the scan-time worker bind
A function-backed catalog data-table declared as Table(function=F, arguments=Arguments(positional=(pa.scalar(7),))) derived its schema correctly but scanned with zero arguments. When F had a required positional Arg(0), the scan failed at bind with: IndexError: Argument 0: index out of range (have 0 positional arguments) Table.arguments was consumed only at catalog-build time for schema derivation (descriptors._get_resolved_columns -> F.bind()) and never threaded into the runtime ScanFunctionResult that drives the scan. Two sites hardcoded empties: - descriptors._inline_function_result -- the inline path, always used by the declarative to_table_info; the C++ extension reads TableInfo.scan_function verbatim and skips the RPC entirely. - ReadOnlyCatalogInterface.table_scan_function_get -- the RPC auto-impl. The default table_scan_branches_get delegates to it, so this also fixes the multi-branch path. The VGI DuckDB extension needed no change: its catalog scan path is symmetric with the direct vgi_table_function() path, and DecodeScanArguments already routes the arg_<N> fields that ScanFunctionResult.to_row_dict emits. Adds scan_arguments_from(), shared by both sites. It rejects a None positional slot rather than skipping it -- dropping one would shift every later argument down an index and silently bind the wrong values. Only the scan function is parameterized by Table.arguments; insert/update/delete receive their rows via the streaming table input, so their inlined results stay argument-free. Backward compatible: a table declaring no arguments still serializes empty. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 44a3b60 commit 43974b5

3 files changed

Lines changed: 255 additions & 16 deletions

File tree

‎tests/catalog/test_declarative.py‎

Lines changed: 187 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,13 +14,13 @@
1414
from __future__ import annotations
1515

1616
from dataclasses import dataclass
17-
from typing import ClassVar
17+
from typing import Annotated, ClassVar
1818

1919
import pyarrow as pa
2020
import pytest
2121
from vgi_rpc.rpc import OutputCollector
2222

23-
from vgi import Worker
23+
from vgi import Arg, Arguments, Worker
2424
from vgi.catalog import (
2525
AttachOpaqueData,
2626
Catalog,
@@ -50,6 +50,7 @@
5050
bind_fixed_schema,
5151
init_single_worker,
5252
)
53+
from vgi.table_in_out_function import TableInOutGenerator
5354

5455
# =============================================================================
5556
# Test Fixtures: Example Functions for Function-Backed Tables
@@ -61,6 +62,45 @@ class EmptyArgs:
6162
"""No arguments."""
6263

6364

65+
@dataclass(frozen=True)
66+
class CountedRowsArgs:
67+
"""A required positional argument plus an optional named one."""
68+
69+
count: Annotated[int, Arg(0, doc="Number of rows to generate", ge=0)]
70+
label: Annotated[str, Arg("label", default="rows", doc="Label for the rows")]
71+
72+
73+
@init_single_worker
74+
@bind_fixed_schema
75+
class CountedRowsFunction(TableFunctionGenerator[CountedRowsArgs]):
76+
"""Table function with a required positional arg, for arg-propagation tests."""
77+
78+
class Meta: # noqa: D106
79+
name = "counted_rows"
80+
description = "Generate N rows"
81+
82+
FIXED_SCHEMA: ClassVar[pa.Schema] = schema({"n": pa.int64()})
83+
84+
@classmethod
85+
def process( # noqa: D102
86+
cls, params: ProcessParams[CountedRowsArgs], state: None, out: OutputCollector
87+
) -> None:
88+
out.emit(
89+
pa.RecordBatch.from_pydict(
90+
{"n": list(range(params.args.count))},
91+
schema=params.output_schema,
92+
)
93+
)
94+
out.finish()
95+
96+
97+
class CountedRowsInsert(TableInOutGenerator[None, None]):
98+
"""Stub insert function, for the write-path arg-propagation test."""
99+
100+
class Meta: # noqa: D106
101+
name = "counted_rows_insert"
102+
103+
64104
@init_single_worker
65105
@bind_fixed_schema
66106
class UsersFunction(TableFunctionGenerator[EmptyArgs]):
@@ -262,6 +302,87 @@ def test_function_backed_table_inlines_scan_function(self) -> None:
262302
assert info.update_function is None
263303
assert info.delete_function is None
264304

305+
def test_inlined_scan_function_carries_declared_positional_arguments(self) -> None:
306+
"""``Table.arguments`` must reach the scan-time bind, not just ``bind()``.
307+
308+
Regression: the inlined ``ScanFunctionResult`` used to hardcode empty
309+
arguments, so a backing function with a required positional ``Arg(0)``
310+
saw zero positional args at scan and failed with
311+
``IndexError: Argument 0: index out of range``.
312+
"""
313+
from vgi_rpc.utils import deserialize_record_batch
314+
315+
table = Table(
316+
name="counted",
317+
function=CountedRowsFunction,
318+
arguments=Arguments(positional=(pa.scalar(7),)),
319+
)
320+
info = table.to_table_info("main")
321+
assert info.scan_function is not None
322+
batch, _ = deserialize_record_batch(info.scan_function)
323+
sfr = ScanFunctionResult.deserialize(batch)
324+
assert sfr.function_name == "counted_rows"
325+
assert [s.as_py() for s in sfr.positional_arguments] == [7]
326+
assert sfr.named_arguments == {}
327+
328+
def test_inlined_scan_function_carries_declared_named_arguments(self) -> None:
329+
"""Named args declared on the Table also reach the scan-time bind."""
330+
from vgi_rpc.utils import deserialize_record_batch
331+
332+
table = Table(
333+
name="counted",
334+
function=CountedRowsFunction,
335+
arguments=Arguments(
336+
positional=(pa.scalar(3),),
337+
named={"label": pa.scalar("widgets")},
338+
),
339+
)
340+
info = table.to_table_info("main")
341+
assert info.scan_function is not None
342+
batch, _ = deserialize_record_batch(info.scan_function)
343+
sfr = ScanFunctionResult.deserialize(batch)
344+
assert [s.as_py() for s in sfr.positional_arguments] == [3]
345+
assert {k: v.as_py() for k, v in sfr.named_arguments.items()} == {"label": "widgets"}
346+
347+
def test_inlined_write_functions_ignore_table_arguments(self) -> None:
348+
"""Only the *scan* function is parameterized by ``Table.arguments``.
349+
350+
Insert/update/delete functions receive their rows via the streaming table
351+
input, so their inlined results stay argument-free.
352+
"""
353+
from vgi_rpc.utils import deserialize_record_batch
354+
355+
table = Table(
356+
name="counted",
357+
function=CountedRowsFunction,
358+
arguments=Arguments(positional=(pa.scalar(7),)),
359+
insert_function=CountedRowsInsert,
360+
)
361+
info = table.to_table_info("main")
362+
assert info.insert_function is not None
363+
batch, _ = deserialize_record_batch(info.insert_function)
364+
sfr = ScanFunctionResult.deserialize(batch)
365+
assert sfr.positional_arguments == []
366+
assert sfr.named_arguments == {}
367+
368+
def test_none_positional_slot_is_rejected(self) -> None:
369+
"""A ``None`` positional slot would shift later args — reject it loudly.
370+
371+
Exercised on the ``scan_arguments_from`` helper directly: a Table carrying
372+
such a slot already fails earlier, in bind()-based schema derivation. This
373+
guards the serializer against ever silently dropping the slot instead.
374+
"""
375+
from vgi.catalog.catalog_interface import scan_arguments_from
376+
377+
with pytest.raises(ValueError, match="positional slot 0 is None"):
378+
scan_arguments_from(Arguments(positional=(None, pa.scalar(7))))
379+
380+
def test_scan_arguments_from_none_is_empty(self) -> None:
381+
"""A table declaring no arguments serializes empty args (unchanged behavior)."""
382+
from vgi.catalog.catalog_interface import scan_arguments_from
383+
384+
assert scan_arguments_from(None) == ([], {})
385+
265386
def test_table_propagates_cardinality_to_table_info(self) -> None:
266387
"""``Table.cardinality_estimate`` / ``cardinality_max`` flow into TableInfo.
267388
@@ -1121,8 +1242,6 @@ def test_estimated_object_count_partitions_functions_by_type(self) -> None:
11211242
client hits a cold miss. Reporting an explicit ``0`` for empty
11221243
categories lets the client skip the bulk RPC entirely.
11231244
"""
1124-
from typing import Annotated
1125-
11261245
from vgi.arguments import Param, Returns
11271246
from vgi.scalar_function import ScalarFunction
11281247

@@ -1174,8 +1293,6 @@ class TestLegacyFunctionsListSchemaInfo:
11741293
"""
11751294

11761295
def _interface(self) -> ReadOnlyCatalogInterface:
1177-
from typing import Annotated
1178-
11791296
from vgi.arguments import Param, Returns
11801297
from vgi.scalar_function import ScalarFunction
11811298

@@ -1694,6 +1811,70 @@ def test_function_backed_table_auto_scan(self, catalog_interface: ReadOnlyCatalo
16941811
)
16951812
assert isinstance(result, ScanFunctionResult)
16961813
assert result.function_name == "users"
1814+
# No declared Table.arguments => empty args, unchanged behavior.
1815+
assert result.positional_arguments == []
1816+
assert result.named_arguments == {}
1817+
1818+
def test_auto_scan_forwards_declared_arguments(self) -> None:
1819+
"""The RPC auto-impl forwards ``Table.arguments`` to the scan bind.
1820+
1821+
Companion to the inlined-path regression: workers that don't inline
1822+
``TableInfo.scan_function`` reach the scan through this RPC instead.
1823+
"""
1824+
counted = Table(
1825+
name="counted",
1826+
function=CountedRowsFunction,
1827+
arguments=Arguments(
1828+
positional=(pa.scalar(7),),
1829+
named={"label": pa.scalar("widgets")},
1830+
),
1831+
)
1832+
1833+
class TestCatalog(ReadOnlyCatalogInterface):
1834+
catalog = Catalog(
1835+
name="test",
1836+
schemas=[Schema(name="main", tables=[counted])],
1837+
)
1838+
1839+
result = TestCatalog().table_scan_function_get(
1840+
attach_opaque_data=AttachOpaqueData(b"test"),
1841+
transaction_opaque_data=None,
1842+
schema_name="main",
1843+
name="counted",
1844+
at_unit=None,
1845+
at_value=None,
1846+
)
1847+
assert result.function_name == "counted_rows"
1848+
assert [s.as_py() for s in result.positional_arguments] == [7]
1849+
assert {k: v.as_py() for k, v in result.named_arguments.items()} == {"label": "widgets"}
1850+
1851+
def test_scan_branches_default_inherits_declared_arguments(self) -> None:
1852+
"""The default ``table_scan_branches_get`` delegates to the legacy method.
1853+
1854+
So the one-branch synthesis must carry the declared arguments too.
1855+
"""
1856+
counted = Table(
1857+
name="counted",
1858+
function=CountedRowsFunction,
1859+
arguments=Arguments(positional=(pa.scalar(7),)),
1860+
)
1861+
1862+
class TestCatalog(ReadOnlyCatalogInterface):
1863+
catalog = Catalog(
1864+
name="test",
1865+
schemas=[Schema(name="main", tables=[counted])],
1866+
)
1867+
1868+
result = TestCatalog().table_scan_branches_get(
1869+
attach_opaque_data=AttachOpaqueData(b"test"),
1870+
transaction_opaque_data=None,
1871+
schema_name="main",
1872+
name="counted",
1873+
at_unit=None,
1874+
at_value=None,
1875+
)
1876+
assert len(result.branches) == 1
1877+
assert [s.as_py() for s in result.branches[0].positional_arguments] == [7]
16971878

16981879
def test_explicit_columns_table_raises(self) -> None:
16991880
"""Tables with explicit columns raise NotImplementedError."""

‎vgi/catalog/catalog_interface.py‎

Lines changed: 52 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@
3737
from vgi_rpc import ArrowSerializableDataclass, ArrowType
3838
from vgi_rpc.utils import deserialize_record_batch, serialize_record_batch_bytes
3939

40-
from vgi.arguments import SecretLookupEntry
40+
from vgi.arguments import Arguments, SecretLookupEntry
4141
from vgi.exceptions import CatalogReadOnlyError
4242
from vgi.metadata import (
4343
DistinctDependence,
@@ -886,6 +886,50 @@ def deserialize(cls, batch: pa.RecordBatch) -> Self:
886886
WriteFunctionResult = ScanFunctionResult
887887

888888

889+
def scan_arguments_from(
890+
arguments: Arguments | None,
891+
) -> tuple[list[pa.Scalar], dict[str, pa.Scalar]]: # type: ignore[type-arg]
892+
"""Split a table's declared [`Arguments`][] into `ScanFunctionResult` shape.
893+
894+
A function-backed ``Table`` declares the arguments its backing function needs
895+
via ``Table.arguments``. Those arguments must reach the *scan-time* worker bind,
896+
not just the catalog-build-time schema derivation — otherwise a backing function
897+
with a required positional ``Arg(0)`` sees zero positional arguments at scan.
898+
899+
``ScanFunctionResult.to_row_dict`` serializes ``positional_arguments`` as the
900+
``arg_<N>`` wire fields the VGI DuckDB extension's ``DecodeScanArguments``
901+
expects, so returning them here is all that is required.
902+
903+
Args:
904+
arguments: The table's declared arguments, or ``None`` when it declares none.
905+
906+
Returns:
907+
``(positional_arguments, named_arguments)`` ready for `ScanFunctionResult`.
908+
``([], {})`` when ``arguments`` is ``None``.
909+
910+
Raises:
911+
ValueError: If a positional slot is ``None``. Dropping it would shift every
912+
later argument down one index and silently bind the wrong values, so an
913+
unfilled slot is rejected rather than skipped.
914+
915+
"""
916+
if arguments is None:
917+
return [], {}
918+
positional: list[pa.Scalar] = [] # type: ignore[type-arg]
919+
for index, scalar in enumerate(arguments.positional):
920+
if scalar is None:
921+
raise ValueError(
922+
f"Table arguments: positional slot {index} is None. Positional "
923+
f"arguments passed to a function-backed table must all be concrete "
924+
f"pa.Scalar values — an unfilled slot would shift the arguments after "
925+
f'it. To make an argument optional, declare it as a named argument '
926+
f'(Arg("name", default=...)) instead.'
927+
)
928+
positional.append(scalar)
929+
named = dict(arguments.named) if arguments.named else {}
930+
return positional, named
931+
932+
889933
# ============================================================================
890934
# Multi-branch scan (catalog_table_scan_branches_get)
891935
# ============================================================================
@@ -2704,12 +2748,16 @@ def table_scan_function_get(
27042748

27052749
# Check if table exists and is function-backed
27062750
if table is not None and table.function is not None:
2707-
# Auto-implement for function-backed tables
2751+
# Auto-implement for function-backed tables. The table's declared
2752+
# ``arguments`` are forwarded so the scan-time bind receives the same
2753+
# arguments the catalog-build-time schema derivation used; a backing
2754+
# function with a required positional ``Arg(0)`` depends on this.
27082755
func_meta = table.function.get_metadata()
2756+
positional_arguments, named_arguments = scan_arguments_from(table.arguments)
27092757
return ScanFunctionResult(
27102758
function_name=func_meta.name,
2711-
positional_arguments=[],
2712-
named_arguments={},
2759+
positional_arguments=positional_arguments,
2760+
named_arguments=named_arguments,
27132761
required_extensions=[],
27142762
)
27152763

‎vgi/catalog/descriptors.py‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
TableColumnStatisticsResult,
3434
TableInfo,
3535
ViewInfo,
36+
scan_arguments_from,
3637
serialize_column_statistics,
3738
)
3839
from vgi.invocation import BindResponse, FunctionType
@@ -171,22 +172,31 @@ def _default_to_sql(value: DefaultValue) -> str:
171172

172173
def _inline_function_result(
173174
func: type[Function] | None,
175+
arguments: Arguments | None = None,
174176
) -> bytes | None:
175177
"""Build inlined ``ScanFunctionResult`` IPC bytes for a function-backed table.
176178
177179
Returns ``None`` when the table is not function-backed for that operation.
178180
Mirrors ``ReadOnlyCatalogInterface._write_function_get`` /
179-
``table_scan_function_get`` auto-impl: empty positional/named arguments,
180-
no required extensions. The C++ extension uses these bytes verbatim and
181-
skips the corresponding ``catalog_table_*_function_get`` RPC.
181+
``table_scan_function_get`` auto-impl: no required extensions, and the
182+
table's declared ``arguments`` (if any) forwarded as the function's
183+
positional/named arguments. The C++ extension uses these bytes verbatim and
184+
skips the corresponding ``catalog_table_*_function_get`` RPC — so omitting
185+
the arguments here means the scan-time worker bind sees none, which breaks
186+
any backing function with a required positional ``Arg(0)``.
187+
188+
``arguments`` defaults to ``None`` (empty args) because only the *scan*
189+
function is parameterized by ``Table.arguments``; the insert/update/delete
190+
functions receive their rows via the streaming table input instead.
182191
"""
183192
if func is None:
184193
return None
185194
func_meta = func.get_metadata()
195+
positional_arguments, named_arguments = scan_arguments_from(arguments)
186196
return ScanFunctionResult(
187197
function_name=func_meta.name,
188-
positional_arguments=[],
189-
named_arguments={},
198+
positional_arguments=positional_arguments,
199+
named_arguments=named_arguments,
190200
required_extensions=[],
191201
).serialize()
192202

@@ -633,7 +643,7 @@ def to_table_info(self, schema_name: str) -> TableInfo:
633643
supports_column_statistics=bool(self.statistics),
634644
comment=self.comment,
635645
tags=dict(self.tags),
636-
scan_function=_inline_function_result(self.function),
646+
scan_function=_inline_function_result(self.function, self.arguments),
637647
insert_function=_inline_function_result(self.insert_function),
638648
update_function=_inline_function_result(self.update_function),
639649
delete_function=_inline_function_result(self.delete_function),

0 commit comments

Comments
 (0)