[ENH] [MEGAPR] Blockfile refactor, compactor job, record segment (#2052)

## Description of changes

*Summarize the changes made by this PR.*
 - Improvements & Bug fixes
	 - /
 - New functionality
- This PR contains a very large refactor the blockfile code to make it
strongly typed and zero-copy. It is a lot of prototype-quality code. I
am getting it merged to test on staging but subsequently will clean this
all up. We should not normalize this and its an extreme deviance from
the norm.

## Test plan
*How are these changes tested?*
- [x] Tests pass locally with `pytest` for python, `yarn test` for js,
`cargo test` for rust

## Documentation Changes
None

.pre-commit-config.yaml

@@ -1,4 +1,4 @@
-exclude: 'chromadb/proto/(chroma_pb2|coordinator_pb2|logservice_pb2)\.(py|pyi|py_grpc\.py)' # Generated files
+exclude: 'chromadb/proto/(chroma_pb2|coordinator_pb2|logservice_pb2|chroma_pb2_grpc|coordinator_pb2_grpc|logservice_pb2_grpc)\.(py|pyi)' # Generated files
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.5.0

chromadb/api/segment.py

@@ -692,9 +692,8 @@ def _query(
         for embedding in query_embeddings:
             self._validate_dimension(coll, len(embedding), update=False)
 
-        metadata_reader = self._manager.get_segment(collection_id, MetadataReader)
-
         if where or where_document:
+            metadata_reader = self._manager.get_segment(collection_id, MetadataReader)
             records = metadata_reader.get_metadata(
                 where=where, where_document=where_document
             )
@@ -729,6 +728,7 @@ def _query(
             all_ids: Set[str] = set()
             for id_list in ids:
                 all_ids.update(id_list)
+            metadata_reader = self._manager.get_segment(collection_id, MetadataReader)
             records = metadata_reader.get_metadata(ids=list(all_ids))
             metadata_by_id = {r["id"]: r["metadata"] for r in records}
             for id_list in ids:

chromadb/proto/convert.py

@@ -21,8 +21,6 @@
 
 
 # TODO: Unit tests for this file, handling optional states etc
-
-
 def to_proto_vector(vector: Vector, encoding: ScalarEncoding) -> proto.Vector:
     if encoding == ScalarEncoding.FLOAT32:
         as_bytes = array.array("f", vector).tobytes()
@@ -158,6 +156,8 @@ def from_proto_segment_scope(segment_scope: proto.SegmentScope) -> SegmentScope:
         return SegmentScope.VECTOR
     elif segment_scope == proto.SegmentScope.METADATA:
         return SegmentScope.METADATA
+    elif segment_scope == proto.SegmentScope.RECORD:
+        return SegmentScope.RECORD
     else:
         raise RuntimeError(f"Unknown segment scope {segment_scope}")
 
@@ -167,6 +167,8 @@ def to_proto_segment_scope(segment_scope: SegmentScope) -> proto.SegmentScope:
         return proto.SegmentScope.VECTOR
     elif segment_scope == SegmentScope.METADATA:
         return proto.SegmentScope.METADATA
+    elif segment_scope == SegmentScope.RECORD:
+        return proto.SegmentScope.RECORD
     else:
         raise RuntimeError(f"Unknown segment scope {segment_scope}")