Add support for storing dictionaries directly in Elasticsearch

[strawgate]

Add support for storing the value as a flattened field in Elasticsearch. We should default to this being off but as an option that has some downsides (content matters more).

It should be a mode on the store that impacts how data is stored but we should always setup the same indices and get should work for either field, i think we use value right now so we'd probably want value_flattened

[claude]

Claude finished @strawgate's task —— View job

---

Investigation Complete ✓

I've thoroughly investigated issue #87 regarding adding support for storing dictionaries directly in Elasticsearch using flattened fields.

---

Recommendation

Implement an optional native_storage mode for ElasticsearchStore that stores values as native Elasticsearch flattened objects instead of JSON strings. This should:

1. Be disabled by default to maintain backward compatibility
2. Add a native_storage: bool parameter to the ElasticsearchStore constructor
3. Use dual-field mapping (value for JSON string, value_flattened for native objects) to support both modes
4. Handle both field types in reads to support migration scenarios

This approach is consistent with the discussions in issue #90 about native storage for document databases.

---

Findings

Current Implementation

Elasticsearch Store (elasticsearch/store.py:57-62):

"value": {
    "type": "keyword",
    "index": False,
    "doc_values": False,
    "ignore_above": 256,
}

Currently stores values as JSON strings:

• Line 80: managed_entry.to_json(include_metadata=False)
• Line 100: load_from_json(value_str)

Flattened Field Type Advantages

Based on research of Elasticsearch documentation and community resources:

Advantages:

• Stores entire object as structured data (not serialized string)
• Prevents "mapping explosion" when objects have many dynamic/unpredictable fields
• Enables basic querying on nested fields without complex mapping
• More efficient storage and retrieval than disabled keyword fields
• All values indexed as keywords (no analysis)

Disadvantages:

• No special support for numerics — all values treated as keywords
• Can't use non-case sensitive queries
• Values stored as-is without text analysis
• Content of the object matters more (affects index size and query performance)

Related Work

Issue #90 extensively discusses native storage for MongoDB and Elasticsearch:

• PR feat: Switch MongoDB and Elasticsearch to native document storage #104 attempted to switch Elasticsearch to flattened field type but was closed due to merge conflicts
• Discussion concluded that document databases (MongoDB, Elasticsearch) should use native storage
• Redis and DynamoDB correctly use JSON strings (no richer types available)

Dual-Field Approach Justification

The issue description suggests using both value (current) and value_flattened (new) fields. This enables:

1. Backward compatibility - existing data remains accessible
2. Gradual migration - users can migrate at their own pace
3. Read resilience - get() can check both fields
4. Opt-in adoption - new deployments can use native mode from day 1

Detailed Action Plan

1. Update ElasticsearchStore Constructor

File: key-value/key-value-aio/src/key_value/aio/stores/elasticsearch/store.py

Add native_storage parameter (around line 123):

def __init__(
    self,
    *,
    elasticsearch_client: AsyncElasticsearch | None = None,
    url: str | None = None,
    api_key: str | None = None,
    index_prefix: str,
    native_storage: bool = False,  # NEW: default to False for backward compat
    default_collection: str | None = None,
) -> None:
    # ... existing validation ...
    self._index_prefix = index_prefix
    self._native_storage = native_storage  # NEW: store the mode
    self._is_serverless = False
    super().__init__(default_collection=default_collection)

2. Update Index Mapping

File: key-value/key-value-aio/src/key_value/aio/stores/elasticsearch/store.py

Modify DEFAULT_MAPPING (lines 43-64) to include both fields:

DEFAULT_MAPPING = {
    "properties": {
        "created_at": {
            "type": "date",
        },
        "expires_at": {
            "type": "date",
        },
        "collection": {
            "type": "keyword",
        },
        "key": {
            "type": "keyword",
        },
        "value": {  # Keep for backward compatibility
            "type": "keyword",
            "index": False,
            "doc_values": False,
            "ignore_above": 256,
        },
        "value_flattened": {  # NEW: for native storage mode
            "type": "flattened",
        },
    },
}

3. Update Write Operations

Modify managed_entry_to_document function (lines 76-88):

def managed_entry_to_document(
    collection: str, 
    key: str, 
    managed_entry: ManagedEntry,
    native_storage: bool = False  # NEW parameter
) -> dict[str, Any]:
    document: dict[str, Any] = {
        "collection": collection,
        "key": key,
    }
    
    # Store in appropriate field based on mode
    if native_storage:
        document["value_flattened"] = dict(managed_entry.value)
    else:
        document["value"] = managed_entry.to_json(include_metadata=False)

    if managed_entry.created_at:
        document["created_at"] = managed_entry.created_at.isoformat()
    if managed_entry.expires_at:
        document["expires_at"] = managed_entry.expires_at.isoformat()

    return document

Update _put_managed_entry (line 268):

document: dict[str, Any] = managed_entry_to_document(
    collection=collection, 
    key=key, 
    managed_entry=managed_entry,
    native_storage=self._native_storage  # Pass the mode
)

Update _put_managed_entries (line 300):

document: dict[str, Any] = managed_entry_to_document(
    collection=collection, 
    key=key, 
    managed_entry=managed_entry,
    native_storage=self._native_storage  # Pass the mode
)

4. Update Read Operations

Update source_to_managed_entry function (lines 91-103):

def source_to_managed_entry(source: dict[str, Any]) -> ManagedEntry:
    # Try flattened field first, fall back to string field
    value_flattened = source.get("value_flattened")
    value_str = source.get("value")
    
    if value_flattened and isinstance(value_flattened, dict):
        # Native storage mode
        value = value_flattened
    elif value_str and isinstance(value_str, str):
        # Legacy JSON string mode
        value = load_from_json(value_str)
    else:
        msg = "Value field not found or invalid type"
        raise DeserializationError(msg)

    created_at: datetime | None = try_parse_datetime_str(value=source.get("created_at"))
    expires_at: datetime | None = try_parse_datetime_str(value=source.get("expires_at"))

    return ManagedEntry(
        value=value,
        created_at=created_at,
        expires_at=expires_at,
    )

Update _get_managed_entry (lines 208-218) - already uses source_to_managed_entry, no change needed

Update _get_managed_entries (line 248) - already uses source_to_managed_entry, no change needed

5. Add Tests

File: key-value/key-value-aio/tests/stores/elasticsearch/test_elasticsearch.py

Add new test class:

@pytest.mark.skipif(should_skip_docker_tests(), reason="Docker is not running")
class TestElasticsearchStoreNativeMode(ContextManagerStoreTestMixin, BaseStoreTests):
    """Test ElasticsearchStore with native_storage=True"""
    
    @pytest.fixture(autouse=True, scope="session", params=ELASTICSEARCH_VERSIONS_TO_TEST)
    async def setup_elasticsearch(self, request: pytest.FixtureRequest) -> AsyncGenerator[None, None]:
        # Same as TestElasticsearchStore
        ...
    
    @override
    @pytest.fixture
    async def store(self) -> AsyncGenerator[ElasticsearchStore, None]:
        async with get_elasticsearch_client() as es_client:
            indices = await es_client.options(ignore_status=404).indices.get(index="kv-store-native-test-*")
            for index in indices:
                _ = await es_client.options(ignore_status=404).indices.delete(index=index)
        async with ElasticsearchStore(
            url=ES_URL, 
            index_prefix="kv-store-native-test",
            native_storage=True  # Enable native mode
        ) as store:
            yield store
    
    async def test_value_stored_as_flattened_object(self, store: ElasticsearchStore, es_client: AsyncElasticsearch):
        """Verify values are stored as flattened objects, not JSON strings"""
        await store.put(collection="test", key="test_key", value={"name": "Alice", "age": 30})
        
        # Check raw Elasticsearch document
        index_name = store._sanitize_index_name(collection="test")
        doc_id = store._sanitize_document_id(key="test_key")
        
        response = await es_client.get(index=index_name, id=doc_id)
        source = response.body["_source"]
        
        # Should have value_flattened field as dict
        assert "value_flattened" in source
        assert isinstance(source["value_flattened"], dict)
        assert source["value_flattened"] == {"name": "Alice", "age": 30}
        
        # Should NOT have value field (JSON string)
        assert "value" not in source or source.get("value") is None
    
    async def test_migration_from_legacy_mode(self, store: ElasticsearchStore, es_client: AsyncElasticsearch):
        """Verify native mode can read legacy JSON string data"""
        # Manually insert a legacy document with JSON string value
        index_name = store._sanitize_index_name(collection="test")
        doc_id = store._sanitize_document_id(key="legacy_key")
        
        await es_client.index(
            index=index_name,
            id=doc_id,
            body={
                "collection": "test",
                "key": "legacy_key",
                "value": '{"legacy": "data"}',  # JSON string
            }
        )
        await es_client.indices.refresh(index=index_name)
        
        # Should be able to read it in native mode
        result = await store.get(collection="test", key="legacy_key")
        assert result == {"legacy": "data"}

6. Update Documentation

File: README.md (in the Elasticsearch section)

Add documentation about the native storage mode:

#### Native Storage Mode (Optional)

By default, ElasticsearchStore stores values as JSON strings in a `keyword` field. You can optionally enable native storage mode to store values as Elasticsearch `flattened` objects:

```python
store = ElasticsearchStore(
    url="http://localhost:9200",
    index_prefix="my-app",
    native_storage=True  # Enable native storage
)

Benefits of native storage:

• More efficient storage and retrieval
• Enables basic querying on nested fields
• Better aligned with Elasticsearch's document model

Trade-offs:

• All values indexed as keywords (no numeric type support)
• Slightly larger index size if objects have many fields
• Breaking change from default mode (requires re-indexing)

Migration: The store can read from both value (JSON string) and value_flattened (native object) fields, allowing gradual migration.

### 7. Sync Library Generation

After implementing changes in the async library, regenerate the sync library:
```bash
make codegen

Related Items

Related Issues and Pull Requests

Repository	Issue or PR	Relevance
strawgate/py-key-value	Review mongodb for inclusion as stable store #90	Extensive discussion about native storage for document databases (MongoDB and Elasticsearch). Concluded that document databases should use native storage instead of JSON strings.
strawgate/py-key-value	Switch MongoDB and Elasticsearch to native document storage #104	Previous attempt to implement flattened field type for Elasticsearch. Closed due to merge conflicts after repository refactor. Provides implementation reference.
strawgate/py-key-value	Store values directly for stores that natively support ttl #88	Related issue about using native features of stores instead of wrapper approaches.

Related Files

Repository	File	Relevance	Sections
strawgate/py-key-value	elasticsearch/store.py	Main Elasticsearch store implementation that needs modification	43-64 (mapping), 76-88 (serialization), 91-103 (deserialization)
strawgate/py-key-value	test_elasticsearch.py	Test file where new native storage tests should be added	All
strawgate/py-key-value	mongodb/store.py	MongoDB store - reference implementation for native document storage pattern	36-69
strawgate/py-key-value	AGENTS.md	Project documentation for understanding development workflow	All

Related Webpages

Name	URL	Relevance
Elasticsearch Flattened Field Type	https://www.elastic.co/guide/en/elasticsearch/reference/current/flattened.html	Official Elasticsearch documentation for the flattened field type
Flattened Datatype Mappings Tutorial	https://coralogix.com/blog/flattened-datatype-mappings-elasticsearch-tutorial/	Tutorial explaining flattened field type use cases and implementation
Elasticsearch Flattened Type Design	elastic/elasticsearch#33003	Original GitHub issue discussing the design and implementation of the flattened field type
Indexing Arbitrary JSON in Elasticsearch	https://smnh.me/indexing-and-searching-arbitrary-json-data-using-elasticsearch	Article comparing approaches for storing JSON data in Elasticsearch

---

Problems Encountered During Investigation

None - all investigation tools and resources were accessible and provided comprehensive information.

---

[strawgate]

@claude make a PR please

[claude]

Claude finished @strawgate's task —— View job • claude/issue-87-20251027-0227 • Create PR ➔

---

Implementation Complete ✓

I've successfully implemented native storage mode for Elasticsearch using flattened fields as requested in issue #87.

Summary of Changes

Key Implementation:

• Added optional native_storage parameter to ElasticsearchStore constructor (defaults to False for backward compatibility)
• Updated Elasticsearch mapping to include value_flattened field using the flattened type
• Modified managed_entry_to_document() to store values in the appropriate field based on the mode
• Updated source_to_managed_entry() to read from both fields, enabling migration from legacy JSON string storage
• Added comprehensive tests including native storage verification and migration testing

Files Modified:

• key-value/key-value-aio/src/key_value/aio/stores/elasticsearch/store.py - Core implementation
• key-value/key-value-aio/tests/stores/elasticsearch/test_elasticsearch.py - Test coverage

Quality Checks:

• ✅ Linting passed (make lint)
• ✅ Type checking passed for async package (make typecheck)
• ✅ Code follows repository conventions

Benefits of This Implementation

1. Backward Compatible - Defaults to legacy mode, new deployments can opt-in
2. Migration Support - Can read from both field types during transition
3. More Efficient - Native storage is better aligned with Elasticsearch's document model
4. Enables Querying - Basic querying on nested fields becomes possible

Known Limitation

⚠️ Sync library not regenerated: The CI environment only has Python 3.12, but the codegen script requires Python 3.10 (per AGENTS.md). The sync library will need to be regenerated when this PR is merged in an environment with Python 3.10.

The sync library generation can be run with: make codegen

Create Pull Request

---

Branch: claude/issue-87-20251027-0227