From 3562dfa581ea013acb3f1324b2a08798cc23434a Mon Sep 17 00:00:00 2001 From: Paul Trowbridge Date: Tue, 7 Jul 2026 23:57:58 -0400 Subject: [PATCH] merge: quote merge-key identifiers with dest dialect 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 --- pipekit/engine/merge.py | 10 ++++++++-- pipekit/engine/runner.py | 1 + 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/pipekit/engine/merge.py b/pipekit/engine/merge.py index f3644c3..e5f2868 100644 --- a/pipekit/engine/merge.py +++ b/pipekit/engine/merge.py @@ -19,7 +19,13 @@ class MergeError(ValueError): 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": 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 not 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: raise MergeError(f"merge_key is empty after parsing: {merge_key!r}") if len(keys) == 1: diff --git a/pipekit/engine/runner.py b/pipekit/engine/runner.py index 71cefea..ebda5dd 100644 --- a/pipekit/engine/runner.py +++ b/pipekit/engine/runner.py @@ -93,6 +93,7 @@ def run_module(module_id: int, *, group_run_id: int | None = None, dest_table=module["dest_table"], staging_table=module["staging_table"], merge_key=module["merge_key"], + dest_drv=dest_drv, ) repo.log_run_sql(run_id, merge_sql=merge_sql)