Compile Merge On clause

Author: rad-pat

databend_sqlalchemy/connector.py

@@ -163,7 +163,7 @@ def mogrify(self, query, parameters):
  def execute(self, operation, parameters=None):
  """Prepare and execute a database operation (query or command)."""
 
- # ToDo - Fix this, which is preventing the execution of blank DDL sunch as CREATE INDEX statements which aren't currently supported
+ # ToDo - Fix this, which is preventing the execution of blank DDL such as CREATE INDEX statements which aren't currently supported
  # Seems hard to fix when statements are coming from metadata.create_all()
  if not operation:
  return

databend_sqlalchemy/databend_dialect.py

@@ -906,12 +906,16 @@ def visit_merge(self, merge, **kw):
  )
  elif isinstance(merge.source, Subquery):
  source = merge.source._compiler_dispatch(self, **source_kw)
+ else:
+ source = merge.source
+
+ merge_on = merge.on._compiler_dispatch(self, **kw)
 
  target_table = self.preparer.format_table(merge.target)
  return (
  f"MERGE INTO {target_table}\n"
  f" USING {source}\n"
- f" ON {merge.on}\n"
+ f" ON {merge_on}\n"
  f"{clauses if clauses else ''}"
  )
 

databend_sqlalchemy/dml.py

@@ -76,6 +76,8 @@ class Merge(UpdateBase):
  __visit_name__ = "merge"
  _bind = None
 
+ inherit_cache = False
+
  def __init__(self, target, source, on):
  if not isinstance(source, (TableClause, Select, Subquery)):
  raise Exception(f"Invalid type for merge source: {source}")

tests/test_merge.py

@@ -11,15 +11,15 @@
 from sqlalchemy import types as sqltypes
 
 # from sqlalchemy.dialects.postgresql import insert
-from sqlalchemy.testing import config
+from sqlalchemy.testing import config, AssertsCompiledSQL
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing.assertions import assert_raises
 from sqlalchemy.testing.assertions import eq_
 
 from databend_sqlalchemy import Merge
 
 
-class MergeIntoTest(fixtures.TablesTest):
+class MergeIntoTest(fixtures.TablesTest, AssertsCompiledSQL):
  __backend__ = True
  run_define_tables = "each"
 
@@ -170,6 +170,11 @@ def test_when_not_matched_insert(self, connection):
  merge = Merge(users, users_xtra, users.c.id == users_xtra.c.id)
  merge.when_not_matched_then_insert()
 
+ self.assert_compile(
+ merge,
+ 'MERGE INTO "users" USING (SELECT users_xtra.id AS id, users_xtra.name AS name, users_xtra.login_email AS login_email FROM users_xtra) AS users_xtra ON "users".id = users_xtra.idWHEN NOT MATCHED THENINSERT *',
+ )
+
  result = connection.execute(merge)
  eq_(
  connection.execute(users.select().order_by(users.c.id)).fetchall(),