Finetuning YDB adapter

lib/dl_connector_ydb/dl_connector_ydb/core/base/adapter.py

@@ -36,10 +36,20 @@
 _DBA_YQL_BASE_DTO_TV = TypeVar("_DBA_YQL_BASE_DTO_TV", bound="BaseSQLConnTargetDTO")
 
 
+DEFAULT_YDB_REQUEST_TIMEOUT_SEC = 30
+
+
 @attr.s
 class YQLAdapterBase(BaseClassicAdapter[_DBA_YQL_BASE_DTO_TV]):
+    use_literal_binds_if_possible = True
+
     execution_options = {
         "ydb_retry_settings": ydb.RetrySettings(retry_cancelled=True),
+        "ydb_request_settings": (
+            ydb.BaseRequestSettings()
+            .with_timeout(DEFAULT_YDB_REQUEST_TIMEOUT_SEC)
+            .with_cancel_after(DEFAULT_YDB_REQUEST_TIMEOUT_SEC)
+        ),
     }
 
     def _get_db_version(self, db_ident: DBIdent) -> Optional[str]:

lib/dl_core/dl_core/connection_executors/adapters/adapters_base_sa.py

@@ -36,6 +36,7 @@
     CursorLogger,
     compile_query_for_debug,
     compile_query_for_inspector,
+    compile_query_with_literal_binds_if_possible,
     get_db_version_query,
 )
 from dl_core.connection_executors.models.db_adapter_data import (
@@ -92,6 +93,7 @@ class BaseSAAdapter(
     SAColumnTypeNormalizer,
 ):
     allow_sa_text_as_columns_source: ClassVar[bool] = False
+    use_literal_binds_if_possible: ClassVar[bool] = False
 
     # Instance attributes
     _default_chunk_size: int = attr.ib()
@@ -166,6 +168,9 @@ def execute_by_steps(self, db_adapter_query: DBAdapterQuery) -> Generator[Execut
         debug_query = compile_query_for_debug(query, engine.dialect)
         inspector_query = compile_query_for_inspector(query, engine.dialect)
 
+        if self.use_literal_binds_if_possible:
+            query = compile_query_with_literal_binds_if_possible(query, engine.dialect)
+
         with (
             db_session_context(backend_type=self.get_backend_type(), db_engine=engine) as db_session,
             self.handle_execution_error(debug_query=debug_query, inspector_query=inspector_query),

lib/dl_core/dl_core/connection_executors/adapters/sa_utils.py

@@ -55,6 +55,15 @@ def compile_query_for_inspector(query: ClauseElement | str, dialect: Dialect) ->
         return "-"
 
 
+def compile_query_with_literal_binds_if_possible(query: ClauseElement | str, dialect: Dialect) -> str:
+    if isinstance(query, str):
+        return query
+    try:
+        return str(query.compile(dialect=dialect, compile_kwargs={"literal_binds": True}))
+    except NotImplementedError:
+        return str(query.compile(dialect=dialect))
+
+
 def make_debug_query(query: str, params: Union[list, dict]) -> str:
     return f"{query} {params!r}"