This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 0aba451b feat(file_index): add indexer contracts and outer-format
composition (#764)
0aba451b is described below
commit 0aba451bd978df08d5ba9245cb353126fc60ec2b
Author: QuakeWang <[email protected]>
AuthorDate: Tue Sep 1 11:54:14 2026 +0800
feat(file_index): add indexer contracts and outer-format composition (#764)
---
crates/paimon/src/file_index/bitmap/mod.rs | 1 +
crates/paimon/src/file_index/bitmap/writer.rs | 11 +-
crates/paimon/src/file_index/bloom_filter/mod.rs | 44 ++-
crates/paimon/src/file_index/file_index_format.rs | 373 ++++++++++++++++++++-
crates/paimon/src/file_index/file_index_reader.rs | 75 +++++
.../file_index/{mod.rs => file_index_writer.rs} | 29 +-
.../paimon/src/file_index/file_indexer_factory.rs | 169 ++++++++++
crates/paimon/src/file_index/mod.rs | 6 +-
8 files changed, 675 insertions(+), 33 deletions(-)
diff --git a/crates/paimon/src/file_index/bitmap/mod.rs
b/crates/paimon/src/file_index/bitmap/mod.rs
index 105cd739..b45f67e9 100644
--- a/crates/paimon/src/file_index/bitmap/mod.rs
+++ b/crates/paimon/src/file_index/bitmap/mod.rs
@@ -953,6 +953,7 @@ impl FileIndexReader for BitmapFileIndexReader {
mod tests {
use super::*;
use crate::common::Options;
+ use crate::file_index::file_index_writer::FileIndexWriter;
use crate::spec::{
BigIntType, BinaryType, BooleanType, CharType, DateType, DoubleType,
FloatType, IntType,
LocalZonedTimestampType, SmallIntType, TimeType, TimestampType,
TinyIntType, VarCharType,
diff --git a/crates/paimon/src/file_index/bitmap/writer.rs
b/crates/paimon/src/file_index/bitmap/writer.rs
index cc63573e..c2c68f78 100644
--- a/crates/paimon/src/file_index/bitmap/writer.rs
+++ b/crates/paimon/src/file_index/bitmap/writer.rs
@@ -23,6 +23,7 @@ use roaring::RoaringBitmap;
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};
@@ -56,8 +57,10 @@ impl BitmapFileIndexWriter {
bitmaps: HashMap::new(),
})
}
+}
- pub(crate) fn write(&mut self, datum: Option<&Datum>) -> Result<()> {
+impl FileIndexWriter for BitmapFileIndexWriter {
+ 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(),
@@ -81,7 +84,7 @@ impl BitmapFileIndexWriter {
Ok(())
}
- pub(crate) fn serialized_bytes(&mut self) -> Result<Bytes> {
+ fn serialized_bytes(&mut self) -> Result<Bytes> {
let null_bytes = serialize_bitmap(&mut self.null_bitmap)?;
let mut body = Vec::new();
let null_entry = if self.null_bitmap.is_empty() {
@@ -162,6 +165,10 @@ impl BitmapFileIndexWriter {
output.extend_from_slice(&body);
Ok(output.freeze())
}
+
+ fn empty(&self) -> bool {
+ self.row_count == 0
+ }
}
struct SerializedEntry<'a> {
diff --git a/crates/paimon/src/file_index/bloom_filter/mod.rs
b/crates/paimon/src/file_index/bloom_filter/mod.rs
index 9d5708ed..d70a5e85 100644
--- a/crates/paimon/src/file_index/bloom_filter/mod.rs
+++ b/crates/paimon/src/file_index/bloom_filter/mod.rs
@@ -23,6 +23,7 @@ use bytes::{Bytes, BytesMut};
use crate::common::Options;
use crate::file_index::file_index_reader::FileIndexReader;
use crate::file_index::file_index_result::FileIndexResult;
+use crate::file_index::file_index_writer::FileIndexWriter;
use crate::spec::{DataType, Datum, PredicateOperator};
use crate::{Error, Result};
@@ -37,6 +38,7 @@ const FPP: &str = "fpp";
pub(crate) struct BloomFilterWriter {
hash_function: FastHash,
filter: BloomFilter64,
+ empty: bool,
}
impl BloomFilterWriter {
@@ -48,21 +50,29 @@ impl BloomFilterWriter {
Ok(Self {
hash_function,
filter,
+ empty: true,
})
}
+}
- pub(crate) fn write(&mut self, datum: Option<&Datum>) -> Result<()> {
+impl FileIndexWriter for BloomFilterWriter {
+ fn write(&mut self, datum: Option<&Datum>) -> Result<()> {
if let Some(datum) = datum {
self.filter.add_hash(self.hash_function.hash(datum)?);
}
+ self.empty = false;
Ok(())
}
- pub(crate) fn serialized_bytes(&self) -> Bytes {
+ fn serialized_bytes(&mut self) -> Result<Bytes> {
let mut serialized = BytesMut::with_capacity(4 +
self.filter.bytes().len());
serialized.extend_from_slice(&self.filter.num_hash_functions().to_be_bytes());
serialized.extend_from_slice(self.filter.bytes());
- serialized.freeze()
+ Ok(serialized.freeze())
+ }
+
+ fn empty(&self) -> bool {
+ self.empty
}
}
@@ -198,7 +208,7 @@ mod tests {
}
writer.write(None).unwrap();
- let serialized = writer.serialized_bytes();
+ let serialized = writer.serialized_bytes().unwrap();
assert_eq!(
serialized.as_ref(),
&hex::decode("00000003818281005001").unwrap()
@@ -253,7 +263,7 @@ mod tests {
.unwrap();
writer.write(Some(&Datum::Long(42))).unwrap();
- let serialized = writer.serialized_bytes();
+ let serialized = writer.serialized_bytes().unwrap();
assert_eq!(&serialized[..4], &133_i32.to_be_bytes());
assert_eq!(serialized.len(), 28);
@@ -279,7 +289,8 @@ mod tests {
.write(Some(&Datum::Float(f32::from_bits(0x7fc0_0001))))
.unwrap();
let reader =
- BloomFilterReader::try_new(float_type.clone(),
writer.serialized_bytes()).unwrap();
+ BloomFilterReader::try_new(float_type.clone(),
writer.serialized_bytes().unwrap())
+ .unwrap();
assert_eq!(
evaluate(
@@ -323,7 +334,8 @@ mod tests {
BloomFilterWriter::try_new(double_type.clone(), &options("10",
"0.1")).unwrap();
writer.write(Some(&Datum::Double(-0.0))).unwrap();
let reader =
- BloomFilterReader::try_new(double_type.clone(),
writer.serialized_bytes()).unwrap();
+ BloomFilterReader::try_new(double_type.clone(),
writer.serialized_bytes().unwrap())
+ .unwrap();
assert_eq!(
evaluate(
&reader,
@@ -342,7 +354,8 @@ mod tests {
let bigint = DataType::BigInt(BigIntType::new());
let mut writer = BloomFilterWriter::try_new(bigint.clone(),
&options).unwrap();
writer.write(Some(&Datum::Long(42))).unwrap();
- let reader = BloomFilterReader::try_new(bigint,
writer.serialized_bytes()).unwrap();
+ let reader =
+ BloomFilterReader::try_new(bigint,
writer.serialized_bytes().unwrap()).unwrap();
assert_eq!(
evaluate(
&reader,
@@ -358,7 +371,8 @@ mod tests {
writer
.write(Some(&Datum::String("abc".to_string())))
.unwrap();
- let reader = BloomFilterReader::try_new(char_type,
writer.serialized_bytes()).unwrap();
+ let reader =
+ BloomFilterReader::try_new(char_type,
writer.serialized_bytes().unwrap()).unwrap();
assert_eq!(
evaluate(
&reader,
@@ -377,7 +391,8 @@ mod tests {
nanos: 456_000,
}))
.unwrap();
- let reader = BloomFilterReader::try_new(timestamp4,
writer.serialized_bytes()).unwrap();
+ let reader =
+ BloomFilterReader::try_new(timestamp4,
writer.serialized_bytes().unwrap()).unwrap();
let missing = [Datum::Timestamp {
millis: 1_700_000_000_124,
nanos: 456_000,
@@ -479,7 +494,8 @@ mod tests {
BloomFilterWriter::try_new(data_type.clone(), &options("10",
"0.1")).unwrap();
writer.write(Some(&datum)).unwrap();
let reader =
- BloomFilterReader::try_new(data_type.clone(),
writer.serialized_bytes()).unwrap();
+ BloomFilterReader::try_new(data_type.clone(),
writer.serialized_bytes().unwrap())
+ .unwrap();
assert_eq!(
evaluate(&reader, &data_type, PredicateOperator::Eq, &[datum]),
@@ -509,7 +525,8 @@ mod tests {
writer.write(Some(&Datum::Long(*value))).unwrap();
}
let reader =
- BloomFilterReader::try_new(data_type.clone(),
writer.serialized_bytes()).unwrap();
+ BloomFilterReader::try_new(data_type.clone(),
writer.serialized_bytes().unwrap())
+ .unwrap();
for value in inserted {
assert_eq!(
@@ -547,7 +564,8 @@ mod tests {
BloomFilterWriter::try_new(data_type.clone(), &options("10",
"0.1")).unwrap();
writer.write(Some(&Datum::Long(42))).unwrap();
let reader =
- BloomFilterReader::try_new(data_type.clone(),
writer.serialized_bytes()).unwrap();
+ BloomFilterReader::try_new(data_type.clone(),
writer.serialized_bytes().unwrap())
+ .unwrap();
for operator in [
PredicateOperator::IsNull,
diff --git a/crates/paimon/src/file_index/file_index_format.rs
b/crates/paimon/src/file_index/file_index_format.rs
index 6bd0f4ca..f5fcb22a 100644
--- a/crates/paimon/src/file_index/file_index_format.rs
+++ b/crates/paimon/src/file_index/file_index_format.rs
@@ -15,12 +15,17 @@
// specific language governing permissions and limitations
// under the License.
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use bytes::{BufMut, Bytes, BytesMut};
use crate::{
+ file_index::{
+ file_index_reader::{EmptyFileIndexReader, FileIndexReader},
+ file_indexer_factory::FileIndexerFactory,
+ },
io::{FileIO, FileRead, FileStatus, InputFile, OutputFile},
+ spec::{DataField, DataType},
Error,
};
@@ -329,12 +334,78 @@ fn calculate_head_length(
Ok(total_length)
}
+fn resolve_index_data_type(
+ fields_by_name: &HashMap<&str, &DataType>,
+ column_name: &str,
+) -> crate::Result<DataType> {
+ let nested_start = column_name.find('[').filter(|_|
column_name.ends_with(']'));
+ let field_name = nested_start
+ .map(|index| &column_name[..index])
+ .unwrap_or(column_name);
+ let data_type = fields_by_name.get(field_name).copied().ok_or_else(|| {
+ format_invalid(format!(
+ "Column '{field_name}' for file index '{column_name}' was not
found in schema"
+ ))
+ })?;
+
+ match (nested_start, data_type) {
+ (Some(_), DataType::Map(map_type)) =>
Ok(map_type.value_type().clone()),
+ (Some(_), data_type) => Err(format_invalid(format!(
+ "Nested file index '{column_name}' requires Map column
'{field_name}', but found {data_type:?}"
+ ))),
+ (None, data_type) => Ok(data_type.clone()),
+ }
+}
+
pub struct FileIndex {
reader: Box<dyn FileRead>,
header: HashMap<String, HashMap<String, IndexInfo>>,
}
impl FileIndex {
+ /// Constructs readers for the required columns described by this
outer-format file.
+ #[allow(dead_code)]
+ pub(crate) async fn create_index_readers(
+ &self,
+ fields: &[DataField],
+ required_columns: &HashSet<String>,
+ ) -> crate::Result<HashMap<String, Vec<Box<dyn FileIndexReader>>>> {
+ let fields_by_name = fields
+ .iter()
+ .map(|field| (field.name(), field.data_type()))
+ .collect::<HashMap<&str, &DataType>>();
+ let mut readers = HashMap::with_capacity(required_columns.len());
+
+ for column_name in required_columns {
+ let Some(index_info) = self.header.get(column_name) else {
+ continue;
+ };
+ let mut column_readers = Vec::with_capacity(index_info.len());
+ for (identifier, info) in index_info {
+ if info.start_pos == EMPTY_INDEX_FLAG {
+ column_readers.push(Box::new(EmptyFileIndexReader) as
Box<dyn FileIndexReader>);
+ continue;
+ }
+
+ let data_type = resolve_index_data_type(&fields_by_name,
column_name)?;
+ let serialized = self
+ .get_bytes_with_start_and_length(info)
+ .await?
+ .ok_or_else(|| {
+ format_invalid(format!(
+ "Non-empty file index '{identifier}' for column
'{column_name}' had no payload"
+ ))
+ })?;
+ column_readers.push(FileIndexerFactory::create_reader(
+ identifier, data_type, serialized,
+ )?);
+ }
+ readers.insert(column_name.clone(), column_readers);
+ }
+
+ Ok(readers)
+ }
+
pub async fn get_column_index(
&self,
column_name: &str,
@@ -557,8 +628,21 @@ impl FileIndexFormatReader {
mod file_index_format_tests {
use super::*;
- use bytes::Bytes;
- use std::collections::HashMap;
+ use bytes::{Bytes, BytesMut};
+ use std::collections::{HashMap, HashSet};
+ use std::ops::Range;
+ use std::sync::{Arc, Mutex};
+
+ use crate::common::Options;
+ use crate::file_index::file_index_predicate::FileIndexPredicate;
+ use crate::file_index::file_index_result::FileIndexResult;
+ use crate::file_index::file_indexer_factory::{
+ FileIndexerFactory, BITMAP_INDEX, BLOOM_FILTER_INDEX,
+ };
+ use crate::spec::{
+ BigIntType, Datum, IntType, MapType, Predicate, PredicateBuilder,
PredicateOperator,
+ VarCharType,
+ };
const JAVA_V1_SIMPLE: &str = concat!(
"00054e4ed01a35ae000000010000002a0000000100016100000001000162",
@@ -581,6 +665,111 @@ mod file_index_format_tests {
Ok(output.to_input_file())
}
+ struct TrackingFileRead {
+ data: Bytes,
+ ranges: Arc<Mutex<Vec<Range<u64>>>>,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for TrackingFileRead {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ self.ranges.lock().unwrap().push(range.clone());
+ Ok(self.data.slice(range.start as usize..range.end as usize))
+ }
+ }
+
+ #[test]
+ fn test_resolve_index_data_type_matches_java_nested_name_rules() {
+ let value_type = DataType::BigInt(BigIntType::new());
+ let map_type = DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::new(20).unwrap()),
+ value_type.clone(),
+ ));
+ let exact_unclosed_type = DataType::Int(IntType::new());
+ let plain_type = DataType::Int(IntType::new());
+ let fields_by_name = HashMap::from([
+ ("metrics", &map_type),
+ ("metrics[k", &exact_unclosed_type),
+ ("plain", &plain_type),
+ ]);
+
+ assert_eq!(
+ resolve_index_data_type(&fields_by_name, "metrics[k]").unwrap(),
+ value_type
+ );
+ assert_eq!(
+ resolve_index_data_type(&fields_by_name, "metrics[k").unwrap(),
+ exact_unclosed_type
+ );
+ assert!(matches!(
+ resolve_index_data_type(&fields_by_name, "metrics[k][nested]"),
+ Ok(DataType::BigInt(_))
+ ));
+ assert!(matches!(
+ resolve_index_data_type(&fields_by_name, "metrics[k][nested"),
+ Err(Error::FileIndexFormatInvalid { .. })
+ ));
+ assert!(matches!(
+ resolve_index_data_type(&fields_by_name, "plain[k]"),
+ Err(Error::FileIndexFormatInvalid { .. })
+ ));
+ }
+
+ #[tokio::test]
+ async fn test_composition_reads_only_required_column_payloads() ->
crate::Result<()> {
+ let data_type = DataType::Int(IntType::new());
+ let mut writer =
+ FileIndexerFactory::create_writer(BITMAP_INDEX, data_type.clone(),
&Options::new())?;
+ writer.write(Some(&Datum::Int(1)))?;
+ let required_payload = writer.serialized_bytes()?;
+ let unrelated_payload = Bytes::from(vec![0; 1024]);
+ let required_end = required_payload.len() as u64;
+ let unrelated_end = required_end + unrelated_payload.len() as u64;
+ let mut data = BytesMut::with_capacity(unrelated_end as usize);
+ data.extend_from_slice(&required_payload);
+ data.extend_from_slice(&unrelated_payload);
+ let ranges = Arc::new(Mutex::new(Vec::new()));
+ let file_index = FileIndex {
+ reader: Box::new(TrackingFileRead {
+ data: data.freeze(),
+ ranges: Arc::clone(&ranges),
+ }),
+ header: HashMap::from([
+ (
+ "required".to_string(),
+ HashMap::from([(
+ BITMAP_INDEX.to_string(),
+ IndexInfo {
+ start_pos: 0,
+ length: required_end as i32,
+ },
+ )]),
+ ),
+ (
+ "unrelated".to_string(),
+ HashMap::from([(
+ "unknown".to_string(),
+ IndexInfo {
+ start_pos: required_end as i32,
+ length: unrelated_payload.len() as i32,
+ },
+ )]),
+ ),
+ ]),
+ };
+ let fields = [DataField::new(0, "required".to_string(), data_type)];
+ let required_columns = HashSet::from(["required".to_string()]);
+
+ let readers = file_index
+ .create_index_readers(&fields, &required_columns)
+ .await?;
+
+ assert_eq!(readers.len(), 1);
+ assert_eq!(readers["required"].len(), 1);
+ assert_eq!(*ranges.lock().unwrap(), vec![0..required_end]);
+ Ok(())
+ }
+
#[tokio::test]
async fn test_writer_matches_java_v1_bytes() -> crate::Result<()> {
let indexes = HashMap::from([(
@@ -835,6 +1024,184 @@ mod file_index_format_tests {
Ok(())
}
+ #[tokio::test]
+ async fn
test_outer_format_builds_grouped_readers_and_evaluates_predicates() ->
crate::Result<()>
+ {
+ let fields = vec![
+ DataField::new(0, "a".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "b".to_string(),
DataType::BigInt(BigIntType::new())),
+ DataField::new(2, "empty".to_string(),
DataType::Int(IntType::new())),
+ ];
+
+ let mut bloom_options = Options::new();
+ bloom_options.set("items", "10");
+ bloom_options.set("fpp", "0.1");
+
+ let mut bitmap = FileIndexerFactory::create_writer(
+ BITMAP_INDEX,
+ fields[0].data_type().clone(),
+ &Options::new(),
+ )?;
+ let mut bloom = FileIndexerFactory::create_writer(
+ BLOOM_FILTER_INDEX,
+ fields[0].data_type().clone(),
+ &bloom_options,
+ )?;
+ for value in [
+ Some(Datum::Int(1)),
+ Some(Datum::Int(2)),
+ None,
+ Some(Datum::Int(1)),
+ ] {
+ bitmap.write(value.as_ref())?;
+ bloom.write(value.as_ref())?;
+ }
+
+ let mut b_bloom = FileIndexerFactory::create_writer(
+ BLOOM_FILTER_INDEX,
+ fields[1].data_type().clone(),
+ &bloom_options,
+ )?;
+ b_bloom.write(Some(&Datum::Long(42)))?;
+
+ let indexes = HashMap::from([
+ (
+ "a".to_string(),
+ HashMap::from([
+ (BITMAP_INDEX.to_string(),
Some(bitmap.serialized_bytes()?)),
+ (
+ BLOOM_FILTER_INDEX.to_string(),
+ Some(bloom.serialized_bytes()?),
+ ),
+ ]),
+ ),
+ (
+ "b".to_string(),
+ HashMap::from([(
+ BLOOM_FILTER_INDEX.to_string(),
+ Some(b_bloom.serialized_bytes()?),
+ )]),
+ ),
+ (
+ "empty".to_string(),
+ HashMap::from([("unregistered-empty-index".to_string(),
None)]),
+ ),
+ ]);
+
+ let output = write_column_indexes("memory:/tmp/composed_file_indexes",
indexes).await?;
+ let file_index =
FileIndexFormatReader::get_file_index(output.to_input_file()).await?;
+ let required_columns =
+ HashSet::from(["a".to_string(), "b".to_string(),
"empty".to_string()]);
+ let readers = file_index
+ .create_index_readers(&fields, &required_columns)
+ .await?;
+ assert_eq!(readers.len(), 3);
+ assert_eq!(readers["a"].len(), 2);
+ assert_eq!(readers["b"].len(), 1);
+ assert_eq!(readers["empty"].len(), 1);
+
+ let predicate = FileIndexPredicate::new(readers);
+ let builder = PredicateBuilder::new(&fields);
+ assert_eq!(
+ predicate.evaluate(&builder.equal("a", Datum::Int(1))?),
+ FileIndexResult::Selection([0_u32, 3].into_iter().collect())
+ );
+ assert_eq!(
+ predicate.evaluate(&builder.equal("b", Datum::Long(43))?),
+ FileIndexResult::Skip
+ );
+ assert_eq!(
+ predicate.evaluate(&builder.equal("empty", Datum::Int(1))?),
+ FileIndexResult::Skip
+ );
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_outer_format_uses_map_value_type_for_nested_index() ->
crate::Result<()> {
+ let value_type = DataType::BigInt(BigIntType::new());
+ let fields = [DataField::new(
+ 0,
+ "metrics".to_string(),
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::new(20).unwrap()),
+ value_type.clone(),
+ )),
+ )];
+ let mut bloom_options = Options::new();
+ bloom_options.set("items", "10");
+ bloom_options.set("fpp", "0.1");
+ let mut bitmap =
+ FileIndexerFactory::create_writer(BITMAP_INDEX,
value_type.clone(), &Options::new())?;
+ let mut bloom = FileIndexerFactory::create_writer(
+ BLOOM_FILTER_INDEX,
+ value_type.clone(),
+ &bloom_options,
+ )?;
+ for value in [Datum::Long(7), Datum::Long(8)] {
+ bitmap.write(Some(&value))?;
+ bloom.write(Some(&value))?;
+ }
+ let indexes = HashMap::from([(
+ "metrics[k]".to_string(),
+ HashMap::from([
+ (BITMAP_INDEX.to_string(), Some(bitmap.serialized_bytes()?)),
+ (
+ BLOOM_FILTER_INDEX.to_string(),
+ Some(bloom.serialized_bytes()?),
+ ),
+ ]),
+ )]);
+
+ let output =
write_column_indexes("memory:/tmp/nested_map_file_indexes", indexes).await?;
+ let file_index =
FileIndexFormatReader::get_file_index(output.to_input_file()).await?;
+ let required_columns = HashSet::from(["metrics[k]".to_string()]);
+ let readers = file_index
+ .create_index_readers(&fields, &required_columns)
+ .await?;
+ assert_eq!(readers["metrics[k]"].len(), 2);
+
+ let predicate = FileIndexPredicate::new(readers);
+ assert_eq!(
+ predicate.evaluate(&Predicate::Leaf {
+ column: "metrics[k]".to_string(),
+ index: 0,
+ data_type: value_type,
+ op: PredicateOperator::Eq,
+ literals: vec![Datum::Long(7)],
+ }),
+ FileIndexResult::Selection([0_u32].into_iter().collect())
+ );
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_composition_rejects_unknown_non_empty_identifier() ->
crate::Result<()> {
+ let indexes = HashMap::from([(
+ "a".to_string(),
+ HashMap::from([("unknown".to_string(),
Some(Bytes::from_static(b"payload")))]),
+ )]);
+ let output = write_column_indexes("memory:/tmp/unknown_file_index",
indexes).await?;
+ let file_index =
FileIndexFormatReader::get_file_index(output.to_input_file()).await?;
+ let fields = [DataField::new(
+ 0,
+ "a".to_string(),
+ DataType::Int(IntType::new()),
+ )];
+
+ let required_columns = HashSet::from(["a".to_string()]);
+ let error = match file_index
+ .create_index_readers(&fields, &required_columns)
+ .await
+ {
+ Ok(_) => panic!("unknown identifier must fail"),
+ Err(error) => error,
+ };
+ assert!(matches!(error, Error::Unsupported { .. }));
+ Ok(())
+ }
+
#[tokio::test]
async fn test_large_data_set() -> crate::Result<()> {
let path = "memory:/tmp/test_large_data_set";
diff --git a/crates/paimon/src/file_index/file_index_reader.rs
b/crates/paimon/src/file_index/file_index_reader.rs
index afc5c472..be986099 100644
--- a/crates/paimon/src/file_index/file_index_reader.rs
+++ b/crates/paimon/src/file_index/file_index_reader.rs
@@ -34,3 +34,78 @@ pub(crate) trait FileIndexReader {
FileIndexResult::Remain
}
}
+
+/// Reader used by the outer format when a writer produced no rows.
+pub(crate) struct EmptyFileIndexReader;
+
+impl FileIndexReader for EmptyFileIndexReader {
+ fn evaluate(
+ &self,
+ _column: &str,
+ _index: usize,
+ _data_type: &DataType,
+ operator: PredicateOperator,
+ _literals: &[Datum],
+ ) -> FileIndexResult {
+ match operator {
+ PredicateOperator::Eq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::In
+ | PredicateOperator::IsNotNull
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains => FileIndexResult::Skip,
+ _ => FileIndexResult::Remain,
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::IntType;
+
+ #[test]
+ fn test_empty_reader_matches_java_supported_predicates() {
+ let reader = EmptyFileIndexReader;
+ let data_type = DataType::Int(IntType::new());
+
+ for operator in [
+ PredicateOperator::Eq,
+ PredicateOperator::Lt,
+ PredicateOperator::LtEq,
+ PredicateOperator::Gt,
+ PredicateOperator::GtEq,
+ PredicateOperator::In,
+ PredicateOperator::IsNotNull,
+ PredicateOperator::StartsWith,
+ PredicateOperator::EndsWith,
+ PredicateOperator::Contains,
+ ] {
+ assert_eq!(
+ reader.evaluate("a", 0, &data_type, operator,
&[Datum::Int(1)]),
+ FileIndexResult::Skip
+ );
+ }
+
+ for operator in [
+ PredicateOperator::IsNull,
+ PredicateOperator::NotEq,
+ PredicateOperator::NotIn,
+ PredicateOperator::ArrayContains,
+ PredicateOperator::ArraysOverlap,
+ PredicateOperator::ArrayContainsAll,
+ PredicateOperator::Like,
+ PredicateOperator::Between,
+ PredicateOperator::NotBetween,
+ ] {
+ assert_eq!(
+ reader.evaluate("a", 0, &data_type, operator,
&[Datum::Int(1)]),
+ FileIndexResult::Remain
+ );
+ }
+ }
+}
diff --git a/crates/paimon/src/file_index/mod.rs
b/crates/paimon/src/file_index/file_index_writer.rs
similarity index 61%
copy from crates/paimon/src/file_index/mod.rs
copy to crates/paimon/src/file_index/file_index_writer.rs
index 48616abb..3fce9bb8 100644
--- a/crates/paimon/src/file_index/mod.rs
+++ b/crates/paimon/src/file_index/file_index_writer.rs
@@ -15,18 +15,19 @@
// specific language governing permissions and limitations
// under the License.
-// Concrete readers/writers and predicate plumbing stay crate-private until
-// factory, data-writer, and scan integration land in later changes.
-#[allow(dead_code)]
-pub(crate) mod bitmap;
-#[allow(dead_code)]
-pub(crate) mod bloom_filter;
-mod file_index_format;
-#[allow(dead_code)]
-pub(crate) mod file_index_predicate;
-#[allow(dead_code)]
-pub(crate) mod file_index_reader;
-#[allow(dead_code)]
-pub(crate) mod file_index_result;
+use bytes::Bytes;
-pub use file_index_format::*;
+use crate::spec::Datum;
+use crate::Result;
+
+/// Writes one concrete file index payload.
+pub(crate) trait FileIndexWriter {
+ /// Adds one row to the index. `None` represents a null value.
+ fn write(&mut self, datum: Option<&Datum>) -> Result<()>;
+
+ /// Serializes the index-specific payload consumed by a matching reader.
+ fn serialized_bytes(&mut self) -> Result<Bytes>;
+
+ /// Returns whether no rows have been added to this writer.
+ fn empty(&self) -> bool;
+}
diff --git a/crates/paimon/src/file_index/file_indexer_factory.rs
b/crates/paimon/src/file_index/file_indexer_factory.rs
new file mode 100644
index 00000000..c6fb4b4b
--- /dev/null
+++ b/crates/paimon/src/file_index/file_indexer_factory.rs
@@ -0,0 +1,169 @@
+// 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 bytes::Bytes;
+
+use crate::common::Options;
+use crate::file_index::bitmap::writer::BitmapFileIndexWriter;
+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::spec::DataType;
+use crate::{Error, Result};
+
+pub(crate) const BITMAP_INDEX: &str = "bitmap";
+pub(crate) const BLOOM_FILTER_INDEX: &str = "bloom-filter";
+
+#[derive(Clone, Copy)]
+enum BuiltinFileIndexer {
+ Bitmap,
+ BloomFilter,
+}
+
+impl BuiltinFileIndexer {
+ fn from_identifier(identifier: &str) -> Result<Self> {
+ match identifier {
+ BITMAP_INDEX => Ok(Self::Bitmap),
+ BLOOM_FILTER_INDEX => Ok(Self::BloomFilter),
+ _ => Err(Error::Unsupported {
+ message: format!("Unknown file index identifier:
{identifier}"),
+ }),
+ }
+ }
+}
+
+/// Factory for the file index implementations built into this crate.
+pub(crate) struct FileIndexerFactory;
+
+impl FileIndexerFactory {
+ pub(crate) fn create_writer(
+ identifier: &str,
+ data_type: DataType,
+ options: &Options,
+ ) -> Result<Box<dyn FileIndexWriter>> {
+ match BuiltinFileIndexer::from_identifier(identifier)? {
+ BuiltinFileIndexer::Bitmap =>
Ok(Box::new(BitmapFileIndexWriter::try_new(
+ data_type, options,
+ )?)),
+ BuiltinFileIndexer::BloomFilter => {
+ Ok(Box::new(BloomFilterWriter::try_new(data_type, options)?))
+ }
+ }
+ }
+
+ pub(crate) fn create_reader(
+ identifier: &str,
+ data_type: DataType,
+ serialized: Bytes,
+ ) -> Result<Box<dyn FileIndexReader>> {
+ match BuiltinFileIndexer::from_identifier(identifier)? {
+ BuiltinFileIndexer::Bitmap =>
Ok(Box::new(BitmapFileIndexReader::try_new(
+ data_type, serialized,
+ )?)),
+ BuiltinFileIndexer::BloomFilter => {
+ Ok(Box::new(BloomFilterReader::try_new(data_type,
serialized)?))
+ }
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{BinaryType, BooleanType, Datum, IntType};
+
+ fn int_type() -> DataType {
+ DataType::Int(IntType::new())
+ }
+
+ #[test]
+ fn test_builtin_writers_track_empty_rows_consistently() {
+ for identifier in [BITMAP_INDEX, BLOOM_FILTER_INDEX] {
+ let mut writer =
+ FileIndexerFactory::create_writer(identifier, int_type(),
&Options::new()).unwrap();
+
+ assert!(writer.empty(), "{identifier}");
+ writer.serialized_bytes().unwrap();
+ assert!(writer.empty(), "{identifier}");
+
+ writer.write(None).unwrap();
+ assert!(!writer.empty(), "{identifier}");
+ writer.serialized_bytes().unwrap();
+ }
+ }
+
+ #[test]
+ fn test_factory_delegates_type_and_option_validation() {
+ let mut bloom_options = Options::new();
+ bloom_options.set("items", "0");
+ assert!(matches!(
+ FileIndexerFactory::create_writer(BLOOM_FILTER_INDEX, int_type(),
&bloom_options),
+ Err(Error::ConfigInvalid { .. })
+ ));
+
+ let mut bitmap_options = Options::new();
+ bitmap_options.set("version", "1");
+ assert!(matches!(
+ FileIndexerFactory::create_writer(BITMAP_INDEX, int_type(),
&bitmap_options),
+ Err(Error::Unsupported { .. })
+ ));
+
+ assert!(matches!(
+ FileIndexerFactory::create_writer(
+ BITMAP_INDEX,
+ DataType::Binary(BinaryType::new(4).unwrap()),
+ &Options::new()
+ ),
+ Err(Error::Unsupported { .. })
+ ));
+ assert!(matches!(
+ FileIndexerFactory::create_writer(
+ BLOOM_FILTER_INDEX,
+ DataType::Boolean(BooleanType::new()),
+ &Options::new()
+ ),
+ Err(Error::Unsupported { .. })
+ ));
+ }
+
+ #[test]
+ fn test_unknown_identifier_is_rejected() {
+ assert!(matches!(
+ FileIndexerFactory::create_writer("unknown", int_type(),
&Options::new()),
+ Err(Error::Unsupported { .. })
+ ));
+ assert!(matches!(
+ FileIndexerFactory::create_reader("unknown", int_type(),
Bytes::new()),
+ Err(Error::Unsupported { .. })
+ ));
+ }
+
+ #[test]
+ fn test_writer_rejects_mismatched_datum() {
+ for identifier in [BITMAP_INDEX, BLOOM_FILTER_INDEX] {
+ let mut writer =
+ FileIndexerFactory::create_writer(identifier, int_type(),
&Options::new()).unwrap();
+
+ assert!(matches!(
+ writer.write(Some(&Datum::Long(1))),
+ Err(Error::DataInvalid { .. })
+ ));
+ assert!(writer.empty(), "{identifier}");
+ }
+ }
+}
diff --git a/crates/paimon/src/file_index/mod.rs
b/crates/paimon/src/file_index/mod.rs
index 48616abb..c070ebdc 100644
--- a/crates/paimon/src/file_index/mod.rs
+++ b/crates/paimon/src/file_index/mod.rs
@@ -16,7 +16,7 @@
// under the License.
// Concrete readers/writers and predicate plumbing stay crate-private until
-// factory, data-writer, and scan integration land in later changes.
+// data-writer and scan integration land in later changes.
#[allow(dead_code)]
pub(crate) mod bitmap;
#[allow(dead_code)]
@@ -28,5 +28,9 @@ pub(crate) mod file_index_predicate;
pub(crate) mod file_index_reader;
#[allow(dead_code)]
pub(crate) mod file_index_result;
+#[allow(dead_code)]
+pub(crate) mod file_index_writer;
+#[allow(dead_code)]
+pub(crate) mod file_indexer_factory;
pub use file_index_format::*;
