diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index cbee1a6e7..d8893243a 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() { @@ -794,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 @@ -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 c55b97e75..262c3217c 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 000000000..55dd673c9 --- /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 { + 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, +} + +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 { + 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 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, + }); + + 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, + _insert_op: InsertOp, + ) -> DFResult> { + 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 { + 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 new file mode 100644 index 000000000..3211b8425 --- /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::>(), + ); + } + } + + 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::(), + 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 0ebd8ab84..45628ad66 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>, } 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,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 2b91ff195..cd3a89016 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), - /// 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
; - /// 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> { 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 be41cfc87..7559a3691 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 63735486c..886033eb7 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> { + self.list_status_recursive_with_limit(path, None).await + } + + pub(crate) async fn list_status_recursive_with_limit( + &self, + 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(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::::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::::from(SystemTime::from(v))), - }); - } - - Ok(statuses) + })) } /// Check if exists. @@ -793,10 +826,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() @@ -909,6 +1103,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 e4a095d23..43166d80c 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 b415f0dd4..d970f5468 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 000000000..7398125e9 --- /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, +} + +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(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 { + 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 { + 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> { + 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) -> Result> { + let mut entries = self + .stream_objects(limit) + .await? + .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 { + 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::>(); + 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 7d716cb7a..133acf698 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 83da7c8a7..41052c453 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!["object.txt"] + ); +} + #[tokio::test] async fn test_load_table_fails_closed_on_query_auth() { let ctx = setup_catalog(vec!["default"]).await;