This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 39e3e14 Support REST format table reads (#492)
39e3e14 is described below
commit 39e3e149e1233eba02543e5c61d3dedea75c7a34
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Jul 9 18:54:36 2026 +0800
Support REST format table reads (#492)
---
crates/paimon/src/spec/core_options.rs | 33 ++
crates/paimon/src/table/format_read_builder.rs | 109 ++++
crates/paimon/src/table/format_table_read.rs | 373 ++++++++++++++
crates/paimon/src/table/format_table_scan.rs | 641 ++++++++++++++++++++++++
crates/paimon/src/table/format_write_builder.rs | 77 +++
crates/paimon/src/table/mod.rs | 8 +
crates/paimon/src/table/read_builder.rs | 136 ++++-
crates/paimon/src/table/rest_env.rs | 20 +-
crates/paimon/src/table/table_read.rs | 87 +++-
crates/paimon/src/table/table_scan.rs | 108 +++-
crates/paimon/src/table/write_builder.rs | 88 +++-
crates/paimon/tests/rest_catalog_test.rs | 247 ++++++++-
12 files changed, 1902 insertions(+), 25 deletions(-)
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index fd685ee..b77ab95 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -31,6 +31,8 @@ const SOURCE_SPLIT_TARGET_SIZE_OPTION: &str =
"source.split.target-size";
const SOURCE_SPLIT_OPEN_FILE_COST_OPTION: &str = "source.split.open-file-cost";
const PARTITION_DEFAULT_NAME_OPTION: &str = "partition.default-name";
const PARTITION_LEGACY_NAME_OPTION: &str = "partition.legacy-name";
+const FORMAT_TABLE_PARTITION_PATH_ONLY_VALUE_OPTION: &str =
+ "format-table.partition-path-only-value";
pub(crate) const BUCKET_KEY_OPTION: &str = "bucket-key";
const BUCKET_FUNCTION_TYPE_OPTION: &str = "bucket-function.type";
const BUCKET_OPTION: &str = "bucket";
@@ -54,6 +56,9 @@ const CHANGELOG_FILE_FORMAT_OPTION: &str =
"changelog-file.format";
const CHANGELOG_FILE_COMPRESSION_OPTION: &str = "changelog-file.compression";
const CHANGELOG_FILE_STATS_MODE_OPTION: &str = "changelog-file.stats-mode";
const ROW_TRACKING_ENABLED_OPTION: &str = "row-tracking.enabled";
+pub(crate) const TABLE_TYPE_OPTION: &str = "type";
+pub(crate) const FORMAT_TABLE_TYPE: &str = "format-table";
+pub(crate) const PATH_OPTION: &str = "path";
const MANIFEST_COMPRESSION_OPTION: &str = "manifest.compression";
const MANIFEST_TARGET_FILE_SIZE_OPTION: &str = "manifest.target-file-size";
const MANIFEST_TARGET_SIZE_OPTION: &str = "manifest.target-size";
@@ -386,6 +391,24 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(false)
}
+ pub fn is_format_table(&self) -> bool {
+ self.options
+ .get(TABLE_TYPE_OPTION)
+ .map(|value| value.eq_ignore_ascii_case(FORMAT_TABLE_TYPE))
+ .unwrap_or(false)
+ }
+
+ pub fn path(&self) -> Option<&str> {
+ self.options.get(PATH_OPTION).map(String::as_str)
+ }
+
+ pub fn format_table_partition_only_value_in_path(&self) -> bool {
+ self.options
+ .get(FORMAT_TABLE_PARTITION_PATH_ONLY_VALUE_OPTION)
+ .map(|value| value.eq_ignore_ascii_case("true"))
+ .unwrap_or(false)
+ }
+
pub fn global_index_enabled(&self) -> bool {
self.options
.get(GLOBAL_INDEX_ENABLED_OPTION)
@@ -1158,6 +1181,16 @@ mod tests {
assert!(!core.legacy_partition_name());
}
+ #[test]
+ fn test_format_table_partition_only_value_in_path() {
+ let options = HashMap::from([(
+ FORMAT_TABLE_PARTITION_PATH_ONLY_VALUE_OPTION.to_string(),
+ "true".to_string(),
+ )]);
+ let core = CoreOptions::new(&options);
+ assert!(core.format_table_partition_only_value_in_path());
+ }
+
#[test]
fn test_try_time_travel_selector_rejects_conflicting_selectors() {
let options = HashMap::from([
diff --git a/crates/paimon/src/table/format_read_builder.rs
b/crates/paimon/src/table/format_read_builder.rs
new file mode 100644
index 0000000..06acf51
--- /dev/null
+++ b/crates/paimon/src/table/format_read_builder.rs
@@ -0,0 +1,109 @@
+// 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.
+
+//! Read builder for Java-compatible `type=format-table` metadata.
+
+use super::partition_filter::PartitionFilter;
+use super::read_builder::resolve_projected_fields;
+use super::read_builder::split_scan_predicates;
+use super::{Table, TableRead, TableScan};
+use crate::spec::{CoreOptions, DataField, Predicate};
+use crate::table::source::RowRange;
+use crate::Result;
+
+#[derive(Debug, Clone)]
+pub(crate) struct FormatReadBuilder<'a> {
+ table: &'a Table,
+ read_type: Option<Vec<DataField>>,
+ partition_filter: Option<PartitionFilter>,
+ data_predicates: Vec<Predicate>,
+ limit: Option<usize>,
+}
+
+impl<'a> FormatReadBuilder<'a> {
+ pub(crate) fn new(table: &'a Table) -> Self {
+ Self {
+ table,
+ read_type: None,
+ partition_filter: None,
+ data_predicates: Vec::new(),
+ limit: None,
+ }
+ }
+
+ pub(crate) fn with_projection(&mut self, columns: &[&str]) -> Result<&mut
Self> {
+ let projection_names = columns.iter().map(|c|
(*c).to_string()).collect::<Vec<_>>();
+ self.read_type = Some(resolve_projected_fields(
+ self.table.identifier().full_name(),
+ self.table.schema().fields(),
+ &projection_names,
+ )?);
+ Ok(self)
+ }
+
+ pub(crate) fn with_read_type(&mut self, read_type: Vec<DataField>) -> &mut
Self {
+ self.read_type = Some(read_type);
+ self
+ }
+
+ pub(crate) fn with_filter(&mut self, filter: Predicate) -> &mut Self {
+ let (partition_predicate, data_predicates) =
split_scan_predicates(self.table, filter);
+ self.partition_filter = partition_predicate.map(|pred| {
+ PartitionFilter::from_predicate(pred,
&self.table.schema().partition_fields())
+ });
+ self.data_predicates = data_predicates;
+ self
+ }
+
+ pub(crate) fn is_exact_filter_pushdown(&self, _filter: &Predicate) -> bool
{
+ false
+ }
+
+ pub(crate) fn with_row_ranges(&mut self, _ranges: Vec<RowRange>) -> &mut
Self {
+ self
+ }
+
+ pub(crate) fn with_limit(&mut self, limit: usize) -> &mut Self {
+ self.limit = Some(limit);
+ self
+ }
+
+ pub(crate) fn new_scan(&self) -> TableScan<'a> {
+ TableScan::new(
+ self.table,
+ self.partition_filter.clone(),
+ Vec::new(),
+ None,
+ self.limit,
+ None,
+ )
+ }
+
+ pub(crate) fn new_read(&self) -> Result<TableRead<'a>> {
+
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ let read_type = match &self.read_type {
+ None => self.table.schema().fields().to_vec(),
+ Some(fields) => fields.clone(),
+ };
+ Ok(TableRead::new_format(
+ self.table,
+ read_type,
+ self.data_predicates.clone(),
+ self.limit,
+ ))
+ }
+}
diff --git a/crates/paimon/src/table/format_table_read.rs
b/crates/paimon/src/table/format_table_read.rs
new file mode 100644
index 0000000..398ec91
--- /dev/null
+++ b/crates/paimon/src/table/format_table_read.rs
@@ -0,0 +1,373 @@
+// 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.
+
+//! Read implementation for Java-compatible `type=format-table` metadata.
+
+use super::data_file_reader::DataFileReader;
+use super::read_builder::split_scan_predicates;
+use super::{ArrowRecordBatchStream, Table};
+use crate::arrow::{build_target_arrow_schema, paimon_type_to_arrow};
+use crate::spec::{extract_datum, BinaryRow, CoreOptions, DataField, DataType,
Datum, Predicate};
+use crate::{DataSplit, Error};
+use arrow_array::{
+ new_null_array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
Float32Array, Float64Array,
+ Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch,
RecordBatchOptions, StringArray,
+ Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use async_stream::try_stream;
+use futures::StreamExt;
+use std::sync::Arc;
+
+#[derive(Debug, Clone)]
+pub(crate) struct FormatTableRead<'a> {
+ table: &'a Table,
+ read_type: Vec<DataField>,
+ data_predicates: Vec<Predicate>,
+ limit: Option<usize>,
+}
+
+impl<'a> FormatTableRead<'a> {
+ pub(crate) fn new(
+ table: &'a Table,
+ read_type: Vec<DataField>,
+ data_predicates: Vec<Predicate>,
+ limit: Option<usize>,
+ ) -> Self {
+ Self {
+ table,
+ read_type,
+ data_predicates,
+ limit,
+ }
+ }
+
+ pub(crate) fn read_type(&self) -> &[DataField] {
+ &self.read_type
+ }
+
+ pub(crate) fn data_predicates(&self) -> &[Predicate] {
+ &self.data_predicates
+ }
+
+ pub(crate) fn table(&self) -> &Table {
+ self.table
+ }
+
+ pub(crate) fn with_filter(mut self, filter: Predicate) -> Self {
+ self.data_predicates = split_scan_predicates(self.table, filter).1;
+ self
+ }
+
+ pub(crate) fn to_arrow(
+ &self,
+ data_splits: &[DataSplit],
+ ) -> crate::Result<ArrowRecordBatchStream> {
+
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ let read_type = self.read_type.clone();
+ let output_schema = build_target_arrow_schema(&read_type)?;
+ let partition_keys = self.table.schema().partition_keys().to_vec();
+ let partition_fields = self.table.schema().partition_fields();
+ let (data_read_type, data_columns, partition_columns) =
+ split_format_read_type(&read_type, &partition_keys);
+ let table_fields = self.table.schema().fields().to_vec();
+ let (data_table_fields, data_predicates) =
+ split_format_table_fields(&table_fields, &partition_keys,
&self.data_predicates);
+
+ let splits = data_splits.to_vec();
+ let file_io = self.table.file_io().clone();
+ let schema_manager = self.table.schema_manager().clone();
+ let schema_id = self.table.schema().id();
+ let mut remaining = self.limit;
+
+ Ok(try_stream! {
+ for split in splits {
+ if matches!(remaining, Some(0)) {
+ break;
+ }
+
+ let mut stream = DataFileReader::new(
+ file_io.clone(),
+ schema_manager.clone(),
+ schema_id,
+ data_table_fields.clone(),
+ data_read_type.clone(),
+ data_predicates.clone(),
+ )
+ .read(std::slice::from_ref(&split))?;
+
+ while let Some(batch) = stream.next().await {
+ if matches!(remaining, Some(0)) {
+ break;
+ }
+
+ let batch = project_format_batch(
+ batch?,
+ &split,
+ &read_type,
+ &data_columns,
+ &partition_columns,
+ &partition_fields,
+ &output_schema,
+ )?;
+ let Some(batch) = apply_limit(batch, &mut remaining) else {
+ break;
+ };
+ yield batch;
+ }
+ }
+ }
+ .boxed())
+ }
+}
+
+fn split_format_read_type(
+ read_type: &[DataField],
+ partition_keys: &[String],
+) -> (Vec<DataField>, Vec<Option<usize>>, Vec<Option<usize>>) {
+ let mut data_read_type = Vec::new();
+ let mut data_columns = Vec::with_capacity(read_type.len());
+ let mut partition_columns = Vec::with_capacity(read_type.len());
+
+ for field in read_type {
+ if let Some(partition_index) = partition_keys.iter().position(|key|
key == field.name()) {
+ data_columns.push(None);
+ partition_columns.push(Some(partition_index));
+ } else {
+ data_columns.push(Some(data_read_type.len()));
+ partition_columns.push(None);
+ data_read_type.push(field.clone());
+ }
+ }
+
+ (data_read_type, data_columns, partition_columns)
+}
+
+fn split_format_table_fields(
+ table_fields: &[DataField],
+ partition_keys: &[String],
+ predicates: &[Predicate],
+) -> (Vec<DataField>, Vec<Predicate>) {
+ let mut data_fields = Vec::new();
+ let mut data_mapping = Vec::with_capacity(table_fields.len());
+
+ for field in table_fields {
+ if partition_keys.iter().any(|key| key == field.name()) {
+ data_mapping.push(None);
+ } else {
+ data_mapping.push(Some(data_fields.len()));
+ data_fields.push(field.clone());
+ }
+ }
+
+ let data_predicates = predicates
+ .iter()
+ .filter_map(|predicate|
predicate.project_field_index_inclusive(&data_mapping))
+ .collect();
+
+ (data_fields, data_predicates)
+}
+
+fn project_format_batch(
+ batch: RecordBatch,
+ split: &DataSplit,
+ read_type: &[DataField],
+ data_columns: &[Option<usize>],
+ partition_columns: &[Option<usize>],
+ partition_fields: &[DataField],
+ output_schema: &Arc<arrow_schema::Schema>,
+) -> crate::Result<RecordBatch> {
+ let num_rows = batch.num_rows();
+ let mut columns = Vec::with_capacity(read_type.len());
+
+ for (idx, field) in read_type.iter().enumerate() {
+ if let Some(data_index) = data_columns[idx] {
+ columns.push(batch.column(data_index).clone());
+ continue;
+ }
+
+ let Some(partition_index) = partition_columns[idx] else {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Format table read field '{}' is neither data nor
partition",
+ field.name()
+ ),
+ source: None,
+ });
+ };
+ let Some(partition_field) = partition_fields.get(partition_index) else
{
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Format table partition field '{}' is missing from
partition schema",
+ field.name()
+ ),
+ source: None,
+ });
+ };
+ columns.push(partition_array(
+ split.partition(),
+ partition_index,
+ partition_field.data_type(),
+ num_rows,
+ )?);
+ }
+
+ if columns.is_empty() {
+ RecordBatch::try_new_with_options(
+ output_schema.clone(),
+ columns,
+ &RecordBatchOptions::new().with_row_count(Some(num_rows)),
+ )
+ } else {
+ RecordBatch::try_new(output_schema.clone(), columns)
+ }
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build format table RecordBatch: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+fn partition_array(
+ partition: &BinaryRow,
+ partition_index: usize,
+ data_type: &DataType,
+ num_rows: usize,
+) -> crate::Result<ArrayRef> {
+ let arrow_type = paimon_type_to_arrow(data_type)?;
+ if partition.arity() <= partition_index as i32 ||
partition.is_null_at(partition_index) {
+ return Ok(new_null_array(&arrow_type, num_rows));
+ }
+
+ let datum = extract_datum(partition, partition_index, data_type)?;
+ let Some(datum) = datum else {
+ return Ok(new_null_array(&arrow_type, num_rows));
+ };
+
+ Ok(match (datum, data_type) {
+ (Datum::Bool(value), DataType::Boolean(_)) => {
+ Arc::new(BooleanArray::from(vec![Some(value); num_rows]))
+ }
+ (Datum::TinyInt(value), DataType::TinyInt(_)) => {
+ Arc::new(Int8Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::SmallInt(value), DataType::SmallInt(_)) => {
+ Arc::new(Int16Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Int(value), DataType::Int(_)) => {
+ Arc::new(Int32Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Long(value), DataType::BigInt(_)) => {
+ Arc::new(Int64Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Float(value), DataType::Float(_)) => {
+ Arc::new(Float32Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Double(value), DataType::Double(_)) => {
+ Arc::new(Float64Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::String(value), DataType::Char(_) | DataType::VarChar(_)) => {
+ let values = std::iter::repeat_with(|| Some(value.as_str()))
+ .take(num_rows)
+ .collect::<Vec<_>>();
+ Arc::new(StringArray::from(values))
+ }
+ (Datum::Bytes(value), DataType::Binary(_) | DataType::VarBinary(_)) =>
{
+ let values = std::iter::repeat_with(|| Some(value.as_slice()))
+ .take(num_rows)
+ .collect::<Vec<_>>();
+ Arc::new(BinaryArray::from(values))
+ }
+ (Datum::Date(value), DataType::Date(_)) => {
+ Arc::new(Date32Array::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Time(value), DataType::Time(_)) => {
+ Arc::new(Time32MillisecondArray::from(vec![Some(value); num_rows]))
+ }
+ (Datum::Timestamp { millis, nanos }, DataType::Timestamp(ts)) => {
+ timestamp_array(millis, nanos, ts.precision(), None, num_rows)?
+ }
+ (Datum::LocalZonedTimestamp { millis, nanos },
DataType::LocalZonedTimestamp(ts)) => {
+ timestamp_array(millis, nanos, ts.precision(), Some("UTC"),
num_rows)?
+ }
+ (_, other) => {
+ return Err(Error::Unsupported {
+ message: format!(
+ "Format table partition column type '{other:?}' is not
supported by the Rust reader yet"
+ ),
+ });
+ }
+ })
+}
+
+fn timestamp_array(
+ millis: i64,
+ nanos: i32,
+ precision: u32,
+ timezone: Option<&'static str>,
+ num_rows: usize,
+) -> crate::Result<ArrayRef> {
+ let array: ArrayRef = match precision {
+ 0..=3 => {
+ let array = TimestampMillisecondArray::from(vec![Some(millis);
num_rows]);
+ match timezone {
+ Some(tz) => Arc::new(array.with_timezone(tz)),
+ None => Arc::new(array),
+ }
+ }
+ 4..=6 => {
+ let value = millis * 1_000 + (nanos as i64) / 1_000;
+ let array = TimestampMicrosecondArray::from(vec![Some(value);
num_rows]);
+ match timezone {
+ Some(tz) => Arc::new(array.with_timezone(tz)),
+ None => Arc::new(array),
+ }
+ }
+ 7..=9 => {
+ let value = millis * 1_000_000 + (nanos as i64);
+ let array = TimestampNanosecondArray::from(vec![Some(value);
num_rows]);
+ match timezone {
+ Some(tz) => Arc::new(array.with_timezone(tz)),
+ None => Arc::new(array),
+ }
+ }
+ _ => {
+ return Err(Error::Unsupported {
+ message: format!(
+ "Unsupported timestamp precision for format table
partition: {precision}"
+ ),
+ });
+ }
+ };
+ Ok(array)
+}
+
+fn apply_limit(batch: RecordBatch, remaining: &mut Option<usize>) ->
Option<RecordBatch> {
+ let Some(value) = remaining else {
+ return Some(batch);
+ };
+ if *value == 0 {
+ return None;
+ }
+ if batch.num_rows() > *value {
+ let limited = batch.slice(0, *value);
+ *value = 0;
+ Some(limited)
+ } else {
+ *value -= batch.num_rows();
+ Some(batch)
+ }
+}
diff --git a/crates/paimon/src/table/format_table_scan.rs
b/crates/paimon/src/table/format_table_scan.rs
new file mode 100644
index 0000000..2890766
--- /dev/null
+++ b/crates/paimon/src/table/format_table_scan.rs
@@ -0,0 +1,641 @@
+// 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.
+
+//! Scan implementation for Java-compatible `type=format-table` metadata.
+
+use super::{Plan, ScanTrace, Table};
+use crate::spec::stats::BinaryTableStats;
+use crate::spec::{
+ extract_datum, BinaryRow, BinaryRowBuilder, CoreOptions, DataField,
DataFileMeta, DataType,
+ Datum, PartitionComputer, Predicate, PredicateOperator,
+};
+use crate::table::partition_filter::PartitionFilter;
+use crate::table::source::DataSplitBuilder;
+use chrono::NaiveDate;
+
+#[derive(Debug, Clone)]
+pub(crate) struct FormatTableScan<'a> {
+ table: &'a Table,
+ partition_filter: Option<PartitionFilter>,
+ limit: Option<usize>,
+}
+
+impl<'a> FormatTableScan<'a> {
+ pub(crate) fn new(
+ table: &'a Table,
+ partition_filter: Option<PartitionFilter>,
+ limit: Option<usize>,
+ ) -> Self {
+ Self {
+ table,
+ partition_filter,
+ limit,
+ }
+ }
+
+ pub(crate) async fn plan(&self) -> crate::Result<Plan> {
+ self.ensure_query_auth_allowed()?;
+ self.plan_inner(None).await
+ }
+
+ pub(crate) async fn plan_with_trace(&self) -> crate::Result<(Plan,
ScanTrace)> {
+ self.ensure_query_auth_allowed()?;
+ let mut trace = ScanTrace::default();
+ let plan = self.plan_inner(Some(&mut trace)).await?;
+ Ok((plan, trace))
+ }
+
+ fn ensure_query_auth_allowed(&self) -> crate::Result<()> {
+
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()
+ }
+
+ async fn plan_inner(&self, trace: Option<&mut ScanTrace>) ->
crate::Result<Plan> {
+ let core_options = CoreOptions::new(self.table.schema().options());
+ let format_extension =
supported_format_table_extension(core_options.file_format())?;
+ let schema_id = self.table.schema().id();
+ let table_path = core_options
+ .path()
+ .unwrap_or_else(|| self.table.location())
+ .trim_end_matches('/')
+ .to_string();
+
+ let partition_fields = self.table.schema().partition_fields();
+ let mut splits = Vec::new();
+ for scan_root in self.scan_roots(&core_options, &table_path)? {
+ let statuses = self
+ .list_status_recursive_if_exists(&scan_root.path)
+ .await?;
+ for status in statuses {
+ if let Some(split) = self
+ .status_to_split(
+ status,
+ &table_path,
+ format_extension,
+ schema_id,
+ &partition_fields,
+ scan_root.partition.clone(),
+ )
+ .await?
+ {
+ splits.push(split);
+ }
+ }
+ }
+
+ splits.sort_by(|left, right| {
+ left.bucket_path().cmp(right.bucket_path()).then_with(|| {
+ left.data_files()[0]
+ .file_name
+ .cmp(&right.data_files()[0].file_name)
+ })
+ });
+ splits = self.apply_limit_pushdown(splits);
+
+ if let Some(trace) = trace {
+ trace.record_final_plan(splits.len(), splits.len(), splits.len());
+ }
+ Ok(Plan::new(splits))
+ }
+
+ fn scan_roots(
+ &self,
+ core_options: &CoreOptions<'_>,
+ table_path: &str,
+ ) -> crate::Result<Vec<ScanRoot>> {
+ let partition_keys = self.table.schema().partition_keys();
+ let partition_fields = self.table.schema().partition_fields();
+ if partition_keys.is_empty() {
+ return Ok(vec![ScanRoot {
+ path: table_path.to_string(),
+ partition: BinaryRow::new(0),
+ }]);
+ }
+
+ let Some(PartitionFilter::PartitionSet { partitions, .. }) =
&self.partition_filter else {
+ if let Some(PartitionFilter::Predicate(predicate)) =
&self.partition_filter {
+ if let Some(path) = leading_equality_partition_path(
+ table_path,
+ partition_keys,
+ &partition_fields,
+ predicate,
+ core_options.partition_default_name(),
+ core_options.legacy_partition_name(),
+ core_options.format_table_partition_only_value_in_path(),
+ ) {
+ return Ok(vec![ScanRoot {
+ path,
+ partition: BinaryRow::new(0),
+ }]);
+ }
+ }
+ return Ok(vec![ScanRoot {
+ path: table_path.to_string(),
+ partition: BinaryRow::new(0),
+ }]);
+ };
+
+ let partition_computer = PartitionComputer::new(
+ partition_keys,
+ self.table.schema().fields(),
+ core_options.partition_default_name(),
+ core_options.legacy_partition_name(),
+ )?;
+ let only_value_in_path =
core_options.format_table_partition_only_value_in_path();
+ let mut roots = Vec::with_capacity(partitions.len());
+ for partition in partitions {
+ let row = BinaryRow::from_serialized_bytes(partition)?;
+ let partition_path = if only_value_in_path {
+ partition_path_from_row(
+ &row,
+ &partition_fields,
+ core_options.partition_default_name(),
+ core_options.legacy_partition_name(),
+ true,
+ )?
+ } else {
+ partition_computer.generate_partition_path(&row)?
+ };
+ roots.push(ScanRoot {
+ path: join_path(table_path, &partition_path),
+ partition: row,
+ });
+ }
+ roots.sort_by(|left, right| left.path.cmp(&right.path));
+ Ok(roots)
+ }
+
+ async fn list_status_recursive_if_exists(
+ &self,
+ path: &str,
+ ) -> crate::Result<Vec<crate::io::FileStatus>> {
+ match self.table.file_io().list_status_recursive(path).await {
+ Ok(statuses) => Ok(statuses),
+ Err(err) => {
+ if !self.table.file_io().exists(path).await.unwrap_or(true) {
+ Ok(Vec::new())
+ } else {
+ Err(err)
+ }
+ }
+ }
+ }
+
+ async fn status_to_split(
+ &self,
+ status: crate::io::FileStatus,
+ table_path: &str,
+ format_extension: &str,
+ schema_id: i64,
+ partition_fields: &[DataField],
+ known_partition: BinaryRow,
+ ) -> crate::Result<Option<crate::DataSplit>> {
+ let Some((parent, file_name)) = split_parent_and_file(&status.path)
else {
+ return Ok(None);
+ };
+ let parent = parent.to_string();
+ let file_name = file_name.to_string();
+ if !is_format_table_data_file_name(&file_name) {
+ return Ok(None);
+ }
+ if !file_name.to_ascii_lowercase().ends_with(format_extension) {
+ return Ok(None);
+ }
+ let status = if status.size == 0 {
+ self.table.file_io().get_status(&status.path).await?
+ } else {
+ status
+ };
+ let file_size = i64::try_from(status.size).map_err(|_|
crate::Error::DataInvalid {
+ message: format!(
+ "Format table file '{}' is too large to fit in i64 metadata",
+ status.path
+ ),
+ source: None,
+ })?;
+ let data_file = data_file_meta(file_name, file_size, schema_id);
+ let partition = if partition_fields.is_empty() {
+ BinaryRow::new(0)
+ } else if known_partition.arity() == partition_fields.len() as i32
+ && !known_partition.is_empty()
+ {
+ known_partition
+ } else {
+ let core_options = CoreOptions::new(self.table.schema().options());
+ let Some(partition) = partition_row_from_path(
+ table_path,
+ &parent,
+ partition_fields,
+ self.table.schema().partition_keys(),
+ core_options.partition_default_name(),
+ core_options.format_table_partition_only_value_in_path(),
+ )?
+ else {
+ return Ok(None);
+ };
+ if !self.partition_matches(&partition)? {
+ return Ok(None);
+ }
+ partition
+ };
+
+ Ok(Some(
+ DataSplitBuilder::new()
+ .with_snapshot(0)
+ .with_partition(partition)
+ .with_bucket(0)
+ .with_bucket_path(parent)
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file])
+ .with_raw_convertible(true)
+ .build()?,
+ ))
+ }
+
+ fn partition_matches(&self, partition: &BinaryRow) -> crate::Result<bool> {
+ match &self.partition_filter {
+ Some(filter) =>
filter.matches_entry(&partition.to_serialized_bytes()),
+ None => Ok(true),
+ }
+ }
+
+ pub(crate) fn apply_limit_pushdown(
+ &self,
+ splits: Vec<crate::DataSplit>,
+ ) -> Vec<crate::DataSplit> {
+ match self.limit {
+ Some(0) => Vec::new(),
+ Some(limit) if splits.len() > limit =>
splits.into_iter().take(limit).collect(),
+ _ => splits,
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct ScanRoot {
+ path: String,
+ partition: BinaryRow,
+}
+
+fn is_format_table_data_file_name(file_name: &str) -> bool {
+ !file_name.is_empty() && !file_name.starts_with('.') &&
!file_name.starts_with('_')
+}
+
+fn split_parent_and_file(path: &str) -> Option<(&str, &str)> {
+ let trimmed = path.trim_end_matches('/');
+ let slash = trimmed.rfind('/')?;
+ Some((&trimmed[..slash], &trimmed[slash + 1..]))
+}
+
+fn join_path(parent: &str, child: &str) -> String {
+ let parent = parent.trim_end_matches('/');
+ let child = child.trim_start_matches('/').trim_end_matches('/');
+ if child.is_empty() {
+ parent.to_string()
+ } else {
+ format!("{parent}/{child}")
+ }
+}
+
+fn leading_equality_partition_path(
+ table_path: &str,
+ partition_keys: &[String],
+ partition_fields: &[DataField],
+ predicate: &Predicate,
+ default_partition_name: &str,
+ legacy_partition_name: bool,
+ only_value_in_path: bool,
+) -> Option<String> {
+ let predicates = predicate.clone().split_and();
+ let mut values: Vec<Option<&Datum>> = vec![None; partition_keys.len()];
+ for predicate in &predicates {
+ let Predicate::Leaf {
+ index,
+ op: PredicateOperator::Eq,
+ literals,
+ ..
+ } = predicate
+ else {
+ continue;
+ };
+ if *index < values.len() {
+ values[*index] = literals.first();
+ }
+ }
+
+ let mut segments = Vec::new();
+ for (idx, key) in partition_keys.iter().enumerate() {
+ let Some(datum) = values[idx] else {
+ break;
+ };
+ let value = partition_value_from_datum(
+ datum,
+ partition_fields[idx].data_type(),
+ default_partition_name,
+ legacy_partition_name,
+ )?;
+ if only_value_in_path {
+ segments.push(escape_path_name(&value));
+ } else {
+ segments.push(format!(
+ "{}={}",
+ escape_path_name(key),
+ escape_path_name(&value)
+ ));
+ }
+ }
+
+ if segments.is_empty() {
+ None
+ } else {
+ Some(join_path(table_path, &segments.join("/")))
+ }
+}
+
+fn partition_path_from_row(
+ row: &BinaryRow,
+ partition_fields: &[DataField],
+ default_partition_name: &str,
+ legacy_partition_name: bool,
+ only_value_in_path: bool,
+) -> crate::Result<String> {
+ let mut segments = Vec::with_capacity(partition_fields.len());
+ for (idx, field) in partition_fields.iter().enumerate() {
+ let value = match extract_datum(row, idx, field.data_type())? {
+ None => default_partition_name.to_string(),
+ Some(datum) => partition_value_from_datum(
+ &datum,
+ field.data_type(),
+ default_partition_name,
+ legacy_partition_name,
+ )
+ .ok_or_else(|| crate::Error::Unsupported {
+ message: format!(
+ "Format table partition path generation does not support
type '{:?}'",
+ field.data_type()
+ ),
+ })?,
+ };
+ if only_value_in_path {
+ segments.push(escape_path_name(&value));
+ } else {
+ segments.push(format!(
+ "{}={}",
+ escape_path_name(field.name()),
+ escape_path_name(&value)
+ ));
+ }
+ }
+ Ok(segments.join("/"))
+}
+
+fn partition_value_from_datum(
+ datum: &Datum,
+ data_type: &DataType,
+ default_partition_name: &str,
+ legacy_partition_name: bool,
+) -> Option<String> {
+ match (datum, data_type) {
+ (Datum::Bool(value), DataType::Boolean(_)) => Some(value.to_string()),
+ (Datum::TinyInt(value), DataType::TinyInt(_)) =>
Some(value.to_string()),
+ (Datum::SmallInt(value), DataType::SmallInt(_)) =>
Some(value.to_string()),
+ (Datum::Int(value), DataType::Int(_)) => Some(value.to_string()),
+ (Datum::Long(value), DataType::BigInt(_)) => Some(value.to_string()),
+ (Datum::String(value), DataType::Char(_) | DataType::VarChar(_)) => {
+ if value.trim().is_empty() {
+ Some(default_partition_name.to_string())
+ } else {
+ Some(value.clone())
+ }
+ }
+ (Datum::Date(value), DataType::Date(_)) => {
+ if legacy_partition_name {
+ Some(value.to_string())
+ } else {
+ Some(format_partition_date(*value))
+ }
+ }
+ (Datum::Time(value), DataType::Time(_)) => Some(value.to_string()),
+ _ => None,
+ }
+}
+
+fn format_partition_date(epoch_days: i32) -> String {
+ let date = NaiveDate::from_num_days_from_ce_opt(epoch_days + 719_163)
+ .unwrap_or(NaiveDate::from_ymd_opt(1970, 1, 1).unwrap());
+ date.format("%Y-%m-%d").to_string()
+}
+
+fn escape_path_name(path: &str) -> String {
+ let mut result = String::with_capacity(path.len());
+ for byte in path.bytes() {
+ if should_escape(byte) {
+ result.push('%');
+ result.push_str(&format!("{byte:02X}"));
+ } else {
+ result.push(byte as char);
+ }
+ }
+ result
+}
+
+fn should_escape(byte: u8) -> bool {
+ byte <= 0x1F
+ || byte >= 0x7F
+ || matches!(
+ byte,
+ b'"' | b'#'
+ | b'%'
+ | b'\''
+ | b'*'
+ | b'/'
+ | b':'
+ | b'='
+ | b'?'
+ | b'\\'
+ | b'\x7F'
+ | b'{'
+ | b'['
+ | b']'
+ | b'^'
+ )
+}
+
+fn partition_row_from_path(
+ table_path: &str,
+ file_parent: &str,
+ partition_fields: &[DataField],
+ partition_keys: &[String],
+ default_partition_name: &str,
+ only_value_in_path: bool,
+) -> crate::Result<Option<BinaryRow>> {
+ let relative = match file_parent
+ .trim_end_matches('/')
+ .strip_prefix(table_path.trim_end_matches('/'))
+ {
+ Some(path) => path.trim_start_matches('/'),
+ None => return Ok(None),
+ };
+ if relative.is_empty() {
+ return Ok(None);
+ }
+
+ let mut values = Vec::with_capacity(partition_keys.len());
+ if only_value_in_path {
+ for segment in relative
+ .split('/')
+ .filter(|segment| !segment.is_empty())
+ .take(partition_keys.len())
+ {
+ let Some(value) = unescape_path_name(segment) else {
+ return Ok(None);
+ };
+ values.push(value);
+ }
+ if values.len() != partition_keys.len() {
+ return Ok(None);
+ }
+ } else {
+ for key in partition_keys {
+ let Some(value) = relative
+ .split('/')
+ .find_map(|segment| partition_segment_value(segment, key))
+ else {
+ return Ok(None);
+ };
+ values.push(value);
+ }
+ }
+
+ let mut builder = BinaryRowBuilder::new(partition_fields.len() as i32);
+ for (idx, value) in values.iter().enumerate() {
+ if value == default_partition_name {
+ builder.set_null_at(idx);
+ continue;
+ }
+ let Some(datum) = parse_partition_datum(value,
partition_fields[idx].data_type()) else {
+ return Ok(None);
+ };
+ builder.write_datum(idx, &datum, partition_fields[idx].data_type());
+ }
+ Ok(Some(builder.build()))
+}
+
+fn partition_segment_value(segment: &str, key: &str) -> Option<String> {
+ let (segment_key, segment_value) = segment.split_once('=')?;
+ if unescape_path_name(segment_key)? == key {
+ unescape_path_name(segment_value)
+ } else {
+ None
+ }
+}
+
+fn unescape_path_name(value: &str) -> Option<String> {
+ let bytes = value.as_bytes();
+ let mut out = Vec::with_capacity(bytes.len());
+ let mut i = 0;
+ while i < bytes.len() {
+ if bytes[i] == b'%' {
+ if i + 2 >= bytes.len() {
+ return None;
+ }
+ let hi = hex_value(bytes[i + 1])?;
+ let lo = hex_value(bytes[i + 2])?;
+ out.push((hi << 4) | lo);
+ i += 3;
+ } else {
+ out.push(bytes[i]);
+ i += 1;
+ }
+ }
+ String::from_utf8(out).ok()
+}
+
+fn hex_value(byte: u8) -> Option<u8> {
+ match byte {
+ b'0'..=b'9' => Some(byte - b'0'),
+ b'a'..=b'f' => Some(byte - b'a' + 10),
+ b'A'..=b'F' => Some(byte - b'A' + 10),
+ _ => None,
+ }
+}
+
+fn parse_partition_datum(value: &str, data_type: &DataType) -> Option<Datum> {
+ match data_type {
+ DataType::Boolean(_) => value.parse::<bool>().ok().map(Datum::Bool),
+ DataType::TinyInt(_) => value.parse::<i8>().ok().map(Datum::TinyInt),
+ DataType::SmallInt(_) =>
value.parse::<i16>().ok().map(Datum::SmallInt),
+ DataType::Int(_) => value.parse::<i32>().ok().map(Datum::Int),
+ DataType::BigInt(_) => value.parse::<i64>().ok().map(Datum::Long),
+ DataType::Char(_) | DataType::VarChar(_) =>
Some(Datum::String(value.to_string())),
+ DataType::Date(_) => parse_partition_date(value).map(Datum::Date),
+ DataType::Time(_) => value.parse::<i32>().ok().map(Datum::Time),
+ _ => None,
+ }
+}
+
+fn parse_partition_date(value: &str) -> Option<i32> {
+ if let Ok(epoch_days) = value.parse::<i32>() {
+ return Some(epoch_days);
+ }
+ let date = NaiveDate::parse_from_str(value, "%Y-%m-%d").ok()?;
+ date.signed_duration_since(NaiveDate::from_ymd_opt(1970, 1, 1).unwrap())
+ .num_days()
+ .try_into()
+ .ok()
+}
+
+fn supported_format_table_extension(format: &str) -> crate::Result<&'static
str> {
+ match format.to_ascii_lowercase().as_str() {
+ "parquet" => Ok(".parquet"),
+ "orc" => Ok(".orc"),
+ "avro" => Ok(".avro"),
+ "row" => Ok(".row"),
+ "mosaic" => Ok(".mosaic"),
+ #[cfg(feature = "vortex")]
+ "vortex" => Ok(".vortex"),
+ other => Err(crate::Error::Unsupported {
+ message: format!(
+ "Format table file.format '{other}' is not supported by the
Rust reader yet"
+ ),
+ }),
+ }
+}
+
+fn data_file_meta(file_name: String, file_size: i64, schema_id: i64) ->
DataFileMeta {
+ DataFileMeta {
+ file_name,
+ file_size,
+ row_count: 0,
+ min_key: Vec::new(),
+ max_key: Vec::new(),
+ key_stats: BinaryTableStats::empty(),
+ value_stats: BinaryTableStats::empty(),
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id,
+ level: 0,
+ extra_files: Vec::new(),
+ creation_time: None,
+ delete_row_count: Some(0),
+ embedded_index: None,
+ file_source: None,
+ value_stats_cols: None,
+ external_path: None,
+ first_row_id: None,
+ write_cols: None,
+ }
+}
diff --git a/crates/paimon/src/table/format_write_builder.rs
b/crates/paimon/src/table/format_write_builder.rs
new file mode 100644
index 0000000..3608e82
--- /dev/null
+++ b/crates/paimon/src/table/format_write_builder.rs
@@ -0,0 +1,77 @@
+// 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.
+
+//! Write builder for Java-compatible `type=format-table` metadata.
+
+use super::write_builder::validate_commit_user;
+use super::{DataEvolutionDeleteWriter, Table, TableCommit, TableUpdate,
TableWrite};
+use uuid::Uuid;
+
+pub(crate) struct FormatWriteBuilder<'a> {
+ table: &'a Table,
+ commit_user: String,
+}
+
+impl<'a> FormatWriteBuilder<'a> {
+ pub(crate) fn new(table: &'a Table) -> Self {
+ Self {
+ table,
+ commit_user: Uuid::new_v4().to_string(),
+ }
+ }
+
+ pub(crate) fn commit_user(&self) -> &str {
+ &self.commit_user
+ }
+
+ pub(crate) fn with_commit_user(
+ mut self,
+ commit_user: impl Into<String>,
+ ) -> crate::Result<Self> {
+ let commit_user = commit_user.into();
+ validate_commit_user(&commit_user)?;
+ self.commit_user = commit_user;
+ Ok(self)
+ }
+
+ pub(crate) fn with_overwrite(self) -> Self {
+ self
+ }
+
+ pub(crate) fn new_commit(&self) -> TableCommit {
+ TableCommit::new(self.table.clone(), self.commit_user.clone())
+ }
+
+ pub(crate) fn new_write(&self) -> crate::Result<TableWrite> {
+ Err(crate::Error::Unsupported {
+ message: "Writing format tables is not supported by the Rust
client yet".to_string(),
+ })
+ }
+
+ pub(crate) fn new_update(&self, _update_columns: Vec<String>) ->
crate::Result<TableUpdate> {
+ Err(crate::Error::Unsupported {
+ message: "Updating format tables is not supported by the Rust
client yet".to_string(),
+ })
+ }
+
+ pub(crate) fn new_delete(&self) ->
crate::Result<DataEvolutionDeleteWriter> {
+ Err(crate::Error::Unsupported {
+ message: "Deleting from format tables is not supported by the Rust
client yet"
+ .to_string(),
+ })
+ }
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index e28e6f4..b0b6060 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -37,6 +37,10 @@ pub mod data_evolution_writer;
mod data_file_reader;
mod data_file_writer;
mod dedicated_format_file_writer;
+mod format_read_builder;
+mod format_table_read;
+mod format_table_scan;
+mod format_write_builder;
#[cfg(feature = "fulltext")]
mod full_text_search_builder;
pub(crate) mod global_index_build_common;
@@ -185,6 +189,10 @@ impl Table {
self.rest_env.as_ref()
}
+ pub(crate) fn is_format_table(&self) -> bool {
+ CoreOptions::new(self.schema.options()).is_format_table()
+ }
+
/// Create a read builder for scan/read.
///
/// Reference: [pypaimon
FileStoreTable.new_read_builder](https://github.com/apache/paimon/blob/release-1.3/paimon-python/pypaimon/table/file_store_table.py).
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index c3c2789..8bca5a7 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -21,6 +21,7 @@
//! and
[TypeUtils.project](https://github.com/apache/paimon/blob/master/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java).
use super::bucket_filter::{extract_predicate_for_keys,
split_partition_and_data_predicates};
+use super::format_read_builder::FormatReadBuilder;
use super::partition_filter::PartitionFilter;
use super::table_read::TableRead;
use super::{Table, TableScan};
@@ -108,7 +109,118 @@ fn normalize_filter(table: &Table, filter: Predicate) ->
NormalizedFilter {
/// Rust keeps a names-based projection API for ergonomics, while aligning the
/// resulting read semantics with Java Paimon's order-preserving projection.
#[derive(Debug, Clone)]
-pub struct ReadBuilder<'a> {
+pub struct ReadBuilder<'a>(ReadBuilderKind<'a>);
+
+#[derive(Debug, Clone)]
+enum ReadBuilderKind<'a> {
+ Paimon(PaimonReadBuilder<'a>),
+ Format(FormatReadBuilder<'a>),
+}
+
+impl<'a> ReadBuilder<'a> {
+ pub(crate) fn new(table: &'a Table) -> Self {
+ if table.is_format_table() {
+ Self(ReadBuilderKind::Format(FormatReadBuilder::new(table)))
+ } else {
+ Self(ReadBuilderKind::Paimon(PaimonReadBuilder::new(table)))
+ }
+ }
+
+ /// Set column projection by name. Output order follows the
caller-specified order.
+ /// Unknown or duplicate names cause this method to fail; an empty list is
a valid
+ /// zero-column projection.
+ pub fn with_projection(&mut self, columns: &[&str]) -> Result<&mut Self> {
+ match &mut self.0 {
+ ReadBuilderKind::Paimon(builder) => {
+ builder.with_projection(columns)?;
+ }
+ ReadBuilderKind::Format(builder) => {
+ builder.with_projection(columns)?;
+ }
+ }
+ Ok(self)
+ }
+
+ /// Set the full read type, including nested field pruning or
connector-defined
+ /// logical read types such as Variant extractions.
+ pub fn with_read_type(&mut self, read_type: Vec<DataField>) -> &mut Self {
+ match &mut self.0 {
+ ReadBuilderKind::Paimon(builder) => {
+ builder.with_read_type(read_type);
+ }
+ ReadBuilderKind::Format(builder) => {
+ builder.with_read_type(read_type);
+ }
+ }
+ self
+ }
+
+ /// Set a filter predicate for scan planning and conservative read pruning.
+ pub fn with_filter(&mut self, filter: Predicate) -> &mut Self {
+ match &mut self.0 {
+ ReadBuilderKind::Paimon(builder) => {
+ builder.with_filter(filter);
+ }
+ ReadBuilderKind::Format(builder) => {
+ builder.with_filter(filter);
+ }
+ }
+ self
+ }
+
+ /// Whether a translated predicate is exact at the table-provider boundary.
+ pub fn is_exact_filter_pushdown(&self, filter: &Predicate) -> bool {
+ match &self.0 {
+ ReadBuilderKind::Paimon(builder) =>
builder.is_exact_filter_pushdown(filter),
+ ReadBuilderKind::Format(builder) =>
builder.is_exact_filter_pushdown(filter),
+ }
+ }
+
+ /// Set row ID ranges `[from, to]` (inclusive) for filtering in data
evolution mode.
+ pub fn with_row_ranges(&mut self, ranges: Vec<RowRange>) -> &mut Self {
+ match &mut self.0 {
+ ReadBuilderKind::Paimon(builder) => {
+ builder.with_row_ranges(ranges);
+ }
+ ReadBuilderKind::Format(builder) => {
+ builder.with_row_ranges(ranges);
+ }
+ }
+ self
+ }
+
+ /// Push a row-limit hint down to scan planning.
+ pub fn with_limit(&mut self, limit: usize) -> &mut Self {
+ match &mut self.0 {
+ ReadBuilderKind::Paimon(builder) => {
+ builder.with_limit(limit);
+ }
+ ReadBuilderKind::Format(builder) => {
+ builder.with_limit(limit);
+ }
+ }
+ self
+ }
+
+ /// Create a table scan. Call [TableScan::plan] to get splits.
+ pub fn new_scan(&self) -> TableScan<'a> {
+ match &self.0 {
+ ReadBuilderKind::Paimon(builder) => builder.new_scan(),
+ ReadBuilderKind::Format(builder) => builder.new_scan(),
+ }
+ }
+
+ /// Create a table read for consuming splits (e.g. from a scan plan).
+ pub fn new_read(&self) -> Result<TableRead<'a>> {
+ match &self.0 {
+ ReadBuilderKind::Paimon(builder) => builder.new_read(),
+ ReadBuilderKind::Format(builder) => builder.new_read(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct PaimonReadBuilder<'a> {
table: &'a Table,
read_type: Option<Vec<DataField>>,
filter: NormalizedFilter,
@@ -116,7 +228,7 @@ pub struct ReadBuilder<'a> {
row_ranges: Option<Vec<RowRange>>,
}
-impl<'a> ReadBuilder<'a> {
+impl<'a> PaimonReadBuilder<'a> {
pub(crate) fn new(table: &'a Table) -> Self {
Self {
table,
@@ -336,7 +448,7 @@ pub(super) fn is_system_projection_field(field_id: i32) ->
bool {
#[cfg(test)]
mod tests {
- use super::ReadBuilder;
+ use super::{PaimonReadBuilder, ReadBuilder, ReadBuilderKind};
use crate::table::TableRead;
mod test_utils {
include!(concat!(env!("CARGO_MANIFEST_DIR"), "/../test_utils.rs"));
@@ -356,6 +468,13 @@ mod tests {
use tempfile::tempdir;
use test_utils::{local_file_path, test_data_file, write_int_parquet_file};
+ fn paimon_builder<'a, 'b>(builder: &'b ReadBuilder<'a>) -> &'b
PaimonReadBuilder<'a> {
+ match &builder.0 {
+ ReadBuilderKind::Paimon(inner) => inner,
+ ReadBuilderKind::Format(_) => panic!("expected Paimon read
builder"),
+ }
+ }
+
fn collect_int_column(batches: &[RecordBatch], column_name: &str) ->
Vec<i32> {
batches
.iter()
@@ -877,15 +996,16 @@ mod tests {
builder.with_filter(filter);
// _ROW_ID predicates should be extracted into row_ranges
- assert!(builder.row_ranges.is_some());
- let ranges = builder.row_ranges.as_ref().unwrap();
+ let inner = paimon_builder(&builder);
+ assert!(inner.row_ranges.is_some());
+ let ranges = inner.row_ranges.as_ref().unwrap();
assert_eq!(ranges.len(), 1);
assert_eq!(ranges[0].from(), 10);
assert_eq!(ranges[0].to(), 20);
// _ROW_ID predicates should be removed from data_predicates
- assert!(!builder.filter.data_predicates.is_empty());
- for p in &builder.filter.data_predicates {
+ assert!(!inner.filter.data_predicates.is_empty());
+ for p in &inner.filter.data_predicates {
if let Predicate::Leaf { column, .. } = p {
assert_ne!(column, crate::spec::ROW_ID_FIELD_NAME);
}
@@ -923,7 +1043,7 @@ mod tests {
builder.with_filter(filter);
// Explicit row_ranges should be preserved, not overwritten
- let ranges = builder.row_ranges.as_ref().unwrap();
+ let ranges = paimon_builder(&builder).row_ranges.as_ref().unwrap();
assert_eq!(ranges.len(), 1);
assert_eq!(ranges[0].from(), 0);
assert_eq!(ranges[0].to(), 5);
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index cb52f2b..b777100 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -23,7 +23,7 @@ use crate::catalog::{Identifier, RESTTokenFileIO};
use crate::common::Options;
use crate::error::Error;
use crate::io::FileIO;
-use crate::spec::TableSchema;
+use crate::spec::{CoreOptions, TableSchema, PATH_OPTION};
use crate::table::snapshot_commit::{RESTSnapshotCommit, SnapshotCommit};
use crate::table::Table;
use crate::Result;
@@ -105,6 +105,11 @@ impl RESTEnv {
source: None,
})?;
+ let table_path = response.path.ok_or_else(|| Error::DataInvalid {
+ message: format!("Table {} response missing path",
identifier.full_name()),
+ source: None,
+ })?;
+
let schema_id = response.schema_id.ok_or_else(|| Error::DataInvalid {
message: format!(
"Table {} response missing schema_id",
@@ -112,12 +117,13 @@ impl RESTEnv {
),
source: None,
})?;
- let table_schema = TableSchema::new(schema_id, &schema);
-
- let table_path = response.path.ok_or_else(|| Error::DataInvalid {
- message: format!("Table {} response missing path",
identifier.full_name()),
- source: None,
- })?;
+ let mut table_schema = TableSchema::new(schema_id, &schema);
+ if CoreOptions::new(table_schema.options()).is_format_table() {
+ table_schema =
table_schema.copy_with_options(std::collections::HashMap::from([(
+ PATH_OPTION.to_string(),
+ table_path.clone(),
+ )]));
+ }
let is_external = response.is_external.ok_or_else(||
Error::DataInvalid {
message: format!(
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 29e8adf..0b8e550 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -17,6 +17,7 @@
use super::data_evolution_reader::DataEvolutionReader;
use super::data_file_reader::DataFileReader;
+use super::format_table_read::FormatTableRead;
use super::kv_file_reader::{KeyValueFileReader, KeyValueReadConfig};
use super::read_builder::split_scan_predicates;
use super::{ArrowRecordBatchStream, Table};
@@ -27,13 +28,95 @@ use crate::DataSplit;
///
/// Reference:
[pypaimon.read.table_read.TableRead](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/read/table_read.py)
#[derive(Debug, Clone)]
-pub struct TableRead<'a> {
+pub struct TableRead<'a>(TableReadKind<'a>);
+
+#[derive(Debug, Clone)]
+enum TableReadKind<'a> {
+ Paimon(PaimonTableRead<'a>),
+ Format(FormatTableRead<'a>),
+}
+
+impl<'a> TableRead<'a> {
+ /// Create a new TableRead with a specific read type (projected fields).
+ pub fn new(
+ table: &'a Table,
+ read_type: Vec<DataField>,
+ data_predicates: Vec<Predicate>,
+ ) -> Self {
+ if table.is_format_table() {
+ Self::new_format(table, read_type, data_predicates, None)
+ } else {
+ Self(TableReadKind::Paimon(PaimonTableRead::new(
+ table,
+ read_type,
+ data_predicates,
+ )))
+ }
+ }
+
+ pub(crate) fn new_format(
+ table: &'a Table,
+ read_type: Vec<DataField>,
+ data_predicates: Vec<Predicate>,
+ limit: Option<usize>,
+ ) -> Self {
+ Self(TableReadKind::Format(FormatTableRead::new(
+ table,
+ read_type,
+ data_predicates,
+ limit,
+ )))
+ }
+
+ /// Schema (fields) that this read will produce.
+ pub fn read_type(&self) -> &[DataField] {
+ match &self.0 {
+ TableReadKind::Paimon(read) => read.read_type(),
+ TableReadKind::Format(read) => read.read_type(),
+ }
+ }
+
+ /// Data predicates for read-side pruning.
+ pub fn data_predicates(&self) -> &[Predicate] {
+ match &self.0 {
+ TableReadKind::Paimon(read) => read.data_predicates(),
+ TableReadKind::Format(read) => read.data_predicates(),
+ }
+ }
+
+ /// Table for this read.
+ pub fn table(&self) -> &Table {
+ match &self.0 {
+ TableReadKind::Paimon(read) => read.table(),
+ TableReadKind::Format(read) => read.table(),
+ }
+ }
+
+ /// Set a filter predicate.
+ pub fn with_filter(self, filter: Predicate) -> Self {
+ match self.0 {
+ TableReadKind::Paimon(read) =>
Self(TableReadKind::Paimon(read.with_filter(filter))),
+ TableReadKind::Format(read) =>
Self(TableReadKind::Format(read.with_filter(filter))),
+ }
+ }
+
+ /// Returns an [`ArrowRecordBatchStream`].
+ pub fn to_arrow(&self, data_splits: &[DataSplit]) ->
crate::Result<ArrowRecordBatchStream> {
+ match &self.0 {
+ TableReadKind::Paimon(read) => read.to_arrow(data_splits),
+ TableReadKind::Format(read) => read.to_arrow(data_splits),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct PaimonTableRead<'a> {
table: &'a Table,
read_type: Vec<DataField>,
data_predicates: Vec<Predicate>,
}
-impl<'a> TableRead<'a> {
+impl<'a> PaimonTableRead<'a> {
/// Create a new TableRead with a specific read type (projected fields).
pub fn new(
table: &'a Table,
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 08b225f..ac879ba 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -21,6 +21,7 @@
//! and
[FullStartingScanner](https://github.com/apache/paimon/blob/release-1.3/paimon-python/pypaimon/read/scanner/full_starting_scanner.py).
use super::bucket_filter::compute_target_buckets;
+use super::format_table_scan::FormatTableScan;
use super::kv_file_reader::retain_primary_key_conjuncts;
use super::partition_filter::PartitionFilter;
use super::stats_filter::{
@@ -616,11 +617,110 @@ async fn prune_data_evolution_group_by_read_fields(
.collect())
}
-/// TableScan for full table scan (no incremental, no predicate).
+#[derive(Debug, Clone)]
+pub struct TableScan<'a>(TableScanKind<'a>);
+
+#[derive(Debug, Clone)]
+enum TableScanKind<'a> {
+ Paimon(PaimonTableScan<'a>),
+ Format(FormatTableScan<'a>),
+}
+
+impl<'a> TableScan<'a> {
+ pub(crate) fn new(
+ table: &'a Table,
+ partition_filter: Option<PartitionFilter>,
+ data_predicates: Vec<Predicate>,
+ bucket_predicate: Option<Predicate>,
+ limit: Option<usize>,
+ row_ranges: Option<Vec<RowRange>>,
+ ) -> Self {
+ if table.is_format_table() {
+ Self(TableScanKind::Format(FormatTableScan::new(
+ table,
+ partition_filter,
+ limit,
+ )))
+ } else {
+ Self(TableScanKind::Paimon(PaimonTableScan::new(
+ table,
+ partition_filter,
+ data_predicates,
+ bucket_predicate,
+ limit,
+ row_ranges,
+ )))
+ }
+ }
+
+ pub fn with_scan_all_files(self) -> Self {
+ match self.0 {
+ TableScanKind::Paimon(scan) =>
Self(TableScanKind::Paimon(scan.with_scan_all_files())),
+ TableScanKind::Format(scan) => Self(TableScanKind::Format(scan)),
+ }
+ }
+
+ pub fn with_row_ranges(self, ranges: Vec<RowRange>) -> Self {
+ match self.0 {
+ TableScanKind::Paimon(scan) => {
+ Self(TableScanKind::Paimon(scan.with_row_ranges(ranges)))
+ }
+ TableScanKind::Format(scan) => Self(TableScanKind::Format(scan)),
+ }
+ }
+
+ pub(super) fn with_projected_read_field_ids(
+ self,
+ projected_read_field_ids: Option<HashSet<i32>>,
+ ) -> Self {
+ match self.0 {
+ TableScanKind::Paimon(scan) => Self(TableScanKind::Paimon(
+ scan.with_projected_read_field_ids(projected_read_field_ids),
+ )),
+ TableScanKind::Format(scan) => Self(TableScanKind::Format(scan)),
+ }
+ }
+
+ pub async fn plan(&self) -> crate::Result<Plan> {
+ match &self.0 {
+ TableScanKind::Paimon(scan) => scan.plan().await,
+ TableScanKind::Format(scan) => scan.plan().await,
+ }
+ }
+
+ pub async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> {
+ match &self.0 {
+ TableScanKind::Paimon(scan) => scan.plan_with_trace().await,
+ TableScanKind::Format(scan) => scan.plan_with_trace().await,
+ }
+ }
+
+ pub(crate) async fn plan_manifest_entries(
+ &self,
+ snapshot: &Snapshot,
+ ) -> crate::Result<Vec<ManifestEntry>> {
+ match &self.0 {
+ TableScanKind::Paimon(scan) =>
scan.plan_manifest_entries(snapshot).await,
+ TableScanKind::Format(_) => Err(crate::Error::Unsupported {
+ message: "Format tables do not have Paimon manifest
entries".to_string(),
+ }),
+ }
+ }
+
+ #[cfg(test)]
+ fn apply_limit_pushdown(&self, splits: Vec<DataSplit>) -> Vec<DataSplit> {
+ match &self.0 {
+ TableScanKind::Paimon(scan) => scan.apply_limit_pushdown(splits),
+ TableScanKind::Format(scan) => scan.apply_limit_pushdown(splits),
+ }
+ }
+}
+
+/// Paimon table scan: resolves snapshots, reads manifests, and builds data
splits.
///
/// Reference:
[pypaimon.read.table_scan.TableScan](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/read/table_scan.py)
#[derive(Debug, Clone)]
-pub struct TableScan<'a> {
+struct PaimonTableScan<'a> {
table: &'a Table,
partition_filter: Option<PartitionFilter>,
data_predicates: Vec<Predicate>,
@@ -636,7 +736,7 @@ pub struct TableScan<'a> {
projected_read_field_ids: Option<HashSet<i32>>,
}
-impl<'a> TableScan<'a> {
+impl<'a> PaimonTableScan<'a> {
pub(crate) fn new(
table: &'a Table,
partition_filter: Option<PartitionFilter>,
@@ -717,11 +817,11 @@ impl<'a> TableScan<'a> {
/// Plan the full scan and return metadata-pruning trace counters.
pub async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> {
self.ensure_query_auth_allowed()?;
- let data_evolution_read_field_ids = self.projected_read_field_ids()?;
let mut trace = ScanTrace {
limit: self.limit,
..Default::default()
};
+ let data_evolution_read_field_ids = self.projected_read_field_ids()?;
let snapshot = match self.resolve_snapshot().await? {
Some(snapshot) => snapshot,
None => return Ok((Plan::new(Vec::new()), trace)),
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index 9671e3b..ab86f4b 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -19,6 +19,7 @@
//!
//! Reference: [pypaimon
WriteBuilder](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/write/write_builder.py)
+use super::format_write_builder::FormatWriteBuilder;
use crate::table::{DataEvolutionDeleteWriter, Table, TableCommit, TableUpdate,
TableWrite};
use uuid::Uuid;
@@ -26,13 +27,94 @@ use uuid::Uuid;
///
/// Provides `new_write` and `new_commit` methods, with optional
/// `overwrite` support for partition-level overwrites.
-pub struct WriteBuilder<'a> {
+pub struct WriteBuilder<'a>(WriteBuilderKind<'a>);
+
+enum WriteBuilderKind<'a> {
+ Paimon(PaimonWriteBuilder<'a>),
+ Format(FormatWriteBuilder<'a>),
+}
+
+impl<'a> WriteBuilder<'a> {
+ pub fn new(table: &'a Table) -> Self {
+ if table.is_format_table() {
+ Self(WriteBuilderKind::Format(FormatWriteBuilder::new(table)))
+ } else {
+ Self(WriteBuilderKind::Paimon(PaimonWriteBuilder::new(table)))
+ }
+ }
+
+ /// Get the commit user shared by writers and committers created by this
builder.
+ pub fn commit_user(&self) -> &str {
+ match &self.0 {
+ WriteBuilderKind::Paimon(builder) => builder.commit_user(),
+ WriteBuilderKind::Format(builder) => builder.commit_user(),
+ }
+ }
+
+ /// Set the commit user shared by writers and committers created by this
builder.
+ pub fn with_commit_user(self, commit_user: impl Into<String>) ->
crate::Result<Self> {
+ match self.0 {
+ WriteBuilderKind::Paimon(builder) =>
Ok(Self(WriteBuilderKind::Paimon(
+ builder.with_commit_user(commit_user)?,
+ ))),
+ WriteBuilderKind::Format(builder) =>
Ok(Self(WriteBuilderKind::Format(
+ builder.with_commit_user(commit_user)?,
+ ))),
+ }
+ }
+
+ /// Mark writers created by this builder as overwrite-aware.
+ pub fn with_overwrite(self) -> Self {
+ match self.0 {
+ WriteBuilderKind::Paimon(builder) => {
+ Self(WriteBuilderKind::Paimon(builder.with_overwrite()))
+ }
+ WriteBuilderKind::Format(builder) => {
+ Self(WriteBuilderKind::Format(builder.with_overwrite()))
+ }
+ }
+ }
+
+ /// Create a new TableCommit for committing write results.
+ pub fn new_commit(&self) -> TableCommit {
+ match &self.0 {
+ WriteBuilderKind::Paimon(builder) => builder.new_commit(),
+ WriteBuilderKind::Format(builder) => builder.new_commit(),
+ }
+ }
+
+ /// Create a new TableWrite for writing Arrow data.
+ pub fn new_write(&self) -> crate::Result<TableWrite> {
+ match &self.0 {
+ WriteBuilderKind::Paimon(builder) => builder.new_write(),
+ WriteBuilderKind::Format(builder) => builder.new_write(),
+ }
+ }
+
+ /// Create a new TableUpdate for data-evolution row-id updates.
+ pub fn new_update(&self, update_columns: Vec<String>) ->
crate::Result<TableUpdate> {
+ match &self.0 {
+ WriteBuilderKind::Paimon(builder) =>
builder.new_update(update_columns),
+ WriteBuilderKind::Format(builder) =>
builder.new_update(update_columns),
+ }
+ }
+
+ /// Create a new writer for data-evolution row-id deletes.
+ pub fn new_delete(&self) -> crate::Result<DataEvolutionDeleteWriter> {
+ match &self.0 {
+ WriteBuilderKind::Paimon(builder) => builder.new_delete(),
+ WriteBuilderKind::Format(builder) => builder.new_delete(),
+ }
+ }
+}
+
+struct PaimonWriteBuilder<'a> {
table: &'a Table,
commit_user: String,
overwrite: bool,
}
-impl<'a> WriteBuilder<'a> {
+impl<'a> PaimonWriteBuilder<'a> {
pub fn new(table: &'a Table) -> Self {
Self {
table,
@@ -117,7 +199,7 @@ impl<'a> WriteBuilder<'a> {
}
}
-fn validate_commit_user(commit_user: &str) -> crate::Result<()> {
+pub(super) fn validate_commit_user(commit_user: &str) -> crate::Result<()> {
let is_invalid = commit_user.is_empty()
|| commit_user == "."
|| commit_user == ".."
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 0a2e20f..b8801b8 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -23,7 +23,7 @@
use std::collections::HashMap;
use std::sync::Arc;
-use arrow_array::{Array, BinaryArray, Int32Array, RecordBatch, StringArray};
+use arrow_array::{Array, BinaryArray, Int32Array, Int64Array, RecordBatch,
StringArray};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
use futures::TryStreamExt;
use paimon::api::ConfigResponse;
@@ -489,6 +489,251 @@ async fn
test_blob_view_prescan_filters_invalid_filtered_out_reference() {
);
}
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_rest_catalog_reads_format_table() {
+ use parquet::arrow::ArrowWriter;
+ use std::fs::File;
+
+ let tmp = tempfile::tempdir().unwrap();
+ let format_path = format!("file://{}", tmp.path().display());
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int64, false),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int64Array::from(vec![1, 2])),
+ Arc::new(StringArray::from(vec!["alice", "bob"])),
+ ],
+ )
+ .unwrap();
+ let file = File::create(tmp.path().join("part-0.parquet")).unwrap();
+ let mut writer = ArrowWriter::try_new(file, arrow_schema, None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .column("name", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .build()
+ .unwrap();
+ let identifier = Identifier::new("default", "format_users");
+ ctx.server
+ .add_table_with_schema("default", "format_users", schema,
&format_path);
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ assert_eq!(table.location(), format_path);
+ assert_eq!(table.schema().options().get("path"), Some(&format_path));
+ assert!(table.new_write_builder().new_write().is_err());
+
+ let read_builder = table.new_read_builder();
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ assert_eq!(plan.splits().len(), 1);
+
+ let read = read_builder.new_read().unwrap();
+ let batches = read
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+
+ let ids = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ let names = batches[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ assert_eq!(ids.values(), &[1, 2]);
+ assert_eq!(names.value(0), "alice");
+ assert_eq!(names.value(1), "bob");
+
+ let mut limited_builder = table.new_read_builder();
+ limited_builder.with_limit(1);
+ let limited_plan = limited_builder.new_scan().plan().await.unwrap();
+ let limited_read = limited_builder.new_read().unwrap();
+ let limited_batches = limited_read
+ .to_arrow(limited_plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(
+ limited_batches
+ .iter()
+ .map(|batch| batch.num_rows())
+ .sum::<usize>(),
+ 1
+ );
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_rest_catalog_prunes_format_table_partition_filter() {
+ use parquet::arrow::ArrowWriter;
+ use std::fs::{create_dir_all, File};
+
+ let tmp = tempfile::tempdir().unwrap();
+ let format_path = format!("file://{}", tmp.path().display());
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int64, false),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ]));
+ for (dt, id, name) in [("2024-01-01", 1_i64, "alice"), ("2024-01-02",
2_i64, "bob")] {
+ let dir = tmp.path().join(format!("dt={dt}"));
+ create_dir_all(&dir).unwrap();
+ let batch = RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int64Array::from(vec![id])),
+ Arc::new(StringArray::from(vec![name])),
+ ],
+ )
+ .unwrap();
+ let file = File::create(dir.join("part-0.parquet")).unwrap();
+ let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(),
None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ }
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = Schema::builder()
+ .column("dt", DataType::VarChar(VarCharType::new(32).unwrap()))
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .column("name", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .partition_keys(["dt"])
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .build()
+ .unwrap();
+ let identifier = Identifier::new("default", "format_partitioned_users");
+ ctx.server
+ .add_table_with_schema("default", "format_partitioned_users", schema,
&format_path);
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("dt", Datum::String("2024-01-02".to_string()))
+ .unwrap();
+ let mut read_builder = table.new_read_builder();
+ read_builder.with_filter(predicate);
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ assert_eq!(plan.splits().len(), 1);
+ assert!(plan.splits()[0].bucket_path().contains("dt=2024-01-02"));
+
+ let read = read_builder.new_read().unwrap();
+ let batches = read
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ let dts = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let ids = batches[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(dts.value(0), "2024-01-02");
+ assert_eq!(ids.value(0), 2);
+
+ let missing_predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("dt", Datum::String("2024-01-03".to_string()))
+ .unwrap();
+ let mut read_builder = table.new_read_builder();
+ read_builder.with_filter(missing_predicate);
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ assert!(plan.splits().is_empty());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_rest_catalog_reads_format_table_value_only_partition_path() {
+ use parquet::arrow::ArrowWriter;
+ use std::fs::{create_dir_all, File};
+
+ let tmp = tempfile::tempdir().unwrap();
+ let format_path = format!("file://{}", tmp.path().display());
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int64, false),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ]));
+ for (dt, id, name) in [("2024-01-01", 1_i64, "alice"), ("2024-01-02",
2_i64, "bob")] {
+ let dir = tmp.path().join(dt);
+ create_dir_all(&dir).unwrap();
+ let batch = RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int64Array::from(vec![id])),
+ Arc::new(StringArray::from(vec![name])),
+ ],
+ )
+ .unwrap();
+ let file = File::create(dir.join("part-0.parquet")).unwrap();
+ let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(),
None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ }
+
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = Schema::builder()
+ .column("dt", DataType::VarChar(VarCharType::new(32).unwrap()))
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .column("name", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .partition_keys(["dt"])
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .option("format-table.partition-path-only-value", "true")
+ .build()
+ .unwrap();
+ let identifier = Identifier::new("default",
"format_value_only_partitioned_users");
+ ctx.server.add_table_with_schema(
+ "default",
+ "format_value_only_partitioned_users",
+ schema,
+ &format_path,
+ );
+
+ let table = ctx.catalog.get_table(&identifier).await.unwrap();
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("dt", Datum::String("2024-01-02".to_string()))
+ .unwrap();
+ let mut read_builder = table.new_read_builder();
+ read_builder.with_filter(predicate);
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ assert_eq!(plan.splits().len(), 1);
+ assert!(plan.splits()[0].bucket_path().ends_with("/2024-01-02"));
+
+ let read = read_builder.new_read().unwrap();
+ let batches = read
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let dts = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ assert_eq!(dts.value(0), "2024-01-02");
+}
+
#[tokio::test]
async fn test_catalog_get_table_not_found() {
let ctx = setup_catalog(vec!["default"]).await;
