This is an automated email from the ASF dual-hosted git repository.
jerry-024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 8001f02d feat: add native object table read support (#749)
8001f02d is described below
commit 8001f02d8cc33bd3d5d77a1e7de781ff3add28d6
Author: shyjsarah <[email protected]>
AuthorDate: Fri Aug 28 11:43:20 2026 +0800
feat: add native object table read support (#749)
---
crates/integrations/datafusion/src/catalog.rs | 35 +-
crates/integrations/datafusion/src/table/mod.rs | 3 +
crates/integrations/datafusion/src/table/object.rs | 303 +++++++++++++++++
.../integrations/datafusion/tests/object_table.rs | 155 +++++++++
crates/paimon/src/catalog/filesystem.rs | 150 ++++++++-
crates/paimon/src/catalog/mod.rs | 23 +-
crates/paimon/src/catalog/rest/rest_catalog.rs | 12 +
crates/paimon/src/io/file_io.rs | 273 ++++++++++++++--
crates/paimon/src/spec/table_type.rs | 9 +-
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/src/table/object_table.rs | 364 +++++++++++++++++++++
crates/paimon/src/table/rest_env.rs | 113 +++++--
crates/paimon/tests/rest_catalog_test.rs | 39 +++
13 files changed, 1421 insertions(+), 60 deletions(-)
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index cbee1a6e..d8893243 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<dyn Fn() -> Option<SessionState> +
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<dyn TableProvider>))
+ }
Ok(paimon::catalog::LoadedTable::External(external)) => {
let declared = external.declared();
if branch.is_some() {
@@ -794,6 +813,10 @@ impl SchemaProvider for PaimonSchemaProvider {
)) as Arc<dyn TableProvider>))
}
Err(e) => Err(to_datafusion_error(e)),
+ Ok(_) => Err(plan_datafusion_err!(
+ "catalog returned an unsupported loaded table kind for
'{}'",
+ identifier.full_name()
+ )),
}
})
.await
@@ -855,6 +878,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.
@@ -912,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/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<DataField> {
diff --git a/crates/integrations/datafusion/src/table/object.rs
b/crates/integrations/datafusion/src/table/object.rs
new file mode 100644
index 00000000..55dd673c
--- /dev/null
+++ b/crates/integrations/datafusion/src/table/object.rs
@@ -0,0 +1,303 @@
+// 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, RecordBatchOptions,
StringArray,
+ StringViewArray,
+};
+use datafusion::arrow::datatypes::SchemaRef;
+use datafusion::catalog::Session;
+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 futures::{StreamExt, TryStreamExt};
+use paimon::table::{ObjectEntry, 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<Self> {
+ let schema = datafusion_arrow_schema(&ObjectTable::fields(),
schema_force_view_types)?;
+ Ok(Self { table, schema })
+ }
+}
+
+#[derive(Debug, Clone)]
+struct ObjectPartitionStream {
+ table: ObjectTable,
+ projection: Arc<[usize]>,
+ schema: SchemaRef,
+ limit: Option<usize>,
+}
+
+impl PartitionStream for ObjectPartitionStream {
+ fn schema(&self) -> &SchemaRef {
+ &self.schema
+ }
+
+ fn execute(&self, ctx: Arc<TaskContext>) -> 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::<DFResult<Vec<_>>>()?;
+ 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 {
+ Arc::clone(&self.schema)
+ }
+
+ fn table_type(&self) -> TableType {
+ TableType::Base
+ }
+
+ async fn scan(
+ &self,
+ _state: &dyn Session,
+ projection: Option<&Vec<usize>>,
+ _filters: &[Expr],
+ limit: Option<usize>,
+ ) -> DFResult<Arc<dyn ExecutionPlan>> {
+ 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<dyn PartitionStream> =
Arc::new(ObjectPartitionStream {
+ table: self.table.clone(),
+ projection: projection.into(),
+ schema: Arc::clone(&projected_schema),
+ limit,
+ });
+
+ Ok(Arc::new(StreamingTableExec::try_new(
+ projected_schema,
+ vec![partition],
+ None,
+ std::iter::empty(),
+ false,
+ limit,
+ )?))
+ }
+
+ async fn insert_into(
+ &self,
+ _state: &dyn Session,
+ _input: Arc<dyn ExecutionPlan>,
+ _insert_op: InsertOp,
+ ) -> DFResult<Arc<dyn ExecutionPlan>> {
+ Err(DataFusionError::NotImplemented(format!(
+ "Object table '{}' is read-only",
+ self.table.identifier().full_name()
+ )))
+ }
+}
+
+fn object_entries_to_batch(
+ entries: &[ObjectEntry],
+ projection: &[usize],
+ schema: &SchemaRef,
+) -> DFResult<RecordBatch> {
+ let columns = projection
+ .iter()
+ .enumerate()
+ .map(|(output_index, source_index)| -> DFResult<ArrayRef> {
+ 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::<DFResult<Vec<_>>>()?;
+ 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<Item = &'a str>,
+ 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::<Vec<_>>();
+ 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::<StreamingTableExec>()
+ .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::<Vec<_>>()
+ .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
new file mode 100644
index 00000000..3211b842
--- /dev/null
+++ b/crates/integrations/datafusion/tests/object_table.rs
@@ -0,0 +1,155 @@
+// 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::<Vec<_>>(),
+ );
+ }
+ }
+
+ assert_eq!(
+ rows,
+ vec![
+ vec!["nested/child.bin", "child.bin", "5", "true", "0", "NULL"],
+ vec!["root.txt", "root.txt", "4", "true", "0", "NULL"],
+ ]
+ );
+
+ 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
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ limited.iter().map(|batch| batch.num_rows()).sum::<usize>(),
+ 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(ordered_limited[0].column(0).as_ref(),
0).unwrap(),
+ "nested/child.bin"
+ );
+
+ 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..45628ad6 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -24,10 +24,10 @@ 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};
+use crate::table::{ObjectTable, SchemaManager, Table};
use async_trait::async_trait;
use bytes::Bytes;
use opendal::raw::get_basename;
@@ -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<std::sync::Arc<LocalCache>>,
}
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<FileIO> {
+ 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,19 @@ 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())
+ .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);
+ }
if declared.requires_table_engine() {
return crate::catalog::LoadedTable::external(
declared,
@@ -395,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?;
@@ -827,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 2b91ff19..cd3a8901 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -263,14 +263,17 @@ 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)]
+#[non_exhaustive]
pub enum LoadedTable {
/// A constructed Paimon table (boxed: far larger than the other variant).
Paimon(Box<Table>),
- /// 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 +374,10 @@ pub trait Catalog: Send + Sync {
/// * [`crate::Error::TableNotExist`] - table does not exist.
async fn get_table(&self, identifier: &Identifier) -> Result<Table>;
- /// 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 +393,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 +538,7 @@ pub trait Catalog: Send + Sync {
async fn list_partitions(&self, identifier: &Identifier) ->
Result<Vec<Partition>> {
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/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index c73f6c0a..79582300 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -26,6 +26,8 @@ use std::time::SystemTime;
use bytes::Bytes;
use chrono::{DateTime, Utc};
+use futures::stream::BoxStream;
+use futures::{StreamExt, TryStreamExt};
use opendal::raw::normalize_root;
use opendal::Operator;
use snafu::ResultExt;
@@ -223,42 +225,73 @@ impl FileIO {
/// List all files recursively under the given directory path.
pub async fn list_status_recursive(&self, path: &str) ->
Result<Vec<FileStatus>> {
+ self.list_status_recursive_with_limit(path, None).await
+ }
+
+ pub(crate) async fn list_status_recursive_with_limit(
+ &self,
+ path: &str,
+ limit: Option<usize>,
+ ) -> Result<Vec<FileStatus>> {
+ self.list_status_recursive_stream(path, limit)
+ .await?
+ .try_collect()
+ .await
+ }
+
+ pub(crate) async fn list_status_recursive_stream(
+ &self,
+ path: &str,
+ limit: Option<usize>,
+ ) -> Result<BoxStream<'static, Result<FileStatus>>> {
+ if limit == Some(0) {
+ 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 entries =
- op.list_with(&list_path)
+ op.lister_with(&list_path)
.recursive(true)
.await
.context(IoUnexpectedSnafu {
message: format!("Failed to list files recursively in
'{path}'"),
})?;
- let mut statuses = Vec::new();
- let list_path_normalized = list_path.trim_start_matches('/');
- for entry in entries {
- 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 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::<Utc>::from(SystemTime::from(v))),
+ };
+ emitted += 1;
+ if limit.is_some_and(|limit| emitted >= limit) {
+ break;
+ }
}
- statuses.push(FileStatus {
- size: meta.content_length(),
- is_dir: false,
- path: status_path(base_path, entry_path),
- last_modified: meta
- .last_modified()
- .map(|v| DateTime::<Utc>::from(SystemTime::from(v))),
- });
- }
-
- Ok(statuses)
+ }))
}
/// Check if exists.
@@ -794,10 +827,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<AtomicUsize>,
+ }
+
+ #[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<AtomicUsize>,
+ }
+
+ 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<RpCreateDir> {
+ Err(unsupported_test_operation())
+ }
+
+ async fn stat(
+ &self,
+ _ctx: &OperationContext,
+ _path: &str,
+ _args: OpStat,
+ ) -> opendal::Result<RpStat> {
+ Err(unsupported_test_operation())
+ }
+
+ fn read(
+ &self,
+ _ctx: &OperationContext,
+ _path: &str,
+ _args: OpRead,
+ ) -> opendal::Result<Self::Reader> {
+ Err(unsupported_test_operation())
+ }
+
+ fn write(
+ &self,
+ _ctx: &OperationContext,
+ _path: &str,
+ _args: OpWrite,
+ ) -> opendal::Result<Self::Writer> {
+ Err(unsupported_test_operation())
+ }
+
+ fn delete(&self, _ctx: &OperationContext) ->
opendal::Result<Self::Deleter> {
+ Err(unsupported_test_operation())
+ }
+
+ fn list(
+ &self,
+ _ctx: &OperationContext,
+ _path: &str,
+ _args: OpList,
+ ) -> opendal::Result<Self::Lister> {
+ Ok(CountingLister {
+ pulls: Arc::clone(&self.pulls),
+ next: 0,
+ })
+ }
+
+ fn copy(
+ &self,
+ _ctx: &OperationContext,
+ _from: &str,
+ _to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> opendal::Result<Self::Copier> {
+ Err(unsupported_test_operation())
+ }
+
+ async fn rename(
+ &self,
+ _ctx: &OperationContext,
+ _from: &str,
+ _to: &str,
+ _args: OpRename,
+ ) -> opendal::Result<RpRename> {
+ Err(unsupported_test_operation())
+ }
+
+ async fn presign(
+ &self,
+ _ctx: &OperationContext,
+ _path: &str,
+ _args: OpPresign,
+ ) -> opendal::Result<RpPresign> {
+ 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<AtomicUsize>,
+ next: usize,
+ }
+
+ impl oio::List for CountingLister {
+ async fn next(&mut self) -> opendal::Result<Option<oio::Entry>> {
+ 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()
@@ -910,6 +1104,39 @@ 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_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/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 f8b99f9f..35f1e45a 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")]
@@ -126,6 +127,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 pk_vector_bucket_split::{BucketVectorPayload, BucketVectorSearchSplit};
pub use postpone_bucket_plan::{PostponeBucketPlan,
POSTPONE_BUCKET_PLAN_TOTAL_BUCKETS_FIELD};
diff --git a/crates/paimon/src/table/object_table.rs
b/crates/paimon/src/table/object_table.rs
new file mode 100644
index 00000000..7398125e
--- /dev/null
+++ b/crates/paimon/src/table/object_table.rs
@@ -0,0 +1,364 @@
+// 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, Schema, TableSchema,
TableType, VarCharType,
+ 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)]
+pub struct ObjectEntry {
+ path: String,
+ name: String,
+ length: i64,
+ mtime: i64,
+ atime: i64,
+ owner: Option<String>,
+}
+
+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<String>,
+}
+
+impl ObjectTable {
+ pub(crate) fn normalize_creation(schema: &Schema, default_location: &str)
-> Result<Schema> {
+ 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> {
+ 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<Self> {
+ 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())
+ .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",
+ 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<DataField> {
+ 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<Vec<ObjectEntry>> {
+ self.list_objects_with_limit(None).await
+ }
+
+ /// 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<usize>) ->
Result<Vec<ObjectEntry>> {
+ let mut entries = self
+ .stream_objects(limit)
+ .await?
+ .try_collect::<Vec<_>>()
+ .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<usize>,
+ ) -> Result<BoxStream<'static, Result<ObjectEntry>>> {
+ 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<ObjectEntry> {
+ 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 {
+ 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()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use bytes::Bytes;
+
+ use super::*;
+
+ #[tokio::test]
+ 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"] {
+ 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::<Vec<_>>();
+ 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());
+ }
+}
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<RESTApi>,
+ options: Options,
+ data_token_enabled: bool,
+ local_cache: Option<Arc<LocalCache>>,
+ ) -> Result<ObjectTable> {
+ 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<RESTApi>,
+ options: &Options,
+ data_token_enabled: bool,
+ is_external: bool,
+ local_cache: Option<Arc<LocalCache>>,
+ ) -> Result<FileIO> {
+ 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<dyn SnapshotCommit> {
Arc::new(RESTSnapshotCommit::new(
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 83da7c8a..41052c45 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -1524,6 +1524,45 @@ 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.location(), object_location);
+ #[cfg(not(windows))]
+ assert_eq!(
+ table
+ .list_objects()
+ .await
+ .unwrap()
+ .iter()
+ .map(|entry| entry.path())
+ .collect::<Vec<_>>(),
+ vec!["object.txt"]
+ );
+}
+
#[tokio::test]
async fn test_load_table_fails_closed_on_query_auth() {
let ctx = setup_catalog(vec!["default"]).await;
