This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 5cd27ec3 perf(core): write the base file bloom filter only when asked
(#933)
5cd27ec3 is described below
commit 5cd27ec350da5f8390a40059d35b2c9ba33c5975
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Oct 30 11:38:20 2026 -0700
perf(core): write the base file bloom filter only when asked (#933)
---
crates/core/src/write/append.rs | 17 ++++++++++++-----
crates/core/tests/spark_parity_tests.rs | 20 ++++++++++++++++++--
crates/core/tests/table_write_tests.rs | 30 +++++++++++++++++++++++++++---
scripts/parity/README.md | 4 +++-
4 files changed, 60 insertions(+), 11 deletions(-)
diff --git a/crates/core/src/write/append.rs b/crates/core/src/write/append.rs
index c3831d1b..e427a53b 100644
--- a/crates/core/src/write/append.rs
+++ b/crates/core/src/write/append.rs
@@ -736,8 +736,11 @@ pub(crate) fn parquet_writer_props(table: &Table) ->
parquet::file::properties::
.build()
}
-/// The record-key bloom filter a base file footer carries, resolved as Java's
-/// `HoodieFileWriterFactory#enableBloomFilter` and `#createBloomFilter` do.
+/// The record-key bloom filter a base file footer carries, sized as Java's
+/// `HoodieFileWriterFactory#createBloomFilter` does. Off unless
+/// `hoodie.parquet.bloom.filter.enabled` is true or the index type is a bloom
+/// index: the record index locates keys without it, and hashing every key
+/// costs a large share of a bulk load.
pub(crate) fn base_file_bloom_filter_config(table: &Table) ->
Option<BloomFilterConfig> {
let populates_meta_fields: bool = table
.hudi_configs
@@ -746,7 +749,7 @@ pub(crate) fn base_file_bloom_filter_config(table: &Table)
-> Option<BloomFilter
let options = table.hudi_configs.as_options();
let enabled = options
.get("hoodie.parquet.bloom.filter.enabled")
- .is_none_or(|v| v.eq_ignore_ascii_case("true"))
+ .is_some_and(|v| v.eq_ignore_ascii_case("true"))
|| options
.get("hoodie.index.type")
.is_some_and(|v| v.contains("BLOOM"));
@@ -1424,7 +1427,8 @@ mod tests {
assert_eq!(rows, 64);
}
- /// The base file bloom filter follows Java's enable rule and sizing
configs.
+ /// The base file bloom filter is off by default, on when asked for or when
+ /// the index type needs it, and sized by Java's configs.
#[tokio::test]
async fn test_base_file_bloom_filter_config_resolution() {
async fn config(options: &[(&str, &str)]) -> Option<BloomFilterConfig>
{
@@ -1437,7 +1441,9 @@ mod tests {
}
base_file_bloom_filter_config(&builder.create().await.unwrap())
}
- assert_eq!(config(&[]).await, Some(BloomFilterConfig::default()));
+ assert_eq!(config(&[]).await, None);
+ let enabled = ("hoodie.parquet.bloom.filter.enabled", "true");
+ assert_eq!(config(&[enabled]).await,
Some(BloomFilterConfig::default()));
let disabled = ("hoodie.parquet.bloom.filter.enabled", "false");
assert_eq!(config(&[disabled]).await, None);
assert_eq!(
@@ -1446,6 +1452,7 @@ mod tests {
);
assert_eq!(
config(&[
+ enabled,
("hoodie.bloom.index.filter.type", "simple"),
("hoodie.index.bloom.num_entries", "100"),
("hoodie.index.bloom.fpp", "0.01"),
diff --git a/crates/core/tests/spark_parity_tests.rs
b/crates/core/tests/spark_parity_tests.rs
index 59281d24..b60fb568 100644
--- a/crates/core/tests/spark_parity_tests.rs
+++ b/crates/core/tests/spark_parity_tests.rs
@@ -276,6 +276,15 @@ fn batch(rows: &[Row]) -> RecordBatch {
}
async fn rs_create(path: &Path, table_type: TableTypeValue, partitioned: bool)
-> Table {
+ rs_create_with(path, table_type, partitioned, &[]).await
+}
+
+async fn rs_create_with(
+ path: &Path,
+ table_type: TableTypeValue,
+ partitioned: bool,
+ options: &[(&str, &str)],
+) -> Table {
let mut builder = Table::create(path.to_str().unwrap())
.with_table_name("parity")
.with_table_type(table_type)
@@ -284,9 +293,16 @@ async fn rs_create(path: &Path, table_type:
TableTypeValue, partitioned: bool) -
if partitioned {
builder = builder.with_partition_fields(["part"]);
}
+ for (key, value) in options {
+ builder = builder.with_option(key, *value);
+ }
builder.create().await.unwrap()
}
+/// Java's bloom index reads the record-key bloom filter from each base file
+/// footer, which rs writes only when asked to.
+const BLOOM_FOOTER: (&str, &str) = ("hoodie.parquet.bloom.filter.enabled",
"true");
+
async fn rs_read(path: &Path) -> Vec<Row> {
rs_read_with(path, ReadOptions::new()).await
}
@@ -1096,7 +1112,7 @@ async fn parity_i_java_index_lookups_on_rs_tables() {
for index in ["SIMPLE", "BLOOM", "GLOBAL_BLOOM"] {
let dir = tempdir().unwrap();
let path = dir.path().join("t");
- let mut table = rs_create(&path, table_type.clone(), true).await;
+ let mut table = rs_create_with(&path, table_type.clone(), true,
&[BLOOM_FOOTER]).await;
table
.append([batch(&[
row("a", Some("sf"), 1, 10),
@@ -1167,7 +1183,7 @@ async fn parity_i_java_index_lookups_on_rs_tables() {
// packing off, a missed lookup writes a duplicate instead of merging.
let dir = tempdir().unwrap();
let path = dir.path().join("t");
- let mut table = rs_create(&path, TableTypeValue::CopyOnWrite, true).await;
+ let mut table = rs_create_with(&path, TableTypeValue::CopyOnWrite, true,
&[BLOOM_FOOTER]).await;
table
.append([batch(&[
row("\u{FF21}", Some("sf"), 1, 10),
diff --git a/crates/core/tests/table_write_tests.rs
b/crates/core/tests/table_write_tests.rs
index de3e714d..41a00d3e 100644
--- a/crates/core/tests/table_write_tests.rs
+++ b/crates/core/tests/table_write_tests.rs
@@ -4326,9 +4326,10 @@ fn base_file_footers(dir: &std::path::Path) ->
Vec<(HashMap<String, String>, Vec
footers
}
-/// Every base file carries the footer Java's bloom index reads: the bloom
-/// filter at Java's default sizing holding every record key, its type code,
-/// and the record key range, which an empty file leaves out.
+/// With the bloom filter enabled, every base file carries the footer Java's
+/// bloom index reads: the bloom filter at Java's default sizing holding every
+/// record key, its type code, and the record key range, which an empty file
+/// leaves out.
#[tokio::test]
async fn test_base_file_footer_carries_bloom_filter_and_key_range() {
for table_type in [TableTypeValue::CopyOnWrite,
TableTypeValue::MergeOnRead] {
@@ -4338,6 +4339,7 @@ async fn
test_base_file_footer_carries_bloom_filter_and_key_range() {
.with_table_type(table_type.clone())
.with_partition_fields(["city"])
.with_record_key_fields(["id"])
+ .with_option("hoodie.parquet.bloom.filter.enabled", "true")
.create()
.await
.unwrap();
@@ -4399,6 +4401,28 @@ fn assert_record_key_footer(footer: &HashMap<String,
String>, keys: &[String]) {
assert_eq!(footer.get("hoodie_max_record_key"), keys.iter().max());
}
+/// By default a base file footer holds no bloom filter and no key range.
+#[tokio::test]
+async fn test_base_file_footer_omits_bloom_filter_by_default() {
+ let dir = tempdir().unwrap();
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("trips")
+ .with_record_key_fields(["id"])
+ .create()
+ .await
+ .unwrap();
+ table.append([sample_batch()]).await.unwrap();
+ let footers = base_file_footers(dir.path());
+ assert!(!footers.is_empty());
+ for (footer, _) in &footers {
+ assert!(
+ footer.keys().all(|key| key == "ARROW:schema"),
+ "{:?}",
+ footer.keys().collect::<Vec<_>>()
+ );
+ }
+}
+
/// A table without meta fields writes no bloom filter, as Java does.
#[tokio::test]
async fn test_base_file_footer_omits_bloom_filter_without_meta_fields() {
diff --git a/scripts/parity/README.md b/scripts/parity/README.md
index 0c694708..69683563 100644
--- a/scripts/parity/README.md
+++ b/scripts/parity/README.md
@@ -38,7 +38,9 @@ Cross-checks hudi-rs against the Apache Hudi Spark
writer/reader:
- **I** — Spark upserts with SIMPLE, BLOOM and GLOBAL_BLOOM indexes on
rs-written COW/MOR tables find every key rs wrote (base files, inserts,
updates) and re-insert one rs deleted without duplicates; includes keys
- whose UTF-8 and UTF-16 orders differ, against the bloom footer key range
+ whose UTF-8 and UTF-16 orders differ, against the bloom footer key range.
+ These tables set `hoodie.parquet.bloom.filter.enabled=true`: rs leaves the
+ footer out by default, and Java's bloom index needs it
- **J** — Spark SQL UPDATE/DELETE/INSERT on rs-written COW/MOR tables
registered in a non-default database, then rs writes on top
