From b8fab081aed7435811daaa00c97a6547a02bb31e Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Fri, 28 Aug 2026 04:44:37 +0800 Subject: [PATCH] feat(file_index): add Java-compatible Bitmap V2 writer Signed-off-by: QuakeWang --- crates/paimon/src/file_index/bitmap/mod.rs | 345 +++++++++++++++++- crates/paimon/src/file_index/bitmap/writer.rs | 344 +++++++++++++++++ 2 files changed, 688 insertions(+), 1 deletion(-) create mode 100644 crates/paimon/src/file_index/bitmap/writer.rs diff --git a/crates/paimon/src/file_index/bitmap/mod.rs b/crates/paimon/src/file_index/bitmap/mod.rs index 9a351c4b..105cd739 100644 --- a/crates/paimon/src/file_index/bitmap/mod.rs +++ b/crates/paimon/src/file_index/bitmap/mod.rs @@ -28,6 +28,8 @@ use crate::file_index::file_index_result::FileIndexResult; use crate::spec::{DataType, Datum, PredicateOperator}; use crate::{Error, Result}; +pub(crate) mod writer; + const VERSION_1: u8 = 1; const VERSION_2: u8 = 2; const JAVA_CANONICAL_FLOAT_NAN_BITS: u32 = 0x7fc0_0000; @@ -950,8 +952,9 @@ impl FileIndexReader for BitmapFileIndexReader { #[cfg(test)] mod tests { use super::*; + use crate::common::Options; use crate::spec::{ - BigIntType, BooleanType, CharType, DateType, DoubleType, FloatType, IntType, + BigIntType, BinaryType, BooleanType, CharType, DateType, DoubleType, FloatType, IntType, LocalZonedTimestampType, SmallIntType, TimeType, TimestampType, TinyIntType, VarCharType, }; @@ -1100,6 +1103,20 @@ mod tests { "00010000000400000000000000143a300000010000000000010010000000000004", "00" ); + const JAVA_INT_MULTIPLE_BODIES_MULTIBLOCK_V2: &str = concat!( + "020000000800000003010000000000000014000000020000000100000000000000", + "030000001c0000002c000000020000000100000014000000140000000200000028", + "0000001400000001000000030000003c000000143a300000010000000000010010", + "000000020005003a300000010000000000010010000000010006003a3000000100", + "00000000010010000000000004003a300000010000000000010010000000030007", + "00" + ); + const JAVA_STRING_BLOCK_BOUNDARY_V2: &str = concat!( + "020000000400000003000000000200000001610000000000000002636300000021", + "0000003300000002000000016100000000000000140000000462626262ffffffff", + "ffffffff00000001000000026363fffffffdffffffff3a3000000100000000000100", + "1000000001000300" + ); const JAVA_SINGLETON_NULL_V1: &str = "01000000030000000201fffffffd00000000ffffffff00000001fffffffe"; const JAVA_SINGLETON_NULL_V2: &str = concat!( @@ -1108,6 +1125,16 @@ mod tests { ); const JAVA_EMPTY_V1: &str = "01000000000000000000"; const JAVA_EMPTY_V2: &str = "020000000000000000000000000000000000"; + const JAVA_FLOAT_SINGLETONS_V2: &str = concat!( + "02000000040000000301fffffffc00000012000000018000000000000000000000", + "280000000380000000fffffffdffffffff00000000fffffffeffffffff7fc00000", + "ffffffffffffffff" + ); + const JAVA_DOUBLE_SINGLETONS_V2: &str = concat!( + "02000000040000000301fffffffc00000012000000018000000000000000000000", + "0000000034000000038000000000000000fffffffdffffffff0000000000000000", + "fffffffeffffffff7ff8000000000000ffffffffffffffff" + ); struct Fixture { name: &'static str, @@ -1268,6 +1295,322 @@ mod tests { ] } + #[test] + fn test_v2_writer_matches_java_golden_payloads() { + for fixture in fixtures() { + let mut writer = + writer::BitmapFileIndexWriter::try_new(fixture.data_type, &Options::new()) + .unwrap_or_else(|error| panic!("{} writer failed: {error}", fixture.name)); + for datum in [ + Some(&fixture.repeated), + Some(&fixture.singleton), + None, + Some(&fixture.repeated), + None, + ] { + writer + .write(datum) + .unwrap_or_else(|error| panic!("{} write failed: {error}", fixture.name)); + } + assert_eq!( + writer.serialized_bytes().unwrap(), + bytes(fixture.v2), + "{} V2 writer", + fixture.name + ); + } + } + + #[test] + fn test_v2_writer_matches_java_floating_value_encoding() { + let float_values = [ + Datum::Float(f32::from_bits(0xffa1_2345)), + Datum::Float(0.0), + Datum::Float(-0.0), + ]; + let mut float_writer = writer::BitmapFileIndexWriter::try_new( + DataType::Float(FloatType::new()), + &Options::new(), + ) + .unwrap(); + for value in &float_values { + float_writer.write(Some(value)).unwrap(); + } + float_writer.write(None).unwrap(); + assert_eq!( + float_writer.serialized_bytes().unwrap(), + bytes(JAVA_FLOAT_SINGLETONS_V2) + ); + + let double_values = [ + Datum::Double(f64::from_bits(0xfff0_1234_5678_9abc)), + Datum::Double(0.0), + Datum::Double(-0.0), + ]; + let mut double_writer = writer::BitmapFileIndexWriter::try_new( + DataType::Double(DoubleType::new()), + &Options::new(), + ) + .unwrap(); + for value in &double_values { + double_writer.write(Some(value)).unwrap(); + } + double_writer.write(None).unwrap(); + assert_eq!( + double_writer.serialized_bytes().unwrap(), + bytes(JAVA_DOUBLE_SINGLETONS_V2) + ); + } + + #[test] + fn test_v2_writer_round_trip_null_singleton_and_multiple_values() { + let data_type = DataType::Int(IntType::new()); + let repeated = Datum::Int(-123_456_789); + let singleton = Datum::Int(42); + let mut writer = + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &Options::new()).unwrap(); + for datum in [ + Some(&repeated), + Some(&singleton), + None, + Some(&repeated), + None, + ] { + writer.write(datum).unwrap(); + } + + let serialized = writer.serialized_bytes().unwrap(); + assert_eq!(serialized, bytes(JAVA_INT_V2)); + let reader = BitmapFileIndexReader::try_new(data_type.clone(), serialized).unwrap(); + assert!(matches!(&reader.index, BitmapIndex::V2(_))); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::Eq, &[repeated]), + selection([0, 3]) + ); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::Eq, &[singleton]), + selection([1]) + ); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::IsNull, &[]), + selection([2, 4]) + ); + } + + #[test] + fn test_v2_writer_empty_and_singleton_null_payloads() { + let data_type = DataType::Int(IntType::new()); + let mut empty = + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &Options::new()).unwrap(); + assert_eq!(empty.serialized_bytes().unwrap(), bytes(JAVA_EMPTY_V2)); + + let mut singleton = + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &Options::new()).unwrap(); + for datum in [Some(&Datum::Int(0)), Some(&Datum::Int(1)), None] { + singleton.write(datum).unwrap(); + } + let serialized = singleton.serialized_bytes().unwrap(); + assert_eq!(serialized, bytes(JAVA_SINGLETON_NULL_V2)); + let reader = BitmapFileIndexReader::try_new(data_type.clone(), serialized).unwrap(); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::IsNull, &[]), + selection([2]) + ); + + let mut all_null = + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &Options::new()).unwrap(); + for _ in 0..3 { + all_null.write(None).unwrap(); + } + let reader = + BitmapFileIndexReader::try_new(data_type.clone(), all_null.serialized_bytes().unwrap()) + .unwrap(); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::IsNull, &[]), + selection(0..3) + ); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::IsNotNull, &[]), + selection([]) + ); + } + + #[test] + fn test_v2_writer_multiple_index_blocks() { + let data_type = DataType::Int(IntType::new()); + let mut options = Options::new(); + options.set("index-block-size", "16"); + let mut writer = + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &options).unwrap(); + for value in [4, 1, 3, 2, 4] { + writer.write(Some(&Datum::Int(value))).unwrap(); + } + + let serialized = writer.serialized_bytes().unwrap(); + assert_eq!(serialized, bytes(JAVA_INT_MULTIBLOCK_V2)); + let reader = BitmapFileIndexReader::try_new(data_type.clone(), serialized).unwrap(); + let blocks = match &reader.index { + BitmapIndex::V2(index) => &index.blocks, + BitmapIndex::V1(_) => panic!("expected V2 index"), + }; + assert_eq!(blocks.len(), 4); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::Eq, &[Datum::Int(4)]), + selection([0, 4]) + ); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::In, + &[Datum::Int(1), Datum::Int(2), Datum::Int(3)] + ), + selection([1, 2, 3]) + ); + } + + #[test] + fn test_v2_writer_multiple_bitmap_bodies_across_index_blocks() { + let data_type = DataType::Int(IntType::new()); + let mut options = Options::new(); + options.set("index-block-size", "28"); + let mut writer = + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &options).unwrap(); + for value in [ + Some(2), + Some(1), + None, + Some(3), + Some(2), + None, + Some(1), + Some(3), + ] { + let datum = value.map(Datum::Int); + writer.write(datum.as_ref()).unwrap(); + } + + let serialized = writer.serialized_bytes().unwrap(); + assert_eq!(serialized, bytes(JAVA_INT_MULTIPLE_BODIES_MULTIBLOCK_V2)); + let reader = BitmapFileIndexReader::try_new(data_type.clone(), serialized).unwrap(); + let blocks = match &reader.index { + BitmapIndex::V2(index) => &index.blocks, + BitmapIndex::V1(_) => panic!("expected V2 index"), + }; + assert_eq!(blocks.len(), 2); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::Eq, &[Datum::Int(1)]), + selection([1, 6]) + ); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::Eq, &[Datum::Int(2)]), + selection([0, 4]) + ); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::Eq, &[Datum::Int(3)]), + selection([3, 7]) + ); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::IsNull, &[]), + selection([2, 5]) + ); + } + + #[test] + fn test_v2_writer_variable_string_index_block_boundary() { + let data_type = DataType::VarChar(VarCharType::new(20).unwrap()); + let mut options = Options::new(); + options.set("index-block-size", "33"); + let mut writer = + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &options).unwrap(); + for value in ["bbbb", "a", "cc", "a"] { + writer + .write(Some(&Datum::String(value.to_string()))) + .unwrap(); + } + + let serialized = writer.serialized_bytes().unwrap(); + assert_eq!(serialized, bytes(JAVA_STRING_BLOCK_BOUNDARY_V2)); + let reader = BitmapFileIndexReader::try_new(data_type.clone(), serialized).unwrap(); + let blocks = match &reader.index { + BitmapIndex::V2(index) => &index.blocks, + BitmapIndex::V1(_) => panic!("expected V2 index"), + }; + // 4-byte header + 13-byte "a" entry + 16-byte "bbbb" entry. + assert_eq!(blocks.len(), 2); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::Eq, + &[Datum::String("a".to_string())] + ), + selection([1, 3]) + ); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::In, + &[ + Datum::String("bbbb".to_string()), + Datum::String("cc".to_string()) + ] + ), + selection([0, 2]) + ); + } + + #[test] + fn test_v2_writer_rejects_invalid_config_and_unsupported_type() { + let data_type = DataType::Int(IntType::new()); + + let mut options = Options::new(); + options.set("version", "2"); + options.set("index-block-size", "16 kb"); + assert!(writer::BitmapFileIndexWriter::try_new(data_type.clone(), &options).is_ok()); + + for version in ["invalid", "256"] { + let mut options = Options::new(); + options.set("version", version); + assert!(matches!( + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &options), + Err(Error::ConfigInvalid { .. }) + )); + } + for version in ["1", "3"] { + let mut options = Options::new(); + options.set("version", version); + assert!(matches!( + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &options), + Err(Error::Unsupported { .. }) + )); + } + for block_size in ["invalid", "0", "15", "9223372036854775807 tb"] { + let mut options = Options::new(); + options.set("index-block-size", block_size); + assert!(matches!( + writer::BitmapFileIndexWriter::try_new(data_type.clone(), &options), + Err(Error::ConfigInvalid { .. }) + )); + } + + let mut options = Options::new(); + options.set("index-block-size", "16"); + assert!(matches!( + writer::BitmapFileIndexWriter::try_new(DataType::BigInt(BigIntType::new()), &options), + Err(Error::ConfigInvalid { .. }) + )); + + assert!(matches!( + writer::BitmapFileIndexWriter::try_new( + DataType::Binary(BinaryType::new(4).unwrap()), + &Options::new() + ), + Err(Error::Unsupported { .. }) + )); + } + #[test] fn test_java_v1_v2_golden_payloads_and_predicates() { for fixture in fixtures() { diff --git a/crates/paimon/src/file_index/bitmap/writer.rs b/crates/paimon/src/file_index/bitmap/writer.rs new file mode 100644 index 00000000..cc63573e --- /dev/null +++ b/crates/paimon/src/file_index/bitmap/writer.rs @@ -0,0 +1,344 @@ +// 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 std::collections::HashMap; +use std::ops::Range; + +use bytes::{BufMut, Bytes, BytesMut}; +use roaring::RoaringBitmap; + +use crate::common::options::parse_memory_size; +use crate::common::Options; +use crate::spec::{DataType, Datum}; +use crate::{Error, Result}; + +use super::{format_invalid, BitmapValue, BitmapValueCodec, VERSION_2}; + +const VERSION_OPTION: &str = "version"; +const INDEX_BLOCK_SIZE_OPTION: &str = "index-block-size"; +const DEFAULT_INDEX_BLOCK_SIZE: &str = "16kb"; +const INDEX_BLOCK_HEADER_SIZE: usize = 4; +const INDEX_ENTRY_FIXED_SIZE: usize = 8; + +/// Writer for Java-compatible Bitmap V2 payloads. +pub(crate) struct BitmapFileIndexWriter { + codec: BitmapValueCodec, + index_block_size: usize, + row_count: u32, + null_bitmap: RoaringBitmap, + bitmaps: HashMap, +} + +impl BitmapFileIndexWriter { + pub(crate) fn try_new(data_type: DataType, options: &Options) -> Result { + let codec = BitmapValueCodec::try_new(&data_type)?; + validate_version(options)?; + let index_block_size = parse_index_block_size(options, codec)?; + Ok(Self { + codec, + index_block_size, + row_count: 0, + null_bitmap: RoaringBitmap::new(), + bitmaps: HashMap::new(), + }) + } + + pub(crate) fn write(&mut self, datum: Option<&Datum>) -> Result<()> { + if self.row_count == i32::MAX as u32 { + return Err(Error::DataInvalid { + message: "Bitmap row count exceeds i32::MAX".to_string(), + source: None, + }); + } + + let value = datum.map(|datum| self.codec.value(datum)).transpose()?; + match value { + Some(value) => { + self.bitmaps + .entry(value) + .or_default() + .insert(self.row_count); + } + None => { + self.null_bitmap.insert(self.row_count); + } + } + self.row_count += 1; + Ok(()) + } + + pub(crate) fn serialized_bytes(&mut self) -> Result { + let null_bytes = serialize_bitmap(&mut self.null_bitmap)?; + let mut body = Vec::new(); + let null_entry = if self.null_bitmap.is_empty() { + None + } else if self.null_bitmap.len() == 1 { + Some(( + singleton_offset(self.null_bitmap.min().unwrap())?, + usize_to_i32(null_bytes.len(), "null bitmap length")?, + )) + } else { + let length = usize_to_i32(null_bytes.len(), "null bitmap length")?; + body.extend_from_slice(&null_bytes); + Some((0, length)) + }; + + let mut bitmaps = self.bitmaps.iter_mut().collect::>(); + bitmaps.sort_unstable_by_key(|(key, _)| *key); + + let mut entries = Vec::with_capacity(bitmaps.len()); + for (key, bitmap) in bitmaps { + let (offset, length) = if bitmap.len() == 1 { + (singleton_offset(bitmap.min().unwrap())?, -1) + } else { + let serialized = serialize_bitmap(bitmap)?; + let offset = usize_to_i32(body.len(), "bitmap body offset")?; + let length = usize_to_i32(serialized.len(), "serialized bitmap length")?; + body.extend_from_slice(&serialized); + (offset, length) + }; + entries.push(SerializedEntry { + key, + offset, + length, + }); + } + usize_to_i32(body.len(), "bitmap body length")?; + + let blocks = build_index_blocks(&entries, self.index_block_size)?; + let mut block_offsets = Vec::with_capacity(blocks.len()); + let mut index_area_length = 0usize; + for block in &blocks { + block_offsets.push(index_area_length); + index_area_length = index_area_length + .checked_add(block_serialized_size(&entries[block.clone()])?) + .ok_or_else(|| format_invalid("Bitmap index area length overflow"))?; + } + + let mut output = BytesMut::new(); + output.put_u8(VERSION_2); + output.put_i32(i32::try_from(self.row_count).map_err(|_| { + format_invalid(format!( + "Bitmap row count exceeds i32::MAX: {}", + self.row_count + )) + })?); + output.put_i32(usize_to_i32(entries.len(), "non-null bitmap count")?); + output.put_u8(u8::from(null_entry.is_some())); + if let Some((offset, length)) = null_entry { + output.put_i32(offset); + output.put_i32(length); + } + + output.put_i32(usize_to_i32(blocks.len(), "bitmap index block count")?); + for (block, offset) in blocks.iter().zip(block_offsets) { + write_value(&mut output, entries[block.start].key)?; + output.put_i32(usize_to_i32(offset, "bitmap index block offset")?); + } + output.put_i32(usize_to_i32(index_area_length, "bitmap index area length")?); + + for block in blocks { + output.put_i32(usize_to_i32(block.len(), "bitmap index block entry count")?); + for entry in &entries[block] { + write_value(&mut output, entry.key)?; + output.put_i32(entry.offset); + output.put_i32(entry.length); + } + } + output.extend_from_slice(&body); + Ok(output.freeze()) + } +} + +struct SerializedEntry<'a> { + key: &'a BitmapValue, + offset: i32, + length: i32, +} + +fn validate_version(options: &Options) -> Result<()> { + let Some(raw) = options.get(VERSION_OPTION) else { + return Ok(()); + }; + let version = raw.parse::().map_err(|error| Error::ConfigInvalid { + message: format!("Invalid Bitmap option {VERSION_OPTION}={raw}: {error}"), + })?; + if version != VERSION_2 { + return Err(Error::Unsupported { + message: format!( + "Bitmap writer only supports version {VERSION_2}, but found {version}" + ), + }); + } + Ok(()) +} + +fn parse_index_block_size(options: &Options, codec: BitmapValueCodec) -> Result { + let raw = options + .get(INDEX_BLOCK_SIZE_OPTION) + .map(String::as_str) + .unwrap_or(DEFAULT_INDEX_BLOCK_SIZE); + let size = parse_memory_size(raw).map_err(|error| Error::ConfigInvalid { + message: format!("Invalid Bitmap option {INDEX_BLOCK_SIZE_OPTION}={raw}: {error:?}"), + })?; + let size = usize::try_from(size).map_err(|_| Error::ConfigInvalid { + message: format!("Invalid Bitmap option {INDEX_BLOCK_SIZE_OPTION}={raw}: out of range"), + })?; + let minimum = INDEX_BLOCK_HEADER_SIZE + INDEX_ENTRY_FIXED_SIZE + minimum_value_size(codec); + if size < minimum { + return Err(Error::ConfigInvalid { + message: format!( + "Bitmap option {INDEX_BLOCK_SIZE_OPTION} must be at least {minimum} bytes for {codec:?}, but was {size}" + ), + }); + } + Ok(size) +} + +fn minimum_value_size(codec: BitmapValueCodec) -> usize { + match codec { + BitmapValueCodec::Boolean | BitmapValueCodec::TinyInt => 1, + BitmapValueCodec::SmallInt => 2, + BitmapValueCodec::Int + | BitmapValueCodec::Float + | BitmapValueCodec::Date + | BitmapValueCodec::Time + | BitmapValueCodec::String => 4, + BitmapValueCodec::BigInt + | BitmapValueCodec::Double + | BitmapValueCodec::TimestampMillis + | BitmapValueCodec::TimestampMicros + | BitmapValueCodec::LocalZonedTimestampMillis + | BitmapValueCodec::LocalZonedTimestampMicros => 8, + } +} + +fn singleton_offset(position: u32) -> Result { + let position = i32::try_from(position) + .map_err(|_| format_invalid(format!("Bitmap row position exceeds i32::MAX: {position}")))?; + (-1_i32) + .checked_sub(position) + .ok_or_else(|| format_invalid(format!("Bitmap singleton offset overflow: {position}"))) +} + +fn serialize_bitmap(bitmap: &mut RoaringBitmap) -> Result> { + bitmap.optimize(); + let mut serialized = Vec::with_capacity(bitmap.serialized_size()); + bitmap + .serialize_into(&mut serialized) + .map_err(|error| Error::UnexpectedError { + message: "Failed to serialize Bitmap RoaringBitmap32".to_string(), + source: Some(Box::new(error)), + })?; + Ok(serialized) +} + +fn build_index_blocks( + entries: &[SerializedEntry<'_>], + block_size_limit: usize, +) -> Result>> { + let mut blocks = Vec::new(); + let mut block_start = 0usize; + let mut block_size = INDEX_BLOCK_HEADER_SIZE; + + for (index, entry) in entries.iter().enumerate() { + let entry_size = INDEX_ENTRY_FIXED_SIZE + .checked_add(value_serialized_size(entry.key)?) + .ok_or_else(|| format_invalid("Bitmap index entry size overflow"))?; + let minimum_block_size = INDEX_BLOCK_HEADER_SIZE + .checked_add(entry_size) + .ok_or_else(|| format_invalid("Bitmap index block size overflow"))?; + if minimum_block_size > block_size_limit { + return Err(Error::ConfigInvalid { + message: format!( + "Bitmap option {INDEX_BLOCK_SIZE_OPTION}={block_size_limit} bytes cannot fit a {minimum_block_size}-byte index block" + ), + }); + } + if block_size + .checked_add(entry_size) + .is_none_or(|size| size > block_size_limit) + { + blocks.push(block_start..index); + block_start = index; + block_size = INDEX_BLOCK_HEADER_SIZE; + } + block_size = block_size + .checked_add(entry_size) + .ok_or_else(|| format_invalid("Bitmap index block size overflow"))?; + } + + if block_start < entries.len() { + blocks.push(block_start..entries.len()); + } + Ok(blocks) +} + +fn block_serialized_size(entries: &[SerializedEntry<'_>]) -> Result { + let mut size = INDEX_BLOCK_HEADER_SIZE; + for entry in entries { + let value_size = value_serialized_size(entry.key)?; + size = size + .checked_add(INDEX_ENTRY_FIXED_SIZE) + .and_then(|size| size.checked_add(value_size)) + .ok_or_else(|| format_invalid("Bitmap index block size overflow"))?; + } + Ok(size) +} + +fn value_serialized_size(value: &BitmapValue) -> Result { + match value { + BitmapValue::Boolean(_) | BitmapValue::TinyInt(_) => Ok(1), + BitmapValue::SmallInt(_) => Ok(2), + BitmapValue::Int(_) + | BitmapValue::Float(_) + | BitmapValue::Date(_) + | BitmapValue::Time(_) => Ok(4), + BitmapValue::BigInt(_) + | BitmapValue::Double(_) + | BitmapValue::Timestamp(_) + | BitmapValue::LocalZonedTimestamp(_) => Ok(8), + BitmapValue::String(value) => 4usize + .checked_add(value.len()) + .ok_or_else(|| format_invalid("Bitmap string value size overflow")), + } +} + +fn write_value(output: &mut BytesMut, value: &BitmapValue) -> Result<()> { + match value { + BitmapValue::Boolean(value) => output.put_u8(u8::from(*value)), + BitmapValue::TinyInt(value) => output.put_i8(*value), + BitmapValue::SmallInt(value) => output.put_i16(*value), + BitmapValue::Int(value) | BitmapValue::Date(value) | BitmapValue::Time(value) => { + output.put_i32(*value) + } + BitmapValue::BigInt(value) + | BitmapValue::Timestamp(value) + | BitmapValue::LocalZonedTimestamp(value) => output.put_i64(*value), + BitmapValue::Float(value) => output.put_u32(value.0), + BitmapValue::Double(value) => output.put_u64(value.0), + BitmapValue::String(value) => { + output.put_i32(usize_to_i32(value.len(), "Bitmap string value length")?); + output.extend_from_slice(value.as_bytes()); + } + } + Ok(()) +} + +fn usize_to_i32(value: usize, field: &str) -> Result { + i32::try_from(value).map_err(|_| format_invalid(format!("{field} exceeds i32::MAX: {value}"))) +}