pt/pipekit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: quote merge-key identifiers with dest dialect

Browse Source 
Incremental merge interpolated merge-key columns raw, so a key with a
character illegal in a bare identifier (e.g. dcord#) produced
WHERE dcord# IN (...) and Postgres errored at "IN". build_merge_sql now
takes the dest driver and quotes each key via quote_identifier (clean
names stay bare, the rest wrap in the dialect's quotes).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Paul Trowbridge 2026-07-07 23:57:58 -04:00
parent 0ddb636f14
commit 3562dfa581
2 changed files with 9 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
pipekit/engine/merge.py
View File

@ -19,7 +19,13 @@ class MergeError(ValueError):
def build_merge_sql(*, strategy: str, dest_table: str, staging_table: str, def build_merge_sql(*, strategy: str, dest_table: str, staging_table: str,
merge_key: str | None) -> str: merge_key: str | None, dest_drv=None) -> str:
# Quote merge-key identifiers with the dest dialect. Keys are user-entered
# and may hold characters illegal in a bare identifier (e.g. '#', spaces,
# mixed case) — PG's quote_identifier leaves clean names bare and wraps the
# rest in "…". Falls back to raw when no driver is supplied.
quote = dest_drv.quote_identifier if dest_drv else (lambda s: s)
if strategy == "full": if strategy == "full":
return f"TRUNCATE TABLE {dest_table};\nINSERT INTO {dest_table} SELECT * FROM {staging_table};" return f"TRUNCATE TABLE {dest_table};\nINSERT INTO {dest_table} SELECT * FROM {staging_table};"
@ -29,7 +35,7 @@ def build_merge_sql(*, strategy: str, dest_table: str, staging_table: str,
if strategy == "incremental": if strategy == "incremental":
if not merge_key: if not merge_key:
raise MergeError("incremental merge requires merge_key") raise MergeError("incremental merge requires merge_key")
keys = [k.strip() for k in merge_key.split(",") if k.strip()] keys = [quote(k.strip()) for k in merge_key.split(",") if k.strip()]
if not keys: if not keys:
raise MergeError(f"merge_key is empty after parsing: {merge_key!r}") raise MergeError(f"merge_key is empty after parsing: {merge_key!r}")
if len(keys) == 1: if len(keys) == 1:

1
pipekit/engine/runner.py
View File

@ -93,6 +93,7 @@ def run_module(module_id: int, *, group_run_id: int | None = None,
dest_table=module["dest_table"], dest_table=module["dest_table"],
staging_table=module["staging_table"], staging_table=module["staging_table"],
merge_key=module["merge_key"], merge_key=module["merge_key"],
dest_drv=dest_drv,
) )
repo.log_run_sql(run_id, merge_sql=merge_sql) repo.log_run_sql(run_id, merge_sql=merge_sql)