Skip to content

Commit d4e7dca

Browse files
authored
support multiple globs in DuckDB and MultiFileDataSource (#7390)
## Summary Adds support for multiple globs in DuckDB, similar to Parquet, E.g. instead of `read_vortex('*.vortex')` now you can also do `read_vortex(['s3://bucket1/path/*.vortex', '/a/b/*.vortex'])` Note that multiple file systems are supported in a single scan. This is implemented and wired through into the underlying Rust MultiFileDataSource. ## Testing We add new DuckDB e2e tests for multi-glob Couldn't figure out a great way to test multiple file systems, just multiple globs --------- Signed-off-by: Andrew Duffy <andrew@a10y.dev>
1 parent 75b1062 commit d4e7dca

7 files changed

Lines changed: 197 additions & 68 deletions

File tree

‎benchmarks/datafusion-bench/src/main.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -317,8 +317,7 @@ async fn register_v2_tables<B: Benchmark + ?Sized>(
317317
};
318318

319319
let multi_ds = MultiFileDataSource::new(SESSION.clone())
320-
.with_filesystem(fs)
321-
.with_glob(glob_pattern)
320+
.with_glob(glob_pattern, Some(fs))
322321
.build()
323322
.await?;
324323

‎vortex-duckdb/cpp/table_function.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -383,6 +383,8 @@ extern "C" duckdb_state duckdb_vx_tfunc_register(duckdb_database ffi_db, const d
383383
auto &system_catalog = Catalog::GetSystemCatalog(*db);
384384
auto data = CatalogTransaction::GetSystemTransaction(*db);
385385
CreateTableFunctionInfo tf_info(tf);
386+
// Allow registering multiple overloads with the same name but different parameter types.
387+
tf_info.on_conflict = OnCreateConflict::ALTER_ON_CONFLICT;
386388
system_catalog.CreateFunction(data, tf_info);
387389
} catch (...) {
388390
return DuckDBError;

‎vortex-duckdb/src/e2e_test/vortex_scan_test.rs‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -350,6 +350,40 @@ fn test_vortex_scan_multiple_files() {
350350
assert_eq!(total_sum, 21);
351351
}
352352

353+
#[test]
354+
fn test_vortex_scan_multiple_globs() {
355+
// Test scanning multiple directories using a list of glob patterns.
356+
let (tempdir1, tempdir2, _file1, _file2, _file3) = RUNTIME.block_on(async {
357+
let tempdir1 = tempfile::tempdir().unwrap();
358+
let tempdir2 = tempfile::tempdir().unwrap();
359+
360+
let file1 = write_vortex_file_to_dir(tempdir1.path(), "numbers", buffer![1i32, 2, 3]).await;
361+
let file2 = write_vortex_file_to_dir(tempdir1.path(), "numbers", buffer![4i32, 5, 6]).await;
362+
let file3 =
363+
write_vortex_file_to_dir(tempdir2.path(), "numbers", buffer![7i32, 8, 9, 10]).await;
364+
365+
(tempdir1, tempdir2, file1, file2, file3)
366+
});
367+
368+
// Create glob patterns for each directory.
369+
let glob_pattern1 = format!("{}/*.vortex", tempdir1.path().display());
370+
let glob_pattern2 = format!("{}/*.vortex", tempdir2.path().display());
371+
372+
// Scan files from both directories using a list of globs.
373+
let conn = database_connection();
374+
let result = conn
375+
.query(&format!(
376+
"SELECT SUM(numbers) FROM read_vortex(['{glob_pattern1}', '{glob_pattern2}'])"
377+
))
378+
.unwrap();
379+
let chunk = result.into_iter().next().unwrap();
380+
let vec = chunk.get_vector(0);
381+
let total_sum = vec.as_slice_with_len::<i64>(chunk.len().as_())[0];
382+
383+
// 1+2+3 + 4+5+6 + 7+8+9+10 = 55
384+
assert_eq!(total_sum, 55);
385+
}
386+
353387
#[test]
354388
fn test_vortex_scan_over_http() {
355389
let file = RUNTIME.block_on(async {

‎vortex-duckdb/src/lib.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ use crate::duckdb::DatabaseRef;
2121
use crate::duckdb::LogicalType;
2222
use crate::duckdb::Value;
2323
use crate::multi_file::VortexMultiFileScan;
24+
use crate::multi_file::VortexMultiFileScanList;
2425

2526
mod convert;
2627
mod datasource;
@@ -56,6 +57,9 @@ pub fn initialize(db: &DatabaseRef) -> VortexResult<()> {
5657
)?;
5758
db.register_table_function::<VortexMultiFileScan>(c"vortex_scan")?;
5859
db.register_table_function::<VortexMultiFileScan>(c"read_vortex")?;
60+
// Register list overloads for multi-glob scanning (e.g., read_vortex(['a.vortex', 'b.vortex']))
61+
db.register_table_function::<VortexMultiFileScanList>(c"vortex_scan")?;
62+
db.register_table_function::<VortexMultiFileScanList>(c"read_vortex")?;
5963
db.register_copy_function::<VortexCopyFunction>(c"vortex", c"vortex")
6064
}
6165

‎vortex-duckdb/src/multi_file.rs‎

Lines changed: 82 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -5,18 +5,23 @@ use std::path::Path;
55
use std::path::absolute;
66
use std::sync::Arc;
77

8+
use itertools::Itertools;
89
use url::Url;
910
use vortex::error::VortexResult;
11+
use vortex::error::vortex_bail;
1012
use vortex::error::vortex_err;
1113
use vortex::file::multi::MultiFileDataSource;
14+
use vortex::io::filesystem::FileSystemRef;
1215
use vortex::io::runtime::BlockingRuntime;
1316
use vortex::scan::DataSourceRef;
17+
use vortex_utils::aliases::hash_map::HashMap;
1418

1519
use crate::RUNTIME;
1620
use crate::SESSION;
1721
use crate::datasource::DataSourceTableFunction;
1822
use crate::duckdb::BindInputRef;
1923
use crate::duckdb::ClientContextRef;
24+
use crate::duckdb::ExtractedValue;
2025
use crate::duckdb::LogicalType;
2126
use crate::filesystem::resolve_filesystem;
2227

@@ -60,34 +65,95 @@ fn normalize_path(path: std::path::PathBuf) -> std::path::PathBuf {
6065
#[derive(Debug)]
6166
pub struct VortexMultiFileScan;
6267

68+
/// Variant of [`VortexMultiFileScan`] that accepts a list of globs.
69+
///
70+
/// This is registered as a separate overload to handle `read_vortex(['a.vortex', 'b.vortex'])`.
71+
#[derive(Debug)]
72+
pub struct VortexMultiFileScanList;
73+
6374
impl DataSourceTableFunction for VortexMultiFileScan {
6475
fn parameters() -> Vec<LogicalType> {
6576
vec![LogicalType::varchar()]
6677
}
6778

6879
fn bind(ctx: &ClientContextRef, input: &BindInputRef) -> VortexResult<DataSourceRef> {
69-
let glob_url_parameter = input
70-
.get_parameter(0)
71-
.ok_or_else(|| vortex_err!("Missing file glob parameter"))?;
80+
bind_multi_file_scan(ctx, input)
81+
}
82+
}
83+
84+
impl DataSourceTableFunction for VortexMultiFileScanList {
85+
fn parameters() -> Vec<LogicalType> {
86+
vec![
87+
LogicalType::list_type(LogicalType::varchar())
88+
.unwrap_or_else(|_| unreachable!("creating list<varchar> type should not fail")),
89+
]
90+
}
91+
92+
fn bind(ctx: &ClientContextRef, input: &BindInputRef) -> VortexResult<DataSourceRef> {
93+
bind_multi_file_scan(ctx, input)
94+
}
95+
}
7296

73-
// Parse the URL and separate the base URL (keep scheme, host, etc.) from the path.
74-
let glob_url_string = glob_url_parameter.as_string();
75-
let glob_url_str = glob_url_string.as_str();
76-
let glob_url = parse_glob_url(glob_url_str)?;
97+
/// Shared bind logic for both single-glob and multi-glob variants.
98+
fn bind_multi_file_scan(
99+
ctx: &ClientContextRef,
100+
input: &BindInputRef,
101+
) -> VortexResult<DataSourceRef> {
102+
let glob_url_parameter = input
103+
.get_parameter(0)
104+
.ok_or_else(|| vortex_err!("Missing file glob parameter"))?;
77105

106+
// The input to the table function can either be a single glob, or a List of glob patterns.
107+
let glob_strings: Vec<String> = match glob_url_parameter.extract() {
108+
ExtractedValue::Varchar(glob) => {
109+
vec![glob.to_string()]
110+
}
111+
ExtractedValue::List(globs) => globs
112+
.into_iter()
113+
.map(|glob| {
114+
let ExtractedValue::Varchar(string) = glob.extract() else {
115+
vortex_bail!("list element must be Varchar type")
116+
};
117+
118+
Ok(string.to_string())
119+
})
120+
.try_collect()?,
121+
_ => vortex_bail!("Invalid argument to read_vortex table function"),
122+
};
123+
124+
// Parse each glob URL and resolve its filesystem.
125+
let mut glob_urls: Vec<Url> = Vec::with_capacity(glob_strings.len());
126+
for glob_str in &glob_strings {
127+
glob_urls.push(parse_glob_url(glob_str)?);
128+
}
129+
130+
// Cache filesystems by base URL to avoid resolving the same filesystem multiple times.
131+
let mut fs_cache: HashMap<Url, FileSystemRef> = HashMap::new();
132+
for glob_url in &glob_urls {
78133
let mut base_url = glob_url.clone();
79134
base_url.set_path("");
135+
if !fs_cache.contains_key(&base_url) {
136+
let fs = resolve_filesystem(&base_url, ctx)?;
137+
fs_cache.insert(base_url, fs);
138+
}
139+
}
80140

81-
let fs = resolve_filesystem(&base_url, ctx)?;
141+
RUNTIME.block_on(async {
142+
let mut builder = MultiFileDataSource::new(SESSION.clone());
82143

83-
RUNTIME.block_on(async {
84-
let builder = MultiFileDataSource::new(SESSION.clone())
85-
.with_filesystem(fs)
86-
.with_glob(glob_url.path());
87-
let ds = builder.build().await?;
88-
VortexResult::Ok(Arc::new(ds) as DataSourceRef)
89-
})
90-
}
144+
for glob_url in &glob_urls {
145+
let mut base_url = glob_url.clone();
146+
base_url.set_path("");
147+
let fs = fs_cache
148+
.get(&base_url)
149+
.map(Arc::clone)
150+
.unwrap_or_else(|| unreachable!("fs should be cached for all base URLs"));
151+
builder = builder.with_glob(glob_url.path(), Some(fs));
152+
}
153+
154+
let ds = builder.build().await?;
155+
Ok(Arc::new(ds) as DataSourceRef)
156+
})
91157
}
92158

93159
#[cfg(test)]

‎vortex-file/public-api.lock‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,7 @@ pub async fn vortex_file::multi::MultiFileDataSource::build(self) -> vortex_erro
1010

1111
pub fn vortex_file::multi::MultiFileDataSource::new(session: vortex_session::VortexSession) -> Self
1212

13-
pub fn vortex_file::multi::MultiFileDataSource::with_filesystem(self, fs: vortex_io::filesystem::FileSystemRef) -> Self
14-
15-
pub fn vortex_file::multi::MultiFileDataSource::with_glob(self, glob: impl core::convert::Into<alloc::string::String>) -> Self
13+
pub fn vortex_file::multi::MultiFileDataSource::with_glob(self, glob: impl core::convert::Into<alloc::string::String>, fs: core::option::Option<vortex_io::filesystem::FileSystemRef>) -> Self
1614

1715
pub fn vortex_file::multi::MultiFileDataSource::with_open_options(self, f: impl core::ops::function::Fn(vortex_file::VortexOpenOptions) -> vortex_file::VortexOpenOptions + core::marker::Send + core::marker::Sync + 'static) -> Self
1816

0 commit comments

Comments
 (0)