Remove bloom filters from TPC-H data

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"] }

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()
                 );

benchmarks/lance-bench/src/main.rs

@@ -218,23 +218,15 @@ async fn generate_lance_data<B: Benchmark + ?Sized>(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?;

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<PathBuf> {
         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<ArrayRef> = 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::<Canonical>(&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::<Canonical>(&mut ctx)?;
+                    Ok(canonical.into_array())
+                }
+            })
+            .into_array_stream()?
+            .try_collect()
+            .await?;
 
         Ok(ChunkedArray::from_iter(chunks).into_array())
     }

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<Pattern> {
         Some(
-            format!("{}_*.{}", table_name, format.ext())
+            format!("{}.{}", table_name, format.ext())
                 .parse()
                 .expect("valid glob pattern"),
         )