From 6c3dbe13e39a9dccd4f9272b4070e3bc710903b9 Mon Sep 17 00:00:00 2001 From: Adam Gutglick Date: Wed, 29 Jul 2026 09:33:44 +0100 Subject: [PATCH] Remove bloom filters from TPC-H data Signed-off-by: Adam Gutglick --- Cargo.lock | 1 + benchmarks/lance-bench/Cargo.toml | 1 + benchmarks/lance-bench/src/convert.rs | 25 +-- benchmarks/lance-bench/src/main.rs | 14 +- vortex-bench/src/datasets/tpch_l_comment.rs | 53 +++---- vortex-bench/src/tpch/benchmark.rs | 5 +- vortex-bench/src/tpch/tpchgen.rs | 163 ++++++-------------- 7 files changed, 91 insertions(+), 171 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 51188781c9a..a1ece6159e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4666,6 +4666,7 @@ dependencies = [ "async-trait", "clap", "futures", + "glob", "lance", "lance-encoding", "parquet 58.4.0", diff --git a/benchmarks/lance-bench/Cargo.toml b/benchmarks/lance-bench/Cargo.toml index 65816db3b50..89bffc20c28 100644 --- a/benchmarks/lance-bench/Cargo.toml +++ b/benchmarks/lance-bench/Cargo.toml @@ -23,6 +23,7 @@ arrow-cast = { version = "58" } async-trait = { workspace = true } clap = { workspace = true, features = ["derive"] } futures = { workspace = true } +glob = { workspace = true } parquet = { version = "58" } tempfile = { workspace = true } tokio = { workspace = true, features = ["full"] } diff --git a/benchmarks/lance-bench/src/convert.rs b/benchmarks/lance-bench/src/convert.rs index ac1db099953..6eb6bf0ffec 100644 --- a/benchmarks/lance-bench/src/convert.rs +++ b/benchmarks/lance-bench/src/convert.rs @@ -9,6 +9,7 @@ use std::sync::Arc; use anyhow::anyhow; use arrow_cast::cast; +use glob::Pattern; use lance::dataset::Dataset as LanceDataset; use lance::dataset::WriteParams; use lance::deps::arrow_array::RecordBatch; @@ -101,15 +102,19 @@ pub async fn convert_parquet_to_lance<'p>( parquet_dir: &'p Path, lance_dir: &'p Path, dataset_name: &str, - file_prefix: Option<&str>, + file_pattern: Option<&str>, convert_utf8view: bool, ) -> anyhow::Result<()> { // let lance_dir = lance_dir.to_path_buf(); // let parquet_dir = parquet_dir.to_path_buf(); - // let file_prefix = file_prefix.to_owned(); + // let file_pattern = file_pattern.to_owned(); // let dataset_name = dataset_name.to_string(); let dataset_path = lance_dir.join(format!("{}.lance", dataset_name)); + let file_pattern = file_pattern + .map(Pattern::new) + .transpose() + .map_err(|e| anyhow!("Invalid Parquet file pattern: {e}"))?; // Use idempotent pattern to avoid reprocessing idempotent_async( @@ -123,14 +128,12 @@ pub async fn convert_parquet_to_lance<'p>( .filter(|entry| { let path = entry.path(); if path.extension().is_some_and(|e| e == "parquet") { - if let Some(prefix) = file_prefix { - // Check if file starts with the prefix - path.file_stem() + if let Some(pattern) = &file_pattern { + path.file_name() .and_then(|s| s.to_str()) - .map(|s| s.starts_with(prefix)) + .map(|s| pattern.matches(s)) .unwrap_or(false) } else { - // No prefix filter, accept all parquet files true } } else { @@ -142,11 +145,11 @@ pub async fn convert_parquet_to_lance<'p>( if parquet_files.is_empty() { anyhow::bail!( - "No Parquet files found{}in {}", - if let Some(p) = file_prefix { - format!(" with prefix '{}' ", p) + "No Parquet files found{} in {}", + if let Some(p) = file_pattern.as_ref() { + format!(" matching '{}' ", p.as_str()) } else { - " ".to_string() + String::new() }, parquet_dir.display() ); diff --git a/benchmarks/lance-bench/src/main.rs b/benchmarks/lance-bench/src/main.rs index 68830c35e22..487762ee9e7 100644 --- a/benchmarks/lance-bench/src/main.rs +++ b/benchmarks/lance-bench/src/main.rs @@ -218,23 +218,15 @@ async fn generate_lance_data(benchmark: &B) -> anyhow::Re // Convert each table to Lance format for table in benchmark.table_specs().iter() { - // Determine file prefix pattern for this table - // TPC-H/TPC-DS use {table}_ prefix, others may use the table name directly - let file_prefix = benchmark + let file_pattern = benchmark .pattern(table.name, Format::Parquet) - .and_then(|p| { - // Extract prefix from pattern like "customer_*.parquet" -> "customer_" - let pattern_str = p.as_str(); - pattern_str - .strip_suffix(&format!("*.{}", Format::Parquet.ext())) - .map(|s| s.to_string()) - }); + .map(|p| p.to_string()); convert_parquet_to_lance( &parquet_dir, &lance_dir, table.name, - file_prefix.as_deref(), + file_pattern.as_deref(), true, // Convert Utf8View to Utf8 for Lance compatibility ) .await?; diff --git a/vortex-bench/src/datasets/tpch_l_comment.rs b/vortex-bench/src/datasets/tpch_l_comment.rs index ebe6ff2e96a..55d8497c2b3 100644 --- a/vortex-bench/src/datasets/tpch_l_comment.rs +++ b/vortex-bench/src/datasets/tpch_l_comment.rs @@ -6,7 +6,6 @@ use std::path::PathBuf; use anyhow::Result; use async_trait::async_trait; use futures::TryStreamExt; -use glob::glob; use vortex::array::ArrayRef; use vortex::array::Canonical; use vortex::array::ExecutionCtx; @@ -39,12 +38,10 @@ async fn ensure_tpch_parquet() -> Result { generate_tpch_tables(options).await?; } - // Return the first lineitem parquet file - let pattern = data_dir.join("lineitem_*.parquet"); - glob(pattern.to_string_lossy().as_ref())? - .next() - .ok_or_else(|| anyhow::anyhow!("No lineitem parquet files found"))? - .map_err(Into::into) + let path = data_dir.join("lineitem.parquet"); + path.exists() + .then_some(path) + .ok_or_else(|| anyhow::anyhow!("No lineitem parquet file found")) } #[async_trait] @@ -58,7 +55,7 @@ impl Dataset for TPCHLCommentChunked { let scale_factor_dir = base_path.join("1.0"); let data_dir = scale_factor_dir.join(Format::OnDiskVortex.name()); - // Generate TPC-H CSV data if it doesn't exist + // Generate TPC-H Vortex data if it doesn't exist if !data_dir.exists() { // Use blocking call like TPC-H benchmark does let options = TpchGenOptions::new("1.0".to_string(), scale_factor_dir) @@ -67,30 +64,22 @@ impl Dataset for TPCHLCommentChunked { futures::executor::block_on(generate_tpch_tables(options))?; } - let mut chunks: Vec = vec![]; - for path in glob( - data_dir - .join("lineitem_*.vortex") - .to_string_lossy() - .as_ref(), - )? { - let file = SESSION.open_options().open_path(path?).await?; - let file_chunks: Vec<_> = file - .scan()? - .with_projection(pack(vec![("l_comment", col("l_comment"))], NonNullable)) - .map({ - let ctx = ctx.clone(); - move |a| { - let mut ctx = ctx.clone(); - let canonical = a.execute::(&mut ctx)?; - Ok(canonical.into_array()) - } - }) - .into_array_stream()? - .try_collect() - .await?; - chunks.extend(file_chunks); - } + let path = data_dir.join("lineitem.vortex"); + let file = SESSION.open_options().open_path(path).await?; + let chunks: Vec<_> = file + .scan()? + .with_projection(pack(vec![("l_comment", col("l_comment"))], NonNullable)) + .map({ + let ctx = ctx.clone(); + move |a| { + let mut ctx = ctx.clone(); + let canonical = a.execute::(&mut ctx)?; + Ok(canonical.into_array()) + } + }) + .into_array_stream()? + .try_collect() + .await?; Ok(ChunkedArray::from_iter(chunks).into_array()) } diff --git a/vortex-bench/src/tpch/benchmark.rs b/vortex-bench/src/tpch/benchmark.rs index 4dfdaf1207d..fe0df1fee41 100644 --- a/vortex-bench/src/tpch/benchmark.rs +++ b/vortex-bench/src/tpch/benchmark.rs @@ -93,8 +93,7 @@ impl Benchmark for TpcHBenchmark { .to_file_path() .map_err(|_| anyhow::anyhow!("Invalid file URL: {}", self.data_url.as_str()))?; - let options = TpchGenOptions::new(self.scale_factor.clone(), base_data_dir) - .with_max_file_size_mb(Some(600)); + let options = TpchGenOptions::new(self.scale_factor.clone(), base_data_dir); tpchgen::generate_tpch_tables(options).await?; @@ -143,7 +142,7 @@ impl Benchmark for TpcHBenchmark { #[expect(clippy::expect_used)] fn pattern(&self, table_name: &str, format: Format) -> Option { Some( - format!("{}_*.{}", table_name, format.ext()) + format!("{}.{}", table_name, format.ext()) .parse() .expect("valid glob pattern"), ) diff --git a/vortex-bench/src/tpch/tpchgen.rs b/vortex-bench/src/tpch/tpchgen.rs index ba2646b365e..9efc8d98bfb 100644 --- a/vortex-bench/src/tpch/tpchgen.rs +++ b/vortex-bench/src/tpch/tpchgen.rs @@ -56,8 +56,6 @@ pub struct TpchGenOptions { pub format: Format, /// Batch size for streaming pub batch_size: usize, - /// The max size of uncompressed file .tbl that we should generate - pub max_file_size_mb: Option, } impl Default for TpchGenOptions { @@ -67,7 +65,6 @@ impl Default for TpchGenOptions { output_dir: "tpch".to_data_path(), format: Format::Parquet, batch_size: 8192 * 64, - max_file_size_mb: None, } } } @@ -90,11 +87,6 @@ impl TpchGenOptions { self.batch_size = batch_size; self } - - pub fn with_max_file_size_mb(mut self, max_file_size_mb: Option) -> Self { - self.max_file_size_mb = max_file_size_mb; - self - } } /// Generate all TPC-H tables for a single scale factor @@ -125,11 +117,10 @@ pub async fn generate_tpch_tables(options: TpchGenOptions) -> Result<()> { ); let table_name = table_name.to_string(); - generate_table_files(table_name, *generator, options.clone()) + generate_table_file(table_name, *generator, options.clone()) }) .collect::>>()? .into_iter() - .flatten() .collect::>>>(); let limiter = Arc::new(Semaphore::new(MAX_CONCURRENT_FILES)); @@ -169,29 +160,12 @@ enum TableGenerator { LineItem, } -impl TableGenerator { - // Returns the size in MBs of the table at scale factor 1, if the table is constant size - // return None - fn uncompressed_data_size(&self) -> Option { - match self { - TableGenerator::Nation => None, - TableGenerator::Region => None, - TableGenerator::Part => Some(113), - TableGenerator::Supplier => Some(1), - TableGenerator::Customer => Some(23), - TableGenerator::PartSupp => Some(113), - TableGenerator::Orders => Some(164), - TableGenerator::LineItem => Some(725), - } - } -} - -/// Generate files for a specific table in streaming fashion -fn generate_table_files( +/// Generate one file for a specific table in streaming fashion +fn generate_table_file( table_name: String, generator: TableGenerator, options: TpchGenOptions, -) -> Result>>> { +) -> Result>> { let write_format = match options.format { Format::Parquet | Format::OnDiskDuckDB => Format::Parquet, Format::OnDiskVortex => Format::OnDiskVortex, @@ -205,98 +179,60 @@ fn generate_table_files( fs::create_dir_all(&output_dir)?; - // Calculate number of partitions without creating expensive iterators - let num_parts = calculate_num_parts(generator, &options)?; - - let mut futures = Vec::new(); - - for partition_idx in 0..num_parts { - let output_file = output_dir.join(format!( - "{table_name}_{partition_idx}.{}", - write_format.ext() - )); - - // Clone necessary data for the async closure - let options_clone = options.clone(); - let table_name = table_name.to_string(); - - let future = async move { - let of = output_file.clone(); - idempotent_async(output_file.to_string_lossy().as_ref(), |path| async move { - info!( - "Generating {table_name} table as {write_format}, at {}", - of.to_string_lossy() - ); - - // Create the specific iterator for this partition only when we need to generate - let iter = create_single_batch_iterator(generator, &options_clone, partition_idx)?; - - // Create generator and process batches in streaming fashion - let schema = Arc::clone(iter.schema()); - - // Create writer based on format - let mut writer: Box = match write_format { - Format::Parquet => Box::new(ParquetWriter::new(path, schema).await?), - Format::OnDiskVortex => Box::new(VortexWriter::new( - path, - schema, - CompactionStrategy::Default, - )?), - Format::VortexCompact => Box::new(VortexWriter::new( - path, - schema, - CompactionStrategy::Compact, - )?), - _ => unreachable!(), - }; - - for batch in iter { - writer.write_batch(&batch).await?; - } - - writer.finalize().await?; - Ok::<(), anyhow::Error>(()) - }) - .await?; - Ok(()) - } - .boxed(); - - futures.push(future); - } - - Ok(futures) -} + let output_file = output_dir.join(format!("{table_name}.{}", write_format.ext())); -/// Calculate the number of partitions without creating expensive iterators -#[expect(clippy::cast_possible_truncation)] -fn calculate_num_parts(generator: TableGenerator, options: &TpchGenOptions) -> Result { - let scale_factor = options.scale_factor.parse::()?; + let future = async move { + let of = output_file.clone(); + idempotent_async(output_file.to_string_lossy().as_ref(), |path| async move { + info!( + "Generating {table_name} table as {write_format}, at {}", + of.to_string_lossy() + ); - let num_parts = if let Some((data_size, max_file_size)) = generator - .uncompressed_data_size() - .zip(options.max_file_size_mb) - { - #[expect(clippy::cast_precision_loss)] - let file_size = (data_size as f64 * scale_factor).ceil() as u64; - file_size.div_ceil(max_file_size) - } else { - 1 - }; + let iter = create_batch_iterator(generator, &options)?; + + // Create generator and process batches in streaming fashion + let schema = Arc::clone(iter.schema()); + + // Create writer based on format + let mut writer: Box = match write_format { + Format::Parquet => Box::new(ParquetWriter::new(path, schema).await?), + Format::OnDiskVortex => Box::new(VortexWriter::new( + path, + schema, + CompactionStrategy::Default, + )?), + Format::VortexCompact => Box::new(VortexWriter::new( + path, + schema, + CompactionStrategy::Compact, + )?), + _ => unreachable!(), + }; + + for batch in iter { + writer.write_batch(&batch).await?; + } + + writer.finalize().await?; + Ok::<(), anyhow::Error>(()) + }) + .await?; + Ok(()) + } + .boxed(); - Ok(num_parts as usize) + Ok(future) } -/// Create a single batch iterator for a specific partition -#[expect(clippy::cast_possible_truncation)] -fn create_single_batch_iterator( +/// Create a batch iterator for one full table +fn create_batch_iterator( generator: TableGenerator, options: &TpchGenOptions, - partition_idx: usize, ) -> Result> { let scale_factor = options.scale_factor.parse::()?; - let num_parts = calculate_num_parts(generator, options)? as i32; - let part = (partition_idx + 1) as i32; // 1-indexed + let part = 1; + let num_parts = 1; let batch_size = options.batch_size; let iterator: Box = match generator { @@ -370,7 +306,6 @@ impl ParquetWriter { let file = TokioFile::create(&path).await?; let props = WriterProperties::builder() .set_compression(Compression::SNAPPY) - .set_bloom_filter_enabled(true) .build(); let writer = AsyncArrowWriter::try_new(file, schema, Some(props))?; Ok(Self { writer })