rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-14 07:52:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor!: remove datanode remote WAL index (#9563)

Browse Source 
* refactor: remove datanode remote WAL index

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(log-store): update new WAL replay tests after index removal

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
Weny Xu
2026-10-13 15:31:04 +00:00
committed by GitHub
parent 863eebdecb
commit f690bbb204
47 changed files with 305 additions and 1805 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-10
View File
@@ -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",
config/config.md
-2
View File
@@ -604,8 +604,6 @@
| `wal.timeout` | String | `5s` | The total request timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
| `wal.max_batch_bytes` | String | `1MB` | The max size of a single producer batch.<br/>Warning: Kafka has a default limit of 1MB per message in a topic.<br/>Defaults to `8MB` when the provider is `experimental_object_store`.<br/>**It's only used when the provider is `kafka` or `experimental_object_store`**. |
| `wal.consumer_wait_timeout` | String | `100ms` | The consumer wait timeout.<br/>**It's only used when the provider is `kafka`**. |
| `wal.create_index` | Bool | `false` | Whether to enable WAL index creation.<br/>**It's only used when the provider is `kafka`**. |
| `wal.dump_index_interval` | String | `60s` | The interval for dumping WAL indexes.<br/>**It's only used when the provider is `kafka`**. |
| `wal.overwrite_entry_start_id` | Bool | `false` | Ignore missing entries during read WAL.<br/>**It's only used when the provider is `kafka`**.<br/><br/>This option ensures that when Kafka messages are deleted, the system<br/>can still successfully replay memtable data without throwing an<br/>out-of-range error.<br/>However, enabling this option might lead to unexpected data loss,<br/>as the system will skip over missing entries instead of treating<br/>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.<br/>**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.<br/>The objects are written under `<prefix>/datanodes/<node_id>/epochs/<generation>`, which is derived from this prefix.<br/>**It's only used when the provider is `experimental_object_store`**. |
config/datanode.example.toml
-8
View File
@@ -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`**.
##
src/cmd/src/datanode/scanbench.rs
+1 -1
View File
@@ -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)?,
src/common/meta/src/instruction.rs
+1 -1
View File
@@ -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<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
src/common/wal/src/config.rs
+24
View File
@@ -346,6 +346,30 @@ mod tests {
assert!(toml::from_str::<DatanodeWalConfig>(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::<DatanodeWalConfig>(&serialized).unwrap(),
expected
);
}
}
#[test]
fn test_kafka_wal_config_debug_redacts_password() {
let config = MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
src/common/wal/src/config/kafka/datanode.rs
-6
View File
@@ -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,
}
src/datanode/src/datanode.rs
+4 -39
View File
@@ -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<GlobalIndexCollector>,
) -> Result<Arc<KafkaLogStore>> {
KafkaLogStore::try_new(config, global_index_collector)
async fn build_kafka_log_store(config: &DatanodeKafkaConfig) -> Result<Arc<KafkaLogStore>> {
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.
src/datanode/src/heartbeat/handler/upgrade_region.rs
+43 -2
View File
@@ -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();
src/datanode/src/store.rs
-19
View File
@@ -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<ObjectStore> {
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() {
src/log-store/Cargo.toml
-1
View File
@@ -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
src/log-store/src/error.rs
+2 -42
View File
@@ -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 { .. }
src/log-store/src/kafka.rs
-2
View File
@@ -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;
src/log-store/src/kafka/client_manager.rs
-16
View File
@@ -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<HashMap<Arc<KafkaProvider>, Client>>,
global_index_collector: Option<GlobalIndexCollector>,
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<GlobalIndexCollector>,
topic_stats: Arc<DashMap<Arc<KafkaProvider>, TopicStat>>,
) -> Result<Self> {
// 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<Arc<KafkaProvider>> {
self.instances.read().await.keys().cloned().collect()
src/log-store/src/kafka/consumer.rs
+37 -78
View File
@@ -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<RecordAndOffset>,
index: Box<dyn RegionWalIndexIterator>,
index: RegionWalRange,
}
impl RecordsBuffer {
/// Creates an empty [`RecordsBuffer`]
pub fn new(index: Box<dyn RegionWalIndexIterator>) -> 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::<Vec<_>>().await.unwrap();
assert_eq!(
records
.into_iter()
.map(|(x, _)| x.offset)
.collect::<Vec<_>>(),
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::<Vec<_>>().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::<Vec<_>>()
.await
.unwrap()
.into_iter()
.map(|(x, _)| x.offset)
.collect::<Vec<_>>(),
[0, 1, 2, 7, 8, 11, 1024, 1025, 1026, 1027]
)
.map(|(record, _)| record.offset as u64)
.collect::<Vec<_>>();
assert_eq!(offsets, (start..end).collect::<Vec<_>>());
}
}
}
src/log-store/src/kafka/consumer/range.rs
+118
View File
@@ -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<EntryId>, 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<u64> {
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<NextBatchHint> {
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<EntryId> {
if self.current_entry_id < self.end_entry_id {
Some(self.current_entry_id)
} else {
None
}
}
pub(crate) fn next(&mut self) -> Option<EntryId> {
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);
}
}
src/log-store/src/kafka/index.rs
-30
View File
@@ -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")
}
src/log-store/src/kafka/index/collector.rs
-414
View File
@@ -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<TokioMutex<HashMap<Arc<KafkaProvider>, Sender<WorkerRequest>>>>,
operator: object_store::ObjectStore,
_handle: CollectionTaskHandle,
}
#[derive(Debug, Clone)]
pub struct CollectionTask {
providers: Arc<TokioMutex<HashMap<Arc<KafkaProvider>, Sender<WorkerRequest>>>>,
dump_index_interval: Duration,
operator: object_store::ObjectStore,
path: String,
running: Arc<AtomicBool>,
}
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<AtomicBool>,
}
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<TokioMutex<HashMap<Arc<KafkaProvider>, Sender<WorkerRequest>>>> =
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<Option<(BTreeSet<EntryId>, 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<KafkaProvider>,
sender: Sender<WorkerRequest>,
) -> Box<dyn IndexCollector> {
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<KafkaProvider>,
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<RegionId, BTreeSet<EntryId>>,
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(&region_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<KafkaProvider>,
}
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);
}
}
src/log-store/src/kafka/index/encoder.rs
-181
View File
@@ -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<RegionId, Vec<u64>>,
last_index: u64,
}
impl DeltaEncodedRegionIndexes {
/// Retrieves the original (decoded) index values for a given region.
pub(crate) fn region(&self, region_id: RegionId) -> Option<BTreeSet<u64>> {
self.regions
.get(&region_id)
.map(|delta| delta.iter().copied().original().collect::<BTreeSet<_>>())
}
/// 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<Vec<u8>>;
}
/// [`DatanodeWalIndexes`] structure holds the WAL indexes for a datanode.
#[derive(Debug, Default, Serialize, Deserialize)]
pub(crate) struct DatanodeWalIndexes(HashMap<String, DeltaEncodedRegionIndexes>);
impl DatanodeWalIndexes {
fn insert(&mut self, topic: String, region_index: &RegionIndexes) {
self.0.insert(topic, region_index.into());
}
fn encode(&mut self) -> Result<Vec<u8>> {
let value = serde_json::to_vec(&self.0).context(error::EncodeJsonSnafu);
self.0.clear();
value
}
pub(crate) fn decode(byte: &[u8]) -> Result<Self> {
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<DatanodeWalIndexes>,
}
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<Vec<u8>> {
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()
);
}
}
src/log-store/src/kafka/index/iterator.rs
-483
View File
@@ -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<NextBatchHint>;
// Peeks the next EntryId without advancing the iterator.
fn peek(&self) -> Option<EntryId>;
// Advances the iterator and returns the next EntryId.
fn next(&mut self) -> Option<EntryId>;
#[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<EntryId>, 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<u64> {
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<NextBatchHint> {
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<EntryId> {
if self.current_entry_id < self.end_entry_id {
Some(self.current_entry_id)
} else {
None
}
}
fn next(&mut self) -> Option<EntryId> {
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<EntryId>,
min_batch_window_size: usize,
}
impl RegionWalVecIndex {
pub fn new<I: IntoIterator<Item = EntryId>>(index: I, min_batch_window_size: usize) -> Self {
Self {
index: index.into_iter().collect::<VecDeque<_>>(),
min_batch_window_size,
}
}
}
impl RegionWalIndexIterator for RegionWalVecIndex {
fn next_batch_hint(&self, avg_size: usize) -> Option<NextBatchHint> {
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<EntryId> {
self.index.front().cloned()
}
fn next(&mut self) -> Option<EntryId> {
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<Box<dyn RegionWalIndexIterator>>,
}
impl MultipleRegionWalIndexIterator {
pub fn new<I: IntoIterator<Item = Box<dyn RegionWalIndexIterator>>>(iterator: I) -> Self {
Self {
iterator: iterator.into_iter().collect::<VecDeque<_>>(),
}
}
}
impl RegionWalIndexIterator for MultipleRegionWalIndexIterator {
fn next_batch_hint(&self, avg_size: usize) -> Option<NextBatchHint> {
for iter in &self.iterator {
if let Some(batch) = iter.next_batch_hint(avg_size) {
return Some(batch);
}
}
None
}
fn peek(&self) -> Option<EntryId> {
for iter in &self.iterator {
let peek = iter.peek();
if peek.is_some() {
return peek;
}
}
None
}
fn next(&mut self) -> Option<EntryId> {
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>, EntryId)>,
max_batch_bytes: usize,
min_window_size: usize,
) -> Option<Box<dyn RegionWalIndexIterator>> {
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<Box<dyn RegionWalIndexIterator>> = 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::<MultipleRegionWalIndexIterator>()
.unwrap();
assert_eq!(iter.iterator.len(), 1);
let vec_index = iter.iterator[0]
.as_any()
.downcast_ref::<RegionWalVecIndex>()
.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::<MultipleRegionWalIndexIterator>()
.unwrap();
assert_eq!(iter.iterator.len(), 2);
let vec_index = iter.iterator[0]
.as_any()
.downcast_ref::<RegionWalVecIndex>()
.unwrap();
assert_eq!(vec_index.index, VecDeque::from([512, 756]));
let wal_range = iter.iterator[1]
.as_any()
.downcast_ref::<RegionWalRange>()
.unwrap();
assert_eq!(wal_range.current_entry_id, 1023);
assert_eq!(wal_range.end_entry_id, 1024);
}
}
src/log-store/src/kafka/log_store.rs
+17 -50
View File
@@ -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<GlobalIndexCollector>,
) -> Result<Self> {
pub async fn try_new(config: &DatanodeKafkaConfig) -> Result<Self> {
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<WalIndex>,
) -> Result<SendableEntryStream<'static, Entry, Self::Error>> {
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::<Vec<_>>()
@@ -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::<Vec<_>>()
src/log-store/src/kafka/periodic_offset_fetcher.rs
+1 -3
View File
@@ -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);
src/log-store/src/kafka/producer.rs
+10 -49
View File
@@ -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<dyn ProducerClient>,
compression: Compression,
max_batch_bytes: usize,
index_collector: Box<dyn IndexCollector>,
topic_stats: Arc<DashMap<Arc<KafkaProvider>, 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<Record>,
) -> Result<ProduceResultHandle> {
let (req, handle) = WorkerRequest::new_produce_request(region_id, batch);
pub(crate) async fn produce(&self, batch: Vec<Record>) -> Result<ProduceResultHandle> {
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);
src/log-store/src/kafka/test_util.rs
+1 -1
View File
@@ -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();
src/log-store/src/kafka/util.rs
-1
View File
@@ -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;
src/log-store/src/kafka/util/range.rs
-146
View File
@@ -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<Item = &'a EntryId>> {
base: Option<EntryId>,
iter: Peekable<I>,
avg_size: usize,
}
impl<'a, I: Iterator<Item = &'a EntryId>> ConvertIndexToRange<'a, I> {
pub fn new(mut iter: Peekable<I>, avg_size: usize) -> Self {
let base = iter.peek().cloned().cloned();
Self {
base,
iter,
avg_size,
}
}
}
impl<'a, I: Iterator<Item = &'a EntryId>> Iterator for ConvertIndexToRange<'a, I> {
type Item = Range<usize>;
fn next(&mut self) -> Option<Self::Item> {
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<I: Iterator<Item = Range<usize>>> {
iter: I,
window_size: usize,
}
impl<I: Iterator<Item = Range<usize>>> MergeRange<I> {
pub fn new(iter: I, window_size: usize) -> Self {
Self { iter, window_size }
}
}
/// Merges ranges.
fn merge(this: &mut Range<usize>, other: &Range<usize>) {
this.start = min(this.start, other.start);
this.end = max(this.end, other.end);
}
impl<I: Iterator<Item = Range<usize>>> MergeRange<I> {
/// Calculates the size of the next merged range.
pub(crate) fn merge(mut self) -> Option<(Range<usize>, 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);
}
}
src/log-store/src/kafka/worker.rs
+2 -56
View File
@@ -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<Record>,
) -> (WorkerRequest, ProduceResultHandle) {
pub(crate) fn new_produce_request(batch: Vec<Record>) -> (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<dyn IndexEncoder>,
sender: oneshot::Sender<()>,
}
impl DumpIndexRequest {
pub fn new(encoder: Arc<dyn IndexEncoder>) -> (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<Record>,
sender: oneshot::Sender<ProduceResultReceiver>,
}
@@ -142,7 +96,6 @@ impl ProduceResultReceiver {
pub(crate) struct PendingRequest {
batch: Vec<Record>,
region_ids: Vec<RegionId>,
size: usize,
sender: oneshot::Sender<Result<Vec<i64>>>,
}
@@ -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<dyn IndexCollector>,
/// The stats of each topic.
pub(crate) topic_stats: Arc<DashMap<Arc<KafkaProvider>, 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;
}
src/log-store/src/kafka/worker/dump_index.rs
-39
View File
@@ -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"),
}
}
}
src/log-store/src/kafka/worker/flush.rs
-5
View File
@@ -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())
src/log-store/src/kafka/worker/produce.rs
+1 -10
View File
@@ -30,16 +30,10 @@ impl BackgroundProducerWorker {
max_batch_bytes: usize,
) -> Vec<PendingRequest> {
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,
});
src/log-store/src/noop/log_store.rs
+1 -2
View File
@@ -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<WalIndex>,
) -> Result<SendableEntryStream<'static, Entry, Self::Error>> {
Ok(Box::pin(stream::empty()))
}
src/log-store/src/object_store_wal/store.rs
+18 -23
View File
@@ -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<WalIndex>,
) -> Result<SendableEntryStream<'static, Entry, Error>> {
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<Entry> {
store
.read(&provider(region_id), start, None)
.read(&provider(region_id), start)
.await
.unwrap()
.try_collect::<Vec<_>>()
@@ -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::<Vec<_>>()
@@ -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::<Vec<_>>()
@@ -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::<Vec<_>>()
src/log-store/src/raft_engine/log_store.rs
+5 -6
View File
@@ -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<WalIndex>,
) -> Result<SendableEntryStream<'static, Entry, Self::Error>> {
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::<Vec<_>>()
@@ -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());
src/log-store/src/test_util/log_store_util.rs
+5 -8
View File
@@ -34,16 +34,13 @@ pub async fn create_tmp_local_file_log_store<P: AsRef<Path>>(path: P) -> RaftEng
/// Create a [KafkaLogStore].
pub async fn create_kafka_log_store(broker_endpoints: Vec<String>) -> 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()
}
src/metric-engine/src/engine/catchup.rs
-2
View File
@@ -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,
src/mito2/src/engine/basic_test.rs
+1 -1
View File
@@ -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(&region.provider, region_id, None);
let mut reader = wal.wal_entry_reader(&region.provider, region_id);
let entries = reader
.read(&region.provider, 1)
.unwrap()
src/mito2/src/engine/batch_catchup_test.rs
-2
View File
@@ -184,7 +184,6 @@ async fn test_batch_catchup_with_format(factory: Option<LogStoreFactory>, 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<LogStoreFactory>, fl
set_writable: true,
entry_id: None,
metadata_entry_id: None,
location_id: None,
checkpoint: None,
},
)
src/mito2/src/engine/object_store_wal_test.rs
+1 -1
View File
@@ -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<EntryId> {
log_store
.read(&object_store_provider(region_id), 0, None)
.read(&object_store_provider(region_id), 0)
.await
.unwrap()
.try_collect::<Vec<_>>()
src/mito2/src/region/catchup.rs
+1 -12
View File
@@ -31,7 +31,6 @@ pub struct RegionCatchupTask<S> {
replay_checkpoint_entry_id: Option<u64>,
expected_last_entry_id: Option<u64>,
allow_stale_entries: bool,
location_id: Option<u64>,
wal: Wal<S>,
}
@@ -43,17 +42,10 @@ impl<S: LogStore> RegionCatchupTask<S> {
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<u64>) -> 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<S: LogStore> RegionCatchupTask<S> {
.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,
src/mito2/src/region/opener.rs
+1 -1
View File
@@ -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(&region_options.storage, &self.object_store_manager)?;
src/mito2/src/wal.rs
+4 -11
View File
@@ -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<S: LogStore> Wal<S> {
&self,
provider: &Provider,
region_id: RegionId,
location_id: Option<u64>,
) -> Box<dyn WalEntryReader> {
match provider {
Provider::RaftEngine(_) => Box::new(LogStoreEntryReader::new(
@@ -171,15 +170,9 @@ impl<S: LogStore> Wal<S> {
)),
// 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<S: LogStore> Wal<S> {
start_id: EntryId,
provider: &'a Provider,
) -> Result<WalEntryStream<'a>> {
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)
}
src/mito2/src/wal/raw_entry_reader.rs
+4 -17
View File
@@ -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<S> {
store: Arc<S>,
wal_index: Option<WalIndex>,
}
impl<S> LogStoreRawEntryReader<S> {
pub fn new(store: Arc<S>) -> 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<S: LogStore> RawEntryReader for LogStoreRawEntryReader<S> {
fn read(&self, provider: &Provider, start_id: EntryId) -> Result<EntryStream<'static>> {
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<WalIndex>,
) -> Result<SendableEntryStream<'static, Entry, Self::Error>, Self::Error> {
Ok(Box::pin(stream::iter(vec![Ok(self.entries.clone())])))
}
src/mito2/src/worker/handle_catchup.rs
-1
View File
@@ -73,7 +73,6 @@ impl<S: LogStore> RegionWorkerLoop<S> {
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 {
src/mito2/src/worker/handle_write.rs
+1 -2
View File
@@ -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<WalIndex>,
) -> LogStoreResult<SendableEntryStream<'static, Entry, Self::Error>> {
Ok(Box::pin(stream::empty()))
}
src/store-api/src/logstore.rs
-17
View File
@@ -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<WalIndex>,
) -> Result<SendableEntryStream<'static, Entry, Self::Error>, Self::Error>;
/// Creates a new `Namespace` from the given ref.
src/store-api/src/logstore/provider.rs
+1 -2
View File
@@ -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<WalIndex>,
) -> Result<SendableEntryStream<'static, Entry, Self::Error>, Self::Error> {
unreachable!()
}
src/store-api/src/region_request.rs
-2
View File
@@ -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<entry::Id>,
/// The hint for replaying memtable.
pub location_id: Option<u64>,
/// Replay checkpoint.
pub checkpoint: Option<ReplayCheckpoint>,
}