Skip to content

Commit

Permalink
fix(compaction): fix max epoch bug of tombstone (#13696)
Browse files Browse the repository at this point in the history
Signed-off-by: Little-Wallace <[email protected]>
  • Loading branch information
Little-Wallace authored Nov 30, 2023
1 parent abc1e0c commit 6ba29a8
Show file tree
Hide file tree
Showing 28 changed files with 156 additions and 147 deletions.
11 changes: 11 additions & 0 deletions src/common/src/util/epoch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,17 @@ pub const EPOCH_AVAILABLE_BITS: u64 = 16;
pub const MAX_SPILL_TIMES: u16 = ((1 << EPOCH_AVAILABLE_BITS) - 1) as u16;
pub const EPOCH_MASK: u64 = (1 << EPOCH_AVAILABLE_BITS) - 1;
pub const MAX_EPOCH: u64 = u64::MAX & !EPOCH_MASK;

pub fn is_max_epoch(epoch: u64) -> bool {
// Since we have write `MAX_EPOCH` as max epoch to sstable in some previous version,
// it means that there may be two value in our system which represent infinite. We must check
// both of them for compatibility. See bug description in https://github.com/risingwavelabs/risingwave/issues/13717
epoch >= MAX_EPOCH
}
pub fn is_compatibility_max_epoch(epoch: u64) -> bool {
// See bug description in https://github.com/risingwavelabs/risingwave/issues/13717
epoch == MAX_EPOCH
}
impl From<u64> for Epoch {
fn from(epoch: u64) -> Self {
Self(epoch)
Expand Down
6 changes: 3 additions & 3 deletions src/ctl/src/cmd_impl/hummock/list_kv.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
use core::ops::Bound::Unbounded;

use risingwave_common::catalog::TableId;
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_common::util::epoch::is_max_epoch;
use risingwave_storage::hummock::CachePolicy;
use risingwave_storage::store::{PrefetchOptions, ReadOptions, StateStoreReadExt};

Expand All @@ -31,8 +31,8 @@ pub async fn list_kv(
let hummock = context
.hummock_store(HummockServiceOpts::from_env(data_dir)?)
.await?;
if epoch == MAX_EPOCH {
tracing::info!("using MAX_EPOCH as epoch");
if is_max_epoch(epoch) {
tracing::info!("using MAX EPOCH as epoch");
}
let scan_result = {
let range = (Unbounded, Unbounded);
Expand Down
4 changes: 2 additions & 2 deletions src/ctl/src/cmd_impl/hummock/list_version_deltas.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.

use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_hummock_sdk::HummockEpoch;

use crate::CtlContext;

Expand All @@ -23,7 +23,7 @@ pub async fn list_version_deltas(
) -> anyhow::Result<()> {
let meta_client = context.meta_client().await?;
let resp = meta_client
.list_version_deltas(start_id, num_epochs, MAX_EPOCH)
.list_version_deltas(start_id, num_epochs, HummockEpoch::MAX)
.await?;
println!("{:#?}", resp.version_deltas);
Ok(())
Expand Down
4 changes: 2 additions & 2 deletions src/ctl/src/cmd_impl/hummock/pause_resume.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.

use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionUpdateExt;
use risingwave_hummock_sdk::HummockEpoch;

use crate::CtlContext;

Expand Down Expand Up @@ -62,7 +62,7 @@ pub async fn replay_version(context: &CtlContext) -> anyhow::Result<()> {
let mut current_delta_id = base_version.id + 1;
loop {
let deltas = meta_client
.list_version_deltas(current_delta_id, delta_fetch_size, MAX_EPOCH)
.list_version_deltas(current_delta_id, delta_fetch_size, HummockEpoch::MAX)
.await
.unwrap();
if deltas.version_deltas.is_empty() {
Expand Down
4 changes: 2 additions & 2 deletions src/ctl/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ use anyhow::Result;
use clap::{Args, Parser, Subcommand};
use cmd_impl::bench::BenchCommands;
use cmd_impl::hummock::SstDumpArgs;
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_hummock_sdk::HummockEpoch;
use risingwave_meta::backup_restore::RestoreOpts;
use risingwave_pb::meta::update_worker_node_schedulability_request::Schedulability;

Expand Down Expand Up @@ -160,7 +160,7 @@ enum HummockCommands {
DisableCommitEpoch,
/// list all Hummock key-value pairs
ListKv {
#[clap(short, long = "epoch", default_value_t = MAX_EPOCH)]
#[clap(short, long = "epoch", default_value_t = HummockEpoch::MAX)]
epoch: u64,

#[clap(short, long = "table-id")]
Expand Down
3 changes: 1 addition & 2 deletions src/frontend/src/meta_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@
use std::collections::HashMap;

use risingwave_common::system_param::reader::SystemParamsReader;
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_pb::backup_service::MetaSnapshotMetadata;
use risingwave_pb::catalog::Table;
use risingwave_pb::common::WorkerNode;
Expand Down Expand Up @@ -230,7 +229,7 @@ impl FrontendMetaClient for FrontendMetaClientImpl {
async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>> {
// FIXME #8612: there can be lots of version deltas, so better to fetch them by pages and refactor `SysRowSeqScanExecutor` to yield multiple chunks.
self.0
.list_version_deltas(0, u32::MAX, MAX_EPOCH)
.list_version_deltas(0, u32::MAX, u64::MAX)
.await
.map(|v| v.version_deltas)
}
Expand Down
7 changes: 4 additions & 3 deletions src/meta/src/hummock/compaction/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
pub mod compaction_config;
mod overlap_strategy;
use risingwave_common::catalog::TableOption;
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_hummock_sdk::prost_key_range::KeyRangeExt;
use risingwave_pb::hummock::compact_task::{self, TaskStatus, TaskType};

Expand All @@ -29,7 +28,9 @@ use std::fmt::{Debug, Formatter};
use std::sync::Arc;

use picker::{LevelCompactionPicker, TierCompactionPicker};
use risingwave_hummock_sdk::{can_concat, CompactionGroupId, HummockCompactionTaskId};
use risingwave_hummock_sdk::{
can_concat, CompactionGroupId, HummockCompactionTaskId, HummockEpoch,
};
use risingwave_pb::hummock::compaction_config::CompactionMode;
use risingwave_pb::hummock::hummock_version::Levels;
use risingwave_pb::hummock::{CompactTask, CompactionConfig, KeyRange, LevelType};
Expand Down Expand Up @@ -130,7 +131,7 @@ impl CompactStatus {
let compact_task = CompactTask {
input_ssts: ret.input.input_levels,
splits: vec![KeyRange::inf()],
watermark: MAX_EPOCH,
watermark: HummockEpoch::MAX,
sorted_output_ssts: vec![],
task_id,
target_level: target_level_id as u32,
Expand Down
5 changes: 2 additions & 3 deletions src/meta/src/hummock/metrics_utils.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,11 @@ use std::time::{SystemTime, UNIX_EPOCH};

use itertools::{enumerate, Itertools};
use prost::Message;
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_hummock_sdk::compaction_group::hummock_version_ext::{
object_size_map, BranchedSstInfo, HummockVersionExt,
};
use risingwave_hummock_sdk::{
CompactionGroupId, HummockContextId, HummockSstableObjectId, HummockVersionId,
CompactionGroupId, HummockContextId, HummockEpoch, HummockSstableObjectId, HummockVersionId,
};
use risingwave_pb::hummock::hummock_version::Levels;
use risingwave_pb::hummock::write_limits::WriteLimit;
Expand Down Expand Up @@ -345,7 +344,7 @@ pub fn trigger_pin_unpin_snapshot_state(
{
metrics.min_pinned_epoch.set(m as i64);
} else {
metrics.min_pinned_epoch.set(MAX_EPOCH as _);
metrics.min_pinned_epoch.set(HummockEpoch::MAX as _);
}
}

Expand Down
11 changes: 8 additions & 3 deletions src/storage/hummock_sdk/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ mod key_cmp;
use std::cmp::Ordering;

pub use key_cmp::*;
use risingwave_common::util::epoch::{EPOCH_MASK, MAX_EPOCH};
use risingwave_common::util::epoch::EPOCH_MASK;
use risingwave_pb::common::{batch_query_epoch, BatchQueryEpoch};
use risingwave_pb::hummock::SstableInfo;

Expand Down Expand Up @@ -279,9 +279,14 @@ pub struct EpochWithGap(u64);
impl EpochWithGap {
#[allow(unused_variables)]
pub fn new(epoch: u64, spill_offset: u16) -> Self {
// We only use 48 high bit to store epoch and use 16 low bit to store spill offset. But for MAX epoch,
// we still keep `u64::MAX` because we have use it in delete range and persist this value to sstable files.
// So for compatibility, we must skip checking it for u64::MAX. See bug description in https://github.com/risingwavelabs/risingwave/issues/13717
#[cfg(not(feature = "enable_test_epoch"))]
{
debug_assert_eq!(epoch & EPOCH_MASK, 0);
debug_assert!(
((epoch & EPOCH_MASK) == 0) || risingwave_common::util::epoch::is_max_epoch(epoch)
);
let epoch_with_gap = epoch + spill_offset as u64;
EpochWithGap(epoch_with_gap)
}
Expand All @@ -300,7 +305,7 @@ impl EpochWithGap {
}

pub fn new_max_epoch() -> Self {
EpochWithGap(MAX_EPOCH)
EpochWithGap(HummockEpoch::MAX)
}

// return the epoch_with_gap(epoch + spill_offset)
Expand Down
4 changes: 2 additions & 2 deletions src/storage/hummock_test/benches/bench_hummock_iter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,8 @@ use bytes::Bytes;
use criterion::{criterion_group, criterion_main, Criterion};
use futures::{pin_mut, TryStreamExt};
use risingwave_common::cache::CachePriority;
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_hummock_sdk::key::TableKey;
use risingwave_hummock_sdk::HummockEpoch;
use risingwave_hummock_test::get_notification_client_for_test;
use risingwave_hummock_test::local_state_store_test_utils::LocalStateStoreTestExt;
use risingwave_hummock_test::test_utils::TestIngestBatch;
Expand Down Expand Up @@ -99,7 +99,7 @@ fn criterion_benchmark(c: &mut Criterion) {
))
.unwrap();
}
hummock_storage.seal_current_epoch(MAX_EPOCH, SealCurrentEpochOptions::for_test());
hummock_storage.seal_current_epoch(HummockEpoch::MAX, SealCurrentEpochOptions::for_test());

c.bench_function("bench-hummock-iter", move |b| {
b.iter(|| {
Expand Down
16 changes: 8 additions & 8 deletions src/storage/src/hummock/compactor/compactor_runner.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,14 @@ use await_tree::InstrumentAwait;
use bytes::Bytes;
use futures::{stream, FutureExt, StreamExt};
use itertools::Itertools;
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_common::util::epoch::is_max_epoch;
use risingwave_hummock_sdk::compact::{
compact_task_to_string, estimate_memory_for_compact_task, statistics_compact_task,
};
use risingwave_hummock_sdk::key::{FullKey, PointRange};
use risingwave_hummock_sdk::key_range::{KeyRange, KeyRangeCommon};
use risingwave_hummock_sdk::table_stats::{add_table_stats_map, TableStats, TableStatsMap};
use risingwave_hummock_sdk::{can_concat, EpochWithGap};
use risingwave_hummock_sdk::{can_concat, EpochWithGap, HummockEpoch};
use risingwave_pb::hummock::compact_task::{TaskStatus, TaskType};
use risingwave_pb::hummock::{BloomFilterType, CompactTask, LevelType};
use tokio::sync::oneshot::Receiver;
Expand Down Expand Up @@ -157,7 +157,7 @@ impl CompactorRunner {
.context
.storage_opts
.compact_iter_recreate_timeout_ms;
let mut del_iter = ForwardMergeRangeIterator::new(MAX_EPOCH);
let mut del_iter = ForwardMergeRangeIterator::new(HummockEpoch::MAX);

for level in &self.compact_task.input_ssts {
if level.table_infos.is_empty() {
Expand Down Expand Up @@ -657,7 +657,7 @@ where
del_iter.seek(full_key.user_key).await?;
if !task_config.gc_delete_keys
&& del_iter.is_valid()
&& del_iter.earliest_epoch() != MAX_EPOCH
&& !is_max_epoch(del_iter.earliest_epoch())
{
sst_builder
.add_monotonic_delete(MonotonicDeleteEvent {
Expand All @@ -680,7 +680,7 @@ where

let mut last_key = FullKey::default();
let mut watermark_can_see_last_key = false;
let mut user_key_last_delete_epoch = MAX_EPOCH;
let mut user_key_last_delete_epoch = HummockEpoch::MAX;
let mut local_stats = StoreLocalStatistic::default();

// Keep table stats changes due to dropping KV.
Expand Down Expand Up @@ -716,7 +716,7 @@ where
}
last_key.set(iter_key);
watermark_can_see_last_key = false;
user_key_last_delete_epoch = MAX_EPOCH;
user_key_last_delete_epoch = HummockEpoch::MAX;
if value.is_delete() {
local_stats.skip_delete_key_count += 1;
}
Expand Down Expand Up @@ -843,7 +843,7 @@ where
sst_builder
.add_monotonic_delete(MonotonicDeleteEvent {
event_key: extended_largest_user_key,
new_epoch: MAX_EPOCH,
new_epoch: HummockEpoch::MAX,
})
.await?;
break;
Expand Down Expand Up @@ -964,7 +964,7 @@ mod tests {
.cloned()
.collect_vec();

let mut iter = ForwardMergeRangeIterator::new(MAX_EPOCH);
let mut iter = ForwardMergeRangeIterator::new(HummockEpoch::MAX);
iter.add_concat_iter(sstable_infos, sstable_store);

let ret = CompactionDeleteRangeIterator::new(iter)
Expand Down
11 changes: 5 additions & 6 deletions src/storage/src/hummock/compactor/shared_buffer_compact.rs
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,10 @@ use itertools::Itertools;
use risingwave_common::cache::CachePriority;
use risingwave_common::catalog::TableId;
use risingwave_common::hash::VirtualNode;
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_hummock_sdk::compaction_group::StaticCompactionGroupId;
use risingwave_hummock_sdk::key::{FullKey, PointRange, TableKey, UserKey};
use risingwave_hummock_sdk::key_range::KeyRange;
use risingwave_hummock_sdk::{CompactionGroupId, EpochWithGap, LocalSstableInfo};
use risingwave_hummock_sdk::{CompactionGroupId, EpochWithGap, HummockEpoch, LocalSstableInfo};
use risingwave_pb::hummock::compact_task;
use tracing::error;

Expand Down Expand Up @@ -241,7 +240,7 @@ async fn compact_shared_buffer(
Box::new(sstable_object_id_manager.clone()),
);
let mut forward_iters = Vec::with_capacity(payload.len());
let mut del_iter = ForwardMergeRangeIterator::new(MAX_EPOCH);
let mut del_iter = ForwardMergeRangeIterator::new(HummockEpoch::MAX);
for imm in &payload {
forward_iters.push(imm.clone().into_forward_iter());
del_iter.add_batch_iter(imm.delete_range_iter());
Expand Down Expand Up @@ -322,7 +321,7 @@ pub async fn merge_imms_in_memory(
let mut largest_table_key = Bound::Included(Bytes::new());

let mut imm_iters = Vec::with_capacity(imms.len());
let mut del_iter = ForwardMergeRangeIterator::new(MAX_EPOCH);
let mut del_iter = ForwardMergeRangeIterator::new(HummockEpoch::MAX);
for imm in imms {
assert!(
imm.kv_count() > 0 || imm.has_range_tombstone(),
Expand Down Expand Up @@ -394,14 +393,14 @@ pub async fn merge_imms_in_memory(

let mut versions: Vec<(EpochWithGap, HummockValue<Bytes>)> = Vec::new();

let mut pivot_last_delete_epoch = MAX_EPOCH;
let mut pivot_last_delete_epoch = HummockEpoch::MAX;

for ((key, value), epoch) in items {
assert!(key >= pivot, "key should be in ascending order");
if key != pivot {
merged_payload.push((pivot, versions));
pivot = key;
pivot_last_delete_epoch = MAX_EPOCH;
pivot_last_delete_epoch = HummockEpoch::MAX;
versions = vec![];
let target_extended_user_key =
PointRange::from_user_key(UserKey::new(table_id, TableKey(pivot.as_ref())), false);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ use arc_swap::ArcSwap;
use await_tree::InstrumentAwait;
use parking_lot::RwLock;
use prometheus::core::{AtomicU64, GenericGauge};
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionUpdateExt;
use risingwave_hummock_sdk::{info_in_release, HummockEpoch, LocalSstableInfo};
use risingwave_pb::hummock::version_update_payload::Payload;
Expand Down Expand Up @@ -396,7 +395,7 @@ impl HummockEventHandler {
}

self.sstable_object_id_manager
.remove_watermark_object_id(TrackerId::Epoch(MAX_EPOCH));
.remove_watermark_object_id(TrackerId::Epoch(HummockEpoch::MAX));

// Notify completion of the Clear event.
let _ = notifier.send(()).inspect_err(|e| {
Expand Down
16 changes: 2 additions & 14 deletions src/storage/src/hummock/iterator/backward_user.rs
Original file line number Diff line number Diff line change
Expand Up @@ -277,13 +277,7 @@ impl<I: HummockIterator<Direction = Backward>> BackwardUserIterator<I> {
impl<I: HummockIterator<Direction = Backward>> BackwardUserIterator<I> {
/// Creates [`BackwardUserIterator`] with maximum epoch.
pub(crate) fn for_test(iterator: I, key_range: UserKeyRange) -> Self {
Self::with_epoch(
iterator,
key_range,
risingwave_common::util::epoch::MAX_EPOCH,
0,
None,
)
Self::with_epoch(iterator, key_range, HummockEpoch::MAX, 0, None)
}

/// Creates [`BackwardUserIterator`] with maximum epoch.
Expand All @@ -292,13 +286,7 @@ impl<I: HummockIterator<Direction = Backward>> BackwardUserIterator<I> {
key_range: UserKeyRange,
min_epoch: HummockEpoch,
) -> Self {
Self::with_epoch(
iterator,
key_range,
risingwave_common::util::epoch::MAX_EPOCH,
min_epoch,
None,
)
Self::with_epoch(iterator, key_range, HummockEpoch::MAX, min_epoch, None)
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -185,7 +185,7 @@ mod tests {

use bytes::Bytes;
use risingwave_common::catalog::TableId;
use risingwave_common::util::epoch::MAX_EPOCH;
use risingwave_common::util::epoch::is_max_epoch;

use super::*;
use crate::hummock::iterator::test_utils::mock_sstable_store;
Expand Down Expand Up @@ -252,7 +252,7 @@ mod tests {
sstable_store,
);
concat_iterator.rewind().await.unwrap();
assert_eq!(concat_iterator.current_epoch(), MAX_EPOCH);
assert!(is_max_epoch(concat_iterator.current_epoch()));
assert_eq!(
concat_iterator.next_extended_user_key().left_user_key,
test_user_key(b"aaaa").as_ref()
Expand Down
Loading

0 comments on commit 6ba29a8

Please sign in to comment.