[ENH] recognize and flush new metadata keys to schema on local compaction

[jairad26]

Description of changes

Summarize the changes made by this PR.

• Improvements & Bug fixes
  • This PR adds support for discovering new metadata keys during log flushing for local chroma
• New functionality
  • ...

Test plan

added e2e test support to run for distributed, local, and single node to ensure metadata discovery and related tests run
How are these changes tested?

• [x ] Tests pass locally with pytest for python, yarn test for js, cargo test for rust

Migration plan

Are there any migrations, or any forwards/backwards compatibility changes needed in order to make sure this change deploys reliably?

Observability plan

What is the plan to instrument and monitor this change?

Documentation Changes

Are all docstrings for user-facing APIs updated if required? Do we need to make documentation changes in the docs section?

[jairad26]

• [ENH] recognize and flush new metadata keys to schema on local compaction #5728 👈 (View in Graphite)
• [ENH] Add local support for schema #5714
• [ENH] Embed query strings in search api #5599
• [BUG] Default create path with no config or schema does not populate default ef in schema #5726
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[github-actions]

Reviewer Checklist

Please leverage this checklist to ensure your code review is thorough before approving

Testing, Bugs, Errors, Logs, Documentation

• Can you think of any use case in which the code does not behave as intended? Have they been tested?
• Can you think of any inputs or external events that could break the code? Is user input validated and safe? Have they been tested?
• If appropriate, are there adequate property based tests?
• If appropriate, are there adequate unit tests?
• Should any logging, debugging, tracing information be added or removed?
• Are error messages user-friendly?
• Have all documentation changes needed been made?
• Have all non-obvious changes been commented?

System Compatibility

• Are there any potential impacts on other parts of the system or backward compatibility?
• Does this change intersect with any items on our roadmap, and if so, is there a plan for fitting them together?

Quality

• Is this code of a unexpectedly high quality (Readability, Modularity, Intuitiveness)

[propel-code-bot]

Dynamic Schema Discovery & Persistence During Local Compaction

Adds end-to-end support for discovering previously unseen metadata keys when processing local log compaction. rust/segment now detects new keys while flushing logs, mutates an in-memory Schema, and returns an ApplyLogsOutcome that signals if the schema changed. rust/log persists that new schema (when the collection already had a schema) and keeps metadata/segment updates in sync. The PR also hardens schema rules (blocks keys starting with CHROMA_), refactors error types, bumps deps (adds serde_json), and updates the large Python E2E suite to cope with conditional SPANN behaviour.

Key Changes

• Extended SqliteMetadataWriter.apply_logs() signature: new schema: Option<Schema> param and returns ApplyLogsOutcome { schema_update }
• Introduced ApplyLogsOutcome struct and helper ensure_schema_for_update_value() to auto-populate Schema on new key discovery while filtering system keys (CHROMA_KEY)
• Added update_collection_schema() method in SqliteMetadataWriter for in-place schema persistence (includes JSON serialization error handling)
• Updated LocalCompactionManager logic: tracks whether a schema was already persisted, writes logs, updates schema inside same backfill flow
• Strengthened Schema.ensure_key_from_metadata() guard against special keys; added CHROMA_KEY constant
• Cargo: added serde_json to chroma-segment
• Large test-suite rewrite (chromadb/test/api/test_schema_e2e.py) to adapt to new conditional schema persistence paths; removed SPANN-skips and performs runtime checks

Affected Areas

• rust/segment/src/sqlite_metadata.rs
• rust/log/src/local_compaction_manager.rs
• rust/types/src/collection_schema.rs
• Python e2e schema tests
• rust/segment/Cargo.toml dependency list

This summary was automatically generated by @propel-code-bot

[propel-code-bot]

[BestPractice]

Error handling gap: MetadataValue::try_from(value) failure is silently ignored in schema discovery logic. If the conversion fails, the key won't be added to the schema even though it might be valid metadata that should be discoverable.

if let Ok(metadata_value) = MetadataValue::try_from(value) {
    return schema_mut
        .ensure_key_from_metadata(key, metadata_value.value_type());
} else {
    // Log the conversion failure for debugging - using standard tracing pattern
    tracing::warn!(key = %key, "Failed to convert metadata value during schema discovery");
}

