Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Deltalake dedup_sort support #2202

Open
wants to merge 2 commits into
base: devel
Choose a base branch
from
Open

Deltalake dedup_sort support #2202

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
daae03f
Deltalake dedup_sort support
guitcastro Jan 9, 2025
f613ac6
fix fmt deltalake.py
guitcastro Jan 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions dlt/common/libs/deltalake.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
from dlt.common.libs.pyarrow import pyarrow as pa
from dlt.common.libs.pyarrow import cast_arrow_schema_types
from dlt.common.schema.typing import TWriteDisposition, TTableSchema
from dlt.common.schema.utils import get_first_column_name_with_prop, get_columns_names_with_prop
from dlt.common.schema.utils import get_first_column_name_with_prop, get_columns_names_with_prop, get_dedup_sort_tuple
from dlt.common.exceptions import MissingDependencyException
from dlt.common.storages import FilesystemConfiguration
from dlt.common.utils import assert_min_pkg_version
Expand Down Expand Up @@ -131,7 +131,13 @@ def merge_delta_table(
else:
primary_keys = get_columns_names_with_prop(schema, "primary_key")
predicate = " AND ".join([f"target.{c} = source.{c}" for c in primary_keys])

dedup_tuple = get_dedup_sort_tuple(schema)
if dedup_tuple:
dedup_column, dedup_sort = dedup_tuple
dedup_operator = ">" if dedup_sort == "desc" else "<"
dedup_query = f"target.{dedup_column} {dedup_operator} source.{dedup_column}"
else:
dedup_query = None
partition_by = get_columns_names_with_prop(schema, "partition")
qry = (
table.merge(
Expand All @@ -140,7 +146,7 @@ def merge_delta_table(
source_alias="source",
target_alias="target",
)
.when_matched_update_all()
.when_matched_update_all(dedup_query)
.when_not_matched_insert_all()
)

Expand Down
Loading