feat: BI-6156 add dt postprocessing to dashsql (#1465)

* add dashsql postprocessing

* some fixes

* refactor

Author: chl0rum

lib/dl_api_lib/dl_api_lib/app/data_api/resources/dashsql.py

@@ -49,13 +49,22 @@
     IncomingDSQLParamType,
     IncomingDSQLParamTypeExt,
 )
+from dl_query_processing.postprocessing.postprocessors.all import (
+    TYPE_PROCESSORS,
+    postprocess_array,
+)
 from dl_query_processing.utils.datetime import parse_datetime
 
 
 if TYPE_CHECKING:
     from dl_constants.types import TJSONLike
     from dl_core.data_processing.dashsql import TResultEvents
 
+import logging
+
+
+LOGGER = logging.getLogger(__name__)
+
 
 TRowProcessor = Callable[[TRow], TRow]
 
@@ -180,12 +189,37 @@ def _postprocess_any(cls, value: Any) -> Any:
         return value
 
     @classmethod
-    def _make_postprocess_row(cls, bi_type_names: tuple[str, ...]) -> TRowProcessor:
-        # Nothing type-specific at the moment
-        funcs = tuple(cls._postprocess_any for _ in bi_type_names)
+    def _get_type_processor(cls, bi_type_name: str) -> Callable[[Any], Any]:
+        bi_type = UserDataType.__members__.get(bi_type_name)
+
+        if bi_type is None:
+            return lambda val: val
 
-        def _postprocess_row(row: TRow, funcs: tuple[Callable, ...] = funcs) -> TRow:
-            return tuple(func(value) for func, value in zip(funcs, row, strict=True))
+        # Handle datetimetz as datetime (without timezone conversion)
+        if bi_type == UserDataType.datetimetz:
+            bi_type = UserDataType.datetime
+
+        type_processor = TYPE_PROCESSORS.get(bi_type)
+
+        # Return arrays as is (no need to dump), otherwise use the processor
+        if type_processor is None or type_processor == postprocess_array:
+            return lambda val: val
+
+        return type_processor
+
+    @classmethod
+    def _make_postprocess_row(cls, bi_type_names: tuple[str, ...]) -> TRowProcessor:
+        type_processors = tuple(cls._get_type_processor(name) for name in bi_type_names)
+
+        def _postprocess_row(row: TRow) -> TRow:
+            result = []
+            for proc, val in zip(type_processors, row, strict=True):
+                try:
+                    result.append(cls._postprocess_any(proc(val)))
+                except (ValueError, TypeError):
+                    LOGGER.warning("Failed to postprocess value %s (processor: %s). Using raw value.", val, proc)
+                    result.append(cls._postprocess_any(val))
+            return tuple(result)
 
         return _postprocess_row
 

lib/dl_api_lib/dl_api_lib_tests/db/data_api/test_dashsql.py

@@ -1,7 +1,71 @@
+import re
+
+from aiohttp.test_utils import TestClient
+import pytest
+
 from dl_api_lib_testing.connector.dashsql_suite import DefaultDashSQLTestSuite
 from dl_api_lib_tests.db.base import DefaultApiTestBase
 from dl_constants.enums import RawSQLLevel
 
 
 class TestDashSQL(DefaultApiTestBase, DefaultDashSQLTestSuite):
     raw_sql_level = RawSQLLevel.dashsql
