Implement YDB specific concat

test/test_suite.py

@@ -546,6 +546,29 @@ def test_from_as_table(self, connection):
         eq_(connection.execute(sa.select(table)).fetchall(), [(1,), (2,), (3,)])
 
 
+class ConcatTest(fixtures.TablesTest):
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "concat_func_test",
+            metadata,
+            Column("A", String),
+            Column("B", String),
+            sa.PrimaryKeyConstraint("A"),
+            schema=None,
+            test_needs_fk=True,
+        )
+
+    def test_concat_func(self, connection):
+        table = self.tables.concat_func_test
+
+        connection.execute(sa.insert(table).values([{"A": "A", "B": "B"}]))
+
+        stmt = select(func.concat(table.c.A, " ", table.c.B)).limit(1)
+
+        eq_(connection.scalar(stmt), "A B")
+
+
 if not OLD_SA:
     from sqlalchemy.testing.suite.test_types import NativeUUIDTest as _NativeUUIDTest
 

ydb_sqlalchemy/sqlalchemy/compiler/base.py

@@ -307,6 +307,10 @@ def visit_function(self, func, add_to_result_map=None, **kwargs):
             + [name]
         ) % {"expr": self.function_argspec(func, **kwargs)}
 
+    def visit_concat_func(self, func, **kwargs):
+        arg_sql = " || ".join(self.process(arg, **kwargs) for arg in func.clauses)
+        return arg_sql
+
     def _is_bound_to_nullable_column(self, bind_name: str) -> bool:
         if bind_name in self.column_keys and hasattr(self.compile_state, "dml_table"):
             if bind_name in self.compile_state.dml_table.c: