mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-30 08:50:48 +01:00
vector: implement drop_index and have DeleteIndex drop the data (#12204)
* 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.
This commit is contained in:
1 parent
2dbf9c9108
commit
15066c8a9f
4 files changed
+97
-6
No files matched your search
@@ -325,10 +325,16 @@ impl VectorService for VectorServer {
|
||||
|
||||
async fn drop_index(
|
||||
&self,
|
||||
_request: Request<pb::DropIndexRequest>,
|
||||
request: Request<pb::DropIndexRequest>,
|
||||
) -> Result<Response<pb::DropIndexResponse>, 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(
|
||||
|
||||
@@ -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<String, String> = 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::<reqwest::Error>()
|
||||
.and_then(|e| e.status())
|
||||
== Some(reqwest::StatusCode::NOT_FOUND)
|
||||
})
|
||||
}
|
||||
|
||||
impl OpenTable {
|
||||
pub async fn stats(&self) -> Result<TableStats> {
|
||||
let versions = self.dataset.versions().await?;
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user