You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "luoyuxia (via GitHub)" <gi...@apache.org> on 2025/12/20 13:14:37 UTC
[PR] Improve log poll [fluss-rust]
luoyuxia opened a new pull request, #103:
URL: https://github.com/apache/fluss-rust/pull/103
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss-rust/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #94
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637144060
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -211,125 +260,292 @@ impl LogFetcher {
projected_fields: Option<Vec<usize>>,
) -> Result<Self> {
let full_arrow_schema = to_arrow_schema(table_info.get_row_type());
- let read_context = Self::create_read_context(full_arrow_schema, projected_fields.clone());
+ let read_context =
+ Self::create_read_context(full_arrow_schema.clone(), projected_fields.clone(), false);
+ let remote_read_context =
+ Self::create_read_context(full_arrow_schema, projected_fields.clone(), true);
let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
Ok(LogFetcher {
- table_path: table_info.table_path.clone(),
- conns,
- table_info,
- metadata,
+ conns: conns.clone(),
+ metadata: metadata.clone(),
log_scanner_status,
read_context,
- remote_log_downloader: RemoteLogDownloader::new(tmp_dir)?,
- credentials_cache: CredentialsCache::new(),
+ remote_read_context,
+ remote_log_downloader: Arc::new(RemoteLogDownloader::new(tmp_dir)?),
+ credentials_cache: Arc::new(CredentialsCache::new(conns.clone(), metadata.clone())),
+ log_fetch_buffer: Arc::new(LogFetchBuffer::new()),
+ nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())),
})
}
fn create_read_context(
full_arrow_schema: SchemaRef,
projected_fields: Option<Vec<usize>>,
+ is_from_remote: bool,
) -> ReadContext {
match projected_fields {
- None => ReadContext::new(full_arrow_schema),
- Some(fields) => ReadContext::with_projection_pushdown(full_arrow_schema, fields),
+ None => ReadContext::new(full_arrow_schema, is_from_remote),
+ Some(fields) => {
+ ReadContext::with_projection_pushdown(full_arrow_schema, fields, is_from_remote)
+ }
}
}
- async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ /// Send fetch requests asynchronously without waiting for responses
+ async fn send_fetches(&self) -> Result<()> {
+ // todo: check update metadata like fluss-java in case leader changes
let fetch_request = self.prepare_fetch_log_requests().await;
- let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+
for (leader, fetch_request) in fetch_request {
- let cluster = self.metadata.get_cluster();
- let server_node = cluster
- .get_tablet_server(leader)
- .expect("todo: handle leader not exist.");
- let con = self.conns.get_connection(server_node).await?;
-
- let fetch_response = con
- .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
- .await?;
-
- for pb_fetch_log_resp in fetch_response.tables_resp {
- let table_id = pb_fetch_log_resp.table_id;
- let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
-
- for fetch_log_for_bucket in fetch_log_for_buckets {
- let bucket: i32 = fetch_log_for_bucket.bucket_id;
- let table_bucket = TableBucket::new(table_id, bucket);
-
- // Check if this is a remote log fetch
- if let Some(ref remote_log_fetch_info) =
- fetch_log_for_bucket.remote_log_fetch_info
- {
- let remote_fs_props = self
- .credentials_cache
- .get_or_refresh(&self.conns, &self.metadata)
- .await?;
- self.remote_log_downloader
- .set_remote_fs_props(remote_fs_props);
- let remote_fetch_info = RemoteLogFetchInfo::from_proto(
- remote_log_fetch_info,
- table_bucket.clone(),
- )?;
-
- if let Some(fetch_offset) =
- self.log_scanner_status.get_bucket_offset(&table_bucket)
- {
- let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
- // Download and process remote log segments
- let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
- let mut current_fetch_offset = fetch_offset;
- // todo: make segment download in parallel
- for (i, segment) in
- remote_fetch_info.remote_log_segments.iter().enumerate()
- {
- if i > 0 {
- pos_in_log_segment = 0;
- current_fetch_offset = segment.start_offset;
- }
+ debug!("Adding pending request for node id {leader}");
+ // Check if we already have a pending request for this node
+ {
+ self.nodes_with_pending_fetch_requests.lock().insert(leader);
+ }
+
+ let cluster = self.metadata.get_cluster().clone();
+
+ let conns = Arc::clone(&self.conns);
+ let log_fetch_buffer = self.log_fetch_buffer.clone();
+ let log_scanner_status = self.log_scanner_status.clone();
+ let read_context = self.read_context.clone();
+ let remote_read_context = self.remote_read_context.clone();
+ let remote_log_downloader = Arc::clone(&self.remote_log_downloader);
+ let creds_cache = self.credentials_cache.clone();
+ let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone();
+
+ // Spawn async task to handle the fetch request
+ tokio::spawn(async move {
+ // make sure it will always remote leader from pending nodes
+ let _guard = scopeguard::guard((), |_| {
+ nodes_with_pending.lock().remove(&leader);
+ });
+
+ let server_node = cluster
+ .get_tablet_server(leader)
+ .expect("todo: handle leader not exist.");
+
+ let con = match conns.get_connection(server_node).await {
+ Ok(con) => con,
+ Err(e) => {
+ // todo: handle failed to get connection
+ warn!("Failed to get connection to destination node: {e:?}");
+ return;
+ }
+ };
+
+ let fetch_response = match con
+ .request(message::FetchLogRequest::new(fetch_request))
+ .await
+ {
+ Ok(resp) => resp,
+ Err(e) => {
+ // todo: handle fetch log from destination node
+ warn!("Failed to fetch log from destination node {server_node:?}: {e:?}");
+ return;
+ }
+ };
+
+ if let Err(e) = Self::handle_fetch_response(
+ fetch_response,
+ &log_fetch_buffer,
+ &log_scanner_status,
+ &read_context,
+ &remote_read_context,
+ &remote_log_downloader,
+ &creds_cache,
+ )
+ .await
+ {
+ // todo: handle fail to handle fetch response
+ error!("Fail to handle fetch response: {e:?}");
+ }
+ });
+ }
+
+ Ok(())
+ }
- let download_future =
- self.remote_log_downloader.request_remote_log(
- &remote_fetch_info.remote_log_tablet_dir,
- segment,
- )?;
- let pending_fetch = RemotePendingFetch::new(
- segment.clone(),
- download_future,
- pos_in_log_segment,
- current_fetch_offset,
- high_watermark,
- self.read_context.clone(),
- );
- let remote_records =
- pending_fetch.convert_to_completed_fetch().await?;
- // Update offset and merge results
- for (tb, records) in remote_records {
- if let Some(last_record) = records.last() {
- self.log_scanner_status
- .update_offset(&tb, last_record.offset() + 1);
- }
- result.entry(tb).or_default().extend(records);
+ /// Handle fetch response and add completed fetches to buffer
+ async fn handle_fetch_response(
+ fetch_response: crate::proto::FetchLogResponse,
+ log_fetch_buffer: &Arc<LogFetchBuffer>,
+ log_scanner_status: &Arc<LogScannerStatus>,
+ read_context: &ReadContext,
+ remote_read_context: &ReadContext,
+ remote_log_downloader: &Arc<RemoteLogDownloader>,
+ credentials_cache: &Arc<CredentialsCache>,
+ ) -> Result<()> {
+ for pb_fetch_log_resp in fetch_response.tables_resp {
+ let table_id = pb_fetch_log_resp.table_id;
+ let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
+
+ for fetch_log_for_bucket in fetch_log_for_buckets {
+ let bucket: i32 = fetch_log_for_bucket.bucket_id;
+ let table_bucket = TableBucket::new(table_id, bucket);
+
+ // todo: check fetch result code for per-bucket
+ let Some(fetch_offset) = log_scanner_status.get_bucket_offset(&table_bucket) else {
+ debug!(
+ "Ignoring fetch log response for bucket {table_bucket} because the bucket has been unsubscribed."
+ );
+ continue;
+ };
+
+ // Check if this is a remote log fetch
+ if let Some(ref remote_log_fetch_info) = fetch_log_for_bucket.remote_log_fetch_info
+ {
+ // set remote fs props
+ let remote_fs_props = credentials_cache.get_or_refresh().await?;
+ remote_log_downloader.set_remote_fs_props(remote_fs_props);
+
+ let remote_fetch_info =
+ RemoteLogFetchInfo::from_proto(remote_log_fetch_info, table_bucket.clone());
+
+ let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
+ Self::pending_remote_fetches(
+ remote_log_downloader.clone(),
+ log_fetch_buffer.clone(),
+ remote_read_context.clone(),
+ &table_bucket,
+ remote_fetch_info,
+ fetch_offset,
+ high_watermark,
+ );
+ } else if fetch_log_for_bucket.records.is_some() {
+ // Handle regular in-memory records - create completed fetch directly
+ let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
+ let records = fetch_log_for_bucket.records.unwrap_or(vec![]);
+ let size_in_bytes = records.len();
+ let log_record_batch = LogRecordsBatches::new(records);
+
+ match DefaultCompletedFetch::new(
+ table_bucket.clone(),
+ log_record_batch,
+ size_in_bytes,
+ read_context.clone(),
+ fetch_offset,
+ high_watermark,
+ ) {
+ Ok(completed_fetch) => {
+ log_fetch_buffer.add(Box::new(completed_fetch));
+ }
+ Err(e) => {
+ // todo: handle error
+ log::warn!("Failed to create completed fetch: {e:?}");
+ }
+ }
+ }
+ }
+ }
+ Ok(())
+ }
+
+ fn pending_remote_fetches(
+ remote_log_downloader: Arc<RemoteLogDownloader>,
+ log_fetch_buffer: Arc<LogFetchBuffer>,
+ read_context: ReadContext,
+ table_bucket: &TableBucket,
+ remote_fetch_info: RemoteLogFetchInfo,
+ fetch_offset: i64,
+ high_watermark: i64,
+ ) {
+ // Download and process remote log segments
+ let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
+ let mut current_fetch_offset = fetch_offset;
+ for (i, segment) in remote_fetch_info.remote_log_segments.iter().enumerate() {
+ if i > 0 {
+ pos_in_log_segment = 0;
+ current_fetch_offset = segment.start_offset;
+ }
+
+ // todo:
+ // 1: control the max threads to download remote segment
+ // 2: introduce priority queue to priority highest for earliest segment
+ let download_future = remote_log_downloader
+ .request_remote_log(&remote_fetch_info.remote_log_tablet_dir, segment);
+
+ // Register callback to be called when download completes
+ // (similar to Java's downloadFuture.onComplete)
+ // This must be done before creating RemotePendingFetch to avoid move issues
+ let table_bucket = table_bucket.clone();
+ let log_fetch_buffer_clone = log_fetch_buffer.clone();
+ download_future.on_complete(move || {
+ log_fetch_buffer_clone.try_complete(&table_bucket);
+ });
+
+ let pending_fetch = RemotePendingFetch::new(
+ segment.clone(),
+ download_future,
+ pos_in_log_segment,
+ current_fetch_offset,
+ high_watermark,
+ read_context.clone(),
+ );
+ // Add to pending fetches in buffer (similar to Java's logFetchBuffer.pend)
+ log_fetch_buffer.pend(Box::new(pending_fetch));
+ }
+ }
+
+ /// Collect completed fetches from buffer
+ /// Reference: LogFetchCollector.collectFetch in Java
+ fn collect_fetches(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ const MAX_POLL_RECORDS: usize = 500; // Default max poll records
+ let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+ let mut records_remaining = MAX_POLL_RECORDS;
+
+ while records_remaining > 0 {
+ // Get the next in line fetch, or get a new one from buffer
+ let next_in_line = self.log_fetch_buffer.next_in_line_fetch();
Review Comment:
Rust's implementation seems to transfer ownership of `next_in_line_fetch` to the current thread. If not all records are consumed in this round, are the unconsumed records dropped?
I've observed that in Java's implementation, `next_in_line_fetch` uses a reference, while Rust's implementation transfers ownership. I'm unclear on how this is handled.
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -211,125 +260,292 @@ impl LogFetcher {
projected_fields: Option<Vec<usize>>,
) -> Result<Self> {
let full_arrow_schema = to_arrow_schema(table_info.get_row_type());
- let read_context = Self::create_read_context(full_arrow_schema, projected_fields.clone());
+ let read_context =
+ Self::create_read_context(full_arrow_schema.clone(), projected_fields.clone(), false);
+ let remote_read_context =
+ Self::create_read_context(full_arrow_schema, projected_fields.clone(), true);
let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
Ok(LogFetcher {
- table_path: table_info.table_path.clone(),
- conns,
- table_info,
- metadata,
+ conns: conns.clone(),
+ metadata: metadata.clone(),
log_scanner_status,
read_context,
- remote_log_downloader: RemoteLogDownloader::new(tmp_dir)?,
- credentials_cache: CredentialsCache::new(),
+ remote_read_context,
+ remote_log_downloader: Arc::new(RemoteLogDownloader::new(tmp_dir)?),
+ credentials_cache: Arc::new(CredentialsCache::new(conns.clone(), metadata.clone())),
+ log_fetch_buffer: Arc::new(LogFetchBuffer::new()),
+ nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())),
})
}
fn create_read_context(
full_arrow_schema: SchemaRef,
projected_fields: Option<Vec<usize>>,
+ is_from_remote: bool,
) -> ReadContext {
match projected_fields {
- None => ReadContext::new(full_arrow_schema),
- Some(fields) => ReadContext::with_projection_pushdown(full_arrow_schema, fields),
+ None => ReadContext::new(full_arrow_schema, is_from_remote),
+ Some(fields) => {
+ ReadContext::with_projection_pushdown(full_arrow_schema, fields, is_from_remote)
+ }
}
}
- async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ /// Send fetch requests asynchronously without waiting for responses
+ async fn send_fetches(&self) -> Result<()> {
+ // todo: check update metadata like fluss-java in case leader changes
let fetch_request = self.prepare_fetch_log_requests().await;
- let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+
for (leader, fetch_request) in fetch_request {
- let cluster = self.metadata.get_cluster();
- let server_node = cluster
- .get_tablet_server(leader)
- .expect("todo: handle leader not exist.");
- let con = self.conns.get_connection(server_node).await?;
-
- let fetch_response = con
- .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
- .await?;
-
- for pb_fetch_log_resp in fetch_response.tables_resp {
- let table_id = pb_fetch_log_resp.table_id;
- let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
-
- for fetch_log_for_bucket in fetch_log_for_buckets {
- let bucket: i32 = fetch_log_for_bucket.bucket_id;
- let table_bucket = TableBucket::new(table_id, bucket);
-
- // Check if this is a remote log fetch
- if let Some(ref remote_log_fetch_info) =
- fetch_log_for_bucket.remote_log_fetch_info
- {
- let remote_fs_props = self
- .credentials_cache
- .get_or_refresh(&self.conns, &self.metadata)
- .await?;
- self.remote_log_downloader
- .set_remote_fs_props(remote_fs_props);
- let remote_fetch_info = RemoteLogFetchInfo::from_proto(
- remote_log_fetch_info,
- table_bucket.clone(),
- )?;
-
- if let Some(fetch_offset) =
- self.log_scanner_status.get_bucket_offset(&table_bucket)
- {
- let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
- // Download and process remote log segments
- let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
- let mut current_fetch_offset = fetch_offset;
- // todo: make segment download in parallel
- for (i, segment) in
- remote_fetch_info.remote_log_segments.iter().enumerate()
- {
- if i > 0 {
- pos_in_log_segment = 0;
- current_fetch_offset = segment.start_offset;
- }
+ debug!("Adding pending request for node id {leader}");
+ // Check if we already have a pending request for this node
+ {
+ self.nodes_with_pending_fetch_requests.lock().insert(leader);
+ }
+
+ let cluster = self.metadata.get_cluster().clone();
+
+ let conns = Arc::clone(&self.conns);
+ let log_fetch_buffer = self.log_fetch_buffer.clone();
+ let log_scanner_status = self.log_scanner_status.clone();
+ let read_context = self.read_context.clone();
+ let remote_read_context = self.remote_read_context.clone();
+ let remote_log_downloader = Arc::clone(&self.remote_log_downloader);
+ let creds_cache = self.credentials_cache.clone();
+ let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone();
+
+ // Spawn async task to handle the fetch request
+ tokio::spawn(async move {
+ // make sure it will always remote leader from pending nodes
+ let _guard = scopeguard::guard((), |_| {
+ nodes_with_pending.lock().remove(&leader);
+ });
+
+ let server_node = cluster
+ .get_tablet_server(leader)
+ .expect("todo: handle leader not exist.");
+
+ let con = match conns.get_connection(server_node).await {
+ Ok(con) => con,
+ Err(e) => {
+ // todo: handle failed to get connection
+ warn!("Failed to get connection to destination node: {e:?}");
+ return;
+ }
+ };
+
+ let fetch_response = match con
+ .request(message::FetchLogRequest::new(fetch_request))
+ .await
+ {
+ Ok(resp) => resp,
+ Err(e) => {
+ // todo: handle fetch log from destination node
+ warn!("Failed to fetch log from destination node {server_node:?}: {e:?}");
+ return;
+ }
+ };
+
+ if let Err(e) = Self::handle_fetch_response(
+ fetch_response,
+ &log_fetch_buffer,
+ &log_scanner_status,
+ &read_context,
+ &remote_read_context,
+ &remote_log_downloader,
+ &creds_cache,
+ )
+ .await
+ {
+ // todo: handle fail to handle fetch response
+ error!("Fail to handle fetch response: {e:?}");
+ }
+ });
+ }
+
+ Ok(())
+ }
- let download_future =
- self.remote_log_downloader.request_remote_log(
- &remote_fetch_info.remote_log_tablet_dir,
- segment,
- )?;
- let pending_fetch = RemotePendingFetch::new(
- segment.clone(),
- download_future,
- pos_in_log_segment,
- current_fetch_offset,
- high_watermark,
- self.read_context.clone(),
- );
- let remote_records =
- pending_fetch.convert_to_completed_fetch().await?;
- // Update offset and merge results
- for (tb, records) in remote_records {
- if let Some(last_record) = records.last() {
- self.log_scanner_status
- .update_offset(&tb, last_record.offset() + 1);
- }
- result.entry(tb).or_default().extend(records);
+ /// Handle fetch response and add completed fetches to buffer
+ async fn handle_fetch_response(
+ fetch_response: crate::proto::FetchLogResponse,
+ log_fetch_buffer: &Arc<LogFetchBuffer>,
+ log_scanner_status: &Arc<LogScannerStatus>,
+ read_context: &ReadContext,
+ remote_read_context: &ReadContext,
+ remote_log_downloader: &Arc<RemoteLogDownloader>,
+ credentials_cache: &Arc<CredentialsCache>,
+ ) -> Result<()> {
+ for pb_fetch_log_resp in fetch_response.tables_resp {
+ let table_id = pb_fetch_log_resp.table_id;
+ let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
+
+ for fetch_log_for_bucket in fetch_log_for_buckets {
+ let bucket: i32 = fetch_log_for_bucket.bucket_id;
+ let table_bucket = TableBucket::new(table_id, bucket);
+
+ // todo: check fetch result code for per-bucket
+ let Some(fetch_offset) = log_scanner_status.get_bucket_offset(&table_bucket) else {
+ debug!(
+ "Ignoring fetch log response for bucket {table_bucket} because the bucket has been unsubscribed."
+ );
+ continue;
+ };
+
+ // Check if this is a remote log fetch
+ if let Some(ref remote_log_fetch_info) = fetch_log_for_bucket.remote_log_fetch_info
+ {
+ // set remote fs props
+ let remote_fs_props = credentials_cache.get_or_refresh().await?;
+ remote_log_downloader.set_remote_fs_props(remote_fs_props);
+
+ let remote_fetch_info =
+ RemoteLogFetchInfo::from_proto(remote_log_fetch_info, table_bucket.clone());
+
+ let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
+ Self::pending_remote_fetches(
+ remote_log_downloader.clone(),
+ log_fetch_buffer.clone(),
+ remote_read_context.clone(),
+ &table_bucket,
+ remote_fetch_info,
+ fetch_offset,
+ high_watermark,
+ );
+ } else if fetch_log_for_bucket.records.is_some() {
+ // Handle regular in-memory records - create completed fetch directly
+ let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
+ let records = fetch_log_for_bucket.records.unwrap_or(vec![]);
+ let size_in_bytes = records.len();
+ let log_record_batch = LogRecordsBatches::new(records);
+
+ match DefaultCompletedFetch::new(
+ table_bucket.clone(),
+ log_record_batch,
+ size_in_bytes,
+ read_context.clone(),
+ fetch_offset,
+ high_watermark,
+ ) {
+ Ok(completed_fetch) => {
+ log_fetch_buffer.add(Box::new(completed_fetch));
+ }
+ Err(e) => {
+ // todo: handle error
+ log::warn!("Failed to create completed fetch: {e:?}");
+ }
+ }
+ }
+ }
+ }
+ Ok(())
+ }
+
+ fn pending_remote_fetches(
+ remote_log_downloader: Arc<RemoteLogDownloader>,
+ log_fetch_buffer: Arc<LogFetchBuffer>,
+ read_context: ReadContext,
+ table_bucket: &TableBucket,
+ remote_fetch_info: RemoteLogFetchInfo,
+ fetch_offset: i64,
+ high_watermark: i64,
+ ) {
+ // Download and process remote log segments
+ let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
+ let mut current_fetch_offset = fetch_offset;
+ for (i, segment) in remote_fetch_info.remote_log_segments.iter().enumerate() {
+ if i > 0 {
+ pos_in_log_segment = 0;
+ current_fetch_offset = segment.start_offset;
+ }
+
+ // todo:
+ // 1: control the max threads to download remote segment
+ // 2: introduce priority queue to priority highest for earliest segment
+ let download_future = remote_log_downloader
+ .request_remote_log(&remote_fetch_info.remote_log_tablet_dir, segment);
+
+ // Register callback to be called when download completes
+ // (similar to Java's downloadFuture.onComplete)
+ // This must be done before creating RemotePendingFetch to avoid move issues
+ let table_bucket = table_bucket.clone();
+ let log_fetch_buffer_clone = log_fetch_buffer.clone();
+ download_future.on_complete(move || {
+ log_fetch_buffer_clone.try_complete(&table_bucket);
+ });
+
+ let pending_fetch = RemotePendingFetch::new(
+ segment.clone(),
+ download_future,
+ pos_in_log_segment,
+ current_fetch_offset,
+ high_watermark,
+ read_context.clone(),
+ );
+ // Add to pending fetches in buffer (similar to Java's logFetchBuffer.pend)
+ log_fetch_buffer.pend(Box::new(pending_fetch));
+ }
+ }
+
+ /// Collect completed fetches from buffer
+ /// Reference: LogFetchCollector.collectFetch in Java
+ fn collect_fetches(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ const MAX_POLL_RECORDS: usize = 500; // Default max poll records
+ let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+ let mut records_remaining = MAX_POLL_RECORDS;
+
+ while records_remaining > 0 {
+ // Get the next in line fetch, or get a new one from buffer
+ let next_in_line = self.log_fetch_buffer.next_in_line_fetch();
+
+ if next_in_line.is_none() || next_in_line.as_ref().unwrap().is_consumed() {
+ // Get a new fetch from buffer
+ if let Some(completed_fetch) = self.log_fetch_buffer.poll() {
+ // Initialize the fetch if not already initialized
+ if !completed_fetch.is_initialized() {
+ let size_in_bytes = completed_fetch.size_in_bytes();
+ match self.initialize_fetch(completed_fetch) {
+ Ok(initialized) => {
+ self.log_fetch_buffer.set_next_in_line_fetch(initialized);
+ continue;
+ }
+ Err(e) => {
+ // Remove a completedFetch upon a parse with exception if
+ // (1) it contains no records, and
+ // (2) there are no fetched records with actual content preceding this
+ // exception.
+ if result.is_empty() && size_in_bytes == 0 {
+ // todo: do we need to consider it like java ?
+ // self.log_fetch_buffer.poll();
}
+ return Err(e);
}
- } else {
- // if the offset is null, it means the bucket has been unsubscribed,
- // skip processing and continue to the next bucket.
- continue;
}
- } else if fetch_log_for_bucket.records.is_some() {
- // Handle regular in-memory records
- let mut fetch_records = vec![];
- let data = fetch_log_for_bucket.records.unwrap();
- for log_record in &mut LogRecordsBatchs::new(&data) {
- let last_offset = log_record.last_log_offset();
- fetch_records.extend(log_record.records(&self.read_context)?);
- self.log_scanner_status
- .update_offset(&table_bucket, last_offset + 1);
- }
- result.insert(table_bucket, fetch_records);
+ } else {
+ self.log_fetch_buffer
+ .set_next_in_line_fetch(Some(completed_fetch));
+ }
+ // Note: peek() already removed the fetch from buffer, so no need to call poll()
Review Comment:
If I understand correctly, are you trying to express the following meaning?
Note: poll() already removed the fetch from buffer, so no need to call poll() again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637515748
##########
crates/fluss/src/client/table/remote_log.rs:
##########
@@ -286,32 +351,54 @@ impl RemotePendingFetch {
read_context,
}
}
+}
+
+impl PendingFetch for RemotePendingFetch {
+ fn table_bucket(&self) -> &TableBucket {
+ &self.segment.table_bucket
+ }
- /// Convert to completed fetch by reading the downloaded file
- pub async fn convert_to_completed_fetch(
- mut self,
- ) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
- let file_path = self.download_future.get_file_path().await?;
- let file_data = tokio::fs::read(&file_path).await?;
+ fn is_completed(&self) -> bool {
+ self.download_future.is_done()
+ }
+
+ fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>> {
+ // Get the file path (this should only be called when is_completed() returns true)
+ let file_path = self.download_future.get_file_path()?;
+ // Read the file data synchronously (we're in a sync context)
+ // Note: This is a limitation - we need to use blocking I/O here
+ let mut file_data = std::fs::read(&file_path).map_err(|e| Error::IoUnexpectedError {
+ message: format!("Failed to read downloaded file: {file_path:?}."),
+ source: e,
+ })?;
// Slice the data if needed
let data = if self.pos_in_log_segment > 0 {
- &file_data[self.pos_in_log_segment as usize..]
+ file_data.split_off(self.pos_in_log_segment as usize)
} else {
- &file_data
+ file_data
};
- // delete the downloaded local file to free disk
- delete_file(file_path).await;
+ let size_in_bytes = data.len();
- // Parse log records (remote log contains full data, need client-side projection)
- let mut fetch_records = vec![];
- for log_record in &mut LogRecordsBatchs::new(data) {
- fetch_records.extend(log_record.records_for_remote_log(&self.read_context)?);
- }
+ let log_record_batch = LogRecordsBatches::new(data);
+
+ // Create DefaultCompletedFetch from the data
+ let completed_fetch = DefaultCompletedFetch::new(
+ self.segment.table_bucket,
+ log_record_batch,
+ size_in_bytes,
+ self.read_context,
+ self.fetch_offset,
+ self.high_watermark,
+ )?;
+
+ // Delete the downloaded local file to free disk (async, but we'll do it in background)
+ let file_path_clone = file_path.clone();
+ tokio::spawn(async move {
+ let _ = delete_file(file_path_clone).await;
Review Comment:
`delete_file` will warn it.
After `RemoteLogDownloader` is moved of out scope, the `TempDir` will be drop, causing the all downloaded files deleted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637522774
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -211,125 +260,292 @@ impl LogFetcher {
projected_fields: Option<Vec<usize>>,
) -> Result<Self> {
let full_arrow_schema = to_arrow_schema(table_info.get_row_type());
- let read_context = Self::create_read_context(full_arrow_schema, projected_fields.clone());
+ let read_context =
+ Self::create_read_context(full_arrow_schema.clone(), projected_fields.clone(), false);
+ let remote_read_context =
+ Self::create_read_context(full_arrow_schema, projected_fields.clone(), true);
let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
Ok(LogFetcher {
- table_path: table_info.table_path.clone(),
- conns,
- table_info,
- metadata,
+ conns: conns.clone(),
+ metadata: metadata.clone(),
log_scanner_status,
read_context,
- remote_log_downloader: RemoteLogDownloader::new(tmp_dir)?,
- credentials_cache: CredentialsCache::new(),
+ remote_read_context,
+ remote_log_downloader: Arc::new(RemoteLogDownloader::new(tmp_dir)?),
+ credentials_cache: Arc::new(CredentialsCache::new(conns.clone(), metadata.clone())),
+ log_fetch_buffer: Arc::new(LogFetchBuffer::new()),
+ nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())),
})
}
fn create_read_context(
full_arrow_schema: SchemaRef,
projected_fields: Option<Vec<usize>>,
+ is_from_remote: bool,
) -> ReadContext {
match projected_fields {
- None => ReadContext::new(full_arrow_schema),
- Some(fields) => ReadContext::with_projection_pushdown(full_arrow_schema, fields),
+ None => ReadContext::new(full_arrow_schema, is_from_remote),
+ Some(fields) => {
+ ReadContext::with_projection_pushdown(full_arrow_schema, fields, is_from_remote)
+ }
}
}
- async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ /// Send fetch requests asynchronously without waiting for responses
+ async fn send_fetches(&self) -> Result<()> {
+ // todo: check update metadata like fluss-java in case leader changes
let fetch_request = self.prepare_fetch_log_requests().await;
- let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+
for (leader, fetch_request) in fetch_request {
- let cluster = self.metadata.get_cluster();
- let server_node = cluster
- .get_tablet_server(leader)
- .expect("todo: handle leader not exist.");
- let con = self.conns.get_connection(server_node).await?;
-
- let fetch_response = con
- .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
- .await?;
-
- for pb_fetch_log_resp in fetch_response.tables_resp {
- let table_id = pb_fetch_log_resp.table_id;
- let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
-
- for fetch_log_for_bucket in fetch_log_for_buckets {
- let bucket: i32 = fetch_log_for_bucket.bucket_id;
- let table_bucket = TableBucket::new(table_id, bucket);
-
- // Check if this is a remote log fetch
- if let Some(ref remote_log_fetch_info) =
- fetch_log_for_bucket.remote_log_fetch_info
- {
- let remote_fs_props = self
- .credentials_cache
- .get_or_refresh(&self.conns, &self.metadata)
- .await?;
- self.remote_log_downloader
- .set_remote_fs_props(remote_fs_props);
- let remote_fetch_info = RemoteLogFetchInfo::from_proto(
- remote_log_fetch_info,
- table_bucket.clone(),
- )?;
-
- if let Some(fetch_offset) =
- self.log_scanner_status.get_bucket_offset(&table_bucket)
- {
- let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
- // Download and process remote log segments
- let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
- let mut current_fetch_offset = fetch_offset;
- // todo: make segment download in parallel
- for (i, segment) in
- remote_fetch_info.remote_log_segments.iter().enumerate()
- {
- if i > 0 {
- pos_in_log_segment = 0;
- current_fetch_offset = segment.start_offset;
- }
+ debug!("Adding pending request for node id {leader}");
+ // Check if we already have a pending request for this node
+ {
+ self.nodes_with_pending_fetch_requests.lock().insert(leader);
+ }
+
+ let cluster = self.metadata.get_cluster().clone();
+
+ let conns = Arc::clone(&self.conns);
+ let log_fetch_buffer = self.log_fetch_buffer.clone();
+ let log_scanner_status = self.log_scanner_status.clone();
+ let read_context = self.read_context.clone();
+ let remote_read_context = self.remote_read_context.clone();
+ let remote_log_downloader = Arc::clone(&self.remote_log_downloader);
+ let creds_cache = self.credentials_cache.clone();
+ let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone();
+
+ // Spawn async task to handle the fetch request
+ tokio::spawn(async move {
+ // make sure it will always remote leader from pending nodes
+ let _guard = scopeguard::guard((), |_| {
+ nodes_with_pending.lock().remove(&leader);
+ });
+
+ let server_node = cluster
+ .get_tablet_server(leader)
+ .expect("todo: handle leader not exist.");
+
+ let con = match conns.get_connection(server_node).await {
+ Ok(con) => con,
+ Err(e) => {
+ // todo: handle failed to get connection
+ warn!("Failed to get connection to destination node: {e:?}");
+ return;
+ }
+ };
+
+ let fetch_response = match con
+ .request(message::FetchLogRequest::new(fetch_request))
+ .await
+ {
+ Ok(resp) => resp,
+ Err(e) => {
+ // todo: handle fetch log from destination node
+ warn!("Failed to fetch log from destination node {server_node:?}: {e:?}");
+ return;
+ }
+ };
+
+ if let Err(e) = Self::handle_fetch_response(
+ fetch_response,
+ &log_fetch_buffer,
+ &log_scanner_status,
+ &read_context,
+ &remote_read_context,
+ &remote_log_downloader,
+ &creds_cache,
+ )
+ .await
+ {
+ // todo: handle fail to handle fetch response
+ error!("Fail to handle fetch response: {e:?}");
+ }
+ });
+ }
+
+ Ok(())
+ }
- let download_future =
- self.remote_log_downloader.request_remote_log(
- &remote_fetch_info.remote_log_tablet_dir,
- segment,
- )?;
- let pending_fetch = RemotePendingFetch::new(
- segment.clone(),
- download_future,
- pos_in_log_segment,
- current_fetch_offset,
- high_watermark,
- self.read_context.clone(),
- );
- let remote_records =
- pending_fetch.convert_to_completed_fetch().await?;
- // Update offset and merge results
- for (tb, records) in remote_records {
- if let Some(last_record) = records.last() {
- self.log_scanner_status
- .update_offset(&tb, last_record.offset() + 1);
- }
- result.entry(tb).or_default().extend(records);
+ /// Handle fetch response and add completed fetches to buffer
+ async fn handle_fetch_response(
+ fetch_response: crate::proto::FetchLogResponse,
+ log_fetch_buffer: &Arc<LogFetchBuffer>,
+ log_scanner_status: &Arc<LogScannerStatus>,
+ read_context: &ReadContext,
+ remote_read_context: &ReadContext,
+ remote_log_downloader: &Arc<RemoteLogDownloader>,
+ credentials_cache: &Arc<CredentialsCache>,
+ ) -> Result<()> {
+ for pb_fetch_log_resp in fetch_response.tables_resp {
+ let table_id = pb_fetch_log_resp.table_id;
+ let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
+
+ for fetch_log_for_bucket in fetch_log_for_buckets {
+ let bucket: i32 = fetch_log_for_bucket.bucket_id;
+ let table_bucket = TableBucket::new(table_id, bucket);
+
+ // todo: check fetch result code for per-bucket
+ let Some(fetch_offset) = log_scanner_status.get_bucket_offset(&table_bucket) else {
+ debug!(
+ "Ignoring fetch log response for bucket {table_bucket} because the bucket has been unsubscribed."
+ );
+ continue;
+ };
+
+ // Check if this is a remote log fetch
+ if let Some(ref remote_log_fetch_info) = fetch_log_for_bucket.remote_log_fetch_info
+ {
+ // set remote fs props
+ let remote_fs_props = credentials_cache.get_or_refresh().await?;
+ remote_log_downloader.set_remote_fs_props(remote_fs_props);
+
+ let remote_fetch_info =
+ RemoteLogFetchInfo::from_proto(remote_log_fetch_info, table_bucket.clone());
+
+ let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
+ Self::pending_remote_fetches(
+ remote_log_downloader.clone(),
+ log_fetch_buffer.clone(),
+ remote_read_context.clone(),
+ &table_bucket,
+ remote_fetch_info,
+ fetch_offset,
+ high_watermark,
+ );
+ } else if fetch_log_for_bucket.records.is_some() {
+ // Handle regular in-memory records - create completed fetch directly
+ let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
+ let records = fetch_log_for_bucket.records.unwrap_or(vec![]);
+ let size_in_bytes = records.len();
+ let log_record_batch = LogRecordsBatches::new(records);
+
+ match DefaultCompletedFetch::new(
+ table_bucket.clone(),
+ log_record_batch,
+ size_in_bytes,
+ read_context.clone(),
+ fetch_offset,
+ high_watermark,
+ ) {
+ Ok(completed_fetch) => {
+ log_fetch_buffer.add(Box::new(completed_fetch));
+ }
+ Err(e) => {
+ // todo: handle error
+ log::warn!("Failed to create completed fetch: {e:?}");
+ }
+ }
+ }
+ }
+ }
+ Ok(())
+ }
+
+ fn pending_remote_fetches(
+ remote_log_downloader: Arc<RemoteLogDownloader>,
+ log_fetch_buffer: Arc<LogFetchBuffer>,
+ read_context: ReadContext,
+ table_bucket: &TableBucket,
+ remote_fetch_info: RemoteLogFetchInfo,
+ fetch_offset: i64,
+ high_watermark: i64,
+ ) {
+ // Download and process remote log segments
+ let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
+ let mut current_fetch_offset = fetch_offset;
+ for (i, segment) in remote_fetch_info.remote_log_segments.iter().enumerate() {
+ if i > 0 {
+ pos_in_log_segment = 0;
+ current_fetch_offset = segment.start_offset;
+ }
+
+ // todo:
+ // 1: control the max threads to download remote segment
+ // 2: introduce priority queue to priority highest for earliest segment
+ let download_future = remote_log_downloader
+ .request_remote_log(&remote_fetch_info.remote_log_tablet_dir, segment);
+
+ // Register callback to be called when download completes
+ // (similar to Java's downloadFuture.onComplete)
+ // This must be done before creating RemotePendingFetch to avoid move issues
+ let table_bucket = table_bucket.clone();
+ let log_fetch_buffer_clone = log_fetch_buffer.clone();
+ download_future.on_complete(move || {
+ log_fetch_buffer_clone.try_complete(&table_bucket);
+ });
+
+ let pending_fetch = RemotePendingFetch::new(
+ segment.clone(),
+ download_future,
+ pos_in_log_segment,
+ current_fetch_offset,
+ high_watermark,
+ read_context.clone(),
+ );
+ // Add to pending fetches in buffer (similar to Java's logFetchBuffer.pend)
+ log_fetch_buffer.pend(Box::new(pending_fetch));
+ }
+ }
+
+ /// Collect completed fetches from buffer
+ /// Reference: LogFetchCollector.collectFetch in Java
+ fn collect_fetches(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ const MAX_POLL_RECORDS: usize = 500; // Default max poll records
+ let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+ let mut records_remaining = MAX_POLL_RECORDS;
+
+ while records_remaining > 0 {
+ // Get the next in line fetch, or get a new one from buffer
+ let next_in_line = self.log_fetch_buffer.next_in_line_fetch();
Review Comment:
Good catch. I think you're right. It'll cause unconsumed records dropped
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637519758
##########
crates/fluss/src/client/table/log_fetch_buffer.rs:
##########
@@ -0,0 +1,367 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 crate::error::Result;
+use crate::metadata::TableBucket;
+use crate::record::{
+ LogRecordBatch, LogRecordIterator, LogRecordsBatches, ReadContext, ScanRecord,
+};
+use parking_lot::Mutex;
+use std::collections::{HashMap, VecDeque};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use tokio::sync::Notify;
+
+/// Represents a completed fetch that can be consumed
+pub trait CompletedFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn fetch_records(&mut self, max_records: usize) -> Result<Vec<ScanRecord>>;
+ fn is_consumed(&self) -> bool;
+ fn drain(&mut self);
+ fn size_in_bytes(&self) -> usize;
+ fn high_watermark(&self) -> i64;
+ fn is_initialized(&self) -> bool;
+ fn set_initialized(&mut self);
+ fn next_fetch_offset(&self) -> i64;
+}
+
+/// Represents a pending fetch that is waiting to be completed
+pub trait PendingFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn is_completed(&self) -> bool;
+ fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>>;
+}
+
+/// Thread-safe buffer for completed fetches
+pub struct LogFetchBuffer {
+ completed_fetches: Mutex<VecDeque<Box<dyn CompletedFetch>>>,
+ pending_fetches: Mutex<HashMap<TableBucket, VecDeque<Box<dyn PendingFetch>>>>,
+ next_in_line_fetch: Mutex<Option<Box<dyn CompletedFetch>>>,
+ not_empty_notify: Notify,
+ woken_up: Arc<AtomicBool>,
+}
+
+impl LogFetchBuffer {
+ pub fn new() -> Self {
+ Self {
+ completed_fetches: Mutex::new(VecDeque::new()),
+ pending_fetches: Mutex::new(HashMap::new()),
+ next_in_line_fetch: Mutex::new(None),
+ not_empty_notify: Notify::new(),
+ woken_up: Arc::new(AtomicBool::new(false)),
+ }
+ }
+
+ /// Check if the buffer is empty
+ pub fn is_empty(&self) -> bool {
+ self.completed_fetches.lock().is_empty()
+ }
+
+ /// Wait for the buffer to become non-empty, with timeout
+ /// Returns true if data became available, false if timeout
+ pub async fn await_not_empty(&self, timeout: Duration) -> bool {
+ let deadline = std::time::Instant::now() + timeout;
+
+ loop {
+ // Check if buffer is not empty
+ if !self.is_empty() {
Review Comment:
I think we can keep it simple now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637514034
##########
crates/fluss/src/client/table/log_fetch_buffer.rs:
##########
@@ -0,0 +1,367 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 crate::error::Result;
+use crate::metadata::TableBucket;
+use crate::record::{
+ LogRecordBatch, LogRecordIterator, LogRecordsBatches, ReadContext, ScanRecord,
+};
+use parking_lot::Mutex;
+use std::collections::{HashMap, VecDeque};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use tokio::sync::Notify;
+
+/// Represents a completed fetch that can be consumed
+pub trait CompletedFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn fetch_records(&mut self, max_records: usize) -> Result<Vec<ScanRecord>>;
+ fn is_consumed(&self) -> bool;
+ fn drain(&mut self);
+ fn size_in_bytes(&self) -> usize;
+ fn high_watermark(&self) -> i64;
+ fn is_initialized(&self) -> bool;
+ fn set_initialized(&mut self);
+ fn next_fetch_offset(&self) -> i64;
+}
+
+/// Represents a pending fetch that is waiting to be completed
+pub trait PendingFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn is_completed(&self) -> bool;
+ fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>>;
+}
+
+/// Thread-safe buffer for completed fetches
+pub struct LogFetchBuffer {
+ completed_fetches: Mutex<VecDeque<Box<dyn CompletedFetch>>>,
+ pending_fetches: Mutex<HashMap<TableBucket, VecDeque<Box<dyn PendingFetch>>>>,
+ next_in_line_fetch: Mutex<Option<Box<dyn CompletedFetch>>>,
+ not_empty_notify: Notify,
+ woken_up: Arc<AtomicBool>,
+}
+
+impl LogFetchBuffer {
+ pub fn new() -> Self {
+ Self {
+ completed_fetches: Mutex::new(VecDeque::new()),
+ pending_fetches: Mutex::new(HashMap::new()),
+ next_in_line_fetch: Mutex::new(None),
+ not_empty_notify: Notify::new(),
+ woken_up: Arc::new(AtomicBool::new(false)),
+ }
+ }
+
+ /// Check if the buffer is empty
+ pub fn is_empty(&self) -> bool {
+ self.completed_fetches.lock().is_empty()
+ }
+
+ /// Wait for the buffer to become non-empty, with timeout
+ /// Returns true if data became available, false if timeout
+ pub async fn await_not_empty(&self, timeout: Duration) -> bool {
+ let deadline = std::time::Instant::now() + timeout;
+
+ loop {
+ // Check if buffer is not empty
+ if !self.is_empty() {
+ return true;
+ }
+
+ // Check if woken up
+ if self.woken_up.swap(false, Ordering::Acquire) {
+ return true;
+ }
+
+ // Check if timeout
+ let now = std::time::Instant::now();
+ if now >= deadline {
+ return false;
+ }
+
+ // Wait for notification with remaining time
+ let remaining = deadline - now;
+ let notified = self.not_empty_notify.notified();
+ tokio::select! {
+ _ = tokio::time::sleep(remaining) => {
+ return false; // Timeout
+ }
+ _ = notified => {
+ // Got notification, check again
+ continue;
+ }
+ }
+ }
+ }
Review Comment:
wakeup current is not used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#issuecomment-3677863720
@luoyuxia Thanks for the pr. Left minor comments from resource management perspective. PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637114720
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -350,25 +643,44 @@ impl LogFetcher {
let offset = match self.log_scanner_status.get_bucket_offset(&bucket) {
Some(offset) => offset,
None => {
- // todo: debug
+ debug!(
+ "Skipping fetch request for bucket {bucket} because the bucket has been unsubscribed."
+ );
continue;
}
};
- if let Some(leader) = self.get_table_bucket_leader(&bucket) {
- let fetch_log_req_for_bucket = PbFetchLogReqForBucket {
- partition_id: None,
- bucket_id: bucket.bucket_id(),
- fetch_offset: offset,
- // 1M
- max_fetch_bytes: 1024 * 1024,
- };
-
- fetch_log_req_for_buckets
- .entry(leader)
- .or_insert_with(Vec::new)
- .push(fetch_log_req_for_bucket);
- ready_for_fetch_count += 1;
+ match self.get_table_bucket_leader(&bucket) {
+ None => {
+ log::trace!(
+ "Skipping fetch request for bucket {bucket} because leader is not available."
+ )
+ }
+ Some(leader) => {
+ if self
+ .nodes_with_pending_fetch_requests
+ .lock()
+ .contains(&leader)
+ {
+ log::trace!(
+ "Skipping fetch request for bucket {bucket} because previous request to server {leader} has not been processed."
+ )
+ } else {
+ let fetch_log_req_for_bucket = PbFetchLogReqForBucket {
+ partition_id: None,
+ bucket_id: bucket.bucket_id(),
+ fetch_offset: offset,
+ // 1M
+ max_fetch_bytes: 1024 * 1024,
+ };
+
+ fetch_log_req_for_buckets
+ .entry(leader)
+ .or_insert_with(Vec::new)
+ .push(fetch_log_req_for_bucket);
+ ready_for_fetch_count += 1;
+ }
+ }
Review Comment:
The fetch request preparation logic now skips buckets with pending requests to a leader (lines 660-667), but this could lead to starvation if a particular leader becomes slow. If one tablet server is slow to respond, all buckets on that server will be blocked from fetching even if other buckets on different servers are ready. Consider adding a timeout for pending requests or allowing concurrent requests per bucket rather than per leader.
##########
crates/fluss/src/client/table/remote_log.rs:
##########
@@ -286,32 +351,54 @@ impl RemotePendingFetch {
read_context,
}
}
+}
+
+impl PendingFetch for RemotePendingFetch {
+ fn table_bucket(&self) -> &TableBucket {
+ &self.segment.table_bucket
+ }
- /// Convert to completed fetch by reading the downloaded file
- pub async fn convert_to_completed_fetch(
- mut self,
- ) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
- let file_path = self.download_future.get_file_path().await?;
- let file_data = tokio::fs::read(&file_path).await?;
+ fn is_completed(&self) -> bool {
+ self.download_future.is_done()
+ }
+
+ fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>> {
+ // Get the file path (this should only be called when is_completed() returns true)
+ let file_path = self.download_future.get_file_path()?;
+ // Read the file data synchronously (we're in a sync context)
+ // Note: This is a limitation - we need to use blocking I/O here
+ let mut file_data = std::fs::read(&file_path).map_err(|e| Error::IoUnexpectedError {
+ message: format!("Failed to read downloaded file: {file_path:?}."),
+ source: e,
+ })?;
Review Comment:
The blocking I/O operation std::fs::read() is called in a potentially async context (within to_completed_fetch which can be called from async code paths). This could block the tokio runtime thread. Consider using tokio::fs::read() with a spawn_blocking wrapper, or documenting that this method should only be called from blocking contexts.
```suggestion
// Read the file data using a Tokio-aware blocking section. This method may be
// called from async code paths, so we use `block_in_place` to avoid stalling
// the runtime worker thread while performing blocking I/O.
let mut file_data = tokio::task::block_in_place(|| std::fs::read(&file_path))
.map_err(|e| Error::IoUnexpectedError {
message: format!("Failed to read downloaded file: {file_path:?}."),
source: e,
})?;
```
##########
crates/fluss/src/client/table/remote_log.rs:
##########
@@ -70,45 +72,110 @@ pub struct RemoteLogFetchInfo {
}
impl RemoteLogFetchInfo {
- pub fn from_proto(info: &PbRemoteLogFetchInfo, table_bucket: TableBucket) -> Result<Self> {
+ pub fn from_proto(info: &PbRemoteLogFetchInfo, table_bucket: TableBucket) -> Self {
let segments = info
.remote_log_segments
.iter()
.map(|s| RemoteLogSegment::from_proto(s, table_bucket.clone()))
.collect();
- Ok(Self {
+ Self {
remote_log_tablet_dir: info.remote_log_tablet_dir.clone(),
partition_name: info.partition_name.clone(),
remote_log_segments: segments,
first_start_pos: info.first_start_pos.unwrap_or(0),
- })
+ }
}
}
+type CompletionCallback = Box<dyn Fn() + Send + Sync>;
+
/// Future for a remote log download request
pub struct RemoteLogDownloadFuture {
- receiver: Option<oneshot::Receiver<Result<PathBuf>>>,
+ result: Arc<Mutex<Option<Result<PathBuf>>>>,
+ completion_callbacks: Arc<Mutex<Vec<CompletionCallback>>>,
+ // todo: add recycleCallback
}
impl RemoteLogDownloadFuture {
pub fn new(receiver: oneshot::Receiver<Result<PathBuf>>) -> Self {
+ let result = Arc::new(Mutex::new(None));
+ let result_clone = Arc::clone(&result);
+ let completion_callbacks: Arc<Mutex<Vec<CompletionCallback>>> =
+ Arc::new(Mutex::new(Vec::new()));
+ let callbacks_clone = Arc::clone(&completion_callbacks);
+
+ // Spawn a task to wait for the download and update result, then call callbacks
+ tokio::spawn(async move {
+ let download_result = match receiver.await {
+ Ok(Ok(path)) => Ok(path),
+ Ok(Err(e)) => Err(e),
+ Err(e) => Err(Error::UnexpectedError {
+ message: format!("Download future cancelled: {e:?}"),
+ source: None,
+ }),
+ };
+ *result_clone.lock() = Some(download_result);
+
+ // Call all registered callbacks
+ // We need to take the callbacks to avoid holding the lock while calling them
+ // This also ensures that any callbacks registered after this point will be called immediately
+ let callbacks: Vec<CompletionCallback> = {
+ let mut callbacks_guard = callbacks_clone.lock();
+ std::mem::take(&mut *callbacks_guard)
+ };
+ for callback in callbacks {
+ callback();
+ }
+
+ // After calling callbacks, any new callbacks registered will see is_done() == true
+ // and will be called immediately in on_complete()
+ });
+
Self {
- receiver: Some(receiver),
+ result,
+ completion_callbacks,
}
}
- /// Get the downloaded file path
- pub async fn get_file_path(&mut self) -> Result<PathBuf> {
- let receiver = self.receiver.take().ok_or_else(|| Error::UnexpectedError {
- message: "Downloaded file already consumed".to_string(),
- source: None,
- })?;
+ /// Register a callback to be called when download completes (similar to Java's onComplete)
+ pub fn on_complete<F>(&self, callback: F)
+ where
+ F: Fn() + Send + Sync + 'static,
+ {
+ // Check if already completed - need to check while holding the lock to avoid race condition
+ let mut callbacks_guard = self.completion_callbacks.lock();
+ let is_done = self.is_done();
+
+ if is_done {
+ // If already completed, call immediately (drop lock first to avoid deadlock)
+ drop(callbacks_guard);
+ callback();
+ } else {
+ // Otherwise, register the callback
+ callbacks_guard.push(Box::new(callback));
+ }
+ }
Review Comment:
The on_complete method has a race condition in checking completion status. The callbacks_guard lock is acquired on line 147, but is_done() on line 148 acquires a separate lock (self.result) to check completion. Between releasing the result lock and checking the boolean, the background task could complete, execute all callbacks, and clear the callback list. This could result in the callback being added to an already-cleared list on line 156, causing it to never be executed. The two locks (callbacks and result) should be checked atomically, or the completion check should be done while holding the callbacks_guard.
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -378,14 +379,17 @@ impl<'a> LogRecordsBatchs<'a> {
}
}
-impl<'a> Iterator for &'a mut LogRecordsBatchs<'a> {
- type Item = LogRecordBatch<'a>;
+impl Iterator for LogRecordsBatches {
+ type Item = LogRecordBatch;
fn next(&mut self) -> Option<Self::Item> {
match self.next_batch_size() {
Some(batch_size) => {
- let data_slice = &self.data[self.current_pos..self.current_pos + batch_size];
- let record_batch = LogRecordBatch::new(data_slice);
+ let start = self.current_pos;
+ let end = start + batch_size;
+ // Since LogRecordsBatchs owns the Vec<u8>, the slice is valid
Review Comment:
The struct name was changed from `LogRecordsBatchs` to `LogRecordsBatches`, but there's a typo in the comment on line 390. The comment refers to "LogRecordsBatchs" (the old, misspelled name) instead of the new correct name "LogRecordsBatches".
```suggestion
// Since LogRecordsBatches owns the Vec<u8>, the slice is valid
```
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -211,125 +260,292 @@ impl LogFetcher {
projected_fields: Option<Vec<usize>>,
) -> Result<Self> {
let full_arrow_schema = to_arrow_schema(table_info.get_row_type());
- let read_context = Self::create_read_context(full_arrow_schema, projected_fields.clone());
+ let read_context =
+ Self::create_read_context(full_arrow_schema.clone(), projected_fields.clone(), false);
+ let remote_read_context =
+ Self::create_read_context(full_arrow_schema, projected_fields.clone(), true);
let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
Ok(LogFetcher {
- table_path: table_info.table_path.clone(),
- conns,
- table_info,
- metadata,
+ conns: conns.clone(),
+ metadata: metadata.clone(),
log_scanner_status,
read_context,
- remote_log_downloader: RemoteLogDownloader::new(tmp_dir)?,
- credentials_cache: CredentialsCache::new(),
+ remote_read_context,
+ remote_log_downloader: Arc::new(RemoteLogDownloader::new(tmp_dir)?),
+ credentials_cache: Arc::new(CredentialsCache::new(conns.clone(), metadata.clone())),
+ log_fetch_buffer: Arc::new(LogFetchBuffer::new()),
+ nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())),
})
}
fn create_read_context(
full_arrow_schema: SchemaRef,
projected_fields: Option<Vec<usize>>,
+ is_from_remote: bool,
) -> ReadContext {
match projected_fields {
- None => ReadContext::new(full_arrow_schema),
- Some(fields) => ReadContext::with_projection_pushdown(full_arrow_schema, fields),
+ None => ReadContext::new(full_arrow_schema, is_from_remote),
+ Some(fields) => {
+ ReadContext::with_projection_pushdown(full_arrow_schema, fields, is_from_remote)
+ }
}
}
- async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ /// Send fetch requests asynchronously without waiting for responses
+ async fn send_fetches(&self) -> Result<()> {
+ // todo: check update metadata like fluss-java in case leader changes
let fetch_request = self.prepare_fetch_log_requests().await;
- let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+
for (leader, fetch_request) in fetch_request {
- let cluster = self.metadata.get_cluster();
- let server_node = cluster
- .get_tablet_server(leader)
- .expect("todo: handle leader not exist.");
- let con = self.conns.get_connection(server_node).await?;
-
- let fetch_response = con
- .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
- .await?;
-
- for pb_fetch_log_resp in fetch_response.tables_resp {
- let table_id = pb_fetch_log_resp.table_id;
- let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
-
- for fetch_log_for_bucket in fetch_log_for_buckets {
- let bucket: i32 = fetch_log_for_bucket.bucket_id;
- let table_bucket = TableBucket::new(table_id, bucket);
-
- // Check if this is a remote log fetch
- if let Some(ref remote_log_fetch_info) =
- fetch_log_for_bucket.remote_log_fetch_info
- {
- let remote_fs_props = self
- .credentials_cache
- .get_or_refresh(&self.conns, &self.metadata)
- .await?;
- self.remote_log_downloader
- .set_remote_fs_props(remote_fs_props);
- let remote_fetch_info = RemoteLogFetchInfo::from_proto(
- remote_log_fetch_info,
- table_bucket.clone(),
- )?;
-
- if let Some(fetch_offset) =
- self.log_scanner_status.get_bucket_offset(&table_bucket)
- {
- let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
- // Download and process remote log segments
- let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
- let mut current_fetch_offset = fetch_offset;
- // todo: make segment download in parallel
- for (i, segment) in
- remote_fetch_info.remote_log_segments.iter().enumerate()
- {
- if i > 0 {
- pos_in_log_segment = 0;
- current_fetch_offset = segment.start_offset;
- }
+ debug!("Adding pending request for node id {leader}");
+ // Check if we already have a pending request for this node
+ {
+ self.nodes_with_pending_fetch_requests.lock().insert(leader);
+ }
+
+ let cluster = self.metadata.get_cluster().clone();
+
+ let conns = Arc::clone(&self.conns);
+ let log_fetch_buffer = self.log_fetch_buffer.clone();
+ let log_scanner_status = self.log_scanner_status.clone();
+ let read_context = self.read_context.clone();
+ let remote_read_context = self.remote_read_context.clone();
+ let remote_log_downloader = Arc::clone(&self.remote_log_downloader);
+ let creds_cache = self.credentials_cache.clone();
+ let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone();
+
+ // Spawn async task to handle the fetch request
+ tokio::spawn(async move {
+ // make sure it will always remote leader from pending nodes
Review Comment:
The error message contains a typo: "remote" should be "remove". The guard is ensuring that the leader will always be removed from the pending nodes set, not "remote" from it.
```suggestion
// make sure it will always remove leader from pending nodes
```
##########
crates/fluss/src/client/table/log_fetch_buffer.rs:
##########
@@ -0,0 +1,367 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 crate::error::Result;
+use crate::metadata::TableBucket;
+use crate::record::{
+ LogRecordBatch, LogRecordIterator, LogRecordsBatches, ReadContext, ScanRecord,
+};
+use parking_lot::Mutex;
+use std::collections::{HashMap, VecDeque};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use tokio::sync::Notify;
+
+/// Represents a completed fetch that can be consumed
+pub trait CompletedFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn fetch_records(&mut self, max_records: usize) -> Result<Vec<ScanRecord>>;
+ fn is_consumed(&self) -> bool;
+ fn drain(&mut self);
+ fn size_in_bytes(&self) -> usize;
+ fn high_watermark(&self) -> i64;
+ fn is_initialized(&self) -> bool;
+ fn set_initialized(&mut self);
+ fn next_fetch_offset(&self) -> i64;
+}
+
+/// Represents a pending fetch that is waiting to be completed
+pub trait PendingFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn is_completed(&self) -> bool;
+ fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>>;
+}
+
+/// Thread-safe buffer for completed fetches
+pub struct LogFetchBuffer {
+ completed_fetches: Mutex<VecDeque<Box<dyn CompletedFetch>>>,
+ pending_fetches: Mutex<HashMap<TableBucket, VecDeque<Box<dyn PendingFetch>>>>,
+ next_in_line_fetch: Mutex<Option<Box<dyn CompletedFetch>>>,
+ not_empty_notify: Notify,
+ woken_up: Arc<AtomicBool>,
+}
+
+impl LogFetchBuffer {
+ pub fn new() -> Self {
+ Self {
+ completed_fetches: Mutex::new(VecDeque::new()),
+ pending_fetches: Mutex::new(HashMap::new()),
+ next_in_line_fetch: Mutex::new(None),
+ not_empty_notify: Notify::new(),
+ woken_up: Arc::new(AtomicBool::new(false)),
+ }
+ }
+
+ /// Check if the buffer is empty
+ pub fn is_empty(&self) -> bool {
+ self.completed_fetches.lock().is_empty()
+ }
+
+ /// Wait for the buffer to become non-empty, with timeout
+ /// Returns true if data became available, false if timeout
+ pub async fn await_not_empty(&self, timeout: Duration) -> bool {
+ let deadline = std::time::Instant::now() + timeout;
+
+ loop {
+ // Check if buffer is not empty
+ if !self.is_empty() {
+ return true;
+ }
+
+ // Check if woken up
+ if self.woken_up.swap(false, Ordering::Acquire) {
+ return true;
+ }
+
+ // Check if timeout
+ let now = std::time::Instant::now();
+ if now >= deadline {
+ return false;
+ }
+
+ // Wait for notification with remaining time
+ let remaining = deadline - now;
+ let notified = self.not_empty_notify.notified();
+ tokio::select! {
+ _ = tokio::time::sleep(remaining) => {
+ return false; // Timeout
+ }
+ _ = notified => {
+ // Got notification, check again
+ continue;
+ }
+ }
+ }
+ }
Review Comment:
The await_not_empty method uses a potentially infinite loop with a select! inside. If the notification system fails or if there's a logic bug, this could result in a tight loop. Additionally, the woken_up atomic is checked with swap but isn't coordinated with the notification, which could lead to missed wakeups if woken_up is set after the swap check but before the select! is reached.
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -723,24 +728,35 @@ struct Projection {
}
impl ReadContext {
- pub fn new(arrow_schema: SchemaRef) -> ReadContext {
+ pub fn new(arrow_schema: SchemaRef, is_from_remote: bool) -> ReadContext {
ReadContext {
target_schema: arrow_schema.clone(),
full_schema: arrow_schema,
projection: None,
+ is_from_remote,
}
}
pub fn with_projection_pushdown(
arrow_schema: SchemaRef,
projected_fields: Vec<usize>,
+ is_from_remote: bool,
) -> ReadContext {
let target_schema = Self::project_schema(arrow_schema.clone(), projected_fields.as_slice());
- let mut sorted_fields = projected_fields.clone();
- sorted_fields.sort_unstable();
+ let (need_do_reorder, sorted_fields) = {
+ // currently, for remote read, arrow log doesn't support projection pushdown,
+ // so, only need to do reordering when is not from remote
+ if !is_from_remote {
+ let mut sorted_fields = projected_fields.clone();
+ sorted_fields.sort_unstable();
+ (!sorted_fields.eq(&projected_fields), sorted_fields)
+ } else {
+ (false, vec![])
+ }
+ };
Review Comment:
When is_from_remote is true, sorted_fields is set to an empty vector on line 754. However, this empty vector is then used to create the Projection structure on lines 780-790 as ordered_fields. This seems inconsistent - if remote reads don't support projection pushdown, consider either returning None for projection or documenting why this structure with empty ordered_fields is necessary.
##########
crates/fluss/src/client/table/log_fetch_buffer.rs:
##########
@@ -0,0 +1,367 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 crate::error::Result;
+use crate::metadata::TableBucket;
+use crate::record::{
+ LogRecordBatch, LogRecordIterator, LogRecordsBatches, ReadContext, ScanRecord,
+};
+use parking_lot::Mutex;
+use std::collections::{HashMap, VecDeque};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use tokio::sync::Notify;
+
+/// Represents a completed fetch that can be consumed
+pub trait CompletedFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn fetch_records(&mut self, max_records: usize) -> Result<Vec<ScanRecord>>;
+ fn is_consumed(&self) -> bool;
+ fn drain(&mut self);
+ fn size_in_bytes(&self) -> usize;
+ fn high_watermark(&self) -> i64;
+ fn is_initialized(&self) -> bool;
+ fn set_initialized(&mut self);
+ fn next_fetch_offset(&self) -> i64;
+}
+
+/// Represents a pending fetch that is waiting to be completed
+pub trait PendingFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn is_completed(&self) -> bool;
+ fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>>;
+}
+
+/// Thread-safe buffer for completed fetches
+pub struct LogFetchBuffer {
+ completed_fetches: Mutex<VecDeque<Box<dyn CompletedFetch>>>,
+ pending_fetches: Mutex<HashMap<TableBucket, VecDeque<Box<dyn PendingFetch>>>>,
+ next_in_line_fetch: Mutex<Option<Box<dyn CompletedFetch>>>,
+ not_empty_notify: Notify,
+ woken_up: Arc<AtomicBool>,
+}
+
+impl LogFetchBuffer {
+ pub fn new() -> Self {
+ Self {
+ completed_fetches: Mutex::new(VecDeque::new()),
+ pending_fetches: Mutex::new(HashMap::new()),
+ next_in_line_fetch: Mutex::new(None),
+ not_empty_notify: Notify::new(),
+ woken_up: Arc::new(AtomicBool::new(false)),
+ }
+ }
+
+ /// Check if the buffer is empty
+ pub fn is_empty(&self) -> bool {
+ self.completed_fetches.lock().is_empty()
+ }
+
+ /// Wait for the buffer to become non-empty, with timeout
+ /// Returns true if data became available, false if timeout
+ pub async fn await_not_empty(&self, timeout: Duration) -> bool {
+ let deadline = std::time::Instant::now() + timeout;
+
+ loop {
+ // Check if buffer is not empty
+ if !self.is_empty() {
+ return true;
+ }
+
+ // Check if woken up
+ if self.woken_up.swap(false, Ordering::Acquire) {
+ return true;
+ }
+
+ // Check if timeout
+ let now = std::time::Instant::now();
+ if now >= deadline {
+ return false;
+ }
+
+ // Wait for notification with remaining time
+ let remaining = deadline - now;
+ let notified = self.not_empty_notify.notified();
+ tokio::select! {
+ _ = tokio::time::sleep(remaining) => {
+ return false; // Timeout
+ }
+ _ = notified => {
+ // Got notification, check again
+ continue;
+ }
+ }
+ }
+ }
+
+ #[allow(dead_code)]
+ /// Wake up any waiting threads
+ pub fn wakeup(&self) {
+ self.woken_up.store(true, Ordering::Release);
+ self.not_empty_notify.notify_waiters();
+ }
+
+ /// Add a pending fetch to the buffer
+ pub fn pend(&self, pending_fetch: Box<dyn PendingFetch>) {
+ let table_bucket = pending_fetch.table_bucket().clone();
+ self.pending_fetches
+ .lock()
+ .entry(table_bucket)
+ .or_default()
+ .push_back(pending_fetch);
+ }
+
+ /// Try to complete pending fetches in order, converting them to completed fetches
+ pub fn try_complete(&self, table_bucket: &TableBucket) {
+ let mut pending_map = self.pending_fetches.lock();
+ if let Some(pendings) = pending_map.get_mut(table_bucket) {
+ let mut has_completed = false;
+ while let Some(front) = pendings.front() {
+ if front.is_completed() {
+ let pending = pendings.pop_front().unwrap();
+ match pending.to_completed_fetch() {
+ Ok(completed) => {
+ self.completed_fetches.lock().push_back(completed);
+ // Signal that buffer is not empty
+ self.not_empty_notify.notify_waiters();
+ has_completed = true;
+ }
+ Err(e) => {
+ // todo: handle exception?
+ log::error!("Error when completing: {e}");
+ }
+ }
+ } else {
+ break;
+ }
+ }
+
+ if has_completed {
+ self.not_empty_notify.notify_waiters();
+ if pendings.is_empty() {
+ pending_map.remove(table_bucket);
+ }
+ }
+ }
Review Comment:
The try_complete method modifies the completed_fetches queue while holding the pending_fetches lock. If the completed_fetches lock is held by another thread, this could cause lock contention. Consider restructuring to minimize the time locks are held or use a single lock if these operations need to be atomic.
```suggestion
// Collect completed fetches while holding the pending_fetches lock,
// then push them to completed_fetches after releasing it to avoid
// holding both locks simultaneously.
let mut completed_to_push: Vec<Box<dyn CompletedFetch>> = Vec::new();
let mut has_completed = false;
{
let mut pending_map = self.pending_fetches.lock();
if let Some(pendings) = pending_map.get_mut(table_bucket) {
while let Some(front) = pendings.front() {
if front.is_completed() {
let pending = pendings.pop_front().unwrap();
match pending.to_completed_fetch() {
Ok(completed) => {
completed_to_push.push(completed);
has_completed = true;
}
Err(e) => {
// todo: handle exception?
log::error!("Error when completing: {e}");
}
}
} else {
break;
}
}
if has_completed && pendings.is_empty() {
pending_map.remove(table_bucket);
}
}
}
if !completed_to_push.is_empty() {
let mut completed_queue = self.completed_fetches.lock();
for completed in completed_to_push {
completed_queue.push_back(completed);
}
}
if has_completed {
// Signal that buffer is not empty
self.not_empty_notify.notify_waiters();
}
```
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -211,125 +260,292 @@ impl LogFetcher {
projected_fields: Option<Vec<usize>>,
) -> Result<Self> {
let full_arrow_schema = to_arrow_schema(table_info.get_row_type());
- let read_context = Self::create_read_context(full_arrow_schema, projected_fields.clone());
+ let read_context =
+ Self::create_read_context(full_arrow_schema.clone(), projected_fields.clone(), false);
+ let remote_read_context =
+ Self::create_read_context(full_arrow_schema, projected_fields.clone(), true);
let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
Ok(LogFetcher {
- table_path: table_info.table_path.clone(),
- conns,
- table_info,
- metadata,
+ conns: conns.clone(),
+ metadata: metadata.clone(),
log_scanner_status,
read_context,
- remote_log_downloader: RemoteLogDownloader::new(tmp_dir)?,
- credentials_cache: CredentialsCache::new(),
+ remote_read_context,
+ remote_log_downloader: Arc::new(RemoteLogDownloader::new(tmp_dir)?),
+ credentials_cache: Arc::new(CredentialsCache::new(conns.clone(), metadata.clone())),
+ log_fetch_buffer: Arc::new(LogFetchBuffer::new()),
+ nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())),
})
}
fn create_read_context(
full_arrow_schema: SchemaRef,
projected_fields: Option<Vec<usize>>,
+ is_from_remote: bool,
) -> ReadContext {
match projected_fields {
- None => ReadContext::new(full_arrow_schema),
- Some(fields) => ReadContext::with_projection_pushdown(full_arrow_schema, fields),
+ None => ReadContext::new(full_arrow_schema, is_from_remote),
+ Some(fields) => {
+ ReadContext::with_projection_pushdown(full_arrow_schema, fields, is_from_remote)
+ }
}
}
- async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ /// Send fetch requests asynchronously without waiting for responses
+ async fn send_fetches(&self) -> Result<()> {
+ // todo: check update metadata like fluss-java in case leader changes
let fetch_request = self.prepare_fetch_log_requests().await;
- let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+
for (leader, fetch_request) in fetch_request {
- let cluster = self.metadata.get_cluster();
- let server_node = cluster
- .get_tablet_server(leader)
- .expect("todo: handle leader not exist.");
- let con = self.conns.get_connection(server_node).await?;
-
- let fetch_response = con
- .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
- .await?;
-
- for pb_fetch_log_resp in fetch_response.tables_resp {
- let table_id = pb_fetch_log_resp.table_id;
- let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
-
- for fetch_log_for_bucket in fetch_log_for_buckets {
- let bucket: i32 = fetch_log_for_bucket.bucket_id;
- let table_bucket = TableBucket::new(table_id, bucket);
-
- // Check if this is a remote log fetch
- if let Some(ref remote_log_fetch_info) =
- fetch_log_for_bucket.remote_log_fetch_info
- {
- let remote_fs_props = self
- .credentials_cache
- .get_or_refresh(&self.conns, &self.metadata)
- .await?;
- self.remote_log_downloader
- .set_remote_fs_props(remote_fs_props);
- let remote_fetch_info = RemoteLogFetchInfo::from_proto(
- remote_log_fetch_info,
- table_bucket.clone(),
- )?;
-
- if let Some(fetch_offset) =
- self.log_scanner_status.get_bucket_offset(&table_bucket)
- {
- let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
- // Download and process remote log segments
- let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
- let mut current_fetch_offset = fetch_offset;
- // todo: make segment download in parallel
- for (i, segment) in
- remote_fetch_info.remote_log_segments.iter().enumerate()
- {
- if i > 0 {
- pos_in_log_segment = 0;
- current_fetch_offset = segment.start_offset;
- }
+ debug!("Adding pending request for node id {leader}");
+ // Check if we already have a pending request for this node
+ {
+ self.nodes_with_pending_fetch_requests.lock().insert(leader);
+ }
+
+ let cluster = self.metadata.get_cluster().clone();
+
+ let conns = Arc::clone(&self.conns);
+ let log_fetch_buffer = self.log_fetch_buffer.clone();
+ let log_scanner_status = self.log_scanner_status.clone();
+ let read_context = self.read_context.clone();
+ let remote_read_context = self.remote_read_context.clone();
+ let remote_log_downloader = Arc::clone(&self.remote_log_downloader);
+ let creds_cache = self.credentials_cache.clone();
+ let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone();
+
+ // Spawn async task to handle the fetch request
+ tokio::spawn(async move {
+ // make sure it will always remote leader from pending nodes
+ let _guard = scopeguard::guard((), |_| {
+ nodes_with_pending.lock().remove(&leader);
+ });
+
+ let server_node = cluster
+ .get_tablet_server(leader)
+ .expect("todo: handle leader not exist.");
+
+ let con = match conns.get_connection(server_node).await {
+ Ok(con) => con,
+ Err(e) => {
+ // todo: handle failed to get connection
+ warn!("Failed to get connection to destination node: {e:?}");
+ return;
+ }
+ };
+
+ let fetch_response = match con
+ .request(message::FetchLogRequest::new(fetch_request))
+ .await
+ {
+ Ok(resp) => resp,
+ Err(e) => {
+ // todo: handle fetch log from destination node
+ warn!("Failed to fetch log from destination node {server_node:?}: {e:?}");
+ return;
+ }
+ };
+
+ if let Err(e) = Self::handle_fetch_response(
+ fetch_response,
+ &log_fetch_buffer,
+ &log_scanner_status,
+ &read_context,
+ &remote_read_context,
+ &remote_log_downloader,
+ &creds_cache,
+ )
+ .await
+ {
+ // todo: handle fail to handle fetch response
+ error!("Fail to handle fetch response: {e:?}");
+ }
+ });
+ }
+
+ Ok(())
+ }
- let download_future =
- self.remote_log_downloader.request_remote_log(
- &remote_fetch_info.remote_log_tablet_dir,
- segment,
- )?;
- let pending_fetch = RemotePendingFetch::new(
- segment.clone(),
- download_future,
- pos_in_log_segment,
- current_fetch_offset,
- high_watermark,
- self.read_context.clone(),
- );
- let remote_records =
- pending_fetch.convert_to_completed_fetch().await?;
- // Update offset and merge results
- for (tb, records) in remote_records {
- if let Some(last_record) = records.last() {
- self.log_scanner_status
- .update_offset(&tb, last_record.offset() + 1);
- }
- result.entry(tb).or_default().extend(records);
+ /// Handle fetch response and add completed fetches to buffer
+ async fn handle_fetch_response(
+ fetch_response: crate::proto::FetchLogResponse,
+ log_fetch_buffer: &Arc<LogFetchBuffer>,
+ log_scanner_status: &Arc<LogScannerStatus>,
+ read_context: &ReadContext,
+ remote_read_context: &ReadContext,
+ remote_log_downloader: &Arc<RemoteLogDownloader>,
+ credentials_cache: &Arc<CredentialsCache>,
+ ) -> Result<()> {
+ for pb_fetch_log_resp in fetch_response.tables_resp {
+ let table_id = pb_fetch_log_resp.table_id;
+ let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
+
+ for fetch_log_for_bucket in fetch_log_for_buckets {
+ let bucket: i32 = fetch_log_for_bucket.bucket_id;
+ let table_bucket = TableBucket::new(table_id, bucket);
+
+ // todo: check fetch result code for per-bucket
+ let Some(fetch_offset) = log_scanner_status.get_bucket_offset(&table_bucket) else {
+ debug!(
+ "Ignoring fetch log response for bucket {table_bucket} because the bucket has been unsubscribed."
+ );
+ continue;
+ };
+
+ // Check if this is a remote log fetch
+ if let Some(ref remote_log_fetch_info) = fetch_log_for_bucket.remote_log_fetch_info
+ {
+ // set remote fs props
+ let remote_fs_props = credentials_cache.get_or_refresh().await?;
+ remote_log_downloader.set_remote_fs_props(remote_fs_props);
+
+ let remote_fetch_info =
+ RemoteLogFetchInfo::from_proto(remote_log_fetch_info, table_bucket.clone());
+
+ let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
+ Self::pending_remote_fetches(
+ remote_log_downloader.clone(),
+ log_fetch_buffer.clone(),
+ remote_read_context.clone(),
+ &table_bucket,
+ remote_fetch_info,
+ fetch_offset,
+ high_watermark,
+ );
+ } else if fetch_log_for_bucket.records.is_some() {
+ // Handle regular in-memory records - create completed fetch directly
+ let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
+ let records = fetch_log_for_bucket.records.unwrap_or(vec![]);
+ let size_in_bytes = records.len();
+ let log_record_batch = LogRecordsBatches::new(records);
+
+ match DefaultCompletedFetch::new(
+ table_bucket.clone(),
+ log_record_batch,
+ size_in_bytes,
+ read_context.clone(),
+ fetch_offset,
+ high_watermark,
+ ) {
+ Ok(completed_fetch) => {
+ log_fetch_buffer.add(Box::new(completed_fetch));
+ }
+ Err(e) => {
+ // todo: handle error
+ log::warn!("Failed to create completed fetch: {e:?}");
+ }
+ }
+ }
+ }
+ }
+ Ok(())
+ }
+
+ fn pending_remote_fetches(
+ remote_log_downloader: Arc<RemoteLogDownloader>,
+ log_fetch_buffer: Arc<LogFetchBuffer>,
+ read_context: ReadContext,
+ table_bucket: &TableBucket,
+ remote_fetch_info: RemoteLogFetchInfo,
+ fetch_offset: i64,
+ high_watermark: i64,
+ ) {
+ // Download and process remote log segments
+ let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
+ let mut current_fetch_offset = fetch_offset;
+ for (i, segment) in remote_fetch_info.remote_log_segments.iter().enumerate() {
+ if i > 0 {
+ pos_in_log_segment = 0;
+ current_fetch_offset = segment.start_offset;
+ }
+
+ // todo:
+ // 1: control the max threads to download remote segment
+ // 2: introduce priority queue to priority highest for earliest segment
+ let download_future = remote_log_downloader
+ .request_remote_log(&remote_fetch_info.remote_log_tablet_dir, segment);
+
+ // Register callback to be called when download completes
+ // (similar to Java's downloadFuture.onComplete)
+ // This must be done before creating RemotePendingFetch to avoid move issues
+ let table_bucket = table_bucket.clone();
+ let log_fetch_buffer_clone = log_fetch_buffer.clone();
+ download_future.on_complete(move || {
+ log_fetch_buffer_clone.try_complete(&table_bucket);
+ });
+
+ let pending_fetch = RemotePendingFetch::new(
+ segment.clone(),
+ download_future,
+ pos_in_log_segment,
+ current_fetch_offset,
+ high_watermark,
+ read_context.clone(),
+ );
+ // Add to pending fetches in buffer (similar to Java's logFetchBuffer.pend)
+ log_fetch_buffer.pend(Box::new(pending_fetch));
+ }
+ }
+
+ /// Collect completed fetches from buffer
+ /// Reference: LogFetchCollector.collectFetch in Java
+ fn collect_fetches(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ const MAX_POLL_RECORDS: usize = 500; // Default max poll records
Review Comment:
The constant MAX_POLL_RECORDS is set to 500 with a comment "Default max poll records", but there's no mechanism to configure this value. If users need to tune this for performance or memory reasons, they cannot. Consider making this configurable through the LogScanner or LogFetcher API.
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -211,125 +260,292 @@ impl LogFetcher {
projected_fields: Option<Vec<usize>>,
) -> Result<Self> {
let full_arrow_schema = to_arrow_schema(table_info.get_row_type());
- let read_context = Self::create_read_context(full_arrow_schema, projected_fields.clone());
+ let read_context =
+ Self::create_read_context(full_arrow_schema.clone(), projected_fields.clone(), false);
+ let remote_read_context =
+ Self::create_read_context(full_arrow_schema, projected_fields.clone(), true);
let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
Ok(LogFetcher {
- table_path: table_info.table_path.clone(),
- conns,
- table_info,
- metadata,
+ conns: conns.clone(),
+ metadata: metadata.clone(),
log_scanner_status,
read_context,
- remote_log_downloader: RemoteLogDownloader::new(tmp_dir)?,
- credentials_cache: CredentialsCache::new(),
+ remote_read_context,
+ remote_log_downloader: Arc::new(RemoteLogDownloader::new(tmp_dir)?),
+ credentials_cache: Arc::new(CredentialsCache::new(conns.clone(), metadata.clone())),
+ log_fetch_buffer: Arc::new(LogFetchBuffer::new()),
+ nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())),
})
}
fn create_read_context(
full_arrow_schema: SchemaRef,
projected_fields: Option<Vec<usize>>,
+ is_from_remote: bool,
) -> ReadContext {
match projected_fields {
- None => ReadContext::new(full_arrow_schema),
- Some(fields) => ReadContext::with_projection_pushdown(full_arrow_schema, fields),
+ None => ReadContext::new(full_arrow_schema, is_from_remote),
+ Some(fields) => {
+ ReadContext::with_projection_pushdown(full_arrow_schema, fields, is_from_remote)
+ }
}
}
- async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ /// Send fetch requests asynchronously without waiting for responses
+ async fn send_fetches(&self) -> Result<()> {
+ // todo: check update metadata like fluss-java in case leader changes
let fetch_request = self.prepare_fetch_log_requests().await;
- let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+
for (leader, fetch_request) in fetch_request {
- let cluster = self.metadata.get_cluster();
- let server_node = cluster
- .get_tablet_server(leader)
- .expect("todo: handle leader not exist.");
- let con = self.conns.get_connection(server_node).await?;
-
- let fetch_response = con
- .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
- .await?;
-
- for pb_fetch_log_resp in fetch_response.tables_resp {
- let table_id = pb_fetch_log_resp.table_id;
- let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
-
- for fetch_log_for_bucket in fetch_log_for_buckets {
- let bucket: i32 = fetch_log_for_bucket.bucket_id;
- let table_bucket = TableBucket::new(table_id, bucket);
-
- // Check if this is a remote log fetch
- if let Some(ref remote_log_fetch_info) =
- fetch_log_for_bucket.remote_log_fetch_info
- {
- let remote_fs_props = self
- .credentials_cache
- .get_or_refresh(&self.conns, &self.metadata)
- .await?;
- self.remote_log_downloader
- .set_remote_fs_props(remote_fs_props);
- let remote_fetch_info = RemoteLogFetchInfo::from_proto(
- remote_log_fetch_info,
- table_bucket.clone(),
- )?;
-
- if let Some(fetch_offset) =
- self.log_scanner_status.get_bucket_offset(&table_bucket)
- {
- let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
- // Download and process remote log segments
- let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
- let mut current_fetch_offset = fetch_offset;
- // todo: make segment download in parallel
- for (i, segment) in
- remote_fetch_info.remote_log_segments.iter().enumerate()
- {
- if i > 0 {
- pos_in_log_segment = 0;
- current_fetch_offset = segment.start_offset;
- }
+ debug!("Adding pending request for node id {leader}");
+ // Check if we already have a pending request for this node
+ {
+ self.nodes_with_pending_fetch_requests.lock().insert(leader);
+ }
+
+ let cluster = self.metadata.get_cluster().clone();
+
+ let conns = Arc::clone(&self.conns);
+ let log_fetch_buffer = self.log_fetch_buffer.clone();
+ let log_scanner_status = self.log_scanner_status.clone();
+ let read_context = self.read_context.clone();
+ let remote_read_context = self.remote_read_context.clone();
+ let remote_log_downloader = Arc::clone(&self.remote_log_downloader);
+ let creds_cache = self.credentials_cache.clone();
+ let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone();
+
+ // Spawn async task to handle the fetch request
+ tokio::spawn(async move {
+ // make sure it will always remote leader from pending nodes
+ let _guard = scopeguard::guard((), |_| {
+ nodes_with_pending.lock().remove(&leader);
+ });
+
+ let server_node = cluster
+ .get_tablet_server(leader)
+ .expect("todo: handle leader not exist.");
+
+ let con = match conns.get_connection(server_node).await {
+ Ok(con) => con,
+ Err(e) => {
+ // todo: handle failed to get connection
+ warn!("Failed to get connection to destination node: {e:?}");
+ return;
+ }
+ };
+
+ let fetch_response = match con
+ .request(message::FetchLogRequest::new(fetch_request))
+ .await
+ {
+ Ok(resp) => resp,
+ Err(e) => {
+ // todo: handle fetch log from destination node
+ warn!("Failed to fetch log from destination node {server_node:?}: {e:?}");
+ return;
+ }
+ };
+
+ if let Err(e) = Self::handle_fetch_response(
+ fetch_response,
+ &log_fetch_buffer,
+ &log_scanner_status,
+ &read_context,
+ &remote_read_context,
+ &remote_log_downloader,
+ &creds_cache,
+ )
+ .await
+ {
+ // todo: handle fail to handle fetch response
+ error!("Fail to handle fetch response: {e:?}");
+ }
+ });
Review Comment:
There's an inconsistency in error handling for the spawned async task. The task logs warnings or errors but doesn't propagate failures back to the caller. If fetch requests consistently fail (e.g., due to network issues), the poll() method could timeout without the user knowing why. Consider adding a mechanism to track and report persistent failures, such as maintaining error counters or surfacing errors through a dedicated channel.
##########
crates/fluss/src/client/table/log_fetch_buffer.rs:
##########
@@ -0,0 +1,367 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 crate::error::Result;
+use crate::metadata::TableBucket;
+use crate::record::{
+ LogRecordBatch, LogRecordIterator, LogRecordsBatches, ReadContext, ScanRecord,
+};
+use parking_lot::Mutex;
+use std::collections::{HashMap, VecDeque};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use tokio::sync::Notify;
+
+/// Represents a completed fetch that can be consumed
+pub trait CompletedFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn fetch_records(&mut self, max_records: usize) -> Result<Vec<ScanRecord>>;
+ fn is_consumed(&self) -> bool;
+ fn drain(&mut self);
+ fn size_in_bytes(&self) -> usize;
+ fn high_watermark(&self) -> i64;
+ fn is_initialized(&self) -> bool;
+ fn set_initialized(&mut self);
+ fn next_fetch_offset(&self) -> i64;
+}
+
+/// Represents a pending fetch that is waiting to be completed
+pub trait PendingFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn is_completed(&self) -> bool;
+ fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>>;
+}
+
+/// Thread-safe buffer for completed fetches
+pub struct LogFetchBuffer {
+ completed_fetches: Mutex<VecDeque<Box<dyn CompletedFetch>>>,
+ pending_fetches: Mutex<HashMap<TableBucket, VecDeque<Box<dyn PendingFetch>>>>,
+ next_in_line_fetch: Mutex<Option<Box<dyn CompletedFetch>>>,
+ not_empty_notify: Notify,
+ woken_up: Arc<AtomicBool>,
+}
+
+impl LogFetchBuffer {
+ pub fn new() -> Self {
+ Self {
+ completed_fetches: Mutex::new(VecDeque::new()),
+ pending_fetches: Mutex::new(HashMap::new()),
+ next_in_line_fetch: Mutex::new(None),
+ not_empty_notify: Notify::new(),
+ woken_up: Arc::new(AtomicBool::new(false)),
+ }
+ }
+
+ /// Check if the buffer is empty
+ pub fn is_empty(&self) -> bool {
+ self.completed_fetches.lock().is_empty()
+ }
+
+ /// Wait for the buffer to become non-empty, with timeout
+ /// Returns true if data became available, false if timeout
+ pub async fn await_not_empty(&self, timeout: Duration) -> bool {
+ let deadline = std::time::Instant::now() + timeout;
+
+ loop {
+ // Check if buffer is not empty
+ if !self.is_empty() {
+ return true;
+ }
+
+ // Check if woken up
+ if self.woken_up.swap(false, Ordering::Acquire) {
+ return true;
+ }
+
+ // Check if timeout
+ let now = std::time::Instant::now();
+ if now >= deadline {
+ return false;
+ }
+
+ // Wait for notification with remaining time
+ let remaining = deadline - now;
+ let notified = self.not_empty_notify.notified();
+ tokio::select! {
+ _ = tokio::time::sleep(remaining) => {
+ return false; // Timeout
+ }
+ _ = notified => {
+ // Got notification, check again
+ continue;
+ }
+ }
+ }
+ }
+
+ #[allow(dead_code)]
+ /// Wake up any waiting threads
+ pub fn wakeup(&self) {
+ self.woken_up.store(true, Ordering::Release);
+ self.not_empty_notify.notify_waiters();
+ }
+
+ /// Add a pending fetch to the buffer
+ pub fn pend(&self, pending_fetch: Box<dyn PendingFetch>) {
+ let table_bucket = pending_fetch.table_bucket().clone();
+ self.pending_fetches
+ .lock()
+ .entry(table_bucket)
+ .or_default()
+ .push_back(pending_fetch);
+ }
+
+ /// Try to complete pending fetches in order, converting them to completed fetches
+ pub fn try_complete(&self, table_bucket: &TableBucket) {
+ let mut pending_map = self.pending_fetches.lock();
+ if let Some(pendings) = pending_map.get_mut(table_bucket) {
+ let mut has_completed = false;
+ while let Some(front) = pendings.front() {
+ if front.is_completed() {
+ let pending = pendings.pop_front().unwrap();
+ match pending.to_completed_fetch() {
+ Ok(completed) => {
+ self.completed_fetches.lock().push_back(completed);
+ // Signal that buffer is not empty
+ self.not_empty_notify.notify_waiters();
+ has_completed = true;
+ }
+ Err(e) => {
+ // todo: handle exception?
+ log::error!("Error when completing: {e}");
+ }
+ }
+ } else {
+ break;
+ }
+ }
+
+ if has_completed {
+ self.not_empty_notify.notify_waiters();
+ if pendings.is_empty() {
+ pending_map.remove(table_bucket);
+ }
+ }
+ }
+ }
+
+ /// Add a completed fetch to the buffer
+ pub fn add(&self, completed_fetch: Box<dyn CompletedFetch>) {
+ let table_bucket = completed_fetch.table_bucket();
+ let mut pending_map = self.pending_fetches.lock();
+
+ if let Some(pendings) = pending_map.get_mut(table_bucket) {
+ if pendings.is_empty() {
+ self.completed_fetches.lock().push_back(completed_fetch);
+ self.not_empty_notify.notify_waiters();
+ } else {
+ pendings.push_back(Box::new(CompletedPendingFetch::new(completed_fetch)));
+ }
+ } else {
+ // If there's no pending fetch for this table_bucket,
+ // directly add to completed_fetches
+ self.completed_fetches.lock().push_back(completed_fetch);
+ self.not_empty_notify.notify_waiters();
+ }
+ }
Review Comment:
The add method has duplicate notification calls. Line 171 and 179 both call not_empty_notify.notify_waiters() in different branches, but this is redundant. Consider consolidating the notification logic or documenting why both are necessary.
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -171,8 +174,43 @@ impl LogScanner {
})
}
- pub async fn poll(&self, _timeout: Duration) -> Result<ScanRecords> {
- Ok(ScanRecords::new(self.poll_for_fetches().await?))
+ pub async fn poll(&self, timeout: Duration) -> Result<ScanRecords> {
+ let start = std::time::Instant::now();
+ let deadline = start + timeout;
+
+ loop {
+ // Try to collect fetches
+ let fetch_result = self.poll_for_fetches().await?;
+
+ if !fetch_result.is_empty() {
+ // We have data, send next round of fetches and return
+ // This enables pipelining while user processes the data
+ self.log_fetcher.send_fetches().await?;
+ return Ok(ScanRecords::new(fetch_result));
+ }
+
+ // No data available, check if we should wait
+ let now = std::time::Instant::now();
+ if now >= deadline {
+ // Timeout reached, return empty result
+ return Ok(ScanRecords::new(HashMap::new()));
+ }
+
+ // Wait for buffer to become non-empty with remaining time
+ let remaining = deadline - now;
+ let has_data = self
+ .log_fetcher
+ .log_fetch_buffer
+ .await_not_empty(remaining)
+ .await;
+
+ if !has_data {
+ // Timeout while waiting
+ return Ok(ScanRecords::new(HashMap::new()));
+ }
+
+ // Buffer became non-empty, try again
+ }
Review Comment:
The poll loop could busy-wait when the buffer is empty but the timeout hasn't been reached. While await_not_empty() waits with the remaining timeout, if it returns true but the subsequent poll_for_fetches() finds no data (race condition), the loop continues immediately without any backoff. Consider adding a small sleep or exponential backoff between retries to avoid excessive CPU usage in edge cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637507210
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -350,25 +643,44 @@ impl LogFetcher {
let offset = match self.log_scanner_status.get_bucket_offset(&bucket) {
Some(offset) => offset,
None => {
- // todo: debug
+ debug!(
+ "Skipping fetch request for bucket {bucket} because the bucket has been unsubscribed."
+ );
continue;
}
};
- if let Some(leader) = self.get_table_bucket_leader(&bucket) {
- let fetch_log_req_for_bucket = PbFetchLogReqForBucket {
- partition_id: None,
- bucket_id: bucket.bucket_id(),
- fetch_offset: offset,
- // 1M
- max_fetch_bytes: 1024 * 1024,
- };
-
- fetch_log_req_for_buckets
- .entry(leader)
- .or_insert_with(Vec::new)
- .push(fetch_log_req_for_bucket);
- ready_for_fetch_count += 1;
+ match self.get_table_bucket_leader(&bucket) {
+ None => {
+ log::trace!(
+ "Skipping fetch request for bucket {bucket} because leader is not available."
+ )
+ }
+ Some(leader) => {
+ if self
+ .nodes_with_pending_fetch_requests
+ .lock()
+ .contains(&leader)
+ {
+ log::trace!(
+ "Skipping fetch request for bucket {bucket} because previous request to server {leader} has not been processed."
+ )
+ } else {
+ let fetch_log_req_for_bucket = PbFetchLogReqForBucket {
+ partition_id: None,
+ bucket_id: bucket.bucket_id(),
+ fetch_offset: offset,
+ // 1M
+ max_fetch_bytes: 1024 * 1024,
+ };
+
+ fetch_log_req_for_buckets
+ .entry(leader)
+ .or_insert_with(Vec::new)
+ .push(fetch_log_req_for_bucket);
+ ready_for_fetch_count += 1;
+ }
+ }
Review Comment:
I agree. But now, let's follow java side logic
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637521822
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -211,125 +260,292 @@ impl LogFetcher {
projected_fields: Option<Vec<usize>>,
) -> Result<Self> {
let full_arrow_schema = to_arrow_schema(table_info.get_row_type());
- let read_context = Self::create_read_context(full_arrow_schema, projected_fields.clone());
+ let read_context =
+ Self::create_read_context(full_arrow_schema.clone(), projected_fields.clone(), false);
+ let remote_read_context =
+ Self::create_read_context(full_arrow_schema, projected_fields.clone(), true);
let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
Ok(LogFetcher {
- table_path: table_info.table_path.clone(),
- conns,
- table_info,
- metadata,
+ conns: conns.clone(),
+ metadata: metadata.clone(),
log_scanner_status,
read_context,
- remote_log_downloader: RemoteLogDownloader::new(tmp_dir)?,
- credentials_cache: CredentialsCache::new(),
+ remote_read_context,
+ remote_log_downloader: Arc::new(RemoteLogDownloader::new(tmp_dir)?),
+ credentials_cache: Arc::new(CredentialsCache::new(conns.clone(), metadata.clone())),
+ log_fetch_buffer: Arc::new(LogFetchBuffer::new()),
+ nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())),
})
}
fn create_read_context(
full_arrow_schema: SchemaRef,
projected_fields: Option<Vec<usize>>,
+ is_from_remote: bool,
) -> ReadContext {
match projected_fields {
- None => ReadContext::new(full_arrow_schema),
- Some(fields) => ReadContext::with_projection_pushdown(full_arrow_schema, fields),
+ None => ReadContext::new(full_arrow_schema, is_from_remote),
+ Some(fields) => {
+ ReadContext::with_projection_pushdown(full_arrow_schema, fields, is_from_remote)
+ }
}
}
- async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ /// Send fetch requests asynchronously without waiting for responses
+ async fn send_fetches(&self) -> Result<()> {
+ // todo: check update metadata like fluss-java in case leader changes
let fetch_request = self.prepare_fetch_log_requests().await;
- let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+
for (leader, fetch_request) in fetch_request {
- let cluster = self.metadata.get_cluster();
- let server_node = cluster
- .get_tablet_server(leader)
- .expect("todo: handle leader not exist.");
- let con = self.conns.get_connection(server_node).await?;
-
- let fetch_response = con
- .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
- .await?;
-
- for pb_fetch_log_resp in fetch_response.tables_resp {
- let table_id = pb_fetch_log_resp.table_id;
- let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
-
- for fetch_log_for_bucket in fetch_log_for_buckets {
- let bucket: i32 = fetch_log_for_bucket.bucket_id;
- let table_bucket = TableBucket::new(table_id, bucket);
-
- // Check if this is a remote log fetch
- if let Some(ref remote_log_fetch_info) =
- fetch_log_for_bucket.remote_log_fetch_info
- {
- let remote_fs_props = self
- .credentials_cache
- .get_or_refresh(&self.conns, &self.metadata)
- .await?;
- self.remote_log_downloader
- .set_remote_fs_props(remote_fs_props);
- let remote_fetch_info = RemoteLogFetchInfo::from_proto(
- remote_log_fetch_info,
- table_bucket.clone(),
- )?;
-
- if let Some(fetch_offset) =
- self.log_scanner_status.get_bucket_offset(&table_bucket)
- {
- let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
- // Download and process remote log segments
- let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
- let mut current_fetch_offset = fetch_offset;
- // todo: make segment download in parallel
- for (i, segment) in
- remote_fetch_info.remote_log_segments.iter().enumerate()
- {
- if i > 0 {
- pos_in_log_segment = 0;
- current_fetch_offset = segment.start_offset;
- }
+ debug!("Adding pending request for node id {leader}");
+ // Check if we already have a pending request for this node
+ {
+ self.nodes_with_pending_fetch_requests.lock().insert(leader);
+ }
+
+ let cluster = self.metadata.get_cluster().clone();
+
+ let conns = Arc::clone(&self.conns);
+ let log_fetch_buffer = self.log_fetch_buffer.clone();
+ let log_scanner_status = self.log_scanner_status.clone();
+ let read_context = self.read_context.clone();
+ let remote_read_context = self.remote_read_context.clone();
+ let remote_log_downloader = Arc::clone(&self.remote_log_downloader);
+ let creds_cache = self.credentials_cache.clone();
+ let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone();
+
+ // Spawn async task to handle the fetch request
+ tokio::spawn(async move {
Review Comment:
If LogFetcher, the task will still run, but after task finishes, it wil be removed. I think it's fine now since java side also have the same behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637517196
##########
crates/fluss/src/client/credentials.rs:
##########
@@ -113,17 +113,13 @@ impl CredentialsCache {
}
}
- self.refresh_from_server(rpc_client, metadata).await
+ self.refresh_from_server().await
}
- async fn refresh_from_server(
- &self,
- rpc_client: &Arc<RpcClient>,
- metadata: &Arc<Metadata>,
- ) -> Result<HashMap<String, String>> {
- let cluster = metadata.get_cluster();
+ async fn refresh_from_server(&self) -> Result<HashMap<String, String>> {
Review Comment:
We still need it be `async` since in this method we will have async code block such as send request
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#issuecomment-3678512025
> @zhaohaidao @leekeiabstraction Thanks for your review. Comments addressed now. PTAL
@luoxia
Thanks very much, LGTM for me. Expecting read performance to take off🚀
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637121366
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -211,125 +260,292 @@ impl LogFetcher {
projected_fields: Option<Vec<usize>>,
) -> Result<Self> {
let full_arrow_schema = to_arrow_schema(table_info.get_row_type());
- let read_context = Self::create_read_context(full_arrow_schema, projected_fields.clone());
+ let read_context =
+ Self::create_read_context(full_arrow_schema.clone(), projected_fields.clone(), false);
+ let remote_read_context =
+ Self::create_read_context(full_arrow_schema, projected_fields.clone(), true);
let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
Ok(LogFetcher {
- table_path: table_info.table_path.clone(),
- conns,
- table_info,
- metadata,
+ conns: conns.clone(),
+ metadata: metadata.clone(),
log_scanner_status,
read_context,
- remote_log_downloader: RemoteLogDownloader::new(tmp_dir)?,
- credentials_cache: CredentialsCache::new(),
+ remote_read_context,
+ remote_log_downloader: Arc::new(RemoteLogDownloader::new(tmp_dir)?),
+ credentials_cache: Arc::new(CredentialsCache::new(conns.clone(), metadata.clone())),
+ log_fetch_buffer: Arc::new(LogFetchBuffer::new()),
+ nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())),
})
}
fn create_read_context(
full_arrow_schema: SchemaRef,
projected_fields: Option<Vec<usize>>,
+ is_from_remote: bool,
) -> ReadContext {
match projected_fields {
- None => ReadContext::new(full_arrow_schema),
- Some(fields) => ReadContext::with_projection_pushdown(full_arrow_schema, fields),
+ None => ReadContext::new(full_arrow_schema, is_from_remote),
+ Some(fields) => {
+ ReadContext::with_projection_pushdown(full_arrow_schema, fields, is_from_remote)
+ }
}
}
- async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ /// Send fetch requests asynchronously without waiting for responses
+ async fn send_fetches(&self) -> Result<()> {
+ // todo: check update metadata like fluss-java in case leader changes
let fetch_request = self.prepare_fetch_log_requests().await;
- let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+
for (leader, fetch_request) in fetch_request {
- let cluster = self.metadata.get_cluster();
- let server_node = cluster
- .get_tablet_server(leader)
- .expect("todo: handle leader not exist.");
- let con = self.conns.get_connection(server_node).await?;
-
- let fetch_response = con
- .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
- .await?;
-
- for pb_fetch_log_resp in fetch_response.tables_resp {
- let table_id = pb_fetch_log_resp.table_id;
- let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
-
- for fetch_log_for_bucket in fetch_log_for_buckets {
- let bucket: i32 = fetch_log_for_bucket.bucket_id;
- let table_bucket = TableBucket::new(table_id, bucket);
-
- // Check if this is a remote log fetch
- if let Some(ref remote_log_fetch_info) =
- fetch_log_for_bucket.remote_log_fetch_info
- {
- let remote_fs_props = self
- .credentials_cache
- .get_or_refresh(&self.conns, &self.metadata)
- .await?;
- self.remote_log_downloader
- .set_remote_fs_props(remote_fs_props);
- let remote_fetch_info = RemoteLogFetchInfo::from_proto(
- remote_log_fetch_info,
- table_bucket.clone(),
- )?;
-
- if let Some(fetch_offset) =
- self.log_scanner_status.get_bucket_offset(&table_bucket)
- {
- let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
- // Download and process remote log segments
- let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
- let mut current_fetch_offset = fetch_offset;
- // todo: make segment download in parallel
- for (i, segment) in
- remote_fetch_info.remote_log_segments.iter().enumerate()
- {
- if i > 0 {
- pos_in_log_segment = 0;
- current_fetch_offset = segment.start_offset;
- }
+ debug!("Adding pending request for node id {leader}");
+ // Check if we already have a pending request for this node
+ {
+ self.nodes_with_pending_fetch_requests.lock().insert(leader);
+ }
+
+ let cluster = self.metadata.get_cluster().clone();
+
+ let conns = Arc::clone(&self.conns);
+ let log_fetch_buffer = self.log_fetch_buffer.clone();
+ let log_scanner_status = self.log_scanner_status.clone();
+ let read_context = self.read_context.clone();
+ let remote_read_context = self.remote_read_context.clone();
+ let remote_log_downloader = Arc::clone(&self.remote_log_downloader);
+ let creds_cache = self.credentials_cache.clone();
+ let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone();
+
+ // Spawn async task to handle the fetch request
+ tokio::spawn(async move {
+ // make sure it will always remote leader from pending nodes
Review Comment:
typo? s/remote/remove
##########
crates/fluss/src/client/table/remote_log.rs:
##########
@@ -286,32 +351,54 @@ impl RemotePendingFetch {
read_context,
}
}
+}
+
+impl PendingFetch for RemotePendingFetch {
+ fn table_bucket(&self) -> &TableBucket {
+ &self.segment.table_bucket
+ }
- /// Convert to completed fetch by reading the downloaded file
- pub async fn convert_to_completed_fetch(
- mut self,
- ) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
- let file_path = self.download_future.get_file_path().await?;
- let file_data = tokio::fs::read(&file_path).await?;
+ fn is_completed(&self) -> bool {
+ self.download_future.is_done()
+ }
+
+ fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>> {
+ // Get the file path (this should only be called when is_completed() returns true)
+ let file_path = self.download_future.get_file_path()?;
+ // Read the file data synchronously (we're in a sync context)
+ // Note: This is a limitation - we need to use blocking I/O here
+ let mut file_data = std::fs::read(&file_path).map_err(|e| Error::IoUnexpectedError {
+ message: format!("Failed to read downloaded file: {file_path:?}."),
+ source: e,
+ })?;
// Slice the data if needed
let data = if self.pos_in_log_segment > 0 {
- &file_data[self.pos_in_log_segment as usize..]
+ file_data.split_off(self.pos_in_log_segment as usize)
} else {
- &file_data
+ file_data
};
- // delete the downloaded local file to free disk
- delete_file(file_path).await;
+ let size_in_bytes = data.len();
- // Parse log records (remote log contains full data, need client-side projection)
- let mut fetch_records = vec![];
- for log_record in &mut LogRecordsBatchs::new(data) {
- fetch_records.extend(log_record.records_for_remote_log(&self.read_context)?);
- }
+ let log_record_batch = LogRecordsBatches::new(data);
+
+ // Create DefaultCompletedFetch from the data
+ let completed_fetch = DefaultCompletedFetch::new(
+ self.segment.table_bucket,
+ log_record_batch,
+ size_in_bytes,
+ self.read_context,
+ self.fetch_offset,
+ self.high_watermark,
+ )?;
+
+ // Delete the downloaded local file to free disk (async, but we'll do it in background)
+ let file_path_clone = file_path.clone();
+ tokio::spawn(async move {
+ let _ = delete_file(file_path_clone).await;
Review Comment:
If deletion fails, will there be any remnants? It's recommended to at least log a warning at the log level if deletion fails.
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -211,125 +260,292 @@ impl LogFetcher {
projected_fields: Option<Vec<usize>>,
) -> Result<Self> {
let full_arrow_schema = to_arrow_schema(table_info.get_row_type());
- let read_context = Self::create_read_context(full_arrow_schema, projected_fields.clone());
+ let read_context =
+ Self::create_read_context(full_arrow_schema.clone(), projected_fields.clone(), false);
+ let remote_read_context =
+ Self::create_read_context(full_arrow_schema, projected_fields.clone(), true);
let tmp_dir = TempDir::with_prefix("fluss-remote-logs")?;
Ok(LogFetcher {
- table_path: table_info.table_path.clone(),
- conns,
- table_info,
- metadata,
+ conns: conns.clone(),
+ metadata: metadata.clone(),
log_scanner_status,
read_context,
- remote_log_downloader: RemoteLogDownloader::new(tmp_dir)?,
- credentials_cache: CredentialsCache::new(),
+ remote_read_context,
+ remote_log_downloader: Arc::new(RemoteLogDownloader::new(tmp_dir)?),
+ credentials_cache: Arc::new(CredentialsCache::new(conns.clone(), metadata.clone())),
+ log_fetch_buffer: Arc::new(LogFetchBuffer::new()),
+ nodes_with_pending_fetch_requests: Arc::new(Mutex::new(HashSet::new())),
})
}
fn create_read_context(
full_arrow_schema: SchemaRef,
projected_fields: Option<Vec<usize>>,
+ is_from_remote: bool,
) -> ReadContext {
match projected_fields {
- None => ReadContext::new(full_arrow_schema),
- Some(fields) => ReadContext::with_projection_pushdown(full_arrow_schema, fields),
+ None => ReadContext::new(full_arrow_schema, is_from_remote),
+ Some(fields) => {
+ ReadContext::with_projection_pushdown(full_arrow_schema, fields, is_from_remote)
+ }
}
}
- async fn send_fetches_and_collect(&self) -> Result<HashMap<TableBucket, Vec<ScanRecord>>> {
+ /// Send fetch requests asynchronously without waiting for responses
+ async fn send_fetches(&self) -> Result<()> {
+ // todo: check update metadata like fluss-java in case leader changes
let fetch_request = self.prepare_fetch_log_requests().await;
- let mut result: HashMap<TableBucket, Vec<ScanRecord>> = HashMap::new();
+
for (leader, fetch_request) in fetch_request {
- let cluster = self.metadata.get_cluster();
- let server_node = cluster
- .get_tablet_server(leader)
- .expect("todo: handle leader not exist.");
- let con = self.conns.get_connection(server_node).await?;
-
- let fetch_response = con
- .request(crate::rpc::message::FetchLogRequest::new(fetch_request))
- .await?;
-
- for pb_fetch_log_resp in fetch_response.tables_resp {
- let table_id = pb_fetch_log_resp.table_id;
- let fetch_log_for_buckets = pb_fetch_log_resp.buckets_resp;
-
- for fetch_log_for_bucket in fetch_log_for_buckets {
- let bucket: i32 = fetch_log_for_bucket.bucket_id;
- let table_bucket = TableBucket::new(table_id, bucket);
-
- // Check if this is a remote log fetch
- if let Some(ref remote_log_fetch_info) =
- fetch_log_for_bucket.remote_log_fetch_info
- {
- let remote_fs_props = self
- .credentials_cache
- .get_or_refresh(&self.conns, &self.metadata)
- .await?;
- self.remote_log_downloader
- .set_remote_fs_props(remote_fs_props);
- let remote_fetch_info = RemoteLogFetchInfo::from_proto(
- remote_log_fetch_info,
- table_bucket.clone(),
- )?;
-
- if let Some(fetch_offset) =
- self.log_scanner_status.get_bucket_offset(&table_bucket)
- {
- let high_watermark = fetch_log_for_bucket.high_watermark.unwrap_or(-1);
- // Download and process remote log segments
- let mut pos_in_log_segment = remote_fetch_info.first_start_pos;
- let mut current_fetch_offset = fetch_offset;
- // todo: make segment download in parallel
- for (i, segment) in
- remote_fetch_info.remote_log_segments.iter().enumerate()
- {
- if i > 0 {
- pos_in_log_segment = 0;
- current_fetch_offset = segment.start_offset;
- }
+ debug!("Adding pending request for node id {leader}");
+ // Check if we already have a pending request for this node
+ {
+ self.nodes_with_pending_fetch_requests.lock().insert(leader);
+ }
+
+ let cluster = self.metadata.get_cluster().clone();
+
+ let conns = Arc::clone(&self.conns);
+ let log_fetch_buffer = self.log_fetch_buffer.clone();
+ let log_scanner_status = self.log_scanner_status.clone();
+ let read_context = self.read_context.clone();
+ let remote_read_context = self.remote_read_context.clone();
+ let remote_log_downloader = Arc::clone(&self.remote_log_downloader);
+ let creds_cache = self.credentials_cache.clone();
+ let nodes_with_pending = self.nodes_with_pending_fetch_requests.clone();
+
+ // Spawn async task to handle the fetch request
+ tokio::spawn(async move {
Review Comment:
If LogFetcher is stopped, is there a task leak issue? I don't see where these asynchronous tasks are maintained.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637123588
##########
crates/fluss/src/client/credentials.rs:
##########
@@ -113,17 +113,13 @@ impl CredentialsCache {
}
}
- self.refresh_from_server(rpc_client, metadata).await
+ self.refresh_from_server().await
}
- async fn refresh_from_server(
- &self,
- rpc_client: &Arc<RpcClient>,
- metadata: &Arc<Metadata>,
- ) -> Result<HashMap<String, String>> {
- let cluster = metadata.get_cluster();
+ async fn refresh_from_server(&self) -> Result<HashMap<String, String>> {
Review Comment:
Does `refresh_from_server` itself need to be `async` as the only caller `get_or_refresh` is already `async`?
##########
crates/fluss/src/client/table/log_fetch_buffer.rs:
##########
@@ -0,0 +1,367 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 crate::error::Result;
+use crate::metadata::TableBucket;
+use crate::record::{
+ LogRecordBatch, LogRecordIterator, LogRecordsBatches, ReadContext, ScanRecord,
+};
+use parking_lot::Mutex;
+use std::collections::{HashMap, VecDeque};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use tokio::sync::Notify;
+
+/// Represents a completed fetch that can be consumed
+pub trait CompletedFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn fetch_records(&mut self, max_records: usize) -> Result<Vec<ScanRecord>>;
+ fn is_consumed(&self) -> bool;
+ fn drain(&mut self);
+ fn size_in_bytes(&self) -> usize;
+ fn high_watermark(&self) -> i64;
+ fn is_initialized(&self) -> bool;
+ fn set_initialized(&mut self);
+ fn next_fetch_offset(&self) -> i64;
+}
+
+/// Represents a pending fetch that is waiting to be completed
+pub trait PendingFetch: Send + Sync {
+ fn table_bucket(&self) -> &TableBucket;
+ fn is_completed(&self) -> bool;
+ fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>>;
+}
+
+/// Thread-safe buffer for completed fetches
+pub struct LogFetchBuffer {
+ completed_fetches: Mutex<VecDeque<Box<dyn CompletedFetch>>>,
+ pending_fetches: Mutex<HashMap<TableBucket, VecDeque<Box<dyn PendingFetch>>>>,
+ next_in_line_fetch: Mutex<Option<Box<dyn CompletedFetch>>>,
+ not_empty_notify: Notify,
+ woken_up: Arc<AtomicBool>,
+}
+
+impl LogFetchBuffer {
+ pub fn new() -> Self {
+ Self {
+ completed_fetches: Mutex::new(VecDeque::new()),
+ pending_fetches: Mutex::new(HashMap::new()),
+ next_in_line_fetch: Mutex::new(None),
+ not_empty_notify: Notify::new(),
+ woken_up: Arc::new(AtomicBool::new(false)),
+ }
+ }
+
+ /// Check if the buffer is empty
+ pub fn is_empty(&self) -> bool {
+ self.completed_fetches.lock().is_empty()
+ }
+
+ /// Wait for the buffer to become non-empty, with timeout
+ /// Returns true if data became available, false if timeout
+ pub async fn await_not_empty(&self, timeout: Duration) -> bool {
+ let deadline = std::time::Instant::now() + timeout;
+
+ loop {
+ // Check if buffer is not empty
+ if !self.is_empty() {
Review Comment:
nit: Instead of using `self.is_empty()` directly which has an indefinite wait on lock, should we consider `self.completed_fetches.try_lock_until(deadline)` here to better honour the timeout (and potentially simplify lines 91 to 95)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#discussion_r2637513921
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -171,8 +174,43 @@ impl LogScanner {
})
}
- pub async fn poll(&self, _timeout: Duration) -> Result<ScanRecords> {
- Ok(ScanRecords::new(self.poll_for_fetches().await?))
+ pub async fn poll(&self, timeout: Duration) -> Result<ScanRecords> {
+ let start = std::time::Instant::now();
+ let deadline = start + timeout;
+
+ loop {
+ // Try to collect fetches
+ let fetch_result = self.poll_for_fetches().await?;
+
+ if !fetch_result.is_empty() {
+ // We have data, send next round of fetches and return
+ // This enables pipelining while user processes the data
+ self.log_fetcher.send_fetches().await?;
+ return Ok(ScanRecords::new(fetch_result));
+ }
+
+ // No data available, check if we should wait
+ let now = std::time::Instant::now();
+ if now >= deadline {
+ // Timeout reached, return empty result
+ return Ok(ScanRecords::new(HashMap::new()));
+ }
+
+ // Wait for buffer to become non-empty with remaining time
+ let remaining = deadline - now;
+ let has_data = self
+ .log_fetcher
+ .log_fetch_buffer
+ .await_not_empty(remaining)
+ .await;
+
+ if !has_data {
+ // Timeout while waiting
+ return Ok(ScanRecords::new(HashMap::new()));
+ }
+
+ // Buffer became non-empty, try again
+ }
Review Comment:
await_not_empty just a timeout
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #103:
URL: https://github.com/apache/fluss-rust/pull/103
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Improve log poll [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #103:
URL: https://github.com/apache/fluss-rust/pull/103#issuecomment-3678420966
@zhaohaidao @leekeiabstraction Thanks for your review. Comments addressed now. PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org