Duplicate records with same _dlt_id with dagster-dlt

[niyas-apron]

I'm replicating some tables from postgres to another postgres using Dagster and DLT. I'm using incremental loading and I noticed that I've duplicate records in the destination.

1. I noticed some duplicate records, where the records have same _dlt_load_id and _dlt_id.
2. I also noticed other cases where, only one of the duplicate records has a _dlt_load_id in _dlt_loads table.
3. I had to rerun the dlt loads sometimes due to db connection error failures. But I was expecting dlt to handle connection errors and not load partial data.
4. There are no records in _dlt_loads table with non-zero status even though I had connection errors from source. Since the destination connection wasn't dropped, shouldn't it have records with non-zero status.
5. I have deployed this with Docker Compose. So each run happens within separate containers.
6. I'm using sql_table like below since I want to specify cursor, primary key, merge and column hints per table. In the specific case where I see duplicates, I'm using write_disposition as merge.

import dlt
from dlt.sources.credentials import ConnectionStringCredentials
from dlt.extract import DltResource
from dlt.sources.sql_database import sql_table

def _base_source(schema: str, conn: str) -> tuple[dlt.source, dict[str, str]]:
    """Return individual table sources combined into one source."""
    config = load_config()[schema]    # Load the configuration for the schema

    # Create individual table resources
    table_resources = []
    cursor_by_res: dict[str, str] = {}

    for table_name, t_config in config["tables"].items():
        # Create individual table source
        table_resource = sql_table(
            credentials=ConnectionStringCredentials(conn),
            schema=schema,
            table=table_name,
            backend=t_config.get("backend", "pyarrow"),     # the table with error has sqlalchemy here
            defer_table_reflect=True,
            reflection_level="full_with_precision",
            chunk_size=10000,
        )

        # Apply hints
        table_resource.apply_hints(
            incremental=dlt.sources.incremental(t_config["cursor"]),
            primary_key=t_config.get("pk"),
            write_disposition={
                "disposition": t_config["write_disposition"],   # mostly uses "merge"
                "strategy": t_config.get("strategy", "delete-insert"),
            },
        )

        cursor_by_res[table_resource.name] = t_config["cursor"]
        table_resources.append(table_resource)

    # Combine all table resources into a single source
    @dlt.source(name=f"{schema}_source")
    def combined_source():
        for resource in table_resources:
            yield resource

    source = combined_source()
    return source, cursor_by_res

def slice_hour(
    source: DltResource, key: str, cursor_by_res: dict[str, str]
) -> DltResource:
    """Clone the source and restrict every resource to one-hour window."""
    start = datetime.strptime(key, "%Y-%m-%d-%H:%M")
    end = start + timedelta(hours=1)
    for res in source.selected_resources.values():
        cursor = cursor_by_res[res.name]
        res.apply_hints(
            incremental=dlt.sources.incremental(
                cursor_path=cursor,
                initial_value=start,
                end_value=end,
            )
        )
    return source

7. I'm creating them using Dagster Asset factory. I've set the pipeline name as unique. For example public_2025-07-08-05-00. I was hoping that this will avoid state conflicts within DLT.

def make_core_db_assets():
    assets = []
    dest = postgres(credentials=settings.dwh_postgres_conn_str)

    def create_asset_for_schema(schema, source_factory):
        """Factory function to properly capture variables."""
        base_source, _ = source_factory()
        asset_prefix = "core_prod_"
        dwh_schema_name = f"{asset_prefix}{schema}"
        pipe = pipeline(
            pipeline_name=f"{schema}_base",
            dataset_name=dwh_schema_name,
            destination=dest,
        )

        @dlt_assets(
            dlt_source=base_source,
            dlt_pipeline=pipe,
            partitions_def=hourly,
            name=dwh_schema_name,
            group_name="dwh_pg_prod",
            pool="core_db_sync_pool",
        )
        def _asset(context: AssetExecutionContext):
            context.log.info(
                f"Executing asset for schema: {schema} to DWH schema: {dwh_schema_name}"
            )
            part_key = slugify(context.partition_key)
            partition_pipeline_name = f"{schema}_{part_key}"
            partition_pipe = pipeline(
                pipeline_name=partition_pipeline_name,
                dataset_name=dwh_schema_name,
                destination=dest,
            )
            base_source, cursor_by_res = source_factory()
            hour_source = slice_hour(base_source, context.partition_key, cursor_by_res)

            # Validate that the required DLT resource exists
            dlt_resource_key = dwh_schema_name
            if dlt_resource_key not in resource_map:
                raise ValueError(
                    f"DLT resource '{dlt_resource_key}' not found in resource_map. Available resources: {list(resource_map.keys())}"
                )

            dlt_resource = resource_map[dlt_resource_key]
            context.log.info(f"Using DLT resource: {dlt_resource_key}")

            yield from dlt_resource.run(
                context=context, dlt_source=hour_source, dlt_pipeline=partition_pipe
            )

        return _asset

    for schema, source_factory in iter_sources(settings.core_postgres_conn_str):
        asset_func = create_asset_for_schema(schema, source_factory)
        assets.append(asset_func)

    return assets

Can someone help me understand the root cause for duplication?

[niyas-apron]

I was facing this issue when I used the merge strategy as upsert, but then I moved to delete-insert which solved the duplication problem. However, now it created a new issue where some partition data are missing in the destination despite the logs saying the data is loaded with X number of records and without any errors!