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

blockstore: Refactor Rocks iterator methods #4030

Merged
merged 6 commits into from
Dec 11, 2024
+104 −142
Merged

blockstore: Refactor Rocks iterator methods #4030

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
0c966f5
hoist key conversion to byte array up to LedgerColumn
steviez Dec 10, 2024
c6211e8
Condense and remove Rocks::iterator_cf_raw_key()
steviez Dec 10, 2024
50a92cd
Remove iterator_cf_raw_key from test_purge_special_columns_with_old_data
steviez Dec 10, 2024
cc1b175
cleanup test_purge_transaction_memos_compaction_filter
steviez Dec 10, 2024
4b349f6
More test cleanup
steviez Dec 10, 2024
66a2548
Fixup for dcou (ledger-tool)
steviez Dec 10, 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
163 changes: 62 additions & 101 deletions ledger/src/blockstore/blockstore_purge.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -804,18 +804,17 @@ pub mod tests {
}

fn get_index_bounds(blockstore: &Blockstore) -> (Box<[u8]>, Box<[u8]>) {
let first_index = {
let mut status_entry_iterator = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::Start);
status_entry_iterator.next().unwrap().unwrap().0
};
let last_index = {
let mut status_entry_iterator = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::End);
status_entry_iterator.next().unwrap().unwrap().0
};
let (first_index, _value) = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::Start)
.next()
.unwrap();
let (last_index, _value) = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::End)
.next()
.unwrap();

(first_index, last_index)
}

Expand Down Expand Up @@ -866,22 +865,22 @@ pub mod tests {
.put(1, &index1)
.unwrap();

let statuses: Vec<_> = blockstore
let num_statuses = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::Start)
.collect();
assert_eq!(statuses.len(), 15);
.iter(IteratorMode::Start)
Comment on lines -871 to +870
Copy link
Author

@steviez steviez Dec 10, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I started off wanting to kill iterator_cf_raw_key() altogether. However, we still need it for the cases where we perform compaction.

Regardless, I opted to use iter() where possible/it made more sense instead of iterator_cf_raw_key() to work towards being able to remove the test only iterator_cf_raw_key() function

.unwrap()
.count();
assert_eq!(num_statuses, 15);

