transformations - updates

.github/workflows/test_common.yml

@@ -110,7 +110,7 @@ jobs:
           pytest tests/pipeline/test_pipeline_extra.py -k arrow ${{ matrix.pytest_args }}
 
       - name: Install pipeline and sources dependencies
-        run: uv sync ${{ matrix.uv_sync_args }} --extra duckdb --extra cli --extra parquet --extra deltalake --extra sql_database --group sentry-sdk --group pipeline --group sources
+        run: uv sync ${{ matrix.uv_sync_args }} --extra duckdb --extra cli --extra parquet --extra deltalake --extra sql_database --group sentry-sdk --group pipeline --group sources --group ibis
 
       - name: Run extract and pipeline tests
         run: |

dlt/__init__.py

@@ -31,7 +31,8 @@
 from dlt.extract.decorators import source, resource, transformer, defer
 from dlt.destinations.decorators import destination
 from dlt.transformations.decorators import transformation
-from dlt.destinations.dataset import dataset, ReadableDBAPIDataset as Dataset
+from dlt.common.destination.dataset import Dataset, Relation
+from dlt.destinations.dataset import dataset
 
 from dlt.pipeline import (
     pipeline as _pipeline,
@@ -82,6 +83,7 @@
     "sources",
     "destinations",
     "Dataset",
+    "Relation",
     "dataset",
     "transformation",
 ]

dlt/common/data_writers/writers.py

@@ -191,16 +191,9 @@ def write_header(self, columns_schema: TTableSchemaColumns) -> None:
     def write_data(self, items: Sequence[TDataItem]) -> None:
         super().write_data(items)
         for item in items:
-            dialect = item.dialect or (self._caps.sqlglot_dialect if self._caps else None)
-            query = item.query
-            parsed_query = sqlglot.parse_one(query, read=dialect)
-
-            # Ensure the parsed query is a SELECT statement
-            if not isinstance(parsed_query, sqlglot.exp.Select):
-                raise ValueError("Only SELECT statements are allowed to write model files.")
-
-            normalized_query = parsed_query.sql(dialect=dialect)
-            self._f.write("dialect: " + (dialect or "") + "\n" + normalized_query + "\n")
+            dialect = item.query_dialect() or (self._caps.sqlglot_dialect if self._caps else None)
+            query = item.to_sql()
+            self._f.write("dialect: " + (dialect or "") + "\n" + query + "\n")
 
     @classmethod
     def writer_spec(cls) -> FileWriterSpec:

dlt/common/destination/capabilities.py

@@ -3,7 +3,6 @@
     Any,
     Callable,
     ClassVar,
-    Iterable,
     Literal,
     Optional,
     Sequence,
@@ -12,7 +11,7 @@
     Protocol,
     Type,
 )
-from dlt.common.data_types import TDataType
+from dlt.common.libs.sqlglot import TSqlGlotDialect
 from dlt.common.exceptions import TerminalValueError
 from dlt.common.normalizers.typing import TNamingConventionReferenceArg
 from dlt.common.typing import TLoaderFileFormat, get_args
@@ -202,7 +201,7 @@ class DestinationCapabilitiesContext(ContainerInjectableContext):
     enforces_nulls_on_alter: bool = True
     """Tells if destination enforces null constraints when adding NOT NULL columns to existing tables"""
 
-    sqlglot_dialect: Optional[str] = None
+    sqlglot_dialect: Optional[TSqlGlotDialect] = None
     """The SQL dialect used by sqlglot to transpile a query to match the destination syntax."""
 
     def generates_case_sensitive_identifiers(self) -> bool: