fix(vikingdb): normalize all date_time range filters in API key client

openviking/storage/vectordb/collection/volcengine_api_key_collection.py

@@ -19,6 +19,7 @@
     ClientForDataApi,
     ClientForDataApiWithApiKey,
 )
+from openviking.utils.search_filters import VALID_TIME_FIELDS
 from openviking_cli.utils.logger import default_logger as logger
 
 
@@ -150,8 +151,28 @@ def _sanitize_list_payload(cls, obj: List[Any]) -> List[Any]:
                 sanitized_list.append(y)
         return sanitized_list
 
+    @classmethod
+    def _normalize_date_time_filter(cls, obj: Any) -> Any:
+        """Rewrite ``range`` nodes on date_time fields to VikingDB ``time_range``.
+
+        OpenViking compiles ``TimeRange`` down to the internal ``range`` DSL, but the
+        commercial data-plane expects ``time_range`` for date_time fields and ``range``
+        only for numeric fields. Numeric ``range`` nodes are left untouched.
+        """
+        if isinstance(obj, list):
+            return [cls._normalize_date_time_filter(item) for item in obj]
+        if not isinstance(obj, dict):
+            return obj
+
+        normalized = {key: cls._normalize_date_time_filter(value) for key, value in obj.items()}
+        if normalized.get("op") == "range" and normalized.get("field") in VALID_TIME_FIELDS:
+            normalized["op"] = "time_range"
+        return normalized
+
     def _data_post(self, path: str, data: Dict[str, Any]):
         safe_data = self._sanitize_payload(data)
+        if isinstance(safe_data, dict) and "filter" in safe_data:
+            safe_data["filter"] = self._normalize_date_time_filter(safe_data["filter"])
         response = self.data_client.do_req("POST", path, req_body=safe_data)
         if response.status_code != 200:
             raise self._build_response_error(response, path)

openviking/storage/vectordb/collection/volcengine_collection.py

@@ -18,6 +18,7 @@
     ClientForConsoleApi,
     ClientForDataApi,
 )
+from openviking.utils.search_filters import VALID_TIME_FIELDS
 from openviking_cli.utils.logger import default_logger as logger
 
 
@@ -255,9 +256,29 @@ def _sanitize_list_payload(cls, obj: List[Any]) -> List[Any]:
                 sanitized_list.append(y)
         return sanitized_list
 
+    @classmethod
+    def _normalize_date_time_filter(cls, obj: Any) -> Any:
+        """Rewrite ``range`` nodes on date_time fields to VikingDB ``time_range``.
+
+        OpenViking compiles ``TimeRange`` down to the internal ``range`` DSL, but the
+        commercial data-plane expects ``time_range`` for date_time fields and ``range``
+        only for numeric fields. Numeric ``range`` nodes are left untouched.
+        """
+        if isinstance(obj, list):
+            return [cls._normalize_date_time_filter(item) for item in obj]
+        if not isinstance(obj, dict):
+            return obj
+
+        normalized = {key: cls._normalize_date_time_filter(value) for key, value in obj.items()}
+        if normalized.get("op") == "range" and normalized.get("field") in VALID_TIME_FIELDS:
+            normalized["op"] = "time_range"
+        return normalized
+
     def _data_post(self, path: str, data: Dict[str, Any]):
         # Centralized sanitization at the request exit, covering all data API inputs
         safe_data = self._sanitize_payload(data)
+        if isinstance(safe_data, dict) and "filter" in safe_data:
+            safe_data["filter"] = self._normalize_date_time_filter(safe_data["filter"])
         response = self.data_client.do_req("POST", path, req_body=safe_data)
         if response.status_code != 200:
             logger.error(f"Request to {path} failed: {response.text}")

tests/storage/test_volcengine_clients.py

@@ -293,6 +293,115 @@ def _fake_do_req(method, path=None, req_params=None, req_body=None):
     }
 
 
+def test_volcengine_collection_uses_date_time_filter_operator(monkeypatch):
+    captured = {}
+
+    class _Response:
+        status_code = 200
+
+        @staticmethod
+        def json():
+            return {"result": {"agg": {"_total": 1}}}
+
+    collection = VolcengineCollection(
+        ak="test-ak",
+        sk="test-sk",
+        region="cn-beijing",
+        meta_data={"ProjectName": "default", "CollectionName": "context"},
+    )
+
+    def _fake_do_req(method, path=None, req_params=None, req_body=None):
+        captured["path"] = path
+        captured["req_body"] = req_body
+        return _Response()
+
+    monkeypatch.setattr(collection.data_client, "do_req", _fake_do_req)
+
+    # Both date_time fields (created_at, updated_at) must be normalized to
+    # time_range, while numeric range nodes are left untouched.
+    collection.aggregate_data(
+        index_name="default",
+        filters={
+            "op": "and",
+            "conds": [
+                {
+                    "op": "range",
+                    "field": "created_at",
+                    "gte": "2026-08-10T00:00:00+00:00",
+                    "lt": "2026-08-11T00:00:00+00:00",
+                },
+                {
+                    "op": "range",
+                    "field": "updated_at",
+                    "gte": "2026-08-10T00:00:00+00:00",
+                },
+                {"op": "range", "field": "level", "gte": 1},
+            ],
+        },
+    )
+
+    assert captured["path"] == "/api/vikingdb/data/agg"
+    assert captured["req_body"]["filter"] == {
+        "op": "and",
+        "conds": [
+            {
+                "op": "time_range",
+                "field": "created_at",
+                "gte": "2026-08-10T00:00:00+00:00",
+                "lt": "2026-08-11T00:00:00+00:00",
+            },
+            {
+                "op": "time_range",
+                "field": "updated_at",
+                "gte": "2026-08-10T00:00:00+00:00",
+            },
+            {"op": "range", "field": "level", "gte": 1},
+        ],
+    }
+
+
+def test_volcengine_collection_date_time_filter_is_idempotent(monkeypatch):
+    captured = {}
+
+    class _Response:
+        status_code = 200
+
+        @staticmethod
+        def json():
+            return {"result": {"data": []}}
+
+    collection = VolcengineCollection(
+        ak="test-ak",
+        sk="test-sk",
+        region="cn-beijing",
+        meta_data={"ProjectName": "default", "CollectionName": "context"},
+    )
+
+    def _fake_do_req(method, path=None, req_params=None, req_body=None):
+        captured["req_body"] = req_body
+        return _Response()
+
+    monkeypatch.setattr(collection.data_client, "do_req", _fake_do_req)
+
+    # Filters already emitted as time_range (e.g. from merge_time_filter) must
+    # pass through unchanged.
+    collection.search_by_scalar(
+        index_name="default",
+        field="created_at",
+        filters={
+            "op": "time_range",
+            "field": "updated_at",
+            "gte": "2026-08-10T00:00:00+00:00",
+        },
+    )
+
+    assert captured["req_body"]["filter"] == {
+        "op": "time_range",
+        "field": "updated_at",
+        "gte": "2026-08-10T00:00:00+00:00",
+    }
+
+
 def test_volcengine_api_key_collection_update_data_posts_to_update_endpoint(monkeypatch):
     captured = {}
 
@@ -373,6 +482,121 @@ def _fake_do_req(method, req_path=None, req_params=None, req_body=None):
     }
 
 
+def test_volcengine_api_key_collection_uses_date_time_filter_operator(monkeypatch):
+    captured = {}
+
+    class _Response:
+        status_code = 200
+
+        @staticmethod
+        def json():
+            return {"result": {"agg": {"_total": 1}}}
+
+    from openviking.storage.vectordb.collection.volcengine_api_key_collection import (
+        VolcengineApiKeyCollection,
+    )
+
+    collection = VolcengineApiKeyCollection(
+        api_key="vk-test-token",
+        region="cn-beijing",
+        meta_data={"ProjectName": "default", "CollectionName": "context", "IndexName": "default"},
+    )
+
+    def _fake_do_req(method, req_path=None, req_params=None, req_body=None):
+        captured["path"] = req_path
+        captured["req_body"] = req_body
+        return _Response()
+
+    monkeypatch.setattr(collection.data_client, "do_req", _fake_do_req)
+
+    # Both date_time fields (created_at, updated_at) must be normalized to
+    # time_range, while numeric range nodes are left untouched.
+    collection.aggregate_data(
+        index_name="default",
+        filters={
+            "op": "and",
+            "conds": [
+                {
+                    "op": "range",
+                    "field": "created_at",
+                    "gte": "2026-08-10T00:00:00+00:00",
+                    "lt": "2026-08-11T00:00:00+00:00",
+                },
+                {
+                    "op": "range",
+                    "field": "updated_at",
+                    "gte": "2026-08-10T00:00:00+00:00",
+                },
+                {"op": "range", "field": "level", "gte": 1},
+            ],
+        },
+    )
+
+    assert captured["path"] == "/api/vikingdb/data/agg"
+    assert captured["req_body"]["filter"] == {
+        "op": "and",
+        "conds": [
+            {
+                "op": "time_range",
+                "field": "created_at",
+                "gte": "2026-08-10T00:00:00+00:00",
+                "lt": "2026-08-11T00:00:00+00:00",
+            },
+            {
+                "op": "time_range",
+                "field": "updated_at",
+                "gte": "2026-08-10T00:00:00+00:00",
+            },
+            {"op": "range", "field": "level", "gte": 1},
+        ],
+    }
+
+
+def test_volcengine_api_key_collection_date_time_filter_is_idempotent(monkeypatch):
+    captured = {}
+
+    class _Response:
+        status_code = 200
+
+        @staticmethod
+        def json():
+            return {"result": {"data": []}}
+
+    from openviking.storage.vectordb.collection.volcengine_api_key_collection import (
+        VolcengineApiKeyCollection,
+    )
+
+    collection = VolcengineApiKeyCollection(
+        api_key="vk-test-token",
+        region="cn-beijing",
+        meta_data={"ProjectName": "default", "CollectionName": "context", "IndexName": "default"},
+    )
+
+    def _fake_do_req(method, req_path=None, req_params=None, req_body=None):
+        captured["req_body"] = req_body
+        return _Response()
+
+    monkeypatch.setattr(collection.data_client, "do_req", _fake_do_req)
+
+    # Filters already emitted as time_range (e.g. from merge_time_filter) must
+    # pass through unchanged.
+    collection.search_by_scalar(
+        index_name="default",
+        field="created_at",
+        filters={
+            "op": "time_range",
+            "field": "updated_at",
+            "gte": "2026-08-10T00:00:00+00:00",
+        },
+    )
+
+    assert captured["req_body"]["filter"] == {
+        "op": "time_range",
+        "field": "updated_at",
+        "gte": "2026-08-10T00:00:00+00:00",
+    }
+
+
 def test_volcengine_adapter_update_data_returns_ids():
     adapter = VolcengineCollectionAdapter(
         ak="test-ak",