feat(file-index): read range bitmap indexes

crates/paimon/src/file_index/file_indexer_factory.rs

@@ -23,23 +23,31 @@ 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::RangeBitmapFileIndexReader;
 use crate::spec::DataType;
 use crate::{Error, Result};
 
 pub(crate) const BITMAP_INDEX: &str = "bitmap";
 pub(crate) const BLOOM_FILTER_INDEX: &str = "bloom-filter";
+pub(crate) const RANGE_BITMAP_INDEX: &str = "range-bitmap";
+
+struct FailOpenFileIndexReader;
+
+impl FileIndexReader for FailOpenFileIndexReader {}
 
 #[derive(Clone, Copy)]
 enum BuiltinFileIndexer {
     Bitmap,
     BloomFilter,
+    RangeBitmap,
 }
 
 impl BuiltinFileIndexer {
     fn from_identifier(identifier: &str) -> Result<Self> {
         match identifier {
             BITMAP_INDEX => Ok(Self::Bitmap),
             BLOOM_FILTER_INDEX => Ok(Self::BloomFilter),
+            RANGE_BITMAP_INDEX => Ok(Self::RangeBitmap),
             _ => Err(Error::Unsupported {
                 message: format!("Unknown file index identifier: {identifier}"),
             }),
@@ -52,7 +60,10 @@ pub(crate) struct FileIndexerFactory;
 
 impl FileIndexerFactory {
     pub(crate) fn is_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(
@@ -67,6 +78,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(),
+            }),
         }
     }
 
@@ -82,14 +96,30 @@ impl FileIndexerFactory {
             BuiltinFileIndexer::BloomFilter => {
                 Ok(Box::new(BloomFilterReader::try_new(data_type, serialized)?))
             }
+            BuiltinFileIndexer::RangeBitmap => {
+                // File indexes are optional accelerators. Rust used to ignore
+                // range-bitmap payloads entirely, so a payload written by a
+                // newer Java version or damaged in storage must conservatively
+                // disable pruning instead of turning a readable data file into
+                // a query failure.
+                Ok(
+                    match RangeBitmapFileIndexReader::try_new(data_type, serialized) {
+                        Ok(reader) => Box::new(reader),
+                        Err(_) => Box::new(FailOpenFileIndexReader),
+                    },
+                )
+            }
         }
     }
 }
 
 #[cfg(test)]
 mod tests {
+    use bytes::{BufMut, BytesMut};
+
     use super::*;
-    use crate::spec::{BinaryType, BooleanType, Datum, IntType};
+    use crate::file_index::file_index_result::FileIndexResult;
+    use crate::spec::{BinaryType, BooleanType, Datum, IntType, PredicateOperator};
 
     fn int_type() -> DataType {
         DataType::Int(IntType::new())
@@ -147,6 +177,11 @@ mod tests {
 
     #[test]
     fn test_unknown_identifier_is_rejected() {
+        assert!(FileIndexerFactory::is_supported(RANGE_BITMAP_INDEX));
+        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 { .. })
@@ -170,4 +205,76 @@ mod tests {
             assert!(writer.empty(), "{identifier}");
         }
     }
+
+    #[test]
+    fn test_range_bitmap_huge_cardinality_fails_open() {
+        let mut dictionary = BytesMut::new();
+        dictionary.put_i32(13);
+        dictionary.put_u8(1);
+        dictionary.put_i32(0);
+        dictionary.put_i32(0);
+        dictionary.put_i32(0);
+
+        let mut serialized = BytesMut::new();
+        serialized.put_i32(21);
+        serialized.put_u8(1);
+        serialized.put_i32(i32::MAX);
+        serialized.put_i32(i32::MAX);
+        serialized.put_i32(0);
+        serialized.put_i32(0);
+        serialized.put_i32(dictionary.len() as i32);
+        serialized.extend_from_slice(&dictionary);
+        let serialized = serialized.freeze();
+
+        assert!(matches!(
+            RangeBitmapFileIndexReader::try_new(int_type(), serialized.clone()),
+            Err(Error::FileIndexFormatInvalid { .. })
+        ));
+
+        let reader =
+            FileIndexerFactory::create_reader(RANGE_BITMAP_INDEX, int_type(), serialized).unwrap();
+        assert_eq!(
+            FileIndexResult::Remain,
+            reader.evaluate("a", 0, &int_type(), PredicateOperator::Eq, &[Datum::Int(0)])
+        );
+    }
+
+    #[test]
+    fn test_range_bitmap_malformed_bsi_fails_open() {
+        // Java V1 index for [1, 3, 5, 7, 9, null, null, 10]. Change its
+        // declared slice count from three to one while leaving the BSI header
+        // and payload otherwise intact.
+        let mut serialized = hex::decode(concat!(
+            "00000015010000000800000006000000010000000a000000420000000d010000",
+            "0001000000040000001900000000010000000100000000000000000000000500",
+            "00001400000004000000030000000500000007000000090000000a0000002201",
+            "030000001300000018000000000000001600000016000000140000002a000000",
+            "143b3000000100000500020000000400070000003a3000000100000000000200",
+            "100000000100030007003a300000010000000000010010000000020003003a30",
+            "000001000000000001001000000004000700"
+        ))
+        .unwrap();
+        let outer_header_length = i32::from_be_bytes(serialized[0..4].try_into().unwrap()) as usize;
+        let dictionary_length_offset = 4 + outer_header_length - 4;
+        let dictionary_length = i32::from_be_bytes(
+            serialized[dictionary_length_offset..dictionary_length_offset + 4]
+                .try_into()
+                .unwrap(),
+        ) as usize;
+        let bsi_offset = 4 + outer_header_length + dictionary_length;
+        serialized[bsi_offset + 5] = 1;
+        let serialized = Bytes::from(serialized);
+
+        assert!(matches!(
+            RangeBitmapFileIndexReader::try_new(int_type(), serialized.clone()),
+            Err(Error::FileIndexFormatInvalid { .. })
+        ));
+
+        let reader =
+            FileIndexerFactory::create_reader(RANGE_BITMAP_INDEX, int_type(), serialized).unwrap();
+        assert_eq!(
+            FileIndexResult::Remain,
+            reader.evaluate("a", 0, &int_type(), PredicateOperator::Eq, &[Datum::Int(1)])
+        );
+    }
 }

crates/paimon/src/file_index/mod.rs

@@ -33,5 +33,7 @@ pub(crate) mod file_index_result;
 pub(crate) mod file_index_writer;
 #[allow(dead_code)]
 pub(crate) mod file_indexer_factory;
+#[allow(dead_code)]
+pub(crate) mod range_bitmap;
 
 pub use file_index_format::*;