From 15066c8a9ff0977528e1d3da24dab8676b22a132 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 29 Oct 2026 11:57:23 -0700 Subject: [PATCH] vector: implement drop_index and have DeleteIndex drop the data (#12204) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * vector: implement drop_index in the engine The RPC was a stub; it now resolves the table's vended location and removes every object under it through the same credential path open uses. A table the catalog can no longer describe is already dropped, so the call stays idempotent. * s3vectors: drop index data after the catalog delete DeleteTable carries the delete authorization, so it runs first; the engine DropIndex then removes the data under the index's location. NotFound and Unimplemented are tolerated so an index whose data went away with the catalog entry, or an older engine, still answers 200. * s3vectors: let the catalog delete carry the index data removal A vector index is colocated: the catalog entry sits at the dataset root, so DeleteTable's recursive delete already removes entry and data. The engine DropIndex call afterwards could only ever describe a missing or recreated record — a no-op in the common case, and a delete of a replacement index's files when the name was reused in between. --- seaweed-vector/crates/server/src/service.rs | 12 ++++-- seaweed-worker/crates/lance/src/dataset.rs | 40 ++++++++++++++++++ weed/s3api/s3vectors/handlers.go | 5 +-- weed/s3api/s3vectors/handlers_test.go | 46 +++++++++++++++++++++ 4 files changed, 97 insertions(+), 6 deletions(-) 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") + } +}