diff --git a/seaweed-vector/crates/server/src/service.rs b/seaweed-vector/crates/server/src/service.rs index f94dbc9415..64819dad67 100644 --- a/seaweed-vector/crates/server/src/service.rs +++ b/seaweed-vector/crates/server/src/service.rs @@ -325,10 +325,16 @@ impl VectorService for VectorServer { async fn drop_index( &self, - _request: Request, + request: Request, ) -> Result, Status> { - // Dropping is a catalog decision; the gateway deletes the table record. - Err(Status::unimplemented("drop through the catalog")) + let req = request.into_inner(); + let table = req + .table + .ok_or_else(|| Status::invalid_argument("missing table"))?; + dataset::drop(&self.client, &table.id, &self.fallback) + .await + .map_err(internal)?; + Ok(Response::new(pb::DropIndexResponse {})) } async fn put_vectors( diff --git a/seaweed-worker/crates/lance/src/dataset.rs b/seaweed-worker/crates/lance/src/dataset.rs index 4b65c3fb31..4ae382c023 100644 --- a/seaweed-worker/crates/lance/src/dataset.rs +++ b/seaweed-worker/crates/lance/src/dataset.rs @@ -6,10 +6,12 @@ //! their lifetime re-describes rather than carrying one set throughout. use std::collections::HashMap; +use std::sync::Arc; use anyhow::{Context, Result}; use lance::dataset::Dataset; use lance::dataset::builder::DatasetBuilder; +use lance::io::{ObjectStore, ObjectStoreParams, ObjectStoreRegistry, StorageOptionsAccessor}; use crate::catalog::{NamespaceClient, TableDescription}; @@ -64,6 +66,44 @@ async fn open_at(description: &TableDescription, fallback: &FallbackOptions) -> .with_context(|| format!("open lance dataset at {}", description.location)) } +/// Remove every object under a table's location. A table the catalog can no +/// longer describe is already dropped, so the call stays idempotent. +pub async fn drop( + client: &NamespaceClient, + id: &[String], + fallback: &FallbackOptions, +) -> Result<()> { + let description = match client.describe_table(id).await { + Ok(description) => description, + Err(err) if describe_not_found(&err) => return Ok(()), + Err(err) => return Err(err), + }; + let mut options: HashMap = fallback.clone(); + options.extend(description.storage_options.clone()); + let (store, path) = ObjectStore::from_uri_and_params( + Arc::new(ObjectStoreRegistry::default()), + &description.location, + &ObjectStoreParams { + storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options( + options, + ))), + ..Default::default() + }, + ) + .await?; + store.remove_dir_all(path).await?; + Ok(()) +} + +fn describe_not_found(err: &anyhow::Error) -> bool { + err.chain().any(|cause| { + cause + .downcast_ref::() + .and_then(|e| e.status()) + == Some(reqwest::StatusCode::NOT_FOUND) + }) +} + impl OpenTable { pub async fn stats(&self) -> Result { let versions = self.dataset.versions().await?; diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index b65d25b9d4..a2630a2cce 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -798,9 +798,8 @@ func (s *Server) handleDeleteIndex(w http.ResponseWriter, r *http.Request) { fmt.Sprintf("index %q is not a vector index", arns.index)) return } - // Deleting the catalog record is the delete: the engine derives an index's - // existence from its table, and DropIndex is unimplemented on the shared - // service. + // Deleting the catalog record is the delete: the catalog purges the + // dataset under the index's location along with the entry. var resp map[string]interface{} err = s.catalog(r.Context(), "DeleteTable", &s3tables.DeleteTableRequest{ TableBucketARN: s.s3tablesArn(arns.bucket), diff --git a/weed/s3api/s3vectors/handlers_test.go b/weed/s3api/s3vectors/handlers_test.go index 597ac27e3f..9b45d9c2d4 100644 --- a/weed/s3api/s3vectors/handlers_test.go +++ b/weed/s3api/s3vectors/handlers_test.go @@ -10,6 +10,7 @@ import ( "strconv" "strings" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/vector_pb" @@ -571,3 +572,48 @@ func TestValidateEncryptionConfigurationRejectsCustomerKeys(t *testing.T) { }) } } +// A vector index is colocated: its catalog entry sits at the dataset root, so +// the catalog delete removes entry and data together. +func TestDeleteIndexRemovesDatasetDirectory(t *testing.T) { + fs := s3tablestest.Start(t) + internal := s3tables.NewManager() + internal.SetTrusted(true) + tables := s3tables.NewManager() + tables.SetDefaultAllow(true) + s := &Server{ + filerClient: s3tables.NewManagerClient(fs.Client), + tables: tables, + internal: internal, + } + ctx := context.Background() + bucketArn := s.s3tablesArn("vec-bucket") + + var resp map[string]interface{} + if err := s.internalCatalog(ctx, "CreateTableBucket", + &s3tables.CreateTableBucketRequest{Name: "vec-bucket", Format: s3tables.FormatLance}, &resp); err != nil { + t.Fatalf("create bucket: %v", err) + } + if err := s.internalCatalog(ctx, "CreateNamespace", &s3tables.CreateNamespaceRequest{ + TableBucketARN: bucketArn, Namespace: []string{indexNamespace}}, &resp); err != nil { + t.Fatalf("create namespace: %v", err) + } + if err := s.internalCatalog(ctx, "CreateTable", &s3tables.CreateTableRequest{ + TableBucketARN: bucketArn, Namespace: []string{indexNamespace}, Name: "idx", + Format: s3tables.FormatLance}, &resp); err != nil { + t.Fatalf("create index record: %v", err) + } + fs.PutFile("/buckets/vec-bucket/vector/idx/data", "part-0.lance", time.Now()) + + body, _ := json.Marshal(&DeleteIndexRequest{VectorBucketName: "vec-bucket", IndexName: "idx"}) + w := httptest.NewRecorder() + s.handleDeleteIndex(w, httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))) + if w.Code != http.StatusOK { + t.Fatalf("delete index: %d %s", w.Code, w.Body.String()) + } + if e := fs.Get("/buckets/vec-bucket/vector", "idx"); e != nil { + t.Fatal("index catalog entry survived DeleteIndex") + } + if e := fs.Get("/buckets/vec-bucket/vector/idx/data", "part-0.lance"); e != nil { + t.Fatal("index dataset survived DeleteIndex") + } +}