feat(stream): add synced logstore #20204

kwannoel · 2025-01-17T10:13:51Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Add synced logstore for unaligned join and decoupled sink into table.

We have 2 futures we repeatedly construct and poll:

One to read from upstream.
Another to read from logstore (state_store / buffer).

May look at #20204 (comment) for the implementation design.

We can't use the desynced (existing) logstore because:

Truncation interface and barrier commit are decoupled. It's designed in such a way because sinks only require at-least-once semantics. So it is fine to read duplicated data after rewind happens. This doesn't apply to join, because downstream will need to ensure it is buffered.
The desynced logstore does not yield barrier downstream directly from the writer side. This is not a requirement for external sinks, because they don't need to receive barrier. However for sink into table or join downstream, we need to propagate barrier typically.

TODOs:

Handle update vnode bitmap
Handle watermark

Checklist

I have written necessary rustdoc comments.
I have added necessary unit tests and integration tests.
I have added test labels as necessary.
I have added fuzzing tests or opened an issue to track them.
My PR contains breaking changes.
My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
My PR contains critical fixes that are necessary to be merged into the latest release.

Documentation

My PR needs documentation updates.

Release note

kwannoel · 2025-01-17T10:14:11Z

refactor #20262
feat(stream): add synced logstore #20204 👈 (View in Graphite)
main

This stack of pull requests is managed by Graphite. Learn more about stacking.

kwannoel · 2025-01-22T10:00:28Z

src/stream/src/executor/sync_kv_log_store.rs

+//!      This is done by constructing a `flushed_chunk_future` which will read the log store
+//!      using the `seq_id`.
+//!   - Barrier,
+//!     because they are directly propagated from the upstream when polling it.


See this docstring for the design.

So what does "synced" mean here? (and what is unsynced logstore?)

unsynced logstore is our current design for the sink logstore, where we have 2 separate streams for reading from and to the logstore.

Synced means that we have a single stream instead, rather than 2 concurrently running ones.

I'm not very familiar with our current design for the sink logstore. So why can't we use the "unsynced" version here? What problems do the "synced" version want to solve?

btw, why sink into table needs this new logstore, instead of the old one

I'm not very familiar with our current design for the sink logstore. So why can't we use the "unsynced" version here? What problems do the "synced" version want to solve?

Updated in PR description. Briefly:

Truncate the logstore before barrier gets committed, rather than asynchronously. This ensures exactly-once semantics.

Yield barriers downstream from the writer side. External sink does not need barriers so the desynced logstore does not have barriers yielded downstream.

src/stream/src/executor/sync_kv_log_store.rs

gru-agent · 2025-01-22T15:11:23Z

This pull request has been modified. If you want me to regenerate unit test for any of the files related, please find the file in "Files Changed" tab and add a comment @gru-agent. (The github "Comment on this file" feature is in the upper right corner of each file in "Files Changed" tab.)

chenzl25

Generally LGTM. The implementation is similar to the sink logstore, but we need to take care of the watermark truncation which is related to compaction.

chenzl25 · 2025-01-24T09:56:28Z

src/stream/src/executor/sync_kv_log_store.rs

+            table_id,
+            metrics,
+            state_store,
+            barrier.epoch.curr, // FIXME(kwannoel): Should this be curr or prev?


It seems to be the prev barrier. It works like backill snapshot read.

wenym1

Leave some early comments. Will continue review the executor implementation. At first glance, the implementation seems to be too complicated. Besides, it seems that update vnode bitmap is not handled in this PR.

wenym1 · 2025-01-26T05:57:29Z

src/stream/Cargo.toml

@@ -72,6 +72,7 @@ tokio-retry = "0.3"
 tokio-stream = { workspace = true }
 tonic = { workspace = true }
 tracing = "0.1"
+tracing-subscriber = "0.3.17"


Seems to be used only in test. May add to dev-dependencies.

wenym1 · 2025-01-26T06:00:22Z

src/stream/src/executor/sync_kv_log_store.rs

