Copilot commented on code in PR #579:
URL: https://github.com/apache/fluss-rust/pull/579#discussion_r3334190016
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -1212,40 +1205,325 @@ mod kv_table_test {
.expect("Failed to build schema"),
)
.distributed_by(Some(2), vec!["a".to_string()])
- .property("table.kv.format-version", "2")
- .property("table.datalake.format", "lance")
+ .property("table.kv.format-version", "1")
+ .property("table.datalake.format", "paimon")
.build()
.expect("Failed to build table");
create_table(&admin, &table_path, &table_descriptor).await;
let table = connection.get_table(&table_path).await.unwrap();
+ assert_eq!(
+ table
+ .get_table_info()
+ .get_properties()
+ .get("table.kv.format-version")
+ .unwrap(),
+ "1"
+ );
// Test PutKv with v0 client - should fail with UNSUPPORTED_VERSION
let table_upsert = table.new_upsert().expect("Failed to create
upsert");
let upsert_writer = table_upsert
.create_writer()
.expect("Failed to create writer");
- let mut row = GenericRow::new(3);
- row.set_field(0, 1);
- row.set_field(1, "a");
- row.set_field(2, "value1");
- let upsert_result = upsert_writer
- .upsert(&row)
- .expect("Failed to upsert row")
- .await;
- assert!(
- upsert_result.is_err(),
- "PutKv with v0 client should be rejected for kv_format_v2 table
with non-default bucket key"
+ let mut upsert_row = GenericRow::new(3);
+ upsert_row.set_field(0, 1);
+ upsert_row.set_field(1, "a");
+ upsert_row.set_field(2, "value1");
+ upsert_writer
+ .upsert(&upsert_row)
+ .unwrap()
+ .await
+ .expect("Failed to upsert row");
+
+ // prefix lookup
+ let mut lookup_row = GenericRow::new(1);
+ lookup_row.set_field(0, 1);
+ let mut prefix_lookup = table
+ .new_lookup()
+ .expect("Failed to create lookup")
+ .lookup_by(vec!["a".to_string()])
+ .create_lookuper()
+ .expect("Failed to create prefix lookuper");
+ let lookup_result = prefix_lookup
+ .lookup(&lookup_row)
+ .await
+ .expect("fail to lookup");
+ let row = lookup_result
+ .get_single_row()
+ .unwrap()
+ .expect("Row should exist");
+ assert_eq!(row.get_string(2).unwrap(), "value1");
+ }
+
+ /// Verifies upsert/lookup/update/delete on a KV table whose data lake
format
+ /// is `paimon`, exercising every scalar `DataType` supported by Paimon's
+ /// BinaryRow encoder. With the default `kv_format_version=1`, the client
+ /// uses `PaimonKeyEncoder` for both the bucket key and the primary key,
+ /// so this also exercises the end-to-end Paimon key encoding path.
+ /// ARRAY/MAP/ROW are intentionally excluded because Paimon's BinaryRow
+ /// writer rejects them.
+ #[tokio::test]
+ async fn upsert_and_lookup_with_paimon_format_v2() {
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+
+ let admin = connection.get_admin().unwrap();
+
+ let table_path = TablePath::new("fluss", "test_kv_paimon_format");
+
+ // Schema covering every Paimon-supported scalar type.
+ // INT primary key keeps the bucket-key/primary-key encoding focused
+ // while the value columns sweep all remaining scalar types.
+ let table_descriptor = TableDescriptor::builder()
+ .schema(
+ Schema::builder()
+ .column("id", DataTypes::int())
+ .column("col_tinyint", DataTypes::tinyint())
+ .column("col_smallint", DataTypes::smallint())
+ .column("col_bigint", DataTypes::bigint())
+ .column("col_float", DataTypes::float())
+ .column("col_double", DataTypes::double())
+ .column("col_boolean", DataTypes::boolean())
+ .column("col_char", DataTypes::char(10))
+ .column("col_string", DataTypes::string())
+ .column("col_decimal", DataTypes::decimal(10, 2))
+ .column("col_date", DataTypes::date())
+ .column("col_time_s", DataTypes::time_with_precision(0))
+ .column("col_time_ms", DataTypes::time_with_precision(3))
+ .column("col_time_us", DataTypes::time_with_precision(6))
+ .column("col_time_ns", DataTypes::time_with_precision(9))
+ .column("col_timestamp",
DataTypes::timestamp_with_precision(0))
+ .column(
+ "col_timestamp_ltz",
+ DataTypes::timestamp_ltz_with_precision(6),
+ )
+ .column("col_bytes", DataTypes::bytes())
+ .column("col_binary", DataTypes::binary(4))
+ .primary_key(vec!["id"])
+ .build()
+ .expect("Failed to build schema"),
+ )
+ .property("table.datalake.format", "paimon")
+ .build()
+ .expect("Failed to build table");
Review Comment:
This test asserts `table.kv.format-version == "2"` (and unwraps the
property), but the table descriptor never sets `table.kv.format-version`. If
the server doesn't materialize the default into the properties map, the unwrap
will panic and the test will fail. Set the property explicitly (or change the
assertion to match the default).
##########
crates/fluss/src/row/paimon/paimon_binary_row_writer.rs:
##########
@@ -0,0 +1,550 @@
+// 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 bytes::Bytes;
+
+use crate::error::{Error, Result};
+use crate::metadata::DataType;
+use crate::row::binary::{BinaryWriter, ValueWriter};
+use crate::row::datum::{TimestampLtz, TimestampNtz};
+use crate::row::{Decimal, FlussArray, FlussMap};
+
+/// Header size in bits (for the ChangeType byte at the front of the null
bitset).
+const HEADER_SIZE_IN_BITS: usize = 8;
+/// Maximum number of bytes that can be packed inline into a fixed 8-byte
field slot
+/// (Paimon's variable-length inline-encoding optimisation).
+const MAX_FIX_PART_DATA_SIZE: usize = 7;
+
+/// A Rust port of Java's
+/// `org.apache.fluss.row.encode.paimon.PaimonBinaryRowWriter`, encoding a
Fluss
+/// `InternalRow` using Paimon's BinaryRow layout.
+///
+/// Layout:
+/// - 1-byte ChangeType header at offset 0 (always `INSERT = 0` for key
encoding).
+/// - Null bitset (`nullBitsSizeInBytes` bytes), where bit `pos + 8` indicates
+/// field `pos` is null.
+/// - Fixed-length region: 8 bytes per field after the null bitset.
+/// - Variable-length tail growing on demand.
+///
+/// This writer implements the [`BinaryWriter`] trait so it can plug into
+/// [`ValueWriter::write_value`]. Because the trait API does not pass the
+/// position to the type-specific write methods, the writer keeps an internal
+/// `current_pos` cursor that is advanced after every field write (including
+/// `set_null_at`). The encoder is required to write fields in field order,
+/// matching the iteration order used by [`crate::row::encode`].
+pub struct PaimonBinaryRowWriter {
+ null_bits_size_in_bytes: usize,
+ fixed_size: usize,
+ buffer: Vec<u8>,
+ cursor: usize,
+ current_pos: usize,
+}
+
+impl PaimonBinaryRowWriter {
+ pub fn new(arity: usize) -> Self {
+ let null_bits_size_in_bytes = calculate_bit_set_width_in_bytes(arity);
+ let fixed_size = get_fixed_length_part_size(null_bits_size_in_bytes,
arity);
+ Self {
+ null_bits_size_in_bytes,
+ fixed_size,
+ buffer: vec![0u8; fixed_size],
+ cursor: fixed_size,
+ current_pos: 0,
+ }
+ }
+
+ /// Mirrors Java's `PaimonBinaryRowWriter.createFieldWriter`, returning the
+ /// Fluss [`ValueWriter`] for a Paimon-supported scalar field type.
+ /// ARRAY/MAP/ROW are explicitly rejected (Java's `default` branch throws).
+ pub fn create_value_writer(field_type: &DataType) -> Result<ValueWriter> {
+ match field_type {
+ DataType::Char(_)
+ | DataType::String(_)
+ | DataType::Boolean(_)
+ | DataType::Binary(_)
+ | DataType::Bytes(_)
+ | DataType::Decimal(_)
+ | DataType::TinyInt(_)
+ | DataType::SmallInt(_)
+ | DataType::Int(_)
+ | DataType::Date(_)
+ | DataType::Time(_)
+ | DataType::BigInt(_)
+ | DataType::Float(_)
+ | DataType::Double(_)
+ | DataType::Timestamp(_)
+ | DataType::TimestampLTz(_) =>
ValueWriter::create_value_writer(field_type, None),
+ _ => Err(Error::UnsupportedOperation {
+ message: format!("Unsupported type for Paimon BinaryRow
writer: {field_type:?}"),
+ }),
+ }
+ }
+
+ /// Writes the Paimon ChangeType byte at offset 0 (always `INSERT = 0`
+ /// for key encoding). Must be called immediately after [`Self::reset`].
+ pub fn write_change_type_insert(&mut self) {
+ self.buffer[0] = 0;
+ }
+
+ pub fn to_bytes(&self) -> Bytes {
+ Bytes::copy_from_slice(&self.buffer[..self.cursor])
+ }
+
+ #[allow(dead_code)]
+ pub fn buffer(&self) -> &[u8] {
+ &self.buffer[..self.cursor]
+ }
+
+ #[allow(dead_code)]
+ pub fn cursor(&self) -> usize {
+ self.cursor
+ }
+
+ fn field_offset(&self, pos: usize) -> usize {
+ self.null_bits_size_in_bytes + 8 * pos
+ }
+
+ fn set_null_bit(&mut self, pos: usize) {
+ let bit = pos + HEADER_SIZE_IN_BITS;
+ let byte_index = bit / 8;
+ let bit_in_byte = bit % 8;
+ self.buffer[byte_index] |= 1u8 << bit_in_byte;
+ }
+
+ fn put_long_le(&mut self, offset: usize, value: i64) {
+ self.buffer[offset..offset + 8].copy_from_slice(&value.to_le_bytes());
+ }
+
+ fn put_int_le(&mut self, offset: usize, value: i32) {
+ self.buffer[offset..offset + 4].copy_from_slice(&value.to_le_bytes());
+ }
+
+ fn put_short_le(&mut self, offset: usize, value: i16) {
+ self.buffer[offset..offset + 2].copy_from_slice(&value.to_le_bytes());
+ }
+
+ /// Set `(offset << 32) | size` as a little-endian i64 at the field slot.
+ fn set_offset_and_size(&mut self, pos: usize, offset: usize, size: u64) {
+ let packed = ((offset as i64) << 32) | (size as i64);
+ let field_offset = self.field_offset(pos);
+ self.put_long_le(field_offset, packed);
+ }
+
+ /// Inline ≤ 7-byte payload into the 8-byte fixed slot using Paimon's
layout
+ /// (`firstByte = len | 0x80` in the high byte, data bytes packed
+ /// little-endian into the low bytes).
+ fn write_bytes_to_fix_len_part(&mut self, pos: usize, bytes: &[u8]) {
+ let len = bytes.len();
+ debug_assert!(len <= MAX_FIX_PART_DATA_SIZE);
+ let field_offset = self.field_offset(pos);
+ // Zero the slot first (in case we're reusing buffer positions on
reset).
+ for b in &mut self.buffer[field_offset..field_offset + 8] {
+ *b = 0;
+ }
+ // Data bytes occupy the low-order positions; first byte (len|0x80)
+ // sits at the high-order byte (index 7) thanks to little-endian
layout.
+ self.buffer[field_offset..field_offset + len].copy_from_slice(bytes);
+ self.buffer[field_offset + 7] = (len as u8) | 0x80;
+ }
+
+ fn ensure_capacity(&mut self, needed_size: usize) {
+ let length = self.cursor + needed_size;
+ if self.buffer.len() < length {
+ self.grow(length);
+ }
+ }
+
+ fn grow(&mut self, min_capacity: usize) {
+ let old_capacity = self.buffer.len();
+ let mut new_capacity = old_capacity + (old_capacity >> 1);
+ if new_capacity < min_capacity {
+ new_capacity = min_capacity;
+ }
+ self.buffer.resize(new_capacity, 0);
+ }
+
+ /// Zero out the padding region between `numBytes` and the next 8-byte
+ /// boundary at the current cursor (matches Java's `zeroOutPaddingBytes`).
+ fn zero_out_padding_bytes(&mut self, num_bytes: usize) {
+ if (num_bytes & 0x07) > 0 {
+ let aligned = (num_bytes >> 3) << 3;
+ // 8 bytes starting at cursor + aligned.
+ let off = self.cursor + aligned;
+ for b in &mut self.buffer[off..off + 8] {
+ *b = 0;
+ }
+ }
+ }
+
+ fn write_bytes_to_var_len_part(&mut self, pos: usize, bytes: &[u8]) {
+ let len = bytes.len();
+ let rounded_size = round_number_of_bytes_to_nearest_word(len);
+
+ self.ensure_capacity(rounded_size);
+ self.zero_out_padding_bytes(len);
+
+ self.buffer[self.cursor..self.cursor + len].copy_from_slice(bytes);
+
+ self.set_offset_and_size(pos, self.cursor, len as u64);
+ self.cursor += rounded_size;
+ }
+
+ fn write_bytes_internal(&mut self, pos: usize, bytes: &[u8]) {
+ if bytes.len() <= MAX_FIX_PART_DATA_SIZE {
+ self.write_bytes_to_fix_len_part(pos, bytes);
+ } else {
+ self.write_bytes_to_var_len_part(pos, bytes);
+ }
+ }
+}
+
+/// Number of bytes occupied by Paimon's null bitset for the given arity,
+/// including the 1-byte (8-bit) ChangeType header.
+fn calculate_bit_set_width_in_bytes(arity: usize) -> usize {
+ ((arity + 63 + HEADER_SIZE_IN_BITS) / 64) * 8
+}
+
+fn get_fixed_length_part_size(null_bits_size_in_bytes: usize, arity: usize) ->
usize {
+ null_bits_size_in_bytes + 8 * arity
+}
+
+fn round_number_of_bytes_to_nearest_word(num_bytes: usize) -> usize {
+ let remainder = num_bytes & 0x07;
+ if remainder == 0 {
+ num_bytes
+ } else {
+ num_bytes + (8 - remainder)
+ }
+}
+
+impl BinaryWriter for PaimonBinaryRowWriter {
+ fn reset(&mut self) {
+ self.cursor = self.fixed_size;
+ self.current_pos = 0;
+ // Zero the null-bits region only (Java semantics: field slots are not
+ // wiped because every field is overwritten in the next encode pass).
+ for b in &mut self.buffer[..self.null_bits_size_in_bytes] {
+ *b = 0;
+ }
+ }
+
+ fn set_null_at(&mut self, pos: usize) {
+ debug_assert_eq!(
+ pos, self.current_pos,
+ "Paimon writer expects in-order writes"
+ );
+ self.set_null_bit(pos);
+ let field_offset = self.field_offset(pos);
+ self.put_long_le(field_offset, 0);
+ self.current_pos = pos + 1;
+ }
+
+ fn write_boolean(&mut self, value: bool) {
+ let pos = self.current_pos;
+ let off = self.field_offset(pos);
+ self.buffer[off] = if value { 1 } else { 0 };
+ self.current_pos = pos + 1;
+ }
+
+ fn write_byte(&mut self, value: u8) {
+ let pos = self.current_pos;
+ let off = self.field_offset(pos);
+ self.buffer[off] = value;
+ self.current_pos = pos + 1;
+ }
Review Comment:
`reset()` intentionally does not clear the fixed-length field slots, but
`write_boolean` / `write_byte` currently only overwrite the first byte of the
8-byte slot. If the writer is reused, the remaining 7 bytes can contain stale
data, making identical values encode to different byte sequences (breaking key
equality / bucketing). Zero the whole slot before writing the value, consistent
with `write_short`/`write_int`/`write_float`.
##########
crates/fluss/src/rpc/api_key.rs:
##########
@@ -78,12 +78,12 @@ impl ApiKey {
| ApiKey::GetDatabaseInfo
| ApiKey::CreatePartition
| ApiKey::DropPartition
- | ApiKey::Authenticate
- // TODO(key-encoding-v1): The Java server supports v0..v1 for these
- // APIs, but the Rust client has not yet implemented the v1 key
- // encoding format. Pinned to v0 until that is done.
- | ApiKey::PutKv | ApiKey::Lookup | ApiKey::PrefixLookup => {
- Some(ApiVersionRange::new(ApiVersion(0), ApiVersion(0)))
+ | ApiKey::Authenticate => Some(ApiVersionRange::new(ApiVersion(0),
ApiVersion(0))),
+ // PutKv / Lookup / PrefixLookup support v0 (legacy key encoding)
+ // and v1 (Paimon BinaryRow key encoding for kv_format_version=2
+ // non-default bucket keys). The Rust client encodes both.
+ ApiKey::PutKv | ApiKey::Lookup | ApiKey::PrefixLookup => {
+ Some(ApiVersionRange::new(ApiVersion(0), ApiVersion(1)))
}
Review Comment:
Bumping PutKv/Lookup/PrefixLookup to advertise max version `1` will change
API-version negotiation results. There are existing unit tests that hardcode
negotiated v0 for PutKv (e.g.
`crates/fluss/src/rpc/server_connection.rs:1176+`), which will now fail and
should be updated alongside this change. Also ensure any request/response
encoding paths are valid for v1 (since `resolve_api_version_for(None,
ApiKey::PutKv)` will now default to v1 before negotiation).
##########
.run/Test.run.xml:
##########
@@ -0,0 +1,20 @@
+<component name="ProjectRunConfigurationManager">
+ <configuration default="false" name="Test"
type="CargoCommandRunConfiguration" factoryName="Cargo Command"
nameIsGenerated="true">
+ <option name="buildProfileId" value="dev" />
+ <option name="command" value="test --features integration_tests --test
test_fluss -- --test-threads=1" />
+ <option name="workingDirectory" value="file://$PROJECT_DIR$" />
+ <envs />
+ <option name="emulateTerminal" value="true" />
+ <option name="channel" value="DEFAULT" />
+ <option name="requiredFeatures" value="true" />
+ <option name="allFeatures" value="true" />
+ <option name="withSudo" value="false" />
+ <option name="buildTarget" value="REMOTE" />
+ <option name="backtrace" value="SHORT" />
+ <option name="isRedirectInput" value="false" />
+ <option name="redirectInputPath" value="" />
+ <method v="2">
+ <option name="CARGO.BUILD_TASK_PROVIDER" enabled="true" />
+ </method>
+ </configuration>
+</component>
Review Comment:
This new XML file is missing the Apache Software Foundation license header.
The repository enforces headers via `.licenserc.yaml`, so this will fail the
license check unless the header is added (or the path is excluded).
##########
crates/fluss/tests/integration/kv_table.rs:
##########
@@ -1212,40 +1205,325 @@ mod kv_table_test {
.expect("Failed to build schema"),
)
.distributed_by(Some(2), vec!["a".to_string()])
- .property("table.kv.format-version", "2")
- .property("table.datalake.format", "lance")
+ .property("table.kv.format-version", "1")
+ .property("table.datalake.format", "paimon")
.build()
.expect("Failed to build table");
create_table(&admin, &table_path, &table_descriptor).await;
let table = connection.get_table(&table_path).await.unwrap();
+ assert_eq!(
+ table
+ .get_table_info()
+ .get_properties()
+ .get("table.kv.format-version")
+ .unwrap(),
+ "1"
+ );
// Test PutKv with v0 client - should fail with UNSUPPORTED_VERSION
Review Comment:
This comment is now misleading: the test no longer expects PutKv to be
rejected, and the code below asserts the upsert succeeds. Please update the
comment to match the actual behavior being tested.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
