Skip to content

Commit 90ac80a

Browse files
codethewebcodetheweb
committed
[ENH]: add batch get version file paths method to Sysdb
1 parent bab5b68 commit 90ac80a

File tree

12 files changed

+219
-11
lines changed

12 files changed

+219
-11
lines changed

‎go/pkg/sysdb/coordinator/coordinator.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -263,6 +263,10 @@ func (s *Coordinator) DeleteCollectionVersion(ctx context.Context, req *coordina
263263
return s.catalog.DeleteCollectionVersion(ctx, req)
264264
}
265265

266+
func (s *Coordinator) BatchGetCollectionVersionFilePaths(ctx context.Context, req *coordinatorpb.BatchGetCollectionVersionFilePathsRequest) (*coordinatorpb.BatchGetCollectionVersionFilePathsResponse, error) {
267+
return s.catalog.BatchGetCollectionVersionFilePaths(ctx, req.CollectionIds)
268+
}
269+
266270
// SetDeleteMode sets the delete mode for testing
267271
func (c *Coordinator) SetDeleteMode(mode DeleteMode) {
268272
c.deleteMode = mode

‎go/pkg/sysdb/coordinator/coordinator_test.go

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1657,6 +1657,37 @@ func (suite *APIsTestSuite) TestForkCollection() {
16571657
suite.True(exists, "Lineage file should exist in S3")
16581658
}
16591659

1660+
func (suite *APIsTestSuite) TestBatchGetCollectionVersionFilePaths() {
1661+
ctx := context.Background()
1662+
1663+
// Create a new collection
1664+
newCollection := &model.CreateCollection{
1665+
ID: types.NewUniqueID(),
1666+
Name: "test_batch_get_collection_version_file_paths",
1667+
TenantID: suite.tenantName,
1668+
DatabaseName: suite.databaseName,
1669+
}
1670+
1671+
newSegments := []*model.Segment{}
1672+
1673+
// Create the collection
1674+
suite.coordinator.catalog.versionFileEnabled = true
1675+
_, _, err := suite.coordinator.CreateCollectionAndSegments(ctx, newCollection, newSegments)
1676+
suite.NoError(err)
1677+
1678+
// Get the version file paths for the collection
1679+
versionFilePaths, err := suite.coordinator.BatchGetCollectionVersionFilePaths(ctx, &coordinatorpb.BatchGetCollectionVersionFilePathsRequest{
1680+
CollectionIds: []string{newCollection.ID.String()},
1681+
})
1682+
suite.NoError(err)
1683+
suite.Len(versionFilePaths.CollectionIdToVersionFilePath, 1)
1684+
1685+
// Verify version file exists in S3
1686+
exists, err := suite.s3MetaStore.HasObjectWithPrefix(ctx, versionFilePaths.CollectionIdToVersionFilePath[newCollection.ID.String()])
1687+
suite.NoError(err)
1688+
suite.True(exists, "Version file should exist in S3")
1689+
}
1690+
16601691
func (suite *APIsTestSuite) TestCountForks() {
16611692
ctx := context.Background()
16621693

‎go/pkg/sysdb/coordinator/table_catalog.go

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2015,6 +2015,20 @@ func (tc *Catalog) DeleteCollectionVersion(ctx context.Context, req *coordinator
20152015
return &result, nil
20162016
}
20172017

2018+
func (tc *Catalog) BatchGetCollectionVersionFilePaths(ctx context.Context, collectionIds []string) (*coordinatorpb.BatchGetCollectionVersionFilePathsResponse, error) {
2019+
result := coordinatorpb.BatchGetCollectionVersionFilePathsResponse{
2020+
CollectionIdToVersionFilePath: make(map[string]string),
2021+
}
2022+
2023+
paths, err := tc.metaDomain.CollectionDb(ctx).BatchGetCollectionVersionFilePaths(collectionIds)
2024+
if err != nil {
2025+
return nil, err
2026+
}
2027+
result.CollectionIdToVersionFilePath = paths
2028+
2029+
return &result, nil
2030+
}
2031+
20182032
func (tc *Catalog) GetVersionFileNamesForCollection(ctx context.Context, tenantID string, collectionID string) (string, error) {
20192033
collectionIDPtr := &collectionID
20202034
collectionEntry, err := tc.metaDomain.CollectionDb(ctx).GetCollectionEntry(collectionIDPtr, nil)

‎go/pkg/sysdb/coordinator/table_catalog_test.go

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -308,8 +308,8 @@ func TestCatalog_FlushCollectionCompactionForVersionedCollection(t *testing.T) {
308308
LogPosition: 10,
309309
Version: int32(currentVersion),
310310
VersionFileName: "version_1.pb",
311-
RootCollectionId: "",
312-
LineageFileName: "",
311+
RootCollectionId: nil,
312+
LineageFileName: nil,
313313
TotalRecordsPostCompaction: 10,
314314
SizeBytesPostCompaction: 100,
315315
LastCompactionTimeSecs: 0,
@@ -512,8 +512,8 @@ func TestCatalog_FlushCollectionCompactionForVersionedCollectionWithEmptyFilePat
512512
LogPosition: 10,
513513
Version: int32(currentVersion),
514514
VersionFileName: "version_1.pb",
515-
RootCollectionId: "",
516-
LineageFileName: "",
515+
RootCollectionId: nil,
516+
LineageFileName: nil,
517517
TotalRecordsPostCompaction: 10,
518518
SizeBytesPostCompaction: 100,
519519
LastCompactionTimeSecs: 0,

‎go/pkg/sysdb/grpc/collection_service.go

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -545,3 +545,11 @@ func (s *Server) DeleteCollectionVersion(ctx context.Context, req *coordinatorpb
545545
}
546546
return res, nil
547547
}
548+
549+
func (s *Server) BatchGetCollectionVersionFilePaths(ctx context.Context, req *coordinatorpb.BatchGetCollectionVersionFilePathsRequest) (*coordinatorpb.BatchGetCollectionVersionFilePathsResponse, error) {
550+
res, err := s.coordinator.BatchGetCollectionVersionFilePaths(ctx, req)
551+
if err != nil {
552+
return nil, grpcutils.BuildInternalGrpcError(err.Error())
553+
}
554+
return res, nil
555+
}

‎go/pkg/sysdb/metastore/db/dao/collection.go

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -531,3 +531,20 @@ func (s *collectionDb) UpdateCollectionLineageFilePath(collectionID string, curr
531531
}).Error
532532

533533
}
534+
535+
func (s *collectionDb) BatchGetCollectionVersionFilePaths(collectionIDs []string) (map[string]string, error) {
536+
var collections []dbmodel.Collection
537+
err := s.read_db.Model(&dbmodel.Collection{}).
538+
Select("id, version_file_name").
539+
Where("id IN ?", collectionIDs).
540+
Find(&collections).Error
541+
if err != nil {
542+
return nil, err
543+
}
544+
545+
result := make(map[string]string)
546+
for _, collection := range collections {
547+
result[collection.ID] = collection.VersionFileName
548+
}
549+
return result, nil
550+
}

‎go/pkg/sysdb/metastore/db/dbmodel/collection.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,4 +68,5 @@ type ICollectionDb interface {
6868
UpdateVersionRelatedFields(collectionID, existingVersionFileName, newVersionFileName string, oldestVersionTs *time.Time, numActiveVersions *int) (int64, error)
6969
LockCollection(collectionID string) error
7070
UpdateCollectionLineageFilePath(collectionID string, currentLineageFilePath *string, newLineageFilePath string) error
71+
BatchGetCollectionVersionFilePaths(collectionIDs []string) (map[string]string, error)
7172
}

‎go/pkg/sysdb/metastore/db/dbmodel/mocks/ICollectionDb.go

Lines changed: 30 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎idl/chromadb/proto/coordinator.proto

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -467,6 +467,14 @@ message DeleteCollectionVersionResponse {
467467
map<string, bool> collection_id_to_success = 1;
468468
}
469469

470+
message BatchGetCollectionVersionFilePathsRequest {
471+
repeated string collection_ids = 1;
472+
}
473+
474+
message BatchGetCollectionVersionFilePathsResponse {
475+
map<string, string> collection_id_to_version_file_path = 1;
476+
}
477+
470478
service SysDB {
471479
rpc CreateDatabase(CreateDatabaseRequest) returns (CreateDatabaseResponse) {}
472480
rpc GetDatabase(GetDatabaseRequest) returns (GetDatabaseResponse) {}
@@ -498,4 +506,5 @@ service SysDB {
498506
rpc ListCollectionsToGc(ListCollectionsToGcRequest) returns (ListCollectionsToGcResponse) {}
499507
rpc MarkVersionForDeletion(MarkVersionForDeletionRequest) returns (MarkVersionForDeletionResponse) {}
500508
rpc DeleteCollectionVersion(DeleteCollectionVersionRequest) returns (DeleteCollectionVersionResponse) {}
509+
rpc BatchGetCollectionVersionFilePaths(BatchGetCollectionVersionFilePathsRequest) returns (BatchGetCollectionVersionFilePathsResponse) {}
501510
}

‎rust/sysdb/src/sysdb.rs

Lines changed: 47 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,9 @@ use chroma_types::{
1919
UpdateCollectionConfiguration, UpdateCollectionError, VectorIndexConfiguration,
2020
};
2121
use chroma_types::{
22-
Collection, CollectionConversionError, CollectionUuid, CountForksError,
23-
FlushCompactionResponse, FlushCompactionResponseConversionError, ForkCollectionError, Segment,
24-
SegmentConversionError, SegmentScope, Tenant,
22+
BatchGetCollectionVersionFilePathsError, Collection, CollectionConversionError, CollectionUuid,
23+
CountForksError, FlushCompactionResponse, FlushCompactionResponseConversionError,
24+
ForkCollectionError, Segment, SegmentConversionError, SegmentScope, Tenant,
2525
};
2626
use std::collections::HashMap;
2727
use std::fmt::Debug;
@@ -400,6 +400,23 @@ impl SysDb {
400400
}
401401
}
402402

403+
pub async fn batch_get_collection_version_file_paths(
404+
&mut self,
405+
collection_ids: Vec<CollectionUuid>,
406+
) -> Result<HashMap<CollectionUuid, String>, BatchGetCollectionVersionFilePathsError> {
407+
match self {
408+
SysDb::Grpc(grpc) => {
409+
grpc.batch_get_collection_version_file_paths(collection_ids)
410+
.await
411+
}
412+
SysDb::Sqlite(_) => todo!(),
413+
SysDb::Test(test) => {
414+
test.batch_get_collection_version_file_paths(collection_ids)
415+
.await
416+
}
417+
}
418+
}
419+
403420
pub async fn get_last_compaction_time(
404421
&mut self,
405422
tanant_ids: Vec<String>,
@@ -1113,6 +1130,33 @@ impl GrpcSysDb {
11131130
})
11141131
}
11151132

1133+
async fn batch_get_collection_version_file_paths(
1134+
&mut self,
1135+
collection_ids: Vec<CollectionUuid>,
1136+
) -> Result<HashMap<CollectionUuid, String>, BatchGetCollectionVersionFilePathsError> {
1137+
let res = self
1138+
.client
1139+
.batch_get_collection_version_file_paths(
1140+
chroma_proto::BatchGetCollectionVersionFilePathsRequest {
1141+
collection_ids: collection_ids
1142+
.into_iter()
1143+
.map(|id| id.0.to_string())
1144+
.collect(),
1145+
},
1146+
)
1147+
.await?;
1148+
let collection_id_to_path = res.into_inner().collection_id_to_version_file_path;
1149+
let mut result = HashMap::new();
1150+
for (key, value) in collection_id_to_path {
1151+
let collection_id = CollectionUuid(
1152+
Uuid::try_parse(&key)
1153+
.map_err(|err| BatchGetCollectionVersionFilePathsError::Uuid(err, key))?,
1154+
);
1155+
result.insert(collection_id, value);
1156+
}
1157+
Ok(result)
1158+
}
1159+
11161160
async fn get_last_compaction_time(
11171161
&mut self,
11181162
tenant_ids: Vec<String>,

‎rust/sysdb/src/test_sysdb.rs

Lines changed: 37 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
use chroma_types::{
2-
Collection, CollectionAndSegments, CollectionUuid, CountForksError, Database,
3-
FlushCompactionResponse, GetCollectionSizeError, GetCollectionWithSegmentsError,
4-
GetSegmentsError, ListDatabasesError, ListDatabasesResponse, Segment, SegmentFlushInfo,
5-
SegmentScope, SegmentType, Tenant,
2+
BatchGetCollectionVersionFilePathsError, Collection, CollectionAndSegments, CollectionUuid,
3+
CountForksError, Database, FlushCompactionResponse, GetCollectionSizeError,
4+
GetCollectionWithSegmentsError, GetSegmentsError, ListDatabasesError, ListDatabasesResponse,
5+
Segment, SegmentFlushInfo, SegmentScope, SegmentType, Tenant,
66
};
77
use chroma_types::{GetCollectionsError, SegmentUuid};
88
use parking_lot::Mutex;
@@ -93,6 +93,18 @@ impl TestSysDb {
9393
.insert(tenant, last_compaction_time);
9494
}
9595

96+
pub fn set_collection_version_file_path(
97+
&mut self,
98+
collection_id: CollectionUuid,
99+
version_file_path: String,
100+
) {
101+
let mut inner = self.inner.lock();
102+
103+
inner
104+
.collection_to_version_file_name
105+
.insert(collection_id, version_file_path);
106+
}
107+
96108
fn filter_collections(
97109
collection: &Collection,
98110
collection_id: Option<CollectionUuid>,
@@ -549,4 +561,25 @@ impl TestSysDb {
549561
) -> Result<usize, CountForksError> {
550562
Ok(10)
551563
}
564+
565+
pub(crate) async fn batch_get_collection_version_file_paths(
566+
&self,
567+
collection_ids: Vec<CollectionUuid>,
568+
) -> Result<HashMap<CollectionUuid, String>, BatchGetCollectionVersionFilePathsError> {
569+
let inner = self.inner.lock();
570+
let mut paths = HashMap::new();
571+
for collection_id in collection_ids {
572+
if let Some(path) = inner.collection_to_version_file_name.get(&collection_id) {
573+
paths.insert(collection_id, path.clone());
574+
} else {
575+
return Err(BatchGetCollectionVersionFilePathsError::Grpc(
576+
tonic::Status::not_found(format!(
577+
"Version file not found for collection: {}",
578+
collection_id
579+
)),
580+
));
581+
}
582+
}
583+
Ok(paths)
584+
}
552585
}

‎rust/types/src/api_types.rs

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,23 @@ impl ChromaError for GetCollectionWithSegmentsError {
9797
}
9898
}
9999

100+
#[derive(Debug, Error)]
101+
pub enum BatchGetCollectionVersionFilePathsError {
102+
#[error("Grpc error: {0}")]
103+
Grpc(#[from] Status),
104+
#[error("Could not parse UUID from string {1}: {0}")]
105+
Uuid(uuid::Error, String),
106+
}
107+
108+
impl ChromaError for BatchGetCollectionVersionFilePathsError {
109+
fn code(&self) -> ErrorCodes {
110+
match self {
111+
BatchGetCollectionVersionFilePathsError::Grpc(status) => status.code().into(),
112+
BatchGetCollectionVersionFilePathsError::Uuid(_, _) => ErrorCodes::InvalidArgument,
113+
}
114+
}
115+
}
116+
100117
#[derive(Serialize, ToSchema)]
101118
pub struct ResetResponse {}
102119

0 commit comments

Comments
 (0)