+type StateStoreStream<S> = Pin<Box<LogStoreItemMergeStream<TimeoutAutoRebuildIter<S>>>>;
+type ReadFlushedChunkFuture = BoxFuture<'static, LogStoreResult<(ChunkId, StreamChunk, u64)>>;
+
+struct SyncedKvLogStoreExecutor<S: StateStore, LS: LocalStateStore> {


Only need a single generic type S: StateStore. LS is S::Local.

wenym1 · 2025-01-26T09:06:33Z

src/stream/src/common/log_store_impl/kv_log_store/buffer.rs

@@ -28,7 +28,7 @@ use crate::common::log_store_impl::kv_log_store::{
 };

 #[derive(Clone)]
-pub(crate) enum LogStoreBufferItem {
+pub enum LogStoreBufferItem {


The synced_kv_log_store is also in the same crate, so not need to extend the visibility.

wenym1 · 2025-01-26T09:06:51Z

src/stream/src/common/log_store_impl/kv_log_store/mod.rs

@@ -32,11 +32,11 @@ use crate::common::log_store_impl::kv_log_store::serde::LogStoreRowSerde;
 use crate::common::log_store_impl::kv_log_store::writer::KvLogStoreWriter;
 use crate::executor::monitor::StreamingMetrics;

-mod buffer;
-mod reader;
+pub mod buffer;


pub(crate) should be enough for the changes in this file.

wenym1 · 2025-01-26T09:09:33Z

src/stream/src/common/log_store_impl/kv_log_store/reader.rs

@@ -672,6 +598,107 @@ impl<S: StateStoreRead + Clone> LogReader for KvLogStoreReader<S> {
    }
 }

+#[allow(clippy::too_many_arguments)]


Are the read_flushed_chunk and read_persisted_log_store just a simple copy-paste from the original code without any change in the logic?

Yes. These are refactored and reused in the synced_log_store executor.

wenym1 · 2025-01-26T09:11:10Z

src/stream/src/executor/sync_kv_log_store.rs

+        let Some(msg) = input.next().await else {
+            bail!("Expected a barrier message, got end of stream")
+        };
+        let barrier = match msg? {


Use expect_first_barrier.

wenym1 · 2025-01-26T09:12:02Z

src/stream/src/executor/sync_kv_log_store.rs

+            Message::Barrier(barrier) => barrier,
+            other => bail!("Expected a barrier message, got {:?}", other),
+        };
+        let mut state_store_stream = Some(


We should yield the barrier first and then init the local state store. Otherwise, we will get stuck at LocalStateStore::init.

wenym1 · 2025-01-26T10:22:25Z

src/stream/src/executor/sync_kv_log_store.rs

+        metrics: &mut KvLogStoreMetrics,
+        truncation_offset: Option<ReaderTruncationOffsetType>,
+        seq_id: &mut SeqIdType,
+        buffer: &mut Mutex<SyncedLogStoreBuffer>,


The current code, the Mutex<SyncedLogStoreBuffer> always has the mutable reference. Therefore, it's likely that we don't have to wrap the buffer with the Mutex.

github-actions bot added the Invalid PR Title label Jan 17, 2025

kwannoel changed the title ~~add interfaces~~ feat(stream): add synced logstore Jan 17, 2025

github-actions bot added type/feature and removed Invalid PR Title labels Jan 17, 2025

kwannoel force-pushed the 01-06-feat_stream_add_synced_logstores branch from e7c0130 to 31d6e5d Compare January 21, 2025 08:27

kwannoel requested review from chenzl25 and wenym1 January 22, 2025 09:58

kwannoel commented Jan 22, 2025

View reviewed changes

src/stream/src/executor/sync_kv_log_store.rs Show resolved Hide resolved

kwannoel marked this pull request as ready for review January 22, 2025 12:47

kwannoel requested a review from a team as a code owner January 22, 2025 12:47

kwannoel requested a review from MrCroxx January 22, 2025 12:47

kwannoel mentioned this pull request Jan 22, 2025

refactor #20262

Draft

8 tasks

kwannoel force-pushed the 01-06-feat_stream_add_synced_logstores branch from ea8d543 to 48bea0d Compare January 23, 2025 06:11

chenzl25 reviewed Jan 24, 2025

View reviewed changes

chenzl25 requested a review from Li0k January 24, 2025 10:00

wenym1 reviewed Jan 26, 2025

View reviewed changes

kwannoel added 10 commits January 27, 2025 17:18

add interfaces

326641a

interim

7c64aea

separate select

0256115

poll writer and reader

e1d8e76

directly read from upstream

4dbfd23

match different upstream items

a3accdc

docs

28d90f8

add skeleton for read path

6b8b872

refactor common fn

ec68692

handle full read path

229f7b5

kwannoel added 28 commits January 27, 2025 17:18

impl execute

f620e9d

fix warn + docs

a6cd2e3

refactor persisted logstore

9a7779b

init state store iter

5468b2e

fix warn

458f02e

pass executor rather than msg stream

c8f2ea8

defer state_store clone

c269c1f

add vnodes

31b535f

test outline pt 1

c26800d

add more test lines

c7b6b94

check test results

9e7b62b

fix calls

93b10a7

make test compile

ca0c40e

yield first barrier

c0e006b

logging

a82fcf5

bias to upstream side

e90e4d2

fmt

e28e3fe

allow unused

123046c

fix test lints

6bca386

fix

e6ded4e

use prev

5a8b58a

fix deps

f0fe699

remove LS

2b3dfd3

use pub(crate) instead of pub

1e4ab0f

use expect_first_barrier + yield before init state store

c0e70d0

fix warn

e0360a4

apply vnode update to write path

8cc063a

update read path

5cccf42

kwannoel force-pushed the 01-06-feat_stream_add_synced_logstores branch from d3629fd to 5cccf42 Compare January 27, 2025 09:18

no need lock

484a5f8

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(stream): add synced logstore #20204

feat(stream): add synced logstore #20204

kwannoel commented Jan 17, 2025 •

edited

Loading

kwannoel commented Jan 17, 2025 •

edited

Loading

kwannoel Jan 22, 2025

xxchan Jan 23, 2025

kwannoel Jan 23, 2025

xxchan Jan 23, 2025

xxchan Jan 23, 2025

kwannoel Jan 23, 2025

gru-agent bot commented Jan 22, 2025

chenzl25 left a comment

chenzl25 Jan 24, 2025

wenym1 left a comment

wenym1 Jan 26, 2025

wenym1 Jan 26, 2025

wenym1 Jan 26, 2025

wenym1 Jan 26, 2025

wenym1 Jan 26, 2025

kwannoel Jan 26, 2025

wenym1 Jan 26, 2025

wenym1 Jan 26, 2025

wenym1 Jan 26, 2025

feat(stream): add synced logstore #20204

Are you sure you want to change the base?

feat(stream): add synced logstore #20204

Conversation

kwannoel commented Jan 17, 2025 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

kwannoel commented Jan 17, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gru-agent bot commented Jan 22, 2025

chenzl25 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wenym1 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kwannoel commented Jan 17, 2025 •

edited

Loading

kwannoel commented Jan 17, 2025 •

edited

Loading