sqalchemy lightweight delete and statement compilation support (#551)

joe-clickhouse · web-flow · commit 629858ae20d9 · 2025-09-16T10:57:35.000-07:00
* sqalchemy lightweight delete support

* linter fix

* explicitly order select

* add sqa support for select and joins

* add explicit delete test

* deal with cloud test latency

* change engine type

* simplify join
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -22,6 +22,7 @@ instead of being passed as ClickHouse server settings. This is in conjunction wi
 The supported method of passing ClickHouse server settings is to prefix such arguments/query parameters with`ch_`.
 
 ## UNRELEASED
+- Add support for lightweight `DELETE` in sqlalchemy. Closes [#382](https://github.com/ClickHouse/clickhouse-connect/issues/382)
 - Added client connection option `rename_response_column` (default `None`) that allows the user to define how response columns are automatically renamed according to a predefined scheme. Helpful for stripping alias prefixes, etc. in potentially complex queries. Closes [#228](https://github.com/ClickHouse/clickhouse-connect/issues/228)
 - Add third-party library identifiers (name/version) in the User-Agent, e.g. pandas/2.2.5. Users can opt out by changing the common setting `send_integration_tags` to `False`.
 - Added support for form encoding query parameters when using HTTP interface. This addresses [#342](https://github.com/ClickHouse/clickhouse-connect/issues/342). Query parameters can now be sent as form-encoded data in the request body by setting `form_encode_query_params=True` when creating the client. This is particularly useful for queries with large parameter payloads that might exceed URL length limits.
diff --git a/clickhouse_connect/cc_sqlalchemy/dialect.py b/clickhouse_connect/cc_sqlalchemy/dialect.py
@@ -6,6 +6,7 @@
 from clickhouse_connect.cc_sqlalchemy.inspector import ChInspector
 from clickhouse_connect.cc_sqlalchemy.sql import full_table
 from clickhouse_connect.cc_sqlalchemy.sql.ddlcompiler import ChDDLCompiler
+from clickhouse_connect.cc_sqlalchemy.sql.compiler import ChStatementCompiler
 from clickhouse_connect.cc_sqlalchemy import ischema_names, dialect_name
 from clickhouse_connect.cc_sqlalchemy.sql.preparer import ChIdentifierPreparer
 from clickhouse_connect.driver.binding import quote_identifier, format_str
@@ -26,6 +27,7 @@ class ClickHouseDialect(DefaultDialect):
     returns_unicode_strings = True
     postfetch_lastrowid = False
     ddl_compiler = ChDDLCompiler
+    statement_compiler = ChStatementCompiler
     preparer = ChIdentifierPreparer
     description_encoding = None
     max_identifier_length = 127
diff --git a/clickhouse_connect/cc_sqlalchemy/sql/compiler.py b/clickhouse_connect/cc_sqlalchemy/sql/compiler.py
@@ -0,0 +1,74 @@
+from sqlalchemy.exc import CompileError
+from sqlalchemy.sql.compiler import SQLCompiler
+
+from clickhouse_connect.cc_sqlalchemy.sql import format_table
+
+
+# pylint: disable=arguments-differ
+class ChStatementCompiler(SQLCompiler):
+
+    # pylint: disable=attribute-defined-outside-init
+    def visit_delete(self, delete_stmt, visiting_cte=None, **kw):
+        table = delete_stmt.table
+        text = f"DELETE FROM {format_table(table)}"
+
+        if delete_stmt.whereclause is not None:
+            self._in_delete_where = True
+            try:
+                text += " WHERE " + self.process(delete_stmt.whereclause, **kw)
+            finally:
+                self._in_delete_where = False
+        else:
+            raise CompileError("ClickHouse DELETE statements require a WHERE clause. To delete all rows, use 'TRUNCATE TABLE' instead.")
+
+        return text
+
+    def visit_select(self, select_stmt, **kw):
+        return super().visit_select(select_stmt, **kw)
+
+    def visit_join(self, join, **kw):
+        left = self.process(join.left, **kw)
+        right = self.process(join.right, **kw)
+        onclause = join.onclause
+
+        if getattr(join, "full", False):
+            join_kw = " FULL OUTER JOIN "
+        elif onclause is None:
+            join_kw = " CROSS JOIN "
+        elif join.isouter:
+            join_kw = " LEFT OUTER JOIN "
+        else:
+            join_kw = " INNER JOIN "
+
+        text = left + join_kw + right
+
+        if onclause is not None:
+            text += " ON " + self.process(onclause, **kw)
+
+        return text
+
+    def visit_column(self, column, add_to_result_map=None, include_table=True, result_map_targets=(), ambiguous_table_name_map=None, **kw):
+        if getattr(self, "_in_delete_where", False):
+            return self.preparer.quote(column.name)
+
+        return super().visit_column(
+            column,
+            add_to_result_map=add_to_result_map,
+            include_table=include_table,
+            result_map_targets=result_map_targets,
+            **kw,
+        )
+
+    # Abstract methods required by SQLCompiler
+    def delete_extra_from_clause(self, delete_stmt, from_table, extra_froms, from_hints, **kw):
+        raise NotImplementedError("ClickHouse doesn't support DELETE with extra FROM clause")
+
+    def update_from_clause(self, update_stmt, from_table, extra_froms, from_hints, **kw):
+        raise NotImplementedError("ClickHouse doesn't support UPDATE with FROM clause")
+
+    # pylint: disable=unused-argument
+    def visit_empty_set_expr(self, element_types, **kw):
+        return "SELECT 1 WHERE 1=0"
+
+    def visit_sequence(self, sequence, **kw):
+        raise NotImplementedError("ClickHouse doesn't support sequences")
diff --git a/tests/integration_tests/test_sqlalchemy/test_delete.py b/tests/integration_tests/test_sqlalchemy/test_delete.py
@@ -0,0 +1,190 @@
+import pytest
+import sqlalchemy as db
+from sqlalchemy import MetaData, delete, text
+from sqlalchemy.engine import Engine
+from sqlalchemy.exc import CompileError
+
+from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import String, UInt64
+from clickhouse_connect.cc_sqlalchemy.ddl.tableengine import engine_map
+
+
+def test_delete_with_table_object(test_engine: Engine, test_db: str, test_table_engine: str):
+    """DELETE using SQLAlchemy Table object"""
+    engine_cls = engine_map[test_table_engine]
+    metadata = MetaData(schema=test_db)
+
+    test_table = db.Table(
+        "delete_test",
+        metadata,
+        db.Column("id", UInt64),
+        db.Column("name", String),
+        db.Column("status", String),
+        engine_cls("id"),
+    )
+
+    with test_engine.begin() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS delete_test"))
+        test_table.create(conn)
+
+        conn.execute(db.insert(test_table).values({"id": 1, "name": "hello world", "status": "active"}))
+        conn.execute(db.insert(test_table).values({"id": 2, "name": "test data", "status": "inactive"}))
+        conn.execute(db.insert(test_table).values({"id": 3, "name": "hello test", "status": "active"}))
+        starting = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(starting) == 3
+        assert [row.id for row in starting] == [1, 2, 3]
+
+        delete_stmt = delete(test_table).where(test_table.c.name.like("%hello%"))
+        conn.execute(delete_stmt)
+
+        remaining = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(remaining) == 1
+        assert remaining[0].name == "test data"
+
+
+def test_delete_with_multiple_conditions(test_engine: Engine, test_db: str, test_table_engine: str):
+    """DELETE with multiple WHERE conditions"""
+    engine_cls = engine_map[test_table_engine]
+    metadata = MetaData(schema=test_db)
+
+    test_table = db.Table(
+        "delete_multi_test",
+        metadata,
+        db.Column("id", UInt64),
+        db.Column("category", String),
+        db.Column("value", UInt64),
+        engine_cls("id"),
+    )
+
+    with test_engine.begin() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS delete_multi_test"))
+        test_table.create(conn)
+
+        conn.execute(db.insert(test_table).values({"id": 1, "category": "A", "value": 100}))
+        conn.execute(db.insert(test_table).values({"id": 2, "category": "B", "value": 200}))
+        conn.execute(db.insert(test_table).values({"id": 3, "category": "A", "value": 300}))
+        conn.execute(db.insert(test_table).values({"id": 4, "category": "C", "value": 50}))
+        starting = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(starting) == 4
+        assert [row.id for row in starting] == [1, 2, 3, 4]
+
+        delete_stmt = delete(test_table).where((test_table.c.category == "A") & (test_table.c.value > 150))
+        conn.execute(delete_stmt)
+
+        remaining = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(remaining) == 3
+        assert [row.id for row in remaining] == [1, 2, 4]
+
+
+def test_delete_all_rows_error(test_engine: Engine, test_db: str, test_table_engine: str):
+    """DELETE without WHERE should raise"""
+    engine_cls = engine_map[test_table_engine]
+    metadata = MetaData(schema=test_db)
+
+    test_table = db.Table(
+        "delete_all_test",
+        metadata,
+        db.Column("id", UInt64),
+        db.Column("data", String),
+        engine_cls("id"),
+    )
+
+    with test_engine.begin() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS delete_all_test"))
+        test_table.create(conn)
+        conn.execute(db.insert(test_table).values({"id": 1, "data": "test1"}))
+
+        delete_stmt = delete(test_table)
+        with pytest.raises(CompileError, match="require a WHERE clause"):
+            conn.execute(delete_stmt)
+
+
+def test_delete_basic_functionality(test_engine: Engine, test_table_engine: str):
+    """Basic DELETE statement compilation and execution"""
+    engine_cls = engine_map[test_table_engine]
+    metadata = MetaData()
+
+    test_table = db.Table(
+        "delete_simple_test",
+        metadata,
+        db.Column("id", UInt64),
+        db.Column("name", String),
+        engine_cls("id"),
+    )
+
+    with test_engine.begin() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS delete_simple_test"))
+        test_table.create(conn)
+
+        conn.execute(db.insert(test_table).values({"id": 1, "name": "test_row"}))
+        starting = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(starting) == 1
+        assert starting[0].name == "test_row"
+
+        delete_stmt = delete(test_table).where(test_table.c.id == 1)
+        conn.execute(delete_stmt)
+
+        result = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(result) == 0
+
+
+def test_delete_with_text_condition(test_engine: Engine, test_table_engine: str):
+    """Test DELETE with text-based WHERE condition"""
+    engine_cls = engine_map[test_table_engine]
+    metadata = MetaData()
+
+    test_table = db.Table(
+        "delete_text_test",
+        metadata,
+        db.Column("id", UInt64),
+        db.Column("status", String),
+        engine_cls("id"),
+    )
+
+    with test_engine.begin() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS delete_text_test"))
+        test_table.create(conn)
+
+        conn.execute(db.insert(test_table).values({"id": 1, "status": "active"}))
+        conn.execute(db.insert(test_table).values({"id": 2, "status": "inactive"}))
+        conn.execute(db.insert(test_table).values({"id": 3, "status": "active"}))
+        starting = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(starting) == 3
+        assert [row.id for row in starting] == [1, 2, 3]
+
+        delete_stmt = delete(test_table).where(test_table.c.status == "inactive")
+        result = conn.execute(delete_stmt)
+
+        result = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(result) == 2
+        assert [row.id for row in result] == [1, 3]
+
+
+def test_explicit_delete(test_engine: Engine, test_table_engine: str):
+    """Test explicit DELETE"""
+    engine_cls = engine_map[test_table_engine]
+    metadata = MetaData()
+
+    test_table = db.Table(
+        "delete_explicit_test",
+        metadata,
+        db.Column("id", UInt64),
+        db.Column("name", String),
+        engine_cls("id"),
+    )
+
+    with test_engine.begin() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS delete_explicit_test"))
+        test_table.create(conn)
+
+        conn.execute(db.insert(test_table).values({"id": 1, "name": "hello world"}))
+        conn.execute(db.insert(test_table).values({"id": 2, "name": "test data"}))
+        conn.execute(db.insert(test_table).values({"id": 3, "name": "hello test"}))
+        starting = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(starting) == 3
+        assert [row.id for row in starting] == [1, 2, 3]
+
+        conn.execute(text("DELETE FROM delete_explicit_test WHERE name LIKE '%hello%'"))
+
+        result = conn.execute(db.select(test_table).order_by(test_table.c.id)).fetchall()
+        assert len(result) == 1
+        assert result[0].name == "test data"
diff --git a/tests/integration_tests/test_sqlalchemy/test_select.py b/tests/integration_tests/test_sqlalchemy/test_select.py
