diff --git a/Cargo.lock b/Cargo.lock
index e97dd203ee3..77e60390cb0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4624,15 +4624,6 @@ dependencies = [
"uuid",
]
-[[package]]
-name = "delta-encoding"
-version = "0.4.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cabe72c7de5725dcd4597e7d2bc8a4b7a4698879481607385fbc50a23203eeb9"
-dependencies = [
- "num-traits",
-]
-
[[package]]
name = "der"
version = "0.7.10"
@@ -8119,7 +8110,6 @@ dependencies = [
"common-wal",
"crc32fast",
"dashmap",
- "delta-encoding",
"derive_builder 0.20.2",
"futures",
"futures-util",
diff --git a/config/config.md b/config/config.md
index 381abde24c4..6faae09c347 100644
--- a/config/config.md
+++ b/config/config.md
@@ -604,8 +604,6 @@
| `wal.timeout` | String | `5s` | The total request timeout for kafka client.
**It's only used when the provider is `kafka`**. |
| `wal.max_batch_bytes` | String | `1MB` | The max size of a single producer batch.
Warning: Kafka has a default limit of 1MB per message in a topic.
Defaults to `8MB` when the provider is `experimental_object_store`.
**It's only used when the provider is `kafka` or `experimental_object_store`**. |
| `wal.consumer_wait_timeout` | String | `100ms` | The consumer wait timeout.
**It's only used when the provider is `kafka`**. |
-| `wal.create_index` | Bool | `false` | Whether to enable WAL index creation.
**It's only used when the provider is `kafka`**. |
-| `wal.dump_index_interval` | String | `60s` | The interval for dumping WAL indexes.
**It's only used when the provider is `kafka`**. |
| `wal.overwrite_entry_start_id` | Bool | `false` | Ignore missing entries during read WAL.
**It's only used when the provider is `kafka`**.
This option ensures that when Kafka messages are deleted, the system
can still successfully replay memtable data without throwing an
out-of-range error.
However, enabling this option might lead to unexpected data loss,
as the system will skip over missing entries instead of treating
them as critical errors. |
| `wal.storage_provider` | String | `""` | The name of the storage provider that holds the WAL objects, an empty name selects the default object store.
**It's only used when the provider is `experimental_object_store`**. |
| `wal.prefix` | String | `wal` | The path prefix of the WAL objects inside the storage provider.
The objects are written under `/datanodes//epochs/`, which is derived from this prefix.
**It's only used when the provider is `experimental_object_store`**. |
diff --git a/config/datanode.example.toml b/config/datanode.example.toml
index a2586908b70..2d72f795189 100644
--- a/config/datanode.example.toml
+++ b/config/datanode.example.toml
@@ -211,14 +211,6 @@ max_batch_bytes = "1MB"
## **It's only used when the provider is `kafka`**.
consumer_wait_timeout = "100ms"
-## Whether to enable WAL index creation.
-## **It's only used when the provider is `kafka`**.
-create_index = false
-
-## The interval for dumping WAL indexes.
-## **It's only used when the provider is `kafka`**.
-dump_index_interval = "60s"
-
## Ignore missing entries during read WAL.
## **It's only used when the provider is `kafka`**.
##
diff --git a/src/cmd/src/datanode/scanbench.rs b/src/cmd/src/datanode/scanbench.rs
index 42eac50ec45..9fbde16fb7b 100644
--- a/src/cmd/src/datanode/scanbench.rs
+++ b/src/cmd/src/datanode/scanbench.rs
@@ -836,7 +836,7 @@ impl ScanbenchCommand {
}
ScanbenchLogStore::Kafka(kafka_config) => {
let log_store = Arc::new(
- KafkaLogStore::try_new(kafka_config, None)
+ KafkaLogStore::try_new(kafka_config)
.await
.map_err(BoxedError::new)
.context(error::BuildCliSnafu)?,
diff --git a/src/common/meta/src/instruction.rs b/src/common/meta/src/instruction.rs
index 827f06e944d..0291f3a0946 100644
--- a/src/common/meta/src/instruction.rs
+++ b/src/common/meta/src/instruction.rs
@@ -381,7 +381,7 @@ pub struct UpgradeRegion {
/// it's helpful to verify whether the leader region is ready.
#[serde(with = "humantime_serde")]
pub replay_timeout: Duration,
- /// The hint for replaying memtable.
+ /// Legacy WAL index location hint, retained for older datanodes.
#[serde(default)]
pub location_id: Option,
#[serde(default, skip_serializing_if = "Option::is_none")]
diff --git a/src/common/wal/src/config.rs b/src/common/wal/src/config.rs
index 0cbebfb20ab..b92c3d13d36 100644
--- a/src/common/wal/src/config.rs
+++ b/src/common/wal/src/config.rs
@@ -346,6 +346,30 @@ mod tests {
assert!(toml::from_str::(toml_str).is_err());
}
+ #[test]
+ fn test_removed_datanode_wal_index_config() {
+ let expected = DatanodeWalConfig::Kafka(DatanodeKafkaConfig::default());
+ for create_index in [false, true] {
+ let config = format!(
+ r#"
+ provider = "kafka"
+ create_index = {create_index}
+ dump_index_interval = "10s"
+ "#
+ );
+ let parsed: DatanodeWalConfig = toml::from_str(&config).unwrap();
+ assert_eq!(parsed, expected);
+ let serialized = toml::to_string(&parsed).unwrap();
+ let table: toml::Table = toml::from_str(&serialized).unwrap();
+ assert!(!table.contains_key("create_index"));
+ assert!(!table.contains_key("dump_index_interval"));
+ assert_eq!(
+ toml::from_str::(&serialized).unwrap(),
+ expected
+ );
+ }
+ }
+
#[test]
fn test_kafka_wal_config_debug_redacts_password() {
let config = MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
diff --git a/src/common/wal/src/config/kafka/datanode.rs b/src/common/wal/src/config/kafka/datanode.rs
index 4003b52e4bc..38836509f0a 100644
--- a/src/common/wal/src/config/kafka/datanode.rs
+++ b/src/common/wal/src/config/kafka/datanode.rs
@@ -40,10 +40,6 @@ pub struct DatanodeKafkaConfig {
pub kafka_topic: KafkaTopicConfig,
// Automatically create topics for WAL.
pub auto_create_topics: bool,
- // Create index for WAL.
- pub create_index: bool,
- #[serde(with = "humantime_serde")]
- pub dump_index_interval: Duration,
/// Internal interval for fetching latest Kafka topic offsets.
#[serde(with = "humantime_serde")]
pub topic_latest_offset_fetch_interval: Duration,
@@ -60,8 +56,6 @@ impl Default for DatanodeKafkaConfig {
consumer_wait_timeout: Duration::from_millis(100),
kafka_topic: KafkaTopicConfig::default(),
auto_create_topics: true,
- create_index: false,
- dump_index_interval: Duration::from_secs(60),
topic_latest_offset_fetch_interval: DEFAULT_TOPIC_LATEST_OFFSET_FETCH_INTERVAL,
overwrite_entry_start_id: false,
}
diff --git a/src/datanode/src/datanode.rs b/src/datanode/src/datanode.rs
index cfbcf948b96..8edcd2ce4a1 100644
--- a/src/datanode/src/datanode.rs
+++ b/src/datanode/src/datanode.rs
@@ -46,7 +46,6 @@ use common_wal::config::raft_engine::RaftEngineConfig;
use file_engine::engine::FileRegionEngine;
use log_store::ObjectStoreLogStore;
use log_store::kafka::log_store::KafkaLogStore;
-use log_store::kafka::{GlobalIndexCollector, default_index_file};
use log_store::noop::log_store::NoopLogStore;
use log_store::raft_engine::log_store::RaftEngineLogStore;
use meta_client::MetaClientRef;
@@ -88,7 +87,7 @@ use crate::heartbeat::HeartbeatTask;
use crate::partition_expr_fetcher::MetaPartitionExprFetcher;
use crate::region_query::DatanodeRegionQueryHandler;
use crate::region_server::{DummyTableProviderFactory, RegionServer};
-use crate::store::{self, new_object_store_without_cache};
+use crate::store;
use crate::utils::{RegionOpenRequests, build_region_open_requests};
/// Datanode service.
@@ -683,29 +682,7 @@ impl DatanodeBuilder {
builder.try_build().await.context(BuildMitoEngineSnafu)?
}
DatanodeWalConfig::Kafka(kafka_config) => {
- if kafka_config.create_index && opts.node_id.is_none() {
- warn!("The WAL index creation only available in distributed mode.")
- }
- let global_index_collector = if kafka_config.create_index
- && let Some(node_id) = opts.node_id
- {
- let operator = new_object_store_without_cache(
- &opts.storage.store,
- &opts.storage.data_home,
- )
- .await?;
- let path = default_index_file(node_id);
- Some(Self::build_global_index_collector(
- kafka_config.dump_index_interval,
- operator,
- path,
- ))
- } else {
- None
- };
-
- let log_store =
- Self::build_kafka_log_store(kafka_config, global_index_collector).await?;
+ let log_store = Self::build_kafka_log_store(kafka_config).await?;
self.topic_stats_reporter = Some(log_store.topic_stats_reporter());
let builder = MitoEngineBuilder::new(
&opts.storage.data_home,
@@ -825,25 +802,13 @@ impl DatanodeBuilder {
}
/// Builds [`KafkaLogStore`].
- async fn build_kafka_log_store(
- config: &DatanodeKafkaConfig,
- global_index_collector: Option,
- ) -> Result> {
- KafkaLogStore::try_new(config, global_index_collector)
+ async fn build_kafka_log_store(config: &DatanodeKafkaConfig) -> Result> {
+ KafkaLogStore::try_new(config)
.await
.map_err(Box::new)
.context(OpenLogStoreSnafu)
.map(Arc::new)
}
-
- /// Builds [`GlobalIndexCollector`]
- fn build_global_index_collector(
- dump_index_interval: Duration,
- operator: object_store::ObjectStore,
- path: String,
- ) -> GlobalIndexCollector {
- GlobalIndexCollector::new(dump_index_interval, operator, path)
- }
}
/// Keeps the first shutdown error and logs the later ones.
diff --git a/src/datanode/src/heartbeat/handler/upgrade_region.rs b/src/datanode/src/heartbeat/handler/upgrade_region.rs
index e2a5677fcad..08f96753ae9 100644
--- a/src/datanode/src/heartbeat/handler/upgrade_region.rs
+++ b/src/datanode/src/heartbeat/handler/upgrade_region.rs
@@ -130,7 +130,6 @@ impl UpgradeRegionsHandler {
let UpgradeRegion {
last_entry_id,
metadata_last_entry_id,
- location_id,
replay_entry_id,
metadata_replay_entry_id,
replay_timeout,
@@ -161,7 +160,6 @@ impl UpgradeRegionsHandler {
set_writable: true,
entry_id: last_entry_id,
metadata_entry_id: metadata_last_entry_id,
- location_id,
checkpoint,
},
));
@@ -320,6 +318,49 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn test_legacy_wal_index_hint() {
+ for location_id in [None, Some(42)] {
+ let mock_region_server = mock_region_server();
+ let region_id = RegionId::new(1024, 1);
+ let (mock_engine, _) =
+ MockRegionEngine::with_custom_apply_fn(MITO_ENGINE_NAME, |engine| {
+ engine.mock_role = Some(Some(RegionRole::Follower));
+ engine.handle_request_mock_fn = Some(Box::new(|_, request| {
+ let store_api::region_request::RegionRequest::Catchup(request) = request
+ else {
+ panic!("expected catchup request");
+ };
+ assert!(request.set_writable);
+ assert_eq!(request.entry_id, Some(10));
+ assert_eq!(request.metadata_entry_id, Some(5));
+ Ok(0)
+ }));
+ });
+ mock_region_server.register_test_region(region_id, mock_engine);
+ let ctx =
+ HandlerContext::new_for_test(mock_region_server, Arc::new(MemoryKvBackend::new()));
+ let reply = UpgradeRegionsHandler::new_test()
+ .handle(
+ &ctx,
+ vec![UpgradeRegion {
+ region_id,
+ last_entry_id: Some(10),
+ metadata_last_entry_id: Some(5),
+ location_id,
+ replay_timeout: Duration::from_secs(1),
+ ..Default::default()
+ }],
+ )
+ .await
+ .unwrap();
+ let reply = &reply.expect_upgrade_regions_reply()[0];
+ assert!(reply.ready);
+ assert!(reply.exists);
+ assert!(reply.error.is_none());
+ }
+ }
+
#[tokio::test]
async fn test_region_not_ready() {
let mock_region_server = mock_region_server();
diff --git a/src/datanode/src/store.rs b/src/datanode/src/store.rs
index 78db3d61033..3369de5aedc 100644
--- a/src/datanode/src/store.rs
+++ b/src/datanode/src/store.rs
@@ -23,25 +23,6 @@ use snafu::prelude::*;
use crate::config::ObjectStoreConfig;
use crate::error::{self, Result};
-pub(crate) async fn new_object_store_without_cache(
- store: &ObjectStoreConfig,
- data_home: &str,
-) -> Result {
- let object_store = new_raw_object_store(store, data_home)
- .await
- .context(error::ObjectStoreSnafu)?;
- // Enable retry layer and cache layer for non-fs object storages
- let object_store = if store.is_object_storage() {
- // Adds retry layer
- with_retry_layers(object_store)
- } else {
- object_store
- };
-
- let object_store = with_instrument_layers(object_store, true);
- Ok(object_store)
-}
-
/// Cleans up old LRU read cache directories that were removed.
fn clean_old_read_cache(store: &ObjectStoreConfig, data_home: &str) {
if !store.is_object_storage() {
diff --git a/src/log-store/Cargo.toml b/src/log-store/Cargo.toml
index 5870110be67..54e0b48c41d 100644
--- a/src/log-store/Cargo.toml
+++ b/src/log-store/Cargo.toml
@@ -30,7 +30,6 @@ common-time.workspace = true
common-wal.workspace = true
crc32fast.workspace = true
dashmap.workspace = true
-delta-encoding = "0.4"
derive_builder.workspace = true
futures.workspace = true
futures-util.workspace = true
diff --git a/src/log-store/src/error.rs b/src/log-store/src/error.rs
index 64d918d5d78..836d1fbfe7d 100644
--- a/src/log-store/src/error.rs
+++ b/src/log-store/src/error.rs
@@ -259,39 +259,6 @@ pub enum Error {
error: tokio::sync::oneshot::error::RecvError,
},
- #[snafu(display("Failed to wait for result of DumpIndex"))]
- WaitDumpIndex {
- #[snafu(implicit)]
- location: Location,
- #[snafu(source)]
- error: tokio::sync::oneshot::error::RecvError,
- },
-
- #[snafu(display("Failed to create writer"))]
- CreateWriter {
- #[snafu(implicit)]
- location: Location,
- #[snafu(source)]
- error: object_store::Error,
- },
-
- #[snafu(display("Failed to write index"))]
- WriteIndex {
- #[snafu(implicit)]
- location: Location,
- #[snafu(source)]
- error: object_store::Error,
- },
-
- #[snafu(display("Failed to read index, path: {path}"))]
- ReadIndex {
- #[snafu(implicit)]
- location: Location,
- #[snafu(source)]
- error: object_store::Error,
- path: String,
- },
-
#[snafu(display(
"The length of meta if exceeded the limit: {}, actual: {}",
limit,
@@ -516,7 +483,6 @@ impl ErrorExt for Error {
| DiscontinuousLogIndex { .. }
| OrderedBatchProducerStopped { .. }
| WaitProduceResultReceiver { .. }
- | WaitDumpIndex { .. }
| MetaLengthExceededLimit { .. }
| ObjectStoreWalStopped { .. }
| WalSegmentFetch { .. } => StatusCode::Internal,
@@ -531,10 +497,7 @@ impl ErrorExt for Error {
ObjectStoreWal { source, .. } => source.status_code(),
// Object store related errors
- CreateWriter { .. }
- | WriteIndex { .. }
- | ReadIndex { .. }
- | WalObjectStore { .. }
+ WalObjectStore { .. }
| StaleWalObject { .. }
| UnconfirmedWalEpochStart { .. }
| Io { .. } => StatusCode::StorageUnavailable,
@@ -563,10 +526,7 @@ impl ErrorExt for Error {
use Error::*;
match self {
- CreateWriter { error, .. }
- | WriteIndex { error, .. }
- | ReadIndex { error, .. }
- | WalObjectStore { error, .. } => retry_hint_from_opendal_error(error),
+ WalObjectStore { error, .. } => retry_hint_from_opendal_error(error),
ObjectStoreWal { source, .. } => source.retry_hint(),
Io { error, .. } => retry_hint_from_io_error(error),
FetchEntry { .. }
diff --git a/src/log-store/src/kafka.rs b/src/log-store/src/kafka.rs
index 2e123cc059c..810e06bb103 100644
--- a/src/log-store/src/kafka.rs
+++ b/src/log-store/src/kafka.rs
@@ -14,7 +14,6 @@
pub(crate) mod client_manager;
pub(crate) mod consumer;
-pub(crate) mod index;
pub mod log_store;
mod periodic_offset_fetcher;
pub(crate) mod producer;
@@ -24,7 +23,6 @@ pub(crate) mod util;
pub(crate) mod worker;
pub use client_manager::DEFAULT_PARTITION;
-pub use index::{GlobalIndexCollector, default_index_file};
use serde::{Deserialize, Serialize};
use store_api::logstore::entry::Id as EntryId;
diff --git a/src/log-store/src/kafka/client_manager.rs b/src/log-store/src/kafka/client_manager.rs
index e594c9148eb..d1cfa2dba41 100644
--- a/src/log-store/src/kafka/client_manager.rs
+++ b/src/log-store/src/kafka/client_manager.rs
@@ -25,7 +25,6 @@ use store_api::logstore::provider::KafkaProvider;
use tokio::sync::{Mutex, RwLock};
use crate::error::{BuildClientSnafu, BuildPartitionClientSnafu, Result, TlsConfigSnafu};
-use crate::kafka::index::{GlobalIndexCollector, NoopCollector};
use crate::kafka::log_store::TopicStat;
use crate::kafka::producer::{OrderedBatchProducer, OrderedBatchProducerRef};
@@ -60,7 +59,6 @@ pub(crate) struct ClientManager {
/// Used to initialize a new [Client].
mutex: Mutex<()>,
instances: RwLock, Client>>,
- global_index_collector: Option,
flush_batch_size: usize,
compression: Compression,
@@ -73,7 +71,6 @@ impl ClientManager {
/// Tries to create a ClientManager.
pub(crate) async fn try_new(
config: &DatanodeKafkaConfig,
- global_index_collector: Option,
topic_stats: Arc, TopicStat>>,
) -> Result {
// Sets backoff config for the top-level kafka client and all clients constructed by it.
@@ -98,7 +95,6 @@ impl ClientManager {
instances: RwLock::new(HashMap::new()),
flush_batch_size: config.max_batch_bytes.as_bytes() as usize,
compression: Compression::Lz4,
- global_index_collector,
topic_stats,
})
}
@@ -151,30 +147,18 @@ impl ClientManager {
.map(Arc::new)?;
let (tx, rx) = OrderedBatchProducer::channel();
- let index_collector = if let Some(global_collector) = self.global_index_collector.as_ref() {
- global_collector
- .provider_level_index_collector(provider.clone(), tx.clone())
- .await
- } else {
- Box::new(NoopCollector)
- };
let producer = Arc::new(OrderedBatchProducer::new(
(tx, rx),
provider.clone(),
client.clone(),
self.compression,
self.flush_batch_size,
- index_collector,
self.topic_stats.clone(),
));
Ok(Client { client, producer })
}
- pub(crate) fn global_index_collector(&self) -> Option<&GlobalIndexCollector> {
- self.global_index_collector.as_ref()
- }
-
/// Lists all topics.
pub(crate) async fn list_topics(&self) -> Vec> {
self.instances.read().await.keys().cloned().collect()
diff --git a/src/log-store/src/kafka/consumer.rs b/src/log-store/src/kafka/consumer.rs
index d064a9bc9ac..d8c22109974 100644
--- a/src/log-store/src/kafka/consumer.rs
+++ b/src/log-store/src/kafka/consumer.rs
@@ -12,6 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
+mod range;
+
+pub(crate) use range::RegionWalRange;
+
use std::collections::VecDeque;
use std::ops::Range;
use std::pin::Pin;
@@ -26,7 +30,7 @@ use pin_project::pin_project;
use rskafka::client::partition::PartitionClient;
use rskafka::record::RecordAndOffset;
-use crate::kafka::index::{NextBatchHint, RegionWalIndexIterator};
+use crate::kafka::consumer::range::NextBatchHint;
pub struct FetchResult {
/// The offsets of the fetched records.
@@ -88,7 +92,7 @@ const MAX_BATCH_SIZE: usize = 52428800;
const AVG_RECORD_SIZE: usize = 256 * 1024;
/// The [`Consumer`] struct represents a Kafka consumer that fetches messages from
-/// a Kafka cluster. Yielding records respecting the [`RegionWalIndexIterator`].
+/// a Kafka cluster. Yielding records respecting the [`RegionWalRange`].
#[pin_project]
#[derive(Builder)]
#[builder(pattern = "owned")]
@@ -142,12 +146,12 @@ impl Consumer {
pub(crate) struct RecordsBuffer {
buffer: VecDeque,
- index: Box,
+ index: RegionWalRange,
}
impl RecordsBuffer {
/// Creates an empty [`RecordsBuffer`]
- pub fn new(index: Box) -> Self {
+ pub fn new(index: RegionWalRange) -> Self {
RecordsBuffer {
buffer: VecDeque::new(),
index,
@@ -297,7 +301,6 @@ mod tests {
use super::*;
use crate::kafka::consumer::{Consumer, RecordsBuffer};
- use crate::kafka::index::{MultipleRegionWalIndexIterator, RegionWalRange, RegionWalVecIndex};
#[derive(Debug)]
struct MockFetchClient {
@@ -345,40 +348,6 @@ mod tests {
}
}
- #[tokio::test]
- async fn test_consumer_with_index() {
- common_telemetry::init_default_ut_logging();
- let record = test_record();
- let record_size = record.approximate_size();
- let mock_client = MockFetchClient {
- record: record.clone(),
- };
- let index = RegionWalVecIndex::new([1, 3, 4, 8, 10, 12], record_size * 3);
- let consumer = Consumer {
- last_high_watermark: -1,
- client: Arc::new(mock_client),
- max_batch_size: usize::MAX,
- max_wait_ms: 500,
- avg_record_size: record_size,
- terminated: false,
- buffer: RecordsBuffer {
- buffer: VecDeque::new(),
- index: Box::new(index),
- },
- fetch_fut: Fuse::terminated(),
- total_fetched_bytes: 0,
- };
-
- let records = consumer.try_collect::>().await.unwrap();
- assert_eq!(
- records
- .into_iter()
- .map(|(x, _)| x.offset)
- .collect::>(),
- vec![1, 3, 4, 8, 10, 12]
- )
- }
-
#[tokio::test]
async fn test_consumer_without_index() {
common_telemetry::init_default_ut_logging();
@@ -396,7 +365,7 @@ mod tests {
terminated: false,
buffer: RecordsBuffer {
buffer: VecDeque::new(),
- index: Box::new(index),
+ index,
},
fetch_fut: Fuse::terminated(),
total_fetched_bytes: 0,
@@ -413,44 +382,34 @@ mod tests {
}
#[tokio::test]
- async fn test_consumer_with_multiple_index() {
- common_telemetry::init_default_ut_logging();
- let record = test_record();
- let mock_client = MockFetchClient {
- record: record.clone(),
- };
-
- let iter0 = Box::new(RegionWalRange::new(0..0, 1024)) as _;
- let iter1 = Box::new(RegionWalVecIndex::new(
- [0, 1, 2, 7, 8, 11],
- record.approximate_size() * 4,
- )) as _;
- let iter2 = Box::new(RegionWalRange::new(12..12, 1024)) as _;
- let iter3 = Box::new(RegionWalRange::new(1024..1028, 1024)) as _;
- let iter = MultipleRegionWalIndexIterator::new([iter0, iter1, iter2, iter3]);
-
- let consumer = Consumer {
- last_high_watermark: -1,
- client: Arc::new(mock_client),
- max_batch_size: usize::MAX,
- max_wait_ms: 500,
- avg_record_size: record.approximate_size(),
- terminated: false,
- buffer: RecordsBuffer {
- buffer: VecDeque::new(),
- index: Box::new(iter),
- },
- fetch_fut: Fuse::terminated(),
- total_fetched_bytes: 0,
- };
-
- let records = consumer.try_collect::>().await.unwrap();
- assert_eq!(
- records
+ async fn test_consumer_range_boundaries() {
+ for (start, end, max_batch_size) in [
+ (0, 0, 1024),
+ (5, 5, 1024),
+ (6, 5, 1024),
+ (5, 6, 1024),
+ (5, 35, 128),
+ ] {
+ let record = test_record();
+ let consumer = ConsumerBuilder::default()
+ .client(Arc::new(MockFetchClient {
+ record: record.clone(),
+ }))
+ .buffer(RecordsBuffer::new(RegionWalRange::new(
+ start..end,
+ max_batch_size,
+ )))
+ .avg_record_size(record.approximate_size())
+ .build()
+ .unwrap();
+ let offsets = consumer
+ .try_collect::>()
+ .await
+ .unwrap()
.into_iter()
- .map(|(x, _)| x.offset)
- .collect::>(),
- [0, 1, 2, 7, 8, 11, 1024, 1025, 1026, 1027]
- )
+ .map(|(record, _)| record.offset as u64)
+ .collect::>();
+ assert_eq!(offsets, (start..end).collect::>());
+ }
}
}
diff --git a/src/log-store/src/kafka/consumer/range.rs b/src/log-store/src/kafka/consumer/range.rs
new file mode 100644
index 00000000000..d4593b6f20c
--- /dev/null
+++ b/src/log-store/src/kafka/consumer/range.rs
@@ -0,0 +1,118 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::cmp::min;
+use std::ops::Range;
+
+use store_api::logstore::EntryId;
+
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) struct NextBatchHint {
+ pub(crate) bytes: usize,
+ pub(crate) len: usize,
+}
+
+/// Represents a range [next_entry_id, end_entry_id) of WAL entries for a region.
+#[derive(Debug)]
+pub struct RegionWalRange {
+ current_entry_id: EntryId,
+ end_entry_id: EntryId,
+ max_batch_size: usize,
+}
+
+impl RegionWalRange {
+ pub fn new(range: Range, max_batch_size: usize) -> Self {
+ Self {
+ current_entry_id: range.start,
+ end_entry_id: range.end,
+ max_batch_size,
+ }
+ }
+
+ fn next_batch_size(&self) -> Option {
+ if self.current_entry_id < self.end_entry_id {
+ Some(self.end_entry_id.saturating_sub(self.current_entry_id))
+ } else {
+ None
+ }
+ }
+
+ pub(crate) fn next_batch_hint(&self, avg_size: usize) -> Option {
+ if let Some(size) = self.next_batch_size() {
+ let bytes = min(size as usize * avg_size, self.max_batch_size);
+ let len = bytes / avg_size;
+
+ return Some(NextBatchHint { bytes, len });
+ }
+
+ None
+ }
+
+ pub(crate) fn peek(&self) -> Option {
+ if self.current_entry_id < self.end_entry_id {
+ Some(self.current_entry_id)
+ } else {
+ None
+ }
+ }
+
+ pub(crate) fn next(&mut self) -> Option {
+ if self.current_entry_id < self.end_entry_id {
+ let next = self.current_entry_id;
+ self.current_entry_id += 1;
+ Some(next)
+ } else {
+ None
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_region_wal_range() {
+ let range = RegionWalRange::new(0..1024, 1024);
+ assert_eq!(
+ range.next_batch_hint(10),
+ Some(NextBatchHint {
+ bytes: 1024,
+ len: 102
+ })
+ );
+
+ let mut range = RegionWalRange::new(0..1, 1024);
+
+ assert_eq!(range.next_batch_size(), Some(1));
+ assert_eq!(range.peek(), Some(0));
+
+ // Advance 1 step
+ assert_eq!(range.next(), Some(0));
+ assert_eq!(range.next_batch_size(), None);
+
+ // Advance 1 step
+ assert_eq!(range.next(), None);
+ assert_eq!(range.next_batch_size(), None);
+ // No effect
+ assert_eq!(range.next(), None);
+ assert_eq!(range.next_batch_size(), None);
+
+ let mut range = RegionWalRange::new(0..0, 1024);
+ assert_eq!(range.next_batch_size(), None);
+ // No effect
+ assert_eq!(range.next(), None);
+ assert_eq!(range.next_batch_size(), None);
+ }
+}
diff --git a/src/log-store/src/kafka/index.rs b/src/log-store/src/kafka/index.rs
deleted file mode 100644
index 23a9378ccb2..00000000000
--- a/src/log-store/src/kafka/index.rs
+++ /dev/null
@@ -1,30 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-mod collector;
-mod encoder;
-mod iterator;
-
-pub use collector::GlobalIndexCollector;
-pub(crate) use collector::{IndexCollector, NoopCollector};
-pub(crate) use encoder::{IndexEncoder, JsonIndexEncoder};
-pub(crate) use iterator::{
- MIN_BATCH_WINDOW_SIZE, NextBatchHint, RegionWalIndexIterator, build_region_wal_index_iterator,
-};
-#[cfg(test)]
-pub(crate) use iterator::{MultipleRegionWalIndexIterator, RegionWalRange, RegionWalVecIndex};
-
-pub fn default_index_file(location_id: u64) -> String {
- format!("__wal/{location_id}/index.json")
-}
diff --git a/src/log-store/src/kafka/index/collector.rs b/src/log-store/src/kafka/index/collector.rs
deleted file mode 100644
index 54b9b21553d..00000000000
--- a/src/log-store/src/kafka/index/collector.rs
+++ /dev/null
@@ -1,414 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::collections::{BTreeSet, HashMap};
-use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::time::Duration;
-
-use common_telemetry::{error, info};
-use futures::future::try_join_all;
-use object_store::ErrorKind;
-use serde::{Deserialize, Serialize};
-use snafu::ResultExt;
-use store_api::logstore::EntryId;
-use store_api::logstore::provider::KafkaProvider;
-use store_api::storage::RegionId;
-use tokio::select;
-use tokio::sync::Mutex as TokioMutex;
-use tokio::sync::mpsc::Sender;
-
-use crate::error::{self, Result};
-use crate::kafka::index::encoder::{DatanodeWalIndexes, IndexEncoder};
-use crate::kafka::index::{JsonIndexEncoder, default_index_file};
-use crate::kafka::worker::{DumpIndexRequest, TruncateIndexRequest, WorkerRequest};
-
-/// The [`IndexCollector`] trait defines the operations for managing and collecting index entries.
-pub trait IndexCollector: Send + Sync {
- /// Appends an [`EntryId`] for a specific region.
- fn append(&mut self, region_id: RegionId, entry_id: EntryId);
-
- /// Truncates the index for a specific region up to a given [`EntryId`].
- ///
- /// It removes all [`EntryId`]s smaller than `entry_id`.
- fn truncate(&mut self, region_id: RegionId, entry_id: EntryId);
-
- /// Sets the latest [`EntryId`].
- fn set_latest_entry_id(&mut self, entry_id: EntryId);
-
- /// Dumps the index.
- fn dump(&mut self, encoder: &dyn IndexEncoder);
-}
-
-/// The [`GlobalIndexCollector`] struct is responsible for managing index entries
-/// across multiple providers.
-#[derive(Debug)]
-pub struct GlobalIndexCollector {
- providers: Arc, Sender>>>,
- operator: object_store::ObjectStore,
- _handle: CollectionTaskHandle,
-}
-
-#[derive(Debug, Clone)]
-pub struct CollectionTask {
- providers: Arc, Sender>>>,
- dump_index_interval: Duration,
- operator: object_store::ObjectStore,
- path: String,
- running: Arc,
-}
-
-impl CollectionTask {
- async fn dump_index(&self) -> Result<()> {
- let encoder = Arc::new(JsonIndexEncoder::default());
- let receivers = {
- let providers = self.providers.lock().await;
- let mut receivers = Vec::with_capacity(providers.len());
- for (provider, sender) in providers.iter() {
- let (req, rx) = DumpIndexRequest::new(encoder.clone());
- receivers.push(rx);
- if sender.send(WorkerRequest::DumpIndex(req)).await.is_err() {
- error!(
- "BackgroundProducerWorker is stopped, topic: {}",
- provider.topic
- )
- }
- }
- receivers
- };
- try_join_all(receivers)
- .await
- .context(error::WaitDumpIndexSnafu)?;
- let bytes = encoder.finish()?;
- let mut writer = self
- .operator
- .writer(&self.path)
- .await
- .context(error::CreateWriterSnafu)?;
- writer.write(bytes).await.context(error::WriteIndexSnafu)?;
- writer.close().await.context(error::WriteIndexSnafu)?;
-
- Ok(())
- }
-
- /// The background task performs two main operations:
- /// - Persists the WAL index to the specified `path` at every `dump_index_interval`.
- /// - Updates the latest index ID for each WAL provider at every `checkpoint_interval`.
- fn run(self) -> CollectionTaskHandle {
- let mut dump_index_interval = tokio::time::interval(self.dump_index_interval);
- let running = self.running.clone();
- let moved_self = self.clone();
- common_runtime::spawn_global(async move {
- loop {
- if !running.load(Ordering::Relaxed) {
- info!("shutdown the index collection task");
- break;
- }
- select! {
- _ = dump_index_interval.tick() => {
- if let Err(err) = moved_self.dump_index().await {
- error!(err; "Failed to persist the WAL index");
- }
- },
- }
- }
- });
- CollectionTaskHandle {
- running: self.running.clone(),
- }
- }
-}
-
-impl Drop for CollectionTaskHandle {
- fn drop(&mut self) {
- self.running.store(false, Ordering::Relaxed);
- }
-}
-
-#[derive(Debug, Default)]
-struct CollectionTaskHandle {
- running: Arc,
-}
-
-impl GlobalIndexCollector {
- /// Constructs a [`GlobalIndexCollector`].
- ///
- /// This method initializes a `GlobalIndexCollector` instance and starts a background task
- /// for managing WAL (Write-Ahead Logging) indexes.
- ///
- /// The background task persists the WAL index to the specified `path` at every `dump_index_interval`.
- pub fn new(
- dump_index_interval: Duration,
- operator: object_store::ObjectStore,
- path: String,
- ) -> Self {
- let providers: Arc, Sender>>> =
- Arc::new(Default::default());
- let task = CollectionTask {
- providers: providers.clone(),
- dump_index_interval,
- operator: operator.clone(),
- path,
- running: Arc::new(AtomicBool::new(true)),
- };
- let handle = task.run();
- Self {
- providers,
- operator,
- _handle: handle,
- }
- }
-
- #[cfg(test)]
- pub fn new_for_test(operator: object_store::ObjectStore) -> Self {
- Self {
- providers: Default::default(),
- operator,
- _handle: Default::default(),
- }
- }
-}
-
-impl GlobalIndexCollector {
- /// Retrieve [`EntryId`]s for a specified `region_id` in `datanode_id`
- /// that are greater than or equal to a given `entry_id`.
- pub(crate) async fn read_remote_region_index(
- &self,
- location_id: u64,
- provider: &KafkaProvider,
- region_id: RegionId,
- entry_id: EntryId,
- ) -> Result, EntryId)>> {
- let path = default_index_file(location_id);
-
- let bytes = match self.operator.read(&path).await {
- Ok(bytes) => bytes.to_vec(),
- Err(err) => {
- if err.kind() == ErrorKind::NotFound {
- return Ok(None);
- } else {
- return Err(err).context(error::ReadIndexSnafu { path });
- }
- }
- };
-
- match DatanodeWalIndexes::decode(&bytes)?.provider(provider) {
- Some(indexes) => {
- let last_index = indexes.last_index();
- let indexes = indexes
- .region(region_id)
- .unwrap_or_default()
- .split_off(&entry_id);
-
- Ok(Some((indexes, last_index)))
- }
- None => Ok(None),
- }
- }
-
- /// Creates a new [`ProviderLevelIndexCollector`] for a specified provider.
- pub(crate) async fn provider_level_index_collector(
- &self,
- provider: Arc,
- sender: Sender,
- ) -> Box {
- self.providers.lock().await.insert(provider.clone(), sender);
- Box::new(ProviderLevelIndexCollector {
- indexes: Default::default(),
- provider,
- })
- }
-
- /// Truncates the index for a specific region up to a given [`EntryId`].
- ///
- /// It removes all [`EntryId`]s smaller than `entry_id`.
- pub(crate) async fn truncate(
- &self,
- provider: &Arc,
- region_id: RegionId,
- entry_id: EntryId,
- ) -> Result<()> {
- if let Some(sender) = self.providers.lock().await.get(provider).cloned()
- && sender
- .send(WorkerRequest::TruncateIndex(TruncateIndexRequest::new(
- region_id, entry_id,
- )))
- .await
- .is_err()
- {
- return error::OrderedBatchProducerStoppedSnafu {}.fail();
- }
-
- Ok(())
- }
-}
-
-/// The [`RegionIndexes`] struct maintains indexes for a collection of regions.
-/// Each region is identified by a `RegionId` and maps to a set of [`EntryId`]s,
-/// representing the entries within that region. It also keeps track of the
-/// latest [`EntryId`] across all regions.
-#[derive(Debug, Clone, Default, Serialize, Deserialize)]
-pub struct RegionIndexes {
- pub(crate) regions: HashMap>,
- pub(crate) latest_entry_id: EntryId,
-}
-
-impl RegionIndexes {
- fn append(&mut self, region_id: RegionId, entry_id: EntryId) {
- self.regions.entry(region_id).or_default().insert(entry_id);
- self.latest_entry_id = self.latest_entry_id.max(entry_id);
- }
-
- fn truncate(&mut self, region_id: RegionId, entry_id: EntryId) {
- if let Some(entry_ids) = self.regions.get_mut(®ion_id) {
- *entry_ids = entry_ids.split_off(&entry_id);
- // The `RegionIndexes` can be empty, keeps to track the latest entry id.
- self.latest_entry_id = self.latest_entry_id.max(entry_id);
- }
- }
-
- fn set_latest_entry_id(&mut self, entry_id: EntryId) {
- self.latest_entry_id = entry_id;
- }
-}
-
-/// The [`ProviderLevelIndexCollector`] struct is responsible for managing index entries
-/// specific to a particular provider.
-#[derive(Debug, Clone)]
-pub struct ProviderLevelIndexCollector {
- indexes: RegionIndexes,
- provider: Arc,
-}
-
-impl IndexCollector for ProviderLevelIndexCollector {
- fn append(&mut self, region_id: RegionId, entry_id: EntryId) {
- self.indexes.append(region_id, entry_id)
- }
-
- fn truncate(&mut self, region_id: RegionId, entry_id: EntryId) {
- self.indexes.truncate(region_id, entry_id)
- }
-
- fn set_latest_entry_id(&mut self, entry_id: EntryId) {
- self.indexes.set_latest_entry_id(entry_id);
- }
-
- fn dump(&mut self, encoder: &dyn IndexEncoder) {
- encoder.encode(&self.provider, &self.indexes)
- }
-}
-
-/// The [`NoopCollector`] struct implements the [`IndexCollector`] trait with no-op methods.
-///
-/// This collector effectively ignores all operations, making it suitable for cases
-/// where index collection is not required or should be disabled.
-pub struct NoopCollector;
-
-impl IndexCollector for NoopCollector {
- fn append(&mut self, _region_id: RegionId, _entry_id: EntryId) {}
-
- fn truncate(&mut self, _region_id: RegionId, _entry_id: EntryId) {}
-
- fn set_latest_entry_id(&mut self, _entry_id: EntryId) {}
-
- fn dump(&mut self, _encoder: &dyn IndexEncoder) {}
-}
-
-#[cfg(test)]
-mod tests {
- use std::collections::{BTreeSet, HashMap};
-
- use store_api::logstore::provider::KafkaProvider;
- use store_api::storage::RegionId;
-
- use crate::kafka::index::JsonIndexEncoder;
- use crate::kafka::index::collector::RegionIndexes;
- use crate::kafka::index::encoder::IndexEncoder;
- use crate::kafka::{GlobalIndexCollector, default_index_file};
-
- #[tokio::test]
- async fn test_read_remote_region_index() {
- let operator =
- object_store::ObjectStore::new(object_store::services::Memory::default()).unwrap();
-
- let path = default_index_file(0);
- let encoder = JsonIndexEncoder::default();
- encoder.encode(
- &KafkaProvider::new("my_topic_0".to_string()),
- &RegionIndexes {
- regions: HashMap::from([(RegionId::new(1, 1), BTreeSet::from([1, 5, 15]))]),
- latest_entry_id: 20,
- },
- );
- let bytes = encoder.finish().unwrap();
- let mut writer = operator.writer(&path).await.unwrap();
- writer.write(bytes).await.unwrap();
- writer.close().await.unwrap();
-
- let collector = GlobalIndexCollector::new_for_test(operator.clone());
- // Index file doesn't exist
- let result = collector
- .read_remote_region_index(
- 1,
- &KafkaProvider::new("my_topic_0".to_string()),
- RegionId::new(1, 1),
- 1,
- )
- .await
- .unwrap();
- assert!(result.is_none());
-
- // RegionId doesn't exist
- let (indexes, last_index) = collector
- .read_remote_region_index(
- 0,
- &KafkaProvider::new("my_topic_0".to_string()),
- RegionId::new(1, 2),
- 5,
- )
- .await
- .unwrap()
- .unwrap();
- assert_eq!(indexes, BTreeSet::new());
- assert_eq!(last_index, 20);
-
- // RegionId(1, 1), Start EntryId: 5
- let (indexes, last_index) = collector
- .read_remote_region_index(
- 0,
- &KafkaProvider::new("my_topic_0".to_string()),
- RegionId::new(1, 1),
- 5,
- )
- .await
- .unwrap()
- .unwrap();
- assert_eq!(indexes, BTreeSet::from([5, 15]));
- assert_eq!(last_index, 20);
-
- // RegionId(1, 1), Start EntryId: 20
- let (indexes, last_index) = collector
- .read_remote_region_index(
- 0,
- &KafkaProvider::new("my_topic_0".to_string()),
- RegionId::new(1, 1),
- 20,
- )
- .await
- .unwrap()
- .unwrap();
- assert_eq!(indexes, BTreeSet::new());
- assert_eq!(last_index, 20);
- }
-}
diff --git a/src/log-store/src/kafka/index/encoder.rs b/src/log-store/src/kafka/index/encoder.rs
deleted file mode 100644
index b8d9e09bfed..00000000000
--- a/src/log-store/src/kafka/index/encoder.rs
+++ /dev/null
@@ -1,181 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::collections::{BTreeSet, HashMap};
-use std::sync::Mutex;
-
-use delta_encoding::{DeltaDecoderExt, DeltaEncoderExt};
-use serde::{Deserialize, Serialize};
-use snafu::ResultExt;
-use store_api::logstore::provider::KafkaProvider;
-use store_api::storage::RegionId;
-
-use crate::error::{self, Result};
-use crate::kafka::index::collector::RegionIndexes;
-
-/// Converts a [`RegionIndexes`] instance into a [`DeltaEncodedRegionIndexes`].
-///
-/// This conversion encodes the index values using delta encoding to reduce storage space.
-impl From<&RegionIndexes> for DeltaEncodedRegionIndexes {
- fn from(value: &RegionIndexes) -> Self {
- let mut regions = HashMap::with_capacity(value.regions.len());
- for (region_id, indexes) in value.regions.iter() {
- let indexes = indexes.iter().copied().deltas().collect();
- regions.insert(*region_id, indexes);
- }
- Self {
- regions,
- last_index: value.latest_entry_id,
- }
- }
-}
-
-/// Represents the delta-encoded version of region indexes for efficient storage.
-#[derive(Debug, Default, Serialize, Deserialize)]
-pub struct DeltaEncodedRegionIndexes {
- regions: HashMap>,
- last_index: u64,
-}
-
-impl DeltaEncodedRegionIndexes {
- /// Retrieves the original (decoded) index values for a given region.
- pub(crate) fn region(&self, region_id: RegionId) -> Option> {
- self.regions
- .get(®ion_id)
- .map(|delta| delta.iter().copied().original().collect::>())
- }
-
- /// Retrieves the last index.
- pub(crate) fn last_index(&self) -> u64 {
- self.last_index
- }
-}
-
-pub trait IndexEncoder: Send + Sync {
- fn encode(&self, provider: &KafkaProvider, region_index: &RegionIndexes);
-
- fn finish(&self) -> Result>;
-}
-
-/// [`DatanodeWalIndexes`] structure holds the WAL indexes for a datanode.
-#[derive(Debug, Default, Serialize, Deserialize)]
-pub(crate) struct DatanodeWalIndexes(HashMap);
-
-impl DatanodeWalIndexes {
- fn insert(&mut self, topic: String, region_index: &RegionIndexes) {
- self.0.insert(topic, region_index.into());
- }
-
- fn encode(&mut self) -> Result> {
- let value = serde_json::to_vec(&self.0).context(error::EncodeJsonSnafu);
- self.0.clear();
- value
- }
-
- pub(crate) fn decode(byte: &[u8]) -> Result {
- serde_json::from_slice(byte).context(error::DecodeJsonSnafu)
- }
-
- /// Retrieves the delta encoded region indexes for a given `provider`.
- pub(crate) fn provider(&self, provider: &KafkaProvider) -> Option<&DeltaEncodedRegionIndexes> {
- self.0.get(&provider.topic)
- }
-}
-
-/// [`JsonIndexEncoder`] encodes the [`RegionIndexes`]s into JSON format.
-#[derive(Debug, Default)]
-pub(crate) struct JsonIndexEncoder {
- buf: Mutex,
-}
-
-impl IndexEncoder for JsonIndexEncoder {
- fn encode(&self, provider: &KafkaProvider, region_index: &RegionIndexes) {
- self.buf
- .lock()
- .unwrap()
- .insert(provider.topic.clone(), region_index);
- }
-
- fn finish(&self) -> Result> {
- let mut buf = self.buf.lock().unwrap();
- buf.encode()
- }
-}
-
-#[cfg(test)]
-mod tests {
- use std::collections::{BTreeSet, HashMap};
-
- use store_api::logstore::provider::KafkaProvider;
- use store_api::storage::RegionId;
-
- use super::{DatanodeWalIndexes, IndexEncoder, JsonIndexEncoder};
- use crate::kafka::index::collector::RegionIndexes;
-
- #[test]
- fn test_json_index_encoder() {
- let encoder = JsonIndexEncoder::default();
- let topic_1 = KafkaProvider::new("my_topic_1".to_string());
- let region_1_indexes = BTreeSet::from([1u64, 2, 4, 5, 20]);
- let region_2_indexes = BTreeSet::from([4u64, 12, 43, 54, 75]);
- encoder.encode(
- &topic_1,
- &RegionIndexes {
- regions: HashMap::from([
- (RegionId::new(1, 1), region_1_indexes.clone()),
- (RegionId::new(1, 2), region_2_indexes.clone()),
- ]),
- latest_entry_id: 1024,
- },
- );
- let topic_2 = KafkaProvider::new("my_topic_2".to_string());
- encoder.encode(
- &topic_2,
- &RegionIndexes {
- regions: HashMap::from([
- (
- RegionId::new(1, 1),
- BTreeSet::from([1024u64, 1025, 1026, 1028, 2048]),
- ),
- (RegionId::new(1, 2), BTreeSet::from([1512])),
- ]),
- latest_entry_id: 2048,
- },
- );
-
- let bytes = encoder.finish().unwrap();
- let datanode_index = DatanodeWalIndexes::decode(&bytes).unwrap();
- assert_eq!(
- datanode_index
- .provider(&topic_1)
- .unwrap()
- .region(RegionId::new(1, 1))
- .unwrap(),
- region_1_indexes,
- );
- assert_eq!(
- datanode_index
- .provider(&topic_1)
- .unwrap()
- .region(RegionId::new(1, 2))
- .unwrap(),
- region_2_indexes,
- );
- assert!(
- datanode_index
- .provider(&KafkaProvider::new("my_topic_3".to_string()))
- .is_none()
- );
- }
-}
diff --git a/src/log-store/src/kafka/index/iterator.rs b/src/log-store/src/kafka/index/iterator.rs
deleted file mode 100644
index ba8e7273d78..00000000000
--- a/src/log-store/src/kafka/index/iterator.rs
+++ /dev/null
@@ -1,483 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::cmp::{max, min};
-use std::collections::{BTreeSet, VecDeque};
-use std::fmt::Debug;
-use std::ops::Range;
-
-use store_api::logstore::EntryId;
-
-use crate::kafka::util::range::{ConvertIndexToRange, MergeRange};
-
-#[derive(Debug, PartialEq, Eq)]
-pub(crate) struct NextBatchHint {
- pub(crate) bytes: usize,
- pub(crate) len: usize,
-}
-
-/// An iterator over WAL (Write-Ahead Log) entries index for a region.
-pub trait RegionWalIndexIterator: Send + Sync + Debug {
- /// Returns next batch hint.
- fn next_batch_hint(&self, avg_size: usize) -> Option;
-
- // Peeks the next EntryId without advancing the iterator.
- fn peek(&self) -> Option;
-
- // Advances the iterator and returns the next EntryId.
- fn next(&mut self) -> Option;
-
- #[cfg(test)]
- fn as_any(&self) -> &dyn std::any::Any;
-}
-
-/// Represents a range [next_entry_id, end_entry_id) of WAL entries for a region.
-#[derive(Debug)]
-pub struct RegionWalRange {
- current_entry_id: EntryId,
- end_entry_id: EntryId,
- max_batch_size: usize,
-}
-
-impl RegionWalRange {
- pub fn new(range: Range, max_batch_size: usize) -> Self {
- Self {
- current_entry_id: range.start,
- end_entry_id: range.end,
- max_batch_size,
- }
- }
-
- fn next_batch_size(&self) -> Option {
- if self.current_entry_id < self.end_entry_id {
- Some(self.end_entry_id.saturating_sub(self.current_entry_id))
- } else {
- None
- }
- }
-}
-
-impl RegionWalIndexIterator for RegionWalRange {
- fn next_batch_hint(&self, avg_size: usize) -> Option {
- if let Some(size) = self.next_batch_size() {
- let bytes = min(size as usize * avg_size, self.max_batch_size);
- let len = bytes / avg_size;
-
- return Some(NextBatchHint { bytes, len });
- }
-
- None
- }
-
- fn peek(&self) -> Option {
- if self.current_entry_id < self.end_entry_id {
- Some(self.current_entry_id)
- } else {
- None
- }
- }
-
- fn next(&mut self) -> Option {
- if self.current_entry_id < self.end_entry_id {
- let next = self.current_entry_id;
- self.current_entry_id += 1;
- Some(next)
- } else {
- None
- }
- }
-
- #[cfg(test)]
- fn as_any(&self) -> &dyn std::any::Any {
- self
- }
-}
-
-pub const MIN_BATCH_WINDOW_SIZE: usize = 4 * 1024 * 1024;
-
-/// Represents an index of Write-Ahead Log entries for a region,
-/// stored as a vector of [EntryId]s.
-#[derive(Debug)]
-pub struct RegionWalVecIndex {
- index: VecDeque,
- min_batch_window_size: usize,
-}
-
-impl RegionWalVecIndex {
- pub fn new>(index: I, min_batch_window_size: usize) -> Self {
- Self {
- index: index.into_iter().collect::>(),
- min_batch_window_size,
- }
- }
-}
-
-impl RegionWalIndexIterator for RegionWalVecIndex {
- fn next_batch_hint(&self, avg_size: usize) -> Option {
- let merger = MergeRange::new(
- ConvertIndexToRange::new(self.index.iter().peekable(), avg_size),
- self.min_batch_window_size,
- );
-
- merger.merge().map(|(range, size)| NextBatchHint {
- bytes: range.end - range.start - 1,
- len: size,
- })
- }
-
- fn peek(&self) -> Option {
- self.index.front().cloned()
- }
-
- fn next(&mut self) -> Option {
- self.index.pop_front()
- }
-
- #[cfg(test)]
- fn as_any(&self) -> &dyn std::any::Any {
- self
- }
-}
-
-/// Represents an iterator over multiple region WAL indexes.
-///
-/// Allowing iteration through multiple WAL indexes.
-#[derive(Debug)]
-pub struct MultipleRegionWalIndexIterator {
- iterator: VecDeque>,
-}
-
-impl MultipleRegionWalIndexIterator {
- pub fn new>>(iterator: I) -> Self {
- Self {
- iterator: iterator.into_iter().collect::>(),
- }
- }
-}
-
-impl RegionWalIndexIterator for MultipleRegionWalIndexIterator {
- fn next_batch_hint(&self, avg_size: usize) -> Option {
- for iter in &self.iterator {
- if let Some(batch) = iter.next_batch_hint(avg_size) {
- return Some(batch);
- }
- }
-
- None
- }
-
- fn peek(&self) -> Option {
- for iter in &self.iterator {
- let peek = iter.peek();
- if peek.is_some() {
- return peek;
- }
- }
-
- None
- }
-
- fn next(&mut self) -> Option {
- while !self.iterator.is_empty() {
- let remove = self.iterator.front().and_then(|iter| iter.peek()).is_none();
- if remove {
- self.iterator.pop_front();
- } else {
- break;
- }
- }
-
- self.iterator.front_mut().and_then(|iter| iter.next())
- }
-
- #[cfg(test)]
- fn as_any(&self) -> &dyn std::any::Any {
- self
- }
-}
-
-/// Builds [`RegionWalIndexIterator`].
-///
-/// Returns None means there are no entries to replay.
-pub fn build_region_wal_index_iterator(
- start_entry_id: EntryId,
- end_entry_id: EntryId,
- region_indexes: Option<(BTreeSet, EntryId)>,
- max_batch_bytes: usize,
- min_window_size: usize,
-) -> Option> {
- if (start_entry_id..end_entry_id).is_empty() {
- return None;
- }
-
- match region_indexes {
- Some((region_indexes, last_index)) => {
- if region_indexes.is_empty() && last_index >= end_entry_id {
- return None;
- }
-
- let mut iterator: Vec> = Vec::with_capacity(2);
- if !region_indexes.is_empty() {
- let index = RegionWalVecIndex::new(region_indexes, min_window_size);
- iterator.push(Box::new(index));
- }
- let known_last_index = max(last_index, start_entry_id);
- if known_last_index < end_entry_id {
- let range = known_last_index..end_entry_id;
- let index = RegionWalRange::new(range, max_batch_bytes);
- iterator.push(Box::new(index));
- }
-
- Some(Box::new(MultipleRegionWalIndexIterator::new(iterator)))
- }
- None => {
- let range = start_entry_id..end_entry_id;
-
- Some(Box::new(RegionWalRange::new(range, max_batch_bytes)))
- }
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
-
- #[test]
- fn test_region_wal_range() {
- let range = RegionWalRange::new(0..1024, 1024);
- assert_eq!(
- range.next_batch_hint(10),
- Some(NextBatchHint {
- bytes: 1024,
- len: 102
- })
- );
-
- let mut range = RegionWalRange::new(0..1, 1024);
-
- assert_eq!(range.next_batch_size(), Some(1));
- assert_eq!(range.peek(), Some(0));
-
- // Advance 1 step
- assert_eq!(range.next(), Some(0));
- assert_eq!(range.next_batch_size(), None);
-
- // Advance 1 step
- assert_eq!(range.next(), None);
- assert_eq!(range.next_batch_size(), None);
- // No effect
- assert_eq!(range.next(), None);
- assert_eq!(range.next_batch_size(), None);
-
- let mut range = RegionWalRange::new(0..0, 1024);
- assert_eq!(range.next_batch_size(), None);
- // No effect
- assert_eq!(range.next(), None);
- assert_eq!(range.next_batch_size(), None);
- }
-
- #[test]
- fn test_region_wal_vec_index() {
- let mut index = RegionWalVecIndex::new([0, 1, 2, 7, 8, 11], 30);
- assert_eq!(
- index.next_batch_hint(10),
- Some(NextBatchHint { bytes: 30, len: 3 })
- );
- assert_eq!(index.peek(), Some(0));
- // Advance 1 step
- assert_eq!(index.next(), Some(0));
- assert_eq!(
- index.next_batch_hint(10),
- Some(NextBatchHint { bytes: 20, len: 2 })
- );
- // Advance 1 step
- assert_eq!(index.next(), Some(1));
- assert_eq!(
- index.next_batch_hint(10),
- Some(NextBatchHint { bytes: 10, len: 1 })
- );
- // Advance 1 step
- assert_eq!(index.next(), Some(2));
- assert_eq!(
- index.next_batch_hint(10),
- Some(NextBatchHint { bytes: 20, len: 2 })
- );
- // Advance 1 step
- assert_eq!(index.next(), Some(7));
- assert_eq!(
- index.next_batch_hint(10),
- Some(NextBatchHint { bytes: 40, len: 2 })
- );
- // Advance 1 step
- assert_eq!(index.next(), Some(8));
- assert_eq!(
- index.next_batch_hint(10),
- Some(NextBatchHint { bytes: 10, len: 1 })
- );
- // Advance 1 step
- assert_eq!(index.next(), Some(11));
- assert_eq!(index.next_batch_hint(10), None);
-
- // No effect
- assert_eq!(index.next(), None);
- assert_eq!(index.next_batch_hint(10), None);
-
- let mut index = RegionWalVecIndex::new([], 1024);
- assert_eq!(index.next_batch_hint(10), None);
- assert_eq!(index.peek(), None);
- // No effect
- assert_eq!(index.peek(), None);
- assert_eq!(index.next(), None);
- assert_eq!(index.next_batch_hint(10), None);
- }
-
- #[test]
- fn test_multiple_region_wal_iterator() {
- let iter0 = Box::new(RegionWalRange::new(0..0, 1024)) as _;
- let iter1 = Box::new(RegionWalVecIndex::new([0, 1, 2, 7, 8, 11], 40)) as _;
- let iter2 = Box::new(RegionWalRange::new(1024..1024, 1024)) as _;
- let mut iter = MultipleRegionWalIndexIterator::new([iter0, iter1, iter2]);
-
- // The next batch is 0, 1, 2
- assert_eq!(
- iter.next_batch_hint(10),
- Some(NextBatchHint { bytes: 30, len: 3 })
- );
- assert_eq!(iter.peek(), Some(0));
- // Advance 1 step
- assert_eq!(iter.next(), Some(0));
-
- // The next batch is 1, 2
- assert_eq!(
- iter.next_batch_hint(10),
- Some(NextBatchHint { bytes: 20, len: 2 })
- );
- assert_eq!(iter.peek(), Some(1));
- // Advance 1 step
- assert_eq!(iter.next(), Some(1));
-
- // The next batch is 2
- assert_eq!(
- iter.next_batch_hint(10),
- Some(NextBatchHint { bytes: 10, len: 1 })
- );
- assert_eq!(iter.peek(), Some(2));
-
- // Advance 1 step
- assert_eq!(iter.next(), Some(2));
- // The next batch is 7, 8, 11
- assert_eq!(
- iter.next_batch_hint(10),
- Some(NextBatchHint { bytes: 50, len: 3 })
- );
- assert_eq!(iter.peek(), Some(7));
-
- // Advance 1 step
- assert_eq!(iter.next(), Some(7));
- // The next batch is 8, 11
- assert_eq!(
- iter.next_batch_hint(10),
- Some(NextBatchHint { bytes: 40, len: 2 })
- );
- assert_eq!(iter.peek(), Some(8));
-
- // Advance 1 step
- assert_eq!(iter.next(), Some(8));
- // The next batch is 11
- assert_eq!(
- iter.next_batch_hint(10),
- Some(NextBatchHint { bytes: 10, len: 1 })
- );
- assert_eq!(iter.peek(), Some(11));
- // Advance 1 step
- assert_eq!(iter.next(), Some(11));
-
- assert_eq!(iter.next_batch_hint(10), None,);
- assert_eq!(iter.peek(), None);
- assert!(!iter.iterator.is_empty());
- assert_eq!(iter.next(), None);
- assert!(iter.iterator.is_empty());
-
- // No effect
- assert_eq!(iter.next(), None);
- assert_eq!(iter.next_batch_hint(10), None,);
- assert_eq!(iter.peek(), None);
- assert_eq!(iter.next(), None);
- }
-
- #[test]
- fn test_build_region_wal_index_iterator() {
- let iterator = build_region_wal_index_iterator(1024, 1024, None, 5, 5);
- assert!(iterator.is_none());
-
- let iterator = build_region_wal_index_iterator(1024, 1023, None, 5, 5);
- assert!(iterator.is_none());
-
- let iterator =
- build_region_wal_index_iterator(1024, 1024, Some((BTreeSet::new(), 1024)), 5, 5);
- assert!(iterator.is_none());
-
- let iterator =
- build_region_wal_index_iterator(1, 1024, Some((BTreeSet::new(), 1024)), 5, 5);
- assert!(iterator.is_none());
-
- let iterator =
- build_region_wal_index_iterator(1, 1024, Some((BTreeSet::new(), 1025)), 5, 5);
- assert!(iterator.is_none());
-
- let iterator = build_region_wal_index_iterator(
- 1,
- 1024,
- Some((BTreeSet::from([512, 756]), 1024)),
- 5,
- 5,
- )
- .unwrap();
- let iter = iterator
- .as_any()
- .downcast_ref::()
- .unwrap();
- assert_eq!(iter.iterator.len(), 1);
- let vec_index = iter.iterator[0]
- .as_any()
- .downcast_ref::()
- .unwrap();
- assert_eq!(vec_index.index, VecDeque::from([512, 756]));
-
- let iterator = build_region_wal_index_iterator(
- 1,
- 1024,
- Some((BTreeSet::from([512, 756]), 1023)),
- 5,
- 5,
- )
- .unwrap();
- let iter = iterator
- .as_any()
- .downcast_ref::()
- .unwrap();
- assert_eq!(iter.iterator.len(), 2);
- let vec_index = iter.iterator[0]
- .as_any()
- .downcast_ref::()
- .unwrap();
- assert_eq!(vec_index.index, VecDeque::from([512, 756]));
- let wal_range = iter.iterator[1]
- .as_any()
- .downcast_ref::()
- .unwrap();
- assert_eq!(wal_range.current_entry_id, 1023);
- assert_eq!(wal_range.end_entry_id, 1024);
- }
-}
diff --git a/src/log-store/src/kafka/log_store.rs b/src/log-store/src/kafka/log_store.rs
index 591a3c88ea5..daa6dfbcf21 100644
--- a/src/log-store/src/kafka/log_store.rs
+++ b/src/log-store/src/kafka/log_store.rs
@@ -31,15 +31,12 @@ use store_api::logstore::entry::{
Entry, Id as EntryId, MultiplePartEntry, MultiplePartHeader, NaiveEntry,
};
use store_api::logstore::provider::{KafkaProvider, Provider};
-use store_api::logstore::{AppendBatchResponse, LogStore, SendableEntryStream, WalIndex};
+use store_api::logstore::{AppendBatchResponse, LogStore, SendableEntryStream};
use store_api::storage::RegionId;
use crate::error::{self, ConsumeRecordSnafu, Error, GetOffsetSnafu, InvalidProviderSnafu, Result};
use crate::kafka::client_manager::{ClientManager, ClientManagerRef};
-use crate::kafka::consumer::{ConsumerBuilder, RecordsBuffer};
-use crate::kafka::index::{
- GlobalIndexCollector, MIN_BATCH_WINDOW_SIZE, build_region_wal_index_iterator,
-};
+use crate::kafka::consumer::{ConsumerBuilder, RecordsBuffer, RegionWalRange};
use crate::kafka::periodic_offset_fetcher::PeriodicOffsetFetcher;
use crate::kafka::producer::OrderedBatchProducerRef;
use crate::kafka::util::record::{
@@ -142,14 +139,9 @@ impl TopicStatsReporter for PeriodicTopicStatsReporter {
impl KafkaLogStore {
/// Tries to create a Kafka log store.
- pub async fn try_new(
- config: &DatanodeKafkaConfig,
- global_index_collector: Option,
- ) -> Result {
+ pub async fn try_new(config: &DatanodeKafkaConfig) -> Result {
let topic_stats = Arc::new(DashMap::new());
- let client_manager = Arc::new(
- ClientManager::try_new(config, global_index_collector, topic_stats.clone()).await?,
- );
+ let client_manager = Arc::new(ClientManager::try_new(config, topic_stats.clone()).await?);
let fetcher = PeriodicOffsetFetcher::new(
config.topic_latest_offset_fetch_interval,
client_manager.clone(),
@@ -291,8 +283,7 @@ impl LogStore for KafkaLogStore {
for (region_id, (producer, records)) in region_grouped_records {
// Safety: `KafkaLogStore::entry` will ensure that the
// `Record`'s `approximate_size` must be less or equal to `max_batch_bytes`.
- region_grouped_result_receivers
- .push((region_id, producer.produce(region_id, records).await?))
+ region_grouped_result_receivers.push((region_id, producer.produce(records).await?))
}
let region_grouped_max_offset =
@@ -318,7 +309,6 @@ impl LogStore for KafkaLogStore {
&self,
provider: &Provider,
mut entry_id: EntryId,
- index: Option,
) -> Result> {
let provider = provider
.as_kafka_provider()
@@ -378,28 +368,13 @@ impl LogStore for KafkaLogStore {
record_num: 0,
});
- let region_indexes = if let (Some(index), Some(collector)) =
- (index, self.client_manager.global_index_collector())
- {
- collector
- .read_remote_region_index(index.location_id, provider, index.region_id, entry_id)
- .await?
- } else {
- None
- };
-
- let Some(iterator) = build_region_wal_index_iterator(
- entry_id,
- end_offset as u64,
- region_indexes,
- self.max_batch_bytes,
- MIN_BATCH_WINDOW_SIZE,
- ) else {
- let range = entry_id..end_offset as u64;
+ let range = entry_id..end_offset as u64;
+ if range.is_empty() {
warn!("No new entries in range {:?} of ns {}", range, provider);
return Ok(futures_util::stream::empty().boxed());
- };
+ }
+ let iterator = RegionWalRange::new(range, self.max_batch_bytes);
debug!("Reading entries with {:?} of ns {}", iterator, provider);
// Safety: Must be ok.
@@ -495,19 +470,11 @@ impl LogStore for KafkaLogStore {
/// that the obsolete entries are deleted immediately.
async fn obsolete(
&self,
- provider: &Provider,
- region_id: RegionId,
- entry_id: EntryId,
+ _provider: &Provider,
+ _region_id: RegionId,
+ _entry_id: EntryId,
) -> Result<()> {
- if let Some(collector) = self.client_manager.global_index_collector() {
- let provider = provider
- .as_kafka_provider()
- .with_context(|| InvalidProviderSnafu {
- expected: KafkaProvider::type_name(),
- actual: provider.type_name(),
- })?;
- collector.truncate(provider, region_id, entry_id).await?;
- }
+ // Kafka WAL pruning is coordinated by metasrv.
Ok(())
}
@@ -682,7 +649,7 @@ mod tests {
max_batch_bytes: ReadableSize::kb(32),
..Default::default()
};
- let logstore = KafkaLogStore::try_new(&config, None).await.unwrap();
+ let logstore = KafkaLogStore::try_new(&config).await.unwrap();
let topic_name = uuid::Uuid::new_v4().to_string();
prepare_topic(&logstore, &topic_name).await;
let provider = Provider::kafka_provider(topic_name);
@@ -708,7 +675,7 @@ mod tests {
// 5 region
assert_eq!(response.last_entry_ids.len(), 5);
let got_entries = logstore
- .read(&provider, 0, None)
+ .read(&provider, 0)
.await
.unwrap()
.try_collect::>()
@@ -766,7 +733,7 @@ mod tests {
max_batch_bytes: ReadableSize::kb(8),
..Default::default()
};
- let logstore = KafkaLogStore::try_new(&config, None).await.unwrap();
+ let logstore = KafkaLogStore::try_new(&config).await.unwrap();
let topic_name = uuid::Uuid::new_v4().to_string();
prepare_topic(&logstore, &topic_name).await;
let provider = Provider::kafka_provider(topic_name);
@@ -792,7 +759,7 @@ mod tests {
// 5 region
assert_eq!(response.last_entry_ids.len(), 5);
let got_entries = logstore
- .read(&provider, 0, None)
+ .read(&provider, 0)
.await
.unwrap()
.try_collect::>()
diff --git a/src/log-store/src/kafka/periodic_offset_fetcher.rs b/src/log-store/src/kafka/periodic_offset_fetcher.rs
index 4063fb75107..b3f8a7f06df 100644
--- a/src/log-store/src/kafka/periodic_offset_fetcher.rs
+++ b/src/log-store/src/kafka/periodic_offset_fetcher.rs
@@ -82,7 +82,6 @@ mod tests {
use common_wal::maybe_skip_kafka_integration_test;
use common_wal::test_util::get_kafka_endpoints;
use store_api::logstore::provider::KafkaProvider;
- use store_api::storage::RegionId;
use super::*;
use crate::kafka::test_util::{prepare, record};
@@ -113,8 +112,7 @@ mod tests {
assert_eq!(current_latest_offset, 0);
let record = vec![record(), record()];
- let region = RegionId::new(1, 1);
- producer.produce(region, record.clone()).await.unwrap();
+ producer.produce(record.clone()).await.unwrap();
tokio::time::sleep(Duration::from_millis(150)).await;
let current_latest_offset = topic_stats.get(&provider).unwrap().latest_offset;
assert_eq!(current_latest_offset, record.len() as u64 - 1);
diff --git a/src/log-store/src/kafka/producer.rs b/src/log-store/src/kafka/producer.rs
index 121bc5ef9cc..d4094a2ba00 100644
--- a/src/log-store/src/kafka/producer.rs
+++ b/src/log-store/src/kafka/producer.rs
@@ -19,11 +19,9 @@ use dashmap::DashMap;
use rskafka::client::partition::{Compression, OffsetAt, PartitionClient};
use rskafka::record::Record;
use store_api::logstore::provider::KafkaProvider;
-use store_api::storage::RegionId;
use tokio::sync::mpsc::{self, Receiver, Sender};
use crate::error::{self, Result};
-use crate::kafka::index::IndexCollector;
use crate::kafka::log_store::TopicStat;
use crate::kafka::worker::{BackgroundProducerWorker, ProduceResultHandle, WorkerRequest};
use crate::metrics::{
@@ -57,7 +55,6 @@ impl OrderedBatchProducer {
client: Arc,
compression: Compression,
max_batch_bytes: usize,
- index_collector: Box,
topic_stats: Arc, TopicStat>>,
) -> Self {
let mut worker = BackgroundProducerWorker {
@@ -67,7 +64,6 @@ impl OrderedBatchProducer {
receiver: rx,
request_batch_size: REQUEST_BATCH_SIZE,
max_batch_bytes,
- index_collector,
topic_stats,
};
tokio::spawn(async move { worker.run().await });
@@ -81,12 +77,8 @@ impl OrderedBatchProducer {
///
/// ## Panic
/// Panic if any [Record]'s `approximate_size` > `max_batch_bytes`.
- pub(crate) async fn produce(
- &self,
- region_id: RegionId,
- batch: Vec,
- ) -> Result {
- let (req, handle) = WorkerRequest::new_produce_request(region_id, batch);
+ pub(crate) async fn produce(&self, batch: Vec) -> Result {
+ let (req, handle) = WorkerRequest::new_produce_request(batch);
if self.sender.send(req).await.is_err() {
warn!("OrderedBatchProducer is already exited");
return error::OrderedBatchProducerStoppedSnafu {}.fail();
@@ -172,10 +164,8 @@ mod tests {
use rskafka::client::partition::Compression;
use rskafka::protocol::error::Error as ProtocolError;
use rskafka::record::Record;
- use store_api::storage::RegionId;
use super::*;
- use crate::kafka::index::NoopCollector;
use crate::kafka::producer::OrderedBatchProducer;
use crate::kafka::test_util::record;
@@ -246,17 +236,12 @@ mod tests {
client.clone(),
Compression::NoCompression,
ReadableSize((record.approximate_size() * 2) as u64).as_bytes() as usize,
- Box::new(NoopCollector),
Arc::new(DashMap::new()),
);
- let region_id = RegionId::new(1, 1);
// Produces 3 records
let handle = producer
- .produce(
- region_id,
- vec![record.clone(), record.clone(), record.clone()],
- )
+ .produce(vec![record.clone(), record.clone(), record.clone()])
.await
.unwrap();
assert_eq!(handle.wait().await.unwrap(), 2);
@@ -264,17 +249,14 @@ mod tests {
// Produces 2 records
let handle = producer
- .produce(region_id, vec![record.clone(), record.clone()])
+ .produce(vec![record.clone(), record.clone()])
.await
.unwrap();
assert_eq!(handle.wait().await.unwrap(), 4);
assert_eq!(client.batch_sizes.lock().unwrap().as_slice(), &[2, 1, 2]);
// Produces 1 records
- let handle = producer
- .produce(region_id, vec![record.clone()])
- .await
- .unwrap();
+ let handle = producer.produce(vec![record.clone()]).await.unwrap();
assert_eq!(handle.wait().await.unwrap(), 5);
assert_eq!(client.batch_sizes.lock().unwrap().as_slice(), &[2, 1, 2, 1]);
}
@@ -295,36 +277,25 @@ mod tests {
client.clone(),
Compression::NoCompression,
ReadableSize((record.approximate_size() * 2) as u64).as_bytes() as usize,
- Box::new(NoopCollector),
Arc::new(DashMap::new()),
);
- let region_id = RegionId::new(1, 1);
let mut futures = FuturesUnordered::new();
futures.push(
producer
- .produce(
- region_id,
- vec![record.clone(), record.clone(), record.clone()],
- )
+ .produce(vec![record.clone(), record.clone(), record.clone()])
.await
.unwrap()
.wait(),
);
futures.push(
producer
- .produce(region_id, vec![record.clone(), record.clone()])
- .await
- .unwrap()
- .wait(),
- );
- futures.push(
- producer
- .produce(region_id, vec![record.clone()])
+ .produce(vec![record.clone(), record.clone()])
.await
.unwrap()
.wait(),
);
+ futures.push(producer.produce(vec![record.clone()]).await.unwrap().wait());
futures.next().await.unwrap().unwrap_err();
futures.next().await.unwrap().unwrap_err();
@@ -348,27 +319,17 @@ mod tests {
client.clone(),
Compression::NoCompression,
ReadableSize((record.approximate_size() * 2) as u64).as_bytes() as usize,
- Box::new(NoopCollector),
Arc::new(DashMap::new()),
);
- let region_id = RegionId::new(1, 1);
let a = producer
- .produce(
- region_id,
- vec![record.clone(), record.clone(), record.clone()],
- )
+ .produce(vec![record.clone(), record.clone(), record.clone()])
.await
.unwrap()
.wait()
.fuse();
- let b = producer
- .produce(region_id, vec![record])
- .await
- .unwrap()
- .wait()
- .fuse();
+ let b = producer.produce(vec![record]).await.unwrap().wait().fuse();
let mut b = Box::pin(b);
diff --git a/src/log-store/src/kafka/test_util.rs b/src/log-store/src/kafka/test_util.rs
index cb4855e467b..99e824a82c8 100644
--- a/src/log-store/src/kafka/test_util.rs
+++ b/src/log-store/src/kafka/test_util.rs
@@ -70,7 +70,7 @@ pub(crate) async fn prepare(
..Default::default()
};
let high_watermark = Arc::new(DashMap::new());
- let manager = ClientManager::try_new(&config, None, high_watermark)
+ let manager = ClientManager::try_new(&config, high_watermark)
.await
.unwrap();
diff --git a/src/log-store/src/kafka/util.rs b/src/log-store/src/kafka/util.rs
index e871feca429..562d107b429 100644
--- a/src/log-store/src/kafka/util.rs
+++ b/src/log-store/src/kafka/util.rs
@@ -12,5 +12,4 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-pub(crate) mod range;
pub(crate) mod record;
diff --git a/src/log-store/src/kafka/util/range.rs b/src/log-store/src/kafka/util/range.rs
deleted file mode 100644
index 282d0aba185..00000000000
--- a/src/log-store/src/kafka/util/range.rs
+++ /dev/null
@@ -1,146 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::cmp::{max, min};
-use std::iter::Peekable;
-use std::ops::Range;
-
-use store_api::logstore::EntryId;
-
-/// Convert a sequence of [`EntryId`]s into size ranges.
-pub(crate) struct ConvertIndexToRange<'a, I: Iterator- > {
- base: Option,
- iter: Peekable,
- avg_size: usize,
-}
-
-impl<'a, I: Iterator
- > ConvertIndexToRange<'a, I> {
- pub fn new(mut iter: Peekable, avg_size: usize) -> Self {
- let base = iter.peek().cloned().cloned();
-
- Self {
- base,
- iter,
- avg_size,
- }
- }
-}
-
-impl<'a, I: Iterator
- > Iterator for ConvertIndexToRange<'a, I> {
- type Item = Range;
-
- fn next(&mut self) -> Option {
- let (base, val) = (&self.base?, self.iter.next()?);
- let start = (*val - *base) as usize * self.avg_size;
- let end = start + self.avg_size + 1;
- Some(start..end)
- }
-}
-
-/// Merge all ranges smaller than the `window_size`.
-///
-/// e.g.,
-///
-/// Case 1
-/// - input: range: [(0..3), (5..6), (5..8)], window_size: 6
-/// - output: range: (0..6)
-///
-/// Case 2
-/// - input: range: [(0..3)], window_size: 6
-/// - output: range: (0..3)
-pub(crate) struct MergeRange>> {
- iter: I,
- window_size: usize,
-}
-
-impl>> MergeRange {
- pub fn new(iter: I, window_size: usize) -> Self {
- Self { iter, window_size }
- }
-}
-
-/// Merges ranges.
-fn merge(this: &mut Range, other: &Range) {
- this.start = min(this.start, other.start);
- this.end = max(this.end, other.end);
-}
-
-impl>> MergeRange {
- /// Calculates the size of the next merged range.
- pub(crate) fn merge(mut self) -> Option<(Range, usize)> {
- let mut merged_range = self.iter.next();
- let this = merged_range.as_mut()?;
- let mut merged = 1;
- for next in self.iter {
- let window = next.start - this.start;
- if window > self.window_size {
- break;
- } else {
- merge(this, &next);
- merged += 1;
- }
- }
- merged_range.map(|range| (range, merged))
- }
-}
-
-#[cfg(test)]
-#[allow(clippy::single_range_in_vec_init)]
-mod tests {
- use super::*;
-
- #[test]
- fn test_convert_index_to_range() {
- let avg_size = 1024;
- let index = [1u64, 4, 10, 15];
- let mut converter = ConvertIndexToRange::new(index.iter().peekable(), avg_size);
-
- assert_eq!(converter.next(), Some(0..avg_size + 1));
- assert_eq!(converter.next(), Some(3 * avg_size..4 * avg_size + 1));
- assert_eq!(converter.next(), Some(9 * avg_size..10 * avg_size + 1));
- assert_eq!(converter.next(), Some(14 * avg_size..15 * avg_size + 1));
- assert_eq!(converter.next(), None);
-
- let index = [];
- let mut converter = ConvertIndexToRange::new(index.iter().peekable(), avg_size);
- assert_eq!(converter.next(), None);
- }
-
- #[test]
- fn test_merge_range() {
- let size_range = [(10usize..13), (12..14), (16..18), (19..29)];
- let merger = MergeRange::new(size_range.into_iter(), 9);
- assert_eq!(merger.merge(), Some((10..29, 4)));
-
- let size_range = [(10usize..13), (12..14), (16..18)];
- let merger = MergeRange::new(size_range.into_iter(), 5);
- assert_eq!(merger.merge(), Some((10..14, 2)));
-
- let size_range = [(10usize..13), (15..17), (16..18)];
- let merger = MergeRange::new(size_range.into_iter(), 5);
- assert_eq!(merger.merge(), Some((10..17, 2)));
-
- let size_range = [(10usize..13)];
- let merger = MergeRange::new(size_range.into_iter(), 4);
- assert_eq!(merger.merge(), Some((10..13, 1)));
-
- let size_range = [(10usize..13)];
- let merger = MergeRange::new(size_range.into_iter(), 2);
- assert_eq!(merger.merge(), Some((10..13, 1)));
-
- let size_range = [];
- let merger = MergeRange::new(size_range.into_iter(), 2);
- assert_eq!(merger.merge(), None);
- }
-}
diff --git a/src/log-store/src/kafka/worker.rs b/src/log-store/src/kafka/worker.rs
index 7e3d00672c8..d73ab076607 100644
--- a/src/log-store/src/kafka/worker.rs
+++ b/src/log-store/src/kafka/worker.rs
@@ -12,7 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-pub(crate) mod dump_index;
pub(crate) mod fetch_latest_offset;
pub(crate) mod flush;
pub(crate) mod produce;
@@ -25,76 +24,31 @@ use futures::future::try_join_all;
use rskafka::client::partition::Compression;
use rskafka::record::Record;
use snafu::{OptionExt, ResultExt};
-use store_api::logstore::EntryId;
use store_api::logstore::provider::KafkaProvider;
-use store_api::storage::RegionId;
use tokio::sync::mpsc::Receiver;
use tokio::sync::oneshot::{self};
use crate::error::{self, NoMaxValueSnafu, Result};
-use crate::kafka::index::{IndexCollector, IndexEncoder};
use crate::kafka::log_store::TopicStat;
use crate::kafka::producer::ProducerClient;
pub(crate) enum WorkerRequest {
Produce(ProduceRequest),
- TruncateIndex(TruncateIndexRequest),
- DumpIndex(DumpIndexRequest),
FetchLatestOffset,
}
impl WorkerRequest {
- pub(crate) fn new_produce_request(
- region_id: RegionId,
- batch: Vec,
- ) -> (WorkerRequest, ProduceResultHandle) {
+ pub(crate) fn new_produce_request(batch: Vec) -> (WorkerRequest, ProduceResultHandle) {
let (tx, rx) = oneshot::channel();
(
- WorkerRequest::Produce(ProduceRequest {
- region_id,
- batch,
- sender: tx,
- }),
+ WorkerRequest::Produce(ProduceRequest { batch, sender: tx }),
ProduceResultHandle { receiver: rx },
)
}
}
-pub(crate) struct DumpIndexRequest {
- encoder: Arc,
- sender: oneshot::Sender<()>,
-}
-
-impl DumpIndexRequest {
- pub fn new(encoder: Arc) -> (DumpIndexRequest, oneshot::Receiver<()>) {
- let (tx, rx) = oneshot::channel();
- (
- DumpIndexRequest {
- encoder,
- sender: tx,
- },
- rx,
- )
- }
-}
-
-pub(crate) struct TruncateIndexRequest {
- region_id: RegionId,
- entry_id: EntryId,
-}
-
-impl TruncateIndexRequest {
- pub fn new(region_id: RegionId, entry_id: EntryId) -> Self {
- Self {
- region_id,
- entry_id,
- }
- }
-}
-
pub(crate) struct ProduceRequest {
- region_id: RegionId,
batch: Vec,
sender: oneshot::Sender,
}
@@ -142,7 +96,6 @@ impl ProduceResultReceiver {
pub(crate) struct PendingRequest {
batch: Vec,
- region_ids: Vec,
size: usize,
sender: oneshot::Sender>>,
}
@@ -159,8 +112,6 @@ pub(crate) struct BackgroundProducerWorker {
pub(crate) request_batch_size: usize,
/// Max bytes size for a single flush.
pub(crate) max_batch_bytes: usize,
- /// Collecting ids of WAL entries.
- pub(crate) index_collector: Box,
/// The stats of each topic.
pub(crate) topic_stats: Arc, TopicStat>>,
}
@@ -195,11 +146,6 @@ impl BackgroundProducerWorker {
for req in buffer.drain(..) {
match req {
WorkerRequest::Produce(req) => produce_requests.push(req),
- WorkerRequest::TruncateIndex(TruncateIndexRequest {
- region_id,
- entry_id,
- }) => self.index_collector.truncate(region_id, entry_id),
- WorkerRequest::DumpIndex(req) => self.dump_index(req).await,
WorkerRequest::FetchLatestOffset => {
self.fetch_latest_offset().await;
}
diff --git a/src/log-store/src/kafka/worker/dump_index.rs b/src/log-store/src/kafka/worker/dump_index.rs
deleted file mode 100644
index 0de02c5c9e7..00000000000
--- a/src/log-store/src/kafka/worker/dump_index.rs
+++ /dev/null
@@ -1,39 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_telemetry::error;
-use rskafka::client::partition::OffsetAt;
-use snafu::ResultExt;
-
-use crate::error;
-use crate::kafka::worker::{BackgroundProducerWorker, DumpIndexRequest};
-
-impl BackgroundProducerWorker {
- pub(crate) async fn dump_index(&mut self, req: DumpIndexRequest) {
- match self
- .client
- .get_offset(OffsetAt::Latest)
- .await
- .context(error::GetOffsetSnafu {
- topic: &self.provider.topic,
- }) {
- Ok(offset) => {
- self.index_collector.set_latest_entry_id(offset as u64);
- self.index_collector.dump(req.encoder.as_ref());
- let _ = req.sender.send(());
- }
- Err(err) => error!(err; "Failed to do checkpoint"),
- }
- }
-}
diff --git a/src/log-store/src/kafka/worker/flush.rs b/src/log-store/src/kafka/worker/flush.rs
index 7348c4ec6ff..4c8e6eab034 100644
--- a/src/log-store/src/kafka/worker/flush.rs
+++ b/src/log-store/src/kafka/worker/flush.rs
@@ -24,7 +24,6 @@ impl BackgroundProducerWorker {
&mut self,
PendingRequest {
batch,
- region_ids,
sender,
size: _size,
}: PendingRequest,
@@ -38,10 +37,6 @@ impl BackgroundProducerWorker {
if let Ok(result) = &result {
let total_record_size = result.encoded_request_size as u64;
- for (idx, region_id) in result.offsets.iter().zip(region_ids) {
- self.index_collector.append(region_id, *idx as u64);
- }
-
let max_offset = result.offsets.iter().max().cloned().unwrap_or_default() as u64;
self.topic_stats
.entry(self.provider.clone())
diff --git a/src/log-store/src/kafka/worker/produce.rs b/src/log-store/src/kafka/worker/produce.rs
index 465d19a57d8..d9e7ba25654 100644
--- a/src/log-store/src/kafka/worker/produce.rs
+++ b/src/log-store/src/kafka/worker/produce.rs
@@ -30,16 +30,10 @@ impl BackgroundProducerWorker {
max_batch_bytes: usize,
) -> Vec {
let mut records_buffer = vec![];
- let mut region_ids = vec![];
let mut batch_size = 0;
let mut pending_requests = Vec::with_capacity(requests.len());
- for ProduceRequest {
- batch,
- sender,
- region_id,
- } in std::mem::take(requests)
- {
+ for ProduceRequest { batch, sender } in std::mem::take(requests) {
let mut receiver = ProduceResultReceiver::default();
for record in batch {
assert!(record.approximate_size() <= max_batch_bytes);
@@ -48,7 +42,6 @@ impl BackgroundProducerWorker {
let (tx, rx) = oneshot::channel();
pending_requests.push(PendingRequest {
batch: std::mem::take(&mut records_buffer),
- region_ids: std::mem::take(&mut region_ids),
size: batch_size,
sender: tx,
});
@@ -58,7 +51,6 @@ impl BackgroundProducerWorker {
batch_size += record.approximate_size();
records_buffer.push(record);
- region_ids.push(region_id);
}
// The remaining records.
if batch_size > 0 {
@@ -66,7 +58,6 @@ impl BackgroundProducerWorker {
let (tx, rx) = oneshot::channel();
pending_requests.push(PendingRequest {
batch: std::mem::take(&mut records_buffer),
- region_ids: std::mem::take(&mut region_ids),
size: batch_size,
sender: tx,
});
diff --git a/src/log-store/src/noop/log_store.rs b/src/log-store/src/noop/log_store.rs
index 11b77e79108..8e9d22d8f1d 100644
--- a/src/log-store/src/noop/log_store.rs
+++ b/src/log-store/src/noop/log_store.rs
@@ -17,7 +17,7 @@ use std::collections::HashMap;
use futures::stream;
use store_api::logstore::entry::{Entry, NaiveEntry};
use store_api::logstore::provider::Provider;
-use store_api::logstore::{AppendBatchResponse, EntryId, LogStore, SendableEntryStream, WalIndex};
+use store_api::logstore::{AppendBatchResponse, EntryId, LogStore, SendableEntryStream};
use store_api::storage::RegionId;
use crate::error::{Error, Result};
@@ -45,7 +45,6 @@ impl LogStore for NoopLogStore {
&self,
_provider: &Provider,
_entry_id: EntryId,
- _index: Option,
) -> Result> {
Ok(Box::pin(stream::empty()))
}
diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs
index 1d75739350a..530f76b9ff6 100644
--- a/src/log-store/src/object_store_wal/store.rs
+++ b/src/log-store/src/object_store_wal/store.rs
@@ -33,7 +33,7 @@ use object_store::ObjectStore;
use snafu::{IntoError, OptionExt, ResultExt, ensure};
use store_api::logstore::entry::{Entry, NaiveEntry};
use store_api::logstore::provider::{ObjectStoreProvider, Provider};
-use store_api::logstore::{AppendBatchResponse, EntryId, LogStore, SendableEntryStream, WalIndex};
+use store_api::logstore::{AppendBatchResponse, EntryId, LogStore, SendableEntryStream};
use store_api::storage::RegionId;
use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot, watch};
use tokio::time::MissedTickBehavior;
@@ -755,7 +755,6 @@ impl LogStore for ObjectStoreLogStore {
&self,
provider: &Provider,
entry_id: EntryId,
- _index: Option,
) -> Result> {
self.check_terminal()?;
let region_id = self.region_of(provider)?;
@@ -3596,7 +3595,7 @@ mod tests {
let errors = [
store.append_batch(vec![entry]).await.unwrap_err(),
store.latest_entry_id(foreign).unwrap_err(),
- store.read(foreign, 0, None).await.err().unwrap(),
+ store.read(foreign, 0).await.err().unwrap(),
store.create_namespace(foreign).await.unwrap_err(),
store.delete_namespace(foreign).await.unwrap_err(),
store.obsolete(foreign, region_id, 1).await.unwrap_err(),
@@ -3705,7 +3704,7 @@ mod tests {
let errors = [
store.append_batch(Vec::new()).await.unwrap_err(),
store.latest_entry_id(&provider).unwrap_err(),
- store.read(&provider, 0, None).await.err().unwrap(),
+ store.read(&provider, 0).await.err().unwrap(),
store.create_namespace(&provider).await.unwrap_err(),
store.delete_namespace(&provider).await.unwrap_err(),
store.list_namespaces().await.unwrap_err(),
@@ -3880,7 +3879,7 @@ mod tests {
start: EntryId,
) -> Vec {
store
- .read(&provider(region_id), start, None)
+ .read(&provider(region_id), start)
.await
.unwrap()
.try_collect::>()
@@ -4198,7 +4197,7 @@ mod tests {
);
reads.lock().unwrap().clear();
let error = store
- .read(&provider(region(2)), 0, None)
+ .read(&provider(region(2)), 0)
.await
.unwrap()
.try_collect::>()
@@ -4262,7 +4261,7 @@ mod tests {
object_store.delete(&path).await.unwrap();
let error = store
- .read(&provider(region(1)), 0, None)
+ .read(&provider(region(1)), 0)
.await
.unwrap()
.try_collect::>()
@@ -4369,7 +4368,7 @@ mod tests {
let objects = depth + 4;
let (store, io, mut parked) = open_parking_range_reads(objects).await;
let reads = METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count();
- let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap();
+ let mut stream = store.read(&provider(region(1)), 0).await.unwrap();
// The first poll starts the fetches of the first `depth` objects.
assert!(stream.next().now_or_never().is_none());
@@ -4424,7 +4423,7 @@ mod tests {
let failed = 5;
let (store, io, mut parked) = open_parking_range_reads(objects).await;
let reads = METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count();
- let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap();
+ let mut stream = store.read(&provider(region(1)), 0).await.unwrap();
assert!(stream.next().now_or_never().is_none());
let mut wave = parked_operations(&mut parked, READ_PREFETCH_DEPTH).await;
@@ -4491,7 +4490,7 @@ mod tests {
let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get();
// A collection removes object 9 after the read listed it.
- let stream = store.read(&provider(region(1)), 0, None).await.unwrap();
+ let stream = store.read(&provider(region(1)), 0).await.unwrap();
store.catalog.write().unwrap().remove_object(collected);
object_store
.delete(&object_path(&object_store, collected))
@@ -4530,7 +4529,7 @@ mod tests {
let objects = READ_PREFETCH_DEPTH as u64 + 4;
let (store, _io, mut parked) = open_parking_range_reads(objects).await;
let reads = METRIC_OBJECT_STORE_WAL_READ_SECONDS.get_sample_count();
- let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap();
+ let mut stream = store.read(&provider(region(1)), 0).await.unwrap();
assert!(stream.next().now_or_never().is_none());
let mut wave = parked_operations(&mut parked, READ_PREFETCH_DEPTH).await;
wave.remove(&0).unwrap().send(true).unwrap();
@@ -4556,7 +4555,7 @@ mod tests {
let (io, _) = RecordingIo::over(object_store);
let store = open_over(io.clone(), &eager()).await;
*io.panic_range_read_of.lock().unwrap() = Some(1);
- let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap();
+ let mut stream = store.read(&provider(region(1)), 0).await.unwrap();
assert_eq!(object_entries([0]), next_batch(&mut stream).await);
let error = timeout(WAIT, stream.next())
.await
@@ -4585,7 +4584,7 @@ mod tests {
.clone()
.try_acquire_many_owned(READ_PREFETCH_BYTES as u32)
.unwrap();
- let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap();
+ let mut stream = store.read(&provider(region(1)), 0).await.unwrap();
assert!(stream.next().now_or_never().is_none());
let (object_seq, release) = next_parked_operation(&mut parked).await;
assert_eq!(0, object_seq);
@@ -4638,7 +4637,7 @@ mod tests {
// The first poll fetches object 0, and object 1 with its bytes
// reserved; the read of object 1 blocks its thread.
- let mut stream = store.read(&provider(region(1)), 0, None).await.unwrap();
+ let mut stream = store.read(&provider(region(1)), 0).await.unwrap();
let _ = stream.next().now_or_never();
began_rx.recv_timeout(WAIT).unwrap();
assert!(budget.available_permits() < READ_PREFETCH_BYTES);
@@ -5771,11 +5770,7 @@ mod tests {
assert_stopped(&timeout(WAIT, pending).await.unwrap().unwrap().unwrap_err());
for error in [
store.latest_entry_id(&provider(region_id)).unwrap_err(),
- store
- .read(&provider(region_id), 1, None)
- .await
- .err()
- .unwrap(),
+ store.read(&provider(region_id), 1).await.err().unwrap(),
] {
assert!(
matches!(unwrap_shared(&error), Error::WalObjectConflict { .. }),
@@ -7108,7 +7103,7 @@ mod tests {
append(&store, region_id, "a2").await.unwrap();
// The read lists objects 1 and 2 before a collection deletes object 1.
- let stream = store.read(&provider(region_id), 0, None).await.unwrap();
+ let stream = store.read(&provider(region_id), 0).await.unwrap();
store
.obsolete(&provider(region_id), region_id, id(1, 1))
.await
@@ -7152,7 +7147,7 @@ mod tests {
// The read lists objects 1 and 2. The delete of object 1 removes
// it, but its outcome is not reported yet.
- let stream = store.read(&provider(region_id), 0, None).await.unwrap();
+ let stream = store.read(&provider(region_id), 0).await.unwrap();
store
.obsolete(&provider(region_id), region_id, id(1, 1))
.await
@@ -7905,7 +7900,7 @@ mod tests {
/// Asserts that a read of the poisoned store fails with the terminal error.
async fn assert_read_is_terminal(store: &ObjectStoreLogStore, region_id: RegionId) {
let error = store
- .read(&provider(region_id), 1, None)
+ .read(&provider(region_id), 1)
.await
.err()
.expect("a read of a poisoned store must fail");
@@ -9183,7 +9178,7 @@ mod tests {
.unwrap();
*io.fail_range_read_at.lock().unwrap() = Some((1, segment.start));
let error = store
- .read(&provider(region_id), 1, None)
+ .read(&provider(region_id), 1)
.await
.unwrap()
.try_collect::>()
diff --git a/src/log-store/src/raft_engine/log_store.rs b/src/log-store/src/raft_engine/log_store.rs
index 923110858b9..32a028f1691 100644
--- a/src/log-store/src/raft_engine/log_store.rs
+++ b/src/log-store/src/raft_engine/log_store.rs
@@ -25,7 +25,7 @@ use raft_engine::{Config, Engine, LogBatch, MessageExt, ReadableSize, RecoveryMo
use snafu::{OptionExt, ResultExt, ensure};
use store_api::logstore::entry::{Entry, Id as EntryId, NaiveEntry};
use store_api::logstore::provider::{Provider, RaftEngineProvider};
-use store_api::logstore::{AppendBatchResponse, LogStore, SendableEntryStream, WalIndex};
+use store_api::logstore::{AppendBatchResponse, LogStore, SendableEntryStream};
use store_api::storage::RegionId;
use crate::error::{
@@ -293,7 +293,6 @@ impl LogStore for RaftEngineLogStore {
&self,
provider: &Provider,
entry_id: EntryId,
- _index: Option,
) -> Result> {
let ns = provider
.as_raft_engine_provider()
@@ -637,7 +636,7 @@ mod tests {
}
let mut entries = HashSet::with_capacity(1024);
let mut s = logstore
- .read(&Provider::raft_engine_provider(1), 0, None)
+ .read(&Provider::raft_engine_provider(1), 0)
.await
.unwrap();
while let Some(r) = s.next().await {
@@ -672,7 +671,7 @@ mod tests {
.is_ok()
);
let entries = logstore
- .read(&Provider::raft_engine_provider(1), 1, None)
+ .read(&Provider::raft_engine_provider(1), 1)
.await
.unwrap()
.collect::>()
@@ -690,7 +689,7 @@ mod tests {
let entries = collect_entries(
logstore
- .read(&Provider::raft_engine_provider(1), 1, None)
+ .read(&Provider::raft_engine_provider(1), 1)
.await
.unwrap(),
)
@@ -776,7 +775,7 @@ mod tests {
logstore.obsolete(&namespace, region_id, 100).await.unwrap();
assert_eq!(101, logstore.engine.first_index(namespace_id).unwrap());
- let res = logstore.read(&namespace, 100, None).await.unwrap();
+ let res = logstore.read(&namespace, 100).await.unwrap();
let mut vec = collect_entries(res).await;
vec.sort_by(|a, b| a.entry_id().partial_cmp(&b.entry_id()).unwrap());
assert_eq!(101, vec.first().unwrap().entry_id());
diff --git a/src/log-store/src/test_util/log_store_util.rs b/src/log-store/src/test_util/log_store_util.rs
index 240006595c6..74b63ae080c 100644
--- a/src/log-store/src/test_util/log_store_util.rs
+++ b/src/log-store/src/test_util/log_store_util.rs
@@ -34,16 +34,13 @@ pub async fn create_tmp_local_file_log_store>(path: P) -> RaftEng
/// Create a [KafkaLogStore].
pub async fn create_kafka_log_store(broker_endpoints: Vec) -> KafkaLogStore {
- KafkaLogStore::try_new(
- &DatanodeKafkaConfig {
- connection: KafkaConnectionConfig {
- broker_endpoints,
- ..Default::default()
- },
+ KafkaLogStore::try_new(&DatanodeKafkaConfig {
+ connection: KafkaConnectionConfig {
+ broker_endpoints,
..Default::default()
},
- None,
- )
+ ..Default::default()
+ })
.await
.unwrap()
}
diff --git a/src/metric-engine/src/engine/catchup.rs b/src/metric-engine/src/engine/catchup.rs
index 61e7591893c..357404545da 100644
--- a/src/metric-engine/src/engine/catchup.rs
+++ b/src/metric-engine/src/engine/catchup.rs
@@ -54,7 +54,6 @@ impl MetricEngineInner {
set_writable: req.set_writable,
entry_id: req.metadata_entry_id,
metadata_entry_id: None,
- location_id: req.location_id,
checkpoint: req.checkpoint.map(|c| ReplayCheckpoint {
entry_id: c.metadata_entry_id.unwrap_or_default(),
metadata_entry_id: None,
@@ -67,7 +66,6 @@ impl MetricEngineInner {
set_writable: req.set_writable,
entry_id: req.entry_id,
metadata_entry_id: None,
- location_id: req.location_id,
checkpoint: req.checkpoint.map(|c| ReplayCheckpoint {
entry_id: c.entry_id,
metadata_entry_id: None,
diff --git a/src/mito2/src/engine/basic_test.rs b/src/mito2/src/engine/basic_test.rs
index 35f15f2393c..47b34870ccd 100644
--- a/src/mito2/src/engine/basic_test.rs
+++ b/src/mito2/src/engine/basic_test.rs
@@ -1714,7 +1714,7 @@ async fn check_request_skip_wal_mixed_batch_recovery(
);
// Inspect persisted WAL mutations, not just an encoder or counter.
- let mut reader = wal.wal_entry_reader(®ion.provider, region_id, None);
+ let mut reader = wal.wal_entry_reader(®ion.provider, region_id);
let entries = reader
.read(®ion.provider, 1)
.unwrap()
diff --git a/src/mito2/src/engine/batch_catchup_test.rs b/src/mito2/src/engine/batch_catchup_test.rs
index 5af92280261..06f2a093523 100644
--- a/src/mito2/src/engine/batch_catchup_test.rs
+++ b/src/mito2/src/engine/batch_catchup_test.rs
@@ -184,7 +184,6 @@ async fn test_batch_catchup_with_format(factory: Option, flat_f
set_writable: true,
entry_id: None,
metadata_entry_id: None,
- location_id: None,
checkpoint: None,
},
)
@@ -231,7 +230,6 @@ async fn test_batch_catchup_err_with_format(factory: Option, fl
set_writable: true,
entry_id: None,
metadata_entry_id: None,
- location_id: None,
checkpoint: None,
},
)
diff --git a/src/mito2/src/engine/object_store_wal_test.rs b/src/mito2/src/engine/object_store_wal_test.rs
index d4225b6fcf4..cd27c63b48d 100644
--- a/src/mito2/src/engine/object_store_wal_test.rs
+++ b/src/mito2/src/engine/object_store_wal_test.rs
@@ -147,7 +147,7 @@ async fn count_rows(engine: &MitoEngine, region_id: RegionId) -> usize {
/// Returns the ids of the entries the store still hands out for `region_id`.
async fn stored_entry_ids(log_store: &ObjectStoreLogStore, region_id: RegionId) -> Vec {
log_store
- .read(&object_store_provider(region_id), 0, None)
+ .read(&object_store_provider(region_id), 0)
.await
.unwrap()
.try_collect::>()
diff --git a/src/mito2/src/region/catchup.rs b/src/mito2/src/region/catchup.rs
index 0d8cfa8ed8c..e9f7f7ef43c 100644
--- a/src/mito2/src/region/catchup.rs
+++ b/src/mito2/src/region/catchup.rs
@@ -31,7 +31,6 @@ pub struct RegionCatchupTask
{
replay_checkpoint_entry_id: Option,
expected_last_entry_id: Option,
allow_stale_entries: bool,
- location_id: Option,
wal: Wal,
}
@@ -43,17 +42,10 @@ impl RegionCatchupTask {
replay_checkpoint_entry_id: None,
expected_last_entry_id: None,
allow_stale_entries,
- location_id: None,
wal,
}
}
- /// Sets the location id.
- pub(crate) fn with_location_id(mut self, location_id: Option) -> Self {
- self.location_id = location_id;
- self
- }
-
/// Sets the expected last entry id.
pub(crate) fn with_expected_last_entry_id(
mut self,
@@ -102,10 +94,7 @@ impl RegionCatchupTask {
.entry_receiver
.take()
.map(|r| Box::new(r) as _)
- .unwrap_or_else(|| {
- self.wal
- .wal_entry_reader(&self.region.provider, region_id, self.location_id)
- });
+ .unwrap_or_else(|| self.wal.wal_entry_reader(&self.region.provider, region_id));
let on_region_opened = self.wal.on_region_opened();
let last_entry_id = replay_memtable(
&self.region.provider,
diff --git a/src/mito2/src/region/opener.rs b/src/mito2/src/region/opener.rs
index 7a93799ae5f..37c91946db3 100644
--- a/src/mito2/src/region/opener.rs
+++ b/src/mito2/src/region/opener.rs
@@ -547,7 +547,7 @@ impl RegionOpener {
let wal_entry_reader = self
.wal_entry_reader
.take()
- .unwrap_or_else(|| wal.wal_entry_reader(&provider, region_id, None));
+ .unwrap_or_else(|| wal.wal_entry_reader(&provider, region_id));
let on_region_opened = wal.on_region_opened();
let object_store = get_object_store(®ion_options.storage, &self.object_store_manager)?;
diff --git a/src/mito2/src/wal.rs b/src/mito2/src/wal.rs
index 772aac336cf..76b3e3a2cdf 100644
--- a/src/mito2/src/wal.rs
+++ b/src/mito2/src/wal.rs
@@ -33,7 +33,7 @@ use futures::stream::BoxStream;
use snafu::ResultExt;
use store_api::logstore::entry::Entry;
use store_api::logstore::provider::Provider;
-use store_api::logstore::{AppendBatchResponse, LogStore, WalIndex};
+use store_api::logstore::{AppendBatchResponse, LogStore};
use store_api::storage::RegionId;
use crate::error::{BuildEntrySnafu, DeleteWalSnafu, Result, WaitWalDurableSnafu, WriteWalSnafu};
@@ -163,7 +163,6 @@ impl Wal {
&self,
provider: &Provider,
region_id: RegionId,
- location_id: Option,
) -> Box {
match provider {
Provider::RaftEngine(_) => Box::new(LogStoreEntryReader::new(
@@ -171,15 +170,9 @@ impl Wal {
)),
// Entries of many regions share one namespace, so the reader filters by region.
Provider::Kafka(_) | Provider::ObjectStore(_) => {
- let reader = if let Some(location_id) = location_id {
- LogStoreRawEntryReader::new(self.store.clone())
- .with_wal_index(WalIndex::new(region_id, location_id))
- } else {
- LogStoreRawEntryReader::new(self.store.clone())
- };
-
Box::new(LogStoreEntryReader::new(RegionRawEntryReader::new(
- reader, region_id,
+ LogStoreRawEntryReader::new(self.store.clone()),
+ region_id,
)))
}
Provider::Noop => Box::new(NoopEntryReader),
@@ -194,7 +187,7 @@ impl Wal {
start_id: EntryId,
provider: &'a Provider,
) -> Result> {
- let mut reader = self.wal_entry_reader(provider, region_id, None);
+ let mut reader = self.wal_entry_reader(provider, region_id);
reader.read(provider, start_id)
}
diff --git a/src/mito2/src/wal/raw_entry_reader.rs b/src/mito2/src/wal/raw_entry_reader.rs
index 8a79d6c3b16..0841d21f669 100644
--- a/src/mito2/src/wal/raw_entry_reader.rs
+++ b/src/mito2/src/wal/raw_entry_reader.rs
@@ -18,9 +18,9 @@ use async_stream::try_stream;
use common_error::ext::BoxedError;
use futures::stream::BoxStream;
use snafu::ResultExt;
+use store_api::logstore::LogStore;
use store_api::logstore::entry::Entry;
use store_api::logstore::provider::Provider;
-use store_api::logstore::{LogStore, WalIndex};
use store_api::storage::RegionId;
use tokio_stream::StreamExt;
@@ -38,20 +38,11 @@ pub(crate) trait RawEntryReader: Send + Sync {
/// Implement the [RawEntryReader] for the [LogStore].
pub struct LogStoreRawEntryReader {
store: Arc,
- wal_index: Option,
}
impl LogStoreRawEntryReader {
pub fn new(store: Arc) -> Self {
- Self {
- store,
- wal_index: None,
- }
- }
-
- pub fn with_wal_index(mut self, wal_index: WalIndex) -> Self {
- self.wal_index = Some(wal_index);
- self
+ Self { store }
}
}
@@ -59,10 +50,9 @@ impl RawEntryReader for LogStoreRawEntryReader {
fn read(&self, provider: &Provider, start_id: EntryId) -> Result> {
let store = self.store.clone();
let provider = provider.clone();
- let wal_index = self.wal_index;
let stream = try_stream!({
let mut stream = store
- .read(&provider, start_id, wal_index)
+ .read(&provider, start_id)
.await
.map_err(BoxedError::new)
.with_context(|_| error::ReadWalSnafu {
@@ -129,9 +119,7 @@ mod tests {
use futures::{TryStreamExt, stream};
use store_api::logstore::entry::{Entry, NaiveEntry};
- use store_api::logstore::{
- AppendBatchResponse, EntryId, LogStore, SendableEntryStream, WalIndex,
- };
+ use store_api::logstore::{AppendBatchResponse, EntryId, LogStore, SendableEntryStream};
use store_api::storage::RegionId;
use super::*;
@@ -161,7 +149,6 @@ mod tests {
&self,
_provider: &Provider,
_id: EntryId,
- _index: Option,
) -> Result, Self::Error> {
Ok(Box::pin(stream::iter(vec![Ok(self.entries.clone())])))
}
diff --git a/src/mito2/src/worker/handle_catchup.rs b/src/mito2/src/worker/handle_catchup.rs
index 81838060385..c526aab8c0c 100644
--- a/src/mito2/src/worker/handle_catchup.rs
+++ b/src/mito2/src/worker/handle_catchup.rs
@@ -73,7 +73,6 @@ impl RegionWorkerLoop {
let mut task = RegionCatchupTask::new(region.clone(), wal, allow_stale_entries)
.with_entry_receiver(entry_receiver)
.with_expected_last_entry_id(request.entry_id)
- .with_location_id(request.location_id)
.with_replay_checkpoint_entry_id(request.checkpoint.map(|c| c.entry_id));
match task.run().await {
diff --git a/src/mito2/src/worker/handle_write.rs b/src/mito2/src/worker/handle_write.rs
index 1383a3e9a5d..721235e71b5 100644
--- a/src/mito2/src/worker/handle_write.rs
+++ b/src/mito2/src/worker/handle_write.rs
@@ -758,7 +758,7 @@ mod tests {
};
use store_api::logstore::entry::{Entry, NaiveEntry};
use store_api::logstore::provider::Provider;
- use store_api::logstore::{AppendBatchResponse, EntryId, SendableEntryStream, WalIndex};
+ use store_api::logstore::{AppendBatchResponse, EntryId, SendableEntryStream};
use store_api::region_request::AffectedRows;
use tokio::sync::oneshot;
@@ -828,7 +828,6 @@ mod tests {
&self,
_provider: &Provider,
_entry_id: EntryId,
- _index: Option,
) -> LogStoreResult> {
Ok(Box::pin(stream::empty()))
}
diff --git a/src/store-api/src/logstore.rs b/src/store-api/src/logstore.rs
index 6084f207641..dd71142def5 100644
--- a/src/store-api/src/logstore.rs
+++ b/src/store-api/src/logstore.rs
@@ -30,22 +30,6 @@ pub use crate::logstore::entry::Id as EntryId;
use crate::logstore::provider::Provider;
use crate::storage::RegionId;
-// The information used to locate WAL index for the specified region.
-#[derive(Debug, Clone, Copy)]
-pub struct WalIndex {
- pub region_id: RegionId,
- pub location_id: u64,
-}
-
-impl WalIndex {
- pub fn new(region_id: RegionId, location_id: u64) -> Self {
- Self {
- region_id,
- location_id,
- }
- }
-}
-
/// `LogStore` serves as a Write-Ahead-Log for storage engine.
#[async_trait::async_trait]
pub trait LogStore: Send + Sync + 'static + std::fmt::Debug {
@@ -64,7 +48,6 @@ pub trait LogStore: Send + Sync + 'static + std::fmt::Debug {
&self,
provider: &Provider,
id: EntryId,
- index: Option,
) -> Result, Self::Error>;
/// Creates a new `Namespace` from the given ref.
diff --git a/src/store-api/src/logstore/provider.rs b/src/store-api/src/logstore/provider.rs
index 4fe7e4dfe7b..4af4caaaa60 100644
--- a/src/store-api/src/logstore/provider.rs
+++ b/src/store-api/src/logstore/provider.rs
@@ -187,7 +187,7 @@ mod tests {
use super::*;
use crate::logstore::entry::Entry;
- use crate::logstore::{AppendBatchResponse, EntryId, SendableEntryStream, WalIndex};
+ use crate::logstore::{AppendBatchResponse, EntryId, SendableEntryStream};
/// A log store whose latest entry id is fixed.
#[derive(Debug)]
@@ -212,7 +212,6 @@ mod tests {
&self,
_provider: &Provider,
_id: EntryId,
- _index: Option,
) -> Result, Self::Error> {
unreachable!()
}
diff --git a/src/store-api/src/region_request.rs b/src/store-api/src/region_request.rs
index e1f0eb22497..ebb2f7c3684 100644
--- a/src/store-api/src/region_request.rs
+++ b/src/store-api/src/region_request.rs
@@ -1929,8 +1929,6 @@ pub struct RegionCatchupRequest {
/// The `entry_id` that was expected to reply to.
/// `None` stands replaying to latest.
pub metadata_entry_id: Option,
- /// The hint for replaying memtable.
- pub location_id: Option,
/// Replay checkpoint.
pub checkpoint: Option,
}