From ab3a563eebfe4a8f731f72e9a6190fc97bf801c1 Mon Sep 17 00:00:00 2001 From: shyjsarah <44659226+shyjsarah@users.noreply.github.com> Date: Wed, 26 Aug 2026 00:00:28 -0700 Subject: [PATCH 1/4] feat: add native object table read support --- crates/integrations/datafusion/src/catalog.rs | 24 +- .../integrations/datafusion/src/table/mod.rs | 3 + .../datafusion/src/table/object.rs | 153 ++++++++++++ .../datafusion/tests/object_table.rs | 114 +++++++++ crates/paimon/src/catalog/filesystem.rs | 40 +++- crates/paimon/src/catalog/mod.rs | 22 +- .../paimon/src/catalog/rest/rest_catalog.rs | 12 + crates/paimon/src/spec/table_type.rs | 9 +- crates/paimon/src/table/mod.rs | 2 + crates/paimon/src/table/object_table.rs | 226 ++++++++++++++++++ crates/paimon/src/table/rest_env.rs | 113 +++++++-- crates/paimon/tests/rest_catalog_test.rs | 37 +++ 12 files changed, 721 insertions(+), 34 deletions(-) create mode 100644 crates/integrations/datafusion/src/table/object.rs create mode 100644 crates/integrations/datafusion/tests/object_table.rs create mode 100644 crates/paimon/src/table/object_table.rs diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index cbee1a6e..f410becc 100644 --- a/crates/integrations/datafusion/src/catalog.rs +++ b/crates/integrations/datafusion/src/catalog.rs @@ -40,7 +40,7 @@ use paimon::spec::TableType as PaimonTableType; use crate::error::to_datafusion_error; use crate::runtime::{await_with_runtime, block_on_with_runtime}; use crate::system_tables; -use crate::table::PaimonTableProvider; +use crate::table::{ObjectTableProvider, PaimonTableProvider}; use crate::{BlobReaderRegistry, DynamicOptions}; pub(crate) type SessionStateProvider = Arc Option + Send + Sync>; @@ -665,6 +665,25 @@ impl SchemaProvider for PaimonSchemaProvider { .clone(); await_with_runtime(async move { match catalog.load_table(&identifier).await { + Ok(paimon::catalog::LoadedTable::Object(table)) => { + if branch.is_some() { + return Err(plan_datafusion_err!( + "branches are not supported for 'object-table' tables ('{}')", + identifier.full_name() + )); + } + let session_options = dynamic_options + .read() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + paimon::spec::CoreOptions::new(&session_options) + .ensure_engine_can_serve(&identifier.full_name()) + .map_err(to_datafusion_error)?; + Ok(Some(Arc::new(ObjectTableProvider::try_new( + table, + schema_force_view_types, + )?) as Arc)) + } Ok(paimon::catalog::LoadedTable::External(external)) => { let declared = external.declared(); if branch.is_some() { @@ -855,6 +874,9 @@ impl SchemaProvider for PaimonSchemaProvider { block_on_with_runtime( async move { match catalog.load_table(&identifier).await { + Ok(paimon::catalog::LoadedTable::Object(_)) => { + branch.is_none() && !has_system_suffix + } Ok(paimon::catalog::LoadedTable::External(external)) => { let declared = external.declared(); // Paimon-only; `table()` rejects them here too. diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index c55b97e7..262c3217 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -46,6 +46,9 @@ use crate::filter_pushdown::{analyze_filters, classify_filter_pushdown}; use crate::physical_plan::PaimonTableScan; use crate::runtime::await_with_runtime; +mod object; +pub(crate) use object::ObjectTableProvider; + const PARQUET_FIELD_ID_META_KEY: &str = "PARQUET:field_id"; pub(crate) fn datafusion_read_fields(table: &Table) -> Vec { diff --git a/crates/integrations/datafusion/src/table/object.rs b/crates/integrations/datafusion/src/table/object.rs new file mode 100644 index 00000000..7d5d16af --- /dev/null +++ b/crates/integrations/datafusion/src/table/object.rs @@ -0,0 +1,153 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::sync::Arc; + +use async_trait::async_trait; +use datafusion::arrow::array::{ + new_null_array, ArrayRef, Int64Array, RecordBatch, StringViewArray, +}; +use datafusion::arrow::datatypes::SchemaRef; +use datafusion::catalog::Session; +use datafusion::datasource::memory::MemorySourceConfig; +use datafusion::datasource::{TableProvider, TableType}; +use datafusion::error::{DataFusionError, Result as DFResult}; +use datafusion::logical_expr::dml::InsertOp; +use datafusion::logical_expr::Expr; +use datafusion::physical_plan::ExecutionPlan; +use paimon::table::ObjectTable; + +use crate::error::to_datafusion_error; + +use super::datafusion_arrow_schema; + +/// DataFusion provider for a native read-only Paimon object table. +#[derive(Debug, Clone)] +pub(crate) struct ObjectTableProvider { + table: ObjectTable, + schema: SchemaRef, +} + +impl ObjectTableProvider { + pub(crate) fn try_new(table: ObjectTable, schema_force_view_types: bool) -> DFResult { + let schema = datafusion_arrow_schema(&ObjectTable::fields(), schema_force_view_types)?; + Ok(Self { table, schema }) + } +} + +#[async_trait] +impl TableProvider for ObjectTableProvider { + fn schema(&self) -> SchemaRef { + Arc::clone(&self.schema) + } + + fn table_type(&self) -> TableType { + TableType::Base + } + + async fn scan( + &self, + _state: &dyn Session, + projection: Option<&Vec>, + _filters: &[Expr], + limit: Option, + ) -> DFResult> { + let mut entries = self + .table + .list_objects() + .await + .map_err(to_datafusion_error)?; + if let Some(limit) = limit { + entries.truncate(limit); + } + + let paths = entries + .iter() + .map(|entry| entry.path().to_string()) + .collect::>(); + let names = entries + .iter() + .map(|entry| entry.name().to_string()) + .collect::>(); + let lengths = entries + .iter() + .map(|entry| entry.length()) + .collect::>(); + let mtimes = entries + .iter() + .map(|entry| entry.mtime()) + .collect::>(); + let atimes = entries + .iter() + .map(|entry| entry.atime()) + .collect::>(); + let owners = entries + .iter() + .map(|entry| entry.owner()) + .collect::>(); + let row_count = entries.len(); + + let string_array = |values: Vec, index: usize| -> ArrayRef { + if matches!( + self.schema.field(index).data_type(), + datafusion::arrow::datatypes::DataType::Utf8View + ) { + Arc::new(StringViewArray::from(values)) + } else { + Arc::new(datafusion::arrow::array::StringArray::from(values)) + } + }; + let batch = RecordBatch::try_new( + Arc::clone(&self.schema), + vec![ + string_array(paths, 0), + string_array(names, 1), + Arc::new(Int64Array::from(lengths)), + Arc::new(Int64Array::from(mtimes)), + Arc::new(Int64Array::from(atimes)), + if owners.iter().all(Option::is_none) { + new_null_array(self.schema.field(5).data_type(), row_count) + } else if matches!( + self.schema.field(5).data_type(), + datafusion::arrow::datatypes::DataType::Utf8View + ) { + Arc::new(StringViewArray::from(owners)) + } else { + Arc::new(datafusion::arrow::array::StringArray::from(owners)) + }, + ], + )?; + + Ok(MemorySourceConfig::try_new_exec( + &[vec![batch]], + Arc::clone(&self.schema), + projection.cloned(), + )?) + } + + async fn insert_into( + &self, + _state: &dyn Session, + _input: Arc, + _insert_op: InsertOp, + ) -> DFResult> { + Err(DataFusionError::NotImplemented(format!( + "Object table '{}' is read-only", + self.table.identifier().full_name() + ))) + } +} diff --git a/crates/integrations/datafusion/tests/object_table.rs b/crates/integrations/datafusion/tests/object_table.rs new file mode 100644 index 00000000..3300f49c --- /dev/null +++ b/crates/integrations/datafusion/tests/object_table.rs @@ -0,0 +1,114 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::HashMap; +use std::fs; +use std::sync::Arc; + +use datafusion::arrow::array::Array; +use datafusion::arrow::util::display::array_value_to_string; +use paimon::catalog::Identifier; +use paimon::spec::{BigIntType, DataType, Schema}; +use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options}; +use paimon_datafusion::SQLContext; +use tempfile::TempDir; + +fn object_table_schema(location: &str) -> Schema { + Schema::builder() + .column("ignored", DataType::BigInt(BigIntType::new())) + .option("type", "object-table") + .option("path", location) + .build() + .unwrap() +} + +#[tokio::test] +async fn object_table_lists_files_recursively() { + let object_dir = TempDir::new().unwrap(); + fs::write(object_dir.path().join("root.txt"), b"root").unwrap(); + fs::create_dir(object_dir.path().join("nested")).unwrap(); + fs::write(object_dir.path().join("nested/child.bin"), b"child").unwrap(); + + let location = format!("file://{}", object_dir.path().display()); + let mut options = Options::new(); + options.set(CatalogOptions::WAREHOUSE, "memory:/warehouse"); + let catalog = Arc::new(FileSystemCatalog::new(options).unwrap()); + catalog + .create_database("db", false, HashMap::new()) + .await + .unwrap(); + catalog + .create_table( + &Identifier::new("db", "objects"), + object_table_schema(&location), + false, + ) + .await + .unwrap(); + + let mut ctx = SQLContext::new(); + ctx.register_catalog("cat", catalog).await.unwrap(); + let batches = ctx + .sql( + "SELECT path, name, length, mtime > 0 AS has_mtime, atime, owner \ + FROM cat.db.objects ORDER BY path", + ) + .await + .unwrap() + .collect() + .await + .unwrap(); + + let mut rows = Vec::new(); + for batch in batches { + for row in 0..batch.num_rows() { + rows.push( + batch + .columns() + .iter() + .map(|column| { + if column.is_null(row) { + "NULL".to_string() + } else { + array_value_to_string(column.as_ref(), row).unwrap() + } + }) + .collect::>(), + ); + } + } + + assert_eq!( + rows, + vec![ + vec!["nested/child.bin", "child.bin", "5", "true", "0", "NULL"], + vec!["root.txt", "root.txt", "4", "true", "0", "NULL"], + ] + ); + + let error = ctx + .sql( + "INSERT INTO cat.db.objects \ + VALUES ('path', 'name', 1, 1, 0, NULL)", + ) + .await + .unwrap() + .collect() + .await + .unwrap_err(); + assert!(error.to_string().contains("read-only"), "{error}"); +} diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs index 0ebd8ab8..e5d84a6a 100644 --- a/crates/paimon/src/catalog/filesystem.rs +++ b/crates/paimon/src/catalog/filesystem.rs @@ -24,7 +24,7 @@ use std::collections::HashMap; use crate::catalog::{Catalog, Database, Identifier, DB_LOCATION_PROP, DB_SUFFIX}; use crate::common::{CatalogOptions, Options}; use crate::error::{ConfigInvalidSnafu, Error, Result}; -use crate::io::cache::create_local_cache; +use crate::io::cache::{create_local_cache, LocalCache}; use crate::io::FileIO; use crate::spec::{CoreOptions, Schema, TableSchema, TableType, TABLE_TYPE_OPTION}; use crate::table::{SchemaManager, Table}; @@ -66,6 +66,8 @@ fn make_path(parent: &str, child: &str) -> String { pub struct FileSystemCatalog { file_io: FileIO, warehouse: String, + options: Options, + local_cache: Option>, } impl FileSystemCatalog { @@ -105,12 +107,17 @@ impl FileSystemCatalog { let local_cache = create_local_cache(&options)?; let mut file_io_builder = FileIO::from_path(&warehouse)?.with_props(options.to_map().iter()); - if let Some(local_cache) = local_cache { - file_io_builder = file_io_builder.with_local_cache(local_cache); + if let Some(local_cache) = &local_cache { + file_io_builder = file_io_builder.with_local_cache(local_cache.clone()); } let file_io = file_io_builder.build()?; - Ok(Self { file_io, warehouse }) + Ok(Self { + file_io, + warehouse, + options, + local_cache, + }) } /// Get the warehouse path. @@ -123,6 +130,14 @@ impl FileSystemCatalog { &self.file_io } + fn build_file_io(&self, path: &str) -> Result { + let mut builder = FileIO::from_path(path)?.with_props(self.options.to_map().iter()); + if let Some(local_cache) = &self.local_cache { + builder = builder.with_local_cache(local_cache.clone()); + } + builder.build() + } + /// Get the path for a database (warehouse / `name` + [DB_SUFFIX]). fn database_path(&self, database_name: &str) -> String { make_path( @@ -360,6 +375,23 @@ impl Catalog for FileSystemCatalog { let (table_path, schema) = self.fetch_table_schema(identifier).await?; let options = CoreOptions::new(schema.options()); let declared = options.table_type()?; + if declared == crate::spec::TableType::ObjectTable { + let object_path = options + .path() + .filter(|path| !path.trim().is_empty()) + .ok_or_else(|| Error::ConfigInvalid { + message: format!( + "Object table '{}' requires a non-empty 'path' option", + identifier.full_name() + ), + })?; + return crate::table::ObjectTable::try_new( + self.build_file_io(object_path)?, + identifier.clone(), + &schema, + ) + .map(crate::catalog::LoadedTable::Object); + } if declared.requires_table_engine() { return crate::catalog::LoadedTable::external( declared, diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs index 2b91ff19..ebccc6a6 100644 --- a/crates/paimon/src/catalog/mod.rs +++ b/crates/paimon/src/catalog/mod.rs @@ -263,14 +263,16 @@ use async_trait::async_trait; use crate::api::PagedList; use crate::spec::{Partition, Schema, SchemaChange, TableType}; -use crate::table::Table; +use crate::table::{ObjectTable, Table}; /// Outcome of [`Catalog::load_table`]. #[derive(Debug)] pub enum LoadedTable { /// A constructed Paimon table (boxed: far larger than the other variant). Paimon(Box), - /// A table this reader cannot construct. + /// A native read-only object table. + Object(ObjectTable), + /// A table that needs a registered external engine. External(ExternalTableMetadata), } @@ -371,9 +373,10 @@ pub trait Catalog: Send + Sync { /// * [`crate::Error::TableNotExist`] - table does not exist. async fn get_table(&self, identifier: &Identifier) -> Result
; - /// Load a table, or classify it as [`LoadedTable::External`] when this - /// reader cannot construct it. One metadata round-trip either way, and the - /// outcome depends only on the table's own metadata. + /// Load a Paimon or native object table, or classify it as + /// [`LoadedTable::External`] when this reader cannot construct it. One + /// metadata round-trip either way, and the outcome depends only on the + /// table's own metadata. /// /// The default implementation classifies from the constructed table, so a /// catalog that only implements [`Catalog::get_table`] still fails closed. @@ -389,6 +392,14 @@ pub trait Catalog: Send + Sync { let table = self.get_table(identifier).await?; let options = crate::spec::CoreOptions::new(table.schema().options()); let declared = options.table_type()?; + if declared == TableType::ObjectTable { + return ObjectTable::try_new( + table.file_io().clone(), + identifier.clone(), + table.schema(), + ) + .map(LoadedTable::Object); + } if declared.requires_table_engine() { return LoadedTable::external(declared, &options, &identifier.full_name()); } @@ -526,6 +537,7 @@ pub trait Catalog: Send + Sync { async fn list_partitions(&self, identifier: &Identifier) -> Result> { match self.load_table(identifier).await? { LoadedTable::Paimon(table) => list_partitions_from_file_system(&table).await, + LoadedTable::Object(_) => Ok(Vec::new()), LoadedTable::External(external) => Err(Error::Unsupported { message: format!( "table '{}' is declared '{}', so it has no Paimon partitions to list", diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs b/crates/paimon/src/catalog/rest/rest_catalog.rs index be41cfc8..7559a369 100644 --- a/crates/paimon/src/catalog/rest/rest_catalog.rs +++ b/crates/paimon/src/catalog/rest/rest_catalog.rs @@ -220,6 +220,18 @@ impl Catalog for RESTCatalog { if let Some(schema) = response.schema.as_ref() { let options = crate::spec::CoreOptions::new(schema.options()); let declared = options.table_type()?; + if declared == crate::spec::TableType::ObjectTable { + return RESTEnv::build_object_table( + identifier, + response, + self.api.clone(), + self.options.clone(), + self.data_token_enabled, + self.local_cache.clone(), + ) + .await + .map(crate::catalog::LoadedTable::Object); + } if declared.requires_table_engine() { return crate::catalog::LoadedTable::external( declared, diff --git a/crates/paimon/src/spec/table_type.rs b/crates/paimon/src/spec/table_type.rs index e4a095d2..43166d80 100644 --- a/crates/paimon/src/spec/table_type.rs +++ b/crates/paimon/src/spec/table_type.rs @@ -53,11 +53,10 @@ impl TableType { } } - /// Whether this type needs an engine of its own (see - /// [`Catalog::load_table`](crate::catalog::Catalog::load_table)). Java - /// builds a dedicated table for each; this client has none, so reading one - /// as Paimon misreads it and writing could put Paimon snapshots over - /// foreign data. + /// Whether this type must not use the Paimon file-store reader (see + /// [`Catalog::load_table`](crate::catalog::Catalog::load_table)). These + /// types need either a dedicated native reader, such as object tables, or + /// a registered external engine. pub fn requires_table_engine(&self) -> bool { matches!( self, diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index b415f0dd..d970f546 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -57,6 +57,7 @@ mod kv_file_reader; mod kv_file_writer; mod lumina_index_build_builder; pub(crate) mod merge_tree_split_generator; +mod object_table; mod partition_filter; mod partition_stat; #[cfg(feature = "fulltext")] @@ -125,6 +126,7 @@ pub use incremental_scan::{ IncrementalPlan, IncrementalScan, IncrementalScanMode, IncrementalSplit, }; pub use lumina_index_build_builder::LuminaIndexBuildBuilder; +pub use object_table::{ObjectEntry, ObjectTable}; pub use partition_stat::PartitionStat; pub use postpone_bucket_plan::{PostponeBucketPlan, POSTPONE_BUCKET_PLAN_TOTAL_BUCKETS_FIELD}; pub use postpone_fixed_bucket_write::{ diff --git a/crates/paimon/src/table/object_table.rs b/crates/paimon/src/table/object_table.rs new file mode 100644 index 00000000..19cb64be --- /dev/null +++ b/crates/paimon/src/table/object_table.rs @@ -0,0 +1,226 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::catalog::Identifier; +use crate::io::FileIO; +use crate::spec::{ + BigIntType, CoreOptions, DataField, DataType, TableSchema, TableType, VarCharType, +}; +use crate::{Error, Result}; + +/// Metadata for one file exposed by an [`ObjectTable`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObjectEntry { + path: String, + name: String, + length: i64, + mtime: i64, + atime: i64, + owner: Option, +} + +impl ObjectEntry { + pub fn path(&self) -> &str { + &self.path + } + + pub fn name(&self) -> &str { + &self.name + } + + pub fn length(&self) -> i64 { + self.length + } + + pub fn mtime(&self) -> i64 { + self.mtime + } + + pub fn atime(&self) -> i64 { + self.atime + } + + pub fn owner(&self) -> Option<&str> { + self.owner.as_deref() + } +} + +/// Read-only view over the files below a configured object location. +#[derive(Debug, Clone)] +pub struct ObjectTable { + file_io: FileIO, + identifier: Identifier, + location: String, + comment: Option, +} + +impl ObjectTable { + pub fn try_new(file_io: FileIO, identifier: Identifier, schema: &TableSchema) -> Result { + let options = CoreOptions::new(schema.options()); + if options.table_type()? != TableType::ObjectTable { + return Err(Error::Unsupported { + message: format!( + "table '{}' is not declared 'object-table'", + identifier.full_name() + ), + }); + } + options.ensure_engine_can_serve(&identifier.full_name())?; + let location = options + .path() + .filter(|path| !path.trim().is_empty()) + .ok_or_else(|| Error::ConfigInvalid { + message: format!( + "Object table '{}' requires a non-empty 'path' option", + identifier.full_name() + ), + })? + .to_string(); + Ok(Self { + file_io, + identifier, + location, + comment: schema.comment().map(str::to_string), + }) + } + + pub fn identifier(&self) -> &Identifier { + &self.identifier + } + + pub fn file_io(&self) -> &FileIO { + &self.file_io + } + + pub fn location(&self) -> &str { + &self.location + } + + pub fn comment(&self) -> Option<&str> { + self.comment.as_deref() + } + + /// Fixed schema matching Java Paimon's `ObjectTable.SCHEMA`. + pub fn fields() -> Vec { + vec![ + DataField::new( + 0, + "path".to_string(), + DataType::VarChar( + VarCharType::with_nullable(false, VarCharType::MAX_LENGTH) + .expect("the maximum varchar length is valid"), + ), + ) + .with_description(Some("Relative path of object".to_string())), + DataField::new( + 1, + "name".to_string(), + DataType::VarChar( + VarCharType::with_nullable(false, VarCharType::MAX_LENGTH) + .expect("the maximum varchar length is valid"), + ), + ) + .with_description(Some("Name of object".to_string())), + DataField::new( + 2, + "length".to_string(), + DataType::BigInt(BigIntType::with_nullable(false)), + ) + .with_description(Some("Bytes length of object".to_string())), + DataField::new( + 3, + "mtime".to_string(), + DataType::BigInt(BigIntType::with_nullable(false)), + ) + .with_description(Some("Modification time of object".to_string())), + DataField::new( + 4, + "atime".to_string(), + DataType::BigInt(BigIntType::with_nullable(false)), + ) + .with_description(Some("Access time of object".to_string())), + DataField::new( + 5, + "owner".to_string(), + DataType::VarChar(VarCharType::string_type()), + ) + .with_description(Some("Owner of object".to_string())), + ] + } + + /// Recursively list all files under the object location. + pub async fn list_objects(&self) -> Result> { + let mut entries = Vec::new(); + let location_path = normalized_path(&self.location); + for status in self.file_io.list_status_recursive(&self.location).await? { + let status_path = normalized_path(&status.path); + let relative = status_path + .strip_prefix(&location_path) + .ok_or_else(|| Error::DataInvalid { + message: format!( + "Object path '{}' is outside table location '{}'", + status.path, self.location + ), + source: None, + })? + .trim_start_matches('/') + .to_string(); + let name = relative + .rsplit('/') + .next() + .filter(|name| !name.is_empty()) + .ok_or_else(|| Error::DataInvalid { + message: format!("Object path '{}' has no file name", status.path), + source: None, + })? + .to_string(); + let length = i64::try_from(status.size).map_err(|_| Error::DataInvalid { + message: format!("Object '{}' is too large to fit in BIGINT", status.path), + source: None, + })?; + entries.push(ObjectEntry { + path: relative, + name, + length, + mtime: status + .last_modified + .map(|modified| modified.timestamp_millis()) + .unwrap_or(0), + // OpenDAL does not expose these values portably. + atime: 0, + owner: None, + }); + } + entries.sort_by(|left, right| left.path.cmp(&right.path)); + Ok(entries) + } +} + +fn normalized_path(value: &str) -> String { + url::Url::parse(value) + .map(|url| trim_trailing_slashes(url.path())) + .unwrap_or_else(|_| trim_trailing_slashes(value)) +} + +fn trim_trailing_slashes(value: &str) -> String { + let trimmed = value.trim_end_matches('/'); + if trimmed.is_empty() && value.starts_with('/') { + "/".to_string() + } else { + trimmed.to_string() + } +} diff --git a/crates/paimon/src/table/rest_env.rs b/crates/paimon/src/table/rest_env.rs index 7d716cb7..133acf69 100644 --- a/crates/paimon/src/table/rest_env.rs +++ b/crates/paimon/src/table/rest_env.rs @@ -26,7 +26,7 @@ use crate::io::cache::LocalCache; use crate::io::FileIO; use crate::spec::{CoreOptions, TableSchema, PATH_OPTION}; use crate::table::snapshot_commit::{RESTSnapshotCommit, SnapshotCommit}; -use crate::table::Table; +use crate::table::{ObjectTable, Table}; use crate::Result; use std::sync::Arc; @@ -192,24 +192,16 @@ impl RESTEnv { source: None, })?; - let file_io = if data_token_enabled && !is_external { - Arc::new(RESTTokenFileIO::new( - identifier.clone(), - table_path.clone(), - options.clone(), - api.clone(), - local_cache.clone(), - )) - .build_file_io() - .await? - } else { - let mut builder = FileIO::from_path(&table_path)?; - builder = builder.with_props(options.to_map()); - if let Some(local_cache) = &local_cache { - builder = builder.with_local_cache(local_cache.clone()); - } - builder.build()? - }; + let file_io = Self::build_file_io( + identifier, + &table_path, + api.clone(), + &options, + data_token_enabled, + is_external, + local_cache.clone(), + ) + .await?; let rest_env = RESTEnv::new( identifier.clone(), @@ -229,6 +221,89 @@ impl RESTEnv { )) } + pub(crate) async fn build_object_table( + identifier: &Identifier, + response: crate::api::GetTableResponse, + api: Arc, + options: Options, + data_token_enabled: bool, + local_cache: Option>, + ) -> Result { + let schema = response.schema.ok_or_else(|| Error::DataInvalid { + message: format!("Table {} response missing schema", identifier.full_name()), + source: None, + })?; + let schema_id = response.schema_id.ok_or_else(|| Error::DataInvalid { + message: format!( + "Table {} response missing schema_id", + identifier.full_name() + ), + source: None, + })?; + let object_path = response + .path + .as_deref() + .filter(|path| !path.trim().is_empty()) + .ok_or_else(|| Error::ConfigInvalid { + message: format!( + "Object table '{}' response requires a non-empty path", + identifier.full_name() + ), + })? + .to_string(); + let mut schema_options = schema.options().clone(); + schema_options.insert(PATH_OPTION.to_string(), object_path.clone()); + let table_schema = TableSchema::new(schema_id, &schema).copy_with_options(schema_options); + let is_external = response.is_external.ok_or_else(|| Error::DataInvalid { + message: format!( + "Table {} response missing is_external", + identifier.full_name() + ), + source: None, + })?; + + let file_io = Self::build_file_io( + identifier, + &object_path, + api, + &options, + data_token_enabled, + is_external, + local_cache, + ) + .await?; + + ObjectTable::try_new(file_io, identifier.clone(), &table_schema) + } + + async fn build_file_io( + identifier: &Identifier, + path: &str, + api: Arc, + options: &Options, + data_token_enabled: bool, + is_external: bool, + local_cache: Option>, + ) -> Result { + if data_token_enabled && !is_external { + return Arc::new(RESTTokenFileIO::new( + identifier.clone(), + path.to_string(), + options.clone(), + api, + local_cache, + )) + .build_file_io() + .await; + } + + let mut builder = FileIO::from_path(path)?.with_props(options.to_map()); + if let Some(local_cache) = local_cache { + builder = builder.with_local_cache(local_cache); + } + builder.build() + } + /// Create a `RESTSnapshotCommit` from this environment. pub fn snapshot_commit(&self) -> Arc { Arc::new(RESTSnapshotCommit::new( diff --git a/crates/paimon/tests/rest_catalog_test.rs b/crates/paimon/tests/rest_catalog_test.rs index 83da7c8a..aaac60dc 100644 --- a/crates/paimon/tests/rest_catalog_test.rs +++ b/crates/paimon/tests/rest_catalog_test.rs @@ -1524,6 +1524,43 @@ async fn test_load_table_returns_external_for_declared_type() { ); } +#[tokio::test] +async fn test_load_table_constructs_native_object_table() { + use paimon::catalog::LoadedTable; + + let object_dir = tempfile::TempDir::new().unwrap(); + std::fs::write(object_dir.path().join("object.txt"), b"object").unwrap(); + let object_location = format!("file://{}", object_dir.path().display()); + + let ctx = setup_catalog(vec!["default"]).await; + let schema = Schema::builder() + .column("ignored", DataType::BigInt(BigIntType::new())) + .option("type", "object-table") + .build() + .unwrap(); + ctx.server + .add_table_with_schema("default", "objects", schema, &object_location); + + let loaded = ctx + .catalog + .load_table(&Identifier::new("default", "objects")) + .await + .unwrap(); + let LoadedTable::Object(table) = loaded else { + panic!("expected a native object table, got {loaded:?}"); + }; + assert_eq!( + table + .list_objects() + .await + .unwrap() + .iter() + .map(|entry| entry.path()) + .collect::>(), + vec!["object.txt"] + ); +} + #[tokio::test] async fn test_load_table_fails_closed_on_query_auth() { let ctx = setup_catalog(vec!["default"]).await; From bb4c346a7900ec0a325724b930904510f4dac1fa Mon Sep 17 00:00:00 2001 From: shyjsarah <44659226+shyjsarah@users.noreply.github.com> Date: Thu, 27 Aug 2026 02:29:57 -0700 Subject: [PATCH 2/4] fix: align object table semantics and bounded listing --- crates/integrations/datafusion/src/catalog.rs | 11 ++ .../datafusion/src/table/object.rs | 7 +- .../datafusion/tests/object_table.rs | 16 +++ crates/paimon/src/catalog/filesystem.rs | 124 ++++++++++++++++-- crates/paimon/src/catalog/mod.rs | 1 + crates/paimon/src/io/file_io.rs | 67 +++++++++- crates/paimon/src/table/object_table.rs | 89 ++++++++++++- crates/paimon/tests/rest_catalog_test.rs | 2 + 8 files changed, 294 insertions(+), 23 deletions(-) diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index f410becc..d8893243 100644 --- a/crates/integrations/datafusion/src/catalog.rs +++ b/crates/integrations/datafusion/src/catalog.rs @@ -813,6 +813,10 @@ impl SchemaProvider for PaimonSchemaProvider { )) as Arc)) } Err(e) => Err(to_datafusion_error(e)), + Ok(_) => Err(plan_datafusion_err!( + "catalog returned an unsupported loaded table kind for '{}'", + identifier.full_name() + )), } }) .await @@ -934,6 +938,13 @@ impl SchemaProvider for PaimonSchemaProvider { log::error!("failed to check table '{}': {e}", identifier); false } + Ok(_) => { + log::error!( + "catalog returned an unsupported loaded table kind for '{}'", + identifier + ); + false + } } }, "paimon catalog access thread panicked", diff --git a/crates/integrations/datafusion/src/table/object.rs b/crates/integrations/datafusion/src/table/object.rs index 7d5d16af..41184d50 100644 --- a/crates/integrations/datafusion/src/table/object.rs +++ b/crates/integrations/datafusion/src/table/object.rs @@ -66,14 +66,11 @@ impl TableProvider for ObjectTableProvider { _filters: &[Expr], limit: Option, ) -> DFResult> { - let mut entries = self + let entries = self .table - .list_objects() + .list_objects_with_limit(limit) .await .map_err(to_datafusion_error)?; - if let Some(limit) = limit { - entries.truncate(limit); - } let paths = entries .iter() diff --git a/crates/integrations/datafusion/tests/object_table.rs b/crates/integrations/datafusion/tests/object_table.rs index 3300f49c..7646d711 100644 --- a/crates/integrations/datafusion/tests/object_table.rs +++ b/crates/integrations/datafusion/tests/object_table.rs @@ -100,6 +100,22 @@ async fn object_table_lists_files_recursively() { ] ); + let limited = ctx + .sql("SELECT path FROM cat.db.objects LIMIT 1") + .await + .unwrap() + .collect() + .await + .unwrap(); + assert_eq!( + limited.iter().map(|batch| batch.num_rows()).sum::(), + 1 + ); + assert_eq!( + array_value_to_string(limited[0].column(0).as_ref(), 0).unwrap(), + "nested/child.bin" + ); + let error = ctx .sql( "INSERT INTO cat.db.objects \ diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs index e5d84a6a..45628ad6 100644 --- a/crates/paimon/src/catalog/filesystem.rs +++ b/crates/paimon/src/catalog/filesystem.rs @@ -27,7 +27,7 @@ use crate::error::{ConfigInvalidSnafu, Error, Result}; use crate::io::cache::{create_local_cache, LocalCache}; use crate::io::FileIO; use crate::spec::{CoreOptions, Schema, TableSchema, TableType, TABLE_TYPE_OPTION}; -use crate::table::{SchemaManager, Table}; +use crate::table::{ObjectTable, SchemaManager, Table}; use async_trait::async_trait; use bytes::Bytes; use opendal::raw::get_basename; @@ -379,16 +379,12 @@ impl Catalog for FileSystemCatalog { let object_path = options .path() .filter(|path| !path.trim().is_empty()) - .ok_or_else(|| Error::ConfigInvalid { - message: format!( - "Object table '{}' requires a non-empty 'path' option", - identifier.full_name() - ), - })?; - return crate::table::ObjectTable::try_new( + .unwrap_or(&table_path); + return crate::table::ObjectTable::try_new_with_default_location( self.build_file_io(object_path)?, identifier.clone(), &schema, + Some(&table_path), ) .map(crate::catalog::LoadedTable::Object); } @@ -427,14 +423,17 @@ impl Catalog for FileSystemCatalog { async fn create_table( &self, identifier: &Identifier, - creation: Schema, + mut creation: Schema, ignore_if_exists: bool, ) -> Result<()> { identifier.validate()?; // Never persist a type nothing can load. - CoreOptions::new(creation.options()).table_type()?; + let declared = CoreOptions::new(creation.options()).table_type()?; let table_path = self.table_path(identifier); + if declared == TableType::ObjectTable { + creation = ObjectTable::normalize_creation(&creation, &table_path)?; + } let table_exists = self.table_exists(identifier).await?; @@ -859,6 +858,111 @@ mod tests { assert!(!catalog.table_exists(&identifier).await.unwrap()); } + #[tokio::test] + async fn test_create_object_table_uses_fixed_schema_and_default_path() { + use crate::catalog::LoadedTable; + use crate::table::ObjectTable; + + let catalog = create_memory_catalog(); + catalog + .create_database("db1", false, HashMap::new()) + .await + .unwrap(); + let identifier = Identifier::new("db1", "objects"); + let schema = Schema::builder() + .column( + "ignored", + crate::spec::DataType::Int(crate::spec::IntType::new()), + ) + .option("type", "object-table") + .build() + .unwrap(); + + catalog + .create_table(&identifier, schema, false) + .await + .unwrap(); + + let expected_path = "memory:/warehouse/db1.db/objects"; + let (_, stored) = catalog.fetch_table_schema(&identifier).await.unwrap(); + assert_eq!(stored.fields(), ObjectTable::fields()); + assert_eq!( + stored.options().get(crate::spec::PATH_OPTION), + Some(&expected_path.to_string()) + ); + + let loaded = catalog.load_table(&identifier).await.unwrap(); + let LoadedTable::Object(table) = loaded else { + panic!("expected a native object table, got {loaded:?}"); + }; + assert_eq!(table.location(), expected_path); + + let explicit_identifier = Identifier::new("db1", "explicit_objects"); + let explicit_path = "memory:/external/objects"; + let explicit_schema = Schema::builder() + .column( + "also_ignored", + crate::spec::DataType::Int(crate::spec::IntType::new()), + ) + .option("type", "object-table") + .option(crate::spec::PATH_OPTION, explicit_path) + .build() + .unwrap(); + catalog + .create_table(&explicit_identifier, explicit_schema, false) + .await + .unwrap(); + let (_, stored) = catalog + .fetch_table_schema(&explicit_identifier) + .await + .unwrap(); + assert_eq!(stored.fields(), ObjectTable::fields()); + assert_eq!( + stored + .options() + .get(crate::spec::PATH_OPTION) + .map(String::as_str), + Some(explicit_path) + ); + let loaded = catalog.load_table(&explicit_identifier).await.unwrap(); + let LoadedTable::Object(table) = loaded else { + panic!("expected a native object table, got {loaded:?}"); + }; + assert_eq!(table.location(), explicit_path); + } + + #[tokio::test] + async fn test_load_legacy_object_table_without_path_uses_table_directory() { + use crate::catalog::LoadedTable; + + let catalog = create_memory_catalog(); + catalog + .create_database("db1", false, HashMap::new()) + .await + .unwrap(); + let identifier = Identifier::new("db1", "legacy_objects"); + let table_path = catalog.table_path(&identifier); + catalog.file_io.mkdirs(&table_path).await.unwrap(); + let legacy = Schema::builder() + .column( + "ignored", + crate::spec::DataType::Int(crate::spec::IntType::new()), + ) + .option("type", "object-table") + .build() + .unwrap(); + catalog + .save_table_schema(&table_path, &TableSchema::new(0, &legacy)) + .await + .unwrap(); + + let loaded = catalog.load_table(&identifier).await.unwrap(); + let LoadedTable::Object(table) = loaded else { + panic!("expected a native object table, got {loaded:?}"); + }; + assert_eq!(table.location(), table_path); + } + #[tokio::test] async fn test_stored_scan_selectors_block_routing() { use crate::catalog::LoadedTable; diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs index ebccc6a6..cd3a8901 100644 --- a/crates/paimon/src/catalog/mod.rs +++ b/crates/paimon/src/catalog/mod.rs @@ -267,6 +267,7 @@ use crate::table::{ObjectTable, Table}; /// Outcome of [`Catalog::load_table`]. #[derive(Debug)] +#[non_exhaustive] pub enum LoadedTable { /// A constructed Paimon table (boxed: far larger than the other variant). Paimon(Box
), diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index 63735486..52c554b5 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -16,7 +16,8 @@ // under the License. use crate::error::*; -use std::collections::HashMap; +use std::cmp::Ordering; +use std::collections::{BinaryHeap, HashMap}; use std::future::Future; use std::ops::Range; use std::pin::Pin; @@ -26,6 +27,7 @@ use std::time::SystemTime; use bytes::Bytes; use chrono::{DateTime, Utc}; +use futures::TryStreamExt; use opendal::raw::normalize_root; use opendal::Operator; use snafu::ResultExt; @@ -223,14 +225,26 @@ impl FileIO { /// List all files recursively under the given directory path. pub async fn list_status_recursive(&self, path: &str) -> Result> { + self.list_status_recursive_with_limit(path, None).await + } + + pub(crate) async fn list_status_recursive_with_limit( + &self, + path: &str, + limit: Option, + ) -> Result> { + if limit == Some(0) { + return Ok(Vec::new()); + } + let (op, relative_path) = self.create(path).await?; // See `list_status`: `relative_path` is a byte-suffix of `path` except // for Windows local paths, where it only swaps separators (same length). let base_path = &path[..path.len() - relative_path.len()]; let list_path = normalize_root(relative_path.as_ref()); - let entries = - op.list_with(&list_path) + let mut entries = + op.lister_with(&list_path) .recursive(true) .await .context(IoUnexpectedSnafu { @@ -238,8 +252,11 @@ impl FileIO { })?; let mut statuses = Vec::new(); + let mut smallest = limit.map(|limit| BinaryHeap::with_capacity(limit.saturating_add(1))); let list_path_normalized = list_path.trim_start_matches('/'); - for entry in entries { + while let Some(entry) = entries.try_next().await.context(IoUnexpectedSnafu { + message: format!("Failed to list files recursively in '{path}'"), + })? { let entry_path = entry.path(); if entry_path.trim_start_matches('/') == list_path_normalized { continue; @@ -248,14 +265,29 @@ impl FileIO { if meta.is_dir() { continue; } - statuses.push(FileStatus { + let status = FileStatus { size: meta.content_length(), is_dir: false, path: status_path(base_path, entry_path), last_modified: meta .last_modified() .map(|v| DateTime::::from(SystemTime::from(v))), - }); + }; + match (&mut smallest, limit) { + (Some(heap), Some(limit)) => { + heap.push(PathOrderedStatus(status)); + if heap.len() > limit { + heap.pop(); + } + } + _ => statuses.push(status), + } + } + if let Some(heap) = smallest { + statuses = heap + .into_iter() + .map(|PathOrderedStatus(status)| status) + .collect(); } Ok(statuses) @@ -600,6 +632,29 @@ pub struct FileStatus { pub last_modified: Option>, } +#[derive(Debug)] +struct PathOrderedStatus(FileStatus); + +impl PartialEq for PathOrderedStatus { + fn eq(&self, other: &Self) -> bool { + self.0.path == other.0.path + } +} + +impl Eq for PathOrderedStatus {} + +impl PartialOrd for PathOrderedStatus { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for PathOrderedStatus { + fn cmp(&self, other: &Self) -> Ordering { + self.0.path.cmp(&other.0.path) + } +} + #[derive(Debug)] pub struct InputFile { op: Operator, diff --git a/crates/paimon/src/table/object_table.rs b/crates/paimon/src/table/object_table.rs index 19cb64be..e2473456 100644 --- a/crates/paimon/src/table/object_table.rs +++ b/crates/paimon/src/table/object_table.rs @@ -18,7 +18,8 @@ use crate::catalog::Identifier; use crate::io::FileIO; use crate::spec::{ - BigIntType, CoreOptions, DataField, DataType, TableSchema, TableType, VarCharType, + BigIntType, CoreOptions, DataField, DataType, Schema, TableSchema, TableType, VarCharType, + PATH_OPTION, }; use crate::{Error, Result}; @@ -69,7 +70,38 @@ pub struct ObjectTable { } impl ObjectTable { + pub(crate) fn normalize_creation(schema: &Schema, default_location: &str) -> Result { + let mut options = schema.options().clone(); + if !options + .get(PATH_OPTION) + .is_some_and(|path| !path.trim().is_empty()) + { + options.insert(PATH_OPTION.to_string(), default_location.to_string()); + } + + let mut builder = Schema::builder() + .options(options) + .comment(schema.comment().map(str::to_string)); + for field in Self::fields() { + builder = builder.column_with_description( + field.name().to_string(), + field.data_type().clone(), + field.description().map(str::to_string), + ); + } + builder.build() + } + pub fn try_new(file_io: FileIO, identifier: Identifier, schema: &TableSchema) -> Result { + Self::try_new_with_default_location(file_io, identifier, schema, None) + } + + pub(crate) fn try_new_with_default_location( + file_io: FileIO, + identifier: Identifier, + schema: &TableSchema, + default_location: Option<&str>, + ) -> Result { let options = CoreOptions::new(schema.options()); if options.table_type()? != TableType::ObjectTable { return Err(Error::Unsupported { @@ -83,6 +115,7 @@ impl ObjectTable { let location = options .path() .filter(|path| !path.trim().is_empty()) + .or_else(|| default_location.filter(|path| !path.trim().is_empty())) .ok_or_else(|| Error::ConfigInvalid { message: format!( "Object table '{}' requires a non-empty 'path' option", @@ -164,9 +197,19 @@ impl ObjectTable { /// Recursively list all files under the object location. pub async fn list_objects(&self) -> Result> { + self.list_objects_with_limit(None).await + } + + /// Recursively list files under the object location, retaining at most the + /// lexicographically smallest `limit` paths when one is supplied. + pub async fn list_objects_with_limit(&self, limit: Option) -> Result> { let mut entries = Vec::new(); let location_path = normalized_path(&self.location); - for status in self.file_io.list_status_recursive(&self.location).await? { + for status in self + .file_io + .list_status_recursive_with_limit(&self.location, limit) + .await? + { let status_path = normalized_path(&status.path); let relative = status_path .strip_prefix(&location_path) @@ -224,3 +267,45 @@ fn trim_trailing_slashes(value: &str) -> String { trimmed.to_string() } } + +#[cfg(test)] +mod tests { + use bytes::Bytes; + + use super::*; + + #[tokio::test] + async fn list_objects_with_limit_keeps_path_smallest_entries() { + let location = "memory:/objects"; + let file_io = FileIO::from_path(location).unwrap().build().unwrap(); + for path in ["z.txt", "nested/b.txt", "a.txt", "nested/a.txt"] { + file_io + .new_output(&format!("{location}/{path}")) + .unwrap() + .write(Bytes::from_static(b"x")) + .await + .unwrap(); + } + let schema = Schema::builder() + .column("ignored", DataType::BigInt(BigIntType::new())) + .option("type", "object-table") + .option(PATH_OPTION, location) + .build() + .unwrap(); + let table = ObjectTable::try_new( + file_io, + Identifier::new("db", "objects"), + &TableSchema::new(0, &schema), + ) + .unwrap(); + + let paths = table + .list_objects_with_limit(Some(2)) + .await + .unwrap() + .into_iter() + .map(|entry| entry.path) + .collect::>(); + assert_eq!(paths, vec!["a.txt", "nested/a.txt"]); + } +} diff --git a/crates/paimon/tests/rest_catalog_test.rs b/crates/paimon/tests/rest_catalog_test.rs index aaac60dc..41052c45 100644 --- a/crates/paimon/tests/rest_catalog_test.rs +++ b/crates/paimon/tests/rest_catalog_test.rs @@ -1549,6 +1549,8 @@ async fn test_load_table_constructs_native_object_table() { let LoadedTable::Object(table) = loaded else { panic!("expected a native object table, got {loaded:?}"); }; + assert_eq!(table.location(), object_location); + #[cfg(not(windows))] assert_eq!( table .list_objects() From fc31528e46ab6f10a93dd99f2384d61b1478ecaa Mon Sep 17 00:00:00 2001 From: shyjsarah <44659226+shyjsarah@users.noreply.github.com> Date: Thu, 27 Aug 2026 03:41:39 -0700 Subject: [PATCH 3/4] fix: stop object listing at pushed limit --- .../datafusion/tests/object_table.rs | 15 +- crates/paimon/src/io/file_io.rs | 221 ++++++++++++++---- crates/paimon/src/table/object_table.rs | 36 ++- 3 files changed, 227 insertions(+), 45 deletions(-) diff --git a/crates/integrations/datafusion/tests/object_table.rs b/crates/integrations/datafusion/tests/object_table.rs index 7646d711..25e0f2f1 100644 --- a/crates/integrations/datafusion/tests/object_table.rs +++ b/crates/integrations/datafusion/tests/object_table.rs @@ -111,8 +111,21 @@ async fn object_table_lists_files_recursively() { limited.iter().map(|batch| batch.num_rows()).sum::(), 1 ); + assert!(["nested/child.bin", "root.txt"].contains( + &array_value_to_string(limited[0].column(0).as_ref(), 0) + .unwrap() + .as_str() + )); + + let ordered_limited = ctx + .sql("SELECT path FROM cat.db.objects ORDER BY path LIMIT 1") + .await + .unwrap() + .collect() + .await + .unwrap(); assert_eq!( - array_value_to_string(limited[0].column(0).as_ref(), 0).unwrap(), + array_value_to_string(ordered_limited[0].column(0).as_ref(), 0).unwrap(), "nested/child.bin" ); diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index 52c554b5..0571197a 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -16,8 +16,7 @@ // under the License. use crate::error::*; -use std::cmp::Ordering; -use std::collections::{BinaryHeap, HashMap}; +use std::collections::HashMap; use std::future::Future; use std::ops::Range; use std::pin::Pin; @@ -252,7 +251,6 @@ impl FileIO { })?; let mut statuses = Vec::new(); - let mut smallest = limit.map(|limit| BinaryHeap::with_capacity(limit.saturating_add(1))); let list_path_normalized = list_path.trim_start_matches('/'); while let Some(entry) = entries.try_next().await.context(IoUnexpectedSnafu { message: format!("Failed to list files recursively in '{path}'"), @@ -273,22 +271,11 @@ impl FileIO { .last_modified() .map(|v| DateTime::::from(SystemTime::from(v))), }; - match (&mut smallest, limit) { - (Some(heap), Some(limit)) => { - heap.push(PathOrderedStatus(status)); - if heap.len() > limit { - heap.pop(); - } - } - _ => statuses.push(status), + statuses.push(status); + if limit.is_some_and(|limit| statuses.len() >= limit) { + break; } } - if let Some(heap) = smallest { - statuses = heap - .into_iter() - .map(|PathOrderedStatus(status)| status) - .collect(); - } Ok(statuses) } @@ -632,29 +619,6 @@ pub struct FileStatus { pub last_modified: Option>, } -#[derive(Debug)] -struct PathOrderedStatus(FileStatus); - -impl PartialEq for PathOrderedStatus { - fn eq(&self, other: &Self) -> bool { - self.0.path == other.0.path - } -} - -impl Eq for PathOrderedStatus {} - -impl PartialOrd for PathOrderedStatus { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} - -impl Ord for PathOrderedStatus { - fn cmp(&self, other: &Self) -> Ordering { - self.0.path.cmp(&other.0.path) - } -} - #[derive(Debug)] pub struct InputFile { op: Operator, @@ -848,10 +812,171 @@ impl OutputFile { mod file_action_test { use std::collections::BTreeSet; use std::fs; + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; use tempfile::tempdir; use super::*; use bytes::Bytes; + use opendal::raw::{ + oio, OpCopier, OpCopy, OpCreateDir, OpList, OpPresign, OpRead, OpRename, OpStat, OpWrite, + RpCreateDir, RpPresign, RpRename, RpStat, Service, ServiceInfo, Servicer, + }; + use opendal::{Capability, EntryMode, Metadata, OperationContext}; + + #[derive(Debug)] + struct CountingListProvider { + pulls: Arc, + } + + #[async_trait::async_trait] + impl FileIOProvider for CountingListProvider { + async fn create(&self, _path: &str) -> crate::Result<(Operator, String)> { + let service: Servicer = Arc::new(CountingListService { + pulls: Arc::clone(&self.pulls), + }); + Ok(( + Operator::from_parts(OperationContext::default(), service), + "objects/".to_string(), + )) + } + } + + #[derive(Debug)] + struct CountingListService { + pulls: Arc, + } + + impl Service for CountingListService { + type Reader = (); + type Writer = (); + type Lister = CountingLister; + type Deleter = (); + type Copier = (); + + fn info(&self) -> ServiceInfo { + ServiceInfo::with_scheme("counting") + } + + fn capability(&self) -> Capability { + Capability { + list: true, + list_with_recursive: true, + ..Default::default() + } + } + + async fn create_dir( + &self, + _ctx: &OperationContext, + _path: &str, + _args: OpCreateDir, + ) -> opendal::Result { + Err(unsupported_test_operation()) + } + + async fn stat( + &self, + _ctx: &OperationContext, + _path: &str, + _args: OpStat, + ) -> opendal::Result { + Err(unsupported_test_operation()) + } + + fn read( + &self, + _ctx: &OperationContext, + _path: &str, + _args: OpRead, + ) -> opendal::Result { + Err(unsupported_test_operation()) + } + + fn write( + &self, + _ctx: &OperationContext, + _path: &str, + _args: OpWrite, + ) -> opendal::Result { + Err(unsupported_test_operation()) + } + + fn delete(&self, _ctx: &OperationContext) -> opendal::Result { + Err(unsupported_test_operation()) + } + + fn list( + &self, + _ctx: &OperationContext, + _path: &str, + _args: OpList, + ) -> opendal::Result { + Ok(CountingLister { + pulls: Arc::clone(&self.pulls), + next: 0, + }) + } + + fn copy( + &self, + _ctx: &OperationContext, + _from: &str, + _to: &str, + _args: OpCopy, + _opts: OpCopier, + ) -> opendal::Result { + Err(unsupported_test_operation()) + } + + async fn rename( + &self, + _ctx: &OperationContext, + _from: &str, + _to: &str, + _args: OpRename, + ) -> opendal::Result { + Err(unsupported_test_operation()) + } + + async fn presign( + &self, + _ctx: &OperationContext, + _path: &str, + _args: OpPresign, + ) -> opendal::Result { + Err(unsupported_test_operation()) + } + } + + fn unsupported_test_operation() -> opendal::Error { + opendal::Error::new( + opendal::ErrorKind::Unsupported, + "operation is not supported by the test service", + ) + } + + struct CountingLister { + pulls: Arc, + next: usize, + } + + impl oio::List for CountingLister { + async fn next(&mut self) -> opendal::Result> { + self.pulls.fetch_add(1, AtomicOrdering::SeqCst); + if self.next == 0 { + self.next += 1; + return Ok(Some(oio::Entry::new( + "objects/first.txt", + Metadata::new(EntryMode::FILE).with_content_length(1), + ))); + } + + Err(opendal::Error::new( + opendal::ErrorKind::Unexpected, + "limited listing polled past the requested row", + )) + } + } fn setup_memory_file_io() -> FileIO { FileIOBuilder::new("memory").build().unwrap() @@ -964,6 +1089,22 @@ mod file_action_test { file_io.delete_dir(dir_path).await.unwrap(); } + #[tokio::test] + async fn test_recursive_listing_stops_after_limit() { + let pulls = Arc::new(AtomicUsize::new(0)); + let file_io = setup_memory_file_io().with_provider(Arc::new(CountingListProvider { + pulls: Arc::clone(&pulls), + })); + + let statuses = file_io + .list_status_recursive_with_limit("counting:/objects/", Some(1)) + .await + .unwrap(); + + assert_eq!(statuses.len(), 1); + assert_eq!(pulls.load(AtomicOrdering::SeqCst), 1); + } + #[tokio::test] async fn test_delete_file_memory() { let file_io = setup_memory_file_io(); diff --git a/crates/paimon/src/table/object_table.rs b/crates/paimon/src/table/object_table.rs index e2473456..1ad9beed 100644 --- a/crates/paimon/src/table/object_table.rs +++ b/crates/paimon/src/table/object_table.rs @@ -200,8 +200,8 @@ impl ObjectTable { self.list_objects_with_limit(None).await } - /// Recursively list files under the object location, retaining at most the - /// lexicographically smallest `limit` paths when one is supplied. + /// Recursively list files under the object location, stopping after + /// `limit` files have been yielded by the storage backend when supplied. pub async fn list_objects_with_limit(&self, limit: Option) -> Result> { let mut entries = Vec::new(); let location_path = normalized_path(&self.location); @@ -275,7 +275,7 @@ mod tests { use super::*; #[tokio::test] - async fn list_objects_with_limit_keeps_path_smallest_entries() { + async fn list_objects_with_limit_returns_at_most_limit_entries() { let location = "memory:/objects"; let file_io = FileIO::from_path(location).unwrap().build().unwrap(); for path in ["z.txt", "nested/b.txt", "a.txt", "nested/a.txt"] { @@ -306,6 +306,34 @@ mod tests { .into_iter() .map(|entry| entry.path) .collect::>(); - assert_eq!(paths, vec!["a.txt", "nested/a.txt"]); + assert_eq!(paths.len(), 2); + assert!(paths.windows(2).all(|pair| pair[0] <= pair[1])); + assert!(paths.iter().all(|path| { + ["z.txt", "nested/b.txt", "a.txt", "nested/a.txt"].contains(&path.as_str()) + })); + } + + #[tokio::test] + async fn list_objects_with_huge_limit_does_not_preallocate() { + let location = "memory:/empty-objects"; + let file_io = FileIO::from_path(location).unwrap().build().unwrap(); + let schema = Schema::builder() + .column("ignored", DataType::BigInt(BigIntType::new())) + .option("type", "object-table") + .option(PATH_OPTION, location) + .build() + .unwrap(); + let table = ObjectTable::try_new( + file_io, + Identifier::new("db", "objects"), + &TableSchema::new(0, &schema), + ) + .unwrap(); + + let entries = table + .list_objects_with_limit(Some(usize::MAX)) + .await + .unwrap(); + assert!(entries.is_empty()); } } From e50ea4fe00710ea69b44c2ea59c1541c8590dfbe Mon Sep 17 00:00:00 2001 From: shyjsarah <44659226+shyjsarah@users.noreply.github.com> Date: Thu, 27 Aug 2026 20:16:40 -0700 Subject: [PATCH 4/4] fix(datafusion): stream object table scans --- .../datafusion/src/table/object.rs | 293 +++++++++++++----- .../datafusion/tests/object_table.rs | 12 + crates/paimon/src/io/file_io.rs | 93 ++++-- crates/paimon/src/table/object_table.rs | 113 ++++--- 4 files changed, 366 insertions(+), 145 deletions(-) diff --git a/crates/integrations/datafusion/src/table/object.rs b/crates/integrations/datafusion/src/table/object.rs index 41184d50..55dd673c 100644 --- a/crates/integrations/datafusion/src/table/object.rs +++ b/crates/integrations/datafusion/src/table/object.rs @@ -19,17 +19,21 @@ use std::sync::Arc; use async_trait::async_trait; use datafusion::arrow::array::{ - new_null_array, ArrayRef, Int64Array, RecordBatch, StringViewArray, + new_null_array, ArrayRef, Int64Array, RecordBatch, RecordBatchOptions, StringArray, + StringViewArray, }; use datafusion::arrow::datatypes::SchemaRef; use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; use datafusion::datasource::{TableProvider, TableType}; use datafusion::error::{DataFusionError, Result as DFResult}; +use datafusion::execution::{SendableRecordBatchStream, TaskContext}; use datafusion::logical_expr::dml::InsertOp; use datafusion::logical_expr::Expr; +use datafusion::physical_plan::stream::RecordBatchStreamAdapter; +use datafusion::physical_plan::streaming::{PartitionStream, StreamingTableExec}; use datafusion::physical_plan::ExecutionPlan; -use paimon::table::ObjectTable; +use futures::{StreamExt, TryStreamExt}; +use paimon::table::{ObjectEntry, ObjectTable}; use crate::error::to_datafusion_error; @@ -49,6 +53,49 @@ impl ObjectTableProvider { } } +#[derive(Debug, Clone)] +struct ObjectPartitionStream { + table: ObjectTable, + projection: Arc<[usize]>, + schema: SchemaRef, + limit: Option, +} + +impl PartitionStream for ObjectPartitionStream { + fn schema(&self) -> &SchemaRef { + &self.schema + } + + fn execute(&self, ctx: Arc) -> SendableRecordBatchStream { + let table = self.table.clone(); + let projection = Arc::clone(&self.projection); + let schema = Arc::clone(&self.schema); + let output_schema = Arc::clone(&self.schema); + let limit = self.limit; + let batch_size = ctx.session_config().batch_size().max(1); + let future = async move { + let entries = table + .stream_objects(limit) + .await + .map_err(to_datafusion_error)?; + let batch_schema = Arc::clone(&schema); + let batches = entries + .map(|entry| entry.map_err(to_datafusion_error)) + .chunks(batch_size) + .map(move |chunk| { + let entries = chunk.into_iter().collect::>>()?; + object_entries_to_batch(&entries, &projection, &batch_schema) + }); + Ok::<_, DataFusionError>(RecordBatchStreamAdapter::new(schema, Box::pin(batches))) + }; + + Box::pin(RecordBatchStreamAdapter::new( + output_schema, + futures::stream::once(future).try_flatten(), + )) + } +} + #[async_trait] impl TableProvider for ObjectTableProvider { fn schema(&self) -> SchemaRef { @@ -66,74 +113,25 @@ impl TableProvider for ObjectTableProvider { _filters: &[Expr], limit: Option, ) -> DFResult> { - let entries = self - .table - .list_objects_with_limit(limit) - .await - .map_err(to_datafusion_error)?; + let projection = projection + .cloned() + .unwrap_or_else(|| (0..self.schema.fields().len()).collect()); + let projected_schema = Arc::new(self.schema.project(&projection)?); + let partition: Arc = Arc::new(ObjectPartitionStream { + table: self.table.clone(), + projection: projection.into(), + schema: Arc::clone(&projected_schema), + limit, + }); - let paths = entries - .iter() - .map(|entry| entry.path().to_string()) - .collect::>(); - let names = entries - .iter() - .map(|entry| entry.name().to_string()) - .collect::>(); - let lengths = entries - .iter() - .map(|entry| entry.length()) - .collect::>(); - let mtimes = entries - .iter() - .map(|entry| entry.mtime()) - .collect::>(); - let atimes = entries - .iter() - .map(|entry| entry.atime()) - .collect::>(); - let owners = entries - .iter() - .map(|entry| entry.owner()) - .collect::>(); - let row_count = entries.len(); - - let string_array = |values: Vec, index: usize| -> ArrayRef { - if matches!( - self.schema.field(index).data_type(), - datafusion::arrow::datatypes::DataType::Utf8View - ) { - Arc::new(StringViewArray::from(values)) - } else { - Arc::new(datafusion::arrow::array::StringArray::from(values)) - } - }; - let batch = RecordBatch::try_new( - Arc::clone(&self.schema), - vec![ - string_array(paths, 0), - string_array(names, 1), - Arc::new(Int64Array::from(lengths)), - Arc::new(Int64Array::from(mtimes)), - Arc::new(Int64Array::from(atimes)), - if owners.iter().all(Option::is_none) { - new_null_array(self.schema.field(5).data_type(), row_count) - } else if matches!( - self.schema.field(5).data_type(), - datafusion::arrow::datatypes::DataType::Utf8View - ) { - Arc::new(StringViewArray::from(owners)) - } else { - Arc::new(datafusion::arrow::array::StringArray::from(owners)) - }, - ], - )?; - - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - Arc::clone(&self.schema), - projection.cloned(), - )?) + Ok(Arc::new(StreamingTableExec::try_new( + projected_schema, + vec![partition], + None, + std::iter::empty(), + false, + limit, + )?)) } async fn insert_into( @@ -148,3 +146,158 @@ impl TableProvider for ObjectTableProvider { ))) } } + +fn object_entries_to_batch( + entries: &[ObjectEntry], + projection: &[usize], + schema: &SchemaRef, +) -> DFResult { + let columns = projection + .iter() + .enumerate() + .map(|(output_index, source_index)| -> DFResult { + let data_type = schema.field(output_index).data_type(); + Ok(match source_index { + 0 => string_array(entries.iter().map(ObjectEntry::path), data_type), + 1 => string_array(entries.iter().map(ObjectEntry::name), data_type), + 2 => Arc::new(Int64Array::from_iter_values( + entries.iter().map(ObjectEntry::length), + )), + 3 => Arc::new(Int64Array::from_iter_values( + entries.iter().map(ObjectEntry::mtime), + )), + 4 => Arc::new(Int64Array::from_iter_values( + entries.iter().map(ObjectEntry::atime), + )), + 5 => owner_array(entries, data_type), + index => { + return Err(DataFusionError::Internal(format!( + "Object table projection index {index} is out of range" + ))); + } + }) + }) + .collect::>>()?; + let options = RecordBatchOptions::new().with_row_count(Some(entries.len())); + Ok(RecordBatch::try_new_with_options( + Arc::clone(schema), + columns, + &options, + )?) +} + +fn string_array<'a>( + values: impl IntoIterator, + data_type: &datafusion::arrow::datatypes::DataType, +) -> ArrayRef { + if matches!(data_type, datafusion::arrow::datatypes::DataType::Utf8View) { + Arc::new(StringViewArray::from_iter_values(values)) + } else { + Arc::new(StringArray::from_iter_values(values)) + } +} + +fn owner_array( + entries: &[ObjectEntry], + data_type: &datafusion::arrow::datatypes::DataType, +) -> ArrayRef { + let owners = entries.iter().map(ObjectEntry::owner).collect::>(); + if owners.iter().all(Option::is_none) { + new_null_array(data_type, entries.len()) + } else if matches!(data_type, datafusion::arrow::datatypes::DataType::Utf8View) { + Arc::new(StringViewArray::from(owners)) + } else { + Arc::new(StringArray::from(owners)) + } +} + +#[cfg(test)] +mod tests { + use bytes::Bytes; + use datafusion::execution::context::SessionConfig; + use datafusion::physical_plan::streaming::StreamingTableExec; + use datafusion::prelude::SessionContext; + use futures::TryStreamExt; + use paimon::catalog::Identifier; + use paimon::io::FileIO; + use paimon::spec::{Schema, TableSchema}; + + use super::*; + + #[tokio::test] + async fn scan_uses_a_projected_streaming_source() { + let location = "memory:/objects"; + let file_io = FileIO::from_path(location).unwrap().build().unwrap(); + let schema = Schema::builder() + .option("type", "object-table") + .option("path", location) + .build() + .unwrap(); + let table = ObjectTable::try_new( + file_io, + Identifier::new("db", "objects"), + &TableSchema::new(0, &schema), + ) + .unwrap(); + let provider = ObjectTableProvider::try_new(table, false).unwrap(); + let projection = vec![0]; + let state = SessionContext::new().state(); + + let plan = provider + .scan(&state, Some(&projection), &[], None) + .await + .unwrap(); + let streaming = plan + .downcast_ref::() + .expect("object scans should use StreamingTableExec"); + + assert_eq!(streaming.partition_schema().fields().len(), 1); + assert_eq!(streaming.partition_schema().field(0).name(), "path"); + } + + #[tokio::test] + async fn scan_streams_projected_batches_at_the_session_batch_size() { + let location = "memory:/streaming-objects"; + let file_io = FileIO::from_path(location).unwrap().build().unwrap(); + for path in ["a.txt", "b.txt", "c.txt"] { + file_io + .new_output(&format!("{location}/{path}")) + .unwrap() + .write(Bytes::from_static(b"x")) + .await + .unwrap(); + } + let schema = Schema::builder() + .option("type", "object-table") + .option("path", location) + .build() + .unwrap(); + let table = ObjectTable::try_new( + file_io, + Identifier::new("db", "objects"), + &TableSchema::new(0, &schema), + ) + .unwrap(); + let provider = ObjectTableProvider::try_new(table, false).unwrap(); + let projection = vec![0]; + let ctx = SessionContext::new_with_config(SessionConfig::new().with_batch_size(1)); + let state = ctx.state(); + + let plan = provider + .scan(&state, Some(&projection), &[], None) + .await + .unwrap(); + let batches = plan + .execute(0, ctx.task_ctx()) + .unwrap() + .try_collect::>() + .await + .unwrap(); + + assert_eq!(batches.len(), 3); + assert!(batches + .iter() + .all(|batch| batch.num_rows() == 1 && batch.num_columns() == 1)); + assert_eq!(batches[0].schema().field(0).name(), "path"); + } +} diff --git a/crates/integrations/datafusion/tests/object_table.rs b/crates/integrations/datafusion/tests/object_table.rs index 25e0f2f1..3211b842 100644 --- a/crates/integrations/datafusion/tests/object_table.rs +++ b/crates/integrations/datafusion/tests/object_table.rs @@ -100,6 +100,18 @@ async fn object_table_lists_files_recursively() { ] ); + let count = ctx + .sql("SELECT COUNT(*) FROM cat.db.objects") + .await + .unwrap() + .collect() + .await + .unwrap(); + assert_eq!( + array_value_to_string(count[0].column(0).as_ref(), 0).unwrap(), + "2" + ); + let limited = ctx .sql("SELECT path FROM cat.db.objects LIMIT 1") .await diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index 0571197a..886033eb 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -26,7 +26,8 @@ use std::time::SystemTime; use bytes::Bytes; use chrono::{DateTime, Utc}; -use futures::TryStreamExt; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; use opendal::raw::normalize_root; use opendal::Operator; use snafu::ResultExt; @@ -232,17 +233,28 @@ impl FileIO { path: &str, limit: Option, ) -> Result> { + self.list_status_recursive_stream(path, limit) + .await? + .try_collect() + .await + } + + pub(crate) async fn list_status_recursive_stream( + &self, + path: &str, + limit: Option, + ) -> Result>> { if limit == Some(0) { - return Ok(Vec::new()); + return Ok(futures::stream::empty().boxed()); } let (op, relative_path) = self.create(path).await?; // See `list_status`: `relative_path` is a byte-suffix of `path` except // for Windows local paths, where it only swaps separators (same length). - let base_path = &path[..path.len() - relative_path.len()]; + let base_path = path[..path.len() - relative_path.len()].to_string(); let list_path = normalize_root(relative_path.as_ref()); - let mut entries = + let entries = op.lister_with(&list_path) .recursive(true) .await @@ -250,34 +262,36 @@ impl FileIO { message: format!("Failed to list files recursively in '{path}'"), })?; - let mut statuses = Vec::new(); - let list_path_normalized = list_path.trim_start_matches('/'); - while let Some(entry) = entries.try_next().await.context(IoUnexpectedSnafu { - message: format!("Failed to list files recursively in '{path}'"), - })? { - let entry_path = entry.path(); - if entry_path.trim_start_matches('/') == list_path_normalized { - continue; - } - let meta = entry.metadata(); - if meta.is_dir() { - continue; - } - let status = FileStatus { - size: meta.content_length(), - is_dir: false, - path: status_path(base_path, entry_path), - last_modified: meta - .last_modified() - .map(|v| DateTime::::from(SystemTime::from(v))), - }; - statuses.push(status); - if limit.is_some_and(|limit| statuses.len() >= limit) { - break; + let path = path.to_string(); + let list_path_normalized = list_path.trim_start_matches('/').to_string(); + Ok(Box::pin(async_stream::try_stream! { + let mut entries = entries; + let mut emitted = 0usize; + while let Some(entry) = entries.try_next().await.context(IoUnexpectedSnafu { + message: format!("Failed to list files recursively in '{path}'"), + })? { + let entry_path = entry.path(); + if entry_path.trim_start_matches('/') == list_path_normalized { + continue; + } + let meta = entry.metadata(); + if meta.is_dir() { + continue; + } + yield FileStatus { + size: meta.content_length(), + is_dir: false, + path: status_path(&base_path, entry_path), + last_modified: meta + .last_modified() + .map(|v| DateTime::::from(SystemTime::from(v))), + }; + emitted += 1; + if limit.is_some_and(|limit| emitted >= limit) { + break; + } } - } - - Ok(statuses) + })) } /// Check if exists. @@ -1105,6 +1119,23 @@ mod file_action_test { assert_eq!(pulls.load(AtomicOrdering::SeqCst), 1); } + #[tokio::test] + async fn test_recursive_listing_stream_yields_before_polling_next_entry() { + let pulls = Arc::new(AtomicUsize::new(0)); + let file_io = setup_memory_file_io().with_provider(Arc::new(CountingListProvider { + pulls: Arc::clone(&pulls), + })); + + let mut statuses = file_io + .list_status_recursive_stream("counting:/objects/", None) + .await + .unwrap(); + let first = statuses.try_next().await.unwrap().unwrap(); + + assert!(first.path.ends_with("first.txt")); + assert_eq!(pulls.load(AtomicOrdering::SeqCst), 1); + } + #[tokio::test] async fn test_delete_file_memory() { let file_io = setup_memory_file_io(); diff --git a/crates/paimon/src/table/object_table.rs b/crates/paimon/src/table/object_table.rs index 1ad9beed..7398125e 100644 --- a/crates/paimon/src/table/object_table.rs +++ b/crates/paimon/src/table/object_table.rs @@ -22,6 +22,8 @@ use crate::spec::{ PATH_OPTION, }; use crate::{Error, Result}; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; /// Metadata for one file exposed by an [`ObjectTable`]. #[derive(Debug, Clone, PartialEq, Eq)] @@ -203,54 +205,77 @@ impl ObjectTable { /// Recursively list files under the object location, stopping after /// `limit` files have been yielded by the storage backend when supplied. pub async fn list_objects_with_limit(&self, limit: Option) -> Result> { - let mut entries = Vec::new(); - let location_path = normalized_path(&self.location); - for status in self - .file_io - .list_status_recursive_with_limit(&self.location, limit) + let mut entries = self + .stream_objects(limit) .await? - { - let status_path = normalized_path(&status.path); - let relative = status_path - .strip_prefix(&location_path) - .ok_or_else(|| Error::DataInvalid { - message: format!( - "Object path '{}' is outside table location '{}'", - status.path, self.location - ), - source: None, - })? - .trim_start_matches('/') - .to_string(); - let name = relative - .rsplit('/') - .next() - .filter(|name| !name.is_empty()) - .ok_or_else(|| Error::DataInvalid { - message: format!("Object path '{}' has no file name", status.path), - source: None, - })? - .to_string(); - let length = i64::try_from(status.size).map_err(|_| Error::DataInvalid { - message: format!("Object '{}' is too large to fit in BIGINT", status.path), - source: None, - })?; - entries.push(ObjectEntry { - path: relative, - name, - length, - mtime: status - .last_modified - .map(|modified| modified.timestamp_millis()) - .unwrap_or(0), - // OpenDAL does not expose these values portably. - atime: 0, - owner: None, - }); - } + .try_collect::>() + .await?; entries.sort_by(|left, right| left.path.cmp(&right.path)); Ok(entries) } + + /// Stream files under the object location in storage listing order. + pub async fn stream_objects( + &self, + limit: Option, + ) -> Result>> { + let statuses = self + .file_io + .list_status_recursive_stream(&self.location, limit) + .await?; + let location = self.location.clone(); + let location_path = normalized_path(&location); + Ok(statuses + .map(move |status| { + status + .and_then(|status| object_entry_from_status(&location, &location_path, status)) + }) + .boxed()) + } +} + +fn object_entry_from_status( + location: &str, + location_path: &str, + status: crate::io::FileStatus, +) -> Result { + let status_path = normalized_path(&status.path); + let relative = status_path + .strip_prefix(location_path) + .ok_or_else(|| Error::DataInvalid { + message: format!( + "Object path '{}' is outside table location '{}'", + status.path, location + ), + source: None, + })? + .trim_start_matches('/') + .to_string(); + let name = relative + .rsplit('/') + .next() + .filter(|name| !name.is_empty()) + .ok_or_else(|| Error::DataInvalid { + message: format!("Object path '{}' has no file name", status.path), + source: None, + })? + .to_string(); + let length = i64::try_from(status.size).map_err(|_| Error::DataInvalid { + message: format!("Object '{}' is too large to fit in BIGINT", status.path), + source: None, + })?; + Ok(ObjectEntry { + path: relative, + name, + length, + mtime: status + .last_modified + .map(|modified| modified.timestamp_millis()) + .unwrap_or(0), + // OpenDAL does not expose these values portably. + atime: 0, + owner: None, + }) } fn normalized_path(value: &str) -> String {