Skip to content

Commit f0bbd0a

Browse files
authored
feat: support core Lumina index build (#347)
1 parent 00a3660 commit f0bbd0a

5 files changed

Lines changed: 1866 additions & 52 deletions

File tree

.github/workflows/ci.yml

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,16 @@ jobs:
120120
pip install lumina-data
121121
echo "LUMINA_LIB_PATH=$(python3 -c 'import lumina_data; print(lumina_data.__path__[0])')/lib/liblumina_py.so" >> $GITHUB_ENV
122122
123+
- name: Core Lumina Native Build Test
124+
run: >
125+
cargo test -p paimon
126+
table::lumina_index_build_builder::tests::test_execute_writes_lumina_index_manifest
127+
--features fulltext,vortex
128+
-- --ignored --exact
129+
env:
130+
RUST_LOG: DEBUG
131+
RUST_BACKTRACE: full
132+
123133
- name: DataFusion Integration Test
124134
run: cargo test -p paimon-datafusion --all-targets
125135
env:

crates/paimon/src/spec/core_options.rs

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ use std::collections::{HashMap, HashSet};
2020
const DELETION_VECTORS_ENABLED_OPTION: &str = "deletion-vectors.enabled";
2121
const DATA_EVOLUTION_ENABLED_OPTION: &str = "data-evolution.enabled";
2222
const GLOBAL_INDEX_ENABLED_OPTION: &str = "global-index.enabled";
23+
const GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION: &str = "global-index.row-count-per-shard";
2324
const SOURCE_SPLIT_TARGET_SIZE_OPTION: &str = "source.split.target-size";
2425
const SOURCE_SPLIT_OPEN_FILE_COST_OPTION: &str = "source.split.open-file-cost";
2526
const PARTITION_DEFAULT_NAME_OPTION: &str = "partition.default-name";
@@ -64,6 +65,7 @@ const DEFAULT_TARGET_FILE_SIZE: i64 = 256 * 1024 * 1024;
6465
const DEFAULT_WRITE_PARQUET_BUFFER_SIZE: i64 = 256 * 1024 * 1024;
6566
const DYNAMIC_BUCKET_TARGET_ROW_NUM_OPTION: &str = "dynamic-bucket.target-row-num";
6667
const DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM: i64 = 200_000;
68+
const DEFAULT_GLOBAL_INDEX_ROW_COUNT_PER_SHARD: i64 = 100_000;
6769
const BLOB_AS_DESCRIPTOR_OPTION: &str = "blob-as-descriptor";
6870
const BLOB_DESCRIPTOR_FIELD_OPTION: &str = "blob-descriptor-field";
6971

@@ -222,6 +224,22 @@ impl<'a> CoreOptions<'a> {
222224
.unwrap_or(false)
223225
}
224226

227+
pub fn global_index_row_count_per_shard(&self) -> crate::Result<i64> {
228+
let value = self
229+
.parse_i64_option(GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION)?
230+
.unwrap_or(DEFAULT_GLOBAL_INDEX_ROW_COUNT_PER_SHARD);
231+
if value <= 0 {
232+
return Err(crate::Error::DataInvalid {
233+
message: format!(
234+
"Option '{}' must be greater than 0, got: {}",
235+
GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION, value
236+
),
237+
source: None,
238+
});
239+
}
240+
Ok(value)
241+
}
242+
225243
pub fn source_split_target_size(&self) -> i64 {
226244
self.options
227245
.get(SOURCE_SPLIT_TARGET_SIZE_OPTION)
@@ -536,6 +554,10 @@ mod tests {
536554

537555
assert_eq!(core_options.source_split_target_size(), 128 * 1024 * 1024);
538556
assert_eq!(core_options.source_split_open_file_cost(), 4 * 1024 * 1024);
557+
assert_eq!(
558+
core_options.global_index_row_count_per_shard().unwrap(),
559+
100_000
560+
);
539561
}
540562

541563
#[test]
@@ -549,11 +571,36 @@ mod tests {
549571
SOURCE_SPLIT_OPEN_FILE_COST_OPTION.to_string(),
550572
"8 mb".to_string(),
551573
),
574+
(
575+
GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION.to_string(),
576+
"2048".to_string(),
577+
),
552578
]);
553579
let core_options = CoreOptions::new(&options);
554580

555581
assert_eq!(core_options.source_split_target_size(), 256 * 1024 * 1024);
556582
assert_eq!(core_options.source_split_open_file_cost(), 8 * 1024 * 1024);
583+
assert_eq!(
584+
core_options.global_index_row_count_per_shard().unwrap(),
585+
2048
586+
);
587+
}
588+
589+
#[test]
590+
fn test_global_index_row_count_per_shard_rejects_invalid_values() {
591+
for value in ["0", "-1", "abc"] {
592+
let options = HashMap::from([(
593+
GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION.to_string(),
594+
value.to_string(),
595+
)]);
596+
let core = CoreOptions::new(&options);
597+
598+
let err = core
599+
.global_index_row_count_per_shard()
600+
.expect_err("invalid rows-per-shard should fail");
601+
assert!(matches!(err, crate::Error::DataInvalid { message, .. }
602+
if message.contains(GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION)));
603+
}
557604
}
558605

559606
#[test]

0 commit comments

Comments
 (0)