Support ColumnPruning #57

zhaohaidao · 2025-11-22T16:45:05Z

Purpose

Support ColumnPruning to save significant network costs

Brief change log

Tests

API and Format

Documentation

zhaohaidao · 2025-11-23T09:42:48Z

@luoyuxia PTAL if u have time.

Copilot

Pull request overview

This PR adds column projection (pruning) support to the Fluss Rust client, enabling users to fetch only specific columns from tables to reduce network transfer costs. The implementation supports both server-side projection pushdown and client-side projection, with the ability to project by column indices or names.

Key Changes:

Added projection support to the scanner API with project() and project_by_name() methods
Implemented custom Arrow IPC message parsing to support projection at the record batch level
Enhanced ReadContext to handle different projection modes (pushdown vs. client-side)

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 12 comments.

Show a summary per file

File	Description
`crates/fluss/src/record/arrow.rs`	Added `parse_ipc_message()` method for custom IPC parsing, modified `records()` to support projection, and extended `ReadContext` with projection capabilities
`crates/fluss/src/client/table/scanner.rs`	Added `project()` and `project_by_name()` methods to `TableScan`, threaded projection fields through `LogScanner` and `LogFetcher`, and updated fetch request generation to include projection information
`crates/examples/src/example_projection.rs`	New example demonstrating column projection usage with both full scan and projected scan scenarios
`crates/examples/src/example_table.rs`	Updated bootstrap server address for consistency
`crates/examples/Cargo.toml`	Added env_logger dependency and registered the new projection example

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-11-23T10:07:31Z

crates/fluss/src/client/table/scanner.rs

+            let (projection_enabled, projected_fields) = if let Some(fields) = &self.projected_fields {
+                if fields.is_empty() {
+                    (false, vec![])
+                } else {


Converting usize to i32 using as i32 could lead to data loss or incorrect values on 64-bit systems where usize can exceed i32::MAX. This could cause issues when projecting columns with very high indices. Consider either:

Using a bounded type for column indices (e.g., validating they're within i32 range)

Changing the protobuf field type to support larger indices

Adding an explicit check and error if the index exceeds i32::MAX

Suggested change

} else {

} else {

// Check for out-of-range indices before converting

if let Some(&idx) = fields.iter().find(|&&i| i > i32::MAX as usize) {

// Return early with error if any index is too large

return HashMap::new(); // Or, if possible, return Err(Error::msg(...))

}

crates/fluss/src/client/table/scanner.rs

crates/fluss/src/record/arrow.rs

Copilot · 2025-11-23T10:07:31Z

crates/fluss/src/record/arrow.rs

+        data: &'a [u8],
+    ) -> Option<(arrow::ipc::RecordBatch<'a>, Buffer, arrow::ipc::MetadataVersion)> {
+        const CONTINUATION_MARKER: u32 = 0xFFFFFFFF;
+


[nitpick] Trailing whitespace detected. Please remove it to maintain code cleanliness.

Suggested change

Copilot · 2025-11-23T10:07:32Z

crates/fluss/src/record/arrow.rs

+        if continuation != CONTINUATION_MARKER {
+            return None;
+        }
+


[nitpick] Trailing whitespace detected. Please remove it to maintain code cleanliness.

Suggested change

Copilot · 2025-11-23T10:07:32Z

crates/fluss/src/client/table/scanner.rs

+                        let projected_schema = arrow_schema::Schema::new(
+                            projected_fields
+                                .iter()
+                                .map(|&idx| full_arrow_schema.field(idx).clone())
+                                .collect::<Vec<_>>(),
+                        );


Similar to the bounds checking issue in ReadContext::to_arrow_metadata, there's no validation when creating the projected schema here. If projected_fields contains an invalid index, calling full_arrow_schema.field(idx) will panic. While TableScan::project validates indices, this code path could still be reached with invalid indices if the projection is set up through other means.

Copilot · 2025-11-23T10:07:33Z

crates/fluss/src/client/table/scanner.rs

+                        let projected_schema = arrow_schema::Schema::new(
+                            projected_fields
+                                .iter()
+                                .map(|&idx| full_arrow_schema.field(idx).clone())
+                                .collect::<Vec<_>>(),
+                        );


This code for building a projected schema is duplicated from the logic in ReadContext::to_arrow_metadata (lines 652-658 in arrow.rs). This duplication creates a maintenance burden and increases the risk of inconsistencies. Consider extracting this logic into a shared helper method.

Suggested change

let projected_schema = arrow_schema::Schema::new(

projected_fields

.iter()

.map(|&idx| full_arrow_schema.field(idx).clone())

.collect::<Vec<_>>(),

);

let projected_schema = build_projected_schema(&full_arrow_schema, &projected_fields);

Copilot · 2025-11-23T10:07:33Z

crates/fluss/src/record/arrow.rs

+        if data.len() < 8 {
+            return None;
+        }
+


[nitpick] Trailing whitespace detected. Please remove it to maintain code cleanliness.

Suggested change

crates/fluss/src/client/table/scanner.rs

Copilot · 2025-11-23T10:07:33Z

crates/fluss/src/client/table/scanner.rs

+                    let (projection_enabled, _) = if !projected_fields.is_empty() {
+                        (true, projected_fields.clone())
+                    } else {
+                        (false, vec![])
+                    };


[nitpick] The check for empty projected fields (line 190) and subsequent handling creates unnecessary complexity. If projected_fields is Some(vec![]) (an empty vector), it should probably be treated the same as None - i.e., fetch all fields. The current logic sets projection_enabled = false for empty vectors, but it's unclear why an empty projection list would ever be intentionally created. Consider either preventing empty vectors from being set in the first place (e.g., in the project method) or simplifying this logic.

luoyuxia

@zhaohaidao Thanks for the pr. Left minor comments. PTAL.
Also, seem this pr doesn't include reordeing the project fields returned by server for final output. See java impl https://github.com/apache/fluss/blob/ae84521aaaef5448a0bc5a63fc83e6ca536ca452/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java#L84

don't forget to create an issue to track it. It's critical when project fields is out of orders like [3, 2, 1]

crates/examples/src/example_projection.rs

crates/fluss/src/client/table/scanner.rs

crates/fluss/src/record/arrow.rs

Copilot

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 9 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-12-03T07:20:45Z

crates/examples/src/example_table.rs

 pub async fn main() -> Result<()> {
    let mut config = Config::parse();
-    config.bootstrap_server = Some("127.0.0.1:56405".to_string());
+    config.bootstrap_server = Some("127.0.0.1:9123".to_string());


Hardcoded bootstrap server changed from port 56405 to 9123. This change appears to be unrelated to the column pruning feature and may have been committed accidentally. If this is an intentional configuration change for development/testing, it should be explained in the PR description or reverted to avoid affecting other developers' local environments.

Suggested change

config.bootstrap_server = Some("127.0.0.1:9123".to_string());

config.bootstrap_server = Some("127.0.0.1:56405".to_string());

crates/examples/Cargo.toml

crates/fluss/src/client/table/scanner.rs

crates/fluss/src/record/arrow.rs

Copilot · 2025-12-03T07:20:47Z

crates/fluss/src/record/arrow.rs

+            return None;
+        }
+
+        // Calculate reordering indexes to transform from sorted order to user-requested order


The .expect() call could panic if there's a mismatch between projected_fields and projection_in_order. While this should never happen with the current code logic (since both are derived from the same source in create_read_context), this represents a potentially unrecoverable error condition. Consider:

Adding debug assertions to validate invariants in with_projection_pushdown():

debug_assert_eq!(projected_fields.len(), projection_in_order.len()); debug_assert!(projected_fields.iter().all(|&f| projection_in_order.contains(&f)));

Or, returning a Result from reordering_indexes() instead of using .expect() to make error handling more explicit and recoverable.

crates/fluss/src/client/table/scanner.rs

crates/fluss/src/record/arrow.rs

crates/fluss/src/client/table/scanner.rs

crates/fluss/src/record/arrow.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

zhaohaidao · 2025-12-03T07:40:07Z

@zhaohaidao Thanks for the pr. Left minor comments. PTAL. Also, seem this pr doesn't include reordeing the project fields returned by server for final output. See java impl https://github.com/apache/fluss/blob/ae84521aaaef5448a0bc5a63fc83e6ca536ca452/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java#L84

don't forget to create an issue to track it. It's critical when project fields is out of orders like [3, 2, 1]

@luoyuxia Thank you for your concise and elegant suggestions. Comments are addressed. PTAL if u have time

Copilot

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 10 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

crates/fluss/src/record/arrow.rs

crates/fluss/src/client/table/scanner.rs

crates/fluss/src/record/arrow.rs

crates/fluss/src/client/table/scanner.rs

crates/fluss/src/record/arrow.rs

crates/fluss/src/client/table/scanner.rs

luoyuxia · 2025-12-03T16:56:20Z

@zhaohaidao Hi, I append a minior commit to improve reordering. PTAL.

zhaohaidao · 2025-12-03T20:19:25Z

@zhaohaidao Hi, I append a minior commit to improve reordering. PTAL.

@luoyuxia Thanks, LGTM

crates/fluss/tests/integration/table.rs

crates/fluss/src/client/table/scanner.rs

luoyuxia · 2025-12-04T02:02:42Z

@zhaohaidao Thanks for updating. Left few comments. Don't forget rebase main branch to resovle conflicts.

zhaohaidao · 2025-12-04T02:45:51Z

@zhaohaidao Thanks for updating. Left few comments. Don't forget rebase main branch to resovle conflicts.

Thanks for reminding. The comments are addressed. PTAL if u have time

luoyuxia

+1

赵海源 added 6 commits November 22, 2025 17:01

test example_table

cce8b53

projection example

590e476

chore refactor

b132031

support non-projection

c955fd5

chore refactor

439f071

address comments

8d19bad

zhaohaidao changed the title ~~(WIP)Support ColumnPruning~~ Support ColumnPruning Nov 23, 2025

luoyuxia requested review from Copilot and luoyuxia November 23, 2025 10:02

Copilot started reviewing on behalf of luoyuxia November 23, 2025 10:03 View session

Copilot finished reviewing on behalf of luoyuxia November 23, 2025 10:05

Copilot AI reviewed Nov 23, 2025

View reviewed changes

address comments

1a92c48

luoyuxia reviewed Nov 23, 2025

View reviewed changes

赵海源 added 5 commits December 1, 2025 09:59

support project fields reordering

396d874

address comments

a0c1ce2

address comments

25f2c99

address comments

361a23a

address comments

29e1dd6

zhaohaidao mentioned this pull request Dec 3, 2025

parse_ipc_message should return exception instead of return None if parse failed #72

Open

2 tasks

luoyuxia requested a review from Copilot December 3, 2025 07:13

zhaohaidao mentioned this pull request Dec 3, 2025

Support ColumnPruning #73

Closed

2 tasks

Copilot started reviewing on behalf of luoyuxia December 3, 2025 07:14 View session

fix fmt check

281cc64

Copilot finished reviewing on behalf of luoyuxia December 3, 2025 07:18

Copilot AI reviewed Dec 3, 2025

View reviewed changes

zhaohaidao and others added 2 commits December 3, 2025 15:21

Update crates/examples/Cargo.toml

2dd9e37

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

address comments

e260f89

luoyuxia requested a review from Copilot December 3, 2025 16:38

Copilot started reviewing on behalf of luoyuxia December 3, 2025 16:38 View session

Copilot finished reviewing on behalf of luoyuxia December 3, 2025 16:41

Copilot AI reviewed Dec 3, 2025

View reviewed changes

luoyuxia force-pushed the poc branch from d138821 to baffd9c Compare December 3, 2025 16:55

luoyuxia force-pushed the poc branch from baffd9c to 257409c Compare December 3, 2025 17:02

improve reorder logic

2d98dc7

luoyuxia force-pushed the poc branch from 257409c to 2d98dc7 Compare December 3, 2025 17:08

赵海源 added 2 commits December 4, 2025 09:27

fix failed case

bb53f97

add test case change

e34a1f8

luoyuxia reviewed Dec 4, 2025

View reviewed changes

crates/fluss/tests/integration/table.rs Outdated Show resolved Hide resolved

luoyuxia reviewed Dec 4, 2025

View reviewed changes

crates/fluss/src/client/table/scanner.rs Outdated Show resolved Hide resolved

address comments

03c9b03

赵海源 added 2 commits December 4, 2025 10:07

add project_fields_in_order

afc048a

Merge remote-tracking branch 'origin/main' into poc

7eb33aa

luoyuxia approved these changes Dec 4, 2025

View reviewed changes

fix fmt check

c398161

luoyuxia merged commit e4e017c into apache:main Dec 4, 2025
13 checks passed

-                } else {
+                } else {
+                    // Check for out-of-range indices before converting
+                    if let Some(&idx) = fields.iter().find(|&&i| i > i32::MAX as usize) {
+                        // Return early with error if any index is too large
+                        return HashMap::new(); // Or, if possible, return Err(Error::msg(...))
+                    }

	config.bootstrap_server = Some("127.0.0.1:9123".to_string());
	config.bootstrap_server = Some("127.0.0.1:56405".to_string());

Support ColumnPruning #57

Support ColumnPruning #57

Conversation

zhaohaidao commented Nov 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

zhaohaidao commented Nov 23, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Nov 23, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Copilot AI Nov 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 23, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 23, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI Nov 23, 2025

Choose a reason for hiding this comment

Uh oh!

luoyuxia left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI Dec 3, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI Dec 3, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

zhaohaidao commented Dec 3, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

zhaohaidao commented Nov 22, 2025 •

edited

Loading