[CLN]: Revert 5742 (#5782)

## Description of changes

_Summarize the changes made by this PR._

- Improvements & Bug fixes
- Reverts https://github.com/chroma-core/chroma/pull/5742/files. We are
cleaning up the schema code a little bit to have separate reconcile
logic for read and write paths so this is in direction of that.
- New functionality
  - ...

## Test plan

_How are these changes tested?_
- [x] Tests pass locally with `pytest` for python, `yarn test` for js,
`cargo test` for rust

## Migration plan
None

## Observability plan
None

## Documentation Changes
None

Author: sanketkedia

rust/segment/src/distributed_hnsw.rs

@@ -97,12 +97,9 @@ impl DistributedHNSWSegmentWriter {
     ) -> Result<Box<DistributedHNSWSegmentWriter>, Box<DistributedHNSWSegmentFromSegmentError>>
     {
         let hnsw_configuration = collection
-            .schema
-            .as_ref()
-            .map(|schema| schema.get_internal_hnsw_config_with_legacy_fallback(segment))
-            .transpose()
+            .config
+            .get_hnsw_config_with_legacy_fallback(segment)
             .map_err(DistributedHNSWSegmentFromSegmentError::InvalidHnswConfiguration)?
-            .flatten()
             .ok_or(DistributedHNSWSegmentFromSegmentError::MissingHnswConfiguration)?;
 
         // TODO: this is hacky, we use the presence of files to determine if we need to load or create the index
@@ -317,12 +314,9 @@ impl DistributedHNSWSegmentReader {
     ) -> Result<Box<DistributedHNSWSegmentReader>, Box<DistributedHNSWSegmentFromSegmentError>>
     {
         let hnsw_configuration = collection
-            .schema
-            .as_ref()
-            .map(|schema| schema.get_internal_hnsw_config_with_legacy_fallback(segment))
-            .transpose()
+            .config
+            .get_hnsw_config_with_legacy_fallback(segment)
             .map_err(DistributedHNSWSegmentFromSegmentError::InvalidHnswConfiguration)?
-            .flatten()
             .ok_or(DistributedHNSWSegmentFromSegmentError::MissingHnswConfiguration)?;
 
         // TODO: this is hacky, we use the presence of files to determine if we need to load or create the index

rust/segment/src/distributed_spann.rs

@@ -15,6 +15,8 @@ use chroma_index::spann::types::{
 use chroma_index::IndexUuid;
 use chroma_index::{hnsw_provider::HnswIndexProvider, spann::types::SpannIndexWriter};
 use chroma_types::Collection;
+use chroma_types::KnnIndex;
+use chroma_types::Schema;
 use chroma_types::SchemaError;
 use chroma_types::SegmentUuid;
 use chroma_types::HNSW_PATH;
@@ -110,13 +112,14 @@ impl SpannSegmentWriter {
             return Err(SpannSegmentWriterError::InvalidArgument);
         }
 
-        let schema = collection.schema.as_ref().ok_or_else(|| {
-            SpannSegmentWriterError::InvalidSchema(SchemaError::InvalidSchema {
-                reason: "Schema is None".to_string(),
-            })
-        })?;
+        let reconciled_schema = Schema::reconcile_schema_and_config(
+            collection.schema.as_ref(),
+            Some(&collection.config),
+            KnnIndex::Spann,
+        )
+        .map_err(SpannSegmentWriterError::InvalidSchema)?;
 
-        let params = schema
+        let params = reconciled_schema
             .get_internal_spann_config()
             .ok_or(SpannSegmentWriterError::MissingSpannConfiguration)?;
 

rust/worker/src/execution/orchestration/compact.rs

@@ -29,7 +29,7 @@ use chroma_system::{
     OrchestratorContext, PanicError, TaskError, TaskMessage, TaskResult,
 };
 use chroma_types::{
-    AttachedFunction, AttachedFunctionUuid, Chunk, Collection, CollectionUuid, KnnIndex, LogRecord,
+    AttachedFunction, AttachedFunctionUuid, Chunk, Collection, CollectionUuid, LogRecord,
     NonceUuid, Schema, SchemaError, Segment, SegmentFlushInfo, SegmentType, SegmentUuid,
 };
 use opentelemetry::trace::TraceContextExt;
@@ -1124,7 +1124,7 @@ impl Handler<TaskResult<GetCollectionAndSegmentsOutput, GetCollectionAndSegments
         }
 
         // Store output collection
-        let mut output_collection = output.output.collection.clone();
+        let output_collection = output.output.collection.clone();
         if self
             .output_collection
             .set(output_collection.clone())
@@ -1295,23 +1295,6 @@ impl Handler<TaskResult<GetCollectionAndSegmentsOutput, GetCollectionAndSegments
             Some(writer) => writer,
             None => return,
         };
-        if self
-            .ok_or_terminate(
-                match vector_segment.r#type {
-                    SegmentType::Spann => output_collection
-                        .reconcile_schema_with_config(KnnIndex::Spann)
-                        .map_err(CompactionError::from),
-                    _ => output_collection
-                        .reconcile_schema_with_config(KnnIndex::Hnsw)
-                        .map_err(CompactionError::from),
-                },
-                ctx,
-            )
-            .await
-            .is_none()
-        {
-            return;
-        }
         let (hnsw_index_uuid, vector_writer, is_vector_segment_spann) = match vector_segment.r#type
         {
             SegmentType::Spann => match self