converter: cast ints to float in float-inferred fields before table build

millpond/arrow_converter.py

@@ -135,6 +135,36 @@ def _stringify_mixed_type_values(records: list[dict], schema: pa.Schema) -> list
     return patched
 
 
+def _floatify_integers_in_float_fields(records: list[dict], schema: pa.Schema) -> list[dict]:
+    """Cast int values to float for fields the schema types as floating.
+
+    JSON producers emit whole floats as integers. PyArrow accepts an int in
+    a double column only when the conversion is exact; an integer above
+    2**53 makes ``pa.Table.from_pylist`` raise ``ArrowInvalid`` and the
+    batch crashes before any column coercion runs. Python ``float()`` is
+    deliberately lossy here — these fields are floating-point measures, so
+    nearest-double is the wanted semantics. Records are only copied when a
+    cast is needed.
+    """
+    float_fields = {f.name for f in schema if pa.types.is_floating(f.type)}
+    if not float_fields:
+        return records
+
+    out = []
+    changed = False
+    for record in records:
+        needs = [k for k in float_fields if type(record.get(k)) is int]
+        if not needs:
+            out.append(record)
+            continue
+        new_record = dict(record)
+        for k in needs:
+            new_record[k] = float(new_record[k])
+        out.append(new_record)
+        changed = True
+    return out if changed else records
+
+
 def _flatten_nested_to_json(records: list[dict]) -> list[dict]:
     """Serialize nested dicts and lists to JSON strings.
 
@@ -320,6 +350,7 @@ def convert(messages: list[bytes]) -> pa.Table | None:
     if patched is not records:
         # Mixed types were found and coerced — re-infer schema with string types
         schema = _build_schema(patched)
+    patched = _floatify_integers_in_float_fields(patched, schema)
     table = pa.Table.from_pylist(patched, schema=schema)
     table = _normalize_numeric_types(table)
     table = _drop_null_typed_columns(table)

tests/unit/test_arrow_converter.py

@@ -144,6 +144,37 @@ def test_mixed_type_int_and_string(self):
         assert len(table) == 2
         assert table.schema.field("employee_count").type == pa.string()
 
+    def test_large_int_in_float_field_converts_lossily(self):
+        """An int above 2^53 in a float-inferred field must not crash the batch.
+
+        The session features topic emits float aggregates as integers when
+        whole; one record carried an integer above 2^53, and PyArrow refuses
+        the inexact int-to-double conversion in from_pylist. The converter
+        pre-casts ints to float when the inferred field type is floating —
+        nearest-double is the wanted semantics for a floating measure.
+        """
+        huge = 130184854372975800  # > 2^53, not exactly representable
+        messages = [
+            orjson.dumps({"mouse_sum_x": 1.5}),
+            orjson.dumps({"mouse_sum_x": huge}),
+        ]
+        table = convert(messages)
+        assert table is not None
+        assert len(table) == 2
+        assert pa.types.is_floating(table.schema.field("mouse_sum_x").type)
+        assert table.column("mouse_sum_x").to_pylist() == [1.5, float(huge)]
+
+    def test_whole_int_in_float_field_stays_float(self):
+        """Whole-number ints in a float-inferred field become floats, not strings."""
+        messages = [
+            orjson.dumps({"velocity": 0.25}),
+            orjson.dumps({"velocity": 3}),
+        ]
+        table = convert(messages)
+        assert table is not None
+        assert pa.types.is_floating(table.schema.field("velocity").type)
+        assert table.column("velocity").to_pylist() == [0.25, 3.0]
+
     def test_large_integer_precision_preserved(self):
         """Integers > 2^53 must not lose precision via float64 cast."""
         large_id = 2**53 + 1  # 9007199254740993 — not representable in float64
@@ -197,9 +228,7 @@ class TestDropNullTypedColumns:
     """
 
     def test_drops_pa_null_column(self):
-        table = pa.table(
-            {"a": pa.array([None, None], pa.null()), "b": ["x", "y"]}
-        )
+        table = pa.table({"a": pa.array([None, None], pa.null()), "b": ["x", "y"]})
         out = _drop_null_typed_columns(table)
         assert out.column_names == ["b"]