feat(gooddata-sdk): add AI Lake service with analyze_statistics

Wrap the AI Lake long-running-operation surface with a typed
`CatalogAILakeService` exposing the three methods consumers need to
trigger and wait on `ANALYZE TABLE` after registering aggregate-aware
LDM shapes:

- `analyze_statistics(instance_id, table_names=None, operation_id=None)`
  → `str`: posts the request and returns the operation ID. The caller
  can pre-supply a UUID; otherwise the service generates one and seeds
  it as the request `operation-id` header so the polling handle is
  known up front (the endpoint returns `Unit` body + the id in the
  response header).
- `get_operation(operation_id)` → `CatalogAILakeOperation`: typed
  handle with `id`, `kind`, `status` (Literal "pending"/"succeeded"/
  "failed" — these are the discriminator values of the OpenAPI
  `Operation` oneOf), and optional `result`/`error` payloads. Convenience
  predicates `is_terminal`, `is_succeeded`, `is_failed`.
- `wait_for_operation(operation_id, timeout_s=300, poll_s=2)`: blocks
  until terminal, raises `CatalogAILakeOperationError` on `failed` and
  `TimeoutError` when the deadline elapses.

Wiring:
- `GoodDataApiClient.ai_lake_api` property now exposes `apis.AILakeApi`
  for callers who need raw access to surfaces this service does not
  yet wrap (database provisioning, pipe-table CRUD, service commands —
  follow-up tickets).
- `GoodDataSdk.catalog_ai_lake` property surfaces the new service.
- Public re-exports: `CatalogAILakeOperation`, `CatalogAILakeOperationError`,
  `CatalogAILakeService`.

Tests (8 tests, all unit-mocked — no live stack needed):
- analyze_statistics seeds caller-supplied UUID, generates one when
  omitted, normalizes empty `table_names`
- get_operation handles success and failure shapes
- wait_for_operation polls until succeeded, raises on failed,
  raises TimeoutError when never terminal

Verified: full SDK unit suite — 425 passed, 2 skipped.

JIRA: CQ-2320
risk: low

Author: jaceksan

packages/gooddata-sdk/src/gooddata_sdk/__init__.py

@@ -7,6 +7,11 @@
 import logging
 
 from gooddata_sdk._version import __version__
