add missing implementation, cleanup, wiring

Author: fresh-borzoni

crates/fluss/Cargo.toml

@@ -48,7 +48,7 @@ tokio = { workspace = true }
 parking_lot = "0.12"
 bytes = "1.10.1"
 dashmap = "6.1.0"
-rust_decimal = "1"
+bigdecimal = { version = "0.4", features = ["serde"] }
 ordered-float = { version = "5", features = ["serde"] }
 parse-display = "0.10"
 ref-cast = "1.0"

crates/fluss/src/record/kv/kv_record_batch.rs

@@ -415,7 +415,7 @@ mod tests {
 
         let key1 = b"key1";
         let mut value1_writer = CompactedRowWriter::new(1);
-        value1_writer.write_bytes(&[1, 2, 3, 4, 5]);
+        value1_writer.write_bytes(&[1, 2, 3, 4, 5]).unwrap();
 
         let row_type = RowType::with_data_types([DataTypes::bytes()].to_vec());
         let row = &CompactedRow::from_bytes(&row_type, value1_writer.buffer());

crates/fluss/src/record/kv/kv_record_batch_builder.rs

@@ -382,7 +382,7 @@ mod tests {
 
         // Test KvFormat validation
         let mut row_writer = CompactedRowWriter::new(1);
-        row_writer.write_int(42);
+        row_writer.write_int(42).unwrap();
         let row_type = RowType::with_data_types(vec![DataTypes::int()]);
         let row = &CompactedRow::from_bytes(&row_type, row_writer.buffer());
 
@@ -498,8 +498,8 @@ mod tests {
 
         // Create and append first record with CompactedRowWriter
         let mut row_writer1 = CompactedRowWriter::new(2);
-        row_writer1.write_int(42);
-        row_writer1.write_string("hello");
+        row_writer1.write_int(42).unwrap();
+        row_writer1.write_string("hello").unwrap();
 
         let row_type = RowType::with_data_types([DataTypes::int(), DataTypes::string()].to_vec());
         let row1 = &CompactedRow::from_bytes(&row_type, row_writer1.buffer());
@@ -510,8 +510,8 @@ mod tests {
 
         // Create and append second record
         let mut row_writer2 = CompactedRowWriter::new(2);
-        row_writer2.write_int(100);
-        row_writer2.write_string("world");
+        row_writer2.write_int(100).unwrap();
+        row_writer2.write_string("world").unwrap();
 
         let row2 = &CompactedRow::from_bytes(&row_type, row_writer2.buffer());