Skip to content

Commit fa710dc

Browse files
rdettairdettai
committed
Refactor mutate_index_metadata to use MutationOccured<IndexMetadata>
1 parent 9e51be7 commit fa710dc

File tree

4 files changed

+46
-39
lines changed

4 files changed

+46
-39
lines changed

‎quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_index/mod.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -506,7 +506,7 @@ impl FileBackedIndex {
506506
}
507507

508508
/// Deletes the source. Returns whether a mutation occurred.
509-
pub(crate) fn delete_source(&mut self, source_id: &str) -> MetastoreResult<bool> {
509+
pub(crate) fn delete_source(&mut self, source_id: &str) -> MetastoreResult<()> {
510510
self.metadata.delete_source(source_id)
511511
}
512512

‎quickwit/quickwit-metastore/src/metastore/file_backed/mod.rs‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -663,9 +663,8 @@ impl MetastoreService for FileBackedMetastore {
663663
let index_uid = request.index_uid();
664664

665665
self.mutate(index_uid, |index| {
666-
index
667-
.delete_source(&request.source_id)
668-
.map(MutationOccurred::from)
666+
index.delete_source(&request.source_id)?;
667+
Ok(MutationOccurred::Yes(()))
669668
})
670669
.await?;
671670
Ok(EmptyResponse {})

‎quickwit/quickwit-metastore/src/metastore/index_metadata/mod.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -126,16 +126,16 @@ impl IndexMetadata {
126126
Ok(mutation_occurred)
127127
}
128128

129-
/// Deletes a source from the index. Returns whether the index was modified (true).
130-
pub(crate) fn delete_source(&mut self, source_id: &str) -> MetastoreResult<bool> {
129+
/// Deletes a source from the index.
130+
pub(crate) fn delete_source(&mut self, source_id: &str) -> MetastoreResult<()> {
131131
self.sources.remove(source_id).ok_or_else(|| {
132132
MetastoreError::NotFound(EntityKind::Source {
133133
index_id: self.index_id().to_string(),
134134
source_id: source_id.to_string(),
135135
})
136136
})?;
137137
self.checkpoint.remove_source(source_id);
138-
Ok(true)
138+
Ok(())
139139
}
140140
}
141141

‎quickwit/quickwit-metastore/src/metastore/postgres/metastore.rs‎

Lines changed: 40 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ use super::utils::{append_query_filters, establish_connection};
6060
use crate::checkpoint::{
6161
IndexCheckpointDelta, PartitionId, SourceCheckpoint, SourceCheckpointDelta,
6262
};
63+
use crate::file_backed::MutationOccurred;
6364
use crate::metastore::postgres::utils::split_maturity_timestamp;
6465
use crate::metastore::{PublishSplitsRequestExt, STREAM_SPLITS_CHUNK_SIZE};
6566
use crate::{
@@ -288,26 +289,31 @@ macro_rules! run_with_tx {
288289
}};
289290
}
290291

291-
async fn mutate_index_metadata<E, M: FnOnce(&mut IndexMetadata) -> Result<bool, E>>(
292+
async fn mutate_index_metadata<
293+
E,
294+
M: FnOnce(IndexMetadata) -> Result<MutationOccurred<IndexMetadata>, E>,
295+
>(
292296
tx: &mut Transaction<'_, Postgres>,
293297
index_uid: IndexUid,
294298
mutate_fn: M,
295-
) -> MetastoreResult<bool>
299+
) -> MetastoreResult<IndexMetadata>
296300
where
297301
MetastoreError: From<E>,
298302
{
299303
let index_id = &index_uid.index_id;
300-
let mut index_metadata = index_metadata(tx, index_id).await?;
304+
let index_metadata = index_metadata(tx, index_id).await?;
301305
if index_metadata.index_uid != index_uid {
302306
return Err(MetastoreError::NotFound(EntityKind::Index {
303307
index_id: index_id.to_string(),
304308
}));
305309
}
306-
let mutation_occurred = mutate_fn(&mut index_metadata)?;
307-
if !mutation_occurred {
308-
return Ok(mutation_occurred);
309-
}
310-
let index_metadata_json = serde_json::to_string(&index_metadata).map_err(|error| {
310+
311+
let mutated_index_metadata = match mutate_fn(index_metadata)? {
312+
MutationOccurred::Yes(index_metadata) => index_metadata,
313+
MutationOccurred::No(index_metadata) => return Ok(index_metadata),
314+
};
315+
316+
let index_metadata_json = serde_json::to_string(&mutated_index_metadata).map_err(|error| {
311317
MetastoreError::JsonSerializeError {
312318
struct_name: "IndexMetadata".to_string(),
313319
message: error.to_string(),
@@ -329,7 +335,7 @@ where
329335
index_id: index_id.to_string(),
330336
}));
331337
}
332-
Ok(mutation_occurred)
338+
Ok(mutated_index_metadata)
333339
}
334340

335341
#[async_trait]
@@ -406,32 +412,25 @@ impl MetastoreService for PostgresqlMetastore {
406412
let retention_policy_opt = request.deserialize_retention_policy()?;
407413
let search_settings = request.deserialize_search_settings()?;
408414
let index_uid: IndexUid = request.index_uid().clone();
409-
let mut mutated_metadata_opt = None;
410-
let mutated_metadata_ref = &mut mutated_metadata_opt;
411-
run_with_tx!(self.connection_pool, tx, {
415+
let updated_metadata = run_with_tx!(self.connection_pool, tx, {
412416
mutate_index_metadata::<MetastoreError, _>(
413417
tx,
414418
index_uid,
415-
|index_metadata: &mut IndexMetadata| {
416-
let mutated = if index_metadata.index_config.search_settings != search_settings
419+
|mut index_metadata: IndexMetadata| {
420+
if index_metadata.index_config.search_settings != search_settings
417421
|| index_metadata.index_config.retention_policy_opt != retention_policy_opt
418422
{
419423
index_metadata.index_config.search_settings = search_settings;
420424
index_metadata.index_config.retention_policy_opt = retention_policy_opt;
421-
true
425+
Ok(MutationOccurred::Yes(index_metadata))
422426
} else {
423-
false
424-
};
425-
*mutated_metadata_ref = Some(index_metadata.clone());
426-
Ok(mutated)
427+
Ok(MutationOccurred::No(index_metadata))
428+
}
427429
},
428430
)
429-
.await?;
430-
Ok(())
431+
.await
431432
})?;
432-
let mutated_metadata =
433-
mutated_metadata_opt.expect("Mutated IndexMetadata should be set by transaction");
434-
IndexMetadataResponse::try_from_index_metadata(&mutated_metadata)
433+
IndexMetadataResponse::try_from_index_metadata(&updated_metadata)
435434
}
436435

437436
#[instrument(skip_all, fields(index_id=%request.index_uid()))]
@@ -975,9 +974,9 @@ impl MetastoreService for PostgresqlMetastore {
975974
mutate_index_metadata::<MetastoreError, _>(
976975
tx,
977976
index_uid,
978-
|index_metadata: &mut IndexMetadata| {
977+
|mut index_metadata: IndexMetadata| {
979978
index_metadata.add_source(source_config)?;
980-
Ok(true)
979+
Ok(MutationOccurred::Yes(index_metadata))
981980
},
982981
)
983982
.await?;
@@ -993,8 +992,12 @@ impl MetastoreService for PostgresqlMetastore {
993992
) -> MetastoreResult<EmptyResponse> {
994993
let index_uid: IndexUid = request.index_uid().clone();
995994
run_with_tx!(self.connection_pool, tx, {
996-
mutate_index_metadata(tx, index_uid, |index_metadata| {
997-
index_metadata.toggle_source(&request.source_id, request.enable)
995+
mutate_index_metadata(tx, index_uid, |mut index_metadata| {
996+
if index_metadata.toggle_source(&request.source_id, request.enable)? {
997+
Ok::<_, MetastoreError>(MutationOccurred::Yes(index_metadata))
998+
} else {
999+
Ok::<_, MetastoreError>(MutationOccurred::No(index_metadata))
1000+
}
9981001
})
9991002
.await?;
10001003
Ok(())
@@ -1010,8 +1013,9 @@ impl MetastoreService for PostgresqlMetastore {
10101013
let index_uid: IndexUid = request.index_uid().clone();
10111014
let source_id = request.source_id.clone();
10121015
run_with_tx!(self.connection_pool, tx, {
1013-
mutate_index_metadata(tx, index_uid.clone(), |index_metadata| {
1014-
index_metadata.delete_source(&source_id)
1016+
mutate_index_metadata(tx, index_uid.clone(), |mut index_metadata| {
1017+
index_metadata.delete_source(&source_id)?;
1018+
Ok::<_, MetastoreError>(MutationOccurred::Yes(index_metadata))
10151019
})
10161020
.await?;
10171021
sqlx::query(
@@ -1038,8 +1042,12 @@ impl MetastoreService for PostgresqlMetastore {
10381042
) -> MetastoreResult<EmptyResponse> {
10391043
let index_uid: IndexUid = request.index_uid().clone();
10401044
run_with_tx!(self.connection_pool, tx, {
1041-
mutate_index_metadata(tx, index_uid, |index_metadata| {
1042-
Ok::<_, MetastoreError>(index_metadata.checkpoint.reset_source(&request.source_id))
1045+
mutate_index_metadata(tx, index_uid, |mut index_metadata| {
1046+
if index_metadata.checkpoint.reset_source(&request.source_id) {
1047+
Ok::<_, MetastoreError>(MutationOccurred::Yes(index_metadata))
1048+
} else {
1049+
Ok::<_, MetastoreError>(MutationOccurred::No(index_metadata))
1050+
}
10431051
})
10441052
.await?;
10451053
Ok(())

0 commit comments

Comments
 (0)