diff --git a/openviking/storage/vectordb/collection/volcengine_api_key_collection.py b/openviking/storage/vectordb/collection/volcengine_api_key_collection.py index 2ca64f0697..89ffbdfe44 100644 --- a/openviking/storage/vectordb/collection/volcengine_api_key_collection.py +++ b/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) diff --git a/openviking/storage/vectordb/collection/volcengine_collection.py b/openviking/storage/vectordb/collection/volcengine_collection.py index 39e2afe606..20196d2bf0 100644 --- a/openviking/storage/vectordb/collection/volcengine_collection.py +++ b/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}") diff --git a/tests/storage/test_volcengine_clients.py b/tests/storage/test_volcengine_clients.py index 4e53b155bf..20d98a0f1e 100644 --- a/tests/storage/test_volcengine_clients.py +++ b/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",