false /* log_err_key */); // TODO
pik_status.PermitUncheckedError();
assert(pik_status.ok());
-
smallest_ = &parsed_smallest;
}
if (largest != nullptr) {
// the truncated end key can cover the largest key in this sstable, reduce
// its sequence number by 1.
parsed_largest.sequence -= 1;
+ // This line is not needed for correctness, but it ensures that the
+ // truncated end key is not covering keys from the next SST file.
+ parsed_largest.type = kValueTypeForSeek;
}
largest_ = &parsed_largest;
}
}
bool TruncatedRangeDelIterator::Valid() const {
+ assert(iter_ != nullptr);
return iter_->Valid() &&
(smallest_ == nullptr ||
icmp_->Compare(*smallest_, iter_->parsed_end_key()) < 0) &&
icmp_->Compare(iter_->parsed_start_key(), *largest_) < 0);
}
-void TruncatedRangeDelIterator::Next() { iter_->TopNext(); }
-
-void TruncatedRangeDelIterator::Prev() { iter_->TopPrev(); }
-
-void TruncatedRangeDelIterator::InternalNext() { iter_->Next(); }
-
-// NOTE: target is a user key
+// NOTE: target is a user key, with timestamp if enabled.
void TruncatedRangeDelIterator::Seek(const Slice& target) {
if (largest_ != nullptr &&
icmp_->Compare(*largest_, ParsedInternalKey(target, kMaxSequenceNumber,
iter_->Seek(target);
}
-// NOTE: target is a user key
+// NOTE: target is a user key, with timestamp if enabled.
void TruncatedRangeDelIterator::SeekForPrev(const Slice& target) {
if (smallest_ != nullptr &&
icmp_->Compare(ParsedInternalKey(target, 0, kTypeRangeDeletion),
std::for_each(
split_untruncated_iters.begin(), split_untruncated_iters.end(),
[&](FragmentedIterPair& iter_pair) {
- std::unique_ptr<TruncatedRangeDelIterator> truncated_iter(
- new TruncatedRangeDelIterator(std::move(iter_pair.second), icmp_,
- smallest_ikey_, largest_ikey_));
+ auto truncated_iter = std::make_unique<TruncatedRangeDelIterator>(
+ std::move(iter_pair.second), icmp_, smallest_ikey_, largest_ikey_);
split_truncated_iters.emplace(iter_pair.first,
std::move(truncated_iter));
});
if (input_iter == nullptr || input_iter->empty()) {
return;
}
- rep_.AddTombstones(
- std::unique_ptr<TruncatedRangeDelIterator>(new TruncatedRangeDelIterator(
- std::move(input_iter), icmp_, smallest, largest)));
+ rep_.AddTombstones(std::make_unique<TruncatedRangeDelIterator>(
+ std::move(input_iter), icmp_, smallest, largest));
}
bool ReadRangeDelAggregator::ShouldDeleteImpl(const ParsedInternalKey& parsed,
if (input_iter == nullptr || input_iter->empty()) {
return;
}
+ // This bounds output of CompactionRangeDelAggregator::NewIterator.
+ if (!trim_ts_.empty()) {
+ assert(icmp_->user_comparator()->timestamp_size() > 0);
+ input_iter->SetTimestampUpperBound(&trim_ts_);
+ }
+
assert(input_iter->lower_bound() == 0);
assert(input_iter->upper_bound() == kMaxSequenceNumber);
parent_iters_.emplace_back(new TruncatedRangeDelIterator(
std::move(input_iter), icmp_, smallest, largest));
+ Slice* ts_upper_bound = nullptr;
+ if (!ts_upper_bound_.empty()) {
+ assert(icmp_->user_comparator()->timestamp_size() > 0);
+ ts_upper_bound = &ts_upper_bound_;
+ }
auto split_iters = parent_iters_.back()->SplitBySnapshot(*snapshots_);
for (auto& split_iter : split_iters) {
auto it = reps_.find(split_iter.first);
assert(inserted);
}
assert(it != reps_.end());
+ // ts_upper_bound is used to bound ShouldDelete() to only consider
+ // range tombstones under full_history_ts_low_ and trim_ts_. Keys covered by
+ // range tombstones that are above full_history_ts_low_ should not be
+ // dropped prematurely: user may read with a timestamp between the range
+ // tombstone and the covered key. Note that we cannot set timestamp
+ // upperbound on the original `input_iter` since `input_iter`s are later
+ // used in CompactionRangeDelAggregator::NewIterator to output range
+ // tombstones for persistence. We do not want to only persist range
+ // tombstones with timestamp lower than ts_upper_bound.
+ split_iter.second->SetTimestampUpperBound(ts_upper_bound);
it->second.AddTombstones(std::move(split_iter.second));
}
}
namespace {
+// Produce a sorted (by start internal key) stream of range tombstones from
+// `children`. lower_bound and upper_bound on user key can be
+// optionally specified. Range tombstones that ends before lower_bound or starts
+// after upper_bound are excluded.
+// If user-defined timestamp is enabled, lower_bound and upper_bound should
+// contain timestamp, but comparison is done ignoring timestamps.
class TruncatedRangeDelMergingIter : public InternalIterator {
public:
TruncatedRangeDelMergingIter(
lower_bound_(lower_bound),
upper_bound_(upper_bound),
upper_bound_inclusive_(upper_bound_inclusive),
- heap_(StartKeyMinComparator(icmp)) {
+ heap_(StartKeyMinComparator(icmp)),
+ ts_sz_(icmp_->user_comparator()->timestamp_size()) {
for (auto& child : children) {
if (child != nullptr) {
assert(child->lower_bound() == 0);
Slice key() const override {
auto* top = heap_.top();
- cur_start_key_.Set(top->start_key().user_key, top->seq(),
- kTypeRangeDeletion);
+ if (ts_sz_) {
+ cur_start_key_.Set(top->start_key().user_key, top->seq(),
+ kTypeRangeDeletion, top->timestamp());
+ } else {
+ cur_start_key_.Set(top->start_key().user_key, top->seq(),
+ kTypeRangeDeletion);
+ }
+ assert(top->start_key().user_key.size() >= ts_sz_);
return cur_start_key_.Encode();
}
Slice value() const override {
auto* top = heap_.top();
- assert(top->end_key().sequence == kMaxSequenceNumber);
- return top->end_key().user_key;
+ if (!ts_sz_) {
+ return top->end_key().user_key;
+ }
+ assert(top->timestamp().size() == ts_sz_);
+ cur_end_key_.clear();
+ cur_end_key_.append(top->end_key().user_key.data(),
+ top->end_key().user_key.size() - ts_sz_);
+ cur_end_key_.append(top->timestamp().data(), ts_sz_);
+ return cur_end_key_;
}
// Unused InternalIterator methods
if (upper_bound_ == nullptr) {
return true;
}
- int cmp = icmp_->user_comparator()->Compare(iter->start_key().user_key,
- *upper_bound_);
+ int cmp = icmp_->user_comparator()->CompareWithoutTimestamp(
+ iter->start_key().user_key, *upper_bound_);
return upper_bound_inclusive_ ? cmp <= 0 : cmp < 0;
}
std::vector<TruncatedRangeDelIterator*> children_;
mutable InternalKey cur_start_key_;
+ mutable std::string cur_end_key_;
+ size_t ts_sz_;
};
-} // namespace
+} // anonymous namespace
std::unique_ptr<FragmentedRangeTombstoneIterator>
CompactionRangeDelAggregator::NewIterator(const Slice* lower_bound,
const Slice* upper_bound,
bool upper_bound_inclusive) {
InvalidateRangeDelMapPositions();
- std::unique_ptr<TruncatedRangeDelMergingIter> merging_iter(
- new TruncatedRangeDelMergingIter(icmp_, lower_bound, upper_bound,
- upper_bound_inclusive, parent_iters_));
+ auto merging_iter = std::make_unique<TruncatedRangeDelMergingIter>(
+ icmp_, lower_bound, upper_bound, upper_bound_inclusive, parent_iters_);
auto fragmented_tombstone_list =
std::make_shared<FragmentedRangeTombstoneList>(
std::move(merging_iter), *icmp_, true /* for_compaction */,
*snapshots_);
- return std::unique_ptr<FragmentedRangeTombstoneIterator>(
- new FragmentedRangeTombstoneIterator(
- fragmented_tombstone_list, *icmp_,
- kMaxSequenceNumber /* upper_bound */));
+ return std::make_unique<FragmentedRangeTombstoneIterator>(
+ fragmented_tombstone_list, *icmp_, kMaxSequenceNumber /* upper_bound */);
}
} // namespace ROCKSDB_NAMESPACE
projects / ceph.git / blobdiff