Fix column prune won't work when compression is enabel

[luoyuxia]

Search before asking

• I searched in the issues and found nothing similar.

Description

as title. #83 (comment). I'm worried about that it only happen in fluss-rust

Willingness to contribute

• I'm willing to submit a PR!

[luoyuxia]

cc @leekeiabstraction

[leekeiabstraction]

From my investigation I find that this issue is primarily occurs after rust client writes into a log table. After rust writes, read with pruning breaks on both rust and java side. Java writes does not break rust read with pruning.

@zhaohaidao When you mentioned that you observed this issue on both sides, did you check if scanning with pruning fails if the table is only written by Java side?

Investigation details:

Vanilla Java client (through Flink SQL) seems to be working fine.

Flink SQL> CREATE TABLE log_table  (order_id BIGINT, item_id BIGINT) WITH ('bucket.num' = '1');
[INFO] Execute statement succeeded.

Flink SQL> INSERT INTO log_table VALUES (1, 1), (2, 2);
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 930a1a418dc3e28c856c94f424d44a79


Flink SQL> SELECT * FROM log_table;
+----+----------------------+----------------------+
| op |             order_id |              item_id |
+----+----------------------+----------------------+
| +I |                    1 |                    1 |
| +I |                    2 |                    2 |
^CQuery terminated, received a total of 2 rows (1.43 seconds)

Flink SQL> SELECT order_id FROM log_table;
+----+----------------------+
| op |             order_id |
+----+----------------------+
| +I |                    1 |
| +I |                    2 |
^CQuery terminated, received a total of 2 rows (2.12 seconds)

Reading the same data inserted by java client from rust side seems to be working fine too.

Get created table:
 TableInfo{ table_path=TablePath { database: "fluss", table: "log_table" }, table_id=5, schema_id=1, schema=Schema { columns: [Column { name: "order_id", data_type: BigInt(BigIntType { nullable: true }), comment: None }, Column { name: "item_id", data_type: BigInt(BigIntType { nullable: true }), comment: None }], primary_key: None, row_type: Row(RowType { nullable: true, fields: [DataField { name: "order_id", data_type: BigInt(BigIntType { nullable: true }), description: None }, DataField { name: "item_id", data_type: BigInt(BigIntType { nullable: true }), description: None }] }) }, physical_primary_keys=[], bucket_keys=[], partition_keys=[], num_buckets=1, properties={"table.replication.factor": "1"}, custom_properties={"bucket.num": "1"}, comment=Some(""), created_time=1766247537221, modified_time=1766247537221 }
Start to poll records......
{1}@0
{2}@1

As soon as data is written from Rust client side, scanning with prune breaks on both Rust and Java side (non pruned scan works).

thread 'main' (2136807) panicked at /Users/keithlee/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/arrow-ipc-57.1.0/src/compression.rs:264:29:
range end index 8 out of range for slice of length 1
stack backtrace:
   0: __rustc::rust_begin_unwind

Flink SQL> SELECT order_id FROM log_table;
+----+----------------------+
| op |             order_id |
+----+----------------------+
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalArgumentException: Not enough data to decompress.  

[leekeiabstraction]

@luoyuxia

It seems like we currently do not specify compression context on writer on rust side, can you confirm that this is the case? Is there another place where compression context is specified?

fluss-rust/crates/fluss/src/record/arrow.rs

Line 292 in 076cbd6

let mut writer = StreamWriter::try_new(&mut arrow_batch_bytes, &table_schema)?;

I believe we should pass in compression context using StreamWriter::try_new_with_options(writer, schema, write_options)

[luoyuxia]

@leekeiabstraction Thank you very much for your investgation.