You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "numinnex (via GitHub)" <gi...@apache.org> on 2026/03/10 09:08:15 UTC
[PR] refactor(journal): refactor partition journal utilize storage trait [iggy]
numinnex opened a new pull request, #2909:
URL: https://github.com/apache/iggy/pull/2909
Refactor the `PartitionJournal` to use the `Strorage` trait as backing storage, rather than storing data inline.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(journal): refactor partition journal utilize storage trait [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz merged PR #2909:
URL: https://github.com/apache/iggy/pull/2909
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(journal): refactor partition journal utilize storage trait [iggy]
Posted by "atharvalade (via GitHub)" <gi...@apache.org>.
atharvalade commented on code in PR #2909:
URL: https://github.com/apache/iggy/pull/2909#discussion_r2912416661
##########
core/partitions/src/journal.rs:
##########
@@ -49,89 +46,313 @@ impl std::ops::Deref for MessageLookup {
}
}
-/// In-memory journal that accumulates message batches as an `IggyMessagesBatchSet`.
-///
-/// This is a pure storage layer — it holds batches and supports lookups via
-/// `MessageLookup`. All tracking metadata (offsets, timestamps, counts) lives
-/// outside the journal in the `SegmentedLog`'s `JournalInfo`.
-///
-/// Uses `UnsafeCell` for interior mutability, matching the single-threaded
-/// per-shard execution model.
-pub struct PartitionJournal {
- batch_set: UnsafeCell<IggyMessagesBatchSet>,
+#[allow(dead_code)]
+pub trait QueryableJournal<S>: Journal<S>
+where
+ S: Storage,
+{
+ type Query;
+
+ fn get(&self, query: &Self::Query) -> impl Future<Output = Option<IggyMessagesBatchSet>>;
+}
+
+#[derive(Debug, Default)]
+pub struct PartitionJournalMemStorage {
+ entries: UnsafeCell<Vec<Bytes>>,
+ op_to_index: UnsafeCell<HashMap<u64, usize>>,
+}
+
+impl Storage for PartitionJournalMemStorage {
+ type Buffer = Bytes;
+
+ async fn write(&self, buf: Self::Buffer) -> usize {
+ let op = Message::<PrepareHeader>::from_bytes(buf.clone())
+ .ok()
+ .map(|message| message.header().op);
+
+ let entries = unsafe { &mut *self.entries.get() };
+ let index = entries.len();
+ entries.push(buf.clone());
+
+ if let Some(op) = op {
+ let op_to_index = unsafe { &mut *self.op_to_index.get() };
+ op_to_index.insert(op, index);
+ }
+
+ buf.len()
+ }
+
+ async fn read(&self, offset: usize, _len: usize) -> Self::Buffer {
+ let op = offset as u64;
+ let Some(index) = ({
+ let op_to_index = unsafe { &*self.op_to_index.get() };
+ op_to_index.get(&op).copied()
+ }) else {
+ return Bytes::new();
+ };
+
+ let entries = unsafe { &*self.entries.get() };
+ entries.get(index).cloned().unwrap_or_default()
+ }
+}
+
+pub struct PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ message_offset_to_op: UnsafeCell<BTreeMap<u64, u64>>,
+ timestamp_to_op: UnsafeCell<BTreeMap<u64, u64>>,
+ headers: UnsafeCell<Vec<PrepareHeader>>,
+ inner: UnsafeCell<JournalInner<S>>,
}
-impl PartitionJournal {
- pub fn new() -> Self {
+impl<S> Default for PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes> + Default,
+{
+ fn default() -> Self {
Self {
- batch_set: UnsafeCell::new(IggyMessagesBatchSet::empty()),
+ message_offset_to_op: UnsafeCell::new(BTreeMap::new()),
+ timestamp_to_op: UnsafeCell::new(BTreeMap::new()),
+ headers: UnsafeCell::new(Vec::new()),
+ inner: UnsafeCell::new(JournalInner {
+ storage: S::default(),
+ }),
}
}
+}
+
+impl<S> std::fmt::Debug for PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("PartitionJournal2Impl").finish()
+ }
+}
+
+struct JournalInner<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ storage: S,
+}
+
+impl PartitionJournalMemStorage {
+ fn drain(&self) -> Vec<Bytes> {
+ let entries = unsafe { &mut *self.entries.get() };
+ let drained = std::mem::take(entries);
+ let op_to_index = unsafe { &mut *self.op_to_index.get() };
+ op_to_index.clear();
+ drained
+ }
- /// Drain all accumulated batches, returning the batch set.
+ fn is_empty(&self) -> bool {
+ let entries = unsafe { &*self.entries.get() };
+ entries.is_empty()
+ }
+}
+
+impl PartitionJournal<PartitionJournalMemStorage> {
+ /// Drain all accumulated batches, matching the legacy `PartitionJournal` API.
pub fn commit(&self) -> IggyMessagesBatchSet {
- let batch_set = unsafe { &mut *self.batch_set.get() };
- std::mem::take(batch_set)
+ let entries = {
+ let inner = unsafe { &*self.inner.get() };
+ inner.storage.drain()
+ };
+
+ let mut messages = Vec::with_capacity(entries.len());
+ for bytes in entries {
+ if let Ok(message) = Message::from_bytes(bytes) {
+ messages.push(message);
+ }
+ }
+
+ let headers = unsafe { &mut *self.headers.get() };
+ headers.clear();
+ let offsets = unsafe { &mut *self.message_offset_to_op.get() };
+ offsets.clear();
+ let timestamps = unsafe { &mut *self.timestamp_to_op.get() };
+ timestamps.clear();
+
+ Self::messages_to_batch_set(&messages)
Review Comment:
`commit()` drains storage destructively before parsing. If any `Message::from_bytes` call returns `Err`, the entry is dropped. The old code had no parsing step on this path, so this failure mode is new and unique to this refactor.
##########
core/partitions/src/journal.rs:
##########
@@ -49,89 +46,313 @@ impl std::ops::Deref for MessageLookup {
}
}
-/// In-memory journal that accumulates message batches as an `IggyMessagesBatchSet`.
-///
-/// This is a pure storage layer — it holds batches and supports lookups via
-/// `MessageLookup`. All tracking metadata (offsets, timestamps, counts) lives
-/// outside the journal in the `SegmentedLog`'s `JournalInfo`.
-///
-/// Uses `UnsafeCell` for interior mutability, matching the single-threaded
-/// per-shard execution model.
-pub struct PartitionJournal {
- batch_set: UnsafeCell<IggyMessagesBatchSet>,
+#[allow(dead_code)]
+pub trait QueryableJournal<S>: Journal<S>
+where
+ S: Storage,
+{
+ type Query;
+
+ fn get(&self, query: &Self::Query) -> impl Future<Output = Option<IggyMessagesBatchSet>>;
+}
+
+#[derive(Debug, Default)]
+pub struct PartitionJournalMemStorage {
+ entries: UnsafeCell<Vec<Bytes>>,
+ op_to_index: UnsafeCell<HashMap<u64, usize>>,
+}
+
+impl Storage for PartitionJournalMemStorage {
+ type Buffer = Bytes;
+
+ async fn write(&self, buf: Self::Buffer) -> usize {
+ let op = Message::<PrepareHeader>::from_bytes(buf.clone())
+ .ok()
+ .map(|message| message.header().op);
+
+ let entries = unsafe { &mut *self.entries.get() };
+ let index = entries.len();
+ entries.push(buf.clone());
+
+ if let Some(op) = op {
+ let op_to_index = unsafe { &mut *self.op_to_index.get() };
+ op_to_index.insert(op, index);
+ }
+
+ buf.len()
+ }
+
+ async fn read(&self, offset: usize, _len: usize) -> Self::Buffer {
+ let op = offset as u64;
+ let Some(index) = ({
+ let op_to_index = unsafe { &*self.op_to_index.get() };
+ op_to_index.get(&op).copied()
+ }) else {
+ return Bytes::new();
+ };
+
+ let entries = unsafe { &*self.entries.get() };
+ entries.get(index).cloned().unwrap_or_default()
+ }
+}
+
+pub struct PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ message_offset_to_op: UnsafeCell<BTreeMap<u64, u64>>,
+ timestamp_to_op: UnsafeCell<BTreeMap<u64, u64>>,
+ headers: UnsafeCell<Vec<PrepareHeader>>,
+ inner: UnsafeCell<JournalInner<S>>,
}
-impl PartitionJournal {
- pub fn new() -> Self {
+impl<S> Default for PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes> + Default,
+{
+ fn default() -> Self {
Self {
- batch_set: UnsafeCell::new(IggyMessagesBatchSet::empty()),
+ message_offset_to_op: UnsafeCell::new(BTreeMap::new()),
+ timestamp_to_op: UnsafeCell::new(BTreeMap::new()),
+ headers: UnsafeCell::new(Vec::new()),
+ inner: UnsafeCell::new(JournalInner {
+ storage: S::default(),
+ }),
}
}
+}
+
+impl<S> std::fmt::Debug for PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("PartitionJournal2Impl").finish()
+ }
+}
+
+struct JournalInner<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ storage: S,
+}
+
+impl PartitionJournalMemStorage {
+ fn drain(&self) -> Vec<Bytes> {
+ let entries = unsafe { &mut *self.entries.get() };
+ let drained = std::mem::take(entries);
+ let op_to_index = unsafe { &mut *self.op_to_index.get() };
+ op_to_index.clear();
+ drained
+ }
- /// Drain all accumulated batches, returning the batch set.
+ fn is_empty(&self) -> bool {
+ let entries = unsafe { &*self.entries.get() };
+ entries.is_empty()
+ }
+}
+
+impl PartitionJournal<PartitionJournalMemStorage> {
+ /// Drain all accumulated batches, matching the legacy `PartitionJournal` API.
pub fn commit(&self) -> IggyMessagesBatchSet {
- let batch_set = unsafe { &mut *self.batch_set.get() };
- std::mem::take(batch_set)
+ let entries = {
+ let inner = unsafe { &*self.inner.get() };
+ inner.storage.drain()
+ };
+
+ let mut messages = Vec::with_capacity(entries.len());
+ for bytes in entries {
+ if let Ok(message) = Message::from_bytes(bytes) {
+ messages.push(message);
+ }
+ }
+
+ let headers = unsafe { &mut *self.headers.get() };
+ headers.clear();
+ let offsets = unsafe { &mut *self.message_offset_to_op.get() };
+ offsets.clear();
+ let timestamps = unsafe { &mut *self.timestamp_to_op.get() };
+ timestamps.clear();
+
+ Self::messages_to_batch_set(&messages)
}
pub fn is_empty(&self) -> bool {
- let batch_set = unsafe { &*self.batch_set.get() };
- batch_set.is_empty()
+ let inner = unsafe { &*self.inner.get() };
+ inner.storage.is_empty()
}
}
-impl Default for PartitionJournal {
- fn default() -> Self {
- Self::new()
+impl<S> PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ fn message_to_batch(message: &Message<PrepareHeader>) -> Option<IggyMessagesBatchMut> {
+ if message.header().operation != Operation::SendMessages {
+ return None;
+ }
+
+ crate::decode_send_messages_batch(message.body_bytes())
}
-}
-impl std::fmt::Debug for PartitionJournal {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- f.debug_struct("PartitionJournal").finish()
+ fn messages_to_batch_set(messages: &[Message<PrepareHeader>]) -> IggyMessagesBatchSet {
+ let mut batch_set = IggyMessagesBatchSet::empty();
+
+ for message in messages {
+ if let Some(batch) = Self::message_to_batch(message) {
+ batch_set.add_batch(batch);
+ }
+ }
+
+ batch_set
+ }
+
+ #[allow(dead_code)]
+ fn candidate_start_op(&self, query: &MessageLookup) -> Option<u64> {
+ match query {
+ MessageLookup::Offset { offset, .. } => {
+ let offsets = unsafe { &*self.message_offset_to_op.get() };
+ offsets
+ .range(..=*offset)
+ .next_back()
+ .or_else(|| offsets.range(*offset..).next())
+ .map(|(_, op)| *op)
+ }
+ MessageLookup::Timestamp { timestamp, .. } => {
+ let timestamps = unsafe { &*self.timestamp_to_op.get() };
+ timestamps
+ .range(..=*timestamp)
+ .next_back()
+ .or_else(|| timestamps.range(*timestamp..).next())
+ .map(|(_, op)| *op)
+ }
+ }
+ }
+
+ async fn message_by_op(&self, op: u64) -> Option<Message<PrepareHeader>> {
+ let offset = usize::try_from(op).ok()?;
+ let bytes = {
+ let inner = unsafe { &*self.inner.get() };
+ inner.storage.read(offset, ZERO_LEN).await
+ };
+
+ Some(
+ Message::from_bytes(bytes)
+ .expect("partition.journal.storage.read: invalid bytes for message"),
+ )
+ }
Review Comment:
`message_by_op` returns `Option` but panics for any op that isn't in storage.
`PartitionJournalMemStorage::read` returns `Bytes::new()` for unknown ops, `Message::from_bytes` on empty bytes returns `Err`, and the `.expect()` panics. This is directly reachable from `load_messages_from_storage` which walks ops with `op += 1`; since consensus ops are globally scoped, so gaps per-partition are expected and will trigger this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(journal): refactor partition journal utilize storage trait [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #2909:
URL: https://github.com/apache/iggy/pull/2909#discussion_r2923060386
##########
core/partitions/src/journal.rs:
##########
@@ -49,89 +46,313 @@ impl std::ops::Deref for MessageLookup {
}
}
-/// In-memory journal that accumulates message batches as an `IggyMessagesBatchSet`.
-///
-/// This is a pure storage layer — it holds batches and supports lookups via
-/// `MessageLookup`. All tracking metadata (offsets, timestamps, counts) lives
-/// outside the journal in the `SegmentedLog`'s `JournalInfo`.
-///
-/// Uses `UnsafeCell` for interior mutability, matching the single-threaded
-/// per-shard execution model.
-pub struct PartitionJournal {
- batch_set: UnsafeCell<IggyMessagesBatchSet>,
+#[allow(dead_code)]
+pub trait QueryableJournal<S>: Journal<S>
+where
+ S: Storage,
+{
+ type Query;
+
+ fn get(&self, query: &Self::Query) -> impl Future<Output = Option<IggyMessagesBatchSet>>;
+}
+
+#[derive(Debug, Default)]
+pub struct PartitionJournalMemStorage {
+ entries: UnsafeCell<Vec<Bytes>>,
+ op_to_index: UnsafeCell<HashMap<u64, usize>>,
+}
+
+impl Storage for PartitionJournalMemStorage {
+ type Buffer = Bytes;
+
+ async fn write(&self, buf: Self::Buffer) -> usize {
+ let op = Message::<PrepareHeader>::from_bytes(buf.clone())
+ .ok()
+ .map(|message| message.header().op);
+
+ let entries = unsafe { &mut *self.entries.get() };
+ let index = entries.len();
+ entries.push(buf.clone());
+
+ if let Some(op) = op {
+ let op_to_index = unsafe { &mut *self.op_to_index.get() };
+ op_to_index.insert(op, index);
+ }
+
+ buf.len()
+ }
+
+ async fn read(&self, offset: usize, _len: usize) -> Self::Buffer {
+ let op = offset as u64;
+ let Some(index) = ({
+ let op_to_index = unsafe { &*self.op_to_index.get() };
+ op_to_index.get(&op).copied()
+ }) else {
+ return Bytes::new();
+ };
+
+ let entries = unsafe { &*self.entries.get() };
+ entries.get(index).cloned().unwrap_or_default()
+ }
+}
+
+pub struct PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ message_offset_to_op: UnsafeCell<BTreeMap<u64, u64>>,
+ timestamp_to_op: UnsafeCell<BTreeMap<u64, u64>>,
+ headers: UnsafeCell<Vec<PrepareHeader>>,
+ inner: UnsafeCell<JournalInner<S>>,
}
-impl PartitionJournal {
- pub fn new() -> Self {
+impl<S> Default for PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes> + Default,
+{
+ fn default() -> Self {
Self {
- batch_set: UnsafeCell::new(IggyMessagesBatchSet::empty()),
+ message_offset_to_op: UnsafeCell::new(BTreeMap::new()),
+ timestamp_to_op: UnsafeCell::new(BTreeMap::new()),
+ headers: UnsafeCell::new(Vec::new()),
+ inner: UnsafeCell::new(JournalInner {
+ storage: S::default(),
+ }),
}
}
+}
+
+impl<S> std::fmt::Debug for PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("PartitionJournal2Impl").finish()
+ }
+}
+
+struct JournalInner<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ storage: S,
+}
+
+impl PartitionJournalMemStorage {
+ fn drain(&self) -> Vec<Bytes> {
+ let entries = unsafe { &mut *self.entries.get() };
+ let drained = std::mem::take(entries);
+ let op_to_index = unsafe { &mut *self.op_to_index.get() };
+ op_to_index.clear();
+ drained
+ }
- /// Drain all accumulated batches, returning the batch set.
+ fn is_empty(&self) -> bool {
+ let entries = unsafe { &*self.entries.get() };
+ entries.is_empty()
+ }
+}
+
+impl PartitionJournal<PartitionJournalMemStorage> {
+ /// Drain all accumulated batches, matching the legacy `PartitionJournal` API.
pub fn commit(&self) -> IggyMessagesBatchSet {
- let batch_set = unsafe { &mut *self.batch_set.get() };
- std::mem::take(batch_set)
+ let entries = {
+ let inner = unsafe { &*self.inner.get() };
+ inner.storage.drain()
+ };
+
+ let mut messages = Vec::with_capacity(entries.len());
+ for bytes in entries {
+ if let Ok(message) = Message::from_bytes(bytes) {
+ messages.push(message);
+ }
+ }
+
+ let headers = unsafe { &mut *self.headers.get() };
+ headers.clear();
+ let offsets = unsafe { &mut *self.message_offset_to_op.get() };
+ offsets.clear();
+ let timestamps = unsafe { &mut *self.timestamp_to_op.get() };
+ timestamps.clear();
+
+ Self::messages_to_batch_set(&messages)
Review Comment:
The goal of this API is to drain entries from journal without validating (panic if validation fails), because all of the validation happens before any entry enters the journal.
I think the point is valid that in case of failed validation, we should make sure that we don't leave journal in inconsistent state, but the way we will achieve that is by crashing the server (if we fail parsing batch read from journal as the commit essentially is like reading from it, then something really bad happend)
##########
core/partitions/src/journal.rs:
##########
@@ -49,89 +46,313 @@ impl std::ops::Deref for MessageLookup {
}
}
-/// In-memory journal that accumulates message batches as an `IggyMessagesBatchSet`.
-///
-/// This is a pure storage layer — it holds batches and supports lookups via
-/// `MessageLookup`. All tracking metadata (offsets, timestamps, counts) lives
-/// outside the journal in the `SegmentedLog`'s `JournalInfo`.
-///
-/// Uses `UnsafeCell` for interior mutability, matching the single-threaded
-/// per-shard execution model.
-pub struct PartitionJournal {
- batch_set: UnsafeCell<IggyMessagesBatchSet>,
+#[allow(dead_code)]
+pub trait QueryableJournal<S>: Journal<S>
+where
+ S: Storage,
+{
+ type Query;
+
+ fn get(&self, query: &Self::Query) -> impl Future<Output = Option<IggyMessagesBatchSet>>;
+}
+
+#[derive(Debug, Default)]
+pub struct PartitionJournalMemStorage {
+ entries: UnsafeCell<Vec<Bytes>>,
Review Comment:
Yeah I agree, the only reason why the op_to_index mapping even exists there, is because we have to handle the `commit` path which resets the storage `offset`, therefore our op_number != storage_offset. I will move that translation to the `PartitionJournal` rather than having it in storage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(journal): refactor partition journal utilize storage trait [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #2909:
URL: https://github.com/apache/iggy/pull/2909#discussion_r2941828755
##########
core/partitions/src/journal.rs:
##########
@@ -49,89 +46,313 @@ impl std::ops::Deref for MessageLookup {
}
}
-/// In-memory journal that accumulates message batches as an `IggyMessagesBatchSet`.
-///
-/// This is a pure storage layer — it holds batches and supports lookups via
-/// `MessageLookup`. All tracking metadata (offsets, timestamps, counts) lives
-/// outside the journal in the `SegmentedLog`'s `JournalInfo`.
-///
-/// Uses `UnsafeCell` for interior mutability, matching the single-threaded
-/// per-shard execution model.
-pub struct PartitionJournal {
- batch_set: UnsafeCell<IggyMessagesBatchSet>,
+#[allow(dead_code)]
+pub trait QueryableJournal<S>: Journal<S>
+where
+ S: Storage,
+{
+ type Query;
+
+ fn get(&self, query: &Self::Query) -> impl Future<Output = Option<IggyMessagesBatchSet>>;
+}
+
+#[derive(Debug, Default)]
+pub struct PartitionJournalMemStorage {
+ entries: UnsafeCell<Vec<Bytes>>,
Review Comment:
So I've actually realized that we cannot do that, the op_to_offset transistion needs to exist if we want to have an possibility to support storages that are disk-backed. The Storage API needs to work on the `offset`, but the on-disk offset != in-memory index, so the usage inside of `PartitionJournal` would differ (it would require different code for disk backed storage and in-memory one). If we would to keep it isomorphic, then we need to operate either on the disk based offset and have offset_to_index translation inside of in-memory storage, or the other way around -- operate on op_number and then for disk-based storage have an index_to_offset translation.
Alternatively we could make the in-memory storage be an contiguous chunk of memory (Vec<u8>) and treat it the same way an block device (be sector addressable).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(journal): refactor partition journal utilize storage trait [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #2909:
URL: https://github.com/apache/iggy/pull/2909#issuecomment-4029826953
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/2909?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 68.63%. Comparing base ([`08845bf`](https://app.codecov.io/gh/apache/iggy/commit/08845bf36bdeee99fad3d544bd3d9b7613924b8e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`b8e0bbe`](https://app.codecov.io/gh/apache/iggy/commit/b8e0bbecc0b10d5b0d4468c4524d93fd2b0d4e12?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #2909 +/- ##
============================================
- Coverage 70.36% 68.63% -1.74%
Complexity 763 763
============================================
Files 1041 965 -76
Lines 86433 80373 -6060
Branches 62708 62473 -235
============================================
- Hits 60818 55163 -5655
+ Misses 23088 22682 -406
- Partials 2527 2528 +1
```
| [Flag](https://app.codecov.io/gh/apache/iggy/pull/2909/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [node](https://app.codecov.io/gh/apache/iggy/pull/2909/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `87.29% <ø> (-4.22%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/2909?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/partitions/src/iggy\_partition.rs](https://app.codecov.io/gh/apache/iggy/pull/2909?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Figgy_partition.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9pZ2d5X3BhcnRpdGlvbi5ycw==) | `0.00% <ø> (ø)` | |
| [core/partitions/src/iggy\_partitions.rs](https://app.codecov.io/gh/apache/iggy/pull/2909?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Figgy_partitions.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9pZ2d5X3BhcnRpdGlvbnMucnM=) | `0.00% <ø> (ø)` | |
| [core/partitions/src/journal.rs](https://app.codecov.io/gh/apache/iggy/pull/2909?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Fjournal.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9qb3VybmFsLnJz) | `0.00% <ø> (ø)` | |
| [core/partitions/src/lib.rs](https://app.codecov.io/gh/apache/iggy/pull/2909?src=pr&el=tree&filepath=core%2Fpartitions%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9wYXJ0aXRpb25zL3NyYy9saWIucnM=) | `0.00% <ø> (ø)` | |
| [core/simulator/src/deps.rs](https://app.codecov.io/gh/apache/iggy/pull/2909?src=pr&el=tree&filepath=core%2Fsimulator%2Fsrc%2Fdeps.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaW11bGF0b3Ivc3JjL2RlcHMucnM=) | `0.00% <ø> (ø)` | |
... and [94 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/2909/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(journal): refactor partition journal utilize storage trait [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #2909:
URL: https://github.com/apache/iggy/pull/2909#discussion_r2923092140
##########
core/partitions/src/journal.rs:
##########
@@ -49,89 +46,313 @@ impl std::ops::Deref for MessageLookup {
}
}
-/// In-memory journal that accumulates message batches as an `IggyMessagesBatchSet`.
-///
-/// This is a pure storage layer — it holds batches and supports lookups via
-/// `MessageLookup`. All tracking metadata (offsets, timestamps, counts) lives
-/// outside the journal in the `SegmentedLog`'s `JournalInfo`.
-///
-/// Uses `UnsafeCell` for interior mutability, matching the single-threaded
-/// per-shard execution model.
-pub struct PartitionJournal {
- batch_set: UnsafeCell<IggyMessagesBatchSet>,
+#[allow(dead_code)]
+pub trait QueryableJournal<S>: Journal<S>
+where
+ S: Storage,
+{
+ type Query;
+
+ fn get(&self, query: &Self::Query) -> impl Future<Output = Option<IggyMessagesBatchSet>>;
+}
+
+#[derive(Debug, Default)]
+pub struct PartitionJournalMemStorage {
+ entries: UnsafeCell<Vec<Bytes>>,
+ op_to_index: UnsafeCell<HashMap<u64, usize>>,
+}
+
+impl Storage for PartitionJournalMemStorage {
+ type Buffer = Bytes;
+
+ async fn write(&self, buf: Self::Buffer) -> usize {
+ let op = Message::<PrepareHeader>::from_bytes(buf.clone())
+ .ok()
+ .map(|message| message.header().op);
+
+ let entries = unsafe { &mut *self.entries.get() };
+ let index = entries.len();
+ entries.push(buf.clone());
+
+ if let Some(op) = op {
+ let op_to_index = unsafe { &mut *self.op_to_index.get() };
+ op_to_index.insert(op, index);
+ }
+
+ buf.len()
+ }
+
+ async fn read(&self, offset: usize, _len: usize) -> Self::Buffer {
+ let op = offset as u64;
+ let Some(index) = ({
+ let op_to_index = unsafe { &*self.op_to_index.get() };
+ op_to_index.get(&op).copied()
+ }) else {
+ return Bytes::new();
+ };
+
+ let entries = unsafe { &*self.entries.get() };
+ entries.get(index).cloned().unwrap_or_default()
+ }
+}
+
+pub struct PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ message_offset_to_op: UnsafeCell<BTreeMap<u64, u64>>,
+ timestamp_to_op: UnsafeCell<BTreeMap<u64, u64>>,
+ headers: UnsafeCell<Vec<PrepareHeader>>,
+ inner: UnsafeCell<JournalInner<S>>,
}
-impl PartitionJournal {
- pub fn new() -> Self {
+impl<S> Default for PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes> + Default,
+{
+ fn default() -> Self {
Self {
- batch_set: UnsafeCell::new(IggyMessagesBatchSet::empty()),
+ message_offset_to_op: UnsafeCell::new(BTreeMap::new()),
+ timestamp_to_op: UnsafeCell::new(BTreeMap::new()),
+ headers: UnsafeCell::new(Vec::new()),
+ inner: UnsafeCell::new(JournalInner {
+ storage: S::default(),
+ }),
}
}
+}
+
+impl<S> std::fmt::Debug for PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("PartitionJournal2Impl").finish()
+ }
+}
+
+struct JournalInner<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ storage: S,
+}
+
+impl PartitionJournalMemStorage {
+ fn drain(&self) -> Vec<Bytes> {
+ let entries = unsafe { &mut *self.entries.get() };
+ let drained = std::mem::take(entries);
+ let op_to_index = unsafe { &mut *self.op_to_index.get() };
+ op_to_index.clear();
+ drained
+ }
- /// Drain all accumulated batches, returning the batch set.
+ fn is_empty(&self) -> bool {
+ let entries = unsafe { &*self.entries.get() };
+ entries.is_empty()
+ }
+}
+
+impl PartitionJournal<PartitionJournalMemStorage> {
+ /// Drain all accumulated batches, matching the legacy `PartitionJournal` API.
pub fn commit(&self) -> IggyMessagesBatchSet {
- let batch_set = unsafe { &mut *self.batch_set.get() };
- std::mem::take(batch_set)
+ let entries = {
+ let inner = unsafe { &*self.inner.get() };
+ inner.storage.drain()
+ };
+
+ let mut messages = Vec::with_capacity(entries.len());
+ for bytes in entries {
+ if let Ok(message) = Message::from_bytes(bytes) {
+ messages.push(message);
+ }
+ }
+
+ let headers = unsafe { &mut *self.headers.get() };
+ headers.clear();
+ let offsets = unsafe { &mut *self.message_offset_to_op.get() };
+ offsets.clear();
+ let timestamps = unsafe { &mut *self.timestamp_to_op.get() };
+ timestamps.clear();
+
+ Self::messages_to_batch_set(&messages)
}
pub fn is_empty(&self) -> bool {
- let batch_set = unsafe { &*self.batch_set.get() };
- batch_set.is_empty()
+ let inner = unsafe { &*self.inner.get() };
+ inner.storage.is_empty()
}
}
-impl Default for PartitionJournal {
- fn default() -> Self {
- Self::new()
+impl<S> PartitionJournal<S>
+where
+ S: Storage<Buffer = Bytes>,
+{
+ fn message_to_batch(message: &Message<PrepareHeader>) -> Option<IggyMessagesBatchMut> {
+ if message.header().operation != Operation::SendMessages {
+ return None;
+ }
+
+ crate::decode_send_messages_batch(message.body_bytes())
}
-}
-impl std::fmt::Debug for PartitionJournal {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- f.debug_struct("PartitionJournal").finish()
+ fn messages_to_batch_set(messages: &[Message<PrepareHeader>]) -> IggyMessagesBatchSet {
+ let mut batch_set = IggyMessagesBatchSet::empty();
+
+ for message in messages {
+ if let Some(batch) = Self::message_to_batch(message) {
+ batch_set.add_batch(batch);
+ }
+ }
+
+ batch_set
+ }
+
+ #[allow(dead_code)]
+ fn candidate_start_op(&self, query: &MessageLookup) -> Option<u64> {
+ match query {
+ MessageLookup::Offset { offset, .. } => {
+ let offsets = unsafe { &*self.message_offset_to_op.get() };
+ offsets
+ .range(..=*offset)
+ .next_back()
+ .or_else(|| offsets.range(*offset..).next())
+ .map(|(_, op)| *op)
+ }
+ MessageLookup::Timestamp { timestamp, .. } => {
+ let timestamps = unsafe { &*self.timestamp_to_op.get() };
+ timestamps
+ .range(..=*timestamp)
+ .next_back()
+ .or_else(|| timestamps.range(*timestamp..).next())
+ .map(|(_, op)| *op)
+ }
+ }
+ }
+
+ async fn message_by_op(&self, op: u64) -> Option<Message<PrepareHeader>> {
+ let offset = usize::try_from(op).ok()?;
+ let bytes = {
+ let inner = unsafe { &*self.inner.get() };
+ inner.storage.read(offset, ZERO_LEN).await
+ };
+
+ Some(
+ Message::from_bytes(bytes)
+ .expect("partition.journal.storage.read: invalid bytes for message"),
+ )
+ }
Review Comment:
Good catch, we shouldn't expect op to be monotonic linearly, as you've mentioned they are shared between partition messages and partition metadata operations, so the op_number could potentially be scattered.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(journal): refactor partition journal utilize storage trait [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on code in PR #2909:
URL: https://github.com/apache/iggy/pull/2909#discussion_r2915817189
##########
core/partitions/src/journal.rs:
##########
@@ -49,89 +46,313 @@ impl std::ops::Deref for MessageLookup {
}
}
-/// In-memory journal that accumulates message batches as an `IggyMessagesBatchSet`.
-///
-/// This is a pure storage layer — it holds batches and supports lookups via
-/// `MessageLookup`. All tracking metadata (offsets, timestamps, counts) lives
-/// outside the journal in the `SegmentedLog`'s `JournalInfo`.
-///
-/// Uses `UnsafeCell` for interior mutability, matching the single-threaded
-/// per-shard execution model.
-pub struct PartitionJournal {
- batch_set: UnsafeCell<IggyMessagesBatchSet>,
+#[allow(dead_code)]
+pub trait QueryableJournal<S>: Journal<S>
+where
+ S: Storage,
+{
+ type Query;
+
+ fn get(&self, query: &Self::Query) -> impl Future<Output = Option<IggyMessagesBatchSet>>;
+}
+
+#[derive(Debug, Default)]
+pub struct PartitionJournalMemStorage {
+ entries: UnsafeCell<Vec<Bytes>>,
Review Comment:
I think `entries` can be made into `UnsafeCell<Vec<u8>>` and store a hashmap in `PartitionJournal` which maps `op -> (offset, byte_length)` inside the above `Vec<u8>`.
This would simplify `impl read` and `write` of `PartitionJournalStorage`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org