diff --git a/CHANGELOG.md b/CHANGELOG.md index f1cdf07..c0f7969 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 85b122f..f43b7bd 100644 --- a/README.md +++ b/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 `/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: ` 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). diff --git a/benchmarks/probe_deployment.py b/benchmarks/probe_deployment.py index 289839d..76242bf 100644 --- a/benchmarks/probe_deployment.py +++ b/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, diff --git a/pyproject.toml b/pyproject.toml index 9f40f13..ca83218 100644 --- a/pyproject.toml +++ b/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] diff --git a/src/tabctx/cache/manager.py b/src/tabctx/cache/manager.py index 02581ff..16dfc10 100644 --- a/src/tabctx/cache/manager.py +++ b/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) diff --git a/src/tabctx/cache/spill.py b/src/tabctx/cache/spill.py index 2307cd0..6fc42b0 100644 --- a/src/tabctx/cache/spill.py +++ b/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)) diff --git a/src/tabctx/engine.py b/src/tabctx/engine.py index 103c002..3494c1c 100644 --- a/src/tabctx/engine.py +++ b/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 diff --git a/src/tabctx/keyed_locks.py b/src/tabctx/keyed_locks.py new file mode 100644 index 0000000..c69cebd --- /dev/null +++ b/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] diff --git a/src/tabctx/labels.py b/src/tabctx/labels.py index 5a55451..54fcf99 100644 --- a/src/tabctx/labels.py +++ b/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 diff --git a/src/tabctx/multidevice.py b/src/tabctx/multidevice.py index 49250b8..1e3e91e 100644 --- a/src/tabctx/multidevice.py +++ b/src/tabctx/multidevice.py @@ -55,6 +55,7 @@ from tabctx.engine import _validate_fit_input from tabctx.errors import DatasetNotFoundError from tabctx.events import EventBus, EventKind, ResidencyEvent +from tabctx.keyed_locks import KeyedLocks from tabctx.labels import resolve_task from tabctx.placement import ( DeviceView, @@ -62,6 +63,7 @@ LeastActiveThatFits, PlacementPolicy, ) +from tabctx.schema import TableSchema from tabctx.types import ArrayLike, EngineStats, PredictOutcome, Task logger = logging.getLogger(__name__) @@ -106,40 +108,18 @@ 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: ... def predict( self, dataset_id: str, X_test: ArrayLike, return_proba: bool = False ) -> PredictOutcome: ... - def feature_names(self, dataset_id: str) -> list[str] | None: ... + def schema(self, dataset_id: str) -> TableSchema | None: ... def evict(self, dataset_id: str) -> None: ... -class _KeyedLocks: - """One lock per key, created on demand and dropped when unused.""" - - 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] - - def _not_found(dataset_id: str) -> DatasetNotFoundError: return DatasetNotFoundError( f"no cached context for dataset_id={dataset_id!r} -- it was never " @@ -165,7 +145,7 @@ def __init__( raise ValueError("every device must serve the same models") self._shards = shards self._devices = devices - self._id_locks = _KeyedLocks() + self._id_locks = KeyedLocks() self._window_s = activity_window_s self._clock = clock self._inflight = [0] * len(shards) @@ -266,9 +246,9 @@ 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: - return self._fit(X, y, task, dataset_id, backend, feature_names, announce=True) + return self._fit(X, y, task, dataset_id, backend, schema, announce=True) def _fit( self, @@ -277,7 +257,7 @@ def _fit( task: Task | None, dataset_id: str | None, backend: str | None, - feature_names: list[str] | None, + schema: TableSchema | None, announce: bool, ) -> str: """fit(); `announce` emits residency events (not for fit_predict's @@ -289,7 +269,7 @@ def _fit( if dataset_id is None: with timing.phase("place"): index = self._place(request) - resolved = self._fit_on(index, X, y, task, None, name, feature_names) + resolved = self._fit_on(index, X, y, task, None, name, schema) if announce: self._emit("placed", resolved, index, name) return resolved @@ -298,7 +278,7 @@ def _fit( held.enter_context(self._id_locks.hold(dataset_id)) with timing.phase("place"): index = self._place(request) - resolved = self._fit_on(index, X, y, task, dataset_id, name, feature_names) + resolved = self._fit_on(index, X, y, task, dataset_id, name, schema) if announce: self._emit("placed", resolved, index, name) # Only now that the new context is in place: drop copies on @@ -329,7 +309,7 @@ def _emit( ResidencyEvent(kind, dataset_id, self._devices[index], backend) ) - def _fit_on(self, index, X, y, task, dataset_id, backend, feature_names) -> str: + def _fit_on(self, index, X, y, task, dataset_id, backend, schema) -> str: with self._active(index): return self._shards[index].fit( X, @@ -337,7 +317,7 @@ def _fit_on(self, index, X, y, task, dataset_id, backend, feature_names) -> str: task=task, dataset_id=dataset_id, backend=backend, - feature_names=feature_names, + schema=schema, ) def predict( @@ -362,12 +342,13 @@ def predict( continue raise _not_found(dataset_id) - def feature_names(self, dataset_id: str) -> list[str] | None: + def schema(self, dataset_id: str) -> TableSchema | None: for index in self._locate(dataset_id): - names = self._shards[index].feature_names(dataset_id) - if names is not None: - return names - return None + try: + return self._shards[index].schema(dataset_id) + except DatasetNotFoundError: + continue # moved or evicted since the snapshot + raise _not_found(dataset_id) def fit_predict( self, diff --git a/src/tabctx/schema.py b/src/tabctx/schema.py new file mode 100644 index 0000000..e4be8d6 --- /dev/null +++ b/src/tabctx/schema.py @@ -0,0 +1,224 @@ +"""A training table's columns, kept with its context so new rows can be +encoded the same way. + +Both models take numbers. Real tables also have text columns ("gold", +"CA"), so at fit time each text column's values become categories, coded +0, 1, 2, ... in sorted order, and the context keeps that encoding. At +predict, the same text maps to the same codes; a value the training +table never had, or a missing one, becomes NaN, which both models treat +as missing. + +A column is numeric when every present value is a number; any text makes +it a text column, and numbers in it are compared as text ("3" == 3). +Columns are matched by name when the table had a header (a file), and by +position otherwise (JSON rows). + +The schema travels with the cached context (spill included) and between +processes, so it is plain data: names and category lists, JSON round +trip via to_json/from_json. +""" + +from __future__ import annotations + +import numbers +from collections.abc import Sequence +from dataclasses import dataclass +from functools import cached_property + +import numpy as np + +from tabctx.errors import InvalidInputError + +# A table as columns: float arrays (NaN = missing) for numeric columns, +# object arrays of str/number/None for columns that may hold text. +Columns = Sequence[np.ndarray] + + +@dataclass(frozen=True) +class TableSchema: + """Per feature column: None if numeric, else its categories in code + order. names: the columns' names, when the table had a header. + offsets: per column, a value subtracted before the model's float32, + for columns given in float64 (times as seconds since the epoch, where + float32 alone would round to 128 s): the training minimum.""" + + categories: tuple[tuple[str, ...] | None, ...] + names: tuple[str, ...] | None = None + offsets: tuple[float | None, ...] | None = None + + @property + def n_features(self) -> int: + return len(self.categories) + + @property + def has_text(self) -> bool: + return any(c is not None for c in self.categories) + + @cached_property + def approx_bytes(self) -> int: + """Roughly what the schema takes in memory (its text dominates).""" + text = [ + *(self.names or ()), + *(c for cats in self.categories if cats for c in cats), + ] + return 64 * len(self.categories) + sum(len(t) + 56 for t in text) + + @property + def text_names(self) -> tuple[str, ...]: + """Names of the text columns (none when the table had no names).""" + if self.names is None: + return () + return tuple( + n for n, c in zip(self.names, self.categories, strict=True) if c is not None + ) + + # ---- fit ----------------------------------------------------------- + + @classmethod + def learn( + cls, columns: Columns, names: Sequence[str] | None = None + ) -> tuple[TableSchema, np.ndarray]: + """The schema of a training table, and the table encoded with it.""" + categories = tuple(_categories(col) for col in columns) + offsets = tuple(_offset(col) for col in columns) + schema = cls( + categories, + tuple(names) if names is not None else None, + offsets if any(o is not None for o in offsets) else None, + ) + return schema, schema.encode(columns) + + # ---- predict ------------------------------------------------------- + + def encode(self, columns: Columns) -> np.ndarray: + """Columns in schema order -> a float32 table for the model.""" + if len(columns) != self.n_features: + raise InvalidInputError( + f"the table has {len(columns)} columns; the training table had " + f"{self.n_features}" + ) + n_rows = len(columns[0]) if columns else 0 + out = np.empty((n_rows, self.n_features), dtype=np.float32) + offsets = self.offsets or (None,) * self.n_features + for j, (col, codes, offset) in enumerate( + zip(columns, self._codes, offsets, strict=True) + ): + if codes is not None: + out[:, j] = _encode(col, codes) + elif offset is not None: + out[:, j] = _numbers(col, self._label(j), np.float64) - offset + else: + out[:, j] = _numbers(col, self._label(j)) + return out + + @cached_property + def _codes(self) -> tuple[dict[str, float] | None, ...]: + """Per text column, value -> code (built once per schema).""" + return tuple( + None if cats is None else {c: float(i) for i, c in enumerate(cats)} + for cats in self.categories + ) + + def __getstate__(self) -> dict: + # Derived values are rebuilt on use, not shipped between processes. + return { + k: v + for k, v in self.__dict__.items() + if k not in ("_codes", "approx_bytes") + } + + def select(self, names: Sequence[str], columns: Columns) -> Columns: + """A named table's columns in schema order (by name when the + training table had names, else by position). Extra columns, such + as the target, are ignored; missing ones are an error.""" + if self.names is None: + return columns + index = {name: i for i, name in enumerate(names)} + missing = [n for n in self.names if n not in index] + if missing: + raise InvalidInputError( + f"the table is missing training column(s) {missing[:10]}; it has " + f"{list(names)[:20]}" + ) + return [columns[index[n]] for n in self.names] + + def _label(self, j: int) -> str: + return repr(self.names[j]) if self.names is not None else f"{j}" + + # ---- storage ------------------------------------------------------- + + def to_json(self) -> dict: + return { + "categories": [list(c) if c is not None else None for c in self.categories], + "names": list(self.names) if self.names is not None else None, + "offsets": list(self.offsets) if self.offsets is not None else None, + } + + @classmethod + def from_json(cls, data: dict) -> TableSchema: + offsets = data.get("offsets") + return cls( + tuple(tuple(c) if c is not None else None for c in data["categories"]), + tuple(data["names"]) if data.get("names") is not None else None, + tuple(offsets) if offsets is not None else None, + ) + + +def _as_text(value: object) -> str: + """A cell's text: numbers as written in a CSV (3, not 3.0).""" + if isinstance(value, float) and value.is_integer(): + return str(int(value)) + return str(value) + + +def _present(value: object) -> bool: + return value is not None and not (isinstance(value, float) and np.isnan(value)) + + +def _categories(col: np.ndarray) -> tuple[str, ...] | None: + if col.dtype.kind in "biuf": + return None + present = [v for v in col if _present(v)] + if all(isinstance(v, numbers.Real) for v in present): + return None + return tuple(sorted({_as_text(v) for v in present})) + + +# float32 holds every integer up to 2**24 exactly; past it, values are +# stored from the training minimum so differences keep their resolution. +_FLOAT32_EXACT = float(2**24) + + +def _offset(col: np.ndarray) -> float | None: + """For a float64 column with values past float32's exact range (times + as seconds since the epoch): its training minimum.""" + if col.dtype != np.float64: + return None + finite = col[np.isfinite(col)] + if not finite.size or np.abs(finite).max() < _FLOAT32_EXACT: + return None + return float(finite.min()) + + +def _numbers(col: np.ndarray, label: str, dtype=np.float32) -> np.ndarray: + if col.dtype.kind in "biuf": + return col.astype(dtype, copy=False) + try: + return np.array([v if _present(v) else np.nan for v in col], dtype=dtype) + except (TypeError, ValueError): + text = next(v for v in col if isinstance(v, str)) + raise InvalidInputError( + f"column {label} held numbers in the training table but has text " + f"here ({text!r})" + ) from None + + +def _encode(col: np.ndarray, code: dict[str, float]) -> np.ndarray: + if col.dtype.kind == "f": # a numeric column where training had text + values = (None if np.isnan(v) else float(v) for v in col) + else: + values = iter(col) + return np.array( + [code.get(_as_text(v), np.nan) if _present(v) else np.nan for v in values], + dtype=np.float32, + ) diff --git a/src/tabctx/serve/app.py b/src/tabctx/serve/app.py index 3a0ffbb..f397c24 100644 --- a/src/tabctx/serve/app.py +++ b/src/tabctx/serve/app.py @@ -26,10 +26,12 @@ import logging import time import uuid +from pathlib import Path from typing import Literal from fastapi import FastAPI, HTTPException, Request -from pydantic import BaseModel, Field +from fastapi.exceptions import RequestValidationError +from pydantic import AliasChoices, BaseModel, Field, ValidationError from ray import serve from ray.serve.config import RequestRouterConfig from starlette.concurrency import run_in_threadpool @@ -49,10 +51,13 @@ from tabctx.events import RecentEvents from tabctx.labels import resolve_task from tabctx.multidevice import MultiDeviceEngine +from tabctx.schema import TableSchema from tabctx.serve.affinity import resolve_dataset_id, session_id_from_headers -from tabctx.serve.csv_io import parse_features_csv, parse_train_csv from tabctx.serve.factory import ServeSettings, build_engine -from tabctx.serve.tables import rows_to_array +from tabctx.serve.files import read_table, read_training_table +from tabctx.serve.multipart import FILE_FIELD, MAX_FIELD_BYTES, MAX_FIELDS, read_form +from tabctx.serve.schemas import SchemaCache +from tabctx.serve.tables import encode_rows, learn_rows from tabctx.serve.tenancy import ( TenantRequiredError, resolve_tenant_id, @@ -85,7 +90,12 @@ class GpuMemory(BaseModel): reserved_mb: float -_TABLE_DOC = "Rows of numbers; null marks a missing value." +_TABLE_DOC = ( + "Rows of cells: numbers, text (a category: the server encodes text " + "columns at fit and reuses that encoding at predict), or null for a " + "missing value." +) +Cell = float | str | None _TIMINGS_DOC = ( "Where this request's time went on the server, in ms per phase (waits " @@ -106,9 +116,9 @@ class GpuMemory(BaseModel): class LegacyPredictRequest(BaseModel): - train_X: list[list[float | None]] = Field(description=_TABLE_DOC) + train_X: list[list[Cell]] = Field(description=_TABLE_DOC) train_y: list[float | str] - test_X: list[list[float | None]] = Field(description=_TABLE_DOC) + test_X: list[list[Cell]] = Field(description=_TABLE_DOC) task: Literal["classification", "regression"] | None = Field( default=None, description=_TASK_DOC ) @@ -139,14 +149,18 @@ class LegacyPredictResponse(BaseModel): class FitRequest(BaseModel): # Inline path (small tables) ... - train_X: list[list[float | None]] | None = Field( - default=None, description=_TABLE_DOC - ) + train_X: list[list[Cell]] | None = Field(default=None, description=_TABLE_DOC) train_y: list[float | str] | None = None - # ... or by-reference path (large tables): a prior POST /v1/tabctx/upload - # of a CSV whose columns are features plus one target column. + # ... or a table file: sent with the request (multipart `file`), or a + # prior POST /v1/tabctx/upload referenced by id. Its columns are the + # features plus one target column. train_upload_id: str | None = None - target_column: str | None = None # default: the CSV's last column + target_column: str | None = Field( + default=None, + validation_alias=AliasChoices("target_column", "target"), + description="The table file's label column (also accepted as `target`); " + "default: its last column.", + ) task: Literal["classification", "regression"] | None = Field( default=None, description=_TASK_DOC ) @@ -183,11 +197,11 @@ class FitResponse(BaseModel): class TabctxPredictRequest(BaseModel): dataset_id: str # Inline path ... - test_X: list[list[float | None]] | None = Field( - default=None, description=_TABLE_DOC - ) - # ... or by-reference: an uploaded CSV of feature columns only, with - # the same header (names and order) the training CSV had. + test_X: list[list[Cell]] | None = Field(default=None, description=_TABLE_DOC) + # ... or a table file (multipart `file`, or an upload's id). When the + # training table was a file, columns are matched by name, so their + # order doesn't matter and extra columns (such as the target) are + # ignored; otherwise by position. test_upload_id: str | None = None return_proba: bool = False @@ -228,6 +242,42 @@ def _map_error(e: Exception) -> HTTPException: raise e +def _request_body(model: type[BaseModel]) -> dict: + """OpenAPI for a body that is JSON, or multipart/form-data with the + same fields as form fields plus the table as a `file` part.""" + schema = model.model_json_schema() + form = { + **schema, + "properties": { + "file": { + "type": "string", + "format": "binary", + "description": "The table: CSV (plain, gzip or zstd) or Parquet, " + "with a header row.", + }, + **schema["properties"], + }, + "required": ["file", *schema.get("required", [])], + } + return { + "requestBody": { + "required": True, + "content": { + "application/json": {"schema": schema}, + "multipart/form-data": {"schema": form}, + }, + } + } + + +def _body_error(e: ValidationError) -> RequestValidationError: + """FastAPI's own 422 shape for a body it validated: locations under + "body".""" + return RequestValidationError( + [{**err, "loc": ("body", *err["loc"])} for err in e.errors()] + ) + + def _replica_tag() -> str | None: try: return serve.get_replica_context().replica_tag @@ -292,6 +342,9 @@ def __init__(self) -> None: # Large-table ingestion (fit/predict-by-reference): replica-local # streamed CSV uploads; see serve/uploads.py for the affinity and # tenancy contracts that make this correct at num_replicas >= 2. + # Each fitted context's schema, kept here too so a predict needn't + # ask the engine (a round trip to a worker process) for it. + self._schemas = SchemaCache() self._uploads = UploadStore( ttl_s=settings.upload_ttl_s, max_upload_bytes=settings.max_upload_bytes, @@ -338,10 +391,11 @@ def _legacy_predict_sync(self, req: LegacyPredictRequest) -> LegacyPredictRespon start = time.monotonic() try: task = resolve_task(req.train_y, req.task) + X, schema = learn_rows(req.train_X) outcome = self._engine.fit_predict( - rows_to_array(req.train_X), + X, req.train_y, - rows_to_array(req.test_X), + encode_rows(req.test_X, schema), task=task, return_proba=req.return_proba, ) @@ -416,52 +470,148 @@ async def tabctx_upload(self, request: Request) -> UploadResponse: served_by=_replica_tag(), ) - @fastapi_app.post("/v1/tabctx/fit", response_model=FitResponse) - async def tabctx_fit(self, req: FitRequest, request: Request) -> FitResponse: - session_id = session_id_from_headers(request.headers) - return await run_in_threadpool( - self._fit_sync, req, dict(request.headers), session_id - ) + # ---- request bodies: JSON, or multipart/form-data with a table file ---- + + async def _read_request[R: BaseModel]( + self, request: Request, model: type[R] + ) -> tuple[R, Path | None]: + """The request's fields, and its table file if it sent one. A JSON + body carries the fields; a multipart/form-data body carries them as + form fields and the table as a `file` part, which streams to disk + under the upload size cap (serve/multipart.py) for the caller to + read and discard.""" + # Refused callers and oversized bodies are turned away from the + # headers alone, before any of the body is read. + try: + resolve_tenant_id(dict(request.headers)) + except (*_AUTH_ERRORS, *_INVALID_INPUT_ERRORS) as e: + raise _map_error(e) from e + content_type = request.headers.get("content-type", "") + if not content_type.lower().startswith("multipart/form-data"): + try: + return model.model_validate_json(await request.body()), None + except ValidationError as e: + raise _body_error(e) from e + self._check_length(request) + try: + fields, writer = await read_form( + request.stream(), content_type, self._uploads.begin + ) + except (InvalidInputError, UploadTooLargeError) as e: + raise _map_error(e) from e + if writer is None: + raise HTTPException( + 422, f"a multipart request carries its table as a `{FILE_FIELD}` part" + ) + path = writer.finish() + try: + return model.model_validate(fields), path + except ValidationError as e: + self._uploads.discard(path) + raise _body_error(e) from e + + def _check_length(self, request: Request) -> None: + declared = request.headers.get("content-length", "") + cap = self._uploads.max_upload_bytes + MAX_FIELDS * MAX_FIELD_BYTES + if declared.isdigit() and int(declared) > cap: + raise _map_error( + UploadTooLargeError( + f"the request is {int(declared)} bytes; tables are capped at " + f"{self._uploads.max_upload_bytes} (TABCTX_MAX_UPLOAD_BYTES)" + ) + ) - def _resolve_train_table( - self, req: FitRequest, tenant_id: str | None - ) -> tuple[object, object, list[str] | None]: - """Returns (X, y, feature_names_or_None) from whichever of the - inline / by-reference paths the request used -- exactly one.""" - inline = req.train_X is not None or req.train_y is not None - if inline and req.train_upload_id is not None: + def _table_file( + self, + sent: Path | None, + upload_id: str | None, + inline: bool, + tenant_id: str | None, + rows: str, + ) -> Path | None: + """The request's table file -- sent with it, or an earlier upload -- + or None for inline rows. Exactly one of the three, checked before + an upload is used up (uploads are single-use).""" + given = (sent is not None) + (upload_id is not None) + inline + if given != 1: raise InvalidInputError( - "provide either inline train_X/train_y or train_upload_id, not both" + f"send the table one way: inline {rows}, a table file, or an " + f"upload id ({'got several' if given else 'got none'})" ) - if req.train_upload_id is not None: - path = self._uploads.consume(req.train_upload_id, tenant_id) - try: - X, y, feature_names = parse_train_csv(path, req.task, req.target_column) - finally: - self._uploads.discard(path) - return X, y, feature_names - if req.train_X is None or req.train_y is None: - raise InvalidInputError( - "fit needs either train_X and train_y (inline) or " - "train_upload_id (an uploaded CSV)" + if upload_id is not None: + return self._uploads.consume(upload_id, tenant_id) + return sent + + def _schema(self, scoped_id: str) -> TableSchema | None: + """The context's schema: this replica recorded it at fit (affinity + sends a dataset's fits and predicts here), else -- after a restart -- + the engine's copy. Raises DatasetNotFoundError for no context. + (A re-fit racing a predict of the same id can encode the predict's + rows with the old schema; re-fitting an id while predicting + against it has no defined result anyway.)""" + schema = self._schemas.get(scoped_id) + return schema if schema is not None else self._engine.schema(scoped_id) + + @fastapi_app.post( + "/v1/tabctx/fit", + response_model=FitResponse, + openapi_extra=_request_body(FitRequest), + ) + async def tabctx_fit(self, request: Request) -> FitResponse: + req, file = await self._read_request(request, FitRequest) + session_id = session_id_from_headers(request.headers) + try: + return await run_in_threadpool( + self._fit_sync, req, file, dict(request.headers), session_id ) - return rows_to_array(req.train_X), req.train_y, None + finally: + if file is not None: + self._uploads.discard(file) + + def _resolve_train_table( + self, req: FitRequest, file: Path | None, tenant_id: str | None + ) -> tuple[object, object, TableSchema | None]: + """Returns (X, y, schema) from whichever of inline rows or a table + file the request used -- exactly one.""" + inline = req.train_X is not None or req.train_y is not None + file = self._table_file( + file, req.train_upload_id, inline, tenant_id, "train_X and train_y" + ) + if file is None: + if req.train_X is None or req.train_y is None: + raise InvalidInputError("an inline fit needs both train_X and train_y") + X, schema = learn_rows(req.train_X) + return X, req.train_y, schema + try: + features, y = read_training_table(file, req.target_column, req.task) + finally: + self._uploads.discard(file) + schema, X = TableSchema.learn(features.columns, features.names) + return X, y, schema def _fit_sync( - self, req: FitRequest, headers: dict[str, str], session_id: str | None + self, + req: FitRequest, + file: Path | None, + headers: dict[str, str], + session_id: str | None, ) -> FitResponse: with timing.collect() as timings: - response = self._fit_timed(req, headers, session_id) + response = self._fit_timed(req, file, headers, session_id) response.timings_ms = timings.as_dict() return response def _fit_timed( - self, req: FitRequest, headers: dict[str, str], session_id: str | None + self, + req: FitRequest, + file: Path | None, + headers: dict[str, str], + session_id: str | None, ) -> FitResponse: try: tenant_id = resolve_tenant_id(headers) with timing.phase("read_table"): - X, y, feature_names = self._resolve_train_table(req, tenant_id) + X, y, schema = self._resolve_train_table(req, file, tenant_id) task = resolve_task(y, req.task) n_train = len(X) n_features = len(X[0]) if n_train else 0 @@ -474,18 +624,20 @@ def _fit_timed( ) scoped_id = scope_dataset_id(tenant_id, dataset_id) model_id = req.model or self._engine.default_backend - # A CSV-fit context records its header on the cached context - # so a test CSV's columns can be checked against it (a - # silently reordered column would mean garbage predictions); - # the schema lives and dies with the context, spill included. - self._engine.fit( - X, - y, - task=task, - dataset_id=scoped_id, - backend=model_id, - feature_names=feature_names, - ) + # The schema (column names, text categories) is recorded on + # the cached context so test tables are encoded the same way; + # it lives and dies with the context, spill included. + with self._schemas.fitting(scoped_id) as record: + self._engine.fit( + X, + y, + task=task, + dataset_id=scoped_id, + backend=model_id, + schema=schema, + ) + if schema is not None: + record(schema) except ( *_AUTH_ERRORS, *_INVALID_INPUT_ERRORS, @@ -504,52 +656,63 @@ def _fit_timed( served_by=_replica_tag(), ) - @fastapi_app.post("/v1/tabctx/predict", response_model=TabctxPredictResponse) - async def tabctx_predict( - self, req: TabctxPredictRequest, request: Request - ) -> TabctxPredictResponse: + @fastapi_app.post( + "/v1/tabctx/predict", + response_model=TabctxPredictResponse, + openapi_extra=_request_body(TabctxPredictRequest), + ) + async def tabctx_predict(self, request: Request) -> TabctxPredictResponse: + req, file = await self._read_request(request, TabctxPredictRequest) session_id = session_id_from_headers(request.headers) - return await run_in_threadpool( - self._predict_sync, req, dict(request.headers), session_id - ) + try: + return await run_in_threadpool( + self._predict_sync, req, file, dict(request.headers), session_id + ) + finally: + if file is not None: + self._uploads.discard(file) def _resolve_test_table( - self, req: TabctxPredictRequest, tenant_id: str | None, scoped_id: str + self, + req: TabctxPredictRequest, + file: Path | None, + tenant_id: str | None, + scoped_id: str, ) -> object: - if req.test_X is not None and req.test_upload_id is not None: - raise InvalidInputError( - "provide either inline test_X or test_upload_id, not both" - ) - if req.test_upload_id is not None: - path = self._uploads.consume(req.test_upload_id, tenant_id) - try: - # Schema check against the training CSV's header when the - # context was fit by reference; inline-fit (or unknown) - # contexts fall back to the engine's feature-count check. - return parse_features_csv(path, self._engine.feature_names(scoped_id)) - finally: - self._uploads.discard(path) - if req.test_X is None: - raise InvalidInputError( - "predict needs either test_X (inline) or test_upload_id " - "(an uploaded CSV)" - ) - return rows_to_array(req.test_X) + file = self._table_file( + file, req.test_upload_id, req.test_X is not None, tenant_id, "test_X" + ) + try: + # Encoded as the training table was (text categories, column + # names); contexts from before schemas existed have none. + schema = self._schema(scoped_id) + if file is None: + return encode_rows(req.test_X, schema) + text = schema.text_names if schema is not None else () + table = read_table(file, text_columns=text) + if schema is None: + schema = TableSchema((None,) * len(table.columns)) + return schema.encode(schema.select(table.names, table.columns)) + finally: + if file is not None: + self._uploads.discard(file) def _predict_sync( self, req: TabctxPredictRequest, + file: Path | None, headers: dict[str, str], session_id: str | None, ) -> TabctxPredictResponse: with timing.collect() as timings: - response = self._predict_timed(req, headers, session_id) + response = self._predict_timed(req, file, headers, session_id) response.timings_ms = timings.as_dict() return response def _predict_timed( self, req: TabctxPredictRequest, + file: Path | None, headers: dict[str, str], session_id: str | None, ) -> TabctxPredictResponse: @@ -559,7 +722,7 @@ def _predict_timed( resolve_dataset_id(session_id, req.dataset_id) scoped_id = scope_dataset_id(tenant_id, req.dataset_id) with timing.phase("read_table"): - X_test = self._resolve_test_table(req, tenant_id, scoped_id) + X_test = self._resolve_test_table(req, file, tenant_id, scoped_id) outcome = self._predictor.predict( scoped_id, X_test, return_proba=req.return_proba ) diff --git a/src/tabctx/serve/csv_io.py b/src/tabctx/serve/csv_io.py deleted file mode 100644 index 1922e3f..0000000 --- a/src/tabctx/serve/csv_io.py +++ /dev/null @@ -1,175 +0,0 @@ -"""CSV -> arrays for fit/predict-by-reference (see serve/uploads.py). - -Parsing lives at the serving layer on purpose: the engine keeps taking -in-memory tables (ArrayLike), so transport format is a serving concern -and the engine/backends never learn about files. - -Format contract (kept deliberately simple -- this is a numeric-table -transport, not a general CSV ingester): - -- First row is a header of column names. -- Feature columns must be numeric; parsed as float32 (halves memory for - the 10^5-10^6-row tables this path exists for; both backends compute - in float32/16 internally anyway). An empty cell, ``NA``, ``N/A``, - ``null``, ``None`` or ``NaN`` is a missing value (NaN), which both - backends handle. -- For training CSVs, one column is the target (``target_column``, or the - LAST column when unspecified). Target values parse as strings for - classification and floats for regression; with no task given, the - values decide (labels.infer_task). -- No quoted fields containing commas (numpy's C parser; numeric tables - don't need them). - -Malformed content raises InvalidInputError naming the problem -- a -caller's bad file must 422, never 500. -""" - -from __future__ import annotations - -from pathlib import Path - -import numpy as np - -from tabctx.errors import InvalidInputError -from tabctx.labels import infer_task -from tabctx.types import Task - - -def _read_header(path: Path) -> list[str]: - with open(path, encoding="utf-8-sig") as f: - first = f.readline().strip() - if not first: - raise InvalidInputError("CSV is empty (expected a header row)") - return [c.strip() for c in first.split(",")] - - -# Spellings of a missing value, compared case-insensitively after strip. -_MISSING = frozenset({"", "na", "n/a", "nan", "null", "none"}) - - -def _cell(text: str) -> float: - text = text.strip() - return np.nan if text.lower() in _MISSING else float(text) - - -def _load_float_columns(path: Path, usecols: list[int], what: str) -> np.ndarray: - load = { - "delimiter": ",", - "skiprows": 1, - "usecols": usecols, - "dtype": np.float32, - "ndmin": 2, - } - try: - arr = np.loadtxt(path, **load) - except ValueError: - # numpy's fast parser rejects empty cells and NA spellings; parse - # again cell by cell (slower, so only for files that need it), - # still rejecting text that isn't a number or a missing value. - try: - arr = np.loadtxt(path, converters=_cell, **load) - except ValueError as e: - raise InvalidInputError( - f"{what}: non-numeric or ragged CSV data ({e})" - ) from e - if arr.shape[0] == 0: - raise InvalidInputError(f"{what}: CSV has a header but no data rows") - return arr - - -def _read_text_column(path: Path, index: int) -> list[str]: - try: - arr = np.loadtxt( - path, delimiter=",", skiprows=1, usecols=[index], dtype=str, ndmin=2 - ) - except ValueError as e: - raise InvalidInputError(f"train target: unreadable CSV column ({e})") from e - labels = [str(v).strip() for v in arr[:, 0]] - if "" in labels: - raise InvalidInputError( - f"train target: row {labels.index('') + 1} has no label; every " - "training row needs one" - ) - return labels - - -def _reject_missing_numeric_labels(labels: list[str]) -> None: - """In a numeric target column, NA/null/NaN is a missing label, not a - class called "NA": every training row needs one.""" - missing = [i for i, t in enumerate(labels) if t.lower() in _MISSING] - if missing and _floats_or_none([t for t in labels if t.lower() not in _MISSING]): - raise InvalidInputError( - f"train target: row {missing[0] + 1} has no label ({labels[missing[0]]!r}); " - "every training row needs one" - ) - - -def _floats_or_none(texts: list[str]) -> list[float] | None: - try: - return [float(t) for t in texts] - except ValueError: - return None - - -def parse_train_csv( - path: Path, task: Task | None, target_column: str | None = None -) -> tuple[np.ndarray, list, list[str]]: - """Returns (X float32 2D, y list, feature_names).""" - header = _read_header(path) - if target_column is None: - target_column = header[-1] - if target_column not in header: - raise InvalidInputError( - f"target_column {target_column!r} not in CSV header {header}" - ) - target_idx = header.index(target_column) - feature_idx = [i for i in range(len(header)) if i != target_idx] - if not feature_idx: - raise InvalidInputError( - "training CSV needs at least one feature column besides the target" - ) - feature_names = [header[i] for i in feature_idx] - - X = _load_float_columns(path, feature_idx, "train features") - - y: list - if task is None: - # CSV has no types: a target column that parses as numbers is - # numbers, and those decide the task; anything else is classes. - y = _read_text_column(path, target_idx) - _reject_missing_numeric_labels(y) - values = _floats_or_none(y) - if values is not None and infer_task(values) == "regression": - y = values - elif task == "regression": - y_arr = _load_float_columns(path, [target_idx], "train target") - y = [float(v) for v in y_arr[:, 0]] - else: - y = _read_text_column(path, target_idx) - - if len(y) != X.shape[0]: # pragma: no cover - # Defensive: X and y are parsed independently (same file, same - # skiprows) so a real CSV can't make them disagree -- found no - # way to construct one in testing. Kept as a guard rather than an - # assert so a future parsing change that could break this - # invariant fails as a clean 422, not a silent shape mismatch. - raise InvalidInputError( - f"target column has {len(y)} values but features have {X.shape[0]} rows" - ) - return X, y, feature_names - - -def parse_features_csv( - path: Path, expected_features: list[str] | None = None -) -> np.ndarray: - """Returns a float32 2D array of all columns. When expected_features - is given (from the training upload), the header must match it -- a - silently reordered or missing column would produce garbage - predictions, which is far worse than a 422.""" - header = _read_header(path) - if expected_features is not None and header != expected_features: - raise InvalidInputError( - f"test CSV columns {header} do not match the training features " - f"{expected_features} (same names, same order, no target column)" - ) - return _load_float_columns(path, list(range(len(header))), "test features") diff --git a/src/tabctx/serve/files.py b/src/tabctx/serve/files.py new file mode 100644 index 0000000..1ceee5a --- /dev/null +++ b/src/tabctx/serve/files.py @@ -0,0 +1,332 @@ +"""Table files -> columns: CSV (plain, gzip or zstd) and Parquet. + +A fit or predict can carry its table as a file (a multipart upload, or +an earlier /v1/tabctx/upload referenced by id). The format comes from +the content, not the filename: Parquet starts with ``PAR1``, gzip and +zstd with their magic bytes, anything else is read as UTF-8 CSV with a +header row. Parquet keeps types and nulls and is several times smaller, +so it is the better way to send a large table. + +Parsing lives at the serving layer on purpose: the engine keeps taking +in-memory arrays, so transport is a serving concern. + +Features: + +- A missing value is a Parquet null, or a CSV cell that is empty or + ``NA``, ``N/A``, ``null``, ``None`` or ``NaN`` (any case, surrounding + spaces ignored). +- Numbers stay numbers; dates and times become seconds (since the epoch, + or since midnight for a time of day); any other text makes its column + a text column, encoded by tabctx.schema. A Parquet string column is + text even when its values look like numbers ("007"); a CSV has no + types, so there a column of number-like text is numbers. + +Labels (the target column) keep their text as written: ``"1.0"`` and +``"01"`` are classes of their own, and ``"NA"`` can be a class among text +labels. A target whose present values are all numbers is numeric: a +missing value among them is a missing label (an error: every training +row needs one), and the numbers decide the task when none is given +(labels.infer_task). Regression needs numbers. + +Malformed content raises InvalidInputError naming the problem, so a +caller's bad file is a 422, never a 500. pyarrow is imported on first +use, so deployments that never read files don't need it. +""" + +from __future__ import annotations + +import csv as stdlib_csv +from collections.abc import Collection +from dataclasses import dataclass +from pathlib import Path + +import numpy as np + +from tabctx.errors import InvalidInputError +from tabctx.labels import infer_task +from tabctx.types import Task + +# Spellings of a missing value, compared case-insensitively after strip. +_MISSING = frozenset({"", "na", "n/a", "nan", "null", "none"}) +# The same for arrow's CSV parser, which compares exactly (no strip). +_ARROW_NULLS = sorted( + {f(m) for m in _MISSING for f in (str.lower, str.upper, str.title)} +) + +_PARQUET_MAGIC = b"PAR1" +_COMPRESSION_MAGIC = {b"\x1f\x8b": "gzip", b"\x28\xb5\x2f\xfd": "zstd"} + + +@dataclass(frozen=True) +class Table: + """Feature columns: float32 arrays (NaN = missing), or object arrays + of text and None.""" + + names: list[str] + columns: list[np.ndarray] + + +def read_table(path: Path, text_columns: Collection[str] = ()) -> Table: + """A table file's columns as features. text_columns: names to keep as + text whatever their values look like (a training text column must + stay text at predict: "007" is not 7).""" + names, arrow = _read(path, text_columns) + keep = _kept_text(path, names, text_columns) + return Table(names, list(_features(arrow, names, keep))) + + +def read_training_table( + path: Path, target: str | None, task: Task | None +) -> tuple[Table, list]: + """(features, labels) of a training table file. target: the label + column's name, default the last column. Labels are floats for + regression and text for classification.""" + pa = _pyarrow() + if target is None and not _is_parquet(path): + # Known before reading, so the label column is read as text. + header = _csv_header(path) + target = header[-1].strip() if header else None + names, arrow = _read(path, (), raw_text=(target,) if target else ()) + target = names[-1] if target is None else target + if target not in names: + raise InvalidInputError( + f"target column {target!r} is not in the table's columns {names[:20]}" + ) + if len(names) < 2: + raise InvalidInputError( + "the training table needs at least one feature column besides the target" + ) + index = names.index(target) + labels = _labels(arrow.column(index), task, pa) + arrow = arrow.remove_column(index) + del names[index] + keep = _kept_text(path, names, ()) + return Table(names, list(_features(arrow, names, keep))), labels + + +# ---- labels ----------------------------------------------------------------- + + +def _labels(col, task: Task | None, pa) -> list: + kind = col.type + if pa.types.is_string(kind) or pa.types.is_large_string(kind): + return _text_labels( + [None if v is None else v.strip() for v in col.to_pylist()], task + ) + if pa.types.is_dictionary(kind): + return _text_labels(col.cast(kind.value_type).to_pylist(), task) + values = col.to_pylist() + missing = [i for i, v in enumerate(values) if v is None or v != v] + if missing: + _no_label(missing[0]) + numeric = ( + pa.types.is_integer(kind) + or pa.types.is_floating(kind) + or pa.types.is_decimal(kind) + ) + if not numeric: + if task == "regression": + raise InvalidInputError( + f"regression needs numeric labels; the target has {kind}" + ) + return [str(v) for v in values] + numbers = col.cast(pa.float64()).to_pylist() + return _numeric_labels([str(v) for v in values], numbers, task) + + +def _text_labels(texts: list[str | None], task: Task | None) -> list: + for row, text in enumerate(texts): + if text is None or text == "": + _no_label(row) + present = [t for t in texts if t.lower() not in _MISSING] + numbers = _floats_or_none(present) + if numbers is None: + if task == "regression": + example = next(t for t in texts if _floats_or_none([t]) is None) + raise InvalidInputError( + f"regression needs numeric labels; the target has text ({example!r})" + ) + return texts # text classes: "NA" among them is a class + if len(present) < len(texts): # "NA" among numbers is a missing label + _no_label(next(i for i, t in enumerate(texts) if t.lower() in _MISSING)) + return _numeric_labels(texts, numbers, task) + + +def _numeric_labels(texts: list[str], numbers: list[float], task: Task | None) -> list: + if task is None: + task = infer_task(numbers) + return numbers if task == "regression" else texts + + +def _floats_or_none(texts: list[str]) -> list[float] | None: + try: + return [float(t) for t in texts] + except ValueError: + return None + + +def _no_label(row: int) -> None: + raise InvalidInputError( + f"train target: row {row + 1} has no label; every training row needs one" + ) + + +# ---- features --------------------------------------------------------------- + + +def _kept_text(path: Path, names: list[str], text_columns: Collection[str]): + """Columns whose text stays text even when it looks like numbers: all + of a Parquet file's (it has types: a string column is text), and in a + CSV, which has none, the named ones.""" + return set(names) if _is_parquet(path) else set(text_columns) + + +def _features(arrow, names: list[str], text_columns: Collection[str]): + """Each column converted in turn.""" + for name in names: + col = arrow.column(0) + arrow = arrow.remove_column(0) + yield _feature(col, name, name in text_columns) + + +def _feature(col, name: str, keep_text: bool) -> np.ndarray: + import pyarrow as pa + + kind = col.type + try: + if pa.types.is_dictionary(kind): + col = col.cast(kind.value_type) + kind = col.type + if pa.types.is_string(kind) or pa.types.is_large_string(kind): + return _text_feature(col.to_pylist(), numeric_ok=not keep_text) + if pa.types.is_temporal(kind): + # float64: the schema stores them from an offset (tabctx.schema). + return _seconds(col, pa).to_numpy() + return col.cast(pa.float32(), safe=False).to_numpy() + except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError) as e: + what = "not UTF-8 text" if pa.types.is_binary(kind) else f"of type {kind}" + raise InvalidInputError( + f"column {name!r} is {what}, which is neither numbers nor text" + ) from e + + +def _seconds(col, pa): + """Dates and times as seconds: since the epoch, or since midnight for + a time of day; durations as their length.""" + import pyarrow.compute as pc + + kind = col.type + if pa.types.is_time(kind): + micros = col.cast(pa.time64("us")).cast(pa.int64()) + elif pa.types.is_duration(kind): + micros = col.cast(pa.duration("us"), safe=False).cast(pa.int64()) + else: + micros = col.cast(pa.timestamp("us"), safe=False).cast(pa.int64()) + return pc.divide(micros.cast(pa.float64()), 1e6) + + +def _text_feature(values: list, numeric_ok: bool) -> np.ndarray: + """Stripped, missing spellings -> None; then, if allowed and every + present value is a number (" 3", "NA" and "4.5" in a CSV that arrow + couldn't type), a numeric column.""" + cells = [ + None if v is None or v.strip().lower() in _MISSING else v.strip() + for v in values + ] + if numeric_ok: + try: + return np.array( + [np.nan if v is None else float(v) for v in cells], dtype=np.float32 + ) + except ValueError: + pass + out = np.empty(len(cells), dtype=object) + out[:] = cells + return out + + +# ---- reading ---------------------------------------------------------------- + + +def _pyarrow(): + try: + import pyarrow as pa + except ImportError as e: # pragma: no cover -- a deployment packaging error + raise InvalidInputError( + "this deployment can't read table files (pyarrow is not installed); " + "send the table as JSON rows" + ) from e + return pa + + +def _is_parquet(path: Path) -> bool: + with open(path, "rb") as f: + return f.read(4) == _PARQUET_MAGIC + + +def _read(path: Path, text_columns: Collection[str], raw_text: Collection[str] = ()): + """(column names, arrow table). raw_text: columns read as text with no + missing-value spellings applied (labels).""" + pa = _pyarrow() + try: + arrow = _read_arrow(path, text_columns, raw_text) + except (pa.ArrowInvalid, pa.ArrowNotImplementedError, OSError) as e: + raise InvalidInputError(f"could not read the table file: {e}") from e + names = [str(n).strip() for n in arrow.column_names] + if not names: + raise InvalidInputError("the table file has no columns") + duplicated = sorted({n for n in names if names.count(n) > 1}) + if duplicated: + raise InvalidInputError(f"the table file repeats column name(s) {duplicated}") + if arrow.num_rows == 0: + raise InvalidInputError("the table file has a header but no rows") + return names, arrow + + +def _open_csv(path: Path): + """The file as a stream of CSV bytes, decompressed if it's gzip or zstd.""" + import pyarrow as pa + + with open(path, "rb") as f: + head = f.read(4) + source = pa.OSFile(str(path)) + compression = _COMPRESSION_MAGIC.get(head[:2]) or _COMPRESSION_MAGIC.get(head) + if compression is not None: + source = pa.CompressedInputStream(source, compression) + return source + + +def _csv_header(path: Path) -> list[str]: + """A CSV's column names as written (unstripped), read ahead of the + table so columns can be typed by name.""" + line = b"" + with _open_csv(path) as source: + while b"\n" not in line and (block := source.read(1 << 16)): + line += block + try: + text = line.split(b"\n", 1)[0].rstrip(b"\r").decode("utf-8-sig") + except UnicodeDecodeError: + return [] # reading the table reports it + return next(stdlib_csv.reader([text]), []) + + +def _read_arrow(path: Path, text_columns: Collection[str], raw_text: Collection[str]): + import pyarrow as pa + import pyarrow.csv as csv + import pyarrow.parquet as pq + + if _is_parquet(path): + return pq.read_table(path) + # Arrow matches column_types by the header's own (unstripped) spelling. + wanted = set(text_columns) | set(raw_text) + convert = csv.ConvertOptions( + null_values=_ARROW_NULLS, + # Text stays text here: missing spellings in text feature columns + # are mapped by _text_feature, and labels keep theirs. + strings_can_be_null=False, + column_types={ + raw: pa.string() for raw in _csv_header(path) if raw.strip() in wanted + }, + ) + with _open_csv(path) as source: + return csv.read_csv(source, convert_options=convert) diff --git a/src/tabctx/serve/multipart.py b/src/tabctx/serve/multipart.py new file mode 100644 index 0000000..bea7ff6 --- /dev/null +++ b/src/tabctx/serve/multipart.py @@ -0,0 +1,171 @@ +"""A multipart/form-data body read as it arrives. + +Text fields go to memory (few and small, both capped); the one ``file`` +part goes straight to an upload writer (serve/uploads.py), which enforces +the upload size cap as bytes arrive. Nothing is spooled or copied first, +so a body over the cap stops at the cap, and a good one is written to +disk once. Disk writes run off the event loop. + +Fields are strict UTF-8 (a charset parameter naming anything else is +refused) and each appears once, so a gateway that routes on them reads +exactly the values this server uses. Malformed bodies raise +InvalidInputError (a 422). +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import AsyncIterator, Callable +from typing import Protocol + +from python_multipart.exceptions import FormParserError +from python_multipart.multipart import MultipartParser, parse_options_header + +from tabctx.errors import InvalidInputError + +# python-multipart logs every malformed body it meets; the 422 says it. +logging.getLogger("python_multipart").setLevel(logging.ERROR) + +FILE_FIELD = "file" +MAX_FIELDS = 16 +MAX_FIELD_BYTES = 64 * 1024 + + +class Writer(Protocol): + def write(self, chunk: bytes) -> None: ... + def abort(self) -> None: ... + + +async def read_form( + stream: AsyncIterator[bytes], content_type: str, begin_file: Callable[[], Writer] +) -> tuple[dict[str, str], Writer | None]: + """(text fields, the file part's writer or None). The caller finishes + the writer; on any error it has been aborted.""" + _, params = parse_options_header(content_type) + charset = params.get(b"charset", b"utf-8").lower() + if charset not in (b"utf-8", b"utf8"): + # Fields are UTF-8, always: a proxy in front that read them in + # another charset would route by values this server never sees. + raise InvalidInputError( + "multipart form fields must be UTF-8 (no other charset)" + ) + boundary = params.get(b"boundary") + if not boundary: + raise InvalidInputError("multipart body without a boundary") + form = _Form(begin_file) + try: + parser = MultipartParser(boundary, form.callbacks()) + async for chunk in stream: + parser.write(chunk) + if form.pending: + await asyncio.to_thread(form.flush) + parser.finalize() + if form.in_part: + raise InvalidInputError("the multipart body ends inside a part") + form.flush() + except FormParserError as e: + if form.writer is not None: + form.writer.abort() + raise InvalidInputError(f"malformed multipart body: {e}") from e + except BaseException: + if form.writer is not None: + form.writer.abort() + raise + return form.fields, form.writer + + +class _Form: + def __init__(self, begin_file: Callable[[], Writer]) -> None: + self._begin_file = begin_file + self.fields: dict[str, str] = {} + self.writer: Writer | None = None + self.pending: list[bytes] = [] # file bytes parsed, not yet written + self.in_part = False + self.open_part = "" + self._n_fields = 0 + self._in_file = False + self._data = bytearray() + self._header_name = b"" + self._header_value = b"" + self._disposition = b"" + + def flush(self) -> None: + pending, self.pending = self.pending, [] + for chunk in pending: + self.writer.write(chunk) + + def callbacks(self) -> dict: + return { + "on_part_begin": self._part_begin, + "on_header_field": self._header_field, + "on_header_value": self._header_value_part, + "on_header_end": self._header_end, + "on_headers_finished": self._headers_finished, + "on_part_data": self._part_data, + "on_part_end": self._part_end, + } + + def _part_begin(self) -> None: + self.in_part = True + self._data = bytearray() + self._disposition = b"" + self._in_file = False + + def _header_field(self, data: bytes, start: int, end: int) -> None: + self._header_name += data[start:end] + + def _header_value_part(self, data: bytes, start: int, end: int) -> None: + self._header_value += data[start:end] + + def _header_end(self) -> None: + if self._header_name.lower() == b"content-disposition": + self._disposition = self._header_value + self._header_name = self._header_value = b"" + + def _headers_finished(self) -> None: + _, options = parse_options_header(self._disposition) + if b"name" not in options: + raise InvalidInputError("a multipart part has no name") + name = options[b"name"].decode("utf-8", "replace") + self.open_part = name + if b"filename" in options: + if name != FILE_FIELD: + raise InvalidInputError( + f"send the table as the {FILE_FIELD!r} part, not {name!r}" + ) + if self.writer is not None: + raise InvalidInputError("a request carries one table file") + self.writer = self._begin_file() + self._in_file = True + elif name == FILE_FIELD: + raise InvalidInputError( + f"send the table as a file part (curl: -F {FILE_FIELD}=@train.csv)" + ) + else: + self._n_fields += 1 + if self._n_fields > MAX_FIELDS: + raise InvalidInputError(f"more than {MAX_FIELDS} form fields") + + def _part_data(self, data: bytes, start: int, end: int) -> None: + if self._in_file: + self.pending.append(bytes(data[start:end])) + return + self._data += data[start:end] + if len(self._data) > MAX_FIELD_BYTES: + raise InvalidInputError( + f"form field {self.open_part!r} is over {MAX_FIELD_BYTES} bytes" + ) + + def _part_end(self) -> None: + if not self._in_file: + if self.open_part in self.fields: + # One value per field, so no reader can pick a different one. + raise InvalidInputError(f"form field {self.open_part!r} is repeated") + try: + self.fields[self.open_part] = self._data.decode("utf-8") + except UnicodeDecodeError as e: + raise InvalidInputError( + f"form field {self.open_part!r} is not UTF-8 text" + ) from e + self.in_part = False diff --git a/src/tabctx/serve/schemas.py b/src/tabctx/serve/schemas.py new file mode 100644 index 0000000..94fb6ca --- /dev/null +++ b/src/tabctx/serve/schemas.py @@ -0,0 +1,83 @@ +"""The schemas of a replica's recently fitted contexts, in its own process. + +A predict encodes its rows with the training table's schema +(tabctx.schema). The engine keeps the schema with the context, but on a +multi-GPU replica asking for it is a round trip to a worker process +(and a large text column makes the schema large), so the serving layer +keeps the schemas it recorded at fit. Session affinity routes a +dataset's fits and predicts to the same replica, so this copy is current +for every context fit since the replica started; a miss (after a +restart, or past the bounds) falls back to the engine. A cached schema +outliving its context is harmless: the predict still gets the engine's +404. + +Fits of one dataset id are recorded one at a time (fitting()): the +engine serializes them too, so the schema recorded last is the resident +context's. A failed fit leaves the engine's previous context serving and +no cached schema, so predicts ask the engine for the one that matches. +""" + +from __future__ import annotations + +import contextlib +import threading +from collections import OrderedDict +from collections.abc import Callable, Iterator + +from tabctx.keyed_locks import KeyedLocks +from tabctx.schema import TableSchema + +DEFAULT_MAX_ENTRIES = 1024 +DEFAULT_MAX_BYTES = 64 * 1024**2 + + +class SchemaCache: + """A thread-safe LRU of dataset id -> TableSchema, bounded in entries + and in (approximate) bytes; a schema over the byte bound isn't kept.""" + + def __init__( + self, max_entries: int = DEFAULT_MAX_ENTRIES, max_bytes: int = DEFAULT_MAX_BYTES + ) -> None: + self._max_entries = max_entries + self._max_bytes = max_bytes + self._entries: OrderedDict[str, TableSchema] = OrderedDict() + self._bytes = 0 + self._lock = threading.Lock() + self._fits = KeyedLocks() + + @contextlib.contextmanager + def fitting(self, dataset_id: str) -> Iterator[Callable[[TableSchema], None]]: + """Around a fit of dataset_id: yields the function that records its + schema once the fit succeeds.""" + with self._fits.hold(dataset_id): + self._drop(dataset_id) + yield lambda schema: self._put(dataset_id, schema) + + def get(self, dataset_id: str) -> TableSchema | None: + with self._lock: + schema = self._entries.get(dataset_id) + if schema is not None: + self._entries.move_to_end(dataset_id) + return schema + + def _put(self, dataset_id: str, schema: TableSchema) -> None: + size = schema.approx_bytes + if size > self._max_bytes: + return + with self._lock: + old = self._entries.pop(dataset_id, None) + if old is not None: + self._bytes -= old.approx_bytes + self._entries[dataset_id] = schema + self._bytes += size + while ( + len(self._entries) > self._max_entries or self._bytes > self._max_bytes + ): + _, old = self._entries.popitem(last=False) + self._bytes -= old.approx_bytes + + def _drop(self, dataset_id: str) -> None: + with self._lock: + old = self._entries.pop(dataset_id, None) + if old is not None: + self._bytes -= old.approx_bytes diff --git a/src/tabctx/serve/tables.py b/src/tabctx/serve/tables.py index 9b463a3..043fc3e 100644 --- a/src/tabctx/serve/tables.py +++ b/src/tabctx/serve/tables.py @@ -1,9 +1,10 @@ """Inline JSON tables -> arrays for the engine. -JSON has no NaN, so a missing cell arrives as ``null`` (None). Both -backends handle missing values natively, so a null becomes NaN here and -the table goes on as a float32 array. Converting once at the edge also -spares the engine its per-row Python checks for large inline tables. +JSON has no NaN, so a missing cell arrives as ``null`` (None) and becomes +NaN; both backends handle missing values natively. A cell may also be +text, which makes its column a text column (tabctx.schema). An all-number +table (the common case) converts in one numpy call, so large numeric +tables pay no per-cell Python cost. """ from __future__ import annotations @@ -12,15 +13,64 @@ import numpy as np +from tabctx.schema import TableSchema -def rows_to_array(rows: Sequence[Sequence[float | None]]) -> np.ndarray | Sequence: - """A float array with null -> NaN, or the rows unchanged when they - aren't a rectangle (ragged, empty), so the engine's own validation - reports the problem in its usual terms.""" +Rows = Sequence[Sequence[float | str | None]] + + +def rows_to_array(rows: Rows) -> np.ndarray | None: + """An all-number table as float32 (null -> NaN), or None when it has + text, isn't a rectangle, or is empty.""" + # numpy would read "007" and "nan" as numbers; a text cell makes a + # text column whatever it says (~10 ns a cell). + if any(str in map(type, row) for row in rows): + return None try: - # float32, as CSV tables are parsed: both backends compute in + # float32, as files are encoded: both backends compute in # float32/16, and it halves what crosses to the worker processes. arr = np.asarray(rows, dtype=np.float32) except (TypeError, ValueError): + return None + return arr if arr.ndim == 2 and arr.size else None + + +def learn_rows(rows: Rows) -> tuple[np.ndarray | Rows, TableSchema | None]: + """A training table, encoded, and the schema to encode test rows with. + A table that isn't a non-empty rectangle comes back as is, with no + schema, for the engine to reject in its usual terms.""" + arr = rows_to_array(rows) + if arr is not None: + return arr, TableSchema((None,) * arr.shape[1]) + columns = _columns(rows) + if columns is None: + return rows, None + schema, X = TableSchema.learn(columns) + return X, schema + + +def encode_rows(rows: Rows, schema: TableSchema | None) -> np.ndarray | Rows: + """Test rows encoded as the training table was.""" + arr = rows_to_array(rows) + if arr is not None and (schema is None or not (schema.has_text or schema.offsets)): + return arr + columns = _columns(rows) + if columns is None: return rows - return arr if arr.ndim == 2 else rows + if schema is None: + # A context fit before schemas existed: numbers only. + schema = TableSchema((None,) * len(columns)) + return schema.encode(columns) + + +def _columns(rows: Rows) -> list[np.ndarray] | None: + if not rows or not rows[0]: + return None + width = len(rows[0]) + if any(len(row) != width for row in rows): + return None + columns = [] + for values in zip(*rows, strict=True): + col = np.empty(len(values), dtype=object) + col[:] = values + columns.append(col) + return columns diff --git a/src/tabctx/serve/uploads.py b/src/tabctx/serve/uploads.py index bcb1138..51333b0 100644 --- a/src/tabctx/serve/uploads.py +++ b/src/tabctx/serve/uploads.py @@ -196,6 +196,15 @@ def commit(self, tenant_id: str | None = None) -> UploadRecord: self._store._register(record) return record + def finish(self) -> Path: + """Close the file and hand it to the caller, unregistered: for a + table sent with the request that uses it, which the caller reads + and then discard()s.""" + assert self._open, "writer already committed/aborted" + self._open = False + self._file.close() + return self._path + def abort(self) -> None: if self._open: self._open = False diff --git a/src/tabctx/serve/workers.py b/src/tabctx/serve/workers.py index 259fad2..eb62b27 100644 --- a/src/tabctx/serve/workers.py +++ b/src/tabctx/serve/workers.py @@ -55,6 +55,7 @@ from tabctx import timing from tabctx.errors import BackendComputeError +from tabctx.schema import TableSchema from tabctx.types import ArrayLike, EngineStats, PredictOutcome, Task logger = logging.getLogger(__name__) @@ -65,7 +66,7 @@ "fit", "predict", "evict", - "feature_names", + "schema", "estimate_fit_bytes", "report", } @@ -636,7 +637,7 @@ 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: return self._call( "fit", @@ -645,7 +646,7 @@ def fit( task=task, dataset_id=dataset_id, backend=backend, - feature_names=feature_names, + schema=schema, ) def predict( @@ -653,8 +654,8 @@ def predict( ) -> PredictOutcome: return self._call("predict", dataset_id, X_test, return_proba=return_proba) - def feature_names(self, dataset_id: str) -> list[str] | None: - return self._call("feature_names", dataset_id) + def schema(self, dataset_id: str) -> TableSchema | None: + return self._call("schema", dataset_id) def evict(self, dataset_id: str) -> None: self._call("evict", dataset_id) diff --git a/tests/integration/test_table_files.py b/tests/integration/test_table_files.py new file mode 100644 index 0000000..e1e5a99 --- /dev/null +++ b/tests/integration/test_table_files.py @@ -0,0 +1,218 @@ +"""A table file in the fit/predict request itself (multipart/form-data), +and text columns, on a real local 2-replica deployment -- the curl path: + + curl $URL/v1/tabctx/fit -F dataset_id=churn -F target=churned -F file=@train.csv +""" + +import gzip +import io + +import pytest + +pytest.importorskip("ray") +httpx = pytest.importorskip("httpx") +pa = pytest.importorskip("pyarrow") +import pyarrow.parquet as pq # noqa: E402 + +TRAIN_CSV = ( + "age,plan,churned\n" + + "".join( + f"{20 + i},{'gold' if i % 2 else 'basic'},{'yes' if i < 7 else 'no'}\n" + for i in range(10) + ) +).encode() + + +def post(url, path, dataset_id, **kwargs): + return httpx.post( + f"{url}{path}", headers={"x-session-id": dataset_id}, timeout=60, **kwargs + ) + + +def fit_file(url, dataset_id, content, **fields): + return post( + url, + "/v1/tabctx/fit", + dataset_id, + data={"dataset_id": dataset_id, **fields}, + files={"file": ("train", content)}, + ) + + +def predict_file(url, dataset_id, content, **fields): + return post( + url, + "/v1/tabctx/predict", + dataset_id, + data={"dataset_id": dataset_id, **fields}, + files={"file": ("test", content)}, + ) + + +def parquet(table) -> bytes: + buf = io.BytesIO() + pq.write_table(table, buf, compression="zstd") + return buf.getvalue() + + +def test_fit_and_predict_with_a_file_in_one_call(two_replica_service): + url, ds = two_replica_service, "file-csv" + fit = fit_file(url, ds, TRAIN_CSV, target="churned") + assert fit.status_code == 200, fit.text + body = fit.json() + assert (body["dataset_id"], body["n_train"], body["n_features"]) == (ds, 10, 2) + assert body["task"] == "classification" + + # Columns by name, any order; text encoded as at fit; unseen -> missing. + test = b"plan,age\ngold,30\nplatinum,\n" + pred = predict_file(url, ds, test, return_proba="true") + assert pred.status_code == 200, pred.text + assert pred.json()["predictions"] == ["yes", "yes"] + assert pred.json()["classes"] == ["no", "yes"] + + +def test_gzip_csv_and_parquet_files(two_replica_service): + url, ds = two_replica_service, "file-formats" + assert fit_file(url, ds, gzip.compress(TRAIN_CSV)).status_code == 200 + table = pa.table({"age": [41.0, None], "plan": ["basic", None]}) + pred = predict_file(url, ds, parquet(table)) + assert pred.status_code == 200, pred.text + assert pred.json()["n_test"] == 2 + + +def test_text_cells_in_json_rows(two_replica_service): + url, ds = two_replica_service, "json-text" + fit = post( + url, + "/v1/tabctx/fit", + ds, + json={ + "dataset_id": ds, + "train_X": [[1.5, "gold", None], [2.0, "basic", 3.0], [None, "gold", 1.0]], + "train_y": ["yes", "no", "yes"], + }, + ) + assert fit.status_code == 200, fit.text + pred = post( + url, + "/v1/tabctx/predict", + ds, + json={"dataset_id": ds, "test_X": [[1.0, "basic", 2.0], [None, "new", None]]}, + ) + assert pred.status_code == 200, pred.text + assert pred.json()["predictions"] == ["yes", "yes"] + # Text where training had numbers is a clear 422. + bad = post( + url, + "/v1/tabctx/predict", + ds, + json={"dataset_id": ds, "test_X": [["x", "gold", 1.0]]}, + ) + assert bad.status_code == 422 and "held numbers" in bad.text + + +@pytest.mark.parametrize( + ("request_kwargs", "status", "match"), + [ + ( + {"data": {"dataset_id": "bad"}, "files": {"other": ("x", b"a\n1\n")}}, + 422, + "'file' part", + ), + ( + { + "data": {"dataset_id": "bad", "task": "nope"}, + "files": {"file": ("x", TRAIN_CSV)}, + }, + 422, + "task", + ), + ( + { + "data": {"dataset_id": "bad"}, + "files": {"file": ("x", b"\x00\x01garbage")}, + }, + 422, + "table", + ), + ( + { + "data": {"dataset_id": "bad", "train_upload_id": "abc"}, + "files": {"file": ("x", TRAIN_CSV)}, + }, + 422, + "got several", + ), + ], +) +def test_bad_multipart_fits_are_422(two_replica_service, request_kwargs, status, match): + resp = post(two_replica_service, "/v1/tabctx/fit", "bad", **request_kwargs) + assert resp.status_code == status, resp.text + assert match in resp.text + + +def test_the_openapi_schema_documents_both_body_types(two_replica_service): + spec = httpx.get(f"{two_replica_service}/openapi.json").json() + content = spec["paths"]["/v1/tabctx/fit"]["post"]["requestBody"]["content"] + assert set(content) == {"application/json", "multipart/form-data"} + assert "file" in content["multipart/form-data"]["schema"]["required"] + + +def test_an_unknown_context_is_404_whatever_the_rows(two_replica_service): + # 404 is what tells a client to fit again; text rows must not turn it + # into a 422 by being encoded against a schema that isn't there. + url = two_replica_service + assert predict_file(url, "never-fit", b"plan,age\ngold,1\n").status_code == 404 + inline = post( + url, + "/v1/tabctx/predict", + "never-fit", + json={"dataset_id": "never-fit", "test_X": [[1.0, "gold"]]}, + ) + assert inline.status_code == 404 + + +def test_an_invalid_request_leaves_its_upload_unused(two_replica_service): + from tabctx.client import TabctxClient + + url, ds = two_replica_service, "upload-kept" + upload_id = TabctxClient(url).upload_csv(TRAIN_CSV, ds) + bad = post( + url, + "/v1/tabctx/fit", + ds, + json={"dataset_id": ds, "train_upload_id": upload_id, "train_y": ["a"]}, + ) + assert bad.status_code == 422 and "got several" in bad.text + good = post( + url, "/v1/tabctx/fit", ds, json={"dataset_id": ds, "train_upload_id": upload_id} + ) + assert good.status_code == 200, good.text + + +def test_an_oversized_body_is_refused_from_its_headers(two_replica_service): + import http.client + from urllib.parse import urlparse + + target = urlparse(two_replica_service) + conn = http.client.HTTPConnection(target.hostname, target.port, timeout=30) + conn.putrequest("POST", "/v1/tabctx/fit") + conn.putheader("Content-Type", "multipart/form-data; boundary=x") + conn.putheader("Content-Length", str(64 * 1024**3)) + conn.putheader("x-session-id", "huge") + conn.endheaders() + conn.send(b"--x\r\n") # a few bytes of 64 GiB: answered without the rest + resp = conn.getresponse() + assert resp.status == 413 + conn.close() + + +def test_validation_errors_keep_fastapis_shape(two_replica_service): + resp = post( + two_replica_service, + "/v1/tabctx/fit", + "shape", + json={"dataset_id": "shape", "task": "nope", "train_X": [[1]], "train_y": [1]}, + ) + assert resp.status_code == 422 + assert resp.json()["detail"][0]["loc"] == ["body", "task"] diff --git a/tests/integration/test_upload_flow.py b/tests/integration/test_upload_flow.py index 71b2267..5e60872 100644 --- a/tests/integration/test_upload_flow.py +++ b/tests/integration/test_upload_flow.py @@ -70,18 +70,21 @@ def test_uploads_are_single_use(two_replica_service): client.fit_uploaded(upload_id, dataset_id) -def test_test_csv_schema_mismatch_is_422(two_replica_service): +def test_test_csv_columns_are_matched_by_name(two_replica_service): client = TabctxClient(two_replica_service) dataset_id = "upload-schema" upload_id = client.upload_csv(_train_csv(30, 3), dataset_id) client.fit_uploaded(upload_id, dataset_id) - # Wrong column ORDER with the right names -- silently accepting this - # would produce garbage predictions. - bad = b"f1,f0,f2\n1,2,3\n" - bad_upload = client.upload_csv(bad, dataset_id) - with pytest.raises(InvalidInputError, match="do not match"): - client.predict(dataset_id, test_upload_id=bad_upload) + # Any order, extra columns (such as the label) ignored ... + reordered = client.upload_csv(b"f2,label,f1,f0\n3,x,1,2\n", dataset_id) + assert client.predict(dataset_id, test_upload_id=reordered).predictions == ["hot"] + # ... but a missing training column is an error, not a guess. + missing = client.upload_csv(b"f1,f0\n1,2\n", dataset_id) + with pytest.raises( + InvalidInputError, match=r"missing training column\(s\) \['f2'\]" + ): + client.predict(dataset_id, test_upload_id=missing) def test_uploads_are_tenant_scoped(two_replica_service): @@ -99,7 +102,7 @@ def test_both_paths_in_one_request_rejected(two_replica_service): client = TabctxClient(two_replica_service) dataset_id = "upload-both" upload_id = client.upload_csv(_train_csv(30, 3), dataset_id) - with pytest.raises(InvalidInputError, match="not both"): + with pytest.raises(InvalidInputError, match="got several"): client._post( "/v1/tabctx/fit", { @@ -114,7 +117,7 @@ def test_both_paths_in_one_request_rejected(two_replica_service): def test_neither_path_rejected(two_replica_service): client = TabctxClient(two_replica_service) - with pytest.raises(InvalidInputError, match="either"): + with pytest.raises(InvalidInputError, match="got none"): client._post( "/v1/tabctx/fit", {"dataset_id": "upload-neither"}, diff --git a/tests/unit/test_csv_io.py b/tests/unit/test_csv_io.py deleted file mode 100644 index 1c67758..0000000 --- a/tests/unit/test_csv_io.py +++ /dev/null @@ -1,138 +0,0 @@ -"""Unit tests for CSV parsing (serve/csv_io.py).""" - -import numpy as np -import pytest - -from tabctx.errors import InvalidInputError -from tabctx.serve.csv_io import parse_features_csv, parse_train_csv - - -def _write(tmp_path, text, name="t.csv"): - p = tmp_path / name - p.write_text(text) - return p - - -class TestParseTrainCsv: - def test_classification_default_target_is_last_column(self, tmp_path): - p = _write(tmp_path, "f0,f1,label\n1,2,cat\n3,4,dog\n") - X, y, names = parse_train_csv(p, "classification") - assert X.dtype == np.float32 and X.shape == (2, 2) - assert X.tolist() == [[1.0, 2.0], [3.0, 4.0]] - assert y == ["cat", "dog"] - assert names == ["f0", "f1"] - - def test_named_target_column_anywhere(self, tmp_path): - p = _write(tmp_path, "label,f0,f1\nyes,1,2\nno,3,4\n") - X, y, names = parse_train_csv(p, "classification", target_column="label") - assert X.tolist() == [[1.0, 2.0], [3.0, 4.0]] - assert y == ["yes", "no"] - assert names == ["f0", "f1"] - - def test_regression_target_parsed_as_float(self, tmp_path): - p = _write(tmp_path, "f0,y\n1,0.5\n2,1.5\n") - _, y, _ = parse_train_csv(p, "regression") - assert y == [0.5, 1.5] - - def test_no_task_infers_regression_from_non_integer_targets(self, tmp_path): - p = _write(tmp_path, "f0,price\n1,212500.5\n2,180000\n") - _, y, _ = parse_train_csv(p, None) - assert y == [212500.5, 180000.0] - - def test_no_task_keeps_integer_and_text_targets_as_classes(self, tmp_path): - p = _write(tmp_path, "f0,label\n1,0\n2,1\n") - assert parse_train_csv(p, None)[1] == ["0", "1"] - p = _write(tmp_path, "f0,label\n1,cat\n2,dog\n", name="text.csv") - assert parse_train_csv(p, None)[1] == ["cat", "dog"] - - def test_missing_feature_values_become_nan(self, tmp_path): - p = _write(tmp_path, "f0,f1,f2,label\n1,,NA,x\nnull, n/a ,2,y\n3,NaN,None,x\n") - X, y, _ = parse_train_csv(p, "classification") - assert np.isnan(X).tolist() == [ - [False, True, True], - [True, True, False], - [False, True, True], - ] - assert X[0, 0] == 1.0 and X[1, 2] == 2.0 - assert y == ["x", "y", "x"] - - def test_text_that_is_not_a_missing_value_is_still_rejected(self, tmp_path): - p = _write(tmp_path, "f0,label\n1,x\nabc,y\n") - with pytest.raises(InvalidInputError, match="non-numeric"): - parse_train_csv(p, "classification") - - def test_a_blank_label_is_rejected(self, tmp_path): - p = _write(tmp_path, "f0,label\n1,x\n2,\n") - with pytest.raises(InvalidInputError, match="row 2 has no label"): - parse_train_csv(p, "classification") - - def test_na_in_a_numeric_target_column_is_a_missing_label(self, tmp_path): - p = _write(tmp_path, "f0,y\n1,2.5\n2,NA\n3,1.5\n") - with pytest.raises(InvalidInputError, match="row 2 has no label"): - parse_train_csv(p, None) - - def test_na_among_text_classes_is_a_class(self, tmp_path): - p = _write(tmp_path, "f0,label\n1,cat\n2,NA\n3,dog\n") - assert parse_train_csv(p, None)[1] == ["cat", "NA", "dog"] - - def test_missing_test_values_become_nan(self, tmp_path): - p = _write(tmp_path, "f0,f1\n,1\n2,NA\n") - assert np.isnan(parse_features_csv(p)).tolist() == [ - [True, False], - [False, True], - ] - - def test_missing_target_column_rejected(self, tmp_path): - p = _write(tmp_path, "f0,f1\n1,2\n") - with pytest.raises(InvalidInputError, match="nope"): - parse_train_csv(p, "classification", target_column="nope") - - def test_no_feature_columns_rejected(self, tmp_path): - p = _write(tmp_path, "label\ncat\n") - with pytest.raises(InvalidInputError): - parse_train_csv(p, "classification") - - def test_non_numeric_feature_rejected(self, tmp_path): - p = _write(tmp_path, "f0,label\noops,cat\n") - with pytest.raises(InvalidInputError): - parse_train_csv(p, "classification") - - def test_empty_file_rejected(self, tmp_path): - with pytest.raises(InvalidInputError): - parse_train_csv(_write(tmp_path, ""), "classification") - - def test_header_only_rejected(self, tmp_path): - with pytest.raises(InvalidInputError): - parse_train_csv(_write(tmp_path, "f0,label\n"), "classification") - - def test_ragged_target_column_rejected(self, tmp_path): - # Feature columns (0, 1) are present in every row, so they parse - # cleanly; the target column (index 2) is missing from the second - # data row -- must surface as a 422-worthy InvalidInputError, not - # an unhandled numpy ValueError. - p = _write(tmp_path, "a,b,label\n1,2,x\n3,4\n") - with pytest.raises(InvalidInputError, match="train target"): - parse_train_csv(p, "classification") - - def test_single_row_still_2d(self, tmp_path): - X, y, _ = parse_train_csv( - _write(tmp_path, "f0,f1,label\n1,2,cat\n"), "classification" - ) - assert X.shape == (1, 2) - assert y == ["cat"] - - -class TestParseFeaturesCsv: - def test_roundtrip(self, tmp_path): - p = _write(tmp_path, "f0,f1\n1,2\n3,4\n") - X = parse_features_csv(p) - assert X.shape == (2, 2) and X.dtype == np.float32 - - def test_schema_match_enforced(self, tmp_path): - p = _write(tmp_path, "f1,f0\n1,2\n") - with pytest.raises(InvalidInputError, match="do not match"): - parse_features_csv(p, expected_features=["f0", "f1"]) - - def test_schema_match_ok(self, tmp_path): - p = _write(tmp_path, "f0,f1\n1,2\n") - assert parse_features_csv(p, expected_features=["f0", "f1"]).shape == (1, 2) diff --git a/tests/unit/test_engine.py b/tests/unit/test_engine.py index e57545c..13cb44d 100644 --- a/tests/unit/test_engine.py +++ b/tests/unit/test_engine.py @@ -7,6 +7,7 @@ from tabctx.errors import AdmissionRejected, DatasetNotFoundError, InvalidInputError from tabctx.memory.calibration_data import A100_40GB_TABICL_CALIBRATION from tabctx.memory.estimator import PowerLawMemoryEstimator +from tabctx.schema import TableSchema TRAIN_X = [[float(i), float(i) * 2] for i in range(20)] TRAIN_Y = ["a" if i % 2 == 0 else "b" for i in range(20)] @@ -163,19 +164,23 @@ def test_predict_chunks_large_test_sets_transparently(): # every row gets exactly one prediction, in order. -def test_feature_names_recorded_at_fit_and_queryable(): +SCHEMA = TableSchema((("lo", "hi"), None), ("band", "x")) + + +def test_schema_recorded_at_fit_and_queryable(): engine, _ = make_engine() - dataset_id = engine.fit(TRAIN_X, TRAIN_Y, feature_names=["f0", "f1"]) - assert engine.feature_names(dataset_id) == ["f0", "f1"] - # Inline fits (no names) and unknown ids both read as None. - assert engine.feature_names(engine.fit(TRAIN_X, TRAIN_Y)) is None - assert engine.feature_names("never-fit") is None - # A re-fit of the same id without names must not keep stale names. + dataset_id = engine.fit(TRAIN_X, TRAIN_Y, schema=SCHEMA) + assert engine.schema(dataset_id) == SCHEMA + # Fits without a schema and unknown ids both read as None. + assert engine.schema(engine.fit(TRAIN_X, TRAIN_Y)) is None + with pytest.raises(DatasetNotFoundError): + engine.schema("never-fit") + # A re-fit of the same id without a schema must not keep a stale one. engine.fit(TRAIN_X, TRAIN_Y, dataset_id=dataset_id) - assert engine.feature_names(dataset_id) is None + assert engine.schema(dataset_id) is None -def test_feature_names_survive_eviction_to_the_spill_tier(tmp_path): +def test_schema_survives_eviction_to_the_spill_tier(tmp_path): from tabctx.cache.spill import DiskSpillStore backend = FakeBackend(bytes_hint=100) @@ -184,9 +189,9 @@ def test_feature_names_survive_eviction_to_the_spill_tier(tmp_path): ) estimator = PowerLawMemoryEstimator(A100_40GB_TABICL_CALIBRATION) engine = TabctxEngine(backend=backend, cache=cache, estimator=estimator) - engine.fit(TRAIN_X, TRAIN_Y, dataset_id="csv", feature_names=["f0", "f1"]) + engine.fit(TRAIN_X, TRAIN_Y, dataset_id="csv", schema=SCHEMA) engine.fit(TRAIN_X, TRAIN_Y, dataset_id="other") # spills "csv" - assert engine.feature_names("csv") == ["f0", "f1"] + assert engine.schema("csv") == SCHEMA def test_stats_reflect_cached_contexts(): diff --git a/tests/unit/test_multidevice.py b/tests/unit/test_multidevice.py index a3458eb..13adbb3 100644 --- a/tests/unit/test_multidevice.py +++ b/tests/unit/test_multidevice.py @@ -15,6 +15,7 @@ from tabctx.engine import TabctxEngine from tabctx.errors import AdmissionRejected, DatasetNotFoundError, InvalidInputError from tabctx.multidevice import MultiDeviceEngine +from tabctx.schema import TableSchema BYTES_PER_CELL = 10 CAPACITY = 10_000 # per device: 1,000 cells resident or in flight @@ -255,11 +256,13 @@ def test_generated_dataset_ids_are_routed(self): dataset_id = engine.fit(*table(10, label="g")) assert engine.predict(dataset_id, [[0.0]]).predictions == ["g"] - def test_feature_names_follow_the_context(self): + def test_schema_follows_the_context(self): engine, _, _ = build(2) - engine.fit(*table(5, features=2), dataset_id="d", feature_names=["p", "q"]) - assert engine.feature_names("d") == ["p", "q"] - assert engine.feature_names("missing") is None + schema = TableSchema((None, ("a", "b")), ("p", "q")) + engine.fit(*table(5, features=2), dataset_id="d", schema=schema) + assert engine.schema("d") == schema + with pytest.raises(DatasetNotFoundError): + engine.schema("missing") def test_fit_predict_leaves_nothing_cached(self): engine, shards, _ = build(2) @@ -433,7 +436,7 @@ def test_lookups_and_stats_never_wait_on_a_busy_device(self): engine.stats, engine.device_stats, engine.admission_headroom_bytes, - lambda: engine.feature_names("missing"), + lambda: engine.schema("missing"), ): t, box = start(call) t.join(3) diff --git a/tests/unit/test_schema.py b/tests/unit/test_schema.py new file mode 100644 index 0000000..b05cdea --- /dev/null +++ b/tests/unit/test_schema.py @@ -0,0 +1,104 @@ +"""Text columns and the training table's schema (tabctx.schema).""" + +import math + +import numpy as np +import pytest + +from tabctx.errors import InvalidInputError +from tabctx.schema import TableSchema + + +def obj(*values): + col = np.empty(len(values), dtype=object) + col[:] = values + return col + + +def nan_mask(X): + return np.isnan(X).tolist() + + +class TestLearn: + def test_numbers_stay_numbers_and_text_becomes_sorted_codes(self): + schema, X = TableSchema.learn( + [np.array([1.5, 2.5, np.nan]), obj("gold", "basic", "gold")] + ) + assert schema.categories == (None, ("basic", "gold")) + assert X.dtype == np.float32 + assert X[:, 1].tolist() == [1.0, 0.0, 1.0] + assert math.isnan(X[2, 0]) + + def test_missing_text_is_nan_not_a_category(self): + schema, X = TableSchema.learn([obj("a", None, "b", float("nan"))]) + assert schema.categories == (("a", "b"),) + assert nan_mask(X) == [[False], [True], [False], [True]] + + def test_numbers_in_a_text_column_are_text(self): + schema, X = TableSchema.learn([obj(3, "x", 3.0, 2.5)]) + # 3 and 3.0 are the same category, as a CSV writes them. + assert schema.categories == (("2.5", "3", "x"),) + assert X[:, 0].tolist() == [1.0, 2.0, 1.0, 0.0] + + def test_an_object_column_of_numbers_is_numeric(self): + schema, X = TableSchema.learn([obj(1, None, 2.5)]) + assert schema.categories == (None,) + assert nan_mask(X) == [[False], [True], [False]] + + +class TestEncode: + def setup_method(self): + self.schema, _ = TableSchema.learn( + [np.array([1.0, 2.0]), obj("lo", "hi")], names=["x", "band"] + ) + + def test_same_text_same_codes_and_unseen_is_missing(self): + X = self.schema.encode([np.array([5.0, 6.0, 7.0]), obj("hi", "mid", None)]) + assert X[:, 0].tolist() == [5.0, 6.0, 7.0] + assert X[0, 1] == 0.0 # "hi" sorts before "lo" + assert nan_mask(X) == [[False, False], [False, True], [False, True]] + + def test_numbers_where_training_had_text_are_matched_as_text(self): + schema, _ = TableSchema.learn([obj("1", "2", "x")]) + X = schema.encode([np.array([2.0, 1.0, 9.0])]) + assert X[:2, 0].tolist() == [1.0, 0.0] + assert math.isnan(X[2, 0]) + + def test_text_where_training_had_numbers_is_rejected(self): + with pytest.raises(InvalidInputError, match="'x' held numbers"): + self.schema.encode([obj("oops", 1.0), obj("lo", "hi")]) + + def test_wrong_column_count_is_rejected(self): + with pytest.raises(InvalidInputError, match="training table had 2"): + self.schema.encode([np.array([1.0])]) + + +class TestSelect: + def test_by_name_in_any_order_ignoring_extras(self): + schema = TableSchema((None, ("a",)), ("x", "band")) + cols = schema.select(["band", "label", "x"], ["B", "L", "X"]) + assert cols == ["X", "B"] + + def test_missing_names_are_reported(self): + schema = TableSchema((None, None), ("x", "y")) + with pytest.raises( + InvalidInputError, match=r"missing training column\(s\) \['y'\]" + ): + schema.select(["x"], ["X"]) + + def test_unnamed_schemas_match_by_position(self): + schema = TableSchema((None, None)) + assert schema.select(["a", "b"], ["A", "B"]) == ["A", "B"] + + +def test_text_names(): + assert TableSchema((None, ("a",)), ("x", "c")).text_names == ("c",) + assert TableSchema((None, ("a",))).text_names == () + + +def test_json_round_trip(): + for schema in ( + TableSchema((None, ("a", "b"))), + TableSchema((("x",), None), ("c", "n")), + ): + assert TableSchema.from_json(schema.to_json()) == schema diff --git a/tests/unit/test_serve_files.py b/tests/unit/test_serve_files.py new file mode 100644 index 0000000..fd404eb --- /dev/null +++ b/tests/unit/test_serve_files.py @@ -0,0 +1,255 @@ +"""Table files -> columns (serve/files.py): CSV, compressed CSV, Parquet.""" + +import datetime +import gzip +import io +import math + +import numpy as np +import pytest + +pa = pytest.importorskip("pyarrow") +import pyarrow.parquet as pq + +from tabctx.errors import InvalidInputError +from tabctx.serve.files import read_table, read_training_table + + +def write(tmp_path, data, name="t"): + p = tmp_path / name + p.write_bytes(data.encode() if isinstance(data, str) else data) + return p + + +def train(tmp_path, data, task=None, target=None): + return read_training_table(write(tmp_path, data), target, task) + + +def labels(tmp_path, data, task=None): + return train(tmp_path, data, task)[1] + + +def parquet(table): + buf = io.BytesIO() + pq.write_table(table, buf, compression="zstd") + return buf.getvalue() + + +def cells(col): + return [None if isinstance(v, float) and math.isnan(v) else v for v in col.tolist()] + + +class TestFormats: + CSV = "x,band,label\n1.5,gold,yes\n2,basic,no\n" + + def check(self, table, y): + assert table.names == ["x", "band"] + assert table.columns[0].tolist() == [1.5, 2.0] + assert table.columns[0].dtype == np.float32 + assert list(table.columns[1]) == ["gold", "basic"] + assert y == ["yes", "no"] + + def test_csv(self, tmp_path): + self.check(*train(tmp_path, self.CSV)) + + def test_gzip_csv_detected_from_content(self, tmp_path): + self.check(*train(tmp_path, gzip.compress(self.CSV.encode()))) + + def test_zstd_csv_detected_from_content(self, tmp_path): + # A raw zstd frame, which is what `zstd file.csv` makes. + self.check( + *train(tmp_path, pa.compress(self.CSV.encode(), "zstd", asbytes=True)) + ) + + def test_parquet_detected_from_content(self, tmp_path): + table = pa.table( + {"x": [1.5, 2.0], "band": ["gold", "basic"], "label": ["yes", "no"]} + ) + self.check(*train(tmp_path, parquet(table))) + + def test_parquet_types(self, tmp_path): + table = pa.table( + { + "n": pa.array([1, None], pa.int64()), + "big": pa.array([2**63, 1], pa.uint64()), + "b": [True, False], + "d": [datetime.date(1970, 1, 2), None], + "c": pa.array(["a", None]).dictionary_encode(), + } + ) + cols = read_table(write(tmp_path, parquet(table))).columns + assert cells(cols[0]) == [1.0, None] + assert cols[1][0] == np.float32(2**63) + assert cols[2].tolist() == [1.0, 0.0] + assert cells(cols[3]) == [86400.0, None] + assert list(cols[4]) == ["a", None] + + def test_times_are_seconds_including_fractions(self, tmp_path): + csv = "t,clock,y\n2020-01-01 10:00:00.5,10:30:00,1\n1970-01-01 00:00:01,00:00:02,2\n" + cols = train(tmp_path, csv)[0].columns + assert cols[0][1] == 1.0 + assert cols[0][0] == 1577872800.5 # float64 here; offset by the schema + assert cols[1].tolist() == [37800.0, 2.0] + table = pa.table( + { + "ts": pa.array([1_500_000], pa.timestamp("us")), + "dur": pa.array([2_500_000], pa.duration("us")), + } + ) + cols = read_table(write(tmp_path, parquet(table))).columns + assert cols[0].tolist() == [1.5] and cols[1].tolist() == [2.5] + + def test_nested_and_undecodable_columns_are_rejected(self, tmp_path): + with pytest.raises(InvalidInputError, match="neither numbers nor text"): + read_table(write(tmp_path, parquet(pa.table({"l": [[1, 2], [3]]})))) + with pytest.raises(InvalidInputError, match="not UTF-8 text"): + read_table(write(tmp_path, b"a,b\n1,caf\xe9\n")) + + +class TestFeatures: + def test_missing_spellings_in_any_case_with_spaces(self, tmp_path): + features, _ = train( + tmp_path, + "a,b,c,d,label\n1,,NA,x,y\nnull, n/a ,2,NONE,z\n3,NaN,None,Null,y\n", + ) + assert [cells(c) for c in features.columns[:3]] == [ + [1.0, None, 3.0], + [None, None, None], + [None, 2.0, None], + ] + assert list(features.columns[3]) == ["x", None, None] + + def test_text_features_are_stripped(self, tmp_path): + features, _ = train(tmp_path, "c,label\ngold,x\nNA,y\n basic ,x\n") + assert list(features.columns[0]) == ["gold", None, "basic"] + + def test_header_names_are_stripped(self, tmp_path): + features, y = train(tmp_path, " a , y \n1, 0\n2, 1\n", target="y") + assert features.names == ["a"] and y == ["0", "1"] + + def test_text_columns_stay_text(self, tmp_path): + p = write(tmp_path, "zip,x\n007,1\n010,2\n") + assert read_table(p).columns[0].tolist() == [7.0, 10.0] + assert list(read_table(p, text_columns=["zip"]).columns[0]) == ["007", "010"] + + def test_a_stripped_text_column_name_types_a_spaced_header(self, tmp_path): + p = write(tmp_path, " zip ,x\n007,1\n") + assert list(read_table(p, text_columns=["zip"]).columns[0]) == ["007"] + + +class TestLabels: + def test_default_target_is_the_last_column(self, tmp_path): + features, y = train(tmp_path, "f0,f1,label\n1,2,cat\n3,4,dog\n") + assert features.names == ["f0", "f1"] and y == ["cat", "dog"] + + def test_named_target_anywhere(self, tmp_path): + features, y = train(tmp_path, "label,f0\nyes,1\nno,3\n", target="label") + assert features.names == ["f0"] and y == ["yes", "no"] + + def test_class_labels_keep_their_text(self, tmp_path): + assert labels(tmp_path, "f,y\n1,1.0\n2,2.0\n") == ["1.0", "2.0"] + assert labels(tmp_path, "f,y\n1,01\n2,02\n") == ["01", "02"] + assert labels(tmp_path, "f,y\n1,true\n2,false\n") == ["true", "false"] + + def test_numbers_decide_the_task(self, tmp_path): + assert labels(tmp_path, "f,y\n1,0.5\n2,1.5\n") == [0.5, 1.5] + assert labels(tmp_path, "f,y\n1,0\n2,1\n") == ["0", "1"] + assert labels(tmp_path, "f,y\n1,0\n2,1\n", task="regression") == [0.0, 1.0] + assert labels(tmp_path, "f,y\n1,0.5\n2,1.5\n", task="classification") == [ + "0.5", + "1.5", + ] + + def test_na_among_text_classes_is_a_class(self, tmp_path): + assert labels(tmp_path, "f,label\n1,cat\n2,NA\n3,dog\n") == ["cat", "NA", "dog"] + + @pytest.mark.parametrize("missing", ["", "NA", "None", "none", "null", "NaN"]) + def test_a_missing_label_among_numbers_is_rejected(self, tmp_path, missing): + with pytest.raises(InvalidInputError, match="row 2 has no label"): + labels(tmp_path, f"f,y\n1,2.5\n2,{missing}\n3,1.5\n") + + def test_a_blank_text_label_is_rejected(self, tmp_path): + with pytest.raises(InvalidInputError, match="row 2 has no label"): + labels(tmp_path, "f,label\n1,x\n2,\n") + + def test_regression_needs_numbers(self, tmp_path): + with pytest.raises(InvalidInputError, match="regression needs numeric labels"): + labels(tmp_path, "a,y\n1,a\n2,b\n", task="regression") + + def test_parquet_labels(self, tmp_path): + def parquet_labels(values, task=None): + table = pa.table({"x": [1.0] * len(values), "y": values}) + return read_training_table(write(tmp_path, parquet(table)), None, task)[1] + + assert parquet_labels([0, 1]) == ["0", "1"] + assert parquet_labels([0.5, 1.25]) == [0.5, 1.25] + assert parquet_labels([True, False]) == ["True", "False"] + assert parquet_labels(pa.array(["a", "b"]).dictionary_encode()) == ["a", "b"] + with pytest.raises(InvalidInputError, match="row 2 has no label"): + parquet_labels([1.0, None]) + with pytest.raises(InvalidInputError, match="regression needs numeric"): + parquet_labels([True, False], task="regression") + + def test_unknown_target_and_no_features_are_rejected(self, tmp_path): + with pytest.raises(InvalidInputError, match="'nope' is not in"): + train(tmp_path, "f0,f1\n1,2\n", target="nope") + with pytest.raises(InvalidInputError, match="at least one feature"): + train(tmp_path, "label\ncat\n") + + +@pytest.mark.parametrize( + ("text", "match"), + [ + ("", "could not read"), + ("f0,label\n", "no rows"), + ("a,b,label\n1,2,x\n3,4\n", "could not read"), + ("a,a\n1,2\n", "repeats column"), + ], +) +def test_malformed_files_are_rejected_with_a_reason(tmp_path, text, match): + with pytest.raises(InvalidInputError, match=match): + read_table(write(tmp_path, text)) + + +def test_parquet_strings_stay_text_even_when_numeric(tmp_path): + table = pa.table({"zip": ["007", "010"], "y": ["a", "b"]}) + features, _ = read_training_table(write(tmp_path, parquet(table)), "y", None) + assert list(features.columns[0]) == ["007", "010"] + + +def test_a_long_csv_with_text_reads_its_header(tmp_path): + # The header is read ahead to type the label column; a multi-byte + # character anywhere later must not break that. + rows = "".join(f"café{i % 7},0{i % 2 + 1}\n" for i in range(120_000)) + p = write(tmp_path, "town,y\n" + rows) + assert p.stat().st_size > 1 << 20 + _, y = read_training_table(p, None, None) + assert set(y) == {"01", "02"} + + +def test_crlf_headers_type_the_label_column(tmp_path): + assert labels(tmp_path, "f,y\r\n1,01\r\n2,02\r\n") == ["01", "02"] + + +def test_decimal_labels_are_numbers(tmp_path): + import decimal + + values = pa.array( + [decimal.Decimal("1.5"), decimal.Decimal("2.5")], pa.decimal128(2, 1) + ) + table = pa.table({"x": [1.0, 2.0], "y": values}) + path = write(tmp_path, parquet(table)) + assert read_training_table(path, None, "regression")[1] == [1.5, 2.5] + assert read_training_table(path, None, None)[1] == [1.5, 2.5] + + +def test_times_keep_subsecond_resolution_through_the_schema(tmp_path): + from tabctx.schema import TableSchema + + csv = "t,y\n2020-01-01 10:00:00.5,1\n2020-01-01 10:00:01,2\n2020-01-01 10:00:02,1\n" + features, _ = train(tmp_path, csv) + schema, X = TableSchema.learn(features.columns, features.names) + assert X[:, 0].tolist() == [0.0, 0.5, 1.5] + test = read_table(write(tmp_path, "t\n2020-01-01 10:00:03\n", name="test")) + assert schema.encode(test.columns).tolist() == [[2.5]] + assert TableSchema.from_json(schema.to_json()) == schema diff --git a/tests/unit/test_serve_multipart.py b/tests/unit/test_serve_multipart.py new file mode 100644 index 0000000..298f960 --- /dev/null +++ b/tests/unit/test_serve_multipart.py @@ -0,0 +1,187 @@ +"""Multipart bodies read as they arrive (serve/multipart.py).""" + +import asyncio +import pickle + +import numpy as np +import pytest + +pytest.importorskip("python_multipart") + +from tabctx.errors import InvalidInputError, UploadTooLargeError +from tabctx.schema import TableSchema +from tabctx.serve.multipart import MAX_FIELD_BYTES, read_form +from tabctx.serve.schemas import SchemaCache +from tabctx.serve.tables import learn_rows +from tabctx.serve.uploads import UploadStore + +BOUNDARY = "b0undary" +CONTENT_TYPE = f"multipart/form-data; boundary={BOUNDARY}" + + +def body(fields=(), files=()): + out = b"" + for name, value in fields: + out += ( + ( + f'--{BOUNDARY}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n' + ).encode() + + value.encode() + + b"\r\n" + ) + for name, data in files: + out += ( + ( + f'--{BOUNDARY}\r\nContent-Disposition: form-data; name="{name}"; ' + 'filename="t.csv"\r\n\r\n' + ).encode() + + data + + b"\r\n" + ) + return out + f"--{BOUNDARY}--\r\n".encode() + + +async def chunks(data, size=7): + for i in range(0, len(data), size): + yield data[i : i + size] + + +def read(data, store, content_type=CONTENT_TYPE): + return asyncio.run(read_form(chunks(data), content_type, store.begin)) + + +@pytest.fixture +def store(tmp_path): + return UploadStore(directory=tmp_path, max_upload_bytes=1000) + + +def test_fields_and_the_file_in_small_chunks(store, tmp_path): + table = b"a,b\n1,2\n" * 20 + fields, writer = read( + body([("dataset_id", "d1"), ("target", "b")], [("file", table)]), store + ) + assert fields == {"dataset_id": "d1", "target": "b"} + path = writer.finish() + assert path.read_bytes() == table + + +def test_no_file_part(store): + fields, writer = read(body([("dataset_id", "d")]), store) + assert fields == {"dataset_id": "d"} and writer is None + + +def test_the_cap_stops_the_upload_as_bytes_arrive(store, tmp_path): + with pytest.raises(UploadTooLargeError): + read(body(files=[("file", b"x" * 5000)]), store) + assert list(tmp_path.iterdir()) == [] # nothing left on disk + + +@pytest.mark.parametrize( + ("data", "match"), + [ + (body(files=[("file", b"a"), ("file", b"b")]), "one table file"), + (body(files=[("table", b"a")]), "as the 'file' part"), + (body([("f", "x" * (MAX_FIELD_BYTES + 1))]), "over"), + (body([(f"f{i}", "x") for i in range(17)]), "more than 16"), + (body(files=[("file", b"abc")])[:-20], "ends inside a part"), + ], +) +def test_malformed_bodies_are_rejected_and_leave_nothing(store, tmp_path, data, match): + with pytest.raises(InvalidInputError, match=match): + read(data, store) + assert list(tmp_path.iterdir()) == [] + + +def test_a_missing_boundary_is_rejected(store): + with pytest.raises(InvalidInputError, match="boundary"): + read(body(), store, content_type="multipart/form-data") + + +def test_numeric_looking_text_cells_make_a_text_column(): + # numpy alone would read "007" as 7, depending on the other columns. + _, schema = learn_rows([[1.0, "007"], [2.0, "010"]]) + assert schema.categories == (None, ("007", "010")) + + +def test_schema_code_maps_are_built_once_and_not_pickled(): + schema = TableSchema((("a", "b"),)) + schema.encode([np.array(["b"], dtype=object)]) + assert "_codes" in schema.__dict__ + copy = pickle.loads(pickle.dumps(schema)) + assert "_codes" not in copy.__dict__ and copy == schema + + +def test_schema_cache_is_a_bounded_lru(): + cache = SchemaCache(max_entries=2) + a, b, c = (TableSchema(((n,),)) for n in "abc") + for key, schema in (("a", a), ("b", b)): + with cache.fitting(key) as record: + record(schema) + assert cache.get("a") is a # now most recent + with cache.fitting("c") as record: + record(c) + assert cache.get("b") is None and cache.get("a") is a and cache.get("c") is c + + +def test_schema_cache_bounds_bytes_and_skips_huge_schemas(): + big = TableSchema((tuple(f"category-{i}" for i in range(2000)),)) + cache = SchemaCache(max_bytes=big.approx_bytes * 2 + 10) + for key in ("x", "y", "z"): + with cache.fitting(key) as record: + record(big) + assert cache.get("x") is None and cache.get("z") is big + tiny = SchemaCache(max_bytes=100) + with tiny.fitting("x") as record: + record(big) + assert tiny.get("x") is None # falls back to the engine's copy + + +def test_a_fit_drops_the_cached_schema_until_it_succeeds(): + import threading + + cache = SchemaCache() + old, new = TableSchema((("old",),)), TableSchema((("new",),)) + with cache.fitting("d") as record: + record(old) + with pytest.raises(RuntimeError), cache.fitting("d"): + assert cache.get("d") is None # a failed fit leaves the engine to answer + raise RuntimeError("fit failed") + assert cache.get("d") is None + + # Fits of one id record one at a time: the second waits for the first. + order = [] + entered = threading.Event() + + def slow_fit(): + with cache.fitting("d") as record: + entered.set() + threading.Event().wait(0.05) + record(old) + order.append("first") + + t = threading.Thread(target=slow_fit) + t.start() + entered.wait() + with cache.fitting("d") as record: + record(new) + order.append("second") + t.join() + assert order == ["first", "second"] and cache.get("d") is new + + +def test_long_boundaries_and_plain_file_fields_are_422(store): + with pytest.raises(InvalidInputError, match="malformed"): + read(body(), store, content_type="multipart/form-data; boundary=" + "b" * 300) + with pytest.raises(InvalidInputError, match="-F file=@"): + read(body([("file", "a,b\n1,2\n")]), store) + + +def test_repeated_fields_and_other_charsets_are_refused(store): + with pytest.raises(InvalidInputError, match="repeated"): + read(body([("model", "a"), ("model", "b")]), store) + with pytest.raises(InvalidInputError, match="UTF-8"): + read(body(), store, content_type=CONTENT_TYPE + "; charset=utf-16-le") + fields, _ = read( + body([("m", "é")]), store, content_type=CONTENT_TYPE + "; charset=UTF-8" + ) + assert fields == {"m": "é"} diff --git a/tests/unit/test_serve_tables.py b/tests/unit/test_serve_tables.py index 8305751..5a40e34 100644 --- a/tests/unit/test_serve_tables.py +++ b/tests/unit/test_serve_tables.py @@ -3,8 +3,11 @@ import math import numpy as np +import pytest -from tabctx.serve.tables import rows_to_array +from tabctx.errors import InvalidInputError +from tabctx.schema import TableSchema +from tabctx.serve.tables import encode_rows, learn_rows, rows_to_array def test_null_cells_become_nan(): @@ -14,10 +17,40 @@ def test_null_cells_become_nan(): assert arr[0, 0] == 1.0 and arr[1, 1] == 2.0 +def test_numeric_tables_learn_an_all_numeric_schema(): + X, schema = learn_rows([[1.0, 2.0], [3.0, None]]) + assert X.dtype == np.float32 and X.shape == (2, 2) + assert schema == TableSchema((None, None)) + + +def test_text_cells_make_text_columns_encoded_the_same_at_predict(): + X, schema = learn_rows([[1.0, "gold"], [2.0, "basic"], [None, None]]) + assert schema.categories == (None, ("basic", "gold")) + assert X[:2].tolist() == [[1.0, 1.0], [2.0, 0.0]] + Xt = encode_rows([[5.0, "basic"], [6.0, "platinum"]], schema) + assert Xt[0].tolist() == [5.0, 0.0] + assert math.isnan(Xt[1, 1]) # unseen category -> missing + + +def test_numbers_for_a_text_column_are_matched_as_text(): + _, schema = learn_rows([["1"], ["2"], ["x"]]) + assert encode_rows([[2], [1]], schema).tolist() == [[1.0], [0.0]] + + +def test_text_where_training_had_numbers_is_rejected(): + _, schema = learn_rows([[1.0], [2.0]]) + with pytest.raises(InvalidInputError, match="held numbers"): + encode_rows([["x"]], schema) + # Contexts from before schemas existed: numbers only, same error. + with pytest.raises(InvalidInputError, match="held numbers"): + encode_rows([["x"]], None) + + def test_non_rectangles_pass_through_for_the_engine_to_report(): - ragged = [[1.0, 2.0], [3.0]] - assert rows_to_array(ragged) is ragged - assert rows_to_array([]) == [] + ragged = [[1.0, "a"], [3.0]] + assert learn_rows(ragged) == (ragged, None) + assert encode_rows(ragged, None) is ragged + assert learn_rows([]) == ([], None) def test_engine_fits_and_predicts_tables_with_missing_values(): @@ -32,7 +65,17 @@ def test_engine_fits_and_predicts_tables_with_missing_values(): cache=ContextCacheManager(capacity_bytes=200_000_000), estimator=PowerLawMemoryEstimator(A100_40GB_TABICL_CALIBRATION), ) - X = rows_to_array([[float(i), None if i % 3 else 1.0] for i in range(10)]) - ds = engine.fit(X, ["a", "b"] * 5) - out = engine.predict(ds, rows_to_array([[None, 1.0], [2.0, None]])) + X, schema = learn_rows([[float(i), None if i % 3 else "a"] for i in range(10)]) + ds = engine.fit(X, ["a", "b"] * 5, schema=schema) + out = engine.predict(ds, encode_rows([[None, "a"], [2.0, None]], schema)) assert len(out.predictions) == 2 + + +def test_json_rows_get_the_offsets_of_a_file_fit_context(): + # A time column fit from a file is stored from its training minimum; + # JSON rows (float64 until encoded) must be shifted the same way. + schema, _ = TableSchema.learn( + [np.array([1577872800.5, 1577872801.0]), np.array([1.0, 2.0], np.float32)] + ) + assert schema.offsets == (1577872800.5, None) + assert encode_rows([[1577872801.5, 1.0]], schema).tolist() == [[1.0, 1.0]] diff --git a/tests/unit/test_spill.py b/tests/unit/test_spill.py index b057367..5889d5e 100644 --- a/tests/unit/test_spill.py +++ b/tests/unit/test_spill.py @@ -13,6 +13,7 @@ AdaptiveMemoryEstimator, PowerLawMemoryEstimator, ) +from tabctx.schema import TableSchema def _ctx(dataset_id: str, est_bytes: int = 100, payload=None) -> CachedContext: @@ -141,24 +142,38 @@ def test_restore_that_no_longer_fits_capacity_is_a_miss(self, tmp_path): cache = ContextCacheManager(capacity_bytes=100, spill_store=spill) assert cache.get("big") is None - def test_feature_names_survive_the_spill_tier(self, tmp_path): + def test_schema_survives_the_spill_tier(self, tmp_path): spill = DiskSpillStore(tmp_path) ctx = _ctx("csv") - ctx.feature_names = ["age", "bmi"] + ctx.schema = TableSchema((None, ("f", "m")), ("age", "sex")) assert spill.spill(ctx) - assert spill.load("csv").feature_names == ["age", "bmi"] + assert spill.load("csv").schema == ctx.schema - def test_meta_without_feature_names_loads_as_none(self, tmp_path): - # Files written by a version before feature_names existed. + def _rewrite_meta(self, spill, dataset_id, edit): import json - spill = DiskSpillStore(tmp_path) - assert spill.spill(_ctx("old")) - meta_path, _ = spill._paths("old") + meta_path, _ = spill._paths(dataset_id) meta = json.loads(meta_path.read_text()) - del meta["feature_names"] + edit(meta) meta_path.write_text(json.dumps(meta)) - assert spill.load("old").feature_names is None + + def test_meta_from_before_schemas_loads_its_names(self, tmp_path): + # 0.14 and earlier recorded only a CSV's column names. + spill = DiskSpillStore(tmp_path) + assert spill.spill(_ctx("old")) + + def old_format(meta): + del meta["schema"] + meta["feature_names"] = ["age", "bmi"] + + self._rewrite_meta(spill, "old", old_format) + assert spill.load("old").schema == TableSchema((None, None), ("age", "bmi")) + + def test_meta_without_any_schema_loads_as_none(self, tmp_path): + spill = DiskSpillStore(tmp_path) + assert spill.spill(_ctx("older")) + self._rewrite_meta(spill, "older", lambda meta: meta.pop("schema")) + assert spill.load("older").schema is None def test_no_spill_store_behaves_as_before(self): cache = ContextCacheManager(capacity_bytes=150) diff --git a/tests/unit/test_workers.py b/tests/unit/test_workers.py index 5e90cf1..f54f24b 100644 --- a/tests/unit/test_workers.py +++ b/tests/unit/test_workers.py @@ -50,7 +50,7 @@ def test_fit_predict_round_trip_and_residency_mirror(built): holders = [w.device for w in built.worker_shards if w.holds("rt")] assert len(holders) == 1 # mirrored before fit() returned assert engine.predict("rt", [[0.0, 0.0]]).predictions == ["x"] - assert engine.feature_names("rt") is None + assert engine.schema("rt") is None engine.evict("rt") assert not any(w.has("rt") for w in built.worker_shards) diff --git a/uv.lock b/uv.lock index df238ff..b3c72b0 100644 --- a/uv.lock +++ b/uv.lock @@ -2504,6 +2504,56 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/6f/ed/1409cdb557e558a6c98003ab12fdd4284699e158c167c187cb0f124eea4c/py_spy-0.4.2-py2.py3-none-win_amd64.whl", hash = "sha256:8b06a353c177677e4e1701b288d8c58e2f8d4208ee81a8048d9f72ba800918f8", size = 1894002, upload-time = "2026-04-24T22:08:53.811Z" }, ] +[[package]] +name = "pyarrow" +version = "26.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/ec/34/17c34cb38e5d940e38f0f0d9fdfa0e8a506676409ea9b85aff7e3079f831/pyarrow-26.0.0.tar.gz", hash = "sha256:0cccd36e00ea3afeb52ded61f2721ce71f604853d70c45365c58324eb773d6ae", size = 1239433, upload-time = "2026-10-09T08:26:25.315Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/60/6793778f2617cce469383dac0ba08c4f2401cf342df0c7b9ca53939d9b46/pyarrow-26.0.0-cp312-cp312-macosx_12_0_arm64.whl", hash = "sha256:90ddaf7c625307ad52f31a9b25c34fe5e4897c7529ee3481135822b2b6842ff1", size = 36333953, upload-time = "2026-10-09T08:14:00.387Z" }, + { url = "https://files.pythonhosted.org/packages/db/81/f944cc63ce8a753e5fbff25de6d1d475ebd7fffdf9cf98c65130294fc896/pyarrow-26.0.0-cp312-cp312-macosx_12_0_x86_64.whl", hash = "sha256:ee341973f78a0b46e073d065e88e75026a9c584051e97f98a0d05d96c6bac7dd", size = 38688456, upload-time = "2026-10-09T08:14:04.344Z" }, + { url = "https://files.pythonhosted.org/packages/f5/2d/7e5c722fa5d5d9f3b75e62fe11694b34217664d4f05ac88031197166b277/pyarrow-26.0.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:01c863a18bd9c8412453dd0d92de6d0ee7b2b3d6fb079d9734a4b2a3c8bd4453", size = 50867603, upload-time = "2026-10-09T08:14:09.115Z" }, + { url = "https://files.pythonhosted.org/packages/88/e4/9cd356d906e71bd79b0c3fc5c9a54e01a0020dcf14c152ccfbcb503c7298/pyarrow-26.0.0-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:6a628922ba20705fa964ca73e4ef959c2fb2f14b9bbec5589a6a1e68e6257c85", size = 53931932, upload-time = "2026-10-09T08:14:24.051Z" }, + { url = "https://files.pythonhosted.org/packages/bb/e4/5bae3133b7fe04c24907a20f3bc1fba388cbbde659199e7b76445982047a/pyarrow-26.0.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:954d971b363b16ee41f89389a4053315dc71265f2ce5c2468eb0a910b1166268", size = 54444720, upload-time = "2026-10-09T08:14:31.214Z" }, + { url = "https://files.pythonhosted.org/packages/ba/b4/ee422493bb6dafdbef776cfe2c2a73106a1063a79bf4e78d1e5f51176885/pyarrow-26.0.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:5d5768d03426abe6526d5274adefa00abf00a7f81118c46e98b5a46390f5549e", size = 57388949, upload-time = "2026-10-09T08:14:38.964Z" }, + { url = "https://files.pythonhosted.org/packages/54/3c/1783aab1dac28e175dcf26dfc7123725efc474caecaed91e8a34cb89cad0/pyarrow-26.0.0-cp312-cp312-win_amd64.whl", hash = "sha256:cc903e1069e9dd5e9dcf780324c0112e27e051e422ecfaff574fb33ed65d9160", size = 28567581, upload-time = "2026-10-09T08:14:44.279Z" }, + { url = "https://files.pythonhosted.org/packages/4d/35/ca95493712af97c46a312945c8e9d16b21c5fe2f148be5466168d0290505/pyarrow-26.0.0-cp313-cp313-macosx_12_0_arm64.whl", hash = "sha256:a6ca849f90cf73fe361f08a5762c783ead9671e4548c1f558cc637b54c9103f2", size = 36336700, upload-time = "2026-10-09T08:14:51.399Z" }, + { url = "https://files.pythonhosted.org/packages/69/ef/b1a675f79c9babfd4fcd99af62141d3c2d1a78a524e311b0c6b80110445a/pyarrow-26.0.0-cp313-cp313-macosx_12_0_x86_64.whl", hash = "sha256:c2ba350957076b1b3a22f549261dc3e9c67ca20816d8bd5f79d7b9c69be4c4c2", size = 38698502, upload-time = "2026-10-09T08:14:57.114Z" }, + { url = "https://files.pythonhosted.org/packages/3b/7c/cea852a832a327a8de797b3a68e5c25ce0f5aa1d20503807671bd90ec642/pyarrow-26.0.0-cp313-cp313-manylinux_2_28_aarch64.whl", hash = "sha256:e3b190ba1d3d22a5a8758597f797111b77d433473744352a184a5ee0a42d672e", size = 50865064, upload-time = "2026-10-09T08:20:01.614Z" }, + { url = "https://files.pythonhosted.org/packages/4f/d6/e95834b29360092376fe4da9956ba41bb7b021869efe6ee9d4172d05cb15/pyarrow-26.0.0-cp313-cp313-manylinux_2_28_x86_64.whl", hash = "sha256:240bd18a7487f8767616a948a69dd4e740a8bc36a1c9da49e4dc9a32c5c2faed", size = 53926722, upload-time = "2026-10-09T08:23:10.829Z" }, + { url = "https://files.pythonhosted.org/packages/e0/7f/98257444e2aea2e1fddceee3af3bd2077236d550428413f80393bd1f888d/pyarrow-26.0.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:2b5fcd69c0e1107b79e55839877db5a6ed04651b73fd6fec581d09e230bed5e4", size = 54443093, upload-time = "2026-10-09T08:23:16.971Z" }, + { url = "https://files.pythonhosted.org/packages/88/ca/dac99cfb25cfa62bf7194600cc99abc14a6bd2af50d7fdb7f15eeaf6e202/pyarrow-26.0.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:f7444ea6975c49a857c68f9bd8fa11acae96dede63d120ffb3bf0a603ea82516", size = 57381937, upload-time = "2026-10-09T08:23:24.95Z" }, + { url = "https://files.pythonhosted.org/packages/c0/ed/138d29fddaf803b90f4527e124bb6aaddc18aaf4a6c50fd0a5f577c94989/pyarrow-26.0.0-cp313-cp313-win_amd64.whl", hash = "sha256:3de30a7432b48b98b9decbd9e25a53bb9251d202c2e6c5a29a50869592ccb117", size = 28478571, upload-time = "2026-10-09T08:23:30.535Z" }, + { url = "https://files.pythonhosted.org/packages/8c/32/01858422a37f083911c2bb4d15cc32c5eeaa9d9b2bf5ddedee995a7146a6/pyarrow-26.0.0-cp314-cp314-macosx_12_0_arm64.whl", hash = "sha256:5780d487ff6c6ed7b42298609680d87fe0036e529a9dc2e1105364bce9697f50", size = 36378402, upload-time = "2026-10-09T08:23:36.537Z" }, + { url = "https://files.pythonhosted.org/packages/00/85/f6b5976c2878b752d0804d371684e0495a71de296b6dc6559e6fbaa4311a/pyarrow-26.0.0-cp314-cp314-macosx_12_0_x86_64.whl", hash = "sha256:a0e4e92eeb088f1d7c2c04d6c7de8434c75abb4b4ccf0bbcd045aa7164c68d93", size = 38733074, upload-time = "2026-10-09T08:23:42.873Z" }, + { url = "https://files.pythonhosted.org/packages/81/bc/c90fcbbcf893631e23dab1b0fb3fa29a508a8614326571b03c0894eda00b/pyarrow-26.0.0-cp314-cp314-manylinux_2_28_aarch64.whl", hash = "sha256:eaf9e7cc7ab59f6c760232bbde18f64d559bbc50544841303bfb32be53533297", size = 50929201, upload-time = "2026-10-09T08:23:50.507Z" }, + { url = "https://files.pythonhosted.org/packages/ec/c1/0c1ff38ab7df1b2cf54cf0ad9f19a516c4e416c6c9b4c966cc2c9d587f77/pyarrow-26.0.0-cp314-cp314-manylinux_2_28_x86_64.whl", hash = "sha256:ab6914db225d7f399652ae1f08588dfbc9efe617612715701e3d9d5cfa5ca19f", size = 53951865, upload-time = "2026-10-09T08:23:57.692Z" }, + { url = "https://files.pythonhosted.org/packages/9f/70/6a6b170496925472adad45a32528770fc8632db35fc60d4edd1e9ce1be0b/pyarrow-26.0.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:41dd3661ef40790a78870052ad7a58ad827b27c67a4511f06962eb9e9b74d19b", size = 54496388, upload-time = "2026-10-09T08:24:05.23Z" }, + { url = "https://files.pythonhosted.org/packages/a8/32/033ef9dba80976820190e292a10a5a23e9406572b76bbeb4d685d90e5c8d/pyarrow-26.0.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:6e949744dcfc2d379808f7013c5f9cafaf0f817656dff7d46c6931528dd1784b", size = 57411588, upload-time = "2026-10-09T08:24:12.043Z" }, + { url = "https://files.pythonhosted.org/packages/1e/ff/a74892c50aaf1f9f744a84493e08a2f99221e77c39d2d4a926de21a99edf/pyarrow-26.0.0-cp314-cp314-win_amd64.whl", hash = "sha256:4a5fa8dc70dd50808990ff36faf44088e357b353d86c7682dd92d4b78d4c97d5", size = 29237858, upload-time = "2026-10-09T08:24:58.106Z" }, + { url = "https://files.pythonhosted.org/packages/03/10/f0ee0976ef08a851a743c57608917ac9a47623f688b9ee0efe5429975ba1/pyarrow-26.0.0-cp314-cp314t-macosx_12_0_arm64.whl", hash = "sha256:e2a1856e9565fe2679863b372478c681806aebbf7d0a6e72f33e77f804e647d6", size = 36495870, upload-time = "2026-10-09T08:24:16.479Z" }, + { url = "https://files.pythonhosted.org/packages/27/ca/0bc431a509bf10b4472dbb94f4184752ecbbddeb7f467152dac0fdaed469/pyarrow-26.0.0-cp314-cp314t-macosx_12_0_x86_64.whl", hash = "sha256:4bcba83299cb2b8f8e443d36c6ba6269a5034431879015fb0719495df8a14de2", size = 38819754, upload-time = "2026-10-09T08:24:20.875Z" }, + { url = "https://files.pythonhosted.org/packages/61/59/2be41d26af7a07fb71581fb753cae396403ba1a2978355fd553929d44a9a/pyarrow-26.0.0-cp314-cp314t-manylinux_2_28_aarch64.whl", hash = "sha256:3a4d235876f14b4136b4d616ec42eb469ea0d6ead336cae631aa1dd29b21c962", size = 50933671, upload-time = "2026-10-09T08:24:27.199Z" }, + { url = "https://files.pythonhosted.org/packages/4b/cb/b6d5048cf3178be9678f5c9c60040199894b2f69c3439c87ced91fd24da9/pyarrow-26.0.0-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:210cc9b83888b87cdc8f793eebb264f22b20d0dedbedefc73b9687a7047b4747", size = 53906419, upload-time = "2026-10-09T08:24:33.536Z" }, + { url = "https://files.pythonhosted.org/packages/09/2b/23e30fbd776c81d18d134d2592eb60daca13e8a57ab087d0fa042f9d9f3d/pyarrow-26.0.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:ca77c43ca55bfc9a4eeb1f0cd5f093f08731b77c24cdba0829035f084959b0bb", size = 54527960, upload-time = "2026-10-09T08:24:41.292Z" }, + { url = "https://files.pythonhosted.org/packages/e2/23/fce251cd6b0546dfc181b00d5c8ef1c95a8c4cae83266bc3dfd5f719c62c/pyarrow-26.0.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:290a74c48e9491b436fd5edacfadf357943f82aa45c81110bd83a69aab33d1cf", size = 57388010, upload-time = "2026-10-09T08:24:48.186Z" }, + { url = "https://files.pythonhosted.org/packages/44/a5/0126fb0ef8d59bf257bdd68bb41623b72afc6e81790a0b4ac863a0f58861/pyarrow-26.0.0-cp314-cp314t-win_amd64.whl", hash = "sha256:515a10dae2a1d236bc9c9209d0317acb6746ea63cd4f98704904af7156d90ed1", size = 29406123, upload-time = "2026-10-09T08:24:53.387Z" }, + { url = "https://files.pythonhosted.org/packages/ed/66/8ada1b5165359d84b4b9b5384742304d1081da670f77d458fd9c9b8a2161/pyarrow-26.0.0-cp315-cp315-macosx_12_0_arm64.whl", hash = "sha256:e890816e5ee89c74a0f8b9379fe8b5ba83f46132b2a0bbb9b1c21359ec30dfda", size = 36373215, upload-time = "2026-10-09T08:25:03.067Z" }, + { url = "https://files.pythonhosted.org/packages/c4/83/74f10c3d803a6834b2acab21847724d4bdbc74d246eb17321432844707f3/pyarrow-26.0.0-cp315-cp315-macosx_12_0_x86_64.whl", hash = "sha256:9db18a9dc0af52135c9eac549d80a7a882696efbe5406cf882b044525d4ecc2e", size = 38730866, upload-time = "2026-10-09T08:25:07.924Z" }, + { url = "https://files.pythonhosted.org/packages/e2/5a/ea2fa2163b1bd8ff73efd39c4060be63fd6ddec03e7887a471acd1e042a4/pyarrow-26.0.0-cp315-cp315-manylinux_2_28_aarch64.whl", hash = "sha256:734312d3d99088d9ec28c5b17bad40389bd8373a1afc10acb60b83fd217af087", size = 50924443, upload-time = "2026-10-09T08:25:13.864Z" }, + { url = "https://files.pythonhosted.org/packages/78/80/8c47b6cf8cfd42826df65193eff026c1cc81fa6cb213a3c3f5d203e6f67a/pyarrow-26.0.0-cp315-cp315-manylinux_2_28_x86_64.whl", hash = "sha256:24f892fdf1ae1942d69d3f7742e2f49960ec95277cfb1a70b8a1d91f4a96d935", size = 53948540, upload-time = "2026-10-09T08:25:19.305Z" }, + { url = "https://files.pythonhosted.org/packages/69/1f/3a506a76d944ec5c5e4b7f01d8d0446b392a6fb384de627a12e503f616b4/pyarrow-26.0.0-cp315-cp315-musllinux_1_2_aarch64.whl", hash = "sha256:879331ddea2a26479fa18fade71e6facf684a6cf19f67daec3775c871569e8e5", size = 54494863, upload-time = "2026-10-09T08:25:24.517Z" }, + { url = "https://files.pythonhosted.org/packages/3d/50/08c4bb04d651788d2eaca78065743f4f6ded974d4ef96ae3c473993e9d0c/pyarrow-26.0.0-cp315-cp315-musllinux_1_2_x86_64.whl", hash = "sha256:5b827650e874f1f9f9392524ea3e9e3e8a245de5ba64acca1f81ab188090afb9", size = 57409877, upload-time = "2026-10-09T08:25:31.157Z" }, + { url = "https://files.pythonhosted.org/packages/d4/f3/c64781fbd7b6d3c07993b698c14944d0d195f07e800fa931c486ae6ab36a/pyarrow-26.0.0-cp315-cp315-win_amd64.whl", hash = "sha256:8e8e28c464552b5ca03e30d4504168c4425ce383884f8611b00e972f9fd933fc", size = 29236658, upload-time = "2026-10-09T08:26:22.607Z" }, + { url = "https://files.pythonhosted.org/packages/06/55/2ee3729daea999f19f061f03898d4895a242c4cd94f26e1324e5fdfbfe10/pyarrow-26.0.0-cp315-cp315t-macosx_12_0_arm64.whl", hash = "sha256:ce28748cbeb0f29c3ce9603782979c7117580fc76f16aa3ca448b38a22281adb", size = 36489011, upload-time = "2026-10-09T08:25:37.64Z" }, + { url = "https://files.pythonhosted.org/packages/6a/7d/3eb17f601f2bf13eda5f2ed28956379ca628b4dda97619cbb1cb1721622d/pyarrow-26.0.0-cp315-cp315t-macosx_12_0_x86_64.whl", hash = "sha256:106bb9290fc6fd9a84138a9440038ef184bac86463543c5ff099229cb30d996c", size = 38808480, upload-time = "2026-10-09T08:25:43.579Z" }, + { url = "https://files.pythonhosted.org/packages/0e/e3/f0047360b0f4bfc031b256dc0aec3837a61f245b2fb70f8363438e2db665/pyarrow-26.0.0-cp315-cp315t-manylinux_2_28_aarch64.whl", hash = "sha256:2e4a413046eba9896e632925066c74095182200ba32e19ff0166bf64d2f936ac", size = 50923273, upload-time = "2026-10-09T08:25:51.445Z" }, + { url = "https://files.pythonhosted.org/packages/38/d9/56d9fb91210407df31cbeb9b91138601c88c7c8fb5f6bf773b20d65509bf/pyarrow-26.0.0-cp315-cp315t-manylinux_2_28_x86_64.whl", hash = "sha256:d58798c4d8d629700058e9afc1e16b9801023f3ce4dc1c92d945e79b5ffe4e98", size = 53900905, upload-time = "2026-10-09T08:25:59.554Z" }, + { url = "https://files.pythonhosted.org/packages/cf/40/8e8a7e9e027c731520c7eb179dd00a153b76ebf0bc11d213c6c8f8502851/pyarrow-26.0.0-cp315-cp315t-musllinux_1_2_aarch64.whl", hash = "sha256:645917e976671debabf854abab6e2b75c571ca4f82adc33a2d338697f7c27d93", size = 54518345, upload-time = "2026-10-09T08:26:07.125Z" }, + { url = "https://files.pythonhosted.org/packages/be/89/1e768a3fdb88d34e708ad2dc00dbf8e4e30290784eb84198d59308963bea/pyarrow-26.0.0-cp315-cp315t-musllinux_1_2_x86_64.whl", hash = "sha256:7c3fda041e7078802589cf257750323ee3d0cd1e56e53a9b20ec845697fb3d28", size = 57379403, upload-time = "2026-10-09T08:26:13.624Z" }, + { url = "https://files.pythonhosted.org/packages/96/be/7b81a44d6a8e70581dcc1d6f01541f9000a973b1e5d75394aec91e7b179a/pyarrow-26.0.0-cp315-cp315t-win_amd64.whl", hash = "sha256:68cd662e9e2b00876a131950cf32336ace2d0865e1f9418763e3d3be8481dfa4", size = 29389953, upload-time = "2026-10-09T08:26:18.277Z" }, +] + [[package]] name = "pyasn1" version = "0.6.4" @@ -2731,6 +2781,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0d/17/c5c6b53ddc18f297992099b3d9ec16c855c0ccc83263a21fe4d1c625ec6c/python_dotenv-1.2.3-py3-none-any.whl", hash = "sha256:904552145e8bfed22162c09dab1c2b9b54fefa7b23ba780f4f26ca0316b0f0d9", size = 22780, upload-time = "2026-08-16T16:54:52.473Z" }, ] +[[package]] +name = "python-multipart" +version = "0.0.32" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5b/42/55c32bb9b12693c092ad250a0e82edb5b31ddeda6eb772de5f308b3804ad/python_multipart-0.0.32.tar.gz", hash = "sha256:be54b7f3fa167bb83e4fcd936b887b708f4e57fe75911c02aebf53efaf8d938e", size = 46881, upload-time = "2026-06-04T16:18:58.647Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e1/04/e8135ebd1ad02c56ec633277529b2602ff99ff634be76cdba5744cf554fd/python_multipart-0.0.32-py3-none-any.whl", hash = "sha256:ff6d3f776f16878c894e52e107296ffc890e913c611b1a4ec6c44e2821fe2e23", size = 30042, upload-time = "2026-06-04T16:18:57.319Z" }, +] + [[package]] name = "pyyaml" version = "6.0.3" @@ -3216,6 +3275,8 @@ dev = [ serve = [ { name = "fastapi" }, { name = "jinja2" }, + { name = "pyarrow" }, + { name = "python-multipart" }, { name = "ray", extra = ["serve"] }, { name = "uvicorn", extra = ["standard"] }, ] @@ -3232,8 +3293,10 @@ tabpfn = [ dev = [ { name = "httpx" }, { name = "pre-commit" }, + { name = "pyarrow" }, { name = "pytest" }, { name = "pytest-cov" }, + { name = "python-multipart" }, { name = "ruff" }, ] @@ -3243,8 +3306,10 @@ requires-dist = [ { name = "httpx", marker = "extra == 'dev'" }, { name = "jinja2", marker = "extra == 'serve'" }, { name = "numpy", specifier = ">=1.24" }, + { name = "pyarrow", marker = "extra == 'serve'", specifier = ">=14" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0" }, { name = "pytest-cov", marker = "extra == 'dev'" }, + { name = "python-multipart", marker = "extra == 'serve'", specifier = ">=0.0.9" }, { name = "ray", extras = ["serve"], marker = "extra == 'serve'", specifier = ">=2.58.0" }, { name = "tabicl", marker = "extra == 'tabicl'", specifier = ">=2.2.0" }, { name = "tabpfn", marker = "extra == 'tabpfn'" }, @@ -3258,8 +3323,10 @@ provides-extras = ["tabicl", "tabpfn", "serve", "dev"] dev = [ { name = "httpx" }, { name = "pre-commit" }, + { name = "pyarrow", specifier = ">=14" }, { name = "pytest", specifier = ">=7.0" }, { name = "pytest-cov" }, + { name = "python-multipart", specifier = ">=0.0.9" }, { name = "ruff" }, ]