feat(serve): table files in the fit/predict request, Parquet, and text columns

CHANGELOG.md

@@ -1,6 +1,16 @@
 # Changelog
 
 ## [Unreleased]
+### Added
+- **A table file in the fit or predict request itself.** `/v1/tabctx/fit` and `/v1/tabctx/predict` take multipart/form-data as well as JSON: the same fields as form fields, the table as a `file` part (`curl -F dataset_id=churn -F target=churned -F file=@train.parquet`), so a file is one call instead of upload-then-reference. The file streams to the replica's disk under the upload size cap, is read, and is deleted. OpenAPI documents both body types.
+- **Parquet and compressed CSV.** Files (sent with the request or through `/v1/tabctx/upload`) may be CSV (plain, gzip or zstd) or Parquet, detected from the content (`serve/files.py`, pyarrow, replacing the numpy CSV parser). A 5,000 x 500 float table is 51 MB as JSON and 13 MB as Parquet; reading it takes 16 ms. Dates and times become seconds, stored from the training minimum so float32 keeps sub-second resolution (`TableSchema.offsets`, for any float64 column past 2^24); header names are stripped; decimal targets are numbers. Labels keep their text as written (`"1.0"`, `"01"`, `"true"` are classes of their own; `"NA"` can be a class among text labels); a missing value among numeric labels, in any spelling, is a 422 naming the row, and regression with text labels is a 422. Reading a CSV peaks at about 3x its size in host memory (arrow's parser; Parquet much less), which matters only for multi-GB direct uploads. `pyarrow` and `python-multipart` join the `serve` extra.
+- **Multipart bodies stream to disk under the upload cap** (`serve/multipart.py`): the file part goes straight to the upload writer as it arrives, so an oversized body stops at the cap (or is refused from its `Content-Length` before any is read), a good one is written once, and disk writes run off the event loop. Tenant checks run before the body is read.
+- **Text columns** (#63). Feature cells may be text in JSON rows and files alike. At fit, each text column's values are coded in sorted order (`tabctx/schema.py`, `TableSchema`); the encoding is kept with the context (spill included) and applied at predict, where a value the training table never had becomes missing. Text where the training table had numbers is a 422 naming the column. All-number JSON tables still convert in one numpy call (validation takes ~20% longer for the wider cell type: 88 -> 108 ms for a 5,000 x 500 body).
+### Changed
+- **Test files are matched to the training file's columns by name**: any order, and extra columns (such as the target) are ignored; a missing column is a 422 naming it. A reordered header used to be a 422.
+- **A request with more than one table** (inline rows, a file, an upload id) is a 422 checked before an upload is used up, so the upload can still be used by a corrected request.
+- Each replica keeps the schemas of the contexts it fit (`serve/schemas.py`, LRU bounded in entries and bytes; fits of one id record in turn, and a failed fit leaves the engine's copy to answer), so a predict doesn't ask a worker process for it; after a restart it falls back to the engine's copy. `engine.schema()` raises `DatasetNotFoundError` for an unknown context (a predict against one is a 404 whatever its rows are).
+- `CachedContext.feature_names` / `engine.feature_names()` became `CachedContext.schema` / `engine.schema()` (a `TableSchema`: names and text categories); `fit(feature_names=...)` is `fit(schema=...)`. Spill files from earlier versions load with their column names, and new ones still record `feature_names` for a rollback.
 
 ## [0.14.1] - 2026-10-21
 ### Fixed

README.md

@@ -261,17 +261,40 @@ behind the same API. No client changes and no routing header needed.
   `per_device`); `/v1/tabctx/limits` reports the best single GPU's
   headroom and each device's.
 
-## Large tables: upload, then fit by reference
+## Table files and text columns
+
+A fit or predict can send its table as rows of JSON, or as a file in
+the same request (multipart/form-data, a `file` part beside the other
+fields as form fields):
+
+```bash
+curl $URL/v1/tabctx/fit -F model=tabicl-v2 -F dataset_id=churn -F target=churned -F file=@train.parquet
+curl $URL/v1/tabctx/predict -F dataset_id=churn -F file=@test.csv
+```
+
+- **Formats**, detected from the content: CSV with a header row (plain,
+  gzip or zstd) and Parquet. Parquet keeps types and nulls and is several
+  times smaller than JSON (a 5,000 x 500 float table: 51 MB as JSON,
+  13 MB as Parquet), so it is the way to send a large table.
+- **Text columns** (`"gold"`, `"CA"`) work in files and JSON rows alike:
+  the server codes each text column's values at fit (`tabctx.schema`),
+  keeps that encoding with the context, and applies it at predict; a
+  value the training table never had becomes missing. **Missing values**:
+  JSON `null`, a Parquet null, or a CSV cell that is empty or `NA`,
+  `N/A`, `null`, `None`, `NaN`.
+- **Columns** of a test file are matched to the training file's by
+  name: any order, and extra columns (such as the target) are ignored. A
+  missing one is a 422.
+- The target is `target` (or `target_column`), default the last column.
 
 Tables too big for GPU and host memory together can still fit: TabICL
 offloads stage outputs to memory-mapped files under `TABCTX_OFFLOAD_DIR`
 (default `<TABCTX_SPILL_DIR>/offload`), slower but without exhausting the
 pod's RAM (one A100 fit 130k x 500 this way).
 
-Inline JSON tables are convenient to ~10^4 rows and painful beyond.
-For big tables, stream a CSV first and reference it (the same
-upload/orchestration split hosted tabular APIs use, minus the object
-store -- uploads stream straight to the serving replica's disk):
+A file can also be uploaded first and referenced by id, for one that
+should travel apart from the request that uses it (uploads stream
+straight to the serving replica's disk):
 
 ```python
 client = TabctxClient("http://localhost:8000")
@@ -282,12 +305,6 @@ test_upload = client.upload_csv_file("scoring_batch.csv", "churn-v1")
 result = client.predict("churn-v1", test_upload_id=test_upload)
 ```
 
-The contract, in short:
-
-- Training CSV = header row + numeric feature columns + one target
-  column (`target_column`, default: the last). Test CSVs carry exactly
-  the training feature columns, same names and order -- a reordered
-  header is rejected 422 rather than silently producing garbage.
 - Uploads are **single-use** (consumed by the fit/predict that
   references them) and expire after `TABCTX_UPLOAD_TTL_S` (default 1h);
   size-capped by `TABCTX_MAX_UPLOAD_BYTES` (default 4GiB, enforced
@@ -296,7 +313,7 @@ The contract, in short:
   request must carry the same `x-session-id: <dataset_id>` header as
   everything else -- affinity routes the upload, fit, and predicts to
   one replica. (The bundled client does this for you.)
-- Transport size and admission are separate layers: a huge CSV uploads
+- Transport size and admission are separate layers: a huge file uploads
   fine, and the memory-admission gate then decides whether the *table*
   fits the GPU budget (413 if not).
 

benchmarks/probe_deployment.py

@@ -145,12 +145,15 @@ def probe_uploads(client: TabctxClient, prefix: str, args) -> None:
         check("uploads are single-use -> 404", False, "no error raised")
     except UploadNotFoundError:
         check("uploads are single-use -> 404", True)
-    bad = client.upload_csv(to_csv(X[:5], None, names[::-1]), ds)
+    reordered = client.upload_csv(to_csv(X[:5], None, names[::-1]), ds)
+    r = client.predict(ds, test_upload_id=reordered)
+    check("reordered test header matched by name", len(r.predictions) == 5)
+    bad = client.upload_csv(to_csv(X[:5, 1:], None, names[1:]), ds)
     try:
         client.predict(ds, test_upload_id=bad)
-        check("reordered test header -> 422", False, "no error raised")
+        check("missing test column -> 422", False, "no error raised")
     except InvalidInputError:
-        check("reordered test header -> 422", True)
+        check("missing test column -> 422", True)
     check(
         "inline predict on CSV-fit context",
         len(client.predict(ds, X[:3].tolist()).predictions) == 3,

pyproject.toml

@@ -40,7 +40,14 @@ tabpfn = ["tabpfn", "torch"]
 # multi-replica deployments correct (see serve/app.py and ROADMAP.md).
 # jinja2 is imported unconditionally by ray.serve's haproxy module but
 # missing from some ray[serve] distributions (seen on a clean CI install).
-serve = ["ray[serve]>=2.58.0", "fastapi", "uvicorn[standard]", "jinja2"]
+serve = [
+    "ray[serve]>=2.58.0",
+    "fastapi",
+    "uvicorn[standard]",
+    "jinja2",
+    "pyarrow>=14",
+    "python-multipart>=0.0.9",
+]
 dev = ["pytest>=7.0", "pytest-cov", "httpx"]
 
 [dependency-groups]
@@ -52,6 +59,8 @@ dev = [
     "httpx",
     "pre-commit",
     "ruff",
+    "pyarrow>=14",
+    "python-multipart>=0.0.9",
 ]
 
 [tool.ruff]

src/tabctx/cache/manager.py

@@ -26,6 +26,7 @@
 
 from tabctx.cache.policies import EvictionPolicy, LRUEvictionPolicy
 from tabctx.errors import CacheCapacityError
+from tabctx.schema import TableSchema
 from tabctx.types import EngineStats, Task
 
 logger = logging.getLogger(__name__)
@@ -40,11 +41,10 @@ class CachedContext:
     n_features: int
     payload: Any
     est_bytes: int
-    # Training-table column names when the context was fit from a CSV
-    # (serve/csv_io.py); None for inline tables. Travels with the
-    # context through the spill tier so a restored context keeps its
-    # schema check.
-    feature_names: list[str] | None = None
+    # How the training table's columns were encoded (tabctx.schema), so
+    # test tables are encoded the same way. Travels with the context
+    # through the spill tier.
+    schema: TableSchema | None = None
     created_at: float = field(default_factory=time.monotonic)
     last_accessed_at: float = field(default_factory=time.monotonic)
 

src/tabctx/cache/spill.py

@@ -46,6 +46,7 @@
 from pathlib import Path
 
 from tabctx.cache.manager import CachedContext
+from tabctx.schema import TableSchema
 
 DEFAULT_SPILL_CAPACITY_BYTES = 50 * 1024**3
 
@@ -122,7 +123,13 @@ def spill(self, context: CachedContext) -> bool:
                         "n_train": context.n_train,
                         "n_features": context.n_features,
                         "est_bytes": context.est_bytes,
-                        "feature_names": context.feature_names,
+                        "schema": context.schema.to_json() if context.schema else None,
+                        # Read by 0.14 and earlier (a rollback keeps names).
+                        "feature_names": (
+                            list(context.schema.names)
+                            if context.schema and context.schema.names
+                            else None
+                        ),
                     }
                 )
             )
@@ -163,8 +170,7 @@ def load(self, dataset_id: str) -> CachedContext | None:
             n_features=meta["n_features"],
             payload=payload,
             est_bytes=meta["est_bytes"],
-            # Absent from files written before feature_names existed.
-            feature_names=meta.get("feature_names"),
+            schema=_schema(meta),
         )
 
     def est_bytes(self, dataset_id: str) -> int | None:
@@ -214,3 +220,14 @@ def stats(self) -> dict:
             "spill_used_bytes": self.used_bytes,
             "spill_capacity_bytes": self._capacity_bytes,
         }
+
+
+def _schema(meta: dict) -> TableSchema | None:
+    """The spilled context's schema; files from before schemas existed
+    may record only column names (all numeric then)."""
+    if meta.get("schema") is not None:
+        return TableSchema.from_json(meta["schema"])
+    names = meta.get("feature_names")
+    if names is None:
+        return None
+    return TableSchema((None,) * len(names), tuple(names))

src/tabctx/engine.py

@@ -26,6 +26,7 @@
 from tabctx.labels import resolve_task
 from tabctx.memory.estimator import MemoryEstimator
 from tabctx.memory.host import HostMemoryGuard
+from tabctx.schema import TableSchema
 from tabctx.types import ArrayLike, EngineStats, PredictOutcome, Task
 
 # Predicts of at most this many test rows against a resident context run
@@ -292,16 +293,16 @@ def fit(
         task: Task | None = None,
         dataset_id: str | None = None,
         backend: str | None = None,
-        feature_names: list[str] | None = None,
+        schema: TableSchema | None = None,
     ) -> str:
         """Encode and cache a training context. Returns a dataset_id to pass
         to predict() -- generated automatically unless the caller supplies
         one (e.g. to reuse a stable, caller-known identifier).
 
-        feature_names (optional) records the training table's column
-        names on the cached context (see CachedContext) for callers that
-        validate a test table's header against it; the engine itself only
-        ever checks feature *count*.
+        schema (optional) records how the training table's columns were
+        encoded (tabctx.schema) on the cached context, for callers that
+        encode test tables the same way; the engine itself only ever
+        checks feature *count*.
 
         Raises InvalidInputError for malformed input (mismatched X/y
         lengths, empty tables, ragged rows) before anything else -- checked
@@ -376,7 +377,7 @@ def fit(
                     n_features=n_features,
                     payload=payload,
                     est_bytes=est_bytes,
-                    feature_names=feature_names,
+                    schema=schema,
                 )
                 self._cache.put(context)
         return resolved_id
@@ -582,17 +583,17 @@ def _restore(self, dataset_id: str) -> None:
             if restored is None:
                 raise _not_found(dataset_id)
 
-    def feature_names(self, dataset_id: str) -> list[str] | None:
-        """Column names recorded at fit() for a cached context, or None
-        when the context is unknown or was fit without names. Reloads a
-        spilled context as a side effect, exactly as predict() would."""
+    def schema(self, dataset_id: str) -> TableSchema | None:
+        """The schema recorded at fit() for a cached context (None if it
+        was fit without one). Reloads a spilled context as a side effect,
+        exactly as predict() would; raises DatasetNotFoundError, as
+        predict() would, when there is no such context."""
         if self._cache.peek(dataset_id) is None:
-            try:
-                self._restore(dataset_id)
-            except DatasetNotFoundError:
-                return None
+            self._restore(dataset_id)
         context = self._cache.peek(dataset_id)
-        return context.feature_names if context is not None else None
+        if context is None:  # evicted again in between
+            raise _not_found(dataset_id)
+        return context.schema
 
     def predict(
         self, dataset_id: str, X_test: ArrayLike, return_proba: bool = False

src/tabctx/keyed_locks.py

@@ -0,0 +1,28 @@
+"""One lock per key (a dataset id), created on demand and dropped when
+unused, so calls for one key run one at a time while others proceed."""
+
+from __future__ import annotations
+
+import contextlib
+import threading
+from collections.abc import Iterator
+
+
+class KeyedLocks:
+    def __init__(self) -> None:
+        self._guard = threading.Lock()
+        self._locks: dict[str, list] = {}  # key -> [lock, users]
+
+    @contextlib.contextmanager
+    def hold(self, key: str) -> Iterator[None]:
+        with self._guard:
+            entry = self._locks.setdefault(key, [threading.Lock(), 0])
+            entry[1] += 1
+        try:
+            with entry[0]:
+                yield
+        finally:
+            with self._guard:
+                entry[1] -= 1
+                if entry[1] == 0:
+                    del self._locks[key]

src/tabctx/labels.py

@@ -6,9 +6,9 @@
 (prices, counts), are a continuous target; anything else is
 classification. Only actual
 numbers count: text labels are categories even when they look numeric
-("250.01" as a diagnosis code), as they always were. CSV labels arrive as
-text, so the CSV parser converts them first when the task is left open
-(serve/csv_io.py).
+("250.01" as a diagnosis code), as they always were. A table file's labels
+are text as written, so the file reader converts them first when they are
+all numbers (serve/files.py).
 
 When the caller does say classification but sends non-integer numbers,
 that's a mistake worth a clear 422 naming the fix, not a backend error