+from gooddata_sdk.catalog.ai_lake.service import (
+    CatalogAILakeOperation,
+    CatalogAILakeOperationError,
+    CatalogAILakeService,
+)
 from gooddata_sdk.catalog.appearance.entity_model.color_palette import (
     CatalogColorPalette,
     CatalogColorPaletteAttributes,

packages/gooddata-sdk/src/gooddata_sdk/catalog/ai_lake/__init__.py

@@ -0,0 +1 @@
+# (C) 2026 GoodData Corporation

packages/gooddata-sdk/src/gooddata_sdk/catalog/ai_lake/service.py

@@ -0,0 +1,144 @@
+# (C) 2026 GoodData Corporation
+"""SDK wrapper for the AI Lake long-running-operation surface.
+
+Today this exposes only the operations needed by aggregate-aware LDMs:
+
+- `analyze_statistics` triggers `ANALYZE TABLE` over a database instance so
+  CBO statistics catch up after a schema or data change. Required after
+  registering a pre-aggregation table whose dim attributes the platform will
+  later resolve via filter pushdown.
+- `get_operation` and `wait_for_operation` cover the polling side of the
+  long-running operation contract that `analyze_statistics` returns.
+
+The full AI Lake API surface (database provisioning, pipe-table
+registration, service commands) is not yet wrapped here; consumers that
+need those should call `client.ai_lake_api.<method>` directly until a
+ticket adds typed wrappers.
+"""
+
+from __future__ import annotations
+
+import time
+import uuid
+from typing import Any, Literal
+
+from attrs import define
+from gooddata_api_client.api.ai_lake_api import AILakeApi
+from gooddata_api_client.model.analyze_statistics_request import AnalyzeStatisticsRequest
+
+from gooddata_sdk.catalog.base import Base
+from gooddata_sdk.client import GoodDataApiClient
+
+# AI Lake operation status values (lower-case on the wire — these are the
+# discriminator values of the `Operation` oneOf on the OpenAPI side).
+OperationStatus = Literal["pending", "succeeded", "failed"]
+TERMINAL_STATUSES: frozenset[OperationStatus] = frozenset({"succeeded", "failed"})
+
+
+@define(kw_only=True)
+class CatalogAILakeOperation(Base):
+    """Long-running-operation handle returned by AI Lake actions."""
+
+    id: str
+    kind: str
+    status: OperationStatus
+    result: dict[str, Any] | None = None
+    error: dict[str, Any] | None = None
+
+    @property
+    def is_terminal(self) -> bool:
+        return self.status in TERMINAL_STATUSES
+
+    @property
+    def is_succeeded(self) -> bool:
+        return self.status == "succeeded"
+
+    @property
+    def is_failed(self) -> bool:
+        return self.status == "failed"
+
+
+class CatalogAILakeOperationError(RuntimeError):
+    """Raised when an AI Lake long-running operation finishes in `failed` state."""
+
+    def __init__(self, operation: CatalogAILakeOperation) -> None:
+        self.operation = operation
+        message = f"AI Lake operation {operation.id} ({operation.kind}) failed"
+        if operation.error:
+            message = f"{message}: {operation.error}"
+        super().__init__(message)
+
+
+class CatalogAILakeService:
+    """Typed access to the AI Lake long-running-operation surface."""
+
+    def __init__(self, api_client: GoodDataApiClient) -> None:
+        self._client = api_client
+        self._ai_lake_api: AILakeApi = AILakeApi(api_client._api_client)
+
+    def analyze_statistics(
+        self,
+        instance_id: str,
+        table_names: list[str] | None = None,
+        operation_id: str | None = None,
+    ) -> str:
+        """Trigger ANALYZE TABLE for tables in an AI Lake database instance.
+
+        Args:
+            instance_id: Database instance name (preferred) or UUID.
+            table_names: Tables to analyze; if `None` or empty, every table
+                in the instance is analyzed.
+            operation_id: Optional client-supplied operation identifier. If
+                omitted, a fresh UUID is generated. Pass the same value that
+                `wait_for_operation` will poll on.
+
+        Returns:
+            The operation ID (UUID string) the platform will track this run
+            under. Pass it to `get_operation` / `wait_for_operation` to poll.
+        """
+        op_id = operation_id or str(uuid.uuid4())
+        request = AnalyzeStatisticsRequest(table_names=list(table_names) if table_names else [])
+        # Body return is `Unit`; the platform tracks the operation under the
+        # client-supplied or server-generated `operation-id` header. We seed
+        # it ourselves so the caller gets a known polling handle without
+        # having to read response headers.
+        self._ai_lake_api.analyze_statistics(instance_id, request, operation_id=op_id)
+        return op_id
+
+    def get_operation(self, operation_id: str) -> CatalogAILakeOperation:
+        """Fetch the current state of a long-running AI Lake operation."""
+        response = self._ai_lake_api.get_ai_lake_operation(operation_id)
+        # The api-client returns the oneOf'd operation as a dict-like object
+        # whose concrete subtype depends on `status`. Normalize via to_dict.
+        data = response.to_dict() if hasattr(response, "to_dict") else dict(response)
+        return CatalogAILakeOperation(
+            id=data["id"],
+            kind=data["kind"],
+            status=data["status"],
+            result=data.get("result"),
+            error=data.get("error"),
+        )
+
+    def wait_for_operation(
+        self,
+        operation_id: str,
+        timeout_s: float = 300.0,
+        poll_s: float = 2.0,
+    ) -> CatalogAILakeOperation:
+        """Block until an AI Lake operation reaches a terminal status.
+
+        Raises `CatalogAILakeOperationError` on `failed` and `TimeoutError`
+        if the operation does not reach a terminal state in time.
+        """
+        deadline = time.monotonic() + timeout_s
+        while True:
+            op = self.get_operation(operation_id)
+            if op.is_succeeded:
+                return op
+            if op.is_failed:
+                raise CatalogAILakeOperationError(op)
+            if time.monotonic() >= deadline:
+                raise TimeoutError(
+                    f"AI Lake operation {operation_id} did not finish within {timeout_s}s (last status: {op.status})"
+                )
+            time.sleep(poll_s)

packages/gooddata-sdk/src/gooddata_sdk/client.py

@@ -71,6 +71,7 @@ def __init__(
         self._actions_api = apis.ActionsApi(self._api_client)
         self._user_management_api = apis.UserManagementApi(self._api_client)
         self._appearance_api = apis.AppearanceApi(self._api_client)
+        self._ai_lake_api = apis.AILakeApi(self._api_client)
         self._executions_cancellable = executions_cancellable
 
     def _do_post_request(
@@ -158,6 +159,10 @@ def user_management_api(self) -> apis.UserManagementApi:
     def appearance_api(self) -> apis.AppearanceApi:
         return self._appearance_api
 
+    @property
+    def ai_lake_api(self) -> apis.AILakeApi:
+        return self._ai_lake_api
+
     @property
     def executions_cancellable(self) -> bool:
         return self._executions_cancellable

packages/gooddata-sdk/src/gooddata_sdk/sdk.py

@@ -3,6 +3,7 @@
 
 from pathlib import Path
 
+from gooddata_sdk.catalog.ai_lake.service import CatalogAILakeService
 from gooddata_sdk.catalog.appearance.service import CatalogAppearanceService
 from gooddata_sdk.catalog.data_source.service import CatalogDataSourceService
 from gooddata_sdk.catalog.export.service import ExportService
@@ -88,6 +89,7 @@ def __init__(self, client: GoodDataApiClient) -> None:
         self._tables = TableService(self._client)
         self._support = SupportService(self._client)
         self._catalog_permission = CatalogPermissionService(self._client)
+        self._catalog_ai_lake = CatalogAILakeService(self._client)
         self._export = ExportService(self._client)
 
     @property
@@ -138,6 +140,10 @@ def catalog_permission(self) -> CatalogPermissionService:
     def export(self) -> ExportService:
         return self._export
 
+    @property
+    def catalog_ai_lake(self) -> CatalogAILakeService:
+        return self._catalog_ai_lake
+
     @property
     def client(self) -> GoodDataApiClient:
         return self._client

packages/gooddata-sdk/tests/catalog/unit_tests/test_ai_lake_service.py

@@ -0,0 +1,146 @@
+# (C) 2026 GoodData Corporation
+"""Unit tests for `CatalogAILakeService`.
+
+These tests use plain mocks against `AILakeApi` rather than vcr cassettes
+because the service is a thin wrapper whose interesting logic
+(`wait_for_operation` polling, status discrimination, UUID-seeding) is
+deterministic and worth verifying without a live stack.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+from gooddata_sdk.catalog.ai_lake.service import (
+    CatalogAILakeOperation,
+    CatalogAILakeOperationError,
+    CatalogAILakeService,
+)
+
+
+def _make_service() -> tuple[CatalogAILakeService, MagicMock]:
+    """Build a service whose api-client side is fully mocked."""
+    fake_ai_lake_api = MagicMock(name="AILakeApi")
+    fake_client = SimpleNamespace(_api_client=MagicMock(name="ApiClient"))
+
+    with patch("gooddata_sdk.catalog.ai_lake.service.AILakeApi", return_value=fake_ai_lake_api):
+        service = CatalogAILakeService(fake_client)  # type: ignore[arg-type]
+
+    assert service._ai_lake_api is fake_ai_lake_api
+    return service, fake_ai_lake_api
+
+
+def _operation_response(**fields: object) -> MagicMock:
+    """Mimic the api-client's deserialized response (`.to_dict()` returns the raw dict)."""
+    response = MagicMock()
+    response.to_dict.return_value = fields
+    return response
+
+
+class TestAnalyzeStatistics:
+    def test_seeds_caller_supplied_operation_id(self) -> None:
+        service, api = _make_service()
+        op_id = service.analyze_statistics(
+            instance_id="demo-db",
+            table_names=["fact_orders", "dim_country"],
+            operation_id="11111111-2222-3333-4444-555555555555",
+        )
+        assert op_id == "11111111-2222-3333-4444-555555555555"
+        assert api.analyze_statistics.call_count == 1
+        call_args = api.analyze_statistics.call_args
+        assert call_args.args[0] == "demo-db"
+        assert call_args.kwargs["operation_id"] == "11111111-2222-3333-4444-555555555555"
+        # Body request carries the table names.
+        body = call_args.args[1]
+        assert list(body.table_names) == ["fact_orders", "dim_country"]
+
+    def test_generates_uuid_when_not_supplied(self) -> None:
+        service, api = _make_service()
+        op_id = service.analyze_statistics(instance_id="demo-db")
+        # UUID4 string format: 8-4-4-4-12 hex chars.
+        assert len(op_id) == 36 and op_id.count("-") == 4
+        assert api.analyze_statistics.call_args.kwargs["operation_id"] == op_id
+
+    def test_empty_table_names_is_normalized_to_empty_list(self) -> None:
+        service, api = _make_service()
+        service.analyze_statistics(instance_id="demo-db", table_names=None)
+        assert list(api.analyze_statistics.call_args.args[1].table_names) == []
+
+
+class TestGetOperation:
+    def test_normalizes_to_typed_handle(self) -> None:
+        service, api = _make_service()
+        api.get_ai_lake_operation.return_value = _operation_response(
+            id="op-1",
+            kind="analyze-statistics",
+            status="succeeded",
+            result={"tablesAnalyzed": 7},
+        )
+        op = service.get_operation("op-1")
+        assert isinstance(op, CatalogAILakeOperation)
+        assert op.id == "op-1"
+        assert op.kind == "analyze-statistics"
+        assert op.is_succeeded
+        assert op.result == {"tablesAnalyzed": 7}
+        assert op.error is None
+
+    def test_carries_error_payload_on_failed(self) -> None:
+        service, api = _make_service()
+        api.get_ai_lake_operation.return_value = _operation_response(
+            id="op-2",
+            kind="analyze-statistics",
+            status="failed",
+            error={"code": "ANALYZE_FAILED", "message": "table not found"},
+        )
+        op = service.get_operation("op-2")
+        assert op.is_failed
+        assert op.error == {"code": "ANALYZE_FAILED", "message": "table not found"}
+
+
+class TestWaitForOperation:
+    def test_polls_until_succeeded(self) -> None:
+        service, api = _make_service()
+        api.get_ai_lake_operation.side_effect = [
+            _operation_response(id="op", kind="analyze-statistics", status="pending"),
+            _operation_response(id="op", kind="analyze-statistics", status="pending"),
+            _operation_response(id="op", kind="analyze-statistics", status="succeeded", result={}),
+        ]
+        with patch("gooddata_sdk.catalog.ai_lake.service.time.sleep") as fake_sleep:
+            op = service.wait_for_operation("op", poll_s=0.5)
+        assert op.is_succeeded
+        assert api.get_ai_lake_operation.call_count == 3
+        # Slept twice between three polls.
+        assert fake_sleep.call_args_list == [call(0.5), call(0.5)]
+
+    def test_raises_on_failed_terminal_status(self) -> None:
+        service, api = _make_service()
+        api.get_ai_lake_operation.return_value = _operation_response(
+            id="op", kind="analyze-statistics", status="failed", error={"message": "boom"}
+        )
+        with (
+            patch("gooddata_sdk.catalog.ai_lake.service.time.sleep"),
+            pytest.raises(CatalogAILakeOperationError) as exc_info,
+        ):
+            service.wait_for_operation("op")
+        assert exc_info.value.operation.is_failed
+        assert "boom" in str(exc_info.value)
+
+    def test_times_out_when_never_terminal(self) -> None:
+        service, api = _make_service()
+        api.get_ai_lake_operation.return_value = _operation_response(
+            id="op", kind="analyze-statistics", status="pending"
+        )
+        # Make `time.monotonic` advance past the deadline on the second call so
+        # the loop runs a few iterations before raising.
+        with (
+            patch("gooddata_sdk.catalog.ai_lake.service.time.sleep"),
+            patch(
+                "gooddata_sdk.catalog.ai_lake.service.time.monotonic",
+                side_effect=[0.0, 1.0, 5.0, 11.0],
+            ),
+            pytest.raises(TimeoutError) as exc_info,
+        ):
+            service.wait_for_operation("op", timeout_s=10.0, poll_s=0.1)
+        assert "did not finish within 10.0s" in str(exc_info.value)