+
+    ISO_DATETIME_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?$")
+
+    @pytest.fixture(scope="class")
+    def dashsql_datetime_query(self) -> str:
+        return """select 
+            toDateTime('2023-06-15 14:30:45'), 
+            toDateTime64('2023-06-15 14:30:45.123', 3),
+            toDateTime('2023-06-15 14:30:45', 'Europe/Moscow'),
+            toDateTime64('2023-06-15 14:30:45.123', 3, 'UTC'),
+            toDateTime(1686838245),
+            cast('2023-06-15 14:30:45' AS DateTime),
+            now()
+            """
+
+    @pytest.fixture(scope="class")
+    def dashsql_array_query(self) -> str:
+        return "select [1, 2, 3]"
+
+    @pytest.mark.asyncio
+    async def test_postprocess_datetime(
+        self,
+        data_api_lowlevel_aiohttp_client: TestClient,
+        saved_connection_id: str,
+        dashsql_datetime_query: str,
+        bi_headers: dict[str, str] | None,
+    ) -> None:
+        resp = await self.get_dashsql_response(
+            data_api_aio=data_api_lowlevel_aiohttp_client,
+            conn_id=saved_connection_id,
+            query=dashsql_datetime_query,
+            headers=bi_headers,
+        )
+        data = await resp.json()
+        row_data = data[1]["data"]
+        for datetime_value in row_data:
+            assert self.ISO_DATETIME_PATTERN.match(
+                datetime_value
+            ), f"Value '{datetime_value}' doesn't match ISO 8601 format"
+
+    @pytest.mark.asyncio
+    async def test_postprocess_array(
+        self,
+        data_api_lowlevel_aiohttp_client: TestClient,
+        saved_connection_id: str,
+        dashsql_array_query: str,
+        bi_headers: dict[str, str] | None,
+    ) -> None:
+        """Test that array values are JSON-dumped to strings."""
+        resp = await self.get_dashsql_response(
+            data_api_aio=data_api_lowlevel_aiohttp_client,
+            conn_id=saved_connection_id,
+            query=dashsql_array_query,
+            headers=bi_headers,
+        )
+        data = await resp.json()
+
+        row_data = data[1]["data"]
+        assert row_data[0] == [1, 2, 3], "Array should be returned as list"

lib/dl_connector_clickhouse/dl_connector_clickhouse_tests/db/api/test_dashsql.py

@@ -27,8 +27,8 @@ async def test_result(self, data_api_lowlevel_aiohttp_client: TestClient, saved_
         assert resp_data[0]["data"]["bi_types"] == ["integer", "unsupported", "genericdatetime"]
 
         assert resp_data[0]["data"]["clickhouse_headers"]["X-ClickHouse-Timezone"] == "UTC", resp_data
-        assert resp_data[1] == {"event": "row", "data": [11, [33, 44], "2020-01-02 03:04:16"]}, resp_data
-        assert resp_data[2] == {"event": "row", "data": [22, [33, 44], "2020-01-02 03:04:27"]}, resp_data
+        assert resp_data[1] == {"event": "row", "data": [11, [33, 44], "2020-01-02T03:04:16"]}, resp_data
+        assert resp_data[2] == {"event": "row", "data": [22, [33, 44], "2020-01-02T03:04:27"]}, resp_data
         assert resp_data[-1]["event"] == "footer", resp_data
         assert isinstance(resp_data[-1]["data"], dict)
 

lib/dl_connector_trino/dl_connector_trino_tests/db/api/test_dashsql.py

@@ -135,7 +135,7 @@ async def test_result_with_params(self, data_api_lowlevel_aiohttp_client: TestCl
                 True,
                 False,
                 "2021-07-19",
-                "2021-07-19 19:35:43.000000",
+                "2021-07-19T19:35:43",
                 "11",
                 "22",
                 True,

lib/dl_query_processing/dl_query_processing/postprocessing/postprocessors/all.py

@@ -41,7 +41,7 @@ def postprocess_array(value: Optional[Iterable[Any]]) -> Optional[Iterable[Optio
     return json.dumps(value, ensure_ascii=False)  # Frontend fully supports non-ASCII JSON
 
 
-TYPE_PROCESSORS = {
+TYPE_PROCESSORS: dict[UserDataType, Callable[[Any], Any]] = {
     UserDataType.datetime: postprocess_datetime,
     # parametrized: UserDataType.datetimetz
     UserDataType.genericdatetime: postprocess_genericdatetime,
@@ -62,7 +62,7 @@ def get_type_processor(field_type_info: Optional[DetailedType]) -> Callable[[Any
     # Basic
     result = TYPE_PROCESSORS.get(field_type_info.data_type)
     if result is not None:
-        return result  # type: ignore  # TODO: fix
+        return result
 
     # Parmetrized
     if field_type_info.data_type == UserDataType.datetimetz: