fix(lint): green the ruff/mypy/pydoclint gates + refresh landing golden

The 0.16 scalar-cache + blended delta left the quality gates red:
- ruff: E501 line-length in protocol.py, D-series docstring nits.
- mypy: ScalarFunction lacked the CACHE_CONTROL attr (declared only on
  the *Generator* base but read off ScalarFunction subclasses in
  protocol.py); TableCardinality(estimate=1) missing the required max;
  cached_label declared LargeStringArray but returns a string() array.
- pydoclint: FunctionInfo/ResolvedMetadata/InitRequest gained
  input_from_args / substream_id without documenting them.
- landing describe.expected.json golden stale after the new fixtures.

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

Author: rustyconover

tests/landing/describe.expected.json

@@ -3,7 +3,7 @@
     {
       "attach_options": [],
       "counts": {
-        "functions": 188,
+        "functions": 199,
         "schemas": 2,
         "tables": 59,
         "views": 3
@@ -655,6 +655,50 @@
               "returns": "binary",
               "type": "scalar"
             },
+            {
+              "args": [
+                {
+                  "desc": "Value",
+                  "name": "value",
+                  "type": "int64"
+                },
+                {
+                  "desc": "Constant addend",
+                  "name": "addend",
+                  "type": "int64"
+                }
+              ],
+              "doc": "value + const addend (advertises vgi.cache.ttl)",
+              "name": "cached_add_const",
+              "returns": "int64",
+              "type": "scalar"
+            },
+            {
+              "args": [
+                {
+                  "desc": "Value to double",
+                  "name": "value",
+                  "type": "int64"
+                }
+              ],
+              "doc": "Doubles a BIGINT value (advertises vgi.cache.ttl for per-value memo)",
+              "name": "cached_double_scalar",
+              "returns": "int64",
+              "type": "scalar"
+            },
+            {
+              "args": [
+                {
+                  "desc": "Value",
+                  "name": "value",
+                  "type": "int64"
+                }
+              ],
+              "doc": "value -> 'lbl-<value>' or NULL for negatives (advertises vgi.cache.ttl)",
+              "name": "cached_label",
+              "returns": "string",
+              "type": "scalar"
+            },
             {
               "args": [
                 {
@@ -1259,6 +1303,18 @@
               "name": "blended_drop",
               "type": "table"
             },
+            {
+              "args": [
+                {
+                  "desc": "Fan-out count: emit rows 0..n-1 for this input row",
+                  "name": "n",
+                  "type": "int64"
+                }
+              ],
+              "doc": "Blended 1->N fan-out (emit 0..n-1 per input row) with row provenance",
+              "name": "blended_explode",
+              "type": "table"
+            },
             {
               "args": [
                 {
@@ -1563,6 +1619,42 @@
               "name": "cacheable_numbers",
               "type": "table"
             },
+            {
+              "args": [
+                {
+                  "desc": "Input column",
+                  "name": "x",
+                  "type": "int64"
+                }
+              ],
+              "doc": "Cacheable blended map x -> x*2 (advertises vgi.cache.ttl)",
+              "name": "cached_double",
+              "type": "table"
+            },
+            {
+              "args": [],
+              "doc": "Cacheable classic (TABLE-input) passthrough (advertises vgi.cache.ttl)",
+              "name": "cached_echo",
+              "type": "table"
+            },
+            {
+              "args": [
+                {
+                  "desc": "Input column",
+                  "name": "x",
+                  "type": "int64"
+                }
+              ],
+              "doc": "Blended map x->x*2 with always-revalidate (304 not_modified) contract",
+              "name": "cached_reval_double",
+              "type": "table"
+            },
+            {
+              "args": [],
+              "doc": "Classic passthrough with always-revalidate (304 not_modified) contract",
+              "name": "cached_reval_echo",
+              "type": "table"
+            },
             {
               "args": [],
               "doc": "Scan colors table (ENUM column)",
@@ -1885,6 +1977,25 @@
               "name": "geo_encode",
               "type": "table"
             },
+            {
+              "args": [
+                {
+                  "desc": "Input column (echoed as output)",
+                  "name": "x",
+                  "type": "int64"
+                },
+                {
+                  "default": "\"range\"",
+                  "desc": "range | length | base64",
+                  "name": "mode",
+                  "named": true,
+                  "type": "string"
+                }
+              ],
+              "doc": "Adversarial blended fixture emitting malformed vgi_rpc.parent_row",
+              "name": "hostile_provenance",
+              "type": "table"
+            },
             {
               "args": [
                 {
@@ -2302,6 +2413,18 @@
               "name": "profiling_demo",
               "type": "table"
             },
+            {
+              "args": [
+                {
+                  "desc": "Input column",
+                  "name": "x",
+                  "type": "int64"
+                }
+              ],
+              "doc": "Blended 1->1 map with projection_pushdown + two output columns",
+              "name": "projectable_blended",
+              "type": "table"
+            },
             {
               "args": [
                 {
@@ -2815,6 +2938,20 @@
               "name": "buffer_input",
               "type": "table_in_out"
             },
+            {
+              "args": [
+                {
+                  "default": "false",
+                  "desc": "Whether to log during processing",
+                  "name": "logging",
+                  "named": true,
+                  "type": "bool"
+                }
+              ],
+              "doc": "Cacheable column-wise sum across all input (advertises vgi.cache.ttl)",
+              "name": "cached_sum_all",
+              "type": "table_in_out"
+            },
             {
               "args": [],
               "doc": "Worker raises during combine (test)",

vgi/_test_fixtures/scalar/arithmetic.py

@@ -202,7 +202,7 @@ class Meta:
     def compute(
         cls,
         value: Annotated[pa.Int64Array, Param(doc="Value")],
-    ) -> Annotated[pa.LargeStringArray, Returns(pa.string())]:
+    ) -> Annotated[pa.StringArray, Returns(pa.string())]:
         """Label non-negative values; NULL for negatives."""
         labels = ["lbl-" + str(v.as_py()) if (v.is_valid and v.as_py() >= 0) else None for v in value]
         return pa.array(labels, type=pa.string())

vgi/_test_fixtures/table_in_out.py

@@ -155,8 +155,8 @@ class Meta:
 
     @classmethod
     def cardinality(cls, params: BindParams[SingleTableArguments]) -> TableCardinality:
-        # One row per substream; unknown up-front, so leave the estimate open.
-        return TableCardinality(estimate=1)
+        # One row per substream; the substream count is unknown up-front.
+        return TableCardinality(estimate=1, max=None)
 
     @classmethod
     def on_bind(cls, params: BindParams[SingleTableArguments]) -> BindResponse:
@@ -1586,9 +1586,7 @@ def process(
         lats = batch.column("latitude").to_pylist()
         lons = batch.column("longitude").to_pylist()
         codes = [
-            None
-            if lat is None or lon is None
-            else f"{round(lat, precision)}:{round(lon, precision)}"
+            None if lat is None or lon is None else f"{round(lat, precision)}:{round(lon, precision)}"
             for lat, lon in zip(lats, lons, strict=True)
         ]
         out.emit(pa.record_batch({"geohash": pa.array(codes, type=pa.string())}))
@@ -1638,9 +1636,7 @@ def process(
         lons = batch.column("longitude").to_pylist()
         alts = batch.column("altitude").to_pylist()
         codes = [
-            None
-            if lat is None or lon is None or alt is None
-            else f"{round(lat, p)}:{round(lon, p)}:{round(alt, p)}"
+            None if lat is None or lon is None or alt is None else f"{round(lat, p)}:{round(lon, p)}:{round(alt, p)}"
             for lat, lon, alt in zip(lats, lons, alts, strict=True)
         ]
         out.emit(pa.record_batch({"geohash": pa.array(codes, type=pa.string())}))
@@ -1994,9 +1990,9 @@ class Meta(SumAllColumnsFunction.Meta):
     @classmethod
     def finalize(
         cls,
-        params: "TableBufferingParams[SumAllColumnsFunctionArguments]",
+        params: TableBufferingParams[SumAllColumnsFunctionArguments],
         finalize_state_id: bytes,
-        state: "_LogDrainState",
+        state: _LogDrainState,
         out: OutputCollector,
     ) -> None:
         rows = params.storage.state_log_scan(state.ns, b"", after_id=state.after_id, limit=1)

vgi/_test_fixtures/worker.py

@@ -75,11 +75,11 @@
     AnyMixedStrFunction,
     BernoulliFunction,
     BinaryPacketFunction,
-    ConcatValuesIntFunction,
-    ConcatValuesStrFunction,
     CachedAddConstScalarFunction,
     CachedDoubleScalarFunction,
     CachedLabelScalarFunction,
+    ConcatValuesIntFunction,
+    ConcatValuesStrFunction,
     ConditionalMessageFunction,
     DoubleFunction,
     FormatNumberDefaultFunction,
@@ -226,34 +226,34 @@
 )
 from vgi._test_fixtures.table_in_out import (
     BatchIndexBufferInputFunction,
+    BlendedDropFunction,
+    BlendedExplodeFunction,
     BufferEmitWideFunction,
     BufferInputFunction,
-    CrashOnCombineFunction,
-    CrashOnFinalizeFunction,
-    CrashOnProcessFunction,
-    BlendedExplodeFunction,
     CachedDoubleFunction,
     CachedEchoFunction,
-    CachedRevalidatingEchoFunction,
     CachedRevalidatingDoubleFunction,
+    CachedRevalidatingEchoFunction,
     CachedSumAllColumnsFunction,
-    HostileProvenanceFunction,
-    ProjectableBlendedFunction,
+    CrashOnCombineFunction,
+    CrashOnFinalizeFunction,
+    CrashOnProcessFunction,
     EchoBufferingFunction,
     EchoFunction,
-    GeoEncode3Function,
-    GeoEncodeFunction,
-    RowSumFunction,
-    BlendedDropFunction,
     EchoWitnessFunction,
     ExceptionFinalizeFunction,
     ExceptionProcessFunction,
     FilterBySettingFunction,
+    GeoEncode3Function,
+    GeoEncodeFunction,
     HangOnProcessFunction,
+    HostileProvenanceFunction,
     LargeStateFunction,
     OrderedBufferInputFunction,
     OrderedSourceFunction,
+    ProjectableBlendedFunction,
     RepeatInputsFunction,
+    RowSumFunction,
     SecretInOutFunction,
     SubstreamPartialSumFunction,
     SumAllColumnsFunction,

vgi/catalog/catalog_interface.py

@@ -695,6 +695,13 @@ class FunctionInfo(CatalogSchemaObject, ArrowSerializableDataclass):
             carries a globally-unique monotonic batch_index from DuckDB's source.
             Workers can sort by it in combine() to reconstruct source order under
             parallel ingest. Mutually exclusive with ``sink_order_dependent``.
+        input_from_args: True for a blended ``RowTransformFunction`` — its
+            positional args ARE its per-row input columns (real typed args, no
+            ``TABLE`` placeholder), so one registration serves the literal /
+            column / LATERAL call shapes. Set from ``RowTransformFunction``
+            subclassing; the C++ extension reads it to enter the in-out
+            registration branch with real-typed args and drive the literal
+            single-row scan-mode.
         required_settings: Settings required by the function.
         required_secrets: Secrets required by the function (each entry has
             secret_type, optional secret_name, optional scope).

vgi/metadata.py

@@ -408,6 +408,10 @@ class ResolvedMetadata:
             BatchIndex()`` so each process() call carries a globally-unique
             monotonic batch_index for reconstructing source order under parallel
             ingest. Mutually exclusive with ``sink_order_dependent``.
+        input_from_args: True for a blended ``RowTransformFunction`` — its
+            positional args ARE its per-row input columns, so one registration
+            serves the literal / column / LATERAL call shapes. Detected from
+            ``RowTransformFunction`` subclassing.
 
     """
 

vgi/protocol.py

@@ -258,6 +258,13 @@ class InitRequest(ArrowSerializableDataclass):
             ``None`` for global init, set for secondary init.
         init_opaque_data: Opaque per-init state threaded back into the stream;
             same producer/wire/consumer contract as ``bind_opaque_data``.
+        substream_id: Stable, client-owned identity of the streaming
+            table-in-out substream this request opens (per DuckDB
+            PipelineExecutor). Rides every InitRequest a substream's worker
+            builds — INPUT and FINALIZE — so a worker can locate that
+            substream's accumulated state even when an HTTP load balancer
+            routes its init / process / finalize to different backends. ``None``
+            for non-substream function types.
         order_by_column_name: Column name of the order-pushdown hint from DuckDB's
             RowGroupPruner optimizer; ``None`` when no hint.
         order_by_direction: Sort direction of the order-pushdown hint; ``None``
@@ -789,7 +796,7 @@ def exchange(self, input: AnnotatedBatch, out: OutputCollector, ctx: CallContext
         # Result-cache opt-in: a scalar declaring CACHE_CONTROL rides its vgi.cache.* keys on
         # the emit path's batch custom_metadata (per-batch — not the fixed IPC schema), so the
         # extension can memoize the output per distinct input value.
-        cc_meta = cls.CACHE_CONTROL.to_metadata() if getattr(cls, "CACHE_CONTROL", None) is not None else None
+        cc_meta = cls.CACHE_CONTROL.to_metadata() if cls.CACHE_CONTROL is not None else None
         out.emit(output, metadata=cc_meta)
 
 
@@ -939,7 +946,9 @@ def _serialize_partition_values_batch(batch: pa.RecordBatch) -> str:
     sink = pa.BufferOutputStream()
     with pa.ipc.new_stream(sink, batch.schema) as writer:
         writer.write_batch(batch)
-    return base64.b64encode(sink.getvalue().to_pybytes()).decode("ascii")
+    # b64encode accepts buffer-protocol objects — no need to copy the
+    # pa.Buffer into an intermediate bytes first.
+    return base64.b64encode(sink.getvalue()).decode("ascii")
 
 
 def _merge_partition_values(
@@ -1351,7 +1360,9 @@ def rehydrate(self, implementation: object) -> None:
         proj_ids = _effective_projection_ids(func_cls, self._init_call.projection_ids)
         output_schema = project_schema(proj_ids, self._init_call.output_schema)
         self._params = ProcessParams(
-            args=func_cls._parse_arguments(func_cls.FunctionArguments, self._init_call.bind_call.arguments, blended=func_cls._is_blended()),
+            args=func_cls._parse_arguments(
+                func_cls.FunctionArguments, self._init_call.bind_call.arguments, blended=func_cls._is_blended()
+            ),
             init_call=self._init_call,
             init_response=self._init_response,
             output_schema=output_schema,
@@ -1527,7 +1538,9 @@ def rehydrate(self, implementation: object) -> None:
         proj_ids = _effective_projection_ids(func_cls, self._init_call.projection_ids)
         output_schema = project_schema(proj_ids, self._init_call.output_schema)
         self._params = ProcessParams(
-            args=func_cls._parse_arguments(func_cls.FunctionArguments, self._init_call.bind_call.arguments, blended=func_cls._is_blended()),
+            args=func_cls._parse_arguments(
+                func_cls.FunctionArguments, self._init_call.bind_call.arguments, blended=func_cls._is_blended()
+            ),
             init_call=self._init_call,
             init_response=self._init_response,
             output_schema=output_schema,