detect XLSX by filename, not table count

src/query/executor/tabular.py

@@ -20,6 +20,7 @@ from typing import Any
 import pandas as pd
 
 from ...catalog.models import Catalog, Source, Table
+from ...knowledge.parquet_service import parquet_blob_name
 from ...middlewares.logging import get_logger
 from ..compiler.pandas import CompiledPandas, PandasCompiler
 from ..ir.models import QueryIR
@@ -131,9 +132,19 @@ class TabularExecutor(BaseExecutor):
 def _resolve_blob_name(source: Source, table: Table) -> str:
     """Map source.location_ref + table → the Parquet blob name to download.
 
-    Convention:
-      CSV / single-sheet Parquet → ``{user_id}/{document_id}.parquet``
-      XLSX (multi-sheet)        → ``{user_id}/{document_id}__{table.name}.parquet``
+    Delegates to ``parquet_service.parquet_blob_name`` so the same naming
+    convention (and ``_safe_sheet_name`` sanitization) is used on both the
+    write side (ingestion) and the read side (query execution).
+
+      CSV / Parquet → ``{user_id}/{document_id}.parquet``
+      XLSX          → ``{user_id}/{document_id}__{safe_sheet}.parquet``
+                      (writer always uploads with sheet suffix for XLSX,
+                      regardless of sheet count — see processing_service
+                      `_build_excel_documents`)
+
+    XLSX is detected via ``Source.name`` (the original filename). This relies
+    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):
         raise ValueError(
@@ -145,9 +156,9 @@ def _resolve_blob_name(source: Source, table: Table) -> str:
     if len(parts) != 2 or not parts[0] or not parts[1]:
         raise ValueError(f"Malformed az_blob location_ref: {source.location_ref!r}")
     user_id, document_id = parts
-    if len(source.tables) == 1:
-        return f"{user_id}/{document_id}.parquet"
-    return f"{user_id}/{document_id}__{table.name}.parquet"
+    is_xlsx = source.name.lower().endswith(".xlsx")
+    sheet_name = table.name if is_xlsx else None
+    return parquet_blob_name(user_id, document_id, sheet_name)
 
 
 def _load_and_apply(blob_bytes: bytes, compiled: CompiledPandas) -> pd.DataFrame: