feat(file_index): support range bitmap append writes

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();
 

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<Item = u32>) -> FileIndexResult {
         FileIndexResult::Selection(rows.into_iter().collect())
     }