This could lead to inconsistent schema discovery where some valid metadata keys are missed due to conversion issues. The enhanced logging follows tracing best practices with structured fields.

Context for Agents

[**BestPractice**]

Error handling gap: `MetadataValue::try_from(value)` failure is silently ignored in schema discovery logic. If the conversion fails, the key won't be added to the schema even though it might be valid metadata that should be discoverable.

```rust
if let Ok(metadata_value) = MetadataValue::try_from(value) {
    return schema_mut
        .ensure_key_from_metadata(key, metadata_value.value_type());
} else {
    // Log the conversion failure for debugging - using standard tracing pattern
    tracing::warn!(key = %key, "Failed to convert metadata value during schema discovery");
}
```

This could lead to inconsistent schema discovery where some valid metadata keys are missed due to conversion issues. The enhanced logging follows tracing best practices with structured fields.

File: rust/segment/src/sqlite_metadata.rs
Line: 319

[propel-code-bot]

[BestPractice]

The apply_logs method signature change adds a new schema parameter, but in the test cases you're passing the collection's schema from test_data. However, the production code in local_compaction_manager.rs passes either the collection's schema or None based on schema_previously_persisted. This inconsistency could lead to different behavior between tests and production.

Consider aligning the test calls with the production logic:

runtime.block_on(sqlite_seg_writer.apply_logs(
    data,
    metadata_seg_id,
    if schema_previously_persisted {
        test_data.collection_and_segments.collection.schema.clone()
    } else {
        None
    },
    &mut *tx,
))

Context for Agents

[**BestPractice**]

The `apply_logs` method signature change adds a new schema parameter, but in the test cases you're passing the collection's schema from `test_data`. However, the production code in `local_compaction_manager.rs` passes either the collection's schema or None based on `schema_previously_persisted`. This inconsistency could lead to different behavior between tests and production.

Consider aligning the test calls with the production logic:
```rust
runtime.block_on(sqlite_seg_writer.apply_logs(
    data,
    metadata_seg_id,
    if schema_previously_persisted {
        test_data.collection_and_segments.collection.schema.clone()
    } else {
        None
    },
    &mut *tx,
))
```

File: rust/segment/src/sqlite_metadata.rs
Line: 999

[propel-code-bot]

[BestPractice]

Missing error handling for schema update: If update_collection_schema() fails (lines 228-234), the error is converted to MetadataApplyLogsFailed which loses important context about what specifically went wrong during schema persistence. This could make debugging schema-related issues very difficult.

// Consider more specific error handling
metadata_writer
    .update_collection_schema(
        collection_and_segments.collection.collection_id,
        &updated_schema,
    )
    .await
    .map_err(|e| CompactionManagerError::SchemaUpdateFailed(e))?;

Context for Agents

[**BestPractice**]

Missing error handling for schema update: If `update_collection_schema()` fails (lines 228-234), the error is converted to `MetadataApplyLogsFailed` which loses important context about what specifically went wrong during schema persistence. This could make debugging schema-related issues very difficult.

```rust
// Consider more specific error handling
metadata_writer
    .update_collection_schema(
        collection_and_segments.collection.collection_id,
        &updated_schema,
    )
    .await
    .map_err(|e| CompactionManagerError::SchemaUpdateFailed(e))?;
```

File: rust/log/src/local_compaction_manager.rs
Line: 234

[propel-code-bot]

[PerformanceOptimization]

Potential performance issue: The ensure_schema_for_update_value function is called for every key-value pair in every metadata update within the log processing loop. For large batches with many metadata fields, this could create significant overhead.

Consider batching schema updates or using a HashSet to track which keys have already been processed to avoid redundant schema checks. For better performance, pre-size the HashSet if you can estimate the number of unique keys:

// Pre-size for better performance
let mut processed_keys = HashSet::with_capacity(estimated_unique_keys);
for (key, value) in meta.iter() {
    if !processed_keys.contains(key) {
        if Self::ensure_schema_for_update_value(&mut schema, key, value) {
            schema_modified = true;
            processed_keys.insert(key.clone());
        }
    }
}

Note: HashSet operations are O(1) average case, but the overhead of hashing and set operations may not be beneficial for very small key sets. Consider the trade-off based on typical batch sizes in your use case.

Context for Agents

[**PerformanceOptimization**]

Potential performance issue: The `ensure_schema_for_update_value` function is called for every key-value pair in every metadata update within the log processing loop. For large batches with many metadata fields, this could create significant overhead.

Consider batching schema updates or using a `HashSet` to track which keys have already been processed to avoid redundant schema checks. For better performance, pre-size the HashSet if you can estimate the number of unique keys:

```rust
// Pre-size for better performance
let mut processed_keys = HashSet::with_capacity(estimated_unique_keys);
for (key, value) in meta.iter() {
    if !processed_keys.contains(key) {
        if Self::ensure_schema_for_update_value(&mut schema, key, value) {
            schema_modified = true;
            processed_keys.insert(key.clone());
        }
    }
}
```

Note: HashSet operations are O(1) average case, but the overhead of hashing and set operations may not be beneficial for very small key sets. Consider the trade-off based on typical batch sizes in your use case.

File: rust/segment/src/sqlite_metadata.rs
Line: 386

[propel-code-bot]

[CriticalError]

The changed apply_logs method signature adds a schema parameter but doesn't handle errors when the schema is None and metadata processing tries to access it. In ensure_schema_for_update_value, if schema.as_mut() returns None, the function silently returns false, but the caller expects schema discovery to work.

Consider adding error handling or initialization:

let mut schema = schema.unwrap_or_else(|| Schema::new());

This ensures schema discovery can always function properly during log flushing.

Context for Agents

[**CriticalError**]

The changed `apply_logs` method signature adds a `schema` parameter but doesn't handle errors when the schema is `None` and metadata processing tries to access it. In `ensure_schema_for_update_value`, if `schema.as_mut()` returns `None`, the function silently returns `false`, but the caller expects schema discovery to work.

Consider adding error handling or initialization:
```rust
let mut schema = schema.unwrap_or_else(|| Schema::new());
```

This ensures schema discovery can always function properly during log flushing.

File: rust/segment/src/sqlite_metadata.rs
Line: 345

[propel-code-bot]

[CriticalError]

Potential Race Condition in Schema State Check: The schema_previously_persisted boolean is determined before schema reconciliation, but the schema could be modified during reconcile_schema_with_config(). This creates a potential inconsistency where:

1. Schema starts as None (schema_previously_persisted = false)
2. reconcile_schema_with_config() creates/modifies schema
3. Later logic uses the stale schema_previously_persisted value

This could lead to schema updates being skipped when they should be persisted.

// Current approach - schema state checked before reconciliation
let schema_previously_persisted = collection_and_segments.collection.schema.is_some();
collection_and_segments
    .collection
    .reconcile_schema_with_config(KnnIndex::Hnsw)?;

// Later usage may be inconsistent with actual schema state
if schema_previously_persisted { /* ... */ }

Consider checking schema state after reconciliation or using a different approach to determine when schema updates should be persisted.

Context for Agents

[**CriticalError**]

**Potential Race Condition in Schema State Check**: The `schema_previously_persisted` boolean is determined before schema reconciliation, but the schema could be modified during `reconcile_schema_with_config()`. This creates a potential inconsistency where:

1. Schema starts as `None` (schema_previously_persisted = false)
2. `reconcile_schema_with_config()` creates/modifies schema
3. Later logic uses the stale `schema_previously_persisted` value

This could lead to schema updates being skipped when they should be persisted.

```rust
// Current approach - schema state checked before reconciliation
let schema_previously_persisted = collection_and_segments.collection.schema.is_some();
collection_and_segments
    .collection
    .reconcile_schema_with_config(KnnIndex::Hnsw)?;

// Later usage may be inconsistent with actual schema state
if schema_previously_persisted { /* ... */ }
```

Consider checking schema state after reconciliation or using a different approach to determine when schema updates should be persisted.

File: rust/log/src/local_compaction_manager.rs
Line: 143

[jairad26]

this is intentional. if theres no schema_str found in the db, we shouldnt use&modify it in compaction