Skip to content
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

fix(storage): use per-table committed epoch in read version update #17744

Merged
merged 4 commits into from
Jul 23, 2024
Merged

fix(storage): use per-table committed epoch in read version update #17744

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8e356b3
fix(storage): use per-table committed epoch in read version update
wenym1 Jul 18, 2024
d645dec
fix
wenym1 Jul 18, 2024
d2e02b0
Merge branch 'main' into yiming/fix-update-version-committed
wenym1 Jul 23, 2024
59d023c
fix latest epoch
wenym1 Jul 23, 2024
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
48 changes: 34 additions & 14 deletions src/storage/hummock_sdk/src/table_watermark.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,16 +45,27 @@ pub struct TableWatermarksIndex {
pub staging_watermarks: VecDeque<(HummockEpoch, Arc<[VnodeWatermark]>)>,
pub committed_watermarks: Option<Arc<TableWatermarks>>,
latest_epoch: HummockEpoch,
committed_epoch: HummockEpoch,
committed_epoch: Option<HummockEpoch>,
}

impl TableWatermarksIndex {
pub fn new(watermark_direction: WatermarkDirection, committed_epoch: HummockEpoch) -> Self {
pub fn new(
watermark_direction: WatermarkDirection,
first_epoch: HummockEpoch,
first_vnode_watermark: Vec<VnodeWatermark>,
committed_epoch: Option<HummockEpoch>,
) -> Self {
if let Some(committed_epoch) = committed_epoch {
assert!(first_epoch > committed_epoch);
}
Self {
watermark_direction,
staging_watermarks: VecDeque::new(),
staging_watermarks: VecDeque::from_iter([(
first_epoch,
Arc::from(first_vnode_watermark),
)]),
committed_watermarks: None,
latest_epoch: committed_epoch,
latest_epoch: first_epoch,
committed_epoch,
}
}
Expand All @@ -66,7 +77,7 @@ impl TableWatermarksIndex {
Self {
watermark_direction: committed_watermarks.direction,
staging_watermarks: VecDeque::new(),
committed_epoch,
committed_epoch: Some(committed_epoch),
latest_epoch: committed_epoch,
committed_watermarks: Some(committed_watermarks),
}
Expand Down Expand Up @@ -238,11 +249,20 @@ impl TableWatermarksIndex {
committed_epoch: HummockEpoch,
) {
assert_eq!(self.watermark_direction, committed_watermark.direction);
assert!(self.committed_epoch <= committed_epoch);
if self.committed_epoch == committed_epoch {
return;
if let Some(prev_committed_epoch) = self.committed_epoch {
assert!(prev_committed_epoch <= committed_epoch);
if prev_committed_epoch == committed_epoch {
return;
}
}
if self.latest_epoch < committed_epoch {
warn!(
latest_epoch = self.latest_epoch,
committed_epoch, "committed_epoch exceed table watermark latest_epoch"
);
self.latest_epoch = committed_epoch;
}
self.committed_epoch = committed_epoch;
self.committed_epoch = Some(committed_epoch);
self.committed_watermarks = Some(committed_watermark);
// keep only watermark higher than committed epoch
while let Some((old_epoch, _)) = self.staging_watermarks.front()
Expand Down Expand Up @@ -902,11 +922,11 @@ mod tests {
watermark2: Bytes,
watermark3: Bytes,
) -> TableWatermarksIndex {
let mut index = TableWatermarksIndex::new(direction, COMMITTED_EPOCH);
index.add_epoch_watermark(
EPOCH1,
vec![VnodeWatermark::new(build_bitmap(0..4), watermark1.clone())].into(),
let mut index = TableWatermarksIndex::new(
direction,
EPOCH1,
vec![VnodeWatermark::new(build_bitmap(0..4), watermark1.clone())],
Some(COMMITTED_EPOCH),
);
index.add_epoch_watermark(
EPOCH2,
Expand Down Expand Up @@ -1057,7 +1077,7 @@ mod tests {
.clone(),
EPOCH1,
);
assert_eq!(EPOCH1, index.committed_epoch);
assert_eq!(EPOCH1, index.committed_epoch.unwrap());
assert_eq!(EPOCH2, index.latest_epoch);
for vnode in 0..VirtualNode::COUNT {
let vnode = VirtualNode::from_index(vnode);
Expand Down
1 change: 1 addition & 0 deletions src/storage/src/hummock/event_handler/uploader/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1165,6 +1165,7 @@ impl HummockUploader {
let UploaderState::Working(data) = &mut self.state else {
return;
};
debug!(epoch, ?table_ids, "start epoch");
for table_id in &table_ids {
let table_data = data
.unsync_data
Expand Down
118 changes: 80 additions & 38 deletions src/storage/src/hummock/store/version.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,13 @@ impl HummockReadVersion {
.map(|table_watermarks| {
TableWatermarksIndex::new_committed(
table_watermarks.clone(),
committed_version.max_committed_epoch(),
committed_version
.version()
.state_table_info
.info()
.get(&table_id)
.expect("should exist")
.committed_epoch,
)
}),
staging: StagingVersion {
Expand Down Expand Up @@ -313,17 +319,28 @@ impl HummockReadVersion {

// old data comes first
for imm_id in imms.iter().rev() {
let valid = match self.staging.imm.pop_back() {
None => false,
Some(prev_imm_id) => prev_imm_id.batch_id() == *imm_id,
let check_err = match self.staging.imm.pop_back() {
None => Some("empty".to_string()),
Some(prev_imm_id) => {
if prev_imm_id.batch_id() == *imm_id {
None
} else {
Some(format!(
"miss match id {} {}",
prev_imm_id.batch_id(),
*imm_id
))
}
}
};
assert!(
valid,
check_err.is_none(),
"should be valid staging_sst.size {},
staging_sst.imm_ids {:?},
staging_sst.epochs {:?},
local_imm_ids {:?},
instance_id {}",
instance_id {}
check_err {:?}",
staging_sst_ref.imm_size,
staging_sst_ref.imm_ids,
staging_sst_ref.epochs,
Expand All @@ -333,6 +350,7 @@ impl HummockReadVersion {
.map(|imm| imm.batch_id())
.collect_vec(),
self.instance_id,
check_err
);
}

Expand All @@ -341,54 +359,78 @@ impl HummockReadVersion {
},

VersionUpdate::CommittedSnapshot(committed_version) => {
let max_committed_epoch = committed_version.max_committed_epoch();
self.committed = committed_version;

if let Some(info) = committed_version
.version()
.state_table_info
.info()
.get(&self.table_id)
{
// TODO: remove it when support update staging local_sst
self.staging
.imm
.retain(|imm| imm.min_epoch() > max_committed_epoch);
let committed_epoch = info.committed_epoch;
self.staging.imm.retain(|imm| {
if self.is_replicated {
imm.min_epoch() > committed_epoch
} else {
assert!(imm.min_epoch() > committed_epoch);
true
}
});

self.staging.sst.retain(|sst| {
sst.epochs.first().expect("epochs not empty") > &max_committed_epoch
sst.epochs.first().expect("epochs not empty") > &committed_epoch
});

// check epochs.last() > MCE
assert!(self.staging.sst.iter().all(|sst| {
sst.epochs.last().expect("epochs not empty") > &max_committed_epoch
sst.epochs.last().expect("epochs not empty") > &committed_epoch
}));
}

if let Some(committed_watermarks) = self
.committed
.version()
.table_watermarks
.get(&self.table_id)
{
if let Some(watermark_index) = &mut self.table_watermarks {
watermark_index.apply_committed_watermarks(
committed_watermarks.clone(),
self.committed.max_committed_epoch(),
);
} else {
self.table_watermarks = Some(TableWatermarksIndex::new_committed(
committed_watermarks.clone(),
self.committed.max_committed_epoch(),
));
if let Some(committed_watermarks) = self
.committed
.version()
.table_watermarks
.get(&self.table_id)
{
if let Some(watermark_index) = &mut self.table_watermarks {
watermark_index.apply_committed_watermarks(
committed_watermarks.clone(),
committed_epoch,
);
} else {
self.table_watermarks = Some(TableWatermarksIndex::new_committed(
committed_watermarks.clone(),
committed_epoch,
));
}
}
}

self.committed = committed_version;
}
VersionUpdate::NewTableWatermark {
direction,
epoch,
vnode_watermarks,
} => self
.table_watermarks
.get_or_insert_with(|| {
TableWatermarksIndex::new(direction, self.committed.max_committed_epoch())
})
.add_epoch_watermark(epoch, Arc::from(vnode_watermarks), direction),
} => {
if let Some(watermark_index) = &mut self.table_watermarks {
watermark_index.add_epoch_watermark(
epoch,
Arc::from(vnode_watermarks),
direction,
);
} else {
self.table_watermarks = Some(TableWatermarksIndex::new(
direction,
epoch,
vnode_watermarks,
self.committed
.version()
.state_table_info
.info()
.get(&self.table_id)
.map(|info| info.committed_epoch),
));
}
}
}
}

Expand Down
Loading