// Delete some of primary-index 0
let oldest_slot = 3;
purge(&blockstore, oldest_slot);
let status_entry_iterator = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::Start);
.iter(IteratorMode::Start)
.unwrap();
let mut count = 0;
for entry in status_entry_iterator {
let (key, _value) = entry.unwrap();
let (_signature, slot) = <cf::TransactionStatus as Column>::index(&key);
for ((_signature, slot), _value) in status_entry_iterator {
assert!(slot >= oldest_slot);
count += 1;
}
Expand All @@ -892,11 +891,10 @@ pub mod tests {
purge(&blockstore, oldest_slot);
let status_entry_iterator = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::Start);
.iter(IteratorMode::Start)
.unwrap();
let mut count = 0;
for entry in status_entry_iterator {
let (key, _value) = entry.unwrap();
let (_signature, slot) = <cf::TransactionStatus as Column>::index(&key);
for ((_signature, slot), _value) in status_entry_iterator {
assert!(slot >= oldest_slot);
count += 1;
}
Expand All @@ -907,11 +905,10 @@ pub mod tests {
purge(&blockstore, oldest_slot);
let status_entry_iterator = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::Start);
.iter(IteratorMode::Start)
.unwrap();
let mut count = 0;
for entry in status_entry_iterator {
let (key, _value) = entry.unwrap();
let (_signature, slot) = <cf::TransactionStatus as Column>::index(&key);
for ((_signature, slot), _value) in status_entry_iterator {
assert!(slot >= oldest_slot);
count += 1;
}
Expand All @@ -922,11 +919,10 @@ pub mod tests {
purge(&blockstore, oldest_slot);
let status_entry_iterator = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::Start);
.iter(IteratorMode::Start)
.unwrap();
let mut count = 0;
for entry in status_entry_iterator {
let (key, _value) = entry.unwrap();
let (_signature, slot) = <cf::TransactionStatus as Column>::index(&key);
for ((_signature, slot), _value) in status_entry_iterator {
assert!(slot >= oldest_slot);
count += 1;
}
Expand All @@ -937,11 +933,10 @@ pub mod tests {
purge(&blockstore, oldest_slot);
let status_entry_iterator = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::Start);
.iter(IteratorMode::Start)
.unwrap();
let mut count = 0;
for entry in status_entry_iterator {
let (key, _value) = entry.unwrap();
let (_signature, slot) = <cf::TransactionStatus as Column>::index(&key);
for ((_signature, slot), _value) in status_entry_iterator {
assert!(slot >= oldest_slot);
count += 1;
}
Expand All @@ -952,7 +947,8 @@ pub mod tests {
purge(&blockstore, oldest_slot);
let mut status_entry_iterator = blockstore
.transaction_status_cf
.iterator_cf_raw_key(IteratorMode::Start);
.iter(IteratorMode::Start)
.unwrap();
assert!(status_entry_iterator.next().is_none());
}

Expand Down Expand Up @@ -993,27 +989,13 @@ pub mod tests {
let max_slot = 19;

clear_and_repopulate_transaction_statuses_for_test(&blockstore, max_slot);
let first_index = {
let mut status_entry_iterator = blockstore
.transaction_status_cf
.iter(IteratorMode::Start)
.unwrap();
status_entry_iterator.next().unwrap().0
};
let last_index = {
let mut status_entry_iterator = blockstore
.transaction_status_cf
.iter(IteratorMode::End)
.unwrap();
status_entry_iterator.next().unwrap().0
};
let (first_index, last_index) = get_index_bounds(&blockstore);

let oldest_slot = 3;
blockstore.db.set_oldest_slot(oldest_slot);
blockstore.transaction_status_cf.compact_range_raw_key(
&cf::TransactionStatus::key(first_index),
&cf::TransactionStatus::key(last_index),
);
blockstore
.transaction_status_cf
.compact_range_raw_key(&first_index, &last_index);

let status_entry_iterator = blockstore
.transaction_status_cf
Expand All @@ -1027,27 +1009,13 @@ pub mod tests {
assert_eq!(count, max_slot - (oldest_slot - 1));

clear_and_repopulate_transaction_statuses_for_test(&blockstore, max_slot);
let first_index = {
let mut status_entry_iterator = blockstore
.transaction_status_cf
.iter(IteratorMode::Start)
.unwrap();
status_entry_iterator.next().unwrap().0
};
let last_index = {
let mut status_entry_iterator = blockstore
.transaction_status_cf
.iter(IteratorMode::End)
.unwrap();
status_entry_iterator.next().unwrap().0
};
let (first_index, last_index) = get_index_bounds(&blockstore);

let oldest_slot = 12;
blockstore.db.set_oldest_slot(oldest_slot);
blockstore.transaction_status_cf.compact_range_raw_key(
&cf::TransactionStatus::key(first_index),
&cf::TransactionStatus::key(last_index),
);
blockstore
.transaction_status_cf
.compact_range_raw_key(&first_index, &last_index);

let status_entry_iterator = blockstore
.transaction_status_cf
Expand Down Expand Up @@ -1103,33 +1071,28 @@ pub mod tests {
)
.unwrap();

let first_index = {
let mut memos_iterator = blockstore
.transaction_memos_cf
.iterator_cf_raw_key(IteratorMode::Start);
memos_iterator.next().unwrap().unwrap().0
};
let last_index = {
let mut memos_iterator = blockstore
.transaction_memos_cf
.iterator_cf_raw_key(IteratorMode::End);
memos_iterator.next().unwrap().unwrap().0
};
let (first_index, _value) = blockstore
.transaction_memos_cf
.iterator_cf_raw_key(IteratorMode::Start)
.next()
.unwrap();
let (last_index, _value) = blockstore
.transaction_memos_cf
.iterator_cf_raw_key(IteratorMode::End)
.next()
.unwrap();
Comment on lines +1074 to +1083
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can't use get_index_bounds() here since this is operating on a different column.

I could hypothetically parameterize get_index_bounds() to take a column instead of Blockstore (currently, get_index_bounds() uses transaction_status_cf). However, I think that will make things more difficult for a future PR so I'd rather hold off on that for now. IMO, what I did here is still cleaner than what was there (ie no .0)


// Purge at slot 0 should not affect any memos
blockstore.db.set_oldest_slot(0);
blockstore
.transaction_memos_cf
.compact_range_raw_key(&first_index, &last_index);
let memos_iterator = blockstore
let num_memos = blockstore
.transaction_memos_cf
.iterator_cf_raw_key(IteratorMode::Start);
let mut count = 0;
for item in memos_iterator {
let _item = item.unwrap();
count += 1;
}
assert_eq!(count, 4);
.iter(IteratorMode::Start)
.unwrap()
.count();
assert_eq!(num_memos, 4);

// Purge at oldest_slot without clean_slot_0 only purges the current memo at slot 4
blockstore.db.set_oldest_slot(oldest_slot);
Expand All @@ -1138,11 +1101,10 @@ pub mod tests {
.compact_range_raw_key(&first_index, &last_index);
let memos_iterator = blockstore
.transaction_memos_cf
.iterator_cf_raw_key(IteratorMode::Start);
.iter(IteratorMode::Start)
.unwrap();
let mut count = 0;
for item in memos_iterator {
let (key, _value) = item.unwrap();
let slot = <cf::TransactionMemos as Column>::index(&key).1;
for ((_signature, slot), _value) in memos_iterator {
assert!(slot == 0 || slot >= oldest_slot);
count += 1;
}
Expand All @@ -1155,11 +1117,10 @@ pub mod tests {
.compact_range_raw_key(&first_index, &last_index);
let memos_iterator = blockstore
.transaction_memos_cf
.iterator_cf_raw_key(IteratorMode::Start);
.iter(IteratorMode::Start)
.unwrap();
let mut count = 0;
for item in memos_iterator {
let (key, _value) = item.unwrap();
let slot = <cf::TransactionMemos as Column>::index(&key).1;
for ((_signature, slot), _value) in memos_iterator {
assert!(slot >= oldest_slot);
count += 1;
}
Expand Down
83 changes: 40 additions & 43 deletions ledger/src/blockstore_db.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -695,36 +695,7 @@ impl Rocks {
Ok(())
}

fn iterator_cf<C>(&self, cf: &ColumnFamily, iterator_mode: IteratorMode<C::Index>) -> DBIterator
where
C: Column,
{
let start_key;
let iterator_mode = match iterator_mode {
IteratorMode::From(start_from, direction) => {
start_key = C::key(start_from);
RocksIteratorMode::From(&start_key, direction)
}
IteratorMode::Start => RocksIteratorMode::Start,
IteratorMode::End => RocksIteratorMode::End,
};
self.db.iterator_cf(cf, iterator_mode)
}

pub(crate) fn iterator_cf_raw_key(
&self,
cf: &ColumnFamily,
iterator_mode: IteratorMode<Vec<u8>>,
) -> DBIterator {
let start_key;
let iterator_mode = match iterator_mode {
IteratorMode::From(start_from, direction) => {
start_key = start_from;
RocksIteratorMode::From(&start_key, direction)
}
IteratorMode::Start => RocksIteratorMode::Start,
IteratorMode::End => RocksIteratorMode::End,
};
fn iterator_cf(&self, cf: &ColumnFamily, iterator_mode: RocksIteratorMode) -> DBIterator {
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The method in Rocks is now a simple pass-through

self.db.iterator_cf(cf, iterator_mode)
}

Expand Down Expand Up @@ -1498,7 +1469,17 @@ where
iterator_mode: IteratorMode<C::Index>,
) -> Result<impl Iterator<Item = (C::Index, Box<[u8]>)> + '_> {
let cf = self.handle();
let iter = self.backend.iterator_cf::<C>(cf, iterator_mode);
let start_key;
let iterator_mode = match iterator_mode {
IteratorMode::Start => RocksIteratorMode::Start,
IteratorMode::End => RocksIteratorMode::End,
IteratorMode::From(start_from, direction) => {
start_key = C::key(start_from);
RocksIteratorMode::From(&start_key, direction)
}
};

let iter = self.backend.iterator_cf(cf, iterator_mode);
Ok(iter.map(|pair| {
let (key, value) = pair.unwrap();
(C::index(&key), value)
Expand Down Expand Up @@ -1808,7 +1789,17 @@ where
iterator_mode: IteratorMode<C::Index>,
) -> Result<impl Iterator<Item = (C::Index, Box<[u8]>)> + '_> {
let cf = self.handle();
let iter = self.backend.iterator_cf::<C>(cf, iterator_mode);
let start_key;
let iterator_mode = match iterator_mode {
IteratorMode::Start => RocksIteratorMode::Start,
IteratorMode::End => RocksIteratorMode::End,
IteratorMode::From(start_from, direction) => {
start_key = C::key(start_from);
RocksIteratorMode::From(&start_key, direction)
}
};

let iter = self.backend.iterator_cf(cf, iterator_mode);
Ok(iter.filter_map(|pair| {
let (key, value) = pair.unwrap();
C::try_current_index(&key).ok().map(|index| (index, value))
Expand All @@ -1820,16 +1811,18 @@ where
iterator_mode: IteratorMode<C::DeprecatedIndex>,
) -> Result<impl Iterator<Item = (C::DeprecatedIndex, Box<[u8]>)> + '_> {
let cf = self.handle();
let iterator_mode_raw_key = match iterator_mode {
IteratorMode::Start => IteratorMode::Start,
IteratorMode::End => IteratorMode::End,
let start_key;
let iterator_mode = match iterator_mode {
IteratorMode::Start => RocksIteratorMode::Start,
IteratorMode::End => RocksIteratorMode::End,
IteratorMode::From(start_from, direction) => {
let raw_key = C::deprecated_key(start_from);
IteratorMode::From(raw_key, direction)
start_key = C::deprecated_key(start_from);
RocksIteratorMode::From(&start_key, direction)
}
};
let iter = self.backend.iterator_cf_raw_key(cf, iterator_mode_raw_key);
Ok(iter.filter_map(|pair| {

let iterator = self.backend.iterator_cf(cf, iterator_mode);
Ok(iterator.filter_map(|pair| {
let (key, value) = pair.unwrap();
C::try_deprecated_index(&key)
.ok()
Expand Down Expand Up @@ -2204,10 +2197,14 @@ pub mod tests {
{
pub(crate) fn iterator_cf_raw_key(
steviez marked this conversation as resolved.
Show resolved Hide resolved
&self,
iterator_mode: IteratorMode<Vec<u8>>,
) -> DBIterator {
let cf = self.handle();
self.backend.iterator_cf_raw_key(cf, iterator_mode)
iterator_mode: IteratorMode<C::Index>,
) -> impl Iterator<Item = (Box<[u8]>, Box<[u8]>)> + '_ {
// The conversion of key back into Box<[u8]> incurs an extra
// allocation. However, this is test code and the goal is to
// maximize code reuse over efficiency
Comment on lines +2205 to +2208
Copy link
Author

@steviez steviez Dec 10, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I added this comment for the interim, but this duplicate allocation/conversion will go away later in upcoming PR. I think the comment is valid regardless tho

self.iter(iterator_mode)
.unwrap()
.map(|(key, value)| (Box::from(C::key(key)), value))
}
}
}
Loading