diff --git a/CHANGELOG.md b/CHANGELOG.md index a6f644c..166a2fd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,12 +33,14 @@ All notable changes to this project will be documented in this file. ### Added * Add an optional Ibis backend with native Parquet results through `Client.query_parquet`. -* Surface errors received during NDJSON query streaming as `QueryError`. * Add `Client.upload` with typed upload modes and support for file streams. +* Add query dialect and response-format options, including all v0.13 compute sizes. +* Add `create_append` uploads and cursor-aware upserts. +* Surface backend NDJSON query stream failures as typed `QueryError` exceptions. ### Fixed -* Require `primary_key` for upserts and stop sending the unsupported `mode` query parameter. +* Continue requiring `primary_key` for upserts and omit the unsupported `mode` query parameter. ## [0.3.1](https://github.com/altertable-ai/altertable-lakehouse-python/compare/altertable-lakehouse-v0.3.0...altertable-lakehouse-v0.3.1) (2026-06-30) diff --git a/README.md b/README.md index 7df9b34..7da2663 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,9 @@ result = client.query_all(req) print(result.metadata.values) print(result.columns) print(result.rows) + +# Optional source dialect and response representation +req = QueryRequest(statement="SELECT 1", dialect="postgres", format="jsonl") ``` ### Append @@ -102,10 +105,16 @@ with open("data.csv", "rb") as f: schema="my_schema", table="my_table", primary_key="id", + cursor_field="updated_at", content=f.read(), + content_type="text/csv", ) ``` +`cursor_field` is optional. On a primary-key collision, the row with the +higher cursor wins; ties leave the stored row unchanged. Composite cursor +fields compare left to right, and `NULL` sorts lower than any non-null value. + ### Upload ```python @@ -116,12 +125,15 @@ with open("data.csv", "rb") as f: catalog="my_cat", schema="my_schema", table="my_table", - mode=UploadMode.CREATE, + mode=UploadMode.CREATE_APPEND, content=f, content_type="text/csv", ) ``` +`UploadMode.CREATE_APPEND` creates the table when it is absent and appends when +it already exists. + ### Validate Query ```python diff --git a/specs b/specs index c55685e..2491b40 160000 --- a/specs +++ b/specs @@ -1 +1 @@ -Subproject commit c55685efe831466bf8ceb6953ecb13cbb19780ab +Subproject commit 2491b404696590588944d576398e44844799432b diff --git a/src/altertable_lakehouse/client.py b/src/altertable_lakehouse/client.py index 61e88ad..07d875c 100644 --- a/src/altertable_lakehouse/client.py +++ b/src/altertable_lakehouse/client.py @@ -19,6 +19,7 @@ QueryMetadata, QueryResult, UploadMode, + ComputeSize, ) from .errors import ( AuthError, @@ -124,8 +125,10 @@ def upsert( catalog: str, schema: str, table: str, - content: bytes, + content: Union[bytes, BinaryIO, Iterable[bytes]], primary_key: str, + cursor_field: Optional[str] = None, + content_type: Optional[str] = None, ) -> None: params = { "catalog": catalog, @@ -133,11 +136,15 @@ def upsert( "table": table, "primary_key": primary_key, } + if cursor_field is not None: + params["cursor_field"] = cursor_field + headers = {"Content-Type": content_type} if content_type else None try: res = self._client.post( "/upsert", params=params, content=content, + headers=headers, ) self._check_response(res) except httpx.RequestError as e: @@ -206,6 +213,9 @@ def autocomplete(self, request: AutocompleteRequest) -> AutocompleteResponse: self._handle_error(e) def query(self, request: QueryRequest) -> Tuple[QueryMetadata, list[Any], Iterator[Any]]: + if request.compute_size == ComputeSize.AUTO and request.session_id: + raise ConfigurationError("compute_size AUTO cannot be combined with session_id") + payload = request.model_dump(exclude_none=True, by_alias=True) res: Optional[httpx.Response] = None try: @@ -232,10 +242,14 @@ def next_item() -> Optional[Any]: except json.JSONDecodeError as exc: raise ParseError("Failed to parse NDJSON line", line_index, line) from exc if line_index > 1 and isinstance(item, dict) and isinstance(item.get("error"), str): - raise QueryError(item["error"], line_index) + raise QueryError(item["error"], line_index, json.dumps(item)) return item return None + def raise_if_stream_error(item: Any) -> None: + if isinstance(item, dict) and isinstance(item.get("error"), str): + raise QueryError(item["error"], line_index, json.dumps(item)) + first_item = next_item() if first_item is None: res.close() @@ -249,6 +263,7 @@ def next_item() -> Optional[Any]: buffered_row: Optional[Any] = None second_item = next_item() if second_item is not None: + raise_if_stream_error(second_item) if isinstance(second_item, list): columns = second_item elif isinstance(second_item, dict) and isinstance(second_item.get("columns"), list): @@ -264,6 +279,7 @@ def row_iter() -> Iterator[Any]: item = next_item() if item is None: break + raise_if_stream_error(item) yield item except httpx.RequestError as e: self._handle_error(e) diff --git a/src/altertable_lakehouse/errors.py b/src/altertable_lakehouse/errors.py index f155626..803f9a2 100644 --- a/src/altertable_lakehouse/errors.py +++ b/src/altertable_lakehouse/errors.py @@ -20,6 +20,15 @@ def __init__(self, message: str, line_index: int, raw_content: str): self.line_index = line_index self.raw_content = raw_content + +class QueryError(AltertableLakehouseError): + """A backend error emitted inside a successful query NDJSON stream.""" + + def __init__(self, message: str, line_index: int, raw_content: str): + super().__init__(f"Query stream error at line {line_index}: {message}") + self.line_index = line_index + self.raw_content = raw_content + class ApiError(AltertableLakehouseError): def __init__(self, message: str, status_code: int): super().__init__(f"{status_code}: {message}") @@ -33,11 +42,3 @@ class BadRequestError(ApiError): class ConfigurationError(AltertableLakehouseError): pass - - -class QueryError(AltertableLakehouseError): - """A query failed after the server started its NDJSON response.""" - - def __init__(self, message: str, line_index: int): - super().__init__(f"{message} at line {line_index}") - self.line_index = line_index diff --git a/src/altertable_lakehouse/models.py b/src/altertable_lakehouse/models.py index 65c8deb..504578b 100644 --- a/src/altertable_lakehouse/models.py +++ b/src/altertable_lakehouse/models.py @@ -9,11 +9,23 @@ class ComputeSize(str, Enum): M = "M" L = "L" XL = "XL" + XXL = "2XL" + XXXL = "3XL" + XXXXL = "4XL" + AUTO = "AUTO" + + +class QueryFormat(str, Enum): + DEFAULT = "default" + CSV = "csv" + JSONL = "jsonl" + PARQUET = "parquet" class UploadMode(str, Enum): CREATE = "create" APPEND = "append" + CREATE_APPEND = "create_append" OVERWRITE = "overwrite" @@ -55,11 +67,13 @@ class QueryRequest(BaseModel): schema_: Optional[str] = Field(default=None, alias="schema") session_id: Optional[str] = None compute_size: Optional[ComputeSize] = None + dialect: Optional[str] = None sanitize: Optional[bool] = None limit: Optional[int] = None offset: Optional[int] = None timezone: Optional[str] = None ephemeral: Optional[bool] = None + format: Optional[QueryFormat] = None visible: Optional[bool] = None requested_by: Optional[str] = None query_id: Optional[str] = None diff --git a/tests/test_client.py b/tests/test_client.py index 6ba0549..aa73245 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,5 +1,6 @@ # type: ignore import os +import json import ssl from io import BytesIO import pytest @@ -75,6 +76,33 @@ def handler(request): assert "mode" not in captured["params"] +def test_upsert_forwards_cursor_field_and_content_type(client): + captured = {} + + def handler(request): + captured["params"] = dict(request.url.params) + captured["content_type"] = request.headers.get("content-type") + return httpx.Response(204, request=request) + + client._client = httpx.Client( + base_url=client.base_url, + transport=httpx.MockTransport(handler), + ) + + client.upsert( + catalog="cat", + schema="sch", + table="tbl", + primary_key="account_id,event_id", + cursor_field="updated_at,sequence", + content=b'[{"account_id":1}]', + content_type="application/json", + ) + + assert captured["params"]["cursor_field"] == "updated_at,sequence" + assert captured["content_type"] == "application/json" + + def test_upsert_requires_primary_key(client): with pytest.raises(TypeError, match="primary_key"): client.upsert(catalog="cat", schema="sch", table="tbl", content=b'{"id":1}') @@ -113,6 +141,29 @@ def handler(request): assert captured["content"] == b"id,name\n1,Alice\n" +def test_upload_supports_create_append(client): + captured = {} + + def handler(request): + captured["params"] = dict(request.url.params) + return httpx.Response(200, request=request) + + client._client = httpx.Client( + base_url=client.base_url, + transport=httpx.MockTransport(handler), + ) + + client.upload( + catalog="cat", + schema="sch", + table="tbl", + mode=models.UploadMode.CREATE_APPEND, + content=b"id,name\n1,Alice\n", + ) + + assert captured["params"]["mode"] == "create_append" + + def test_upload_omits_content_type_and_surfaces_api_errors(client): captured = {} @@ -182,6 +233,95 @@ def test_query_returns_stream_metadata_and_columns(client): assert columns == [{"name": "1", "type": "INTEGER"}] assert row_values == [[1]] + +def test_query_forwards_v013_options_and_rejects_auto_with_session(client): + captured = {} + + def handler(request): + captured["payload"] = json.loads(request.content) + return httpx.Response( + 200, + text='{"query_id":"query"}\n[]\n', + headers={"content-type": "application/x-ndjson"}, + request=request, + ) + + client._client = httpx.Client( + base_url=client.base_url, + transport=httpx.MockTransport(handler), + ) + + metadata, columns, rows = client.query( + models.QueryRequest( + statement="SELECT 1", + compute_size=models.ComputeSize.XXL, + dialect="postgres", + format=models.QueryFormat.JSONL, + ) + ) + + assert metadata.values["query_id"] == "query" + assert columns == [] + assert list(rows) == [] + assert captured["payload"] == { + "statement": "SELECT 1", + "compute_size": "2XL", + "dialect": "postgres", + "format": "jsonl", + } + + with pytest.raises(errors.ConfigurationError, match="AUTO"): + client.query( + models.QueryRequest( + statement="SELECT 1", + compute_size=models.ComputeSize.AUTO, + session_id="existing-session", + ) + ) + + +def test_query_raises_typed_error_from_ndjson_stream(client): + def handler(request): + return httpx.Response( + 200, + text='{"query_id":"query"}\n[{"name":"id","type":"INTEGER"}]\n{"error":"worker failed"}\n', + headers={"content-type": "application/x-ndjson"}, + request=request, + ) + + client._client = httpx.Client( + base_url=client.base_url, + transport=httpx.MockTransport(handler), + ) + + _, _, rows = client.query(models.QueryRequest(statement="SELECT 1")) + + with pytest.raises(errors.QueryError, match="worker failed") as exc_info: + list(rows) + + assert exc_info.value.line_index == 3 + assert exc_info.value.raw_content == '{"error": "worker failed"}' + + +def test_query_raises_typed_error_before_columns(client): + def handler(request): + return httpx.Response( + 200, + text='{"query_id":"query"}\n{"error":"schema failed"}\n', + headers={"content-type": "application/x-ndjson"}, + request=request, + ) + + client._client = httpx.Client( + base_url=client.base_url, + transport=httpx.MockTransport(handler), + ) + + with pytest.raises(errors.QueryError, match="schema failed") as exc_info: + client.query(models.QueryRequest(statement="SELECT 1")) + + assert exc_info.value.line_index == 2 + def test_client_forwards_verify_false(monkeypatch, base_url): captured = {}