Bump version to 0.1.15 and add cast-compatible exchange conformance tests

Add exchange_cast_compatible method and tests verifying server-side
RecordBatch.cast() for compatible schema mismatches (int32/int64/float32
to float64) and rejection of incompatible column names.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rustyconover

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "vgi-rpc"
-version = "0.1.14"
+version = "0.1.15"
 description = "Vector Gateway Interface - RPC framework based on Apache Arrow"
 readme = "README.md"
 requires-python = ">=3.13"

uv.lock

@@ -307,6 +307,14 @@ def exchange_error_on_nth(self, fail_on: int) -> Stream[FailOnExchangeNState]:
             input_schema=_SCALE_INPUT_SCHEMA,
         )
 
+    def exchange_cast_compatible(self) -> Stream[ScaleExchangeState]:
+        """Exchange expecting float64 input — echoes values via factor=1.0."""
+        return Stream(
+            output_schema=_SCALE_OUTPUT_SCHEMA,
+            state=ScaleExchangeState(factor=1.0),
+            input_schema=_SCALE_INPUT_SCHEMA,
+        )
+
     def exchange_error_on_init(self) -> Stream[ScaleExchangeState]:
         """Raise during exchange init."""
         raise RuntimeError("intentional exchange init error")

vgi_rpc/conformance/_impl.py

@@ -3,7 +3,7 @@
 
 """ConformanceService Protocol definition.
 
-Defines ~44 RPC methods covering every framework capability:
+Defines ~45 RPC methods covering every framework capability:
 scalar echo, void, complex types, optionals, dataclass round-trip,
 annotated types, multi-param, errors, logging, producer streams,
 exchange streams, headers, and introspection.
@@ -296,6 +296,10 @@ def produce_dynamic_schema(
         """
         ...
 
+    def exchange_cast_compatible(self) -> Stream[StreamState]:
+        """Exchange expecting float64 input — tests server-side cast for compatible schemas."""
+        ...
+
     def exchange_with_rich_header(self, seed: int, factor: float) -> Stream[StreamState, RichHeader]:
         """Exchange stream with a rich multi-type header.
 

vgi_rpc/conformance/_protocol.py

@@ -834,6 +834,76 @@ def test_zero_row_input(self, conformance_conn: ConnFactory) -> None:
             assert out.batch.num_rows == 0
 
 
+# ---------------------------------------------------------------------------
+# Exchange Streams: Cast-Compatible Schemas
+# ---------------------------------------------------------------------------
+
+
+class TestExchangeCastCompatible:
+    """Test that exchange streams cast compatible input schemas (e.g. int32 -> float64)."""
+
+    def test_cast_int32_to_float64(self, conformance_conn: ConnFactory) -> None:
+        """Send int32 values to a float64 exchange, expect float64 output."""
+        with conformance_conn() as proxy, proxy.exchange_cast_compatible() as session:
+            batch = pa.record_batch(
+                [pa.array([1, 2, 3], type=pa.int32())],
+                schema=pa.schema([pa.field("value", pa.int32())]),
+            )
+            out = session.exchange(AnnotatedBatch(batch=batch))
+            assert out.batch.schema.field("value").type == pa.float64()
+            assert out.batch.column("value").to_pylist() == [
+                pytest.approx(1.0),
+                pytest.approx(2.0),
+                pytest.approx(3.0),
+            ]
+
+    def test_cast_int64_to_float64(self, conformance_conn: ConnFactory) -> None:
+        """Send int64 values to a float64 exchange, expect float64 output."""
+        with conformance_conn() as proxy, proxy.exchange_cast_compatible() as session:
+            batch = pa.record_batch(
+                [pa.array([10, 20, 30], type=pa.int64())],
+                schema=pa.schema([pa.field("value", pa.int64())]),
+            )
+            out = session.exchange(AnnotatedBatch(batch=batch))
+            assert out.batch.schema.field("value").type == pa.float64()
+            assert out.batch.column("value").to_pylist() == [
+                pytest.approx(10.0),
+                pytest.approx(20.0),
+                pytest.approx(30.0),
+            ]
+
+    def test_cast_float32_to_float64(self, conformance_conn: ConnFactory) -> None:
+        """Send float32 values to a float64 exchange, expect float64 output."""
+        with conformance_conn() as proxy, proxy.exchange_cast_compatible() as session:
+            batch = pa.record_batch(
+                [pa.array([1.5, 2.5, 3.5], type=pa.float32())],
+                schema=pa.schema([pa.field("value", pa.float32())]),
+            )
+            out = session.exchange(AnnotatedBatch(batch=batch))
+            assert out.batch.schema.field("value").type == pa.float64()
+            assert out.batch.column("value").to_pylist() == [
+                pytest.approx(1.5),
+                pytest.approx(2.5),
+                pytest.approx(3.5),
+            ]
+
+    def test_cast_exact_schema(self, conformance_conn: ConnFactory) -> None:
+        """Send matching float64 values — no cast needed."""
+        with conformance_conn() as proxy, proxy.exchange_cast_compatible() as session:
+            out = session.exchange(AnnotatedBatch.from_pydict({"value": [5.0, 10.0]}))
+            assert out.batch.column("value").to_pylist() == [pytest.approx(5.0), pytest.approx(10.0)]
+
+    def test_cast_incompatible_column_name(self, conformance_conn: ConnFactory) -> None:
+        """Send wrong column name, expect RpcError."""
+        with conformance_conn() as proxy, proxy.exchange_cast_compatible() as session:
+            batch = pa.record_batch(
+                [pa.array([1.0], type=pa.float64())],
+                schema=pa.schema([pa.field("wrong", pa.float64())]),
+            )
+            with pytest.raises(RpcError):
+                session.exchange(AnnotatedBatch(batch=batch))
+
+
 # ---------------------------------------------------------------------------
 # Exchange Streams With Headers
 # ---------------------------------------------------------------------------
@@ -933,7 +1003,7 @@ def test_run_describe_conformance(self, service_description: ServiceDescription)
 
     def test_describe_via_rpc(self, service_description: ServiceDescription) -> None:
         """Smoke test: basic transport-level describe call works."""
-        assert len(service_description.methods) == 47
+        assert len(service_description.methods) == 48
         assert service_description.protocol_name == "ConformanceService"
         echo_str = service_description.methods["echo_string"]
         assert echo_str.method_type == MethodType.UNARY

vgi_rpc/conformance/_pytest_suite.py

@@ -783,6 +783,79 @@ def _test_exchange_empty_session(proxy: ConformanceService, logs: LogCollector)
     assert proxy.echo_int(value=42) == 42
 
 
+# ---------------------------------------------------------------------------
+# Exchange cast-compatible tests
+# ---------------------------------------------------------------------------
+
+
+@_conformance_test(category="exchange_stream", name="cast_int32_to_float64")
+def _test_cast_int32(proxy: ConformanceService, logs: LogCollector) -> None:
+    with proxy.exchange_cast_compatible() as session:
+        batch = pa.record_batch(
+            [pa.array([1, 2, 3], type=pa.int32())],
+            schema=pa.schema([pa.field("value", pa.int32())]),
+        )
+        out = session.exchange(AnnotatedBatch(batch=batch))
+        assert out.batch.schema.field("value").type == pa.float64()
+        values = cast("list[float]", out.batch.column("value").to_pylist())
+        assert abs(values[0] - 1.0) < 1e-6
+        assert abs(values[1] - 2.0) < 1e-6
+        assert abs(values[2] - 3.0) < 1e-6
+
+
+@_conformance_test(category="exchange_stream", name="cast_int64_to_float64")
+def _test_cast_int64(proxy: ConformanceService, logs: LogCollector) -> None:
+    with proxy.exchange_cast_compatible() as session:
+        batch = pa.record_batch(
+            [pa.array([10, 20, 30], type=pa.int64())],
+            schema=pa.schema([pa.field("value", pa.int64())]),
+        )
+        out = session.exchange(AnnotatedBatch(batch=batch))
+        assert out.batch.schema.field("value").type == pa.float64()
+        values = cast("list[float]", out.batch.column("value").to_pylist())
+        assert abs(values[0] - 10.0) < 1e-6
+        assert abs(values[1] - 20.0) < 1e-6
+        assert abs(values[2] - 30.0) < 1e-6
+
+
+@_conformance_test(category="exchange_stream", name="cast_float32_to_float64")
+def _test_cast_float32(proxy: ConformanceService, logs: LogCollector) -> None:
+    with proxy.exchange_cast_compatible() as session:
+        batch = pa.record_batch(
+            [pa.array([1.5, 2.5, 3.5], type=pa.float32())],
+            schema=pa.schema([pa.field("value", pa.float32())]),
+        )
+        out = session.exchange(AnnotatedBatch(batch=batch))
+        assert out.batch.schema.field("value").type == pa.float64()
+        values = cast("list[float]", out.batch.column("value").to_pylist())
+        assert abs(values[0] - 1.5) < 1e-6
+        assert abs(values[1] - 2.5) < 1e-6
+        assert abs(values[2] - 3.5) < 1e-6
+
+
+@_conformance_test(category="exchange_stream", name="cast_exact_schema")
+def _test_cast_exact(proxy: ConformanceService, logs: LogCollector) -> None:
+    with proxy.exchange_cast_compatible() as session:
+        out = session.exchange(AnnotatedBatch.from_pydict({"value": [5.0, 10.0]}))
+        values = cast("list[float]", out.batch.column("value").to_pylist())
+        assert abs(values[0] - 5.0) < 1e-6
+        assert abs(values[1] - 10.0) < 1e-6
+
+
+@_conformance_test(category="exchange_stream", name="cast_incompatible_column_name")
+def _test_cast_incompatible(proxy: ConformanceService, logs: LogCollector) -> None:
+    with proxy.exchange_cast_compatible() as session:
+        batch = pa.record_batch(
+            [pa.array([1.0], type=pa.float64())],
+            schema=pa.schema([pa.field("wrong", pa.float64())]),
+        )
+        try:
+            session.exchange(AnnotatedBatch(batch=batch))
+            raise AssertionError("Expected RpcError")
+        except RpcError as e:
+            assert "TypeError" in str(e) or "type" in str(e).lower()
+
+
 # ---------------------------------------------------------------------------
 # Exchange header tests
 # ---------------------------------------------------------------------------
@@ -1074,6 +1147,7 @@ def decorator(
         "echo_with_log_extras",
         "echo_with_multi_logs",
         "exchange_accumulate",
+        "exchange_cast_compatible",
         "exchange_error_on_init",
         "exchange_error_on_nth",
         "exchange_scale",
@@ -1138,6 +1212,7 @@ def decorator(
 _STREAM_METHODS = frozenset(
     {
         "exchange_accumulate",
+        "exchange_cast_compatible",
         "exchange_error_on_init",
         "exchange_error_on_nth",
         "exchange_scale",
@@ -1203,7 +1278,7 @@ def _test_desc_describe_version(desc: ServiceDescription) -> None:
 
 @_describe_test(category="describe_service", name="method_count")
 def _test_desc_method_count(desc: ServiceDescription) -> None:
-    assert len(desc.methods) == 47
+    assert len(desc.methods) == 48
 
 
 # ---------------------------------------------------------------------------