From 92b8224a00c6bae897fec1d6ea04882854f94b90 Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Sat, 19 Sep 2026 21:01:45 +0800 Subject: [PATCH] feat(file_index): support range bitmap append writes Generate Java-compatible Range Bitmap V1 indexes for configured append columns instead of ignoring range-bitmap options. Reuse the existing embedded and sidecar index paths and validate per-column chunk sizes. Cover Java payload compatibility, range pruning, and append lifecycle. Signed-off-by: QuakeWang --- .../src/file_index/file_indexer_factory.rs | 22 +- crates/paimon/src/file_index/range_bitmap.rs | 60 ++- .../src/file_index/range_bitmap/writer.rs | 255 ++++++++++++ .../writer/fixtures/GenerateRangeBitmap.java | 66 +++ .../range_bitmap/writer/fixtures/README.md | 39 ++ .../file_index/range_bitmap/writer/tests.rs | 389 ++++++++++++++++++ .../src/table/data_file_index_writer.rs | 4 +- .../src/table/data_file_index_writer/tests.rs | 193 ++++++--- docs/src/getting-started.md | 18 +- 9 files changed, 979 insertions(+), 67 deletions(-) create mode 100644 crates/paimon/src/file_index/range_bitmap/writer.rs create mode 100644 crates/paimon/src/file_index/range_bitmap/writer/fixtures/GenerateRangeBitmap.java create mode 100644 crates/paimon/src/file_index/range_bitmap/writer/fixtures/README.md create mode 100644 crates/paimon/src/file_index/range_bitmap/writer/tests.rs diff --git a/crates/paimon/src/file_index/file_indexer_factory.rs b/crates/paimon/src/file_index/file_indexer_factory.rs index e7404758f..298225c3f 100644 --- a/crates/paimon/src/file_index/file_indexer_factory.rs +++ b/crates/paimon/src/file_index/file_indexer_factory.rs @@ -23,6 +23,7 @@ use crate::file_index::bitmap::BitmapFileIndexReader; use crate::file_index::bloom_filter::{BloomFilterReader, BloomFilterWriter}; use crate::file_index::file_index_reader::FileIndexReader; use crate::file_index::file_index_writer::FileIndexWriter; +use crate::file_index::range_bitmap::writer::RangeBitmapFileIndexWriter; use crate::file_index::range_bitmap::RangeBitmapFileIndexReader; use crate::spec::DataType; use crate::{Error, Result}; @@ -69,7 +70,10 @@ impl FileIndexerFactory { /// Reader support does not imply that the index can be generated. pub(crate) fn is_write_supported(identifier: &str) -> bool { - matches!(identifier, BITMAP_INDEX | BLOOM_FILTER_INDEX) + matches!( + identifier, + BITMAP_INDEX | BLOOM_FILTER_INDEX | RANGE_BITMAP_INDEX + ) } pub(crate) fn create_writer( @@ -84,9 +88,9 @@ impl FileIndexerFactory { BuiltinFileIndexer::BloomFilter => { Ok(Box::new(BloomFilterWriter::try_new(data_type, options)?)) } - BuiltinFileIndexer::RangeBitmap => Err(Error::Unsupported { - message: "Writing range-bitmap indexes is not supported yet".to_string(), - }), + BuiltinFileIndexer::RangeBitmap => Ok(Box::new(RangeBitmapFileIndexWriter::try_new( + data_type, options, + )?)), } } @@ -133,7 +137,7 @@ mod tests { #[test] fn test_builtin_writers_track_empty_rows_consistently() { - for identifier in [BITMAP_INDEX, BLOOM_FILTER_INDEX] { + for identifier in [BITMAP_INDEX, BLOOM_FILTER_INDEX, RANGE_BITMAP_INDEX] { assert!(FileIndexerFactory::is_write_supported(identifier)); let mut writer = FileIndexerFactory::create_writer(identifier, int_type(), &Options::new()).unwrap(); @@ -185,12 +189,8 @@ mod tests { #[test] fn test_unknown_identifier_is_rejected() { assert!(FileIndexerFactory::is_supported(RANGE_BITMAP_INDEX)); - assert!(!FileIndexerFactory::is_write_supported(RANGE_BITMAP_INDEX)); + assert!(FileIndexerFactory::is_write_supported(RANGE_BITMAP_INDEX)); assert!(!FileIndexerFactory::is_write_supported("unknown")); - assert!(matches!( - FileIndexerFactory::create_writer(RANGE_BITMAP_INDEX, int_type(), &Options::new()), - Err(Error::Unsupported { .. }) - )); assert!(matches!( FileIndexerFactory::create_writer("unknown", int_type(), &Options::new()), Err(Error::Unsupported { .. }) @@ -203,7 +203,7 @@ mod tests { #[test] fn test_writer_rejects_mismatched_datum() { - for identifier in [BITMAP_INDEX, BLOOM_FILTER_INDEX] { + for identifier in [BITMAP_INDEX, BLOOM_FILTER_INDEX, RANGE_BITMAP_INDEX] { let mut writer = FileIndexerFactory::create_writer(identifier, int_type(), &Options::new()).unwrap(); diff --git a/crates/paimon/src/file_index/range_bitmap.rs b/crates/paimon/src/file_index/range_bitmap.rs index c488640e7..b9ecf78f1 100644 --- a/crates/paimon/src/file_index/range_bitmap.rs +++ b/crates/paimon/src/file_index/range_bitmap.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -//! Reader for Java Paimon's `range-bitmap` file index. +//! Java-compatible `range-bitmap` file index. //! //! The index maps ordered dictionary codes to row positions with a bit-sliced //! bitmap. Evaluating it produces the same conservative row selection consumed @@ -33,6 +33,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 JAVA_CANONICAL_FLOAT_NAN_BITS: u32 = 0x7fc0_0000; const JAVA_CANONICAL_DOUBLE_NAN_BITS: u64 = 0x7ff8_0000_0000_0000; @@ -1054,6 +1056,62 @@ mod tests { RangeBitmapFileIndexReader::try_new(int_type(), bytes).unwrap() } + #[test] + fn test_writer_matches_java_v1_bytes() { + use crate::common::Options; + use crate::file_index::file_index_writer::FileIndexWriter; + use writer::RangeBitmapFileIndexWriter; + + let cases = [ + ( + int_type(), + vec![ + Some(Datum::Int(1)), + Some(Datum::Int(3)), + Some(Datum::Int(5)), + Some(Datum::Int(7)), + Some(Datum::Int(9)), + None, + None, + Some(Datum::Int(10)), + ], + JAVA_INT_V1, + ), + ( + DataType::VarChar(crate::spec::VarCharType::new(32).unwrap()), + vec![ + Some(Datum::String("aa".into())), + Some(Datum::String("b".into())), + Some(Datum::String("你好".into())), + None, + Some(Datum::String("ccc".into())), + ], + JAVA_STRING_V1, + ), + ( + DataType::Float(crate::spec::FloatType::new()), + vec![ + Some(Datum::Float(-0.0)), + Some(Datum::Float(0.0)), + Some(Datum::Float(1.5)), + Some(Datum::Float(f32::NAN)), + None, + ], + JAVA_FLOAT_V1, + ), + ]; + for (data_type, values, golden) in cases { + let mut writer = + RangeBitmapFileIndexWriter::try_new(data_type, &Options::new()).unwrap(); + for value in &values { + writer.write(value.as_ref()).unwrap(); + } + let bytes = writer.serialized_bytes().unwrap(); + assert_eq!(hex::encode(&bytes), golden); + assert_eq!(writer.serialized_bytes().unwrap(), bytes); + } + } + fn selection(rows: impl IntoIterator) -> FileIndexResult { FileIndexResult::Selection(rows.into_iter().collect()) } diff --git a/crates/paimon/src/file_index/range_bitmap/writer.rs b/crates/paimon/src/file_index/range_bitmap/writer.rs new file mode 100644 index 000000000..c608fdc85 --- /dev/null +++ b/crates/paimon/src/file_index/range_bitmap/writer.rs @@ -0,0 +1,255 @@ +// 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::BTreeMap; + +use bytes::{BufMut, Bytes}; +use roaring::RoaringBitmap; + +use super::{RangeValue, RangeValueCodec, VERSION_1}; +use crate::common::options::parse_memory_size; +use crate::common::Options; +use crate::file_index::file_index_writer::FileIndexWriter; +use crate::spec::{DataType, Datum}; +use crate::{Error, Result}; + +/// Java V1 writer for boolean, numeric, string, date and time values. +/// Decimal precision is limited to 18 and timestamp precision to 6. +pub(crate) struct RangeBitmapFileIndexWriter { + codec: RangeValueCodec, + chunk_size: usize, + row_count: u32, + bitmaps: BTreeMap, +} + +impl RangeBitmapFileIndexWriter { + pub(crate) fn try_new(data_type: DataType, options: &Options) -> Result { + let codec = RangeValueCodec::try_new(&data_type)?; + let default = match codec { + RangeValueCodec::Boolean | RangeValueCodec::TinyInt | RangeValueCodec::SmallInt => "0b", + _ => "16kb", + }; + let raw = options + .get("chunk-size") + .map(String::as_str) + .unwrap_or(default); + let size = parse_memory_size(raw).map_err(|error| Error::ConfigInvalid { + message: format!("Invalid range-bitmap chunk-size '{raw}': {error:?}"), + })?; + if !(0..=i64::from(i32::MAX)).contains(&size) { + return Err(Error::ConfigInvalid { + message: "Range-bitmap chunk-size must be between 0 and 2147483647 bytes" + .to_string(), + }); + } + Ok(Self { + codec, + chunk_size: size as usize, + row_count: 0, + bitmaps: BTreeMap::new(), + }) + } + + fn dictionary(&self) -> Result> { + let mut offsets = Vec::new(); + let mut chunks = Vec::new(); + let mut keys = Vec::new(); + let mut values = self.bitmaps.keys().enumerate().peekable(); + while let Some((code, first)) = values.next() { + put_count(&mut offsets, chunks.len())?; + chunks.put_u8(VERSION_1); + write_value(&mut chunks, first)?; + put_count(&mut chunks, code)?; + put_count(&mut chunks, keys.len())?; + + // Java stores the first key in the chunk header, outside its size budget. + let mut chunk_keys = Vec::new(); + let mut key_offsets = Vec::new(); + let mut count = 0; + while let Some((_, value)) = values.peek() { + let mut encoded = Vec::new(); + write_value(&mut encoded, value)?; + if encoded.len() > self.chunk_size - chunk_keys.len() + || (self.codec.fixed_length().is_none() + && key_offsets.len() + 4 > self.chunk_size) + { + break; + } + if self.codec.fixed_length().is_none() { + put_count(&mut key_offsets, chunk_keys.len())?; + } + chunk_keys.extend_from_slice(&encoded); + count += 1; + values.next(); + } + put_count(&mut chunks, count)?; + if let Some(length) = self.codec.fixed_length() { + put_count(&mut chunks, chunk_keys.len())?; + put_count(&mut chunks, length)?; + } else { + put_count(&mut chunks, key_offsets.len())?; + put_count(&mut chunks, chunk_keys.len())?; + } + keys.extend_from_slice(&key_offsets); + keys.extend_from_slice(&chunk_keys); + checked_count(keys.len())?; + } + let mut result = Vec::new(); + result.put_i32(13); + result.put_u8(VERSION_1); + put_count(&mut result, offsets.len() / 4)?; + put_count(&mut result, offsets.len())?; + put_count(&mut result, chunks.len())?; + result.extend_from_slice(&offsets); + result.extend_from_slice(&chunks); + result.extend_from_slice(&keys); + checked_count(result.len())?; + Ok(result) + } + + fn bsi(&self) -> Result> { + // Java sign-extends cardinality - 1 to a long: an empty dictionary has 64 slices. + let width = if self.bitmaps.is_empty() { + 64 + } else { + (usize::BITS - (self.bitmaps.len() - 1).leading_zeros()).max(1) as usize + }; + let mut existing = RoaringBitmap::new(); + let mut slices = vec![RoaringBitmap::new(); width]; + for (code, rows) in self.bitmaps.values().enumerate() { + existing |= rows; + let mut bits = code; + while bits != 0 { + slices[bits.trailing_zeros() as usize] |= rows; + bits &= bits - 1; + } + } + let existing = serialize_bitmap(existing)?; + let mut indexes = Vec::new(); + let mut body = Vec::new(); + for slice in slices { + let bytes = serialize_bitmap(slice)?; + put_count(&mut indexes, body.len())?; + put_count(&mut indexes, bytes.len())?; + body.extend_from_slice(&bytes); + } + let mut result = Vec::new(); + put_count(&mut result, 10 + indexes.len())?; + result.put_u8(VERSION_1); + result.put_u8(width as u8); + put_count(&mut result, existing.len())?; + put_count(&mut result, indexes.len())?; + result.extend_from_slice(&indexes); + result.extend_from_slice(&existing); + result.extend_from_slice(&body); + checked_count(result.len())?; + Ok(result) + } +} + +impl FileIndexWriter for RangeBitmapFileIndexWriter { + fn write(&mut self, datum: Option<&Datum>) -> Result<()> { + if self.row_count == i32::MAX as u32 { + return Err(Error::DataInvalid { + message: "Range-bitmap row count exceeds i32::MAX".to_string(), + source: None, + }); + } + if let Some(value) = datum.map(|datum| self.codec.value(datum)).transpose()? { + self.bitmaps + .entry(value) + .or_default() + .insert(self.row_count); + } + self.row_count += 1; + Ok(()) + } + + fn serialized_bytes(&mut self) -> Result { + let dictionary = self.dictionary()?; + let bsi = self.bsi()?; + let mut header = Vec::new(); + header.put_u8(VERSION_1); + header.put_u32(self.row_count); + put_count(&mut header, self.bitmaps.len())?; + if let Some((min, _)) = self.bitmaps.first_key_value() { + write_value(&mut header, min)?; + write_value(&mut header, self.bitmaps.last_key_value().unwrap().0)?; + } + put_count(&mut header, dictionary.len())?; + let mut result = Vec::new(); + put_count(&mut result, header.len())?; + result.extend_from_slice(&header); + result.extend_from_slice(&dictionary); + result.extend_from_slice(&bsi); + checked_count(result.len())?; + Ok(Bytes::from(result)) + } + + fn empty(&self) -> bool { + self.row_count == 0 + } +} + +fn checked_count(value: usize) -> Result { + i32::try_from(value).map_err(|_| Error::DataInvalid { + message: "Range-bitmap size exceeds i32::MAX".to_string(), + source: None, + }) +} + +fn put_count(output: &mut Vec, value: usize) -> Result<()> { + output.put_i32(checked_count(value)?); + Ok(()) +} + +fn write_value(output: &mut Vec, value: &RangeValue) -> Result<()> { + match value { + RangeValue::Boolean(value) => output.put_u8(u8::from(*value)), + RangeValue::TinyInt(value) => output.put_i8(*value), + RangeValue::SmallInt(value) => output.put_i16(*value), + RangeValue::Int(value) | RangeValue::Date(value) | RangeValue::Time(value) => { + output.put_i32(*value) + } + RangeValue::BigInt(value) + | RangeValue::Decimal(value) + | RangeValue::Timestamp(value) + | RangeValue::LocalZonedTimestamp(value) => output.put_i64(*value), + RangeValue::Float(value) => output.put_u32(value.0), + RangeValue::Double(value) => output.put_u64(value.0), + RangeValue::String(value) => { + put_count(output, value.len())?; + output.extend_from_slice(value.as_bytes()); + } + } + Ok(()) +} + +fn serialize_bitmap(mut bitmap: RoaringBitmap) -> Result> { + bitmap.optimize(); + let mut bytes = Vec::with_capacity(bitmap.serialized_size()); + bitmap + .serialize_into(&mut bytes) + .map_err(|error| Error::DataInvalid { + message: format!("Failed to serialize range-bitmap: {error}"), + source: None, + })?; + Ok(bytes) +} + +#[cfg(test)] +mod tests; diff --git a/crates/paimon/src/file_index/range_bitmap/writer/fixtures/GenerateRangeBitmap.java b/crates/paimon/src/file_index/range_bitmap/writer/fixtures/GenerateRangeBitmap.java new file mode 100644 index 000000000..c22a5863a --- /dev/null +++ b/crates/paimon/src/file_index/range_bitmap/writer/fixtures/GenerateRangeBitmap.java @@ -0,0 +1,66 @@ +/* + * 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. + */ + +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.fileindex.FileIndexWriter; +import org.apache.paimon.fileindex.rangebitmap.RangeBitmapFileIndex; +import org.apache.paimon.options.Options; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.IntType; +import org.apache.paimon.types.VarCharType; + +import java.security.MessageDigest; + +/** Generates the SHA-256 golden payloads used by the Rust writer tests. */ +public class GenerateRangeBitmap { + private static void emit(String name, DataType type, String chunkSize, Object[] values) + throws Exception { + Options options = new Options(); + options.setString("chunk-size", chunkSize); + FileIndexWriter writer = new RangeBitmapFileIndex(type, options).createWriter(); + for (Object value : values) { + writer.writeRecord(value); + } + byte[] digest = MessageDigest.getInstance("SHA-256").digest(writer.serializedBytes()); + StringBuilder hex = new StringBuilder(); + for (byte b : digest) { + hex.append(String.format("%02x", b & 0xff)); + } + System.out.println(name + " " + hex); + } + + public static void main(String[] args) throws Exception { + emit("empty", new IntType(), "0b", new Object[] {}); + emit("nulls", new IntType(), "0b", new Object[] {null, null, null}); + emit("singleton", new IntType(), "0b", new Object[] {7, null, 7}); + Object[] ints = {9, -1, 3, null, 1, 7, 5, 3, Integer.MIN_VALUE, Integer.MAX_VALUE}; + emit("int-zero", new IntType(), "0b", ints); + emit("int-chunks", new IntType(), "8b", ints); + Object[] strings = {BinaryString.fromString("z"), BinaryString.fromString(""), + BinaryString.fromString("a\u0000"), null, BinaryString.fromString("\u4f60\u597d"), + BinaryString.fromString("ab"), BinaryString.fromString("abc"), + BinaryString.fromString("\ud83e\udd80")}; + emit("string-chunks", new VarCharType(), "8b", strings); + Object[] containers = new Object[70000]; + for (int i = 0; i < containers.length; i++) { + containers[i] = i % 11 == 0 ? null : (i * 37) % 101; + } + emit("containers", new IntType(), "12b", containers); + } +} diff --git a/crates/paimon/src/file_index/range_bitmap/writer/fixtures/README.md b/crates/paimon/src/file_index/range_bitmap/writer/fixtures/README.md new file mode 100644 index 000000000..3b4412599 --- /dev/null +++ b/crates/paimon/src/file_index/range_bitmap/writer/fixtures/README.md @@ -0,0 +1,39 @@ + + +# Range Bitmap Java fixtures + +`GenerateRangeBitmap.java` generates the full-payload SHA-256 digests in +`../tests.rs`. The reference is Apache Paimon commit `1d368b4a5`, JDK 8, +and RoaringBitmap 1.2.1. Cases cover empty/all-null/singleton indexes, +zero-sized chunks, fixed/variable dictionary boundaries, and multiple +Roaring containers. + +Set `PAIMON_CLASSPATH` to the reference checkout's compiled `paimon-common` +and `paimon-api` classes plus RoaringBitmap 1.2.1, jsr305 3.0.2 and slf4j-api +jars. From this directory: + +```sh +fixture_classes=$(mktemp -d) +javac -cp "$PAIMON_CLASSPATH" -d "$fixture_classes" GenerateRangeBitmap.java +java -cp "$fixture_classes:$PAIMON_CLASSPATH" GenerateRangeBitmap +``` + +The Rust tests compare exact bytes for the small INT/STRING/FLOAT samples +in `range_bitmap.rs`; digests keep the larger boundary fixtures compact. diff --git a/crates/paimon/src/file_index/range_bitmap/writer/tests.rs b/crates/paimon/src/file_index/range_bitmap/writer/tests.rs new file mode 100644 index 000000000..2209b01c7 --- /dev/null +++ b/crates/paimon/src/file_index/range_bitmap/writer/tests.rs @@ -0,0 +1,389 @@ +// 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 super::*; +use crate::file_index::file_index_result::FileIndexResult; +use crate::file_index::range_bitmap::RangeBitmapFileIndexReader; +use crate::spec::PredicateOperator; + +fn selection(rows: impl IntoIterator) -> FileIndexResult { + FileIndexResult::Selection(rows.into_iter().collect()) +} + +fn data_type(name: &str) -> DataType { + serde_json::from_value(serde_json::Value::String(name.to_string())).unwrap() +} + +#[test] +fn java_chunk_and_container_golden_payloads() { + use sha2::{Digest, Sha256}; + + // Generated by fixtures/GenerateRangeBitmap.java against Java Paimon 1d368b4a5 + // with RoaringBitmap 1.2.1. Hash the entire payload, including all headers. + let ints = vec![ + Some(9), + Some(-1), + Some(3), + None, + Some(1), + Some(7), + Some(5), + Some(3), + Some(i32::MIN), + Some(i32::MAX), + ] + .into_iter() + .map(|v| v.map(Datum::Int)) + .collect::>(); + let strings = [ + Some("z"), + Some(""), + Some("a\0"), + None, + Some("你好"), + Some("ab"), + Some("abc"), + Some("🦀"), + ] + .into_iter() + .map(|v| v.map(|v| Datum::String(v.into()))) + .collect(); + let containers = (0..70_000) + .map(|i| { + if i % 11 == 0 { + None + } else { + Some(Datum::Int((i * 37) % 101)) + } + }) + .collect(); + for (name, size, values, digest) in [ + ( + "INT", + "0b", + vec![], + "930b0b08191aa151e382586494360c5408fc7d4e2fd6971e0637004e2099e017", + ), + ( + "INT", + "0b", + vec![None; 3], + "ca2a1588451c1c3c353abd33b7f447721bf61f65ce06c8d004646821c4123edd", + ), + ( + "INT", + "0b", + vec![Some(Datum::Int(7)), None, Some(Datum::Int(7))], + "69007c0772b87a68420a8ed271849a45072e7868c1ba2fe972d3e7a4c9be850f", + ), + ( + "INT", + "0b", + ints.clone(), + "82999022f0517a867a0d2a109467de891c0d11a41f20e207c8502d7b31414e21", + ), + ( + "INT", + "8b", + ints, + "b5b2b34a69e6c25ca7cf8bbb799df142b44d1c9011840fa2862fef996f2c4892", + ), + ( + "VARCHAR(100)", + "8b", + strings, + "11cc7028788d995ca15b8d94bdd7c91dd9ba1110a941cfce7b8412bcfc7b3f5f", + ), + ( + "INT", + "12b", + containers, + "668a624b4c77633891f7280529ce3344149a9efb6ac2509b78c75ceca3bfe59e", + ), + ] { + let mut options = Options::new(); + options.set("chunk-size", size); + let mut writer = RangeBitmapFileIndexWriter::try_new(data_type(name), &options).unwrap(); + for value in values { + writer.write(value.as_ref()).unwrap(); + } + assert_eq!( + hex::encode(Sha256::digest(writer.serialized_bytes().unwrap())), + digest, + "{name}, {size}" + ); + } +} + +#[test] +fn supported_types_round_trip_across_chunks() { + let cases = [ + ("BOOLEAN", Datum::Bool(false), Datum::Bool(true)), + ("TINYINT", Datum::TinyInt(i8::MIN), Datum::TinyInt(i8::MAX)), + ( + "SMALLINT", + Datum::SmallInt(i16::MIN), + Datum::SmallInt(i16::MAX), + ), + ("INT", Datum::Int(i32::MIN), Datum::Int(i32::MAX)), + ("BIGINT", Datum::Long(i64::MIN), Datum::Long(i64::MAX)), + ( + "FLOAT", + Datum::Float(f32::NEG_INFINITY), + Datum::Float(f32::INFINITY), + ), + ( + "DOUBLE", + Datum::Double(f64::NEG_INFINITY), + Datum::Double(f64::INFINITY), + ), + ( + "DECIMAL(18, 2)", + Datum::Decimal { + unscaled: -999999999999999999, + precision: 18, + scale: 2, + }, + Datum::Decimal { + unscaled: 999999999999999999, + precision: 18, + scale: 2, + }, + ), + ("DATE", Datum::Date(-1), Datum::Date(1)), + ("TIME(3)", Datum::Time(0), Datum::Time(86399999)), + ( + "TIMESTAMP(3)", + Datum::Timestamp { + millis: -1, + nanos: 0, + }, + Datum::Timestamp { + millis: 1, + nanos: 0, + }, + ), + ( + "TIMESTAMP(6)", + Datum::Timestamp { + millis: -1, + nanos: 999000, + }, + Datum::Timestamp { + millis: 0, + nanos: 1000, + }, + ), + ( + "TIMESTAMP_LTZ(3)", + Datum::LocalZonedTimestamp { + millis: -1, + nanos: 0, + }, + Datum::LocalZonedTimestamp { + millis: 1, + nanos: 0, + }, + ), + ( + "TIMESTAMP_LTZ(6)", + Datum::LocalZonedTimestamp { + millis: -1, + nanos: 999000, + }, + Datum::LocalZonedTimestamp { + millis: 0, + nanos: 1000, + }, + ), + ( + "CHAR(10)", + Datum::String("".into()), + Datum::String("你好".into()), + ), + ( + "VARCHAR(100)", + Datum::String("a\0".into()), + Datum::String("🦀".into()), + ), + ]; + for (name, low, high) in cases { + let data_type = data_type(name); + for size in ["0b", "1b", "8b", "16kb"] { + let mut options = Options::new(); + options.set("chunk-size", size); + let mut writer = + RangeBitmapFileIndexWriter::try_new(data_type.clone(), &options).unwrap(); + for value in [Some(&high), None, Some(&low), Some(&high)] { + writer.write(value).unwrap(); + } + let bytes = writer.serialized_bytes().unwrap(); + assert_eq!(bytes, writer.serialized_bytes().unwrap()); + let reader = RangeBitmapFileIndexReader::try_new(data_type.clone(), bytes).unwrap(); + for (op, literals, expected) in [ + (PredicateOperator::Eq, vec![high.clone()], selection([0, 3])), + (PredicateOperator::Lt, vec![high.clone()], selection([2])), + (PredicateOperator::Gt, vec![low.clone()], selection([0, 3])), + ( + PredicateOperator::Between, + vec![low.clone(), high.clone()], + selection([0, 2, 3]), + ), + (PredicateOperator::IsNull, vec![], selection([1])), + ] { + assert_eq!( + reader.try_evaluate(&data_type, op, &literals).unwrap(), + expected, + "{name}, {size}, {op:?}" + ); + } + } + } +} + +#[test] +fn ranges_match_unindexed_values_across_bitmap_containers() { + use rand::{rngs::StdRng, Rng, SeedableRng}; + let data_type = data_type("INT"); + let mut options = Options::new(); + options.set("chunk-size", "12b"); + let mut writer = RangeBitmapFileIndexWriter::try_new(data_type.clone(), &options).unwrap(); + let mut rng = StdRng::seed_from_u64(42); + let values: Vec> = (0..70_000) + .map(|_| { + if rng.gen_ratio(1, 8) { + None + } else { + Some(rng.gen_range(-1000..=1000)) + } + }) + .collect(); + for value in &values { + writer.write(value.map(Datum::Int).as_ref()).unwrap(); + } + let reader = + RangeBitmapFileIndexReader::try_new(data_type.clone(), writer.serialized_bytes().unwrap()) + .unwrap(); + for bound in [-1001, -1000, -1, 0, 1, 1000, 1001] { + for op in [ + PredicateOperator::Eq, + PredicateOperator::NotEq, + PredicateOperator::Lt, + PredicateOperator::LtEq, + PredicateOperator::Gt, + PredicateOperator::GtEq, + ] { + let expected = values.iter().enumerate().filter_map(|(row, value)| { + value + .filter(|value| match op { + PredicateOperator::Eq => *value == bound, + PredicateOperator::NotEq => *value != bound, + PredicateOperator::Lt => *value < bound, + PredicateOperator::LtEq => *value <= bound, + PredicateOperator::Gt => *value > bound, + PredicateOperator::GtEq => *value >= bound, + _ => unreachable!(), + }) + .map(|_| row as u32) + }); + assert_eq!( + reader + .try_evaluate(&data_type, op, &[Datum::Int(bound)]) + .unwrap(), + selection(expected) + ); + } + } +} + +#[test] +fn empty_null_singleton_and_write_after_serialization() { + let data_type = data_type("INT"); + let mut writer = + RangeBitmapFileIndexWriter::try_new(data_type.clone(), &Options::new()).unwrap(); + for nulls in 0..=3 { + assert_eq!(writer.empty(), nulls == 0); + let reader = RangeBitmapFileIndexReader::try_new( + data_type.clone(), + writer.serialized_bytes().unwrap(), + ) + .unwrap(); + assert_eq!( + reader + .try_evaluate(&data_type, PredicateOperator::IsNull, &[]) + .unwrap(), + selection(0..nulls) + ); + assert_eq!(reader.bsi.slices.len(), 64); + writer.write(None).unwrap(); + } + writer.write(Some(&Datum::Int(7))).unwrap(); + let reader = + RangeBitmapFileIndexReader::try_new(data_type.clone(), writer.serialized_bytes().unwrap()) + .unwrap(); + assert_eq!(reader.bsi.slices.len(), 1); + assert_eq!( + reader + .try_evaluate(&data_type, PredicateOperator::Eq, &[Datum::Int(7)]) + .unwrap(), + selection([4]) + ); +} + +#[test] +fn rejects_invalid_configuration_values_and_overflow() { + for name in [ + "DECIMAL(19, 2)", + "TIMESTAMP(9)", + "TIMESTAMP_LTZ(9)", + "BINARY(4)", + "VARBINARY(4)", + ] { + assert!(matches!( + RangeBitmapFileIndexWriter::try_new(data_type(name), &Options::new()), + Err(Error::Unsupported { .. }) + )); + } + for size in ["-1", "bad", "2gb", "4gb", "9223372036854775807tb"] { + let mut options = Options::new(); + options.set("chunk-size", size); + assert!(matches!( + RangeBitmapFileIndexWriter::try_new(data_type("INT"), &options), + Err(Error::ConfigInvalid { .. }) + )); + } + let mut writer = + RangeBitmapFileIndexWriter::try_new(data_type("TIMESTAMP(6)"), &Options::new()).unwrap(); + for value in [ + Datum::Int(1), + Datum::Timestamp { + millis: 0, + nanos: 1, + }, + Datum::Timestamp { + millis: i64::MAX, + nanos: 0, + }, + ] { + assert!(writer.write(Some(&value)).is_err()); + assert!(writer.empty()); + } + writer.row_count = i32::MAX as u32; + assert!(writer.write(None).is_err()); + assert_eq!(writer.row_count, i32::MAX as u32); + assert!(checked_count(i32::MAX as usize + 1).is_err()); +} diff --git a/crates/paimon/src/table/data_file_index_writer.rs b/crates/paimon/src/table/data_file_index_writer.rs index a21739890..532d5a77d 100644 --- a/crates/paimon/src/table/data_file_index_writer.rs +++ b/crates/paimon/src/table/data_file_index_writer.rs @@ -104,7 +104,9 @@ impl FileIndexOptions { }; if !matches!( (identifier, option), - ("bitmap", "version" | "index-block-size") | ("bloom-filter", "items" | "fpp") + ("bitmap", "version" | "index-block-size") + | ("bloom-filter", "items" | "fpp") + | ("range-bitmap", "chunk-size") ) { return Err(Error::ConfigInvalid { message: format!("Unknown file index option: {key}"), diff --git a/crates/paimon/src/table/data_file_index_writer/tests.rs b/crates/paimon/src/table/data_file_index_writer/tests.rs index b7f6bed2a..15615aa92 100644 --- a/crates/paimon/src/table/data_file_index_writer/tests.rs +++ b/crates/paimon/src/table/data_file_index_writer/tests.rs @@ -147,7 +147,7 @@ async fn evaluate( #[tokio::test] async fn test_file_index_append_commit_reload_and_rolling() { - for identifier in ["bitmap", "bloom-filter", "both"] { + for identifier in ["bitmap", "bloom-filter", "both", "range-bitmap", "all"] { for rolling in [false, true] { for threshold in ["0 B", "1 MB"] { let mut options = vec![ @@ -155,15 +155,19 @@ async fn test_file_index_append_commit_reload_and_rolling() { ("file-index.in-manifest-threshold", threshold), ("file-index.read.enabled", "false"), ]; - if identifier != "bloom-filter" { + if matches!(identifier, "bitmap" | "both" | "all") { options.push(("file-index.bitmap.columns", " id, value, id ")); } - if identifier != "bitmap" { + if matches!(identifier, "bloom-filter" | "both" | "all") { options.extend([ ("file-index.bloom-filter.columns", "id"), ("file-index.bloom-filter.id.items", "10"), ]); } + if matches!(identifier, "range-bitmap" | "all") { + options.push(("file-index.range-bitmap.columns", "id, value")); + options.push(("file-index.range-bitmap.id.chunk-size", "0b")); + } let table = table(memory_io(), schema(&options)).await; let builder = table.new_write_builder(); let mut writer = builder.new_write().unwrap(); @@ -303,9 +307,9 @@ async fn test_file_index_skips_unsupported_identifier_groups() { let mut options = vec![ ("file-index.bsi.columns", "id"), ("file-index.bsi.id.version", "upstream-specific"), - ("file-index.range-bitmap.columns", "missing[nested]"), + ("file-index.future-index.columns", "missing[nested]"), ( - "file-index.range-bitmap.missing[nested].version", + "file-index.future-index.missing[nested].version", "upstream-specific", ), ("file-index.unknown.columns", ""), @@ -357,19 +361,19 @@ async fn test_file_index_skips_unsupported_identifier_groups() { } #[test] -fn test_file_index_read_only_index_does_not_enable_generation() { +fn test_file_index_range_bitmap_enables_generation() { assert!(FileIndexerFactory::is_supported("range-bitmap")); let schema = schema(&[("file-index.range-bitmap.columns", "id")]); assert!(FileIndexOptions::parse(schema.options(), schema.fields()) .unwrap() - .is_none()); + .is_some()); } #[test] fn test_file_index_skips_unsupported_options_without_columns() { let schema = schema(&[ ("file-index.bsi.id.version", "upstream-specific"), - ("file-index.range-bitmap.version", "upstream-specific"), + ("file-index.future-index.version", "upstream-specific"), ]); assert!(FileIndexOptions::parse(schema.options(), schema.fields()) .unwrap() @@ -430,6 +434,17 @@ async fn test_file_index_threshold_boundary_and_abort() { #[tokio::test] async fn test_file_index_invalid_configuration_fails_before_writing() { let cases = vec![ + vec![("file-index.range-bitmap.columns", "missing")], + vec![("file-index.range-bitmap.columns", "id[nested]")], + vec![("file-index.range-bitmap.id.chunk-size", "0b")], + vec![ + ("file-index.range-bitmap.columns", "id"), + ("file-index.range-bitmap.id.chunk-size", "2gb"), + ], + vec![ + ("file-index.range-bitmap.columns", "id"), + ("file-index.range-bitmap.id.version", "1"), + ], vec![("file-index.bitmap.columns", "missing")], vec![("file-index.bitmap.version", "2")], vec![("file-index.bitmap.columns", "")], @@ -487,33 +502,35 @@ async fn test_file_index_invalid_configuration_fails_before_writing() { #[tokio::test] async fn test_file_index_rejects_unsupported_table_write_modes() { - for schema in [ - Schema::builder() - .column("id", crate::spec::DataType::Int(IntType::new())) - .primary_key(["id"]) - .option("bucket", "1") - .option("file-index.bitmap.columns", "id") - .build() - .unwrap(), - Schema::builder() - .column("id", crate::spec::DataType::Int(IntType::new())) - .option("data-evolution.enabled", "true") - .option("row-tracking.enabled", "true") - .option("file-index.bitmap.columns", "id") - .build() - .unwrap(), - ] { - let table = table(memory_io(), schema).await; - let error = match table.new_write_builder().new_write() { - Ok(_) => panic!("unsupported write mode must reject index generation"), - Err(error) => error, - }; - assert!( - error - .to_string() - .contains("FileIndex generation supports ordinary append writes only"), - "{error}" - ); + for identifier in ["bitmap", "range-bitmap"] { + for schema in [ + Schema::builder() + .column("id", crate::spec::DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option(format!("file-index.{identifier}.columns"), "id") + .build() + .unwrap(), + Schema::builder() + .column("id", crate::spec::DataType::Int(IntType::new())) + .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") + .option(format!("file-index.{identifier}.columns"), "id") + .build() + .unwrap(), + ] { + let table = table(memory_io(), schema).await; + let error = match table.new_write_builder().new_write() { + Ok(_) => panic!("unsupported write mode must reject index generation"), + Err(error) => error, + }; + assert!( + error + .to_string() + .contains("FileIndex generation supports ordinary append writes only"), + "{error}" + ); + } } } @@ -526,6 +543,7 @@ async fn test_file_index_uses_partition_bucket_file_row_order() { .option("bucket", "2") .option("bucket-key", "id") .option("file-index.bitmap.columns", "id,value") + .option("file-index.range-bitmap.columns", "id,value") .option("file-index.in-manifest-threshold", "0 B") .build() .unwrap(); @@ -568,7 +586,7 @@ async fn test_file_index_uses_partition_bucket_file_row_order() { .try_collect() .await .unwrap(); - let expected = batches + let expected: roaring::RoaringBitmap = batches .iter() .flat_map(|batch| { batch @@ -581,19 +599,17 @@ async fn test_file_index_uses_partition_bucket_file_row_order() { .enumerate() .filter_map(|(row, id)| (id == Some(3)).then_some(row as u32)) .collect(); - let actual = evaluate( - &table, - split.bucket_path(), - file, - PredicateBuilder::new(table.schema().fields()) - .equal("id", Datum::Int(3)) - .unwrap(), - ) - .await; - match actual { - FileIndexResult::Selection(rows) => assert_eq!(rows, expected), - FileIndexResult::Skip => assert!(roaring::RoaringBitmap::is_empty(&expected)), - FileIndexResult::Remain => panic!("bitmap equality must select physical rows"), + let predicates = PredicateBuilder::new(table.schema().fields()); + for predicate in [ + predicates.equal("id", Datum::Int(3)).unwrap(), + predicates.greater_than("id", Datum::Int(2)).unwrap(), + ] { + let actual = evaluate(&table, split.bucket_path(), file, predicate).await; + match actual { + FileIndexResult::Selection(rows) => assert_eq!(rows, expected), + FileIndexResult::Skip => assert!(expected.is_empty()), + FileIndexResult::Remain => panic!("index predicate must select physical rows"), + } } } } @@ -667,7 +683,7 @@ impl StorageProbe { impl FileIOProvider for StorageProbe { async fn create(&self, path: &str) -> Result<(Operator, String)> { let relative = path.strip_prefix("memory:/").unwrap().to_string(); - if path.ends_with(".parquet") { + if path.ends_with(".parquet") || path.ends_with(".row") { self.data_accesses.fetch_add(1, Ordering::SeqCst); } if path.ends_with(".index") @@ -726,6 +742,83 @@ async fn test_file_index_prunes_without_opening_data_file() { } } +#[tokio::test] +async fn test_range_bitmap_append_range_pruning() { + for format in ["parquet", "row"] { + for threshold in ["0 B", "1 MB"] { + let storage = StorageProbe::new(0); + let table = table( + storage.io(), + schema(&[ + ("file.format", format), + ("file-index.range-bitmap.columns", "id"), + ("file-index.range-bitmap.id.chunk-size", "0b"), + ("file-index.in-manifest-threshold", threshold), + ]), + ) + .await; + let builder = table.new_write_builder(); + let mut writer = builder.new_write().unwrap(); + writer + .write_arrow_batch(&batch(vec![Some(9), None, Some(1), Some(9)], vec![None; 4])) + .await + .unwrap(); + builder + .new_commit() + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); + let predicates = PredicateBuilder::new(table.schema().fields()); + let missing = predicates + .between("id", Datum::Int(3), Datum::Int(7)) + .unwrap(); + let mut scan = table.new_read_builder(); + scan.with_filter(missing.clone()); + let (_, trace) = scan.new_scan().plan_with_trace().await.unwrap(); + assert_eq!(trace.final_files, 1, "statistics must retain the file"); + storage.data_accesses.store(0, Ordering::SeqCst); + assert!(query(&table, true, Some(missing.clone())).await.is_empty()); + assert_eq!(storage.data_accesses.load(Ordering::SeqCst), 0); + assert!(query(&table, false, Some(missing)).await.is_empty()); + assert!(storage.data_accesses.load(Ordering::SeqCst) > 0); + + let plan = table.new_read_builder().new_scan().plan().await.unwrap(); + let split = &plan.splits()[0]; + let file = &split.data_files()[0]; + assert_eq!(file.embedded_index.is_some(), threshold != "0 B"); + assert_eq!(file.extra_files.len(), usize::from(threshold == "0 B")); + for (predicate, positions, expected) in [ + ( + predicates.less_than("id", Datum::Int(5)).unwrap(), + vec![2], + vec![(Some(1), None)], + ), + ( + predicates.greater_or_equal("id", Datum::Int(9)).unwrap(), + vec![0, 3], + vec![(Some(9), None); 2], + ), + ( + predicates.is_null("id").unwrap(), + vec![1], + vec![(None, None)], + ), + ] { + assert_eq!( + evaluate(&table, split.bucket_path(), file, predicate.clone()).await, + FileIndexResult::Selection(positions.into_iter().collect()) + ); + for enabled in [false, true] { + assert_eq!( + query(&table, enabled, Some(predicate.clone())).await, + expected + ); + } + } + } + } +} + #[tokio::test] async fn test_file_index_bloom_false_positive_keeps_residual_filter() { let table = table( diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md index c04e54b9d..aecf9bb2b 100644 --- a/docs/src/getting-started.md +++ b/docs/src/getting-started.md @@ -128,7 +128,7 @@ Mosaic data file reads are always available. The current Mosaic support is read- ## FileIndexes for Append Writes -Ordinary append writes can generate Bitmap and Bloom Filter indexes for supported +Ordinary append writes can generate Bitmap, Bloom Filter, and Range Bitmap indexes for supported top-level columns using table options: ```text @@ -136,16 +136,26 @@ file-index.bitmap.columns = category file-index.bloom-filter.columns = id file-index.bloom-filter.id.items = 100000 file-index.bloom-filter.id.fpp = 0.01 +file-index.range-bitmap.columns = score +file-index.range-bitmap.score.chunk-size = 16kb file-index.in-manifest-threshold = 500 B ``` Column lists are comma-separated. Bitmap supports `version` (currently `2` only) and `index-block-size` per column. Bloom Filter supports `items` and `fpp`. +Range Bitmap writes Java V1 payloads and supports `chunk-size` per column +(0 through 2147483647 bytes). The default is `16kb`, except for Boolean, +TinyInt, and SmallInt, which default to `0b`. A chunk's first key is stored +in its header, so `0b` is valid and creates one dictionary chunk per distinct key. +Supported Range Bitmap types are Boolean, TinyInt, SmallInt, Int, BigInt, +Float, Double, Decimal (precision <= 18), Date, Time, Timestamp and +LocalZonedTimestamp (precision <= 6), and Char/VarChar/String. +Range Bitmap enables range predicate pruning through the existing file index +reader; it does not add TopN support. For index types supported for writing, invalid columns, unsupported data types, and invalid index options fail when creating the writer. Index types without a -writer (such as `bsi` and `range-bitmap`) and all their options are ignored, so -these table properties do not prevent append writes. Range Bitmap indexes can -still be read from existing files. +writer (such as `bsi`) and all their options are ignored, so +these table properties do not prevent append writes. When supported indexes are configured, each data file gets its own index. The complete serialized index is embedded in the manifest when its size