Add index management system with datetime partitioning

Author: Grzegorz Pustulka

CHANGELOG.md

@@ -11,6 +11,28 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Added
 
 - Added the ability to set timeout for Opensearch and Elasticsearch clients by setting the environmental variable `ES_TIMEOUT` [#408](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/408)
+- Added comprehensive index management system with dynamic selection and insertion strategies for improved performance and scalability [#405](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/405)
+- Added `ENABLE_DATETIME_INDEX_FILTERING` environment variable to enable datetime-based index selection using collection IDs. Requires indexes in format: `STAC_ITEMS_INDEX_PREFIX_collection-id_start_year-start_month-start_day-end_year-end_month-end_day`, e.g. `items_sentinel-2-l2a_2025-06-06-2025-09-22`. Default is `false`. [#405](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/405)
+- Added `DATETIME_INDEX_MAX_SIZE_GB` environment variable to set maximum size limit in GB for datetime-based indexes. When an index exceeds this size, a new time-partitioned index will be created. Default is `25` GB. Only applies when `ENABLE_DATETIME_INDEX_FILTERING` is enabled. [#405](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/405)
+- Added search engine adapter system with support for both Elasticsearch and OpenSearch [#405](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/405):
+  - `SearchEngineAdapter` base class with engine-specific implementations
+  - `ElasticsearchAdapter` and `OpenSearchAdapter` with tailored index creation methods
+  - Automatic engine type detection based on client class
+  - `SearchEngineAdapterFactory` for creating appropriate adapters
+- Added datetime-based index selection strategies with caching support [#405](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/405):
+  - `AsyncDatetimeBasedIndexSelector` and `SyncDatetimeBasedIndexSelector` for temporal filtering
+  - `IndexCacheManager` with configurable TTL-based cache expiration (default 1 hour)
+  - `AsyncIndexAliasLoader` and `SyncIndexAliasLoader` for alias management
+  - `UnfilteredIndexSelector` as fallback for returning all available indexes
+- Added index insertion strategies with automatic partitioning [#405](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/405):
+  - Simple insertion strategy (`AsyncSimpleIndexInserter`, `SyncSimpleIndexInserter`) for traditional single-index-per-collection approach
+  - Datetime-based insertion strategy (`AsyncDatetimeIndexInserter`, `SyncDatetimeIndexInserter`) with time-based partitioning
+  - Automatic index size monitoring and splitting when limits exceeded
+  - Handling of chronologically early data and bulk operations
+- Added index management utilities [#405](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/405):
+  - `IndexSizeManager` for size monitoring and overflow handling
+  - `DatetimeIndexManager` for datetime-based index operations
+  - Factory patterns (`IndexInsertionFactory`, `IndexSelectorFactory`) for strategy creation based on configuration
 
 ## [v6.0.0] - 2025-06-22
 
@@ -22,6 +44,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 - Updated stac-fastapi parent libraries to v6.0.0 [#291](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/291)
 
+
 ## [v5.0.0] - 2025-06-11
 
 ### Added

Makefile

@@ -73,10 +73,10 @@ test-opensearch:
 
 .PHONY: test
 test:
-	-$(run_es) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh elasticsearch:9200 && cd stac_fastapi/tests/ && pytest --cov=stac_fastapi --cov-report=term-missing'
+	-$(run_es) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh elasticsearch:9200 && cd stac_fastapi/tests/ && pytest -s --cov=stac_fastapi --cov-report=term-missing'
 	docker compose down
 
-	-$(run_os) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh opensearch:9202 && cd stac_fastapi/tests/ && pytest --cov=stac_fastapi --cov-report=term-missing'
+	-$(run_os) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh opensearch:9202 && cd stac_fastapi/tests/ && pytest -s --cov=stac_fastapi --cov-report=term-missing'
 	docker compose down
 
 .PHONY: run-database-es

README.md

@@ -21,6 +21,7 @@ services:
       - ES_USE_SSL=false
       - ES_VERIFY_CERTS=false
       - BACKEND=elasticsearch
+      - ENABLE_DATETIME_INDEX_FILTERING=true
     ports:
       - "8080:8080"
     volumes:
@@ -55,6 +56,7 @@ services:
       - ES_VERIFY_CERTS=false
       - BACKEND=opensearch
       - STAC_FASTAPI_RATE_LIMIT=200/minute
+      - ENABLE_DATETIME_INDEX_FILTERING=true
     ports:
       - "8082:8082"
     volumes:
@@ -69,9 +71,11 @@ services:
   elasticsearch:
     container_name: es-container
     image: docker.elastic.co/elasticsearch/elasticsearch:${ELASTICSEARCH_VERSION:-8.11.0}
+    platform: linux/amd64
     hostname: elasticsearch
     environment:
       ES_JAVA_OPTS: -Xms512m -Xmx1g
+      action.destructive_requires_name: false
     volumes:
       - ./elasticsearch/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml
       - ./elasticsearch/snapshots:/usr/share/elasticsearch/snapshots
@@ -81,11 +85,13 @@ services:
   opensearch:
     container_name: os-container
     image: opensearchproject/opensearch:${OPENSEARCH_VERSION:-2.11.1}
+    platform: linux/amd64
     hostname: opensearch
     environment:
       - discovery.type=single-node
       - plugins.security.disabled=true
       - OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m
+      - action.destructive_requires_name=false
     volumes:
       - ./opensearch/config/opensearch.yml:/usr/share/opensearch/config/opensearch.yml
       - ./opensearch/snapshots:/usr/share/opensearch/snapshots

compose.yml

@@ -37,6 +37,7 @@
     BulkTransactionMethod,
     Items,
 )
+from stac_fastapi.sfeos_helpers.database import return_date
 from stac_fastapi.types import stac as stac_types
 from stac_fastapi.types.conformance import BASE_CONFORMANCE_CLASSES
 from stac_fastapi.types.core import AsyncBaseCoreClient
@@ -324,10 +325,15 @@ async def item_collection(
             search=search, collection_ids=[collection_id]
         )
 
-        if datetime:
+        try:
+            datetime_search = return_date(datetime)
             search = self.database.apply_datetime_filter(
-                search=search, interval=datetime
+                search=search, datetime_search=datetime_search
             )
+        except (ValueError, TypeError) as e:
+            # Handle invalid interval formats if return_date fails
+            logger.error(f"Invalid interval format: {datetime}, error: {e}")
+            datetime_search = None
 
         if bbox:
             bbox = [float(x) for x in bbox]
@@ -342,6 +348,7 @@ async def item_collection(
             sort=None,
             token=token,
             collection_ids=[collection_id],
+            datetime_search=datetime_search,
         )
 
         items = [
@@ -500,10 +507,15 @@ async def post_search(
                 search=search, collection_ids=search_request.collections
             )
 
-        if search_request.datetime:
+        try:
+            datetime_search = return_date(search_request.datetime)
             search = self.database.apply_datetime_filter(
-                search=search, interval=search_request.datetime
+                search=search, datetime_search=datetime_search
             )
+        except (ValueError, TypeError) as e:
+            # Handle invalid interval formats if return_date fails
+            logger.error(f"Invalid interval format: {search_request.datetime}, error: {e}")
+            datetime_search = None
 
         if search_request.bbox:
             bbox = search_request.bbox
@@ -560,6 +572,7 @@ async def post_search(
             token=search_request.token,
             sort=sort,
             collection_ids=search_request.collections,
+            datetime_search=datetime_search,
         )
 
         fields = (

stac_fastapi/core/stac_fastapi/core/core.py

@@ -36,11 +36,8 @@
     get_queryables_mapping_shared,
     index_alias_by_collection_id,
     index_by_collection_id,
-    indices,
-    mk_actions,
     mk_item_id,
     populate_sort_shared,
-    return_date,
     validate_refresh,
 )
 from stac_fastapi.sfeos_helpers.database.utils import (
@@ -55,9 +52,14 @@
     ITEMS_INDEX_PREFIX,
     Geometry,
 )
+from stac_fastapi.sfeos_helpers.search_engine import (
+    BaseIndexInserter,
+    IndexInsertionFactory,
+    IndexSelectionStrategy,
+    IndexSelectorFactory,
+)
 from stac_fastapi.types.errors import ConflictError, NotFoundError
 from stac_fastapi.types.links import resolve_links
-from stac_fastapi.types.rfc3339 import DateTimeType
 from stac_fastapi.types.stac import Collection, Item
 
 logger = logging.getLogger(__name__)
@@ -135,6 +137,10 @@ class DatabaseLogic(BaseDatabaseLogic):
     sync_settings: SyncElasticsearchSettings = attr.ib(
         factory=SyncElasticsearchSettings
     )
+    async_index_selector: IndexSelectionStrategy = attr.ib(init=False)
+    sync_index_selector: IndexSelectionStrategy = attr.ib(init=False)
+    async_index_inserter: BaseIndexInserter = attr.ib(init=False)
+    sync_index_inserter: BaseIndexInserter = attr.ib(init=False)
 
     client = attr.ib(init=False)
     sync_client = attr.ib(init=False)
@@ -143,6 +149,18 @@ def __attrs_post_init__(self):
         """Initialize clients after the class is instantiated."""
         self.client = self.async_settings.create_client
         self.sync_client = self.sync_settings.create_client
+        self.async_index_inserter = (
+            IndexInsertionFactory.create_async_insertion_strategy(self.client)
+        )
+        self.sync_index_inserter = IndexInsertionFactory.create_sync_insertion_strategy(
+            self.sync_client
+        )
+        self.async_index_selector = IndexSelectorFactory.create_async_selector(
+            self.client
+        )
+        self.sync_index_selector = IndexSelectorFactory.create_sync_selector(
+            self.sync_client
+        )
 
     item_serializer: Type[ItemSerializer] = attr.ib(default=ItemSerializer)
     collection_serializer: Type[CollectionSerializer] = attr.ib(
@@ -256,30 +274,18 @@ def apply_collections_filter(search: Search, collection_ids: List[str]):
 
     @staticmethod
     def apply_datetime_filter(
-        search: Search, interval: Optional[Union[DateTimeType, str]]
+        search: Search, datetime_search: Dict[str, Optional[str]]
     ) -> Search:
         """Apply a filter to search on datetime, start_datetime, and end_datetime fields.
 
         Args:
             search: The search object to filter.
-            interval: Optional datetime interval to filter by. Can be:
-                - A single datetime string (e.g., "2023-01-01T12:00:00")
-                - A datetime range string (e.g., "2023-01-01/2023-12-31")
-                - A datetime object
-                - A tuple of (start_datetime, end_datetime)
+            datetime_search: Dict[str, Optional[str]]
 
         Returns:
             The filtered search object.
         """
-        if not interval:
-            return search
-
-        should = []
-        try:
-            datetime_search = return_date(interval)
-        except (ValueError, TypeError) as e:
-            # Handle invalid interval formats if return_date fails
-            logger.error(f"Invalid interval format: {interval}, error: {e}")
+        if not datetime_search:
             return search
 
         if "eq" in datetime_search:
@@ -489,6 +495,7 @@ async def execute_search(
         token: Optional[str],
         sort: Optional[Dict[str, Dict[str, str]]],
         collection_ids: Optional[List[str]],
+        datetime_search: Dict[str, Optional[str]],
         ignore_unavailable: bool = True,
     ) -> Tuple[Iterable[Dict[str, Any]], Optional[int], Optional[str]]:
         """Execute a search query with limit and other optional parameters.
@@ -499,6 +506,7 @@ async def execute_search(
             token (Optional[str]): The token used to return the next set of results.
             sort (Optional[Dict[str, Dict[str, str]]]): Specifies how the results should be sorted.
             collection_ids (Optional[List[str]]): The collection ids to search.
+            datetime_search (Dict[str, Optional[str]]): Datetime range used for index selection.
             ignore_unavailable (bool, optional): Whether to ignore unavailable collections. Defaults to True.
 
         Returns:
@@ -519,7 +527,9 @@ async def execute_search(
 
         query = search.query.to_dict() if search.query else None
 
-        index_param = indices(collection_ids)
+        index_param = await self.async_index_selector.select_indexes(
+            collection_ids, datetime_search
+        )
 
         max_result_window = MAX_LIMIT
 
@@ -583,6 +593,7 @@ async def aggregate(
         geometry_geohash_grid_precision: int,
         geometry_geotile_grid_precision: int,
         datetime_frequency_interval: str,
+        datetime_search,
         ignore_unavailable: Optional[bool] = True,
     ):
         """Return aggregations of STAC Items."""
@@ -618,7 +629,10 @@ def _fill_aggregation_parameters(name: str, agg: dict) -> dict:
             if k in aggregations
         }
 
-        index_param = indices(collection_ids)
+        index_param = await self.async_index_selector.select_indexes(
+            collection_ids, datetime_search
+        )
+
         search_task = asyncio.create_task(
             self.client.search(
                 index=index_param,
@@ -816,9 +830,12 @@ async def create_item(
             item=item, base_url=base_url, exist_ok=exist_ok
         )
 
+        target_index = await self.async_index_inserter.get_target_index(
+            collection_id, item
+        )
         # Index the item in the database
         await self.client.index(
-            index=index_alias_by_collection_id(collection_id),
+            index=target_index,
             id=mk_item_id(item_id, collection_id),
             document=item,
             refresh=refresh,
@@ -983,9 +1000,9 @@ async def delete_item(self, item_id: str, collection_id: str, **kwargs: Any):
 
         try:
             # Perform the delete operation
-            await self.client.delete(
+            await self.client.delete_by_query(
                 index=index_alias_by_collection_id(collection_id),
-                id=mk_item_id(item_id, collection_id),
+                body={"query": {"term": {"_id": mk_item_id(item_id, collection_id)}}},
                 refresh=refresh,
             )
         except ESNotFoundError:
@@ -1085,8 +1102,10 @@ async def create_collection(self, collection: Collection, **kwargs: Any):
             refresh=refresh,
         )
 
-        # Create the item index for the collection
-        await create_item_index(collection_id)
+        if self.async_index_inserter.should_create_collection_index():
+            await self.async_index_inserter.create_simple_index(
+                self.client, collection_id
+            )
 
     async def find_collection(self, collection_id: str) -> Collection:
         """Find and return a collection from the database.
@@ -1360,9 +1379,12 @@ async def bulk_async(
 
         # Perform the bulk insert
         raise_on_error = self.async_settings.raise_on_bulk_error
+        actions = await self.async_index_inserter.prepare_bulk_actions(
+            collection_id, processed_items
+        )
         success, errors = await helpers.async_bulk(
             self.client,
-            mk_actions(collection_id, processed_items),
+            actions,
             refresh=refresh,
             raise_on_error=raise_on_error,
         )
@@ -1426,9 +1448,12 @@ def bulk_sync(
 
         # Perform the bulk insert
         raise_on_error = self.sync_settings.raise_on_bulk_error
+        actions = self.sync_index_inserter.prepare_bulk_actions(
+            collection_id, processed_items
+        )
         success, errors = helpers.bulk(
             self.sync_client,
-            mk_actions(collection_id, processed_items),
+            actions,
             refresh=refresh,
             raise_on_error=raise_on_error,
         )