You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/02/25 15:45:47 UTC
(iggy) branch zero_copy_schema updated: poll start work
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch zero_copy_schema
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/zero_copy_schema by this push:
new 08879348 poll start work
08879348 is described below
commit 08879348ae638411c8582775c864a2fec7f50aee
Author: numinex <g....@gmail.com>
AuthorDate: Tue Feb 25 16:45:41 2025 +0100
poll start work
---
bench/src/actors/consumer.rs | 62 +++----
bench/src/actors/producing_consumer.rs | 2 +-
sdk/src/messages/send_messages.rs | 4 +-
sdk/src/models/batch.rs | 100 ++++++++++-
server/src/streaming/partitions/messages.rs | 4 +-
server/src/streaming/segments/logs/fetch_result.rs | 34 ++++
server/src/streaming/segments/logs/log_reader.rs | 200 +++++++++++----------
server/src/streaming/segments/logs/mod.rs | 2 +
server/src/streaming/segments/reading_messages.rs | 66 +++----
9 files changed, 305 insertions(+), 169 deletions(-)
diff --git a/bench/src/actors/consumer.rs b/bench/src/actors/consumer.rs
index 2fe93471..5a92bfb2 100644
--- a/bench/src/actors/consumer.rs
+++ b/bench/src/actors/consumer.rs
@@ -130,39 +130,39 @@ impl Consumer {
);
}
/*
- let warmup_end = Instant::now() + self.warmup_time.get_duration();
- while Instant::now() < warmup_end {
- let offset = current_iteration * messages_per_batch as u64;
- let (strategy, auto_commit) = match self.polling_kind {
- PollingKind::Offset => (PollingStrategy::offset(offset), false),
- PollingKind::Next => (PollingStrategy::next(), true),
- _ => panic!(
- "Unsupported polling kind for benchmark: {:?}",
- self.polling_kind
- ),
- };
- let polled_messages = client
- .poll_messages(
- &stream_id,
- &topic_id,
- partition_id,
- &consumer,
- &strategy,
- messages_per_batch,
- auto_commit,
- )
- .await?;
+ let warmup_end = Instant::now() + self.warmup_time.get_duration();
+ while Instant::now() < warmup_end {
+ let offset = current_iteration * messages_per_batch as u64;
+ let (strategy, auto_commit) = match self.polling_kind {
+ PollingKind::Offset => (PollingStrategy::offset(offset), false),
+ PollingKind::Next => (PollingStrategy::next(), true),
+ _ => panic!(
+ "Unsupported polling kind for benchmark: {:?}",
+ self.polling_kind
+ ),
+ };
+ let polled_messages = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ partition_id,
+ &consumer,
+ &strategy,
+ messages_per_batch,
+ auto_commit,
+ )
+ .await?;
- if polled_messages.messages.is_empty() {
- warn!(
- "Consumer: {} - Messages are empty for offset: {}, retrying...",
- self.consumer_id, offset
- );
- continue;
+ if polled_messages.messages.is_empty() {
+ warn!(
+ "Consumer: {} - Messages are empty for offset: {}, retrying...",
+ self.consumer_id, offset
+ );
+ continue;
+ }
+ current_iteration += 1;
}
- current_iteration += 1;
- }
- */
+ */
}
if let Some(cg_id) = self.consumer_group_id {
diff --git a/bench/src/actors/producing_consumer.rs b/bench/src/actors/producing_consumer.rs
index fac44f9c..b238b7ff 100644
--- a/bench/src/actors/producing_consumer.rs
+++ b/bench/src/actors/producing_consumer.rs
@@ -255,7 +255,7 @@ impl ProducingConsumer {
)
.await?;
- /*
+ /*
if polled_messages.messages.len() != messages_per_batch as usize {
let should_warn = last_warning_time
.map(|t| t.elapsed() >= Duration::from_secs(1))
diff --git a/sdk/src/messages/send_messages.rs b/sdk/src/messages/send_messages.rs
index f1bf768a..c2185107 100644
--- a/sdk/src/messages/send_messages.rs
+++ b/sdk/src/messages/send_messages.rs
@@ -467,8 +467,8 @@ pub(crate) fn as_bytes(
// TODO: create a writer method on the `Message` and in the future
// once the `Message` struct is dropped on the `IggyMessage`.
let headers_len = header::get_headers_size_bytes(&message.headers).as_bytes_u64();
- let payload_len = message.payload.len() as u64 + 16;
- let total_len = headers_len + payload_len;
+ let payload_len = message.payload.len() as u64 + 16;
+ let total_len = headers_len + payload_len;
bytes.extend_from_slice(&0u32.to_le_bytes()); // offset_delta
bytes.extend_from_slice(&0u32.to_le_bytes()); // timestamp_delta
diff --git a/sdk/src/models/batch.rs b/sdk/src/models/batch.rs
index 474708bf..cfa60846 100644
--- a/sdk/src/models/batch.rs
+++ b/sdk/src/models/batch.rs
@@ -226,7 +226,7 @@ impl<const N: usize> std::io::Write for HeaderWriter<N> {
}
#[serde_as]
-#[derive(Clone, Debug, Default, PartialEq)]
+#[derive(Debug, Default, PartialEq)]
pub struct IggyBatch {
pub header: IggyHeader,
pub batch: Vec<u8>,
@@ -237,7 +237,7 @@ impl IggyBatch {
Self { header, batch }
}
- pub fn update_offsets_and_timestamps(
+ pub fn update_header_and_messages_offsets_and_timestamps(
&mut self,
base_offset: u64,
base_timestamp: IggyTimestamp,
@@ -268,6 +268,7 @@ impl IggyBatch {
}
header.last_offset_delta = messages_count - 1;
header.last_timestamp_delta = timestamp_delta;
+ header.batch_length = batch.len() as u64;
fn write_value_at<const N: usize>(slice: &mut [u8], value: [u8; N], position: usize) {
let slice = &mut slice[position..position + N];
@@ -297,3 +298,98 @@ impl IggyBatch {
header_size + batch_size
}
}
+
+#[derive(Debug, PartialEq)]
+pub struct IggyMessageView<'msg> {
+ pub id: u128,
+ pub offset: u64,
+ pub timestamp: IggyTimestamp,
+ pub payload: &'msg [u8],
+ pub headers: &'msg [u8],
+}
+
+impl<'msg> IggyMessageView<'msg> {
+ pub fn new(
+ id: u128,
+ offset: u64,
+ timestamp: IggyTimestamp,
+ payload: &'msg [u8],
+ headers: &'msg [u8],
+ ) -> Self {
+ Self {
+ id,
+ offset,
+ timestamp,
+ payload,
+ headers,
+ }
+ }
+}
+
+impl<'batch> IntoIterator for &'batch IggyBatch {
+ type Item = IggyMessageView<'batch>;
+ type IntoIter = IggyMessageIterator<'batch>;
+
+ fn into_iter(self) -> Self::IntoIter {
+ IggyMessageIterator {
+ batch: self,
+ position: 0,
+ }
+ }
+}
+
+pub struct IggyMessageIterator<'batch> {
+ batch: &'batch IggyBatch,
+ position: usize,
+}
+
+impl<'msg> Iterator for IggyMessageIterator<'msg> {
+ type Item = IggyMessageView<'msg>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ let data = &self.batch.batch;
+ let mut pos = self.position;
+
+ /*
+ let mut position = 0;
+ let mut messages_count = 0u32;
+ let mut timestamp_delta = 0;
+ while position < batch.len() {
+ let offset_delta = messages_count.to_le_bytes();
+ write_value_at(batch, offset_delta, position);
+ position += 4;
+ let duration_delta = IggyTimestamp::now() - base_timestamp;
+ timestamp_delta = duration_delta.as_micros() as u32;
+ let timestamp_delta_value = timestamp_delta.to_le_bytes();
+ write_value_at(batch, timestamp_delta_value, position);
+
+ position += 4;
+ let msg_len = u64::from_le_bytes(batch[position..position + 8].try_into().unwrap()) + 8;
+ position += 8 + msg_len as usize;
+ messages_count += 1;
+ }
+ */
+ //TODO: replace all of those oks with an Faillable iterator I guess ?
+ let offset_delta = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
+ pos += 4;
+ let timestamp_delta = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
+ pos += 4;
+ let total_length = u64::from_le_bytes(data[pos..pos + 8].try_into().ok()?);
+ pos += 8;
+ let payload_length = u64::from_le_bytes(data[pos..pos + 8].try_into().ok()?);
+ pos += 8;
+ let headers_length = total_length - payload_length;
+ let id = u128::from_le_bytes(data[pos..pos+16].try_into().ok()?);
+ pos += 16;
+
+ let payload = &data[pos..payload_length as usize];
+ pos += payload_length as usize;
+ let headers = &data[pos..headers_length as usize];
+ pos += headers_length as usize;
+ self.position += pos;
+
+ let offset = self.batch.header.base_offset + u64::from(offset_delta);
+ let timestamp = self.batch.header.base_timestamp + timestamp_delta as u64;
+ Some(IggyMessageView::new(id, offset, timestamp.into(), payload, headers))
+ }
+}
diff --git a/server/src/streaming/partitions/messages.rs b/server/src/streaming/partitions/messages.rs
index d8e75c4f..cd24afcb 100644
--- a/server/src/streaming/partitions/messages.rs
+++ b/server/src/streaming/partitions/messages.rs
@@ -430,9 +430,9 @@ impl Partition {
}
*/
- batch.update_offsets_and_timestamps(base_offset, base_timestamp);
+ batch.update_header_and_messages_offsets_and_timestamps(base_offset, base_timestamp);
let last_offset_delta = batch.header.last_offset_delta;
- let messages_count = last_offset_delta + 1;
+ let messages_count = last_offset_delta + 1;
let last_offset = base_offset + (last_offset_delta) as u64;
if self.should_increment_offset {
self.current_offset = last_offset;
diff --git a/server/src/streaming/segments/logs/fetch_result.rs b/server/src/streaming/segments/logs/fetch_result.rs
new file mode 100644
index 00000000..a91090f9
--- /dev/null
+++ b/server/src/streaming/segments/logs/fetch_result.rs
@@ -0,0 +1,34 @@
+use iggy::models::batch::IggyHeader;
+use std::ops::Range;
+
+#[derive(Default)]
+pub struct IggyBatchSlice {
+ pub range: Range<usize>,
+ pub header: IggyHeader,
+ pub bytes: Vec<u8>,
+}
+
+impl IggyBatchSlice {
+ pub fn new(range: Range<usize>, header: IggyHeader, bytes: Vec<u8>) -> Self {
+ Self {
+ range,
+ header,
+ bytes,
+ }
+ }
+}
+
+#[derive(Default)]
+pub struct IggyBatchFetchResult {
+ pub batch_slices: Vec<IggyBatchSlice>,
+ pub header: IggyHeader,
+}
+
+impl IggyBatchFetchResult {
+ pub fn new(batch_slices: Vec<IggyBatchSlice>, header: IggyHeader) -> Self {
+ Self {
+ batch_slices,
+ header,
+ }
+ }
+}
diff --git a/server/src/streaming/segments/logs/log_reader.rs b/server/src/streaming/segments/logs/log_reader.rs
index bfa8b79e..955026a8 100644
--- a/server/src/streaming/segments/logs/log_reader.rs
+++ b/server/src/streaming/segments/logs/log_reader.rs
@@ -1,8 +1,13 @@
use crate::streaming::segments::indexes::IndexRange;
use bytes::BytesMut;
use error_set::ErrContext;
-use iggy::{error::IggyError, utils::byte_size::IggyByteSize};
+use iggy::{
+ error::IggyError,
+ models::batch::{IggyBatch, IggyHeader, IGGY_BATCH_OVERHEAD},
+ utils::{byte_size::IggyByteSize, timestamp::IggyTimestamp},
+};
use std::{
+ fmt,
fs::{File, OpenOptions},
os::unix::prelude::FileExt,
};
@@ -16,6 +21,8 @@ use std::{
use tokio::task::spawn_blocking;
use tracing::{error, trace, warn};
+use super::{fetch_result::IggyBatchSlice, IggyBatchFetchResult};
+
/// A dedicated struct for reading from the log file.
#[derive(Debug)]
pub struct SegmentLogReader {
@@ -24,6 +31,32 @@ pub struct SegmentLogReader {
log_size_bytes: Arc<AtomicU64>,
}
+enum BatchSkipCriteria {
+ None,
+ Offset(u64),
+ Timestamp(IggyTimestamp),
+}
+
+impl BatchSkipCriteria {
+ pub fn is_batch_skippable(&self, header: IggyHeader) -> bool {
+ match self {
+ BatchSkipCriteria::None => false,
+ BatchSkipCriteria::Offset(offset) => header.base_offset < *offset,
+ BatchSkipCriteria::Timestamp(timestamp) => todo!(),
+ }
+ }
+}
+
+impl fmt::Display for BatchSkipCriteria {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ BatchSkipCriteria::None => write!(f, "None"),
+ BatchSkipCriteria::Offset(offset) => write!(f, "offset: {}", offset),
+ BatchSkipCriteria::Timestamp(ts) => write!(f, "timestamp: {:?}", ts),
+ }
+ }
+}
+
impl SegmentLogReader {
/// Opens the log file in read mode.
pub async fn new(file_path: &str, log_size_bytes: Arc<AtomicU64>) -> Result<Self, IggyError> {
@@ -66,46 +99,57 @@ impl SegmentLogReader {
})
}
+ // TODO: Implement this as a stream/iterator that streams batches and handle filtering level above.
/// Loads message batches given an index range.
pub async fn load_batches_by_range_impl(
&self,
index_range: &IndexRange,
- ) -> Result<Vec<()>, IggyError> {
- //TODO: Fix me
- /*
+ start_offset: u64,
+ end_offset: u64,
+ ) -> Result<Vec<IggyBatch>, IggyError> {
let mut file_size = self.file_size();
+ // TODO: Fix me
+ /*
if file_size == 0 {
trace!("Log file {} is empty.", self.file_path);
return Ok(Vec::new());
}
+ */
- let mut offset = index_range.start.position as u64;
+ let mut position = index_range.start.position as u64;
let mut batches = Vec::new();
let mut last_batch_to_read = false;
- while !last_batch_to_read && offset < file_size {
+ // TODO: This can be improved, instead of reading all the batches starting from the idx start,
+ // We can skip some of the batches, based on filtering by header
+ while !last_batch_to_read && position < file_size {
file_size = self.file_size();
- match self.read_next_batch(offset, file_size).await? {
- Some((batch, bytes_read)) => {
- offset += bytes_read;
- let last_offset_in_batch = batch.base_offset + batch.last_offset_delta as u64;
-
- if last_offset_in_batch >= index_range.end.offset as u64 || offset >= file_size
+ match self
+ .maybe_read_next_batch(position, file_size, BatchSkipCriteria::Offset(start_offset))
+ .await?
+ {
+ (Some(batch), bytes_read) => {
+ position += bytes_read;
+ let last_offset_in_batch =
+ batch.header.base_offset + batch.header.last_offset_delta as u64;
+ if last_offset_in_batch >= index_range.end.offset as u64
+ || position >= file_size
{
last_batch_to_read = true;
}
batches.push(batch);
}
- None => {
- break;
+ (None, bytes_read) => {
+ if bytes_read == 0 {
+ break;
+ }
+ position += bytes_read;
}
}
}
-
- trace!("Loaded {} message batches.", batches.len());
+ //TODO: Fix me
+ //trace!("Loaded {} message batches.", batches);
Ok(batches)
- */
- todo!()
}
/// Loads and returns all message IDs from the log file.
@@ -226,116 +270,74 @@ impl SegmentLogReader {
todo!()
}
- async fn read_next_batch(
+ async fn maybe_read_next_batch(
&self,
- offset: u64,
+ position: u64,
file_size: u64,
- ) -> Result<Option<((), u64)>, IggyError> {
- //TODO: Fix me
- /*
- let batch_header_size = RETAINED_BATCH_HEADER_LEN;
- if offset + batch_header_size > file_size {
- return Ok(None);
+ skip_criteria: BatchSkipCriteria,
+ ) -> Result<(Option<IggyBatch>, u64), IggyError> {
+ let batch_header_size = IGGY_BATCH_OVERHEAD;
+ if position + batch_header_size > file_size {
+ return Ok((None, 0));
}
- let header_buf = match self.read_at(offset, batch_header_size).await {
+ let header_buf = match self.read_at(position, batch_header_size).await {
Ok(buf) => buf,
- Err(error) if error.kind() == ErrorKind::UnexpectedEof => return Ok(None),
+ Err(error) if error.kind() == ErrorKind::UnexpectedEof => return Ok((None, 0)),
Err(error) => {
error!(
- "Error reading batch header at offset {} in file {}: {error}",
- offset, self.file_path
+ "Error reading batch header at position {} in file {}: {error}",
+ position, self.file_path
);
return Err(IggyError::CannotReadBatchBaseOffset);
}
};
if header_buf.len() < batch_header_size as usize {
warn!(
- "Cannot read batch header at offset {} in file {}. Possibly truncated.",
- offset, self.file_path
+ "Cannot read batch header at position {} in file {}. Possibly truncated.",
+ position, self.file_path
);
- return Ok(None);
+ return Ok((None, 0));
+ }
+ let header = IggyHeader::from_bytes(&header_buf);
+
+ if skip_criteria.is_batch_skippable(header) {
+ trace!(
+ "Skipping batch with base offset: {}, base timestamp: {}, desired {}",
+ header.base_offset,
+ header.base_timestamp,
+ skip_criteria
+ );
+ let bytes_read = batch_header_size + header.batch_length;
+ return Ok((None, bytes_read));
}
- let batch_base_offset = u64::from_le_bytes(
- header_buf[0..8]
- .try_into()
- .with_error_context(|error| {
- format!(
- "Failed to parse batch base offset at offset {offset} in file {}: {error}",
- self.file_path
- )
- })
- .map_err(|_| IggyError::CannotReadBatchBaseOffset)?,
- );
- let batch_length = u32::from_le_bytes(
- header_buf[8..12]
- .try_into()
- .with_error_context(|error| {
- format!(
- "Failed to parse batch length at offset {offset} in file {}: {error}",
- self.file_path
- )
- })
- .map_err(|_| IggyError::CannotReadBatchLength)?,
- );
- let last_offset_delta = u32::from_le_bytes(
- header_buf[12..16]
- .try_into()
- .with_error_context(|error| {
- format!(
- "Failed to parse last offset delta at offset {offset} in file {}: {error}",
- self.file_path
- )
- })
- .map_err(|_| IggyError::CannotReadLastOffsetDelta)?,
- );
- let max_timestamp = u64::from_le_bytes(
- header_buf[16..24]
- .try_into()
- .with_error_context(|error| {
- format!(
- "Failed to parse max timestamp at offset {offset} in file {}: {error}",
- self.file_path
- )
- })
- .map_err(|_| IggyError::CannotReadMaxTimestamp)?,
- );
-
- let payload_len = batch_length as usize;
- let payload_offset = offset + batch_header_size;
- if payload_offset + payload_len as u64 > file_size {
+ let batch_length = header.batch_length as usize;
+ let batch_position = position + batch_header_size;
+ if batch_position + batch_length as u64 > file_size {
warn!(
- "It's not possible to read the full batch payload ({} bytes) at offset {} in file {} of size {}. Possibly truncated.",
- payload_len, payload_offset, self.file_path, file_size
+ "It's not possible to read the full batch payload ({} bytes) at position {} in file {} of size {}. Possibly truncated.",
+ batch_length, batch_position, self.file_path, file_size
);
- return Ok(None);
+ return Ok((None, 0));
}
- let payload_buf = match self.read_at(payload_offset, payload_len as u64).await {
+ let batch_bytes = match self.read_at(batch_position, batch_length as u64).await {
Ok(buf) => buf,
- Err(error) if error.kind() == ErrorKind::UnexpectedEof => return Ok(None),
+ Err(error) if error.kind() == ErrorKind::UnexpectedEof => return Ok((None, 0)),
Err(error) => {
error!(
- "Error reading batch payload at offset {} in file {}: {error}",
- payload_offset, self.file_path
+ "Error reading batch payload at position {} in file {}: {error}",
+ batch_position, self.file_path
);
return Err(IggyError::CannotReadBatchPayload);
}
};
- let bytes_read = batch_header_size + payload_len as u64;
- let batch = RetainedMessageBatch::new(
- batch_base_offset,
- last_offset_delta,
- max_timestamp,
- IggyByteSize::from(payload_len as u64),
- BytesMut::from(&payload_buf[..]).freeze(),
- );
+ let bytes_read = batch_header_size + batch_length as u64;
+ let batch = IggyBatch::new(header, batch_bytes);
- Ok(Some((batch, bytes_read)))
- */
- todo!()
+ Ok((Some(batch), bytes_read))
}
fn file_size(&self) -> u64 {
diff --git a/server/src/streaming/segments/logs/mod.rs b/server/src/streaming/segments/logs/mod.rs
index 74c57b06..148ece2a 100644
--- a/server/src/streaming/segments/logs/mod.rs
+++ b/server/src/streaming/segments/logs/mod.rs
@@ -1,7 +1,9 @@
+mod fetch_result;
mod log_reader;
mod log_writer;
mod persister_task;
+pub use fetch_result::IggyBatchFetchResult;
pub use log_reader::SegmentLogReader;
pub use log_writer::SegmentLogWriter;
pub use persister_task::PersisterTask;
diff --git a/server/src/streaming/segments/reading_messages.rs b/server/src/streaming/segments/reading_messages.rs
index ede1e535..87046ee5 100644
--- a/server/src/streaming/segments/reading_messages.rs
+++ b/server/src/streaming/segments/reading_messages.rs
@@ -1,4 +1,4 @@
-use super::indexes::*;
+use super::{indexes::*, logs::IggyBatchFetchResult};
use crate::streaming::segments::segment::Segment;
use error_set::ErrContext;
use iggy::{
@@ -59,12 +59,12 @@ impl Segment {
&self,
mut offset: u64,
count: u32,
- ) -> Result<Vec<Arc<()>>, IggyError> {
- //TODO: Fix me
+ ) -> Result<IggyBatchFetchResult, IggyError> {
/*
if count == 0 {
return Ok(EMPTY_MESSAGES.into_iter().map(Arc::new).collect());
}
+ */
if offset < self.start_offset {
offset = self.start_offset;
@@ -82,6 +82,8 @@ impl Segment {
return self.load_messages_from_disk(offset, end_offset).await;
}
+ // TODO: Fix me
+ /*
let first_buffer_offset = batch_accumulator.batch_base_offset();
let last_buffer_offset = batch_accumulator.batch_max_offset();
@@ -89,12 +91,15 @@ impl Segment {
if offset >= first_buffer_offset && end_offset <= last_buffer_offset {
return Ok(self.load_messages_from_unsaved_buffer(offset, end_offset));
}
+ */
// Case 2: All messages are on disk
- if end_offset < first_buffer_offset {
- return self.load_messages_from_disk(offset, end_offset).await;
- }
+ //if end_offset < first_buffer_offset {
+ return self.load_messages_from_disk(offset, end_offset).await;
+ //}
+ // TODO: Fix me
+ /*
// Case 3: Messages span disk and messages_require_to_save buffer boundary
let mut messages = Vec::new();
@@ -116,7 +121,6 @@ impl Segment {
Ok(messages)
*/
- todo!()
}
pub async fn get_all_messages(&self) -> Result<Vec<Arc<()>>, IggyError> {
@@ -184,16 +188,16 @@ impl Segment {
pub async fn load_batches_by_range(
&self,
index_range: &IndexRange,
- ) -> Result<Vec<()>, IggyError> {
- //TODO: Fix me
- /*
+ start_offset: u64,
+ end_offset: u64,
+ ) -> Result<IggyBatchFetchResult, IggyError> {
trace!("Loading message batches for index range: {:?}", index_range);
let batches = self
.log_reader
.as_ref()
.unwrap()
- .load_batches_by_range_impl(index_range)
+ .load_batches_by_range_impl(index_range, start_offset, end_offset)
.await
.with_error_context(|error| {
format!(
@@ -201,11 +205,11 @@ impl Segment {
index_range, self
)
})?;
+ // Filter the batches
+ // Create IggyBatchFetchResult
trace!("Loaded {} message batches.", batches.len());
Ok(batches)
- */
- todo!()
}
pub async fn load_index_for_timestamp(
@@ -327,9 +331,7 @@ impl Segment {
&self,
start_offset: u64,
end_offset: u64,
- ) -> Result<Vec<Arc<()>>, IggyError> {
- // TODO: Fix me
- /*
+ ) -> Result<IggyBatchFetchResult, IggyError> {
trace!(
"Loading messages from disk, start offset: {}, end offset: {}, current offset: {}...",
start_offset,
@@ -337,6 +339,8 @@ impl Segment {
self.current_offset
);
+ //TODO: Fix me
+ /*
if start_offset > end_offset {
warn!(
"Cannot load messages from disk, invalid offset range: {} - {}.",
@@ -344,6 +348,7 @@ impl Segment {
);
return Ok(EMPTY_MESSAGES.into_iter().map(Arc::new).collect());
}
+ */
if let Some(indices) = &self.indexes {
let relative_start_offset = (start_offset - self.start_offset) as u32;
@@ -360,7 +365,11 @@ impl Segment {
start_offset,
end_offset
);
+ panic!("todo");
+ // TODO: Fix me
+ /*
return Ok(EMPTY_MESSAGES.into_iter().map(Arc::new).collect());
+ */
}
};
@@ -382,10 +391,8 @@ impl Segment {
self.load_messages_from_segment_file(&index_range, start_offset, end_offset)
.await
}
- None => Ok(EMPTY_MESSAGES.into_iter().map(Arc::new).collect()),
+ None => panic!("todo"),
}
- */
- todo!()
}
async fn load_messages_from_segment_file(
@@ -393,9 +400,7 @@ impl Segment {
index_range: &IndexRange,
start_offset: u64,
end_offset: u64,
- ) -> Result<Vec<Arc<()>>, IggyError> {
- //TODO: Fix me
- /*
+ ) -> Result<IggyBatchFetchResult, IggyError> {
trace!(
"Loading messages from disk, index range: {:?}, start offset: {}, end offset: {}.",
index_range,
@@ -403,27 +408,24 @@ impl Segment {
end_offset
);
let messages_count = (start_offset + end_offset + 1) as usize;
- let messages = self
- .load_batches_by_range(index_range)
+ let batches = self
+ .load_batches_by_range(index_range, start_offset, end_offset)
.await
.with_error_context(|error| format!(
"{COMPONENT} (error: {error}) - failed to load message batches, stream ID: {}, topic ID: {}, partition ID: {}, start offset: {start_offset}, end offset: {end_offset}",
self.stream_id, self.topic_id, self.partition_id
- ))?
- .iter()
- .to_messages_with_filter(messages_count, &|msg| {
- msg.offset >= start_offset && msg.offset <= end_offset
- });
+ ))?;
+ //TODO: Fix me
+ /*
trace!(
"Loaded {} messages from disk, segment start offset: {}, end offset: {}.",
messages.len(),
self.start_offset,
self.current_offset
);
-
- Ok(messages.into_iter().map(Arc::new).collect())
*/
- todo!()
+
+ Ok(batches)
}
}