fresh-borzoni commented on code in PR #156: URL: https://github.com/apache/fluss-rust/pull/156#discussion_r2686347737
########## crates/fluss/src/record/kv/kv_record_batch.rs: ########## @@ -0,0 +1,370 @@ +// 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. + +//! KV record batch implementation. +//! +//! The schema of a KvRecordBatch is: +//! - Length => Int32 +//! - Magic => Int8 +//! - CRC => Uint32 +//! - SchemaId => Int16 +//! - Attributes => Int8 +//! - WriterId => Int64 +//! - BatchSequence => Int32 +//! - RecordCount => Int32 +//! - Records => [Record] +//! +//! The CRC covers data from the SchemaId to the end of the batch. + +use bytes::Bytes; +use std::io; + +use crate::record::kv::KvRecord; + +// Field lengths in bytes +pub const LENGTH_LENGTH: usize = 4; +pub const MAGIC_LENGTH: usize = 1; +pub const CRC_LENGTH: usize = 4; +pub const SCHEMA_ID_LENGTH: usize = 2; +pub const ATTRIBUTE_LENGTH: usize = 1; +pub const WRITER_ID_LENGTH: usize = 8; +pub const BATCH_SEQUENCE_LENGTH: usize = 4; +pub const RECORDS_COUNT_LENGTH: usize = 4; + +// Field offsets +pub const LENGTH_OFFSET: usize = 0; +pub const MAGIC_OFFSET: usize = LENGTH_OFFSET + LENGTH_LENGTH; +pub const CRC_OFFSET: usize = MAGIC_OFFSET + MAGIC_LENGTH; +pub const SCHEMA_ID_OFFSET: usize = CRC_OFFSET + CRC_LENGTH; +pub const ATTRIBUTES_OFFSET: usize = SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH; +pub const WRITER_ID_OFFSET: usize = ATTRIBUTES_OFFSET + ATTRIBUTE_LENGTH; +pub const BATCH_SEQUENCE_OFFSET: usize = WRITER_ID_OFFSET + WRITER_ID_LENGTH; +pub const RECORDS_COUNT_OFFSET: usize = BATCH_SEQUENCE_OFFSET + BATCH_SEQUENCE_LENGTH; +pub const RECORDS_OFFSET: usize = RECORDS_COUNT_OFFSET + RECORDS_COUNT_LENGTH; + +/// Total header size +pub const RECORD_BATCH_HEADER_SIZE: usize = RECORDS_OFFSET; + +/// Overhead of the batch (length field) +pub const KV_OVERHEAD: usize = LENGTH_OFFSET + LENGTH_LENGTH; + +/// A KV record batch. +/// +/// This struct provides read access to a serialized KV record batch. +pub struct KvRecordBatch { + data: Bytes, + position: usize, +} + +impl KvRecordBatch { + /// Create a new KvRecordBatch pointing to the given data at the specified position. + pub fn new(data: Bytes, position: usize) -> Self { + Self { data, position } + } + + /// Get the size in bytes of this batch. + /// Returns 0 if the batch is invalid or length is negative. + pub fn size_in_bytes(&self) -> usize { + if self.data.len() < self.position.saturating_add(LENGTH_LENGTH) { + return 0; + } + let length_i32 = i32::from_be_bytes([ Review Comment: Good catch, thank you! -- 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]
