diff --git a/conf/config.yaml b/conf/config.yaml
index fa78f1d79..be716c5ec 100644
--- a/conf/config.yaml
+++ b/conf/config.yaml
@@ -55,6 +55,7 @@ vectordb:
collection_name: vdb_test
hybrid_search: true
enable: true
+ schema_version: 1 # Increment when the collection schema changes and a migration is required
# --- Relational Database (PostgreSQL) ---
# Env: POSTGRES_HOST, POSTGRES_PORT, POSTGRES_USER, POSTGRES_PASSWORD, DEFAULT_FILE_QUOTA
diff --git a/docs/content/docs/documentation/API.mdx b/docs/content/docs/documentation/API.mdx
index c915b1ad9..dc2407f55 100644
--- a/docs/content/docs/documentation/API.mdx
+++ b/docs/content/docs/documentation/API.mdx
@@ -90,6 +90,18 @@ Upload a new file to a specific partition for indexing.
- `201 Created`: Returns task status URL
- `409 Conflict`: File already exists in partition
+##### Temporal Filtering
+OpenRAG supports temporal filtering to retrieve documents from specific time periods.
+The client can include the temporal field to allow temporal-aware search in search endpoints.
+
+* `created_at`: ISO 8601 format date of when the file was created
+
+:::info
+`created_at` is provided by the client in the metadata of the file during upload.
+This is a first iteration — additional temporal fields (e.g. `updated_at`) may be added in future releases as needed.
+:::
+
+
##### Upload files while modeling relations between them
OpenRAG supports document relationships to enable context-aware retrieval.
diff --git a/docs/content/docs/documentation/milvus_migration.md b/docs/content/docs/documentation/milvus_migration.md
deleted file mode 100644
index 10dbd5c2d..000000000
--- a/docs/content/docs/documentation/milvus_migration.md
+++ /dev/null
@@ -1,126 +0,0 @@
----
-title: Milvus Migrations
----
-
-# Milvus Upgrade
-OpenRAG has been upgraded from Milvus **2.5.4** to **2.6.11** to leverage the enhancements introduced in the latest releases, particularly the new temporal querying capabilities added in version **2.6.6+**.
-
-## What's New in 2.6.x
-
-Milvus 2.6.6+ introduced the **`TIMESTAMPTZ`** field type, which enables:
-
-- **Comparison and range filtering** using standard operators (`=`, `!=`, `<`, `>`, etc.)
-- **Interval arithmetic** — add or subtract durations (days, hours, minutes) directly in filter expressions
-- **Time-based indexing** for faster temporal queries
-- **Combined filtering** — pair timestamp conditions with vector similarity search
-
-**Example — basic comparison:**
-```python
-expr = "tsz != ISO '2025-01-03T00:00:00+08:00'"
-results = client.query(
- collection_name,
- filter=expr,
- output_fields=["id", "tsz"],
- limit=10
-)
-```
-
-**Example — interval arithmetic:**
-```python
-expr = "tsz + INTERVAL 'P1D' > ISO '2025-01-03T00:00:00+08:00'"
-results = client.query(
- collection_name,
- filter=expr,
- output_fields=["id", "tsz"],
- limit=10
-)
-```
-
-> `INTERVAL` values follow [ISO 8601 duration](https://en.wikipedia.org/wiki/ISO_8601#Durations) syntax:
-> * `P1D` = 1 day
-> * `PT3H` = 3 hours
-> * `P2DT6H` = 2 days and 6 hours.
-
-## Current State
-
-:::info
-Temporal fields are currently stored as **strings**, not **`TIMESTAMPTZ`**. Migrating to `TIMESTAMPTZ` requires a schema and index change, and Milvus doesn't support migrations on schema and index changes: it has to be handled manually.
-
-Until a Milvus schema & index migration strategy is defined, filtering still works via **lexicographic string comparison** on ISO 8601 strings:
-```python
-expr = "tsz != '2025-01-03T00:00:00+08:00'" # No ISO/INTERVAL keywords
-results = client.query(
- collection_name,
- filter=expr,
- output_fields=["id", "tsz"],
- limit=10
-)
-```
-Full `TIMESTAMPTZ` support will be activated in a future release once the migration is established.
-:::
-
-## Milvus Version Upgrade Steps
-
-:::danger[Who needs this migration?]
-This migration is only required if you are upgrading from **OpenRAG <= 1.1.7**, which shipped with Milvus <= 2.5.x. If your deployment already runs Milvus 2.6.x, skip this section.
-:::
-
-> For the full official reference, see the [Milvus upgrade guide](https://milvus.io/docs/upgrade_milvus_standalone-docker.md#Upgrade-process).
-
-### Step 1 — Upgrade Milvus to 2.5.16 (intermediate step)
-
-:::caution[Do not update OpenRAG yet]
-During this step, keep your current version of OpenRAG (<= 1.1.7) running. Only the Milvus image is changed here. OpenRAG itself is updated in Step 2.
-:::
-
-Milvus requires an intermediate upgrade to **v2.5.16** before jumping to 2.6.x. This step must be done manually **before** updating OpenRAG.
-
-Temporarily edit `vdb/milvus.yaml` to set the intermediate Milvus image:
-
-```diff lang=yaml
-// vdb/milvus.yaml
-milvus:
-- image: milvusdb/milvus:v2.5.4
-+ image: milvusdb/milvus:v2.5.16
-```
-
-Then restart Milvus and wait for it to be healthy:
-
-```bash
-docker compose down
-docker compose up milvus -d
-```
-
-Verify it is running and healthy before continuing:
-
-```bash
-docker inspect milvus-standalone --format '{{ .Config.Image }}'
-# Expected: milvusdb/milvus:v2.5.16
-```
-
-### Step 2 — Update OpenRAG
-
-Once Milvus 2.5.16 is healthy, stop all services and update OpenRAG to the new version. The updated `vdb/milvus.yaml` already includes Milvus 2.6.11 and the required MinIO and etcd upgrades.
-
-```bash
-docker compose down
-```
-
-Verify that all containers are stopped:
-
-```bash
-docker ps | grep milvus
-```
-
-Pull or checkout the new OpenRAG release, then start the stack:
-
-```bash
-docker compose up -d
-```
-
-Confirm the running Milvus version:
-
-```bash
-docker inspect milvus-standalone --format '{{ .Config.Image }}'
-# Expected: milvusdb/milvus:v2.6.11
-```
diff --git a/docs/content/docs/documentation/milvus_migration.mdx b/docs/content/docs/documentation/milvus_migration.mdx
new file mode 100644
index 000000000..8dd937f0c
--- /dev/null
+++ b/docs/content/docs/documentation/milvus_migration.mdx
@@ -0,0 +1,305 @@
+---
+title: Milvus Migrations
+---
+
+import { Tabs, TabItem } from '@astrojs/starlight/components';
+
+# Milvus Upgrade
+OpenRAG has been upgraded from Milvus **2.5.4** to **2.6.11** to leverage the enhancements introduced in the latest releases, particularly the new temporal querying capabilities added in version **2.6.6+**.
+
+## What's New in 2.6.x
+
+Milvus 2.6.6+ introduced the **`TIMESTAMPTZ`** field type, which enables:
+
+- **Comparison and range filtering** using standard operators (`=`, `!=`, `<`, `>`, etc.)
+- **Interval arithmetic** — add or subtract durations (days, hours, minutes) directly in filter expressions
+- **Time-based indexing** for faster temporal queries
+- **Combined filtering** — pair timestamp conditions with vector similarity search
+
+**Example — basic comparison:**
+```python
+expr = "tsz != ISO '2025-01-03T00:00:00+08:00'"
+results = client.query(
+ collection_name,
+ filter=expr,
+ output_fields=["id", "tsz"],
+ limit=10
+)
+```
+
+**Example — interval arithmetic:**
+```python
+expr = "tsz + INTERVAL 'P1D' > ISO '2025-01-03T00:00:00+08:00'"
+results = client.query(
+ collection_name,
+ filter=expr,
+ output_fields=["id", "tsz"],
+ limit=10
+)
+```
+
+> `INTERVAL` values follow [ISO 8601 duration](https://en.wikipedia.org/wiki/ISO_8601#Durations) syntax:
+> * `P1D` = 1 day
+> * `PT3H` = 3 hours
+> * `P2DT6H` = 2 days and 6 hours.
+
+## Milvus Version Upgrade Steps
+
+:::danger[Who needs this migration?]
+This migration is only required if you are upgrading from **OpenRAG <= 1.1.7**, which shipped with Milvus <= 2.5.x. If your deployment already runs Milvus 2.6.x, skip this section.
+:::
+
+> For the full official reference, see the [Milvus upgrade guide](https://milvus.io/docs/upgrade_milvus_standalone-docker.md#Upgrade-process).
+
+### Step 1 — Upgrade Milvus to 2.5.16 (intermediate step)
+
+:::caution[Do not update OpenRAG yet]
+During this step, keep your current version of OpenRAG (< 1.1.7) running. Only the Milvus image is changed here. OpenRAG itself is updated in Step 2.
+:::
+
+Milvus requires an intermediate upgrade to **v2.5.16** before jumping to 2.6.x. This step must be done manually **before** updating OpenRAG.
+
+Temporarily edit `vdb/milvus.yaml` to set the intermediate Milvus image:
+
+```diff lang=yaml
+// vdb/milvus.yaml
+milvus:
+- image: milvusdb/milvus:v2.5.4
++ image: milvusdb/milvus:v2.5.16
+```
+
+Then restart Milvus and wait for it to be healthy:
+
+```bash
+docker compose down
+docker compose up milvus -d
+```
+
+Verify it is running and healthy before continuing:
+
+```bash
+docker inspect milvus-standalone --format '{{ .Config.Image }}'
+# Expected: milvusdb/milvus:v2.5.16
+```
+
+### Step 2 — Update OpenRAG
+
+Once Milvus 2.5.16 is healthy, stop all services and update OpenRAG to the new version. The updated `vdb/milvus.yaml` already includes Milvus 2.6.11 and the required MinIO and etcd upgrades.
+
+```bash
+docker compose down
+```
+
+Verify that all containers are stopped:
+
+```bash
+docker ps | grep milvus
+```
+
+Pull or checkout the new OpenRAG release, then start the stack:
+
+```bash
+docker compose up -d
+```
+
+Confirm the running Milvus version:
+
+```bash
+docker inspect milvus-standalone --format '{{ .Config.Image }}'
+# Expected: milvusdb/milvus:v2.6.11
+```
+
+## Schema Migrations
+
+OpenRAG ships a generic migration runner that discovers and applies all pending Milvus schema migrations in order. You never need to invoke individual migration scripts by hand.
+
+:::info
+Migrations are versioned. The runner reads the current schema version stored in the collection's properties and only applies scripts that bring the collection forward (or backward) from that version.
+:::
+
+:::danger[OpenRAG must be stopped]
+Stop the OpenRAG application before running any migration.
+:::
+
+### Step 1 — Start only the Milvus container
+
+```bash
+docker compose up -d milvus
+```
+
+Wait until Milvus is healthy:
+
+```bash
+docker compose ps milvus
+```
+
+### Step 2 — Dry-run (inspect, no changes)
+
+
+
+```bash
+docker compose run --no-deps --rm --build --entrypoint "" openrag \
+ uv run python scripts/migrations/milvus/migrate.py --dry-run
+```
+
+
+```bash
+docker compose --profile cpu run --no-deps --rm --build --entrypoint "" openrag-cpu \
+ uv run python scripts/migrations/milvus/migrate.py --dry-run
+```
+
+
+
+Review the output to confirm which migrations are pending and what changes they would apply.
+
+### Step 3 — Apply all pending migrations
+
+
+
+```bash
+docker compose run --no-deps --rm --entrypoint "" openrag \
+ uv run python scripts/migrations/milvus/migrate.py
+```
+
+
+```bash
+docker compose --profile cpu run --no-deps --rm --entrypoint "" openrag-cpu \
+ uv run python scripts/migrations/milvus/migrate.py
+```
+
+
+
+The runner will apply each pending migration in order. For the `add_temporal_fields` migration (v0 → v1) this means:
+1. Adding the nullable `TIMESTAMPTZ` field `created_at`
+2. Creating an `STL_SORT` index on that field
+3. Stamping the collection with `schema_version=1` so OpenRAG no longer reports a migration error on startup
+
+### Step 4 — Restart OpenRAG
+
+
+
+```bash
+docker compose up --build -d
+```
+
+
+```bash
+docker compose --profile cpu up --build -d
+```
+
+
+
+### Targeting a specific version
+
+To upgrade or downgrade to a specific schema version rather than the latest:
+
+
+
+```bash
+# Upgrade to version 2 only
+docker compose run --no-deps --rm --entrypoint "" openrag \
+ uv run python scripts/migrations/milvus/migrate.py --target 2
+
+# Downgrade to version 0 (resets version stamp and drops indexes)
+docker compose run --no-deps --rm --entrypoint "" openrag \
+ uv run python scripts/migrations/milvus/migrate.py --downgrade --target 0
+```
+
+
+```bash
+# Upgrade to version 2 only
+docker compose --profile cpu run --no-deps --rm --entrypoint "" openrag-cpu \
+ uv run python scripts/migrations/milvus/migrate.py --target 2
+
+# Downgrade to version 0 (resets version stamp and drops indexes)
+docker compose --profile cpu run --no-deps --rm --entrypoint "" openrag-cpu \
+ uv run python scripts/migrations/milvus/migrate.py --downgrade --target 0
+```
+
+
+
+### Rollback
+
+Milvus does not support dropping fields. A downgrade only removes indexes and resets the version stamp — fields remain in the schema but are unused by the application:
+
+
+
+```bash
+docker compose run --no-deps --rm --entrypoint "" openrag \
+ uv run python scripts/migrations/milvus/migrate.py --downgrade
+```
+
+
+```bash
+docker compose --profile cpu run --no-deps --rm --entrypoint "" openrag-cpu \
+ uv run python scripts/migrations/milvus/migrate.py --downgrade
+```
+
+
+
+To fully remove the fields you would need to recreate the collection from scratch.
+
+---
+
+## Adding a New Migration Script
+
+Migration scripts live in `openrag/scripts/migrations/milvus/`. The runner discovers them automatically — no registration step required.
+
+### Naming convention
+
+Files must follow the pattern `N.short_description.py`, where `N` is the **target schema version** as a positive integer:
+
+```
+openrag/scripts/migrations/milvus/
+ 1.add_temporal_fields.py ← brings the schema to version 1
+ 2.your_new_migration.py ← brings the schema to version 2
+ migrate.py ← generic runner (do not rename)
+```
+
+The numeric prefix determines execution order. Never reuse or change an existing version number.
+
+### Required module interface
+
+Each migration script must expose the following at module level:
+
+| Name | Type | Description |
+|------|------|-------------|
+| `TARGET_VERSION` | `int` | The schema version this script brings the collection to |
+| `upgrade(client, collection_name, dry_run)` | `function` | Applies the migration |
+| `downgrade(client, collection_name, dry_run)` | `function` | Reverts the migration (indexes only — fields cannot be dropped) |
+
+### Minimal template
+
+```python
+"""
+Milvus migration: (schema version N-1 → N)
+"""
+
+from pymilvus import DataType, MilvusClient
+from utils.logger import get_logger
+
+TARGET_VERSION = N # replace with the actual version number
+
+FIELDS_2_ADD = [
+ {"field_name": "my_field", "data_type": DataType.VARCHAR, "max_length": 256, "nullable": True},
+]
+
+INDEXES_2_ADD = [
+ # add index specs here if needed
+]
+
+logger = get_logger()
+
+
+def upgrade(client: MilvusClient, collection_name: str, dry_run: bool = False) -> None:
+ # Add fields and indexes, then bump the version property.
+ ...
+
+
+def downgrade(client: MilvusClient, collection_name: str, dry_run: bool = False) -> None:
+ # Drop indexes and reset the version property.
+ # Note: Milvus does not support dropping fields.
+ ...
+```
+
+Use `1.add_temporal_fields.py` as a reference implementation for the full upgrade/downgrade pattern.
\ No newline at end of file
diff --git a/extern/indexer-ui b/extern/indexer-ui
index 3620b98b7..92e8875ee 160000
--- a/extern/indexer-ui
+++ b/extern/indexer-ui
@@ -1 +1 @@
-Subproject commit 3620b98b715d1b9bd4869fef251549e10f0edf1e
+Subproject commit 92e8875ee1537f7156e46a8dcb2a6085d887ddc8
diff --git a/openrag/components/indexer/utils/files.py b/openrag/components/indexer/utils/files.py
index f3d11b48f..e64ed100e 100644
--- a/openrag/components/indexer/utils/files.py
+++ b/openrag/components/indexer/utils/files.py
@@ -1,12 +1,13 @@
import re
import secrets
import time
+from datetime import UTC, datetime
from pathlib import Path
import aiofiles
import consts
from components.utils import load_config
-from fastapi import UploadFile
+from fastapi import HTTPException, UploadFile, status
config = load_config()
SERIALIZE_TIMEOUT = config.ray.indexer.serialize_timeout
@@ -84,3 +85,25 @@ async def serialize_file(task_id: str, path: str, metadata: dict | None = None):
timeout=SERIALIZE_TIMEOUT,
task_description=f"Serialization task {task_id}",
)
+
+
+def extract_temporal_fields(metadata: dict, temporal_fields: list) -> dict:
+ result = {}
+ for field in temporal_fields:
+ if field not in metadata or metadata[field] is None:
+ continue
+
+ datetime_str = metadata[field]
+ try:
+ # Try parsing the provided datetime to ensure it's valid
+ d = datetime.fromisoformat(datetime_str)
+ if d.tzinfo is None:
+ d = d.replace(tzinfo=UTC)
+ result[field] = d.isoformat()
+ except Exception:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"Invalid ISO 8601 datetime field ({datetime_str}) for field '{field}'.",
+ )
+
+ return result
diff --git a/openrag/components/indexer/utils/test_files.py b/openrag/components/indexer/utils/test_files.py
index 0c84336ab..ff46da5fd 100644
--- a/openrag/components/indexer/utils/test_files.py
+++ b/openrag/components/indexer/utils/test_files.py
@@ -2,9 +2,9 @@
from pathlib import Path
import pytest
-from fastapi import UploadFile
+from fastapi import HTTPException, UploadFile
-from .files import sanitize_filename, save_file_to_disk
+from .files import extract_temporal_fields, sanitize_filename, save_file_to_disk
@pytest.mark.asyncio
@@ -83,3 +83,30 @@ def fake_make_unique_filename(filename: str) -> str:
)
def test_sanitize_filename(input_name, expected):
assert sanitize_filename(input_name) == expected
+
+
+# --- extract_temporal_fields ---
+
+
+def test_extract_temporal_fields_field_not_in_metadata():
+ assert extract_temporal_fields({}, ["created_at"]) == {}
+
+
+def test_extract_temporal_fields_naive_datetime_defaults_to_utc():
+ metadata = {"created_at": "2024-06-15T12:30:00"}
+ result = extract_temporal_fields(metadata, ["created_at"])
+ assert result == {"created_at": "2024-06-15T12:30:00+00:00"}
+
+
+def test_extract_temporal_fields_with_timezone():
+ metadata = {"created_at": "2024-06-15T12:30:00+02:00"}
+ result = extract_temporal_fields(metadata, ["created_at"])
+ assert result == {"created_at": "2024-06-15T12:30:00+02:00"}
+
+
+def test_extract_temporal_fields_invalid_datetime_raises_400():
+ with pytest.raises(HTTPException) as exc_info:
+ extract_temporal_fields({"created_at": "not-a-date"}, ["created_at"])
+ assert exc_info.value.status_code == 400
+ assert "not-a-date" in exc_info.value.detail
+ assert "created_at" in exc_info.value.detail
diff --git a/openrag/components/indexer/vectordb/vectordb.py b/openrag/components/indexer/vectordb/vectordb.py
index 6f40c0b1e..82d1d4a70 100644
--- a/openrag/components/indexer/vectordb/vectordb.py
+++ b/openrag/components/indexer/vectordb/vectordb.py
@@ -1,6 +1,7 @@
import asyncio
import time
from abc import ABC, abstractmethod
+from datetime import UTC, datetime
import numpy as np
import ray
@@ -105,12 +106,9 @@ async def get_file_chunks(self, file_id: str, partition: str, include_id: bool =
async def get_chunk_by_id(self, chunk_id: str):
pass
- # @abstractmethod
- # def sample_chunk_ids(
- # self, partition: str, n_ids: int = 100, seed: int | None = None
- # ):
- # pass
+SCHEMA_VERSION_PROPERTY_KEY = "openrag.schema_version"
+INDEXED_TIME_FIELDS = ["created_at"]
MAX_LENGTH = 65_535
@@ -192,6 +190,7 @@ def load_collection(self):
try:
if self._client.has_collection(self.collection_name):
self.logger.warning(f"Collection `{self.collection_name}` already exists. Loading it.")
+ self._check_schema_version()
else:
self.logger.info("Creating empty collection")
index_params = self._create_index()
@@ -215,6 +214,7 @@ def load_collection(self):
collection_name=self.collection_name,
operation="create_collection",
)
+ self._store_schema_version()
try:
self._client.load_collection(self.collection_name)
self.collection_loaded = True
@@ -286,6 +286,9 @@ def _create_schema(self):
dim=self.embedder.embedding_dimension,
)
+ for time_field in INDEXED_TIME_FIELDS:
+ schema.add_field(field_name=time_field, datatype=DataType.TIMESTAMPTZ, nullable=True)
+
if self.hybrid_search:
# Add sparse field for BM25 - this will be auto-generated
schema.add_field(
@@ -339,9 +342,54 @@ def _create_index(self):
"bm25_b": 0.75,
},
)
+ # indexes for dates TIMESTAMPTZ field
+ for time_field in INDEXED_TIME_FIELDS:
+ index_params.add_index(
+ field_name=time_field,
+ index_type="STL_SORT", # Index for TIMESTAMPTZ
+ index_name=f"{time_field}_idx",
+ )
return index_params
+ def _store_schema_version(self) -> None:
+ """Persist the configured schema_version as a collection property after collection creation."""
+ schema_version = self.config.vectordb.schema_version
+ self._client.alter_collection_properties(
+ collection_name=self.collection_name,
+ properties={SCHEMA_VERSION_PROPERTY_KEY: str(schema_version)},
+ )
+ self.logger.info(f"Schema version {schema_version} stored on collection `{self.collection_name}`.")
+
+ def _check_schema_version(self) -> None:
+ """
+ Read the stored schema version from collection properties and compare it
+ against the configured schema_version. Raises VDBSchemaMigrationRequiredError
+ if they diverge so the application fails fast instead of silently working on a
+ stale schema.
+ """
+ expected_version = self.config.vectordb.schema_version
+ desc = self._client.describe_collection(self.collection_name)
+ props = desc.get("properties", {})
+ raw = props.get(SCHEMA_VERSION_PROPERTY_KEY)
+
+ try:
+ stored_version = int(raw) if raw is not None else 0
+ except (ValueError, TypeError):
+ stored_version = 0
+
+ if stored_version != expected_version:
+ raise VDBSchemaMigrationRequiredError(
+ f"Collection `{self.collection_name}` is at schema version {stored_version} "
+ f"but the application requires version {expected_version}. "
+ "Please perform the migration script.",
+ collection_name=self.collection_name,
+ stored_version=stored_version,
+ expected_version=expected_version,
+ )
+
+ self.logger.info(f"Collection `{self.collection_name}` schema version {stored_version} — OK.")
+
async def list_collections(self) -> list[str]:
return self._client.list_collections()
@@ -382,12 +430,14 @@ async def async_add_documents(self, chunks: list[Document], user: dict) -> None:
entities = []
vectors = await self.embedder.aembed_documents(chunks)
order_metadata_l: list[dict] = _gen_chunk_order_metadata(n=len(chunks))
+ indexed_at = datetime.now(UTC).isoformat()
for chunk, vector, order_metadata in zip(chunks, vectors, order_metadata_l):
entities.append(
{
"text": chunk.page_content,
"vector": vector,
+ "indexed_at": indexed_at,
**order_metadata,
**chunk.metadata,
}
@@ -399,6 +449,7 @@ async def async_add_documents(self, chunks: list[Document], user: dict) -> None:
)
# insert file_id and partition into partition_file_manager
+ file_metadata.update({"indexed_at": indexed_at})
self.partition_file_manager.add_file_to_partition(
file_id=file_id,
partition=partition,
@@ -500,7 +551,6 @@ async def async_search(
# Join all parts with " and " only if there are multiple conditions
expr = " and ".join(expr_parts) if expr_parts else ""
- filter_params = filter_params or {}
try:
query_vector = await self.embedder.aembed_query(query)
@@ -517,7 +567,6 @@ async def async_search(
},
"limit": top_k,
"expr": expr,
- "expr_params": filter_params,
}
if self.hybrid_search:
sparse_param = {
@@ -529,7 +578,6 @@ async def async_search(
},
"limit": top_k,
"expr": expr,
- "expr_params": filter_params,
}
reqs = [
AnnSearchRequest(**vector_param),
@@ -560,7 +608,6 @@ async def async_search(
collection_name=self.collection_name,
output_fields=["*"],
filter=expr,
- filter_params=filter_params,
**vector_param,
)
diff --git a/openrag/config/models.py b/openrag/config/models.py
index 59eabd7e8..bdfdaf3ad 100644
--- a/openrag/config/models.py
+++ b/openrag/config/models.py
@@ -110,6 +110,7 @@ class VectorDBConfig(ConfigMixin):
collection_name: str = "vdb_test"
hybrid_search: bool = True
enable: bool = True
+ schema_version: int = 1
# ---------------------------------------------------------------------------
diff --git a/openrag/routers/extract.py b/openrag/routers/extract.py
index 5179a1462..c47d39121 100644
--- a/openrag/routers/extract.py
+++ b/openrag/routers/extract.py
@@ -31,9 +31,6 @@
- `filename`: Original filename
- `partition`: Partition name
- `page`: Page number in source document
- - `datetime`: Document date (if set)
- - `modified_at`: File modification timestamp
- - `created_at`: File creation timestamp
- `indexed_at`: Chunk indexing timestamp
- Additional custom metadata
diff --git a/openrag/routers/indexer.py b/openrag/routers/indexer.py
index 861a75080..0d057e27b 100644
--- a/openrag/routers/indexer.py
+++ b/openrag/routers/indexer.py
@@ -1,10 +1,9 @@
import json
-from datetime import datetime
from pathlib import Path
from typing import Any
import ray
-from components.indexer.utils.files import sanitize_filename, save_file_to_disk
+from components.indexer.utils.files import extract_temporal_fields, sanitize_filename, save_file_to_disk
from components.ray_utils import call_ray_actor_with_timeout
from config import load_config
from fastapi import (
@@ -51,6 +50,9 @@
# URL scheme configuration
PREFERRED_URL_SCHEME = config.server.preferred_url_scheme
+# DATETIME FIELDS: Fields provided by the client
+TEMPORAL_FIELDS = ["created_at"]
+
def build_url(request: Request, route_name: str, **path_params) -> str:
"""Build a URL using the preferred scheme if configured."""
@@ -102,9 +104,14 @@ async def get_supported_types():
"mimetype": "text/plain",
"author": "John Doe",
...
+ "created_at": "2025-01-03T00:00:00+08:00" // Optional temporal field (ISO 8601)
}
```
+**Temporal Fields:**
+- You can provide a temporal fields such as `created_at` in the metadata for time-based queries and filtering.
+- Datetime values must be in ISO 8601 format (e.g., `2025-01-03T00:00:00+08:00`).
+
**Common Mimetypes:**
- `text/plain` - Plain text files
- `text/markdown` - Markdown files
@@ -157,9 +164,12 @@ async def add_file(
# Append extra metadata
metadata["file_size"] = human_readable_size(file_stat.st_size)
- metadata["created_at"] = datetime.fromtimestamp(file_stat.st_ctime).isoformat()
metadata["file_id"] = file_id
+ ## Add temporal fields to metadata, using provided values if available, otherwise extracting from file system
+ temporal_fields = extract_temporal_fields(metadata, temporal_fields=TEMPORAL_FIELDS)
+ metadata.update(temporal_fields)
+
# Validate and parse workspace_ids
parsed_workspace_ids = None
if workspace_ids:
@@ -247,9 +257,14 @@ async def delete_file(
"mimetype": "text/plain",
"author": "John Doe",
...
+ "created_at": "2024-01-01T12:00:00+00:00" // Optional temporal field (ISO 8601)
}
```
+**Temporal Fields:**
+- You can provide the temporal fields `created_at` in the metadata for time-based queries and filtering.
+- Datetime values must be in ISO 8601 format (e.g., `2024-01-01T12:00:00+00:00`).
+
**Response:**
Returns 202 Accepted with a task status URL for tracking indexing progress.
""",
@@ -298,9 +313,12 @@ async def put_file(
# Append extra metadata
metadata["file_size"] = human_readable_size(file_stat.st_size)
- metadata["created_at"] = datetime.fromtimestamp(file_stat.st_ctime).isoformat()
metadata["file_id"] = file_id
+ ## Add temporal fields to metadata, using provided values if available, otherwise extracting from file system
+ temporal_fields = extract_temporal_fields(metadata, temporal_fields=TEMPORAL_FIELDS)
+ metadata.update(temporal_fields)
+
# Indexing the file — restore pre-existing workspace memberships on the new version.
task = indexer.add_file.remote(
path=file_path,
diff --git a/openrag/routers/search.py b/openrag/routers/search.py
index 6b19e41a1..05d3595c1 100644
--- a/openrag/routers/search.py
+++ b/openrag/routers/search.py
@@ -27,9 +27,11 @@ def __init__(
self,
include_related: bool = Query(False, description="Include chunks from files with same relationship_id"),
include_ancestors: bool = Query(False, description="Include chunks from ancestor files in hierarchy"),
- related_limit: int = Query(20, description="Maximum number of related/ancestor chunks to fetch per result"),
+ related_limit: int = Query(
+ 20, ge=0, description="Maximum number of related/ancestor chunks to fetch per result"
+ ),
max_ancestor_depth: int | None = Query(
- None, description="Maximum depth of ancestor files to include. None means unlimited."
+ None, ge=0, description="Maximum depth of ancestor files to include. None means unlimited."
),
):
self.include_related = include_related
@@ -42,8 +44,10 @@ class CommonSearchParams:
def __init__(
self,
text: str = Query(..., description="Text to search semantically"),
- top_k: int = Query(5, description="Number of top results to return"),
- similarity_threshold: float = Query(0.6, description="Minimum similarity score for results (0 to 1)"),
+ top_k: int = Query(5, ge=1, description="Number of top results to return"),
+ similarity_threshold: float = Query(
+ 0.75, ge=0, le=1, description="Minimum similarity score for results (0 to 1)"
+ ),
filter: str | None = Query(
default=None,
description="""Milvus filter expression string.""",
@@ -75,7 +79,7 @@ def __init__(
- Logical: AND, OR, NOT (see https://milvus.io/docs/boolean.md)
Examples:
- `file_id == "abc123"`
- - `created_at > "2024-01-01"`
+ - `created_at > ISO "2024-01-01T00:00:00+00:00"`
- `page >= 5 AND page <= 10`
- `file_id in ["id1", "id2", "id3"]`
@@ -176,7 +180,6 @@ async def search_multiple_partitions(
}
for doc in results
]
-
return JSONResponse(status_code=status.HTTP_200_OK, content={"documents": documents})
@@ -202,7 +205,7 @@ async def search_multiple_partitions(
- Logical: AND, OR, NOT (see https://milvus.io/docs/boolean.md)
Examples:
- `file_id == "abc123"`
- - `created_at > "2024-01-01"`
+ - `created_at > ISO "2024-01-01T00:00:00+00:00"`
- `page >= 5 AND page <= 10`
- `file_id in ["id1", "id2", "id3"]`
@@ -284,7 +287,6 @@ async def search_one_partition(
}
for doc in results
]
-
return JSONResponse(status_code=status.HTTP_200_OK, content={"documents": documents})
@@ -307,7 +309,7 @@ async def search_one_partition(
- Logical: AND, OR, NOT (see https://milvus.io/docs/boolean.md)
Examples:
- `file_id == "abc123"`
- - `created_at > "2024-01-01"`
+ - `created_at > ISO "2024-01-01T00:00:00+00:00"`
- `page >= 5 AND page <= 10`
- `file_id in ["id1", "id2", "id3"]`
@@ -356,5 +358,4 @@ async def search_file(
}
for doc in results
]
-
return JSONResponse(status_code=status.HTTP_200_OK, content={"documents": documents})
diff --git a/openrag/scripts/migrations/milvus/1.add_created_at_temporal_fields.py b/openrag/scripts/migrations/milvus/1.add_created_at_temporal_fields.py
new file mode 100644
index 000000000..563f0193a
--- /dev/null
+++ b/openrag/scripts/migrations/milvus/1.add_created_at_temporal_fields.py
@@ -0,0 +1,231 @@
+"""
+Milvus migration: add temporal fields (schema version 0 → 1)
+==============================================================
+Adds the following TIMESTAMPTZ field (nullable) to the existing collection:
+ - created_at
+
+The field also gets an STL_SORT index. After a successful upgrade the
+collection's ``openrag.schema_version`` property is set to 1 so the application
+no longer raises VDBSchemaMigrationRequiredError on startup.
+
+Existing documents will retain null for these fields; new documents will have
+them populated at index time by the application code.
+
+Usage — prefer the generic runner (from repo root, inside the container):
+ docker compose run --no-deps --rm --entrypoint "" openrag \\
+ uv run python scripts/migrations/milvus/migrate.py [--dry-run] [--downgrade] [--target N]
+
+Or run this script directly:
+ # Dry-run first (inspect only, no changes):
+ docker compose run --no-deps --rm --build --entrypoint "" openrag \\
+ uv run python scripts/migrations/milvus/1.add_temporal_fields.py --dry-run
+
+ # Apply:
+ docker compose run --no-deps --rm --build --entrypoint "" openrag \\
+ uv run python scripts/migrations/milvus/1.add_temporal_fields.py
+
+ # Roll back indexes and reset version (fields cannot be dropped in Milvus):
+ docker compose run --no-deps --rm --entrypoint "" openrag \\
+ uv run python scripts/migrations/milvus/1.add_temporal_fields.py --downgrade
+"""
+
+import argparse
+import sys
+
+from components.indexer.vectordb.vectordb import SCHEMA_VERSION_PROPERTY_KEY
+from config import load_config
+from pymilvus import DataType, MilvusClient
+from utils.logger import get_logger
+
+TARGET_VERSION = 1 # The schema version this migration brings the collection to.
+
+# ---------------------------------------------------------------------------
+# Declarative migration spec — edit here to change what gets added/removed.
+# ---------------------------------------------------------------------------
+
+FIELDS_2_ADD = [
+ {"field_name": "created_at", "data_type": DataType.TIMESTAMPTZ, "nullable": True},
+]
+
+INDEXES_2_ADD = [
+ {"field_name": "created_at", "index_type": "STL_SORT", "index_name": "created_at_idx"},
+]
+
+# ---------------------------------------------------------------------------
+
+logger = get_logger()
+
+
+def _get_existing_field_names(client: MilvusClient, collection_name: str) -> set[str]:
+ desc = client.describe_collection(collection_name)
+ return {f["name"] for f in desc["fields"]}
+
+
+def _get_existing_index_names(client: MilvusClient, collection_name: str) -> set[str]:
+ return set(client.list_indexes(collection_name))
+
+
+def _get_stored_version(client: MilvusClient, collection_name: str) -> int:
+ desc = client.describe_collection(collection_name)
+ raw = desc.get("properties", {}).get(SCHEMA_VERSION_PROPERTY_KEY)
+ if raw is None:
+ return 0
+ try:
+ return int(raw)
+ except ValueError:
+ return 0
+
+
+def _print_state(client: MilvusClient, collection_name: str, required_version: int) -> None:
+ """Query and display the current state of fields, indexes and schema version."""
+ desc = client.describe_collection(collection_name)
+ field_map = {f["name"]: f for f in desc["fields"]}
+ existing_indexes = _get_existing_index_names(client, collection_name)
+ stored_version = _get_stored_version(client, collection_name)
+
+ field_to_index = {idx["field_name"]: idx["index_name"] for idx in INDEXES_2_ADD}
+
+ logger.info(f"--- Collection '{collection_name}' state ---")
+ logger.info(f" Schema version : stored={stored_version} required={required_version}")
+ logger.info(" Fields:")
+ for field_spec in FIELDS_2_ADD:
+ field_name = field_spec["field_name"]
+ index_name = field_to_index.get(field_name)
+ field_present = field_name in field_map
+ index_present = index_name is not None and index_name in existing_indexes
+
+ index_detail = ""
+ if index_present:
+ info = client.describe_index(collection_name=collection_name, index_name=index_name)
+ index_detail = f" (index_type={info.get('index_type')})"
+
+ index_status = f"OK{index_detail}" if index_present else ("N/A" if not index_name else "MISSING")
+ logger.info(f" {field_name}: field={'OK' if field_present else 'MISSING'} | index={index_status}")
+
+
+def upgrade(client: MilvusClient, collection_name: str, dry_run: bool = False) -> None:
+ stored_version = _get_stored_version(client, collection_name)
+ if stored_version >= TARGET_VERSION:
+ logger.info(f"Collection is already at version {stored_version} — nothing to do.")
+ _print_state(client, collection_name, TARGET_VERSION)
+ return
+
+ existing_fields = _get_existing_field_names(client, collection_name)
+ existing_indexes = _get_existing_index_names(client, collection_name)
+ fields_added: list[str] = []
+
+ for field_spec in FIELDS_2_ADD:
+ field_name = field_spec["field_name"]
+ if field_name in existing_fields:
+ logger.info(f"Field '{field_name}' already exists — skipping.")
+ continue
+
+ logger.info(f"{'[DRY-RUN] ' if dry_run else ''}Adding field '{field_name}'.")
+ if not dry_run:
+ client.add_collection_field(collection_name=collection_name, **field_spec)
+ fields_added.append(field_name)
+
+ index_params = client.prepare_index_params()
+ needs_index = False
+
+ for idx_spec in INDEXES_2_ADD:
+ field_name = idx_spec["field_name"]
+ index_name = idx_spec["index_name"]
+
+ if index_name in existing_indexes:
+ logger.info(f"Index '{index_name}' already exists — skipping.")
+ continue
+
+ if field_name not in existing_fields and field_name not in fields_added:
+ logger.warning(f"Field '{field_name}' could not be added — skipping index.")
+ continue
+
+ logger.info(f"{'[DRY-RUN] ' if dry_run else ''}Scheduling index '{index_name}' for '{field_name}'.")
+ index_params.add_index(**idx_spec)
+ needs_index = True
+
+ if not dry_run:
+ if needs_index:
+ logger.info("Creating indexes...")
+ client.create_index(collection_name=collection_name, index_params=index_params)
+ logger.info("Indexes created.")
+
+ # Bump stored version so the application stops raising VDBSchemaMigrationRequiredError.
+ logger.info(f"Storing schema version {TARGET_VERSION} on collection '{collection_name}'.")
+ client.alter_collection_properties(
+ collection_name=collection_name,
+ properties={SCHEMA_VERSION_PROPERTY_KEY: str(TARGET_VERSION)},
+ )
+ logger.info("Migration complete.")
+ else:
+ logger.info("Dry-run complete. No changes were made.")
+
+ _print_state(client, collection_name, TARGET_VERSION)
+
+
+def downgrade(client: MilvusClient, collection_name: str, dry_run: bool = False) -> None:
+ """
+ Milvus does NOT support dropping fields from an existing collection.
+ The only way to fully roll back is to drop and recreate the collection,
+ which would lose all data.
+
+ This function only removes the indexes and resets the version property.
+ """
+ existing_indexes = _get_existing_index_names(client, collection_name)
+
+ for idx_spec in INDEXES_2_ADD:
+ index_name = idx_spec["index_name"]
+ if index_name not in existing_indexes:
+ logger.info(f"Index '{index_name}' does not exist — skipping.")
+ continue
+
+ logger.info(f"{'[DRY-RUN] ' if dry_run else ''}Dropping index '{index_name}'.")
+ if not dry_run:
+ client.drop_index(collection_name=collection_name, index_name=index_name)
+
+ if not dry_run:
+ client.alter_collection_properties(
+ collection_name=collection_name,
+ properties={SCHEMA_VERSION_PROPERTY_KEY: "0"},
+ )
+ logger.info("Schema version reset to 0.")
+
+ logger.warning(
+ f"Fields {[f['field_name'] for f in FIELDS_2_ADD]} cannot be dropped from Milvus. "
+ "To fully remove them you would need to recreate the collection."
+ )
+
+ _print_state(client, collection_name, required_version=0)
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description="Milvus migration: add temporal fields (v0 → v1)")
+ parser.add_argument("--dry-run", action="store_true", help="Inspect only, make no changes")
+ parser.add_argument(
+ "--downgrade",
+ action="store_true",
+ help="Drop indexes and reset version (fields cannot be dropped in Milvus)",
+ )
+ args = parser.parse_args()
+
+ cfg = load_config()
+ host = cfg.vectordb.host
+ port = cfg.vectordb.port
+ collection_name = cfg.vectordb.collection_name
+ uri = f"http://{host}:{port}"
+
+ logger.info(f"Connecting to Milvus at {uri}, collection='{collection_name}'")
+ client = MilvusClient(uri=uri)
+
+ if not client.has_collection(collection_name):
+ logger.error(f"Collection '{collection_name}' does not exist. Aborting.")
+ sys.exit(1)
+
+ if args.downgrade:
+ downgrade(client, collection_name, dry_run=args.dry_run)
+ else:
+ upgrade(client, collection_name, dry_run=args.dry_run)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/openrag/scripts/migrations/milvus/migrate.py b/openrag/scripts/migrations/milvus/migrate.py
new file mode 100644
index 000000000..33df55c8c
--- /dev/null
+++ b/openrag/scripts/migrations/milvus/migrate.py
@@ -0,0 +1,213 @@
+"""
+Generic Milvus migration runner
+================================
+Discovers all migration scripts in this directory (files matching ``N.*.py``),
+sorts them by their numeric prefix, and runs ``upgrade()`` / ``downgrade()``
+in order based on the current schema version stored in the collection.
+
+Usage (from repo root, inside the container):
+
+ # Dry-run — inspect what would change, no writes:
+ docker compose run --no-deps --rm --build --entrypoint "" openrag \\
+ uv run python scripts/migrations/milvus/migrate.py --dry-run
+
+ # Upgrade to latest:
+ docker compose run --no-deps --rm --entrypoint "" openrag \\
+ uv run python scripts/migrations/milvus/migrate.py
+
+ # Upgrade to a specific version:
+ docker compose run --no-deps --rm --entrypoint "" openrag \\
+ uv run python scripts/migrations/milvus/migrate.py --target 2
+
+ # Downgrade to version 0 (resets version property, drops indexes):
+ docker compose run --no-deps --rm --entrypoint "" openrag \\
+ uv run python scripts/migrations/milvus/migrate.py --downgrade --target 0
+
+Convention — each migration module must expose:
+ TARGET_VERSION: int # the version this script brings the DB to
+ upgrade(client, collection_name, dry_run=False)
+ downgrade(client, collection_name, dry_run=False)
+"""
+
+import argparse
+import importlib.util
+import re
+import sys
+from pathlib import Path
+from types import ModuleType
+
+from components.indexer.vectordb.vectordb import SCHEMA_VERSION_PROPERTY_KEY
+from config import load_config
+from pymilvus import MilvusClient
+from utils.logger import get_logger
+
+logger = get_logger()
+
+_MIGRATION_PATTERN = re.compile(r"^(\d+)\..+\.py$")
+_MIGRATIONS_DIR = Path(__file__).parent
+
+
+# ---------------------------------------------------------------------------
+# Discovery
+# ---------------------------------------------------------------------------
+
+
+def _discover_migrations() -> list[tuple[int, Path]]:
+ """Return (version, path) pairs sorted by version for all migration files."""
+ found: list[tuple[int, Path]] = []
+ for p in _MIGRATIONS_DIR.iterdir():
+ if p.name == Path(__file__).name:
+ continue # skip this runner
+ m = _MIGRATION_PATTERN.match(p.name)
+ if m:
+ found.append((int(m.group(1)), p))
+ found.sort(key=lambda x: x[0])
+ return found
+
+
+def _load_module(path: Path) -> ModuleType:
+ spec = importlib.util.spec_from_file_location(path.stem, path)
+ if spec is None or spec.loader is None:
+ raise ImportError(f"Cannot load migration module: {path}")
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module) # type: ignore[union-attr]
+ return module
+
+
+def _validate_module(module: ModuleType, path: Path) -> None:
+ for attr in ("TARGET_VERSION", "upgrade", "downgrade"):
+ if not hasattr(module, attr):
+ raise AttributeError(f"Migration '{path.name}' is missing required attribute '{attr}'")
+
+
+# ---------------------------------------------------------------------------
+# Version helpers
+# ---------------------------------------------------------------------------
+
+
+def _get_stored_version(client: MilvusClient, collection_name: str) -> int:
+ desc = client.describe_collection(collection_name)
+ raw = desc.get("properties", {}).get(SCHEMA_VERSION_PROPERTY_KEY)
+ if raw is None:
+ return 0
+ try:
+ return int(raw)
+ except ValueError:
+ return 0
+
+
+# ---------------------------------------------------------------------------
+# Runner
+# ---------------------------------------------------------------------------
+
+
+def run_upgrade(
+ client: MilvusClient,
+ collection_name: str,
+ migrations: list[tuple[int, Path]],
+ target_version: int,
+ dry_run: bool,
+) -> None:
+ current = _get_stored_version(client, collection_name)
+ logger.info(f"Current schema version: {current} → target: {target_version}")
+
+ pending = [(v, p) for v, p in migrations if current < v <= target_version]
+ if not pending:
+ logger.info("Collection is already up to date — nothing to do.")
+ return
+
+ for version, path in pending:
+ logger.info(f"--- Applying migration {path.name} (v{version}) ---")
+ module = _load_module(path)
+ _validate_module(module, path)
+ module.upgrade(client, collection_name, dry_run=dry_run)
+
+ logger.info("All pending migrations applied.")
+
+
+def run_downgrade(
+ client: MilvusClient,
+ collection_name: str,
+ migrations: list[tuple[int, Path]],
+ target_version: int,
+ dry_run: bool,
+) -> None:
+ current = _get_stored_version(client, collection_name)
+ logger.info(f"Current schema version: {current} → downgrade target: {target_version}")
+
+ # Run in reverse: undo the highest version first
+ pending = [(v, p) for v, p in migrations if target_version < v <= current]
+ pending.sort(key=lambda x: x[0], reverse=True)
+
+ if not pending:
+ logger.info("Nothing to downgrade.")
+ return
+
+ for version, path in pending:
+ logger.info(f"--- Reverting migration {path.name} (v{version}) ---")
+ module = _load_module(path)
+ _validate_module(module, path)
+ module.downgrade(client, collection_name, dry_run=dry_run)
+
+ logger.info("Downgrade complete.")
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+ migrations = _discover_migrations()
+ latest_version = migrations[-1][0] if migrations else 0
+
+ parser = argparse.ArgumentParser(
+ description="Generic Milvus migration runner — discovers and applies all pending migrations."
+ )
+ parser.add_argument("--dry-run", action="store_true", help="Inspect only, make no changes")
+ parser.add_argument(
+ "--downgrade",
+ action="store_true",
+ help="Run downgrade instead of upgrade",
+ )
+ parser.add_argument(
+ "--target",
+ type=int,
+ default=None,
+ help=f"Target schema version (default: {latest_version} for upgrade, 0 for downgrade)",
+ )
+ args = parser.parse_args()
+
+ # Resolve default target
+ if args.target is None:
+ target_version = 0 if args.downgrade else latest_version
+ else:
+ target_version = args.target
+
+ if not migrations:
+ logger.warning("No migration files found in this directory.")
+ sys.exit(0)
+
+ logger.info(f"Discovered {len(migrations)} migration(s): {[p.name for _, p in migrations]}")
+
+ cfg = load_config()
+ host = cfg.vectordb.host
+ port = cfg.vectordb.port
+ collection_name = cfg.vectordb.collection_name
+ uri = f"http://{host}:{port}"
+
+ logger.info(f"Connecting to Milvus at {uri}, collection='{collection_name}'")
+ client = MilvusClient(uri=uri)
+
+ if not client.has_collection(collection_name):
+ logger.error(f"Collection '{collection_name}' does not exist. Aborting.")
+ sys.exit(1)
+
+ if args.downgrade:
+ run_downgrade(client, collection_name, migrations, target_version, dry_run=args.dry_run)
+ else:
+ run_upgrade(client, collection_name, migrations, target_version, dry_run=args.dry_run)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/openrag/utils/exceptions/vectordb.py b/openrag/utils/exceptions/vectordb.py
index 964f9ed68..e9bf7cbc1 100644
--- a/openrag/utils/exceptions/vectordb.py
+++ b/openrag/utils/exceptions/vectordb.py
@@ -106,6 +106,18 @@ def __init__(self, message: str, **kwargs):
)
+class VDBSchemaMigrationRequiredError(VDBError):
+ """Raised when the collection schema version does not match the expected version."""
+
+ def __init__(self, message: str, **kwargs):
+ super().__init__(
+ message=message,
+ code="VDB_SCHEMA_MIGRATION_REQUIRED",
+ status_code=503,
+ **kwargs,
+ )
+
+
class UnexpectedVDBError(VDBError):
"""Raised for unexpected errors in vector database operations."""
diff --git a/tests/api_tests/api_run/docker-compose.yaml b/tests/api_tests/api_run/docker-compose.yaml
index 9bf54913e..f3085533a 100644
--- a/tests/api_tests/api_run/docker-compose.yaml
+++ b/tests/api_tests/api_run/docker-compose.yaml
@@ -28,12 +28,13 @@ services:
retries: 10
etcd:
- image: quay.io/coreos/etcd:v3.5.16
+ image: quay.io/coreos/etcd:v3.5.25
environment:
- ETCD_AUTO_COMPACTION_MODE=revision
- ETCD_AUTO_COMPACTION_RETENTION=1000
- ETCD_QUOTA_BACKEND_BYTES=4294967296
- command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd
+ - ETCD_SNAPSHOT_COUNT=50000
+ command: etcd -advertise-client-urls=http://etcd:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd
healthcheck:
test: ["CMD", "etcdctl", "endpoint", "health"]
interval: 5s
@@ -41,7 +42,7 @@ services:
retries: 5
minio:
- image: minio/minio:RELEASE.2023-03-20T20-16-18Z
+ image: minio/minio:RELEASE.2024-12-18T13-15-44Z
environment:
MINIO_ACCESS_KEY: minioadmin
MINIO_SECRET_KEY: minioadmin
@@ -53,10 +54,10 @@ services:
retries: 5
milvus:
- image: milvusdb/milvus:v2.5.4
+ image: milvusdb/milvus:v2.6.11
command: ["milvus", "run", "standalone"]
security_opt:
- - seccomp:unconfined
+ - seccomp:unconfined
environment:
ETCD_ENDPOINTS: etcd:2379
MINIO_ADDRESS: minio:9000
diff --git a/tests/api_tests/test_search.py b/tests/api_tests/test_search.py
index 04c328887..57af79439 100644
--- a/tests/api_tests/test_search.py
+++ b/tests/api_tests/test_search.py
@@ -3,6 +3,7 @@
import json
import time
import uuid
+from datetime import datetime
import pytest
@@ -496,12 +497,42 @@ class TestSearchFiltering:
def filter_test_files(self, tmp_path):
"""Create 6 files with the same content but different metadata."""
files_config = [
- {"file_id": "filter-file-1", "origin": "source_A", "file_number": 1},
- {"file_id": "filter-file-2", "origin": "source_A", "file_number": 2},
- {"file_id": "filter-file-3", "origin": "source_B", "file_number": 3},
- {"file_id": "filter-file-4", "origin": "source_B", "file_number": 4},
- {"file_id": "filter-file-5", "origin": "source_C", "file_number": 5},
- {"file_id": "filter-file-6", "origin": "source_C", "file_number": 6},
+ {
+ "file_id": "filter-file-1",
+ "origin": "source_A",
+ "file_number": 1,
+ "created_at": "2020-06-15T00:00:00+00:00",
+ },
+ {
+ "file_id": "filter-file-2",
+ "origin": "source_A",
+ "file_number": 2,
+ "created_at": "2021-06-15T00:00:00+00:00",
+ },
+ {
+ "file_id": "filter-file-3",
+ "origin": "source_B",
+ "file_number": 3,
+ "created_at": "2022-06-15T00:00:00+00:00",
+ },
+ {
+ "file_id": "filter-file-4",
+ "origin": "source_B",
+ "file_number": 4,
+ "created_at": "2023-06-15T00:00:00+00:00",
+ },
+ {
+ "file_id": "filter-file-5",
+ "origin": "source_C",
+ "file_number": 5,
+ "created_at": "2024-06-15T00:00:00+00:00",
+ },
+ {
+ "file_id": "filter-file-6",
+ "origin": "source_C",
+ "file_number": 6,
+ "created_at": "2024-07-15T00:00:00+00:00",
+ },
]
file_paths = {}
@@ -709,3 +740,91 @@ def test_logical_operator_OR(self, api_client, indexed_filter_partition):
file_ids = {doc["metadata"].get("file_id") for doc in documents}
expected_ids = {"filter-file-1", "filter-file-2", "filter-file-6"}
assert file_ids == expected_ids, f"Expected {expected_ids}, got {file_ids}"
+
+ # =========================================================================
+ # Temporal filtering tests (datetime field, ISO 8601)
+ # =========================================================================
+
+ def test_temporal_fields_present_in_metadata(self, api_client, indexed_filter_partition):
+ """Test temporal fields are present in the metadata."""
+ response = api_client.get(
+ f"/search/partition/{indexed_filter_partition}",
+ params={
+ "text": self.COMMON_CONTENT,
+ "top_k": 10,
+ },
+ )
+ assert response.status_code == 200
+ data = response.json()
+ assert "documents" in data
+
+ documents = data["documents"]
+ assert len(documents) > 0, "Should find at least one document"
+
+ # Verify that each document has temporal fields in metadata
+ for doc in documents:
+ metadata = doc.get("metadata", {})
+ for temp_field in ["created_at"]:
+ k = metadata.get(temp_field)
+ assert k is not None, (
+ f"Document {doc['metadata'].get('file_id')} is missing temporal field '{temp_field}'"
+ )
+ # Verify it's a valid ISO 8601 datetime string
+ try:
+ datetime.fromisoformat(k)
+ except ValueError:
+ assert False, (
+ f"Document {doc['metadata'].get('file_id')} has invalid datetime format in field '{temp_field}': {k}"
+ )
+
+ def test_temporal_filter_with_iso_format(self, api_client, indexed_filter_partition):
+ """Test that temporal filtering on the created_at field works with ISO 8601 strings."""
+ partition = indexed_filter_partition
+
+ # --- before 2022: should return files 1 and 2 ---
+ resp = api_client.get(
+ f"/search/partition/{partition}",
+ params={
+ "text": self.COMMON_CONTENT,
+ "top_k": 10,
+ "filter": 'created_at < ISO "2022-01-01T00:00:00+00:00"',
+ },
+ )
+ assert resp.status_code == 200
+ data = resp.json()
+ assert "documents" in data
+ file_ids = {doc["metadata"].get("file_id") for doc in data["documents"]}
+
+ assert file_ids == {"filter-file-1", "filter-file-2"}, f"Expected files before 2022, got {file_ids}"
+
+ # --- after 2024-01-01: should return files 5 and 6 ---
+ resp = api_client.get(
+ f"/search/partition/{partition}",
+ params={
+ "text": self.COMMON_CONTENT,
+ "top_k": 10,
+ "filter": 'created_at > ISO "2024-01-01T00:00:00+00:00"',
+ },
+ )
+ assert resp.status_code == 200
+ data = resp.json()
+ assert "documents" in data
+ file_ids = {doc["metadata"].get("file_id") for doc in data["documents"]}
+ assert file_ids == {"filter-file-5", "filter-file-6"}, f"Expected files after 2024, got {file_ids}"
+
+ # --- range [2022, 2024]: should return files 3 and 4 ---
+ resp = api_client.get(
+ f"/search/partition/{partition}",
+ params={
+ "text": self.COMMON_CONTENT,
+ "top_k": 10,
+ "filter": 'created_at >= ISO "2022-01-01T00:00:00+00:00" AND created_at <= ISO "2024-01-01T00:00:00+00:00"',
+ },
+ )
+ assert resp.status_code == 200
+ data = resp.json()
+ assert "documents" in data
+ file_ids = {doc["metadata"].get("file_id") for doc in data["documents"]}
+ assert file_ids == {"filter-file-3", "filter-file-4"}, (
+ f"Expected filter-file-3 and filter-file-4 in range [2022, 2024], got {file_ids}"
+ )