Skip to content

Commit 6c5dc8b

Browse files
committed
fix: Fixed versioned replication error
1 parent e194c65 commit 6c5dc8b

1 file changed

Lines changed: 18 additions & 3 deletions

File tree

components/data_proxy/src/replication/replication_handler.rs

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
use crate::structs::FileFormat;
1+
use crate::structs::{FileFormat, VersionVariant};
22
use crate::CONFIG;
33
use crate::{
44
caching::cache::Cache, data_backends::storage_backend::StorageBackend,
@@ -24,7 +24,7 @@ use pithos_lib::{streamreadwrite::GenericStreamReadWriter, transformer::ReadWrit
2424
use std::{str::FromStr, sync::Arc};
2525
use tokio::pin;
2626
use tokio::sync::RwLock;
27-
use tracing::{info_span, trace, Instrument};
27+
use tracing::{error, info_span, trace, Instrument};
2828

2929
pub struct ReplicationMessage {
3030
pub direction: Direction,
@@ -515,11 +515,26 @@ impl ReplicationHandler {
515515
trace!("skipping object");
516516
continue;
517517
} else {
518+
let parent = {
519+
let mut current_version_id = object_id.clone();
520+
if let Some(version) = &object.versions {
521+
version.iter().next().map(|v| match v {
522+
VersionVariant::IsVersion(id) => {
523+
current_version_id = id.clone()
524+
}
525+
_ => {}
526+
});
527+
}
528+
cache.get_single_parent(&current_version_id).await.map_err(|e| {
529+
error!(error = ?e, msg = e.to_string());
530+
e
531+
})?
532+
};
518533
backend
519534
.initialize_location(
520535
&object,
521536
object_state.read().await.get_size(),
522-
cache.get_single_parent(&object.id).await?,
537+
parent,
523538
false,
524539
)
525540
.await

0 commit comments

Comments
 (0)