[NOTICKET] feat(storage): read tabular Parquet from Supabase S3 via provider toggle

TabularExecutor was hard-wired to Azure Blob (az_blob:// location_ref +
AzureBlobStorage). After the Go data plane switched storage provider to
Supabase S3, location_ref became object_storage://... and tabular queries
failed with a ValueError before download.

- settings: add STORAGE_PROVIDER toggle + SUPABASE_S3_* config (mirrors Go)
- storage/object_storage: new isolated boto3 Supabase S3 read client
- tabular executor: accept object_storage:// prefix; pick backend by toggle

Bridge (Option C) until Go exposes a source-data endpoint; see
docs/proposal_go_source_data_endpoint.md.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/config/settings.py

@@ -63,6 +63,21 @@ class Settings(BaseSettings):
     azureai_container_name: str = Field(alias="azureai__container__name", default="")
     azureai_container_account_name: str = Field(alias="azureai__container__account__name", default="")
 
+    # Object storage provider toggle. Mirrors the Go data plane's `storage.provider`
+    # (see Orchestrator config: azure_blob | supabase_s3). Blank falls back to azure_blob.
+    # Tabular query execution reads the processed Parquet from whichever backend is active.
+    storage_provider: str = Field(alias="STORAGE_PROVIDER", default="azure_blob")
+
+    # Supabase S3-compatible object storage (active when storage_provider == "supabase_s3").
+    # BRIDGE (Option C): lets Python read the same bucket Go writes to, until Go exposes a
+    # source-data endpoint (see docs/proposal_go_source_data_endpoint.md) — then this client
+    # is removed and Python stops touching storage directly.
+    supabase_s3_bucket: str = Field(alias="SUPABASE_S3_BUCKET", default="")
+    supabase_s3_endpoint: str = Field(alias="SUPABASE_S3_ENDPOINT", default="")
+    supabase_s3_region: str = Field(alias="SUPABASE_S3_REGION", default="")
+    supabase_s3_access_key_id: str = Field(alias="SUPABASE_S3_ACCESS_KEY_ID", default="")
+    supabase_s3_secret_access_key: str = Field(alias="SUPABASE_S3_SECRET_ACCESS_KEY", default="")
+
     # Langfuse
     LANGFUSE_PUBLIC_KEY: str
     LANGFUSE_SECRET_KEY: str

src/query/executor/tabular.py

@@ -29,13 +29,18 @@ from .base import BaseExecutor, QueryResult
 logger = get_logger("tabular_executor")
 
 _AZ_BLOB_PREFIX = "az_blob://"
+# Go's Supabase S3 data plane writes location_ref with this prefix instead of az_blob://.
+# Both encode the same path structure after the prefix: {user_id}/{document_id}.
+_OBJECT_STORAGE_PREFIX = "object_storage://"
+_LOCATION_REF_PREFIXES = (_AZ_BLOB_PREFIX, _OBJECT_STORAGE_PREFIX)
 _ROW_HARD_CAP = 10_000
 
 
 class TabularExecutor(BaseExecutor):
     """Executes compiled pandas chain on a Parquet blob.
 
-    `fetch_blob` is injectable for tests — defaults to AzureBlobStorage.
+    `fetch_blob` is injectable for tests — defaults to the storage backend
+    selected by `settings.storage_provider` (azure_blob | supabase_s3).
     """
 
     def __init__(
@@ -49,6 +54,16 @@ class TabularExecutor(BaseExecutor):
 
     @staticmethod
     async def _default_fetch_blob(blob_name: str) -> bytes:
+        # Pick the storage backend by the same toggle the Go data plane uses.
+        # Blank/unknown falls back to Azure Blob (the pre-migration default).
+        from ...config.settings import settings
+
+        provider = (settings.storage_provider or "").strip().lower()
+        if provider == "supabase_s3":
+            from ...storage.object_storage import object_storage
+
+            return await object_storage.download_file(blob_name)
+
         from ...storage.az_blob.az_blob import blob_storage
 
         return await blob_storage.download_file(blob_name)
@@ -157,15 +172,19 @@ def _resolve_blob_name(source: Source, table: Table) -> str:
     on the upload pipeline preserving the file extension, which it does today
     because `Document.filename` is set once at upload and never renamed.
     """
-    if not source.location_ref.startswith(_AZ_BLOB_PREFIX):
+    matched_prefix = next(
+        (p for p in _LOCATION_REF_PREFIXES if source.location_ref.startswith(p)),
+        None,
+    )
+    if matched_prefix is None:
         raise ValueError(
-            f"TabularExecutor expects 'az_blob://...' location_ref, "
-            f"got {source.location_ref!r}"
+            f"TabularExecutor expects 'az_blob://...' or 'object_storage://...' "
+            f"location_ref, got {source.location_ref!r}"
         )
-    path = source.location_ref[len(_AZ_BLOB_PREFIX):]
+    path = source.location_ref[len(matched_prefix):]
     parts = path.split("/", 1)
     if len(parts) != 2 or not parts[0] or not parts[1]:
-        raise ValueError(f"Malformed az_blob location_ref: {source.location_ref!r}")
+        raise ValueError(f"Malformed location_ref: {source.location_ref!r}")
     user_id, document_id = parts
     is_xlsx = source.name.lower().endswith(".xlsx")
     sheet_name = table.name if is_xlsx else None

src/storage/object_storage/__init__.py

@@ -0,0 +1,13 @@
+"""Supabase S3-compatible object storage (read side).
+
+BRIDGE MODULE (Option C) — self-contained on purpose. It exists only so the
+tabular query path can read the processed Parquet from the same bucket the Go
+data plane writes to, after Go switched storage provider Azure Blob -> Supabase S3.
+
+Once Go exposes a source-data endpoint (docs/proposal_go_source_data_endpoint.md),
+delete this whole package and point TabularExecutor at that endpoint instead.
+"""
+
+from .supabase_s3 import object_storage
+
+__all__ = ["object_storage"]

src/storage/object_storage/supabase_s3.py

@@ -0,0 +1,85 @@
+"""Supabase S3-compatible object storage client (read side).
+
+Mirrors the Go data plane's Supabase S3 client (Orchestrator
+`internal/documents/supabase_s3.go`): path-style addressing against a custom
+endpoint, static credentials, single bucket.
+
+boto3 is synchronous; download runs in a worker thread so the async call sites
+(TabularExecutor._fetch_blob) stay non-blocking. Only the read path is
+implemented — Python is read-only here; Go owns all writes.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from src.config.settings import settings
+from src.middlewares.logging import get_logger
+
+logger = get_logger("supabase_s3")
+
+
+class SupabaseS3Storage:
+    """Read-only client for the Supabase S3 bucket Go writes to."""
+
+    def __init__(self) -> None:
+        self._bucket = settings.supabase_s3_bucket
+        self._endpoint = settings.supabase_s3_endpoint.rstrip("/")
+        self._region = settings.supabase_s3_region
+        self._access_key_id = settings.supabase_s3_access_key_id
+        self._secret_access_key = settings.supabase_s3_secret_access_key
+        self._client = None  # lazy — avoid import-time failure on Azure deployments
+
+    def _is_configured(self) -> bool:
+        return all(
+            (
+                self._bucket,
+                self._endpoint,
+                self._region,
+                self._access_key_id,
+                self._secret_access_key,
+            )
+        )
+
+    def _get_client(self):
+        if self._client is None:
+            if not self._is_configured():
+                raise RuntimeError(
+                    "Supabase S3 storage is not fully configured "
+                    "(set STORAGE_PROVIDER=supabase_s3 and SUPABASE_S3_* env vars)"
+                )
+            import boto3
+            from botocore.config import Config
+
+            self._client = boto3.client(
+                "s3",
+                endpoint_url=self._endpoint,
+                region_name=self._region,
+                aws_access_key_id=self._access_key_id,
+                aws_secret_access_key=self._secret_access_key,
+                config=Config(
+                    signature_version="s3v4",
+                    s3={"addressing_style": "path"},  # path-style, like Go's UsePathStyle
+                ),
+            )
+        return self._client
+
+    def _download_sync(self, object_name: str) -> bytes:
+        client = self._get_client()
+        resp = client.get_object(Bucket=self._bucket, Key=object_name)
+        return resp["Body"].read()
+
+    async def download_file(self, object_name: str) -> bytes:
+        """Download an object's bytes. Interface-compatible with AzureBlobStorage.download_file."""
+        try:
+            logger.info(f"Downloading object {object_name}")
+            content = await asyncio.to_thread(self._download_sync, object_name)
+            logger.info(f"Successfully downloaded {object_name} ({len(content)} bytes)")
+            return content
+        except Exception as e:
+            logger.error(f"Failed to download object {object_name}", error=str(e))
+            raise
+
+
+# Singleton (lazy client construction — safe to import even when not configured)
+object_storage = SupabaseS3Storage()