Skip to content

Add create-only split staging mode #6714

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Mallets wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Add create-only split staging mode #6714

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_index/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,6 +333,17 @@ impl FileBackedIndex {
Ok(())
}

/// Stages a split only when its ID is absent, preserving any concurrent writer's row.
pub(crate) fn stage_split_create_only(
&mut self,
split_metadata: SplitMetadata,
) -> Result<(), MetastoreError> {
if self.splits.contains_key(split_metadata.split_id()) {
return Ok(());
Comment on lines +341 to +342

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Report skipped create-only stages as no mutation

When a file-backed create-only request contains only split IDs that already exist, this helper returns the same result as an insertion, and stage_splits consequently returns MutationOccurred::Yes. That makes mutate rewrite the entire index file via put_index for a logical no-op, so repeated recovery retries incur unnecessary storage writes and can fail solely because storage is temporarily unavailable. Return whether an insertion occurred and aggregate that result so an all-skipped batch uses MutationOccurred::No.

Useful? React with 👍 / 👎.

}
self.stage_split(split_metadata)
}

/// Marks the splits for deletion. Returns whether a mutation occurred.
pub(crate) fn mark_splits_for_deletion(
&mut self,
Expand Down
8 changes: 7 additions & 1 deletion quickwit/quickwit-metastore/src/metastore/file_backed/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -656,13 +656,19 @@ impl MetastoreService for FileBackedMetastore {
#[instrument(name = "metastore.file_backed.stage_splits", skip_all, fields(index_uid = %request.index_uid()))]
async fn stage_splits(&self, request: StageSplitsRequest) -> MetastoreResult<EmptyResponse> {
let index_uid = request.index_uid().clone();
let create_only = request.create_only;
let splits_metadata = request.deserialize_splits_metadata()?;

self.mutate(&index_uid, |index| {
let mut failed_split_ids = Vec::new();

for split_metadata in splits_metadata {
match index.stage_split(split_metadata) {
let stage_result = if create_only {
index.stage_split_create_only(split_metadata)
} else {
index.stage_split(split_metadata)
};
match stage_result {
Ok(()) => {}
Err(MetastoreError::FailedPrecondition {
entity: EntityKind::Split { split_id },
Expand Down
2 changes: 2 additions & 0 deletions quickwit/quickwit-metastore/src/metastore/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -650,6 +650,7 @@ impl StageSplitsRequestExt for StageSplitsRequest {
let request = Self {
index_uid: Some(index_uid.into()),
split_metadata_list_serialized_json,
create_only: false,
};
Ok(request)
}
Expand All @@ -663,6 +664,7 @@ impl StageSplitsRequestExt for StageSplitsRequest {
let request = Self {
index_uid: Some(index_uid.into()),
split_metadata_list_serialized_json,
create_only: false,
};
Ok(request)
}
Expand Down
8 changes: 6 additions & 2 deletions quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -679,6 +679,7 @@ impl MetastoreService for PostgresqlMetastore {
#[instrument(name = "metastore.postgres.stage_splits", skip_all, fields(split_ids))]
async fn stage_splits(&self, request: StageSplitsRequest) -> MetastoreResult<EmptyResponse> {
let index_uid: IndexUid = request.index_uid().clone();
let create_only = request.create_only;
let splits_metadata = request.deserialize_splits_metadata()?;

if splits_metadata.is_empty() {
Expand Down Expand Up @@ -745,7 +746,9 @@ impl MetastoreService for PostgresqlMetastore {
node_id = excluded.node_id,
update_timestamp = CURRENT_TIMESTAMP,
create_timestamp = CURRENT_TIMESTAMP
WHERE splits.split_id = excluded.split_id AND splits.split_state = 'Staged'
WHERE splits.split_id = excluded.split_id
AND splits.split_state = 'Staged'
AND NOT $11
Comment on lines +749 to +751

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Use DO NOTHING for create-only conflicts

When concurrent create-only batches contain the same already-existing split IDs in different orders, this remains an ON CONFLICT DO UPDATE statement, and PostgreSQL locks each conflicting row even though the final predicate is false. The workers can therefore acquire different first locks and deadlock, causing one transaction to abort and consume the metastore client's retry budget even though both requests should be conflict no-ops. Use a create-only query with ON CONFLICT DO NOTHING instead.

Useful? React with 👍 / 👎.

RETURNING split_id;
"#)
.bind(&split_ids)
Expand All @@ -758,11 +761,12 @@ impl MetastoreService for PostgresqlMetastore {
.bind(&node_ids)
.bind(SplitState::Staged.as_str())
.bind(&index_uid)
.bind(create_only)
.fetch_all(tx.as_mut())
.await
.map_err(|sqlx_error| convert_sqlx_err(&index_uid.index_id, sqlx_error))?;

if upserted_split_ids.len() != split_ids.len() {
if !create_only && upserted_split_ids.len() != split_ids.len() {
let failed_split_ids: Vec<String> = split_ids
.into_iter()
.filter(|split_id| !upserted_split_ids.contains(split_id))
Expand Down
37 changes: 37 additions & 0 deletions quickwit/quickwit-metastore/src/tests/split.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1621,6 +1621,43 @@ pub async fn test_metastore_stage_splits<MetastoreToTest: MetastoreServiceExt +
.await
.expect("Pre-existing staged splits should be updated.");

// Recovery staging inserts missing rows but must not overwrite an existing writer's staged row.
let mut conflicting_split_metadata = split_metadata_1.clone();
conflicting_split_metadata.node_id = "recovery".to_string();
let split_metadata_3 = SplitMetadata {
split_id: format!("{index_id}--split-3").into(),
index_uid: index_uid.clone(),
node_id: "recovery".to_string(),
..Default::default()
};
let mut stage_splits_request = StageSplitsRequest::try_from_splits_metadata(
index_uid.clone(),
[conflicting_split_metadata, split_metadata_3.clone()],
)
.unwrap();
stage_splits_request.create_only = true;
metastore.stage_splits(stage_splits_request).await.unwrap();

let query = ListSplitsQuery::for_index(index_uid.clone()).with_split_state(SplitState::Staged);
let splits = metastore
.list_splits(ListSplitsRequest::try_from_list_splits_query(&query).unwrap())
.await
.unwrap()
.collect_splits()
.await
.unwrap();
assert_eq!(splits.len(), 3);
let existing_split = splits
.iter()
.find(|split| split.split_id().as_str() == split_id_1)
.unwrap();
assert_eq!(existing_split.split_metadata.node_id, "node-1");
assert!(
splits
.iter()
.any(|split| split.split_id() == split_metadata_3.split_id())
);

let publish_splits_request = PublishSplitsRequest {
index_uid: Some(index_uid.clone()),
staged_split_ids: vec![split_id_1.clone(), split_id_2.clone()],
Expand Down
2 changes: 2 additions & 0 deletions quickwit/quickwit-proto/protos/quickwit/metastore.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -388,6 +388,8 @@ message ListSplitsResponse {
message StageSplitsRequest {
quickwit.common.IndexUid index_uid = 1;
string split_metadata_list_serialized_json = 2;
// Create-only mode: insert missing rows without upserting an existing split row.
bool create_only = 3;

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Gate create-only staging on server support

During a rolling or mixed-version deployment, an older metastore silently ignores unknown protobuf field 3 and executes the existing upsert path, so a new recovery client can overwrite the authoritative split row while receiving a successful response. This semantic mode needs a version-safe RPC or capability gate before callers can rely on it, or an explicitly enforced metastore-first upgrade procedure documented for this protocol change.

AGENTS.md reference: AGENTS.md:L23-L24

Useful? React with 👍 / 👎.

}

message PublishSplitsRequest {
Expand Down
3 changes: 3 additions & 0 deletions quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading