Upgrade dlt to 1.20

Datetime handling has been overhauled and as a result databases
such as mssql now properly reflect tz-naive columns. We
add in a mapping for the tz-naive column to the Iceberg type.

Author: martyngigg

elt-common/pyproject.toml

@@ -8,7 +8,10 @@ version = "0.1.0"
 description = "A set of common utility routines for data pipelines."
 readme = "README.md"
 requires-python = ">=3.13"
-dependencies = ["dlt[parquet,s3]~=1.15.0", "pyiceberg~=0.9.1"]
+dependencies = [
+    "dlt[parquet,s3]~=1.20.0",
+    "pyiceberg~=0.9.1",
+]
 
 
 [project.optional-dependencies]

elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py

@@ -29,6 +29,7 @@
     PrimitiveType,
     StringType,
     TimeType,
+    TimestampType,
     TimestamptzType,
 )
 from pyiceberg.typedef import Identifier
@@ -107,7 +108,10 @@ def dlt_type_to_iceberg(column: TColumnType) -> PrimitiveType:
             raise TypeError(
                 f"Iceberg v1 & v2 does not support timestamps in '{TIMESTAMP_PRECISION_TO_UNIT[9]}' precision."  # type:ignore
             )
-        return TimestamptzType()
+        if column.get("timezone", True):
+            return TimestamptzType()
+        else:
+            return TimestampType()
     elif dlt_type == "binary":
         return BinaryType()
     else:

elt-common/uv.lock

@@ -21,11 +21,11 @@
 from dlt.sources.sql_database import sql_table
 from html2text import html2text
 import pyarrow as pa
-from sqlalchemy.sql import Select
+import sqlalchemy as sa
 
 import elt_common.cli as cli_utils
 
-OPRALOG_EPOCH = dt.datetime(2017, 4, 25, 0, 0, 0, tzinfo=dt.UTC)
+OPRALOG_EPOCH = dt.datetime(2017, 4, 25, 0, 0, 0)
 SQL_TABLE_KWARGS = dict(
     schema=dlt.config.value,
     backend="pyarrow",
@@ -36,8 +36,19 @@
 EXTRACTED_ENTRY_IDS: List[int] = []
 
 
+def with_resource_limit(
+    resource: DltResource, limit_max_items: int | None = None
+) -> DltResource:
+    if limit_max_items is not None:
+        resource.add_limit(limit_max_items)
+
+    return resource
+
+
 @dlt.source()
-def opralogwebdb(chunk_size: int = 50000) -> Generator[DltResource]:
+def opralogwebdb(
+    chunk_size: int = 50000, limit_max_items: int | None = None
+) -> Generator[DltResource]:
     """Opralog usage began in 04/2017. We split tables into two categories:
 
       - append-only tables: previous records are never updated, use 'append' write_disposition
@@ -48,6 +59,8 @@ def opralogwebdb(chunk_size: int = 50000) -> Generator[DltResource]:
     were updated. We use the 'LastChangedDate' column of 'Entries' to find the list of
     new or updated EntryId values and load the MoreEntryColumn records for these Entries
     into the destination.
+
+    `chunk_size` and `limit_max_items` are primarily used for testing and debugging
     """
 
     tables_append_records = {
@@ -65,19 +78,20 @@ def opralogwebdb(chunk_size: int = 50000) -> Generator[DltResource]:
             chunk_size=chunk_size,
             **SQL_TABLE_KWARGS,
         )
-        yield resource
+        yield with_resource_limit(resource, limit_max_items)
 
     # Now the Entries table, with incremental cursor, that tells us what EntryIds have been updated
-    yield entries_table(chunk_size)
+    yield with_resource_limit(entries_table(chunk_size), limit_max_items)
 
     # Finally the MoreEntryColumns table based on the loaded EntryIds
-    yield more_entry_columns_table(chunk_size)
+    yield with_resource_limit(more_entry_columns_table(chunk_size), limit_max_items)
 
 
 def entries_table(chunk_size: int) -> DltResource:
     """Return a resource wrapper for the Entries table"""
     resource = sql_table(
         table="Entries",
+        #        query_adapter_callback=entries_query_adapter_callback,
         incremental=dlt.sources.incremental(
             "LastChangedDate",
             initial_value=OPRALOG_EPOCH,
@@ -92,6 +106,39 @@ def entries_table(chunk_size: int) -> DltResource:
     )
 
 
+def entries_query_adapter_callback(
+    query, table, incremental=None, _engine=None
+) -> sa.Select:
+    """Generate a SQLAlchemy text clause compatible with SQL Server datetime columns.
+
+    Ensures timezone-aware datetimes from DLT are cast/converted safely.
+    """
+    if not incremental:
+        return query
+
+    # Handle incremental extraction
+    select_clause = table.select()
+    cursor_col = incremental.cursor_path
+    start_value = incremental.start_value
+
+    # DLT may store UTC-aware datetime, which SQL Server datetime cannot handle
+    if isinstance(start_value, dt.datetime) and start_value.tzinfo is not None:
+        start_value = start_value.replace(tzinfo=None)
+
+    # truncate to milliseconds (3 digits)
+    if isinstance(start_value, dt.datetime):
+        start_value = start_value.replace(
+            microsecond=(start_value.microsecond // 1000) * 1000
+        )
+        formatted = start_value.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+    else:
+        formatted = start_value
+
+    # Construct WHERE clause with an explicit CAST for safety
+    where_clause = sa.text(f"{cursor_col} >= CAST('{formatted}' AS datetime)")
+    return select_clause.where(where_clause)
+
+
 def additional_comment_to_markdown(table: pa.Table) -> pa.Table:
     """Transform the AdditionalComment column to markdown"""
     column_name = "AdditionalComment"
@@ -123,7 +170,7 @@ def store_extracted_entry_ids(table: pa.Table) -> pa.Table:
 def more_entry_columns_table(chunk_size: int) -> DltResource:
     """Return a resource wrapper for the MoreEntryColumns table"""
 
-    def more_entry_columns_query(query: Select, table):
+    def more_entry_columns_query(query: sa.Select, table):
         return query.filter(table.c.EntryId.in_(EXTRACTED_ENTRY_IDS))
 
     resource = sql_table(