feat: update artifact handling and schema management in execution pipeline

- Enhanced artifact handling by introducing a unified method for creating artifact references across different storage backends (S3, ADLS, Local).
- Refactored the execution contracts to replace the deprecated ExecutorBaseModel with ExecutorResponse, improving clarity and consistency.
- Updated the ExecutorNode to include tenant_id in requests, facilitating multi-tenant support.
- Improved error handling and logging across various pipeline nodes, ensuring better traceability and debugging capabilities.
- Removed unused schema management methods and streamlined the datasource resolution process for improved performance.

Author: nadeem4

.gitignore

@@ -26,4 +26,6 @@ site
 
 data
 
-last_reasoning.json
+last_reasoning.json
+
+artifacts/

packages/adapter-sdk/src/nl2sql_adapter_sdk/__init__.py

@@ -1,7 +1,7 @@
 """Adapter SDK: shared contracts for core and adapters."""
 
 from .capabilities import DatasourceCapability
-from .contracts import AdapterRequest, ResultColumn, ResultError, ResultFrame
+from .contracts import AdapterRequest, ResultError, ResultFrame
 from .protocols import DatasourceAdapterProtocol
 from .schema import (
     Column,
@@ -21,7 +21,6 @@
 __all__ = [
     "DatasourceCapability",
     "AdapterRequest",
-    "ResultColumn",
     "ResultError",
     "ResultFrame",
     "DatasourceAdapterProtocol",

packages/adapter-sdk/src/nl2sql_adapter_sdk/contracts.py

@@ -26,12 +26,6 @@ class AdapterRequest(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
 
-class ResultColumn(BaseModel):
-    """Column metadata for a ResultFrame."""
-
-    name: str
-    type: str = Field(default="unknown", description="Logical or native column type.")
-
 
 class ResultError(BaseModel):
     """Standardized error envelope for adapter results."""
@@ -49,7 +43,7 @@ class ResultFrame(BaseModel):
     """Adapter-agnostic, DataFrame-like result contract."""
 
     success: bool = Field(default=True)
-    columns: List[ResultColumn] = Field(default_factory=list)
+    columns: List[str] = Field(default_factory=list)
     rows: List[List[Any]] = Field(default_factory=list)
     row_count: int = Field(default=0)
     truncated: bool = Field(default=False)
@@ -75,11 +69,10 @@ def from_row_dicts(
         if columns is None:
             columns = list(rows[0].keys()) if rows else []
 
-        col_specs = [ResultColumn(name=col, type="unknown") for col in columns]
         row_values = [[row.get(col) for col in columns] for row in rows]
 
         return cls(
-            columns=col_specs,
+            columns=columns,
             rows=row_values,
             row_count=row_count if row_count is not None else len(row_values),
             **kwargs,
@@ -90,5 +83,5 @@ def to_row_dicts(self) -> List[Dict[str, Any]]:
 
         if not self.rows or not self.columns:
             return []
-        names = [col.name for col in self.columns]
+        names = self.columns
         return [dict(zip(names, row)) for row in self.rows]

packages/adapter-sqlalchemy/src/nl2sql_sqlalchemy_adapter/adapter.py

@@ -5,7 +5,6 @@
 from nl2sql_adapter_sdk.capabilities import DatasourceCapability
 from nl2sql_adapter_sdk.contracts import (
     AdapterRequest,
-    ResultColumn,
     ResultError,
     ResultFrame,
 )
@@ -155,7 +154,7 @@ def execute_sql(self, sql: str) -> ResultFrame:
 
         return ResultFrame(
             success=True,
-            columns=[ResultColumn(name=col, type="unknown") for col in cols],
+            columns=cols,
             rows=rows,
             row_count=row_count,
             bytes=total_bytes,

packages/core/src/nl2sql/aggregation/aggregator.py

@@ -34,7 +34,7 @@ def execute(
 
         computed: Dict[str, pl.DataFrame] = {}
 
-        for layer in dag.layers or []:
+        for layer in dag.layers:
             for node_id in layer:
                 node = node_index.get(node_id)
                 if not node:

packages/core/src/nl2sql/aggregation/engines/polars_duckdb.py

@@ -16,13 +16,7 @@ def __init__(self):
         self.artifact_store = build_artifact_store()
 
     def load_scan(self, artifact: ArtifactRef) -> pl.DataFrame:
-        if artifact.uri.startswith("s3://") or artifact.uri.startswith("abfs://"):
-            result_frame = self.artifact_store.read_result_frame(artifact)
-            return pl.from_dicts(result_frame.to_row_dicts())
-
-        relation = duckdb.query(f"SELECT * FROM '{artifact.uri}'")
-        table = relation.arrow()
-        return pl.from_arrow(table)
+        return self.artifact_store.read_parquet(artifact)
 
     def combine(
         self,

packages/core/src/nl2sql/common/errors.py

@@ -47,7 +47,8 @@ class ErrorCode(str, Enum):
     PIPELINE_TIMEOUT = "PIPELINE_TIMEOUT"
     SERVICE_UNAVAILABLE = "SERVICE_UNAVAILABLE"
     EXECUTION_TIMEOUT = "EXECUTION_TIMEOUT"
-    CANCELLED = "CANCELLED"
+    CANCELLED = "CANCELLED",
+    EXECUTION_FAILED = "EXECUTION_FAILED"
 
 
 

packages/core/src/nl2sql/context.py

@@ -48,6 +48,7 @@ def __init__(
             vector_store_path = pathlib.Path(settings.vector_store_path)
 
         cm = ConfigManager()
+        self.tenant_id = settings.tenant_id
         self.config_manager = cm
 
         secret_configs = cm.load_secrets(secrets_config_path)

packages/core/src/nl2sql/datasources/registry.py

@@ -112,24 +112,6 @@ def register_datasource(self, config: DatasourceConfig) -> DatasourceAdapterProt
                 f"No adapter found for engine type: '{conn_type}' in datasource '{ds_id}'"
             )
 
-    def refresh_schema(self, datasource_id: str, vector_store: Any) -> Dict[str, int]:
-        """Refreshes the schema for a specific datasource.
-        
-        This triggers a fresh intrusion of the database schema via the adapter
-        and updates the vector store index.
-
-        Args:
-            datasource_id: The ID of the datasource to refresh.
-            vector_store: The VectorStore instance.
-
-        Returns:
-            Dict[str, int]: Statistics of the refreshed components.
-
-        Raises:
-            ValueError: If the datasource ID is unknown.
-        """
-        adapter = self.get_adapter(datasource_id)
-        return vector_store.refresh_schema(adapter, datasource_id)
 
     def get_adapter(self, datasource_id: str) -> DatasourceAdapterProtocol:
         """Retrieves the adapter for a datasource.

packages/core/src/nl2sql/execution/__init__.py

@@ -1,4 +1,4 @@
-from .contracts import ArtifactRef, ExecutorBaseModel, ExecutorRequest
+from .contracts import ArtifactRef, ExecutorResponse, ExecutorRequest
 from .execution_store import ExecutionStore
 
-__all__ = ["ArtifactRef", "ExecutorBaseModel", "ExecutorRequest", "ExecutionStore"]
+__all__ = ["ArtifactRef", "ExecutorResponse", "ExecutorRequest", "ExecutionStore"]