diff --git a/Cargo.lock b/Cargo.lock index 0efb91445..61d3be37f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4677,6 +4677,7 @@ dependencies = [ "http", "indexmap 2.14.0", "jiff", + "lexical-write-float", "libc", "libloading 0.9.0", "log", @@ -4765,7 +4766,6 @@ dependencies = [ "datafusion", "flate2", "futures", - "lexical-write-float", "log", "paimon", "paimon-ftindex-core", diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 3af2afc51..aa2829821 100644 --- a/bindings/python/python/pypaimon_rust/datafusion.pyi +++ b/bindings/python/python/pypaimon_rust/datafusion.pyi @@ -283,6 +283,7 @@ class Table: def branch(self) -> str: ... def location(self) -> str: ... def schema(self) -> TableSchema: ... + def new_system_read(self, name: str) -> SystemRead: ... def new_read_builder(self, options: Optional[Dict[str, str]] = None) -> ReadBuilder: ... def new_stream_scan(self) -> StreamTableScan: ... def new_full_text_search_builder(self) -> FullTextSearchBuilder: ... @@ -863,3 +864,13 @@ class ShardTableUpdator: """Require a finished read and exactly as many update rows; repeated calls reuse messages.""" ... def close(self) -> None: ... + + +class SystemRead: + def read_type(self) -> List[DataField]: ... + def with_projection(self, columns: List[str]) -> SystemRead: ... + def with_filter(self, predicate: Dict[str, Any]) -> SystemRead: ... + def with_limit(self, limit: int) -> SystemRead: ... + def read(self) -> pyarrow.RecordBatch: + """Materialize metadata once, then filter, project and limit the result.""" + ... diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs index edd492055..2d9699094 100644 --- a/bindings/python/src/context.rs +++ b/bindings/python/src/context.rs @@ -796,6 +796,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> let this = PyModule::new(py, "datafusion")?; this.add_class::()?; this.add_class::()?; + this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs index 9b92ae834..0b4508c82 100644 --- a/bindings/python/src/lib.rs +++ b/bindings/python/src/lib.rs @@ -33,6 +33,7 @@ mod predicate; mod read; mod schema; mod shard_update; +mod system_table; mod table; mod udf; mod update_assignment; diff --git a/bindings/python/src/system_table.rs b/bindings/python/src/system_table.rs new file mode 100644 index 000000000..41d51f4a9 --- /dev/null +++ b/bindings/python/src/system_table.rs @@ -0,0 +1,76 @@ +// 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. + +//! Thin bindings for core metadata reads. + +use crate::error::to_py_err; +use crate::schema::PyDataField; +use arrow::pyarrow::ToPyArrow; +use paimon::table::SystemRead; +use paimon_datafusion::runtime::runtime; +use pyo3::prelude::*; +use pyo3::types::PyDict; + +#[pyclass(name = "SystemRead", module = "pypaimon_rust.datafusion")] +pub struct PySystemRead { + pub(crate) inner: SystemRead, +} + +#[pymethods] +impl PySystemRead { + fn read_type(&self) -> Vec { + self.inner + .read_type() + .into_iter() + .map(PyDataField::new) + .collect() + } + + fn with_projection( + mut slf: PyRefMut<'_, Self>, + columns: Vec, + ) -> PyResult> { + let columns: Vec<_> = columns.iter().map(String::as_str).collect(); + slf.inner.with_projection(&columns).map_err(to_py_err)?; + Ok(slf) + } + + fn with_filter<'py>( + mut slf: PyRefMut<'py, Self>, + predicate: &Bound<'_, PyDict>, + ) -> PyResult> { + let predicate = crate::predicate::dict_to_predicate( + predicate, + slf.inner.full_row_type().fields(), + true, + )?; + slf.inner.with_filter(predicate).map_err(to_py_err)?; + Ok(slf) + } + + fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_, Self> { + slf.inner.with_limit(limit); + slf + } + + fn read(&self, py: Python<'_>) -> PyResult> { + let batch = py + .detach(|| runtime().block_on(self.inner.read())) + .map_err(to_py_err)?; + Ok(batch.to_pyarrow(py)?.unbind()) + } +} diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs index 14e4298f0..216423943 100644 --- a/bindings/python/src/table.rs +++ b/bindings/python/src/table.rs @@ -198,6 +198,13 @@ impl PyTable { } } + /// Read metadata through the core system reader. + fn new_system_read(&self, name: &str) -> PyResult { + Ok(crate::system_table::PySystemRead { + inner: self.inner.new_system_read(name).map_err(to_py_err)?, + }) + } + fn new_full_text_search_builder(&self) -> crate::full_text_search::PyFullTextSearchBuilder { crate::full_text_search::PyFullTextSearchBuilder::new(Arc::clone(&self.inner)) } diff --git a/crates/integrations/datafusion/Cargo.toml b/crates/integrations/datafusion/Cargo.toml index dfca7278e..b3b33a99c 100644 --- a/crates/integrations/datafusion/Cargo.toml +++ b/crates/integrations/datafusion/Cargo.toml @@ -45,7 +45,6 @@ futures = "0.3" serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { workspace = true, features = ["rt", "rt-multi-thread", "time", "fs"] } -lexical-write-float = "1.0.6" uuid = { version = "1", features = ["v4"] } [dev-dependencies] diff --git a/crates/integrations/datafusion/src/system_tables/consumers.rs b/crates/integrations/datafusion/src/system_tables/consumers.rs deleted file mode 100644 index 40c922cee..000000000 --- a/crates/integrations/datafusion/src/system_tables/consumers.rs +++ /dev/null @@ -1,190 +0,0 @@ -// 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. - -//! Mirrors Java [ConsumersTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/ConsumersTable.java). - -use std::collections::HashSet; -use std::sync::{Arc, OnceLock}; - -use async_trait::async_trait; -use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray}; -use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef}; -use datafusion::catalog::Session; -use datafusion::common::ScalarValue; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::Result as DFResult; -use datafusion::logical_expr::{Expr, Operator, TableProviderFilterPushDown}; -use datafusion::physical_plan::ExecutionPlan; -use paimon::table::Table; - -use crate::error::to_datafusion_error; - -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(ConsumersTable { table })) -} - -fn consumers_schema() -> SchemaRef { - static SCHEMA: OnceLock = OnceLock::new(); - SCHEMA - .get_or_init(|| { - Arc::new(Schema::new(vec![ - Field::new("consumer_id", DataType::Utf8, false), - Field::new("next_snapshot_id", DataType::Int64, false), - ])) - }) - .clone() -} - -#[derive(Debug)] -pub(super) struct ConsumersTable { - table: Table, -} - -#[async_trait] -impl TableProvider for ConsumersTable { - fn schema(&self) -> SchemaRef { - consumers_schema() - } - - fn table_type(&self) -> TableType { - TableType::View - } - - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - filters: &[Expr], - _limit: Option, - ) -> DFResult> { - let manager = self.table.consumer_manager(); - let requested_ids = requested_consumer_ids(filters); - let consumers = crate::runtime::await_with_runtime(async move { - match requested_ids { - Some(ids) => manager.list_by_ids(&ids).await, - None => manager.list_all().await, - } - }) - .await - .map_err(to_datafusion_error)?; - - let schema = consumers_schema(); - let batch = RecordBatch::try_new( - schema.clone(), - vec![ - Arc::new(StringArray::from_iter_values( - consumers.iter().map(|(id, _)| id.as_str()), - )), - Arc::new(Int64Array::from_iter_values( - consumers.iter().map(|(_, next_snapshot)| *next_snapshot), - )), - ], - )?; - - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema, - projection.cloned(), - )?) - } - - fn supports_filters_pushdown( - &self, - filters: &[&Expr], - ) -> DFResult> { - Ok(filters - .iter() - .map(|filter| { - if consumer_ids_from_filter(filter).is_some() { - TableProviderFilterPushDown::Inexact - } else { - TableProviderFilterPushDown::Unsupported - } - }) - .collect()) - } -} - -fn requested_consumer_ids(filters: &[Expr]) -> Option> { - let ids = filters - .iter() - .filter_map(consumer_ids_from_filter) - .reduce(|mut left, right| { - left.retain(|id| right.contains(id)); - left - })?; - let mut ids = ids.into_iter().collect::>(); - ids.sort_unstable(); - Some(ids) -} - -fn consumer_ids_from_filter(filter: &Expr) -> Option> { - match filter { - Expr::BinaryExpr(binary) if binary.op == Operator::Eq => { - consumer_id_literal(binary.left.as_ref(), binary.right.as_ref()) - .or_else(|| consumer_id_literal(binary.right.as_ref(), binary.left.as_ref())) - .map(|id| HashSet::from([id])) - } - Expr::BinaryExpr(binary) if binary.op == Operator::And => { - match ( - consumer_ids_from_filter(binary.left.as_ref()), - consumer_ids_from_filter(binary.right.as_ref()), - ) { - (Some(mut left), Some(right)) => { - left.retain(|id| right.contains(id)); - Some(left) - } - (Some(ids), None) | (None, Some(ids)) => Some(ids), - (None, None) => None, - } - } - Expr::BinaryExpr(binary) if binary.op == Operator::Or => { - let mut left = consumer_ids_from_filter(binary.left.as_ref())?; - left.extend(consumer_ids_from_filter(binary.right.as_ref())?); - Some(left) - } - Expr::InList(in_list) - if !in_list.negated && is_consumer_id_column(in_list.expr.as_ref()) => - { - in_list.list.iter().map(string_literal).collect() - } - _ => None, - } -} - -fn consumer_id_literal(column: &Expr, literal: &Expr) -> Option { - is_consumer_id_column(column) - .then_some(literal) - .and_then(string_literal) -} - -fn is_consumer_id_column(expr: &Expr) -> bool { - matches!(expr, Expr::Column(column) if column.name == "consumer_id") -} - -fn string_literal(expr: &Expr) -> Option { - match expr { - Expr::Literal( - ScalarValue::Utf8(Some(value)) - | ScalarValue::LargeUtf8(Some(value)) - | ScalarValue::Utf8View(Some(value)), - _, - ) => Some(value.clone()), - _ => None, - } -} diff --git a/crates/integrations/datafusion/src/system_tables/metadata.rs b/crates/integrations/datafusion/src/system_tables/metadata.rs new file mode 100644 index 000000000..6c635be9b --- /dev/null +++ b/crates/integrations/datafusion/src/system_tables/metadata.rs @@ -0,0 +1,112 @@ +// 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. + +//! DataFusion adaptation of core read-only system tables. + +use crate::error::to_datafusion_error; +use async_trait::async_trait; +use datafusion::arrow::datatypes::SchemaRef; +use datafusion::catalog::Session; +use datafusion::datasource::memory::MemorySourceConfig; +use datafusion::datasource::{TableProvider, TableType}; +use datafusion::error::Result; +use datafusion::logical_expr::{Expr, TableProviderFilterPushDown}; +use datafusion::physical_plan::ExecutionPlan; +use paimon::table::{SystemRead, Table}; +use std::sync::Arc; + +pub(super) fn build(table: Table, name: &str) -> Result> { + Ok(Arc::new(MetadataTableProvider { + read: table.new_system_read(name).map_err(to_datafusion_error)?, + })) +} + +#[derive(Debug)] +pub(super) struct MetadataTableProvider { + read: SystemRead, +} + +#[async_trait] +impl TableProvider for MetadataTableProvider { + fn schema(&self) -> SchemaRef { + self.read.schema() + } + fn table_type(&self) -> TableType { + TableType::View + } + + async fn scan( + &self, + _state: &dyn Session, + projection: Option<&Vec>, + filters: &[Expr], + _limit: Option, + ) -> Result> { + let mut read = self.read.clone(); + if let Some(indices) = projection { + let schema = self.read.schema(); + let columns: Vec<_> = indices + .iter() + .map(|&index| schema.field(index).name().as_str()) + .collect(); + read.with_projection(&columns) + .map_err(to_datafusion_error)?; + } + if let Some(predicate) = crate::filter_pushdown::analyze_filters( + filters, + self.read.full_row_type().fields(), + true, + ) + .pushed_predicate + { + read.with_filter(predicate).map_err(to_datafusion_error)?; + } + let schema = read.schema(); + let batch = crate::runtime::await_with_runtime(read.read()) + .await + .map_err(to_datafusion_error)?; + Ok(MemorySourceConfig::try_new_exec( + &[vec![batch]], + schema, + None, + )?) + } + + fn supports_filters_pushdown( + &self, + filters: &[&Expr], + ) -> Result> { + let fields = self.read.full_row_type(); + Ok(filters + .iter() + .map(|filter| { + if crate::filter_pushdown::analyze_filters( + &[(*filter).clone()], + fields.fields(), + true, + ) + .pushed_predicate + .is_some() + { + TableProviderFilterPushDown::Inexact + } else { + TableProviderFilterPushDown::Unsupported + } + }) + .collect()) + } +} diff --git a/crates/integrations/datafusion/src/system_tables/mod.rs b/crates/integrations/datafusion/src/system_tables/mod.rs index 0a206fdf7..6b7d36d2c 100644 --- a/crates/integrations/datafusion/src/system_tables/mod.rs +++ b/crates/integrations/datafusion/src/system_tables/mod.rs @@ -32,20 +32,10 @@ use crate::error::to_datafusion_error; mod aggregation_fields; mod audit_log; -mod branches; -mod buckets; -mod consumers; -mod files; -mod manifests; -mod options; +mod metadata; mod partitions; mod physical_files_size; mod referenced_files_size; -mod row_string_cast; -mod schemas; -mod snapshots; -mod table_indexes; -mod tags; type Builder = fn(Table) -> DFResult>; @@ -55,18 +45,23 @@ type Builder = fn(Table) -> DFResult>; const TABLES: &[(&str, Builder)] = &[ ("aggregation_fields", aggregation_fields::build), ("audit_log", audit_log::build), - ("branches", branches::build), - ("buckets", buckets::build), - ("consumers", consumers::build), - ("files", files::build), - ("manifests", manifests::build), - ("options", options::build), + ("branches", |table| metadata::build(table, "branches")), + ("buckets", |table| metadata::build(table, "buckets")), + ("consumers", |table| metadata::build(table, "consumers")), + ("files", |table| metadata::build(table, "files")), + ("file_key_ranges", |table| { + metadata::build(table, "file_key_ranges") + }), + ("manifests", |table| metadata::build(table, "manifests")), + ("options", |table| metadata::build(table, "options")), ("physical_files_size", physical_files_size::build), ("referenced_files_size", referenced_files_size::build), - ("schemas", schemas::build), - ("snapshots", snapshots::build), - ("table_indexes", table_indexes::build), - ("tags", tags::build), + ("schemas", |table| metadata::build(table, "schemas")), + ("snapshots", |table| metadata::build(table, "snapshots")), + ("table_indexes", |table| { + metadata::build(table, "table_indexes") + }), + ("tags", |table| metadata::build(table, "tags")), ]; const SYSTEM_TABLE_NAMES: &[&str] = &[ @@ -76,6 +71,7 @@ const SYSTEM_TABLE_NAMES: &[&str] = &[ "buckets", "consumers", "files", + "file_key_ranges", "manifests", "options", "partitions", @@ -113,20 +109,11 @@ pub(crate) fn is_registered(name: &str) -> bool { /// with a `$`. Keep in step with `TABLES`: one missing here goes back to /// having its time-travel clause silently dropped. pub(crate) fn is_system_table_provider(provider: &dyn TableProvider) -> bool { - provider.is::() - || provider.is::() - || provider.is::() - || provider.is::() - || provider.is::() - || provider.is::() - || provider.is::() + provider.is::() + || provider.is::() || provider.is::() || provider.is::() || provider.is::() - || provider.is::() - || provider.is::() - || provider.is::() - || provider.is::() } fn wrap_to_system_table(name: &str, base_table: Table) -> Option>> { @@ -267,20 +254,14 @@ mod tests { .unwrap(); let ctx = SessionContext::new(); - for (name, builder) in [ - ("manifests", super::manifests::build as super::Builder), - ( - "table_indexes", - super::table_indexes::build as super::Builder, - ), - ] { + for name in ["manifests", "table_indexes"] { for empty in [false, true] { let view = if empty { table.copy_with_pinned_snapshot(None) } else { table.clone() }; - let provider = builder(view).unwrap(); + let provider = super::metadata::build(view, name).unwrap(); let batches = ctx.read_table(provider).unwrap().collect().await.unwrap(); let count: usize = batches.iter().map(RecordBatch::num_rows).sum(); assert_eq!(count == 0, empty, "{name}, empty={empty}"); diff --git a/crates/integrations/datafusion/src/system_tables/options.rs b/crates/integrations/datafusion/src/system_tables/options.rs deleted file mode 100644 index 04d85f87f..000000000 --- a/crates/integrations/datafusion/src/system_tables/options.rs +++ /dev/null @@ -1,87 +0,0 @@ -// 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. - -//! Mirrors Java [OptionsTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/OptionsTable.java). - -use std::sync::{Arc, OnceLock}; - -use async_trait::async_trait; -use datafusion::arrow::array::{RecordBatch, StringArray}; -use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef}; -use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::Result as DFResult; -use datafusion::logical_expr::Expr; -use datafusion::physical_plan::ExecutionPlan; -use paimon::table::Table; - -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(OptionsTable { table })) -} - -fn options_schema() -> SchemaRef { - static SCHEMA: OnceLock = OnceLock::new(); - SCHEMA - .get_or_init(|| { - Arc::new(Schema::new(vec![ - Field::new("key", DataType::Utf8, false), - Field::new("value", DataType::Utf8, false), - ])) - }) - .clone() -} - -#[derive(Debug)] -pub(super) struct OptionsTable { - table: Table, -} - -#[async_trait] -impl TableProvider for OptionsTable { - fn schema(&self) -> SchemaRef { - options_schema() - } - - fn table_type(&self) -> TableType { - TableType::View - } - - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - _filters: &[Expr], - _limit: Option, - ) -> DFResult> { - // Java uses LinkedHashMap insertion order; HashMap has none — sort for stable output. - let mut entries: Vec<(&String, &String)> = self.table.schema().options().iter().collect(); - entries.sort_by(|a, b| a.0.cmp(b.0)); - - let keys = StringArray::from_iter_values(entries.iter().map(|(k, _)| k.as_str())); - let values = StringArray::from_iter_values(entries.iter().map(|(_, v)| v.as_str())); - - let schema = options_schema(); - let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(keys), Arc::new(values)])?; - - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema, - projection.cloned(), - )?) - } -} diff --git a/crates/integrations/datafusion/src/system_tables/schemas.rs b/crates/integrations/datafusion/src/system_tables/schemas.rs deleted file mode 100644 index 7575b3b02..000000000 --- a/crates/integrations/datafusion/src/system_tables/schemas.rs +++ /dev/null @@ -1,134 +0,0 @@ -// 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. - -//! Mirrors Java [SchemasTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java). - -use std::sync::{Arc, OnceLock}; - -use async_trait::async_trait; -use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray, TimestampMillisecondArray}; -use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit}; -use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::{DataFusionError, Result as DFResult}; -use datafusion::logical_expr::Expr; -use datafusion::physical_plan::ExecutionPlan; -use paimon::table::Table; -use serde::Serialize; - -use crate::error::to_datafusion_error; - -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(SchemasTable { table })) -} - -fn schemas_schema() -> SchemaRef { - static SCHEMA: OnceLock = OnceLock::new(); - SCHEMA - .get_or_init(|| { - Arc::new(Schema::new(vec![ - Field::new("schema_id", DataType::Int64, false), - Field::new("fields", DataType::Utf8, false), - Field::new("partition_keys", DataType::Utf8, false), - Field::new("primary_keys", DataType::Utf8, false), - Field::new("options", DataType::Utf8, false), - Field::new("comment", DataType::Utf8, true), - Field::new( - "update_time", - DataType::Timestamp(TimeUnit::Millisecond, None), - false, - ), - ])) - }) - .clone() -} - -#[derive(Debug)] -pub(super) struct SchemasTable { - table: Table, -} - -#[async_trait] -impl TableProvider for SchemasTable { - fn schema(&self) -> SchemaRef { - schemas_schema() - } - - fn table_type(&self) -> TableType { - TableType::View - } - - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - _filters: &[Expr], - _limit: Option, - ) -> DFResult> { - let table = self.table.clone(); - let schemas = - crate::runtime::await_with_runtime( - async move { table.schema_manager().list_all().await }, - ) - .await - .map_err(to_datafusion_error)?; - - let n = schemas.len(); - let mut schema_ids: Vec = Vec::with_capacity(n); - let mut fields_json: Vec = Vec::with_capacity(n); - let mut partition_keys_json: Vec = Vec::with_capacity(n); - let mut primary_keys_json: Vec = Vec::with_capacity(n); - let mut options_json: Vec = Vec::with_capacity(n); - let mut comments: Vec> = Vec::with_capacity(n); - let mut update_times: Vec = Vec::with_capacity(n); - - for schema in &schemas { - schema_ids.push(schema.id()); - fields_json.push(to_json(schema.fields())?); - partition_keys_json.push(to_json(schema.partition_keys())?); - primary_keys_json.push(to_json(schema.primary_keys())?); - options_json.push(to_json(schema.options())?); - comments.push(schema.comment().map(str::to_string)); - update_times.push(schema.time_millis()); - } - - let schema = schemas_schema(); - let batch = RecordBatch::try_new( - schema.clone(), - vec![ - Arc::new(Int64Array::from(schema_ids)), - Arc::new(StringArray::from(fields_json)), - Arc::new(StringArray::from(partition_keys_json)), - Arc::new(StringArray::from(primary_keys_json)), - Arc::new(StringArray::from(options_json)), - Arc::new(StringArray::from(comments)), - Arc::new(TimestampMillisecondArray::from(update_times)), - ], - )?; - - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema, - projection.cloned(), - )?) - } -} - -fn to_json(value: &T) -> DFResult { - serde_json::to_string(value).map_err(|e| DataFusionError::External(Box::new(e))) -} diff --git a/crates/integrations/datafusion/src/system_tables/snapshots.rs b/crates/integrations/datafusion/src/system_tables/snapshots.rs deleted file mode 100644 index 040c51c38..000000000 --- a/crates/integrations/datafusion/src/system_tables/snapshots.rs +++ /dev/null @@ -1,154 +0,0 @@ -// 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. - -//! Mirrors Java [SnapshotsTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java). - -use std::sync::{Arc, OnceLock}; - -use async_trait::async_trait; -use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray, TimestampMillisecondArray}; -use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit}; -use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::Result as DFResult; -use datafusion::logical_expr::Expr; -use datafusion::physical_plan::ExecutionPlan; -use paimon::table::Table; - -use crate::error::to_datafusion_error; - -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(SnapshotsTable { table })) -} - -fn snapshots_schema() -> SchemaRef { - static SCHEMA: OnceLock = OnceLock::new(); - SCHEMA - .get_or_init(|| { - Arc::new(Schema::new(vec![ - Field::new("snapshot_id", DataType::Int64, false), - Field::new("schema_id", DataType::Int64, false), - Field::new("commit_user", DataType::Utf8, false), - Field::new("commit_identifier", DataType::Int64, false), - Field::new("commit_kind", DataType::Utf8, false), - Field::new( - "commit_time", - DataType::Timestamp(TimeUnit::Millisecond, None), - false, - ), - Field::new("base_manifest_list", DataType::Utf8, false), - Field::new("delta_manifest_list", DataType::Utf8, false), - Field::new("changelog_manifest_list", DataType::Utf8, true), - Field::new("total_record_count", DataType::Int64, true), - Field::new("delta_record_count", DataType::Int64, true), - Field::new("changelog_record_count", DataType::Int64, true), - Field::new("watermark", DataType::Int64, true), - Field::new("next_row_id", DataType::Int64, true), - ])) - }) - .clone() -} - -#[derive(Debug)] -pub(super) struct SnapshotsTable { - table: Table, -} - -#[async_trait] -impl TableProvider for SnapshotsTable { - fn schema(&self) -> SchemaRef { - snapshots_schema() - } - - fn table_type(&self) -> TableType { - TableType::View - } - - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - _filters: &[Expr], - _limit: Option, - ) -> DFResult> { - let sm = self.table.snapshot_manager(); - let snapshots = crate::runtime::await_with_runtime(async move { sm.list_all().await }) - .await - .map_err(to_datafusion_error)?; - - let n = snapshots.len(); - let mut snapshot_ids = Vec::with_capacity(n); - let mut schema_ids = Vec::with_capacity(n); - let mut commit_users: Vec = Vec::with_capacity(n); - let mut commit_identifiers = Vec::with_capacity(n); - let mut commit_kinds: Vec = Vec::with_capacity(n); - let mut commit_times = Vec::with_capacity(n); - let mut base_manifest_lists: Vec = Vec::with_capacity(n); - let mut delta_manifest_lists: Vec = Vec::with_capacity(n); - let mut changelog_manifest_lists: Vec> = Vec::with_capacity(n); - let mut total_record_counts: Vec> = Vec::with_capacity(n); - let mut delta_record_counts: Vec> = Vec::with_capacity(n); - let mut changelog_record_counts: Vec> = Vec::with_capacity(n); - let mut watermarks: Vec> = Vec::with_capacity(n); - let mut next_row_ids: Vec> = Vec::with_capacity(n); - - for snap in &snapshots { - snapshot_ids.push(snap.id()); - schema_ids.push(snap.schema_id()); - commit_users.push(snap.commit_user().to_string()); - commit_identifiers.push(snap.commit_identifier()); - commit_kinds.push(snap.commit_kind().to_string()); - commit_times.push(snap.time_millis() as i64); - base_manifest_lists.push(snap.base_manifest_list().to_string()); - delta_manifest_lists.push(snap.delta_manifest_list().to_string()); - changelog_manifest_lists.push(snap.changelog_manifest_list().map(str::to_string)); - total_record_counts.push(snap.total_record_count()); - delta_record_counts.push(snap.delta_record_count()); - changelog_record_counts.push(snap.changelog_record_count()); - watermarks.push(snap.watermark()); - next_row_ids.push(snap.next_row_id()); - } - - let schema = snapshots_schema(); - let batch = RecordBatch::try_new( - schema.clone(), - vec![ - Arc::new(Int64Array::from(snapshot_ids)), - Arc::new(Int64Array::from(schema_ids)), - Arc::new(StringArray::from(commit_users)), - Arc::new(Int64Array::from(commit_identifiers)), - Arc::new(StringArray::from(commit_kinds)), - Arc::new(TimestampMillisecondArray::from(commit_times)), - Arc::new(StringArray::from(base_manifest_lists)), - Arc::new(StringArray::from(delta_manifest_lists)), - Arc::new(StringArray::from(changelog_manifest_lists)), - Arc::new(Int64Array::from(total_record_counts)), - Arc::new(Int64Array::from(delta_record_counts)), - Arc::new(Int64Array::from(changelog_record_counts)), - Arc::new(Int64Array::from(watermarks)), - Arc::new(Int64Array::from(next_row_ids)), - ], - )?; - - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema, - projection.cloned(), - )?) - } -} diff --git a/crates/integrations/datafusion/tests/system_tables.rs b/crates/integrations/datafusion/tests/system_tables.rs index 73dc6ba36..48f299bcf 100644 --- a/crates/integrations/datafusion/tests/system_tables.rs +++ b/crates/integrations/datafusion/tests/system_tables.rs @@ -75,6 +75,14 @@ async fn run_sql(ctx: &SQLContext, sql: &str) -> Vec { .unwrap_or_else(|e| panic!("Failed to execute `{sql}`: {e}")) } +async fn read_metadata( + table: paimon::table::Table, + name: &str, +) -> paimon::Result { + let batch = table.new_system_read(name)?.read().await?; + Ok(Box::pin(futures::stream::iter([Ok(batch)]))) +} + // Error text of `sql`, whether it surfaces at planning or execution. async fn query_error(ctx: &SQLContext, sql: &str) -> String { match ctx.sql(sql).await { @@ -793,6 +801,7 @@ async fn test_snapshots_system_table() { ("changelog_record_count", DataType::Int64), ("watermark", DataType::Int64), ("next_row_id", DataType::Int64), + ("writer_version", DataType::Utf8), ]; for (i, (name, dtype)) in expected_columns.iter().enumerate() { let field = arrow_schema.field(i); @@ -1329,7 +1338,7 @@ async fn test_files_system_table() { "creation_time", DataType::Timestamp(TimeUnit::Millisecond, None), ), - ("delete_row_count", DataType::Int64), + ("deleteRowCount", DataType::Int64), ("file_source", DataType::Utf8), ("first_row_id", DataType::Int64), ( @@ -1729,3 +1738,157 @@ async fn test_buckets_system_table() { "$buckets must aggregate $files by (partition, bucket)" ); } + +#[tokio::test] +async fn test_file_key_ranges_matches_file_metadata() { + use futures::TryStreamExt; + let (ctx, catalog, _tmp) = create_context().await; + let table = catalog + .get_table(&Identifier::new("default", FIXTURE_TABLE)) + .await + .unwrap(); + let expected = run_sql(&ctx, &format!("SELECT partition, bucket, file_path, file_format, schema_id, level, record_count, file_size_in_bytes, min_key, max_key, first_row_id FROM paimon.default.{FIXTURE_TABLE}$files")).await; + let actual: Vec<_> = read_metadata(table.clone(), "file_key_ranges") + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(actual, expected); + assert_eq!(actual[0].num_columns(), 11); + assert!(actual[0].num_rows() > 0); + let projected = run_sql(&ctx, &format!("SELECT file_path FROM paimon.default.{FIXTURE_TABLE}$file_key_ranges WHERE record_count > 0 LIMIT 1")).await; + assert_eq!(projected[0].num_rows(), 1); + assert_eq!(projected[0].num_columns(), 1); + let empty = table.copy_with_pinned_snapshot(None); + let batches: Vec<_> = read_metadata(empty, "file_key_ranges") + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 0); +} + +#[tokio::test] +async fn test_resolved_system_table_reader_preserves_guards_and_schema() { + use futures::TryStreamExt; + let (_ctx, catalog, _tmp) = create_context().await; + let table = catalog + .get_table(&Identifier::new("default", FIXTURE_TABLE)) + .await + .unwrap(); + for name in [ + "snapshots", + "schemas", + "options", + "manifests", + "files", + "buckets", + "tags", + "branches", + "consumers", + "file_key_ranges", + "table_indexes", + ] { + let stream = read_metadata(table.clone(), name).await.unwrap(); + let schema = table.new_system_read(name).unwrap().schema(); + let batches: Vec<_> = stream.try_collect().await.unwrap(); + assert!(!schema.fields().is_empty(), "{name}"); + assert!( + batches.iter().all(|batch| batch.schema() == schema), + "{name}" + ); + } + for name in ["partitions", "unknown"] { + let error = match read_metadata(table.clone(), name).await { + Ok(_) => panic!("{name} must require a catalog or be unknown"), + Err(error) => error.to_string(), + }; + assert!(error.contains("requires a catalog or is unknown")); + } + let denied = table.copy_with_options(std::collections::HashMap::from([( + "query-auth.enabled".to_string(), + "true".to_string(), + )])); + let error = match read_metadata(denied, "options").await { + Ok(_) => panic!("Authorization must be checked before reading metadata"), + Err(error) => error.to_string(), + }; + assert!(error.contains("query-auth.enabled")); +} + +#[tokio::test] +async fn test_options_ignore_resolved_dynamic_overrides() { + use futures::TryStreamExt; + let (_ctx, catalog, _tmp) = create_context().await; + let table = catalog + .get_table(&Identifier::new("default", FIXTURE_TABLE)) + .await + .unwrap(); + let copy = table.copy_with_options(std::collections::HashMap::from([ + ("file.format".to_string(), "csv".to_string()), + ("metadata.runtime-only".to_string(), "private".to_string()), + ])); + let batches: Vec<_> = read_metadata(copy, "options") + .await + .unwrap() + .try_collect() + .await + .unwrap(); + let rows: std::collections::HashMap<_, _> = batches + .iter() + .flat_map(|batch| { + let keys = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .map(|index| { + ( + keys.value(index).to_string(), + values.value(index).to_string(), + ) + }) + .collect::>() + }) + .collect(); + assert_eq!(&rows, table.schema().options()); + assert!(!rows.contains_key("metadata.runtime-only")); +} + +#[tokio::test] +async fn test_core_system_read_filters_before_projection() { + use paimon::spec::{Datum, Predicate, PredicateOperator}; + let (_ctx, catalog, _tmp) = create_context().await; + let table = catalog + .get_table(&Identifier::new("default", FIXTURE_TABLE)) + .await + .unwrap(); + let mut read = table.new_system_read("snapshots").unwrap(); + let field = read.read_type()[0].clone(); + read.with_projection(&["commit_user"]).unwrap(); + read.with_filter(Predicate::Leaf { + column: field.name().to_string(), + index: 0, + data_type: field.data_type().clone(), + op: PredicateOperator::Gt, + literals: vec![Datum::Long(0)], + }) + .unwrap(); + read.with_limit(1); + let rows = read.read().await.unwrap(); + assert_eq!(rows.num_rows(), 1); + assert_eq!(rows.schema().fields().len(), 1); + assert_eq!(rows.schema().field(0).name(), "commit_user"); + read.with_limit(0); + let rows = read.read().await.unwrap(); + assert_eq!(rows.num_rows(), 0); + assert_eq!(rows.schema(), read.schema()); +} diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml index f9aaffa83..31302b4a8 100644 --- a/crates/paimon/Cargo.toml +++ b/crates/paimon/Cargo.toml @@ -67,6 +67,7 @@ storage-jindo = ["storage-oss"] storage-oss-cpp = ["storage-oss"] [dependencies] +lexical-write-float = "1.0.6" rand = "0.8.5" url = "2.5.2" # Already in the tree via `url`; direct dep for `RESTUtil::decode_string`. diff --git a/crates/paimon/src/common/date_time.rs b/crates/paimon/src/common/date_time.rs index d9f800b12..ecb9fd530 100644 --- a/crates/paimon/src/common/date_time.rs +++ b/crates/paimon/src/common/date_time.rs @@ -83,7 +83,7 @@ pub(crate) fn format_epoch_timestamp(seconds: i64, nanos: u32, precision: u32) - } /// Java Timestamp.toString uses LocalDateTime's ISO form, not a SQL CAST form. -pub(crate) fn format_iso_timestamp(seconds: i64, nanos: u32) -> String { +pub fn format_iso_timestamp(seconds: i64, nanos: u32) -> String { let (year, month, day) = epoch_day_to_date(i64::from(seconds.div_euclid(86_400) as i32)); let year = if (0..=9999).contains(&year) { format!("{year:04}") diff --git a/crates/paimon/src/common/mod.rs b/crates/paimon/src/common/mod.rs index 543fb6fa9..d29489f24 100644 --- a/crates/paimon/src/common/mod.rs +++ b/crates/paimon/src/common/mod.rs @@ -18,6 +18,7 @@ //! Common utilities for Paimon. pub(crate) mod date_time; +pub use date_time::format_iso_timestamp; pub(crate) mod number; pub mod options; diff --git a/crates/paimon/src/spec/binary_row.rs b/crates/paimon/src/spec/binary_row.rs index 15b81289f..98cf25a1d 100644 --- a/crates/paimon/src/spec/binary_row.rs +++ b/crates/paimon/src/spec/binary_row.rs @@ -383,6 +383,16 @@ impl BinaryRow { Ok(Some(datum)) } + /// Decode a scalar field as a single-element Arrow array. + /// Uses the same typed decoding as partition expansion in table reads. + pub fn field_to_arrow( + &self, + index: usize, + data_type: &DataType, + ) -> crate::Result { + crate::arrow::partition::partition_array(self, index, data_type, 1) + } + /// Build a BinaryRow from selected columns of an Arrow RecordBatch at a given row. /// /// `field_indices` maps each position in the output BinaryRow to a column index diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 69b13bbe3..8970213b1 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -99,6 +99,33 @@ impl TableSchema { &self.fields } + /// Fields physically present in a file, following Java TableSchema.dataFileSchema. + /// Explicit write columns retain their order and omit metadata columns. + pub fn data_file_fields(&self, write_cols: Option<&[String]>) -> Vec { + if let Some(columns) = write_cols { + return columns + .iter() + .filter_map(|name| self.fields.iter().find(|f| f.name() == name).cloned()) + .collect(); + } + let options = CoreOptions::new(&self.options); + if !options.data_evolution_enabled() + || !options.data_evolution_write_cols_optimization_enabled() + { + return self.fields.clone(); + } + let inline = options.blob_inline_fields(); + self.fields + .iter() + .filter(|field| { + !(field.data_type().is_blob_file_field() && !inline.contains(field.name()) + || options.vector_file_format().is_some() + && matches!(field.data_type(), DataType::Vector(_))) + }) + .cloned() + .collect() + } + /// Logical fields followed by the metadata available to table reads. /// Row tracking provides ROW_ID and SEQUENCE_NUMBER; primary-key files /// provide SEQUENCE_NUMBER independently of row tracking. Keeping this diff --git a/crates/paimon/src/spec/snapshot.rs b/crates/paimon/src/spec/snapshot.rs index 70b366dbc..1af888b02 100644 --- a/crates/paimon/src/spec/snapshot.rs +++ b/crates/paimon/src/spec/snapshot.rs @@ -119,6 +119,10 @@ pub struct Snapshot { #[builder(default = None)] #[serde(skip_serializing_if = "Option::is_none")] next_row_id: Option, + /// Engine version that produced this snapshot, absent in older metadata. + #[builder(default = None)] + #[serde(default, skip_serializing_if = "Option::is_none")] + writer_version: Option, } impl Snapshot { @@ -240,6 +244,11 @@ impl Snapshot { self.next_row_id } + /// Get the writer version, when recorded by the producer. + pub fn writer_version(&self) -> Option<&str> { + self.writer_version.as_deref() + } + /// Get the commit kind of this snapshot. #[inline] pub fn commit_kind(&self) -> &CommitKind { diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 0608b3316..abc961122 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -129,6 +129,9 @@ mod postpone_fixed_bucket_write_builder; mod postpone_retract; mod prepared_files; mod query_auth; +mod system_table; +pub use system_table::SystemRead; + mod read_builder; mod read_limit; pub mod referenced_files; @@ -544,6 +547,11 @@ impl Table { && options.partitioned_table_in_metastore() } + /// Read a small system table using this table's resolved environment. + pub fn new_system_read(&self, name: &str) -> Result { + SystemRead::new(self.clone(), name) + } + /// 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/integrations/datafusion/src/system_tables/branches.rs b/crates/paimon/src/table/system_table/branches.rs similarity index 50% rename from crates/integrations/datafusion/src/system_tables/branches.rs rename to crates/paimon/src/table/system_table/branches.rs index c925def71..3cba2ec02 100644 --- a/crates/integrations/datafusion/src/system_tables/branches.rs +++ b/crates/paimon/src/table/system_table/branches.rs @@ -19,24 +19,13 @@ use std::sync::{Arc, OnceLock}; -use async_trait::async_trait; -use datafusion::arrow::array::{RecordBatch, StringArray, TimestampMillisecondArray}; -use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit}; -use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::Result as DFResult; -use datafusion::logical_expr::Expr; -use datafusion::physical_plan::ExecutionPlan; -use paimon::table::{BranchManager, Table}; +use crate::table::{BranchManager, Table}; +use arrow_array::{RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; -use crate::error::to_datafusion_error; +use crate::Result; -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(BranchesTable { table })) -} - -fn branches_schema() -> SchemaRef { +pub(super) fn branches_schema() -> SchemaRef { static SCHEMA: OnceLock = OnceLock::new(); SCHEMA .get_or_init(|| { @@ -52,52 +41,26 @@ fn branches_schema() -> SchemaRef { .clone() } -#[derive(Debug)] -pub(super) struct BranchesTable { - table: Table, -} +pub(super) async fn read( + table: &Table, + _projection: Option<&[usize]>, + _predicate: Option<&crate::spec::Predicate>, +) -> Result { + let (names, create_times) = collect_branches(table).await?; -#[async_trait] -impl TableProvider for BranchesTable { - fn schema(&self) -> SchemaRef { - branches_schema() - } + let schema = branches_schema(); + let batch = super::record_batch( + schema.clone(), + vec![ + Arc::new(StringArray::from(names)), + super::local_timestamp_column(create_times)?, + ], + )?; - fn table_type(&self) -> TableType { - TableType::View - } - - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - _filters: &[Expr], - _limit: Option, - ) -> DFResult> { - let table = self.table.clone(); - let (names, create_times) = - crate::runtime::await_with_runtime(async move { collect_branches(&table).await }) - .await - .map_err(to_datafusion_error)?; - - let schema = branches_schema(); - let batch = RecordBatch::try_new( - schema.clone(), - vec![ - Arc::new(StringArray::from(names)), - Arc::new(TimestampMillisecondArray::from(create_times)), - ], - )?; - - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema, - projection.cloned(), - )?) - } + Ok(batch) } -async fn collect_branches(table: &Table) -> paimon::Result<(Vec, Vec)> { +async fn collect_branches(table: &Table) -> crate::Result<(Vec, Vec)> { let file_io = table.file_io(); let bm = BranchManager::new(file_io.clone(), table.location().to_string()); let names = bm.list_all().await?; diff --git a/crates/integrations/datafusion/src/system_tables/buckets.rs b/crates/paimon/src/table/system_table/buckets.rs similarity index 78% rename from crates/integrations/datafusion/src/system_tables/buckets.rs rename to crates/paimon/src/table/system_table/buckets.rs index f34b6a803..82fb1af16 100644 --- a/crates/integrations/datafusion/src/system_tables/buckets.rs +++ b/crates/paimon/src/table/system_table/buckets.rs @@ -20,28 +20,16 @@ use std::collections::BTreeMap; use std::sync::{Arc, OnceLock}; -use async_trait::async_trait; -use datafusion::arrow::array::{ - Int32Array, Int64Array, RecordBatch, StringArray, TimestampMillisecondArray, -}; -use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, Schema, SchemaRef, TimeUnit}; -use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::Result as DFResult; -use datafusion::logical_expr::Expr; -use datafusion::physical_plan::ExecutionPlan; -use paimon::spec::{BinaryRow, DataField}; -use paimon::table::{DataSplit, Table}; +use crate::spec::{BinaryRow, DataField}; +use crate::table::{DataSplit, Table}; +use arrow_array::{Int32Array, Int64Array, RecordBatch, StringArray, TimestampMillisecondArray}; +use arrow_schema::{DataType as ArrowDataType, Field, Schema, SchemaRef, TimeUnit}; use super::row_string_cast::format_row_as_java_cast_string; -use crate::error::to_datafusion_error; -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(BucketsTable { table })) -} +use crate::Result; -fn buckets_schema() -> SchemaRef { +pub(super) fn buckets_schema() -> SchemaRef { static SCHEMA: OnceLock = OnceLock::new(); SCHEMA .get_or_init(|| { @@ -61,40 +49,14 @@ fn buckets_schema() -> SchemaRef { .clone() } -#[derive(Debug)] -pub(super) struct BucketsTable { - table: Table, -} - -#[async_trait] -impl TableProvider for BucketsTable { - fn schema(&self) -> SchemaRef { - buckets_schema() - } - - fn table_type(&self) -> TableType { - TableType::View - } - - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - _filters: &[Expr], - _limit: Option, - ) -> DFResult> { - let table = self.table.clone(); - let rows = - crate::runtime::await_with_runtime(async move { collect_bucket_rows(&table).await }) - .await - .map_err(to_datafusion_error)?; - let batch = bucket_rows_to_record_batch(&rows)?; - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - buckets_schema(), - projection.cloned(), - )?) - } +pub(super) async fn read( + table: &Table, + _projection: Option<&[usize]>, + _predicate: Option<&crate::spec::Predicate>, +) -> Result { + let rows = collect_bucket_rows(table).await?; + let batch = bucket_rows_to_record_batch(&rows)?; + Ok(batch) } /// A bucket's files aggregated: mirrors Java `BucketEntry`. @@ -112,7 +74,7 @@ struct BucketRow { agg: BucketAgg, } -async fn collect_bucket_rows(table: &Table) -> paimon::Result> { +async fn collect_bucket_rows(table: &Table) -> crate::Result> { let scan = table .new_read_builder() .new_scan() @@ -137,7 +99,7 @@ async fn collect_bucket_rows(table: &Table) -> paimon::Result> { fn aggregate_bucket_rows( splits: &[DataSplit], partition_fields: &[DataField], -) -> paimon::Result> { +) -> crate::Result> { let mut aggs: BTreeMap<(Option, i32, Vec), BucketAgg> = BTreeMap::new(); for split in splits { let partition = format_partition(split.partition(), partition_fields)?; @@ -166,7 +128,7 @@ fn aggregate_bucket_rows( .collect()) } -fn bucket_rows_to_record_batch(rows: &[BucketRow]) -> DFResult { +fn bucket_rows_to_record_batch(rows: &[BucketRow]) -> Result { let n = rows.len(); let mut partitions = Vec::with_capacity(n); let mut buckets = Vec::with_capacity(n); @@ -182,7 +144,7 @@ fn bucket_rows_to_record_batch(rows: &[BucketRow]) -> DFResult { file_counts.push(row.agg.file_count); last_update_times.push(row.agg.last_update_time); } - Ok(RecordBatch::try_new( + super::record_batch( buckets_schema(), vec![ Arc::new(StringArray::from(partitions)), @@ -192,7 +154,7 @@ fn bucket_rows_to_record_batch(rows: &[BucketRow]) -> DFResult { Arc::new(Int64Array::from(file_counts)), Arc::new(TimestampMillisecondArray::from(last_update_times)), ], - )?) + ) } /// Format `partition` as Java's cast-to-string, matching `$files`; `{}` when the @@ -200,7 +162,7 @@ fn bucket_rows_to_record_batch(rows: &[BucketRow]) -> DFResult { fn format_partition( partition: &BinaryRow, partition_fields: &[DataField], -) -> paimon::Result> { +) -> crate::Result> { if partition_fields.is_empty() { return Ok(Some("{}".to_string())); } @@ -210,7 +172,7 @@ fn format_partition( #[cfg(test)] mod tests { use super::*; - use paimon::spec::{DataType, Datum, VarCharType}; + use crate::spec::{DataType, Datum, VarCharType}; fn string_type() -> DataType { DataType::VarChar(VarCharType::new(64).unwrap()) diff --git a/crates/paimon/src/table/system_table/consumers.rs b/crates/paimon/src/table/system_table/consumers.rs new file mode 100644 index 000000000..373d0dc4c --- /dev/null +++ b/crates/paimon/src/table/system_table/consumers.rs @@ -0,0 +1,105 @@ +// 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. + +//! Mirrors Java [ConsumersTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/ConsumersTable.java). + +use std::collections::HashSet; +use std::sync::{Arc, OnceLock}; + +use crate::table::Table; +use arrow_array::{Int64Array, RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema, SchemaRef}; + +use crate::Result; + +pub(super) fn consumers_schema() -> SchemaRef { + static SCHEMA: OnceLock = OnceLock::new(); + SCHEMA + .get_or_init(|| { + Arc::new(Schema::new(vec![ + Field::new("consumer_id", DataType::Utf8, false), + Field::new("next_snapshot_id", DataType::Int64, false), + ])) + }) + .clone() +} + +pub(super) async fn read( + table: &Table, + _projection: Option<&[usize]>, + predicate: Option<&crate::spec::Predicate>, +) -> Result { + let manager = table.consumer_manager(); + let consumers = match predicate.and_then(consumer_ids) { + Some(ids) => { + manager + .list_by_ids(&ids.into_iter().collect::>()) + .await? + } + None => manager.list_all().await?, + }; + + let schema = consumers_schema(); + let batch = super::record_batch( + schema.clone(), + vec![ + Arc::new(StringArray::from_iter_values( + consumers.iter().map(|(id, _)| id.as_str()), + )), + Arc::new(Int64Array::from_iter_values( + consumers.iter().map(|(_, next_snapshot)| *next_snapshot), + )), + ], + )?; + + Ok(batch) +} + +fn consumer_ids(predicate: &crate::spec::Predicate) -> Option> { + use crate::spec::{Datum, Predicate, PredicateOperator}; + match predicate { + Predicate::Leaf { + column, + op, + literals, + .. + } if column == "consumer_id" + && matches!(op, PredicateOperator::Eq | PredicateOperator::In) => + { + literals + .iter() + .map(|value| { + if let Datum::String(value) = value { + Some(value.clone()) + } else { + None + } + }) + .collect() + } + Predicate::And(children) => children.iter().filter_map(consumer_ids).reduce(|mut a, b| { + a.retain(|id| b.contains(id)); + a + }), + Predicate::Or(children) => children + .iter() + .map(consumer_ids) + .collect::>>() + .map(|sets| sets.into_iter().flatten().collect()), + _ => None, + } +} diff --git a/crates/integrations/datafusion/src/system_tables/files.rs b/crates/paimon/src/table/system_table/files.rs similarity index 56% rename from crates/integrations/datafusion/src/system_tables/files.rs rename to crates/paimon/src/table/system_table/files.rs index e9749007a..66f7100fc 100644 --- a/crates/integrations/datafusion/src/system_tables/files.rs +++ b/crates/paimon/src/table/system_table/files.rs @@ -20,32 +20,29 @@ use std::collections::{BTreeMap, HashMap}; use std::sync::{Arc, OnceLock}; -use async_trait::async_trait; -use datafusion::arrow::array::{ - Int32Array, Int64Array, ListBuilder, RecordBatch, StringArray, StringBuilder, - TimestampMillisecondArray, +use crate::spec::{BinaryRow, DataField, DataFileMeta, TableSchema}; +use crate::table::{DataSplit, Table}; +use arrow_array::builder::{ListBuilder, StringBuilder}; +use arrow_array::{ + ArrayRef, Int32Array, Int64Array, RecordBatch, StringArray, TimestampMillisecondArray, }; -use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, Schema, SchemaRef, TimeUnit}; -use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::Result as DFResult; -use datafusion::logical_expr::Expr; -use datafusion::physical_plan::ExecutionPlan; -use paimon::spec::{BinaryRow, DataField, DataFileMeta, TableSchema}; -use paimon::table::{DataSplit, Table}; +use arrow_schema::{DataType as ArrowDataType, Field, Schema, SchemaRef, TimeUnit}; use super::row_string_cast::{ format_row_as_java_array_string, format_row_as_java_cast_string, - format_row_field_as_java_cast_string, + format_row_field_as_java_object_string, }; -use crate::error::to_datafusion_error; -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(FilesTable { table })) +use crate::Result; + +pub(super) fn key_ranges_schema() -> SchemaRef { + let files = files_schema(); + let mut fields = files.fields()[..10].to_vec(); + fields.push(files.field(18).clone().into()); + Arc::new(Schema::new(fields)) } -fn files_schema() -> SchemaRef { +pub(super) fn files_schema() -> SchemaRef { static SCHEMA: OnceLock = OnceLock::new(); SCHEMA .get_or_init(|| { @@ -70,7 +67,7 @@ fn files_schema() -> SchemaRef { ArrowDataType::Timestamp(TimeUnit::Millisecond, None), true, ), - Field::new("delete_row_count", ArrowDataType::Int64, true), + Field::new("deleteRowCount", ArrowDataType::Int64, true), Field::new("file_source", ArrowDataType::Utf8, true), Field::new("first_row_id", ArrowDataType::Int64, true), Field::new( @@ -83,43 +80,30 @@ fn files_schema() -> SchemaRef { .clone() } -#[derive(Debug)] -pub(super) struct FilesTable { - table: Table, -} - -#[async_trait] -impl TableProvider for FilesTable { - fn schema(&self) -> SchemaRef { +pub(super) async fn read( + table: &Table, + key_ranges: bool, + _projection: Option<&[usize]>, + _predicate: Option<&crate::spec::Predicate>, +) -> Result { + let rows = collect_file_rows(table, !key_ranges).await?; + + let schema = if key_ranges { + key_ranges_schema() + } else { files_schema() - } - - fn table_type(&self) -> TableType { - TableType::View - } + }; + let batch = if key_ranges { + let mut columns = common_file_columns(&rows); + columns.push(Arc::new(Int64Array::from_iter( + rows.iter().map(|row| row.first_row_id), + ))); + super::record_batch(schema.clone(), columns)? + } else { + file_rows_to_record_batch(&rows)? + }; - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - _filters: &[Expr], - _limit: Option, - ) -> DFResult> { - let table = self.table.clone(); - let rows = - crate::runtime::await_with_runtime(async move { collect_file_rows(&table).await }) - .await - .map_err(to_datafusion_error)?; - - let schema = files_schema(); - let batch = file_rows_to_record_batch(&rows)?; - - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema, - projection.cloned(), - )?) - } + Ok(batch) } #[derive(Debug)] @@ -134,9 +118,7 @@ struct FileRow { file_size_in_bytes: i64, min_key: Option, max_key: Option, - null_value_counts: String, - min_value_stats: String, - max_value_stats: String, + stats: Option, min_sequence_number: Option, max_sequence_number: Option, creation_time: Option, @@ -146,7 +128,7 @@ struct FileRow { write_cols: Option>, } -async fn collect_file_rows(table: &Table) -> paimon::Result> { +async fn collect_file_rows(table: &Table, include_stats: bool) -> crate::Result> { let scan = table .new_read_builder() .new_scan() @@ -159,7 +141,16 @@ async fn collect_file_rows(table: &Table) -> paimon::Result> { let mut rows = Vec::new(); for split in scan.splits() { - rows.extend(data_split_rows(table, split, &partition_fields, &mut schema_cache).await?); + rows.extend( + data_split_rows( + table, + split, + &partition_fields, + &mut schema_cache, + include_stats, + ) + .await?, + ); } Ok(rows) } @@ -169,7 +160,8 @@ async fn data_split_rows( split: &DataSplit, partition_fields: &[DataField], schema_cache: &mut HashMap>, -) -> paimon::Result> { + include_stats: bool, +) -> crate::Result> { let partition = format_partition(split.partition(), partition_fields)?; let mut rows = Vec::with_capacity(split.data_files().len()); for file in split.data_files() { @@ -177,7 +169,11 @@ async fn data_split_rows( let key_fields = key_fields_for_schema(&file_schema); let min_key = format_key(&file.min_key, &key_fields)?; let max_key = format_key(&file.max_key, &key_fields)?; - let stats = format_value_stats(table, schema_cache, file).await?; + let stats = if include_stats { + Some(format_value_stats(table, schema_cache, file).await?) + } else { + None + }; rows.push(FileRow { partition: partition.clone(), @@ -185,7 +181,7 @@ async fn data_split_rows( file_path: file .external_path .clone() - .unwrap_or_else(|| split.data_file_path(file)), + .unwrap_or_else(|| java_file_path(&split.data_file_path(file))), file_format: data_file_format_identifier(&file.file_name)?, schema_id: file.schema_id, level: file.level, @@ -193,9 +189,7 @@ async fn data_split_rows( file_size_in_bytes: file.file_size, min_key, max_key, - null_value_counts: stats.null_value_counts, - min_value_stats: stats.min_value_stats, - max_value_stats: stats.max_value_stats, + stats, min_sequence_number: Some(file.min_sequence_number), max_sequence_number: Some(file.max_sequence_number), creation_time: file.creation_time.map(|t| t.timestamp_millis()), @@ -212,7 +206,7 @@ async fn schema_for_id( table: &Table, schema_cache: &mut HashMap>, schema_id: i64, -) -> paimon::Result> { +) -> crate::Result> { if let Some(schema) = schema_cache.get(&schema_id) { return Ok(schema.clone()); } @@ -224,7 +218,7 @@ async fn schema_for_id( fn format_partition( partition: &BinaryRow, partition_fields: &[DataField], -) -> paimon::Result> { +) -> crate::Result> { if partition_fields.is_empty() { return Ok(Some("{}".to_string())); } @@ -249,7 +243,7 @@ fn key_fields_for_schema(schema: &TableSchema) -> Vec { .collect() } -fn format_key(bytes: &[u8], key_fields: &[DataField]) -> paimon::Result> { +fn format_key(bytes: &[u8], key_fields: &[DataField]) -> crate::Result> { if bytes.is_empty() { return Ok(None); } @@ -271,10 +265,18 @@ async fn format_value_stats( table: &Table, schema_cache: &mut HashMap>, file: &DataFileMeta, -) -> paimon::Result { - let table_fields = table.schema().fields(); +) -> crate::Result { let file_schema = schema_for_id(table, schema_cache, file.schema_id).await?; - let data_fields = file_schema.fields(); + format_file_value_stats(table.schema().fields(), &file_schema, file) +} + +fn format_file_value_stats( + table_fields: &[DataField], + file_schema: &TableSchema, + file: &DataFileMeta, +) -> crate::Result { + let physical_fields = file_schema.data_file_fields(file.write_cols.as_deref()); + let data_fields = physical_fields.as_slice(); let field_mapping = table_to_data_field_mapping(table_fields, data_fields); let dense_mapping = dense_stats_mapping(data_fields, file.value_stats_cols.as_deref()); let min_row = decode_stats_row(file.value_stats.min_values())?; @@ -302,11 +304,21 @@ async fn format_value_stats( .unwrap_or_else(|| table_field.data_type()); lower_bounds.insert( table_field.name().to_string(), - format_stats_value(min_row.as_ref(), stats_index, value_type)?, + format_stats_value( + min_row.as_ref(), + stats_index, + value_type, + table_field.data_type(), + )?, ); upper_bounds.insert( table_field.name().to_string(), - format_stats_value(max_row.as_ref(), stats_index, value_type)?, + format_stats_value( + max_row.as_ref(), + stats_index, + value_type, + table_field.data_type(), + )?, ); } @@ -377,7 +389,7 @@ fn stats_index_for_data_field( } } -fn decode_stats_row(bytes: &[u8]) -> paimon::Result> { +fn decode_stats_row(bytes: &[u8]) -> crate::Result> { if bytes.is_empty() { Ok(None) } else { @@ -388,8 +400,9 @@ fn decode_stats_row(bytes: &[u8]) -> paimon::Result> { fn format_stats_value( row: Option<&BinaryRow>, stats_index: Option, - data_type: &paimon::spec::DataType, -) -> paimon::Result> { + data_type: &crate::spec::DataType, + target_type: &crate::spec::DataType, +) -> crate::Result> { let Some(row) = row else { return Ok(None); }; @@ -399,7 +412,24 @@ fn format_stats_value( if stats_index >= row.arity() as usize { return Ok(None); } - format_row_field_as_java_cast_string(row, stats_index, data_type) + if data_type == target_type { + return format_row_field_as_java_object_string(row, stats_index, data_type); + } + let source = row.field_to_arrow(stats_index, data_type)?; + let target = crate::arrow::cast_array_for_schema_evolution(&source, data_type, target_type)?; + let fields = [DataField::new(0, "value".to_string(), target_type.clone())]; + let schema = Arc::new(Schema::new(vec![Field::new( + "value", + target.data_type().clone(), + true, + )])); + let batch = + super::record_batch(schema, vec![target]).map_err(|error| crate::Error::DataInvalid { + message: format!("Invalid evolved file statistics: {error}"), + source: Some(Box::new(error)), + })?; + let row = BinaryRow::from_arrow(&batch, 0, &[0], &fields)?; + format_row_field_as_java_object_string(&row, 0, target_type) } fn format_java_map(map: &BTreeMap, value_to_string: F) -> String @@ -419,9 +449,17 @@ where out } -fn data_file_format_identifier(file_name: &str) -> paimon::Result { +fn java_file_path(path: &str) -> String { + if let Some((scheme, tail)) = path.split_once(":///") { + format!("{scheme}:/{tail}") + } else { + path.to_string() + } +} + +fn data_file_format_identifier(file_name: &str) -> crate::Result { let Some(dot) = file_name.rfind('.') else { - return Err(paimon::Error::DataInvalid { + return Err(crate::Error::DataInvalid { message: format!("{file_name} is not a legal file name."), source: None, }); @@ -430,7 +468,7 @@ fn data_file_format_identifier(file_name: &str) -> paimon::Result { let extension = &file_name[dot + 1..]; if is_hadoop_compression_extension(extension) { let Some(second_dot) = file_name[..dot].rfind('.') else { - return Err(paimon::Error::DataInvalid { + return Err(crate::Error::DataInvalid { message: format!("{file_name} is not a legal file name."), source: None, }); @@ -455,18 +493,39 @@ fn file_source_to_string(file_source: Option) -> Option { }) } -fn file_rows_to_record_batch(rows: &[FileRow]) -> DFResult { +fn common_file_columns(rows: &[FileRow]) -> Vec { + vec![ + Arc::new(StringArray::from_iter( + rows.iter().map(|r| r.partition.as_deref()), + )), + Arc::new(Int32Array::from_iter_values(rows.iter().map(|r| r.bucket))), + Arc::new(StringArray::from_iter_values( + rows.iter().map(|r| r.file_path.as_str()), + )), + Arc::new(StringArray::from_iter_values( + rows.iter().map(|r| r.file_format.as_str()), + )), + Arc::new(Int64Array::from_iter_values( + rows.iter().map(|r| r.schema_id), + )), + Arc::new(Int32Array::from_iter_values(rows.iter().map(|r| r.level))), + Arc::new(Int64Array::from_iter_values( + rows.iter().map(|r| r.record_count), + )), + Arc::new(Int64Array::from_iter_values( + rows.iter().map(|r| r.file_size_in_bytes), + )), + Arc::new(StringArray::from_iter( + rows.iter().map(|r| r.min_key.as_deref()), + )), + Arc::new(StringArray::from_iter( + rows.iter().map(|r| r.max_key.as_deref()), + )), + ] +} + +fn file_rows_to_record_batch(rows: &[FileRow]) -> Result { let n = rows.len(); - let mut partitions = Vec::with_capacity(n); - let mut buckets = Vec::with_capacity(n); - let mut file_paths = Vec::with_capacity(n); - let mut file_formats = Vec::with_capacity(n); - let mut schema_ids = Vec::with_capacity(n); - let mut levels = Vec::with_capacity(n); - let mut record_counts = Vec::with_capacity(n); - let mut file_sizes = Vec::with_capacity(n); - let mut min_keys = Vec::with_capacity(n); - let mut max_keys = Vec::with_capacity(n); let mut null_value_counts = Vec::with_capacity(n); let mut min_value_stats = Vec::with_capacity(n); let mut max_value_stats = Vec::with_capacity(n); @@ -479,19 +538,16 @@ fn file_rows_to_record_batch(rows: &[FileRow]) -> DFResult { let mut write_cols = ListBuilder::new(StringBuilder::new()); for row in rows { - partitions.push(row.partition.clone()); - buckets.push(row.bucket); - file_paths.push(row.file_path.clone()); - file_formats.push(row.file_format.clone()); - schema_ids.push(row.schema_id); - levels.push(row.level); - record_counts.push(row.record_count); - file_sizes.push(row.file_size_in_bytes); - min_keys.push(row.min_key.clone()); - max_keys.push(row.max_key.clone()); - null_value_counts.push(row.null_value_counts.clone()); - min_value_stats.push(row.min_value_stats.clone()); - max_value_stats.push(row.max_value_stats.clone()); + let stats = row + .stats + .as_ref() + .ok_or_else(|| crate::Error::DataInvalid { + message: "Files table requires value statistics".to_string(), + source: None, + })?; + null_value_counts.push(stats.null_value_counts.clone()); + min_value_stats.push(stats.min_value_stats.clone()); + max_value_stats.push(stats.max_value_stats.clone()); min_sequence_numbers.push(row.min_sequence_number); max_sequence_numbers.push(row.max_sequence_number); creation_times.push(row.creation_time); @@ -509,37 +565,136 @@ fn file_rows_to_record_batch(rows: &[FileRow]) -> DFResult { } } - Ok(RecordBatch::try_new( - files_schema(), - vec![ - Arc::new(StringArray::from(partitions)), - Arc::new(Int32Array::from(buckets)), - Arc::new(StringArray::from(file_paths)), - Arc::new(StringArray::from(file_formats)), - Arc::new(Int64Array::from(schema_ids)), - Arc::new(Int32Array::from(levels)), - Arc::new(Int64Array::from(record_counts)), - Arc::new(Int64Array::from(file_sizes)), - Arc::new(StringArray::from(min_keys)), - Arc::new(StringArray::from(max_keys)), - Arc::new(StringArray::from(null_value_counts)), - Arc::new(StringArray::from(min_value_stats)), - Arc::new(StringArray::from(max_value_stats)), - Arc::new(Int64Array::from(min_sequence_numbers)), - Arc::new(Int64Array::from(max_sequence_numbers)), - Arc::new(TimestampMillisecondArray::from(creation_times)), - Arc::new(Int64Array::from(delete_row_counts)), - Arc::new(StringArray::from(file_sources)), - Arc::new(Int64Array::from(first_row_ids)), - Arc::new(write_cols.finish()), - ], - )?) + let mut columns = common_file_columns(rows); + columns.extend(vec![ + Arc::new(StringArray::from(null_value_counts)) as ArrayRef, + Arc::new(StringArray::from(min_value_stats)), + Arc::new(StringArray::from(max_value_stats)), + Arc::new(Int64Array::from(min_sequence_numbers)), + Arc::new(Int64Array::from(max_sequence_numbers)), + Arc::new(TimestampMillisecondArray::from(creation_times)), + Arc::new(Int64Array::from(delete_row_counts)), + Arc::new(StringArray::from(file_sources)), + Arc::new(Int64Array::from(first_row_ids)), + Arc::new(write_cols.finish()), + ]); + super::record_batch(files_schema(), columns) } #[cfg(test)] mod tests { use super::*; + fn stats_file(columns: Option>) -> DataFileMeta { + use crate::spec::{BinaryTableStats, DataType, Datum, IntType}; + let typ = DataType::Int(IntType::new()); + let value = Datum::Int(42); + let row = BinaryRow::from_datums(&[(Some(&value), &typ)]).to_serialized_bytes(); + DataFileMeta { + file_name: "data.parquet".to_string(), + file_size: 100, + row_count: 2, + min_key: vec![], + max_key: vec![], + key_stats: BinaryTableStats::empty(), + value_stats: BinaryTableStats::new(row.clone(), row, vec![Some(0)]), + min_sequence_number: 0, + max_sequence_number: 1, + schema_id: 0, + level: 0, + extra_files: vec![], + creation_time: None, + delete_row_count: None, + embedded_index: None, + file_source: None, + value_stats_cols: None, + external_path: None, + first_row_id: None, + write_cols: columns, + column_max_sequence_numbers: None, + } + } + + #[test] + fn file_stats_follow_physical_columns_and_schema_casts() { + use crate::spec::{DataType, DecimalType, IntType, Schema as PaimonSchema}; + let schema = PaimonSchema::builder() + .column("a", DataType::Int(IntType::new())) + .column("b", DataType::Int(IntType::new())) + .build() + .unwrap(); + let schema = TableSchema::new(0, &schema); + let mut file = stats_file(Some(vec!["b".to_string()])); + let result = format_file_value_stats(schema.fields(), &schema, &file).unwrap(); + assert_eq!(result.min_value_stats, "{a=null, b=42}"); + assert_eq!(result.null_value_counts, "{a=2, b=0}"); + // Missing dense stats are unknown; missing physical fields are all NULL. + file.value_stats_cols = Some(vec![]); + let result = format_file_value_stats(schema.fields(), &schema, &file).unwrap(); + assert_eq!(result.min_value_stats, "{a=null, b=null}"); + assert_eq!(result.null_value_counts, "{a=2, b=null}"); + file.value_stats_cols = None; + let target = vec![DataField::new( + 1, + "renamed".to_string(), + DataType::Decimal(DecimalType::new(10, 2).unwrap()), + )]; + let result = format_file_value_stats(&target, &schema, &file).unwrap(); + assert_eq!(result.min_value_stats, "{renamed=42.00}"); + } + + #[test] + fn stats_cast_overflow_is_nullable_even_for_a_required_field() { + use crate::spec::{DataType, DecimalType, IntType}; + let source_type = DataType::Int(IntType::new()); + let value = crate::spec::Datum::Int(42); + let row = BinaryRow::from_datums(&[(Some(&value), &source_type)]); + let target = DataType::Decimal(DecimalType::with_nullable(false, 1, 0).unwrap()); + assert_eq!( + format_stats_value(Some(&row), Some(0), &source_type, &target).unwrap(), + None + ); + } + + #[test] + fn key_range_batch_does_not_require_value_statistics() { + let row = FileRow { + partition: Some("{}".into()), + bucket: 0, + file_path: "s3://bucket/data.parquet".into(), + file_format: "parquet".into(), + schema_id: 0, + level: 0, + record_count: 2, + file_size_in_bytes: 100, + min_key: Some("[1, a]".into()), + max_key: Some("[2, b]".into()), + stats: None, + min_sequence_number: None, + max_sequence_number: None, + creation_time: None, + delete_row_count: None, + file_source: None, + first_row_id: Some(8), + write_cols: None, + }; + let rows = [row]; + let mut columns = common_file_columns(&rows); + columns.push(Arc::new(Int64Array::from(vec![Some(8)]))); + let batch = RecordBatch::try_new(key_ranges_schema(), columns).unwrap(); + assert_eq!(batch.num_rows(), 1); + assert_eq!(batch.num_columns(), 11); + assert!(file_rows_to_record_batch(&rows).is_err()); + assert_eq!( + java_file_path("file:///tmp/data.parquet"), + "file:/tmp/data.parquet" + ); + assert_eq!( + java_file_path("s3://bucket/data.parquet"), + "s3://bucket/data.parquet" + ); + } + #[test] fn test_data_file_format_identifier() { assert_eq!( diff --git a/crates/integrations/datafusion/src/system_tables/manifests.rs b/crates/paimon/src/table/system_table/manifests.rs similarity index 58% rename from crates/integrations/datafusion/src/system_tables/manifests.rs rename to crates/paimon/src/table/system_table/manifests.rs index 1acf7a522..d75c20338 100644 --- a/crates/integrations/datafusion/src/system_tables/manifests.rs +++ b/crates/paimon/src/table/system_table/manifests.rs @@ -19,29 +19,19 @@ use std::sync::{Arc, OnceLock}; -use async_trait::async_trait; -use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray}; -use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef}; -use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::Result as DFResult; -use datafusion::logical_expr::Expr; -use datafusion::physical_plan::ExecutionPlan; -use paimon::spec::{BinaryRow, DataField, ManifestFileMeta, ManifestList}; -use paimon::table::Table; +use crate::spec::{BinaryRow, DataField, ManifestFileMeta, ManifestList}; +use crate::table::Table; +use arrow_array::{Int64Array, RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema, SchemaRef}; use super::row_string_cast::format_row_as_java_cast_string; -use crate::error::to_datafusion_error; const MIN_PARTITION_STATS_INDEX: usize = 5; const MAX_PARTITION_STATS_INDEX: usize = 6; -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(ManifestsTable { table })) -} +use crate::Result; -fn manifests_schema() -> SchemaRef { +pub(super) fn manifests_schema() -> SchemaRef { static SCHEMA: OnceLock = OnceLock::new(); SCHEMA .get_or_init(|| { @@ -60,108 +50,76 @@ fn manifests_schema() -> SchemaRef { .clone() } -#[derive(Debug)] -pub(super) struct ManifestsTable { - table: Table, -} - -#[async_trait] -impl TableProvider for ManifestsTable { - fn schema(&self) -> SchemaRef { - manifests_schema() +pub(super) async fn read( + table: &Table, + projection: Option<&[usize]>, + _predicate: Option<&crate::spec::Predicate>, +) -> Result { + let metas = collect_manifests(table).await?; + + let n = metas.len(); + let mut file_names: Vec = Vec::with_capacity(n); + let mut file_sizes = Vec::with_capacity(n); + let mut num_added = Vec::with_capacity(n); + let mut num_deleted = Vec::with_capacity(n); + let mut schema_ids = Vec::with_capacity(n); + let mut min_partition_stats: Vec> = Vec::with_capacity(n); + let mut max_partition_stats: Vec> = Vec::with_capacity(n); + let mut min_row_ids: Vec> = Vec::with_capacity(n); + let mut max_row_ids: Vec> = Vec::with_capacity(n); + let partition_fields = table.schema().partition_fields(); + let projected_columns = projection; + let materialize_min_partition_stats = + should_materialize_column(projected_columns, MIN_PARTITION_STATS_INDEX); + let materialize_max_partition_stats = + should_materialize_column(projected_columns, MAX_PARTITION_STATS_INDEX); + + for meta in metas { + let stats = meta.partition_stats(); + file_names.push(meta.file_name().to_string()); + file_sizes.push(meta.file_size()); + num_added.push(meta.num_added_files()); + num_deleted.push(meta.num_deleted_files()); + schema_ids.push(meta.schema_id()); + min_partition_stats.push(materialize_partition_stats_value( + materialize_min_partition_stats, + stats.min_values(), + stats.null_counts(), + &partition_fields, + )?); + max_partition_stats.push(materialize_partition_stats_value( + materialize_max_partition_stats, + stats.max_values(), + stats.null_counts(), + &partition_fields, + )?); + min_row_ids.push(meta.min_row_id()); + max_row_ids.push(meta.max_row_id()); } - fn table_type(&self) -> TableType { - TableType::View - } - - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - _filters: &[Expr], - _limit: Option, - ) -> DFResult> { - let table = self.table.clone(); - let metas = - crate::runtime::await_with_runtime(async move { collect_manifests(&table).await }) - .await - .map_err(to_datafusion_error)?; - - let n = metas.len(); - let mut file_names: Vec = Vec::with_capacity(n); - let mut file_sizes = Vec::with_capacity(n); - let mut num_added = Vec::with_capacity(n); - let mut num_deleted = Vec::with_capacity(n); - let mut schema_ids = Vec::with_capacity(n); - let mut min_partition_stats: Vec> = Vec::with_capacity(n); - let mut max_partition_stats: Vec> = Vec::with_capacity(n); - let mut min_row_ids: Vec> = Vec::with_capacity(n); - let mut max_row_ids: Vec> = Vec::with_capacity(n); - let partition_fields = self.table.schema().partition_fields(); - let projected_columns = projection.map(Vec::as_slice); - let materialize_min_partition_stats = - should_materialize_column(projected_columns, MIN_PARTITION_STATS_INDEX); - let materialize_max_partition_stats = - should_materialize_column(projected_columns, MAX_PARTITION_STATS_INDEX); - - for meta in metas { - let stats = meta.partition_stats(); - file_names.push(meta.file_name().to_string()); - file_sizes.push(meta.file_size()); - num_added.push(meta.num_added_files()); - num_deleted.push(meta.num_deleted_files()); - schema_ids.push(meta.schema_id()); - min_partition_stats.push( - materialize_partition_stats_value( - materialize_min_partition_stats, - stats.min_values(), - stats.null_counts(), - &partition_fields, - ) - .map_err(to_datafusion_error)?, - ); - max_partition_stats.push( - materialize_partition_stats_value( - materialize_max_partition_stats, - stats.max_values(), - stats.null_counts(), - &partition_fields, - ) - .map_err(to_datafusion_error)?, - ); - min_row_ids.push(meta.min_row_id()); - max_row_ids.push(meta.max_row_id()); - } - - let schema = manifests_schema(); - let batch = RecordBatch::try_new( - schema.clone(), - vec![ - Arc::new(StringArray::from(file_names)), - Arc::new(Int64Array::from(file_sizes)), - Arc::new(Int64Array::from(num_added)), - Arc::new(Int64Array::from(num_deleted)), - Arc::new(Int64Array::from(schema_ids)), - Arc::new(StringArray::from(min_partition_stats)), - Arc::new(StringArray::from(max_partition_stats)), - Arc::new(Int64Array::from(min_row_ids)), - Arc::new(Int64Array::from(max_row_ids)), - ], - )?; - - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema, - projection.cloned(), - )?) - } + let schema = manifests_schema(); + let batch = super::record_batch( + schema.clone(), + vec![ + Arc::new(StringArray::from(file_names)), + Arc::new(Int64Array::from(file_sizes)), + Arc::new(Int64Array::from(num_added)), + Arc::new(Int64Array::from(num_deleted)), + Arc::new(Int64Array::from(schema_ids)), + Arc::new(StringArray::from(min_partition_stats)), + Arc::new(StringArray::from(max_partition_stats)), + Arc::new(Int64Array::from(min_row_ids)), + Arc::new(Int64Array::from(max_row_ids)), + ], + )?; + + Ok(batch) } -async fn collect_manifests(table: &Table) -> paimon::Result> { +async fn collect_manifests(table: &Table) -> crate::Result> { let file_io = table.file_io(); let manifest_sm = - paimon::table::SnapshotManager::new(file_io.clone(), table.location().to_string()); + crate::table::SnapshotManager::new(file_io.clone(), table.location().to_string()); let Some(snapshot) = table.resolve_read_snapshot().await? else { return Ok(Vec::new()); }; @@ -198,7 +156,7 @@ fn materialize_partition_stats_value( value_bytes: &[u8], null_counts: &[Option], partition_fields: &[DataField], -) -> paimon::Result> { +) -> crate::Result> { if materialize { format_partition_stats_value(value_bytes, null_counts, partition_fields) } else { @@ -210,7 +168,7 @@ fn format_partition_stats_value( value_bytes: &[u8], null_counts: &[Option], partition_fields: &[DataField], -) -> paimon::Result> { +) -> crate::Result> { if value_bytes.is_empty() { return if partition_fields.is_empty() || null_counts.len() == partition_fields.len() { Ok(Some(format_all_null_partition_row(partition_fields.len()))) @@ -233,7 +191,7 @@ fn format_all_null_partition_row(arity: usize) -> String { #[cfg(test)] mod tests { use super::*; - use paimon::spec::{DataType as PaimonDataType, Datum, FloatType, IntType, VarCharType}; + use crate::spec::{DataType as PaimonDataType, Datum, FloatType, IntType, VarCharType}; fn field(name: &str, data_type: PaimonDataType) -> DataField { DataField::new(0, name.to_string(), data_type) diff --git a/crates/paimon/src/table/system_table/mod.rs b/crates/paimon/src/table/system_table/mod.rs new file mode 100644 index 000000000..b7893dbe6 --- /dev/null +++ b/crates/paimon/src/table/system_table/mod.rs @@ -0,0 +1,427 @@ +// 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-only metadata tables shared by Native bindings and query engines. + +use crate::spec::{CoreOptions, DataField, Predicate, RowType}; +use crate::table::Table; +use crate::{Error, Result}; +use arrow_array::{ArrayRef, RecordBatch}; +use arrow_schema::{Schema, SchemaRef}; +use std::sync::Arc; + +mod branches; +mod buckets; +mod consumers; +mod files; +mod manifests; +mod options; +mod row_string_cast; +mod schemas; +mod snapshots; +mod table_indexes; +mod tags; + +const NAMES: &[&str] = &[ + "branches", + "buckets", + "consumers", + "files", + "file_key_ranges", + "manifests", + "options", + "schemas", + "snapshots", + "table_indexes", + "tags", +]; + +/// Reads a small metadata table in one batch using the resolved table environment. +#[derive(Debug, Clone)] +pub struct SystemRead { + table: Table, + name: &'static str, + fields: Vec, + projection: Option>, + predicate: Option, + limit: Option, +} + +impl SystemRead { + pub(crate) fn new(table: Table, name: &str) -> Result { + let options = CoreOptions::new(table.schema().options()); + if options.table_type()?.requires_table_engine() { + return Err(invalid("System tables require a Paimon table")); + } + options.ensure_read_authorized()?; + let name = NAMES + .iter() + .copied() + .find(|candidate| candidate.eq_ignore_ascii_case(name)) + .ok_or_else(|| { + invalid(format!( + "System table '{name}' requires a catalog or is unknown" + )) + })?; + let fields = schema_for_name(name) + .fields() + .iter() + .enumerate() + .map(|(index, field)| { + let id = if name == "snapshots" && field.name() == "writer_version" { + 15 + } else { + index as i32 + }; + Ok(DataField::new( + id, + field.name().clone(), + crate::arrow::arrow_to_paimon_type(field.data_type(), field.is_nullable())?, + )) + }) + .collect::>()?; + Ok(Self { + table, + name, + fields, + projection: None, + predicate: None, + limit: None, + }) + } + + /// Full logical schema used to resolve predicates, including unprojected columns. + pub fn full_row_type(&self) -> RowType { + RowType::with_nullable(false, self.fields.clone()) + } + + /// Logical output fields, preserving the requested order and repeated columns. + pub fn read_type(&self) -> Vec { + match &self.projection { + Some(indices) => indices + .iter() + .map(|&index| self.fields[index].clone()) + .collect(), + None => self.fields.clone(), + } + } + + /// Declared output schema, including an explicitly empty projection. + pub fn schema(&self) -> SchemaRef { + let schema = schema_for_name(self.name); + match &self.projection { + Some(indices) => Arc::new(Schema::new( + indices + .iter() + .map(|&index| schema.field(index).clone()) + .collect::>(), + )), + None => schema, + } + } + + /// Select output columns by name. An empty list selects zero columns. + pub fn with_projection(&mut self, columns: &[&str]) -> Result<&mut Self> { + let schema = schema_for_name(self.name); + self.projection = Some( + columns + .iter() + .map(|name| schema.index_of(name).map_err(arrow_error)) + .collect::>()?, + ); + Ok(self) + } + + /// Filter metadata using the full schema, before output projection and limit. + pub fn with_filter(&mut self, predicate: Predicate) -> Result<&mut Self> { + self.predicate = Some(normalize_predicate(predicate, &self.fields)?); + Ok(self) + } + + /// Limit returned rows after filtering. + pub fn with_limit(&mut self, limit: usize) -> &mut Self { + self.limit = Some(limit); + self + } + + /// Materialize metadata once per call. The returned batch is independent of + /// the catalog and file IO, and retains its declared schema even when empty. + pub async fn read(&self) -> Result { + if self.limit == Some(0) { + return Ok(RecordBatch::new_empty(self.schema())); + } + let mut needed = self.projection.clone(); + if let (Some(indices), Some(predicate)) = (&mut needed, &self.predicate) { + let mut leaves = Vec::new(); + crate::arrow::residual::collect_predicate_leaf_refs(predicate, &mut leaves); + for (column, _) in leaves { + let index = schema_for_name(self.name) + .index_of(column) + .map_err(arrow_error)?; + if !indices.contains(&index) { + indices.push(index); + } + } + } + let mut batch = match self.name { + "branches" => { + branches::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? + } + "buckets" => { + buckets::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? + } + "consumers" => { + consumers::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? + } + "files" | "file_key_ranges" => { + files::read( + &self.table, + self.name == "file_key_ranges", + needed.as_deref(), + self.predicate.as_ref(), + ) + .await? + } + "manifests" => { + manifests::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? + } + "options" => { + options::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? + } + "schemas" => { + schemas::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? + } + "snapshots" => { + snapshots::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? + } + "table_indexes" => { + table_indexes::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await? + } + "tags" => tags::read(&self.table, needed.as_deref(), self.predicate.as_ref()).await?, + _ => unreachable!("validated system-table name"), + }; + if let Some(predicate) = &self.predicate { + let mask = crate::arrow::residual::evaluate_predicates_mask( + &batch, + std::slice::from_ref(predicate), + &self.fields, + &self.fields, + )? + .ok_or_else(|| invalid("System-table predicate could not be evaluated exactly"))?; + batch = + arrow_select::filter::filter_record_batch(&batch, &mask).map_err(arrow_error)?; + } + if let Some(indices) = &self.projection { + batch = batch.project(indices).map_err(arrow_error)?; + } + if let Some(limit) = self.limit { + batch = batch.slice(0, limit.min(batch.num_rows())); + } + Ok(batch) + } +} + +fn normalize_predicate(predicate: Predicate, fields: &[DataField]) -> Result { + Ok(match predicate { + Predicate::Leaf { + column, + op, + literals, + .. + } => { + let (index, field) = fields + .iter() + .enumerate() + .find(|(_, field)| field.name() == column) + .ok_or_else(|| invalid(format!("Unknown system-table filter column '{column}'")))?; + Predicate::Leaf { + column, + index, + data_type: field.data_type().clone(), + op, + literals, + } + } + Predicate::And(children) => Predicate::And( + children + .into_iter() + .map(|p| normalize_predicate(p, fields)) + .collect::>()?, + ), + Predicate::Or(children) => Predicate::Or( + children + .into_iter() + .map(|p| normalize_predicate(p, fields)) + .collect::>()?, + ), + Predicate::Not(child) => Predicate::Not(Box::new(normalize_predicate(*child, fields)?)), + other => other, + }) +} + +fn schema_for_name(name: &str) -> SchemaRef { + match name { + "branches" => branches::branches_schema(), + "buckets" => buckets::buckets_schema(), + "consumers" => consumers::consumers_schema(), + "files" => files::files_schema(), + "file_key_ranges" => files::key_ranges_schema(), + "manifests" => manifests::manifests_schema(), + "options" => options::options_schema(), + "schemas" => schemas::schemas_schema(), + "snapshots" => snapshots::snapshots_schema(), + "table_indexes" => table_indexes::table_indexes_schema(), + "tags" => tags::tags_schema(), + _ => unreachable!("validated name"), + } +} + +fn record_batch(schema: SchemaRef, columns: Vec) -> Result { + RecordBatch::try_new(schema, columns).map_err(arrow_error) +} + +fn local_timestamp_column(values: Vec) -> Result { + use crate::spec::{DataType, LocalZonedTimestampType, TimestampType}; + use arrow_array::TimestampMillisecondArray; + let values: ArrayRef = Arc::new(TimestampMillisecondArray::from(values)); + crate::arrow::cast_array_for_schema_evolution( + &values, + &DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3)?), + &DataType::Timestamp(TimestampType::new(3)?), + ) +} + +fn invalid(message: impl Into) -> Error { + Error::DataInvalid { + message: message.into(), + source: None, + } +} +fn arrow_error(error: arrow_schema::ArrowError) -> Error { + Error::DataInvalid { + message: error.to_string(), + source: Some(Box::new(error)), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::{ + DataType, Datum, IntType, PredicateOperator, Schema as PaimonSchema, TableSchema, + }; + use arrow_array::StringArray; + + async fn table() -> Table { + let file_io = crate::io::FileIO::from_path("memory:///") + .unwrap() + .build() + .unwrap(); + let schema = PaimonSchema::builder() + .column("id", DataType::Int(IntType::new())) + .option("a", "skip") + .option("b", "match") + .option("c", "match") + .build() + .unwrap(); + let schema = TableSchema::new(0, &schema); + file_io + .new_output("memory:///metadata/schema/schema-0") + .unwrap() + .write(serde_json::to_vec(&schema).unwrap().into()) + .await + .unwrap(); + Table::from_resolved_schema( + file_io, + crate::catalog::Identifier::new("default", "metadata"), + "memory:///metadata".to_string(), + schema, + "main", + ) + .unwrap() + } + + #[tokio::test] + async fn filter_precedes_projection_and_limit() { + let mut read = table().await.new_system_read("options").unwrap(); + read.with_projection(&["key"]).unwrap(); + read.with_filter(Predicate::Leaf { + column: "value".into(), + index: 999, + data_type: DataType::Int(IntType::new()), + op: PredicateOperator::Eq, + literals: vec![Datum::String("match".into())], + }) + .unwrap(); + read.with_limit(1); + let batch = read.read().await.unwrap(); + assert_eq!(batch.schema(), read.schema()); + assert_eq!(read.read_type()[0].name(), "key"); + assert_eq!(batch.num_rows(), 1); + assert_eq!( + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + "b" + ); + read.with_limit(0); + let batch = read.read().await.unwrap(); + assert_eq!(batch.schema(), read.schema()); + assert_eq!(batch.num_rows(), 0); + } + + #[tokio::test] + async fn empty_and_repeated_projections_preserve_rows() { + let mut read = table().await.new_system_read("options").unwrap(); + read.with_projection(&[]).unwrap(); + let batch = read.read().await.unwrap(); + assert_eq!(batch.num_rows(), 3); + assert_eq!(batch.num_columns(), 0); + assert!(read.read_type().is_empty()); + read.with_projection(&["value", "key", "value"]).unwrap(); + let batch = read.read().await.unwrap(); + assert_eq!(batch.num_rows(), 3); + assert_eq!( + batch + .schema() + .fields() + .iter() + .map(|field| field.name().as_str()) + .collect::>(), + ["value", "key", "value"] + ); + assert_eq!(batch.column(0), batch.column(2)); + assert!(read.with_projection(&["missing"]).is_err()); + } + + #[tokio::test] + async fn unknown_filter_column_is_rejected() { + let mut read = table().await.new_system_read("options").unwrap(); + assert!(read + .with_filter(Predicate::Leaf { + column: "missing".into(), + index: 0, + data_type: DataType::Int(IntType::new()), + op: PredicateOperator::IsNull, + literals: vec![], + }) + .is_err()); + } +} diff --git a/crates/paimon/src/table/system_table/options.rs b/crates/paimon/src/table/system_table/options.rs new file mode 100644 index 000000000..cce58793c --- /dev/null +++ b/crates/paimon/src/table/system_table/options.rs @@ -0,0 +1,64 @@ +// 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. + +//! Mirrors Java [OptionsTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/OptionsTable.java). + +use std::sync::{Arc, OnceLock}; + +use crate::table::Table; +use arrow_array::{RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema, SchemaRef}; + +use crate::Result; + +pub(super) fn options_schema() -> SchemaRef { + static SCHEMA: OnceLock = OnceLock::new(); + SCHEMA + .get_or_init(|| { + Arc::new(Schema::new(vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", DataType::Utf8, false), + ])) + }) + .clone() +} + +pub(super) async fn read( + table: &Table, + _projection: Option<&[usize]>, + _predicate: Option<&crate::spec::Predicate>, +) -> Result { + let manager = table.schema_manager(); + let latest = manager + .latest() + .await? + .ok_or_else(|| crate::Error::DataInvalid { + message: "Table does not exist".to_string(), + source: None, + })?; + // Java uses LinkedHashMap insertion order; HashMap has none — sort for stable output. + let mut entries: Vec<(&String, &String)> = latest.options().iter().collect(); + entries.sort_by(|a, b| a.0.cmp(b.0)); + + let keys = StringArray::from_iter_values(entries.iter().map(|(k, _)| k.as_str())); + let values = StringArray::from_iter_values(entries.iter().map(|(_, v)| v.as_str())); + + let schema = options_schema(); + let batch = super::record_batch(schema.clone(), vec![Arc::new(keys), Arc::new(values)])?; + + Ok(batch) +} diff --git a/crates/integrations/datafusion/src/system_tables/row_string_cast.rs b/crates/paimon/src/table/system_table/row_string_cast.rs similarity index 81% rename from crates/integrations/datafusion/src/system_tables/row_string_cast.rs rename to crates/paimon/src/table/system_table/row_string_cast.rs index 11c1d46a5..1101f606f 100644 --- a/crates/integrations/datafusion/src/system_tables/row_string_cast.rs +++ b/crates/paimon/src/table/system_table/row_string_cast.rs @@ -17,10 +17,10 @@ use std::num::NonZeroI32; -use chrono::{Local, NaiveDate, NaiveDateTime, TimeZone, Timelike}; +use crate::spec::{BinaryRow, DataField, DataType, Datum}; +use crate::{Error, Result}; +use chrono::NaiveDate; use lexical_write_float::{format::STANDARD, Options, ToLexicalWithOptions}; -use paimon::spec::{BinaryRow, DataField, DataType, Datum}; -use paimon::{Error, Result}; const MILLIS_PER_DAY: i64 = 86_400_000; const JAVA_FLOAT_OPTIONS: Options = Options::builder() @@ -58,13 +58,15 @@ pub(super) fn format_row_as_java_array_string( if pos > 0 { out.push_str(", "); } - out.push_str(&format_field(row, pos, field.data_type())?); + out.push_str( + &format_object_field(row, pos, field.data_type())?.unwrap_or_else(|| "null".into()), + ); } out.push(']'); Ok(out) } -pub(super) fn format_row_field_as_java_cast_string( +pub(super) fn format_row_field_as_java_object_string( row: &BinaryRow, pos: usize, data_type: &DataType, @@ -73,7 +75,26 @@ pub(super) fn format_row_field_as_java_cast_string( if row.is_null_at(pos) { return Ok(None); } - format_field(row, pos, data_type).map(Some) + format_object_field(row, pos, data_type) +} + +fn format_object_field( + row: &BinaryRow, + pos: usize, + data_type: &DataType, +) -> Result> { + let value = match row.get_datum(pos, data_type)? { + None => return Ok(None), + Some(Datum::Date(value) | Datum::Time(value)) => value.to_string(), + Some(Datum::Timestamp { millis, nanos } | Datum::LocalZonedTimestamp { millis, nanos }) => { + crate::common::format_iso_timestamp( + millis.div_euclid(1000), + (millis.rem_euclid(1000) * 1_000_000 + i64::from(nanos)) as u32, + ) + } + Some(_) => format_field(row, pos, data_type)?, + }; + Ok(Some(value)) } fn validate_row(row: &BinaryRow, fields: &[DataField]) -> Result<()> { @@ -154,11 +175,18 @@ fn format_field(row: &BinaryRow, pos: usize, data_type: &DataType) -> Result Ok(format_decimal_plain(unscaled, scale)), - (Datum::Timestamp { millis, nanos }, DataType::Timestamp(t)) => { - format_timestamp(millis, nanos, t.precision()) - } - (Datum::LocalZonedTimestamp { millis, nanos }, DataType::LocalZonedTimestamp(t)) => { - format_local_zoned_timestamp(millis, nanos, t.precision()) + (Datum::Timestamp { .. } | Datum::LocalZonedTimestamp { .. }, _) => { + let source = row.field_to_arrow(pos, data_type)?; + let target = DataType::VarChar(crate::spec::VarCharType::string_type()); + let strings = + crate::arrow::cast_array_for_schema_evolution(&source, data_type, &target)?; + let strings = strings + .as_any() + .downcast_ref::() + .ok_or_else(|| { + data_invalid("Timestamp string cast returned a non-string array".into()) + })?; + Ok(strings.value(0).to_string()) } (datum, _) => Err(data_invalid(format!( "Decoded row string cast datum {datum:?} does not match type {data_type:?}" @@ -264,76 +292,6 @@ fn format_decimal_plain(unscaled: i128, scale: u32) -> String { } } -fn format_timestamp(millis: i64, nano_of_milli: i32, precision: u32) -> Result { - format_timestamp_naive(millis_to_naive_datetime(millis, nano_of_milli)?, precision) -} - -fn format_local_zoned_timestamp(millis: i64, nano_of_milli: i32, precision: u32) -> Result { - let nanos = timestamp_nanos(millis, nano_of_milli)?; - let secs = millis.div_euclid(1000); - let local = Local - .timestamp_opt(secs, nanos) - .single() - .ok_or_else(|| data_invalid(format!("Invalid local zoned timestamp millis {millis}")))?; - format_timestamp_naive(local.naive_local(), precision) -} - -fn format_timestamp_naive(dt: NaiveDateTime, precision: u32) -> Result { - let precision = usize::try_from(precision).map_err(|e| Error::DataInvalid { - message: format!("Timestamp row string cast precision {precision} is invalid"), - source: Some(Box::new(e)), - })?; - if precision > 9 { - return Err(data_invalid(format!( - "Timestamp row string cast precision {precision} is outside 0..=9" - ))); - } - - let mut out = dt.format("%Y-%m-%d %H:%M:%S").to_string(); - if precision > 0 { - let fraction = format!("{:09}", dt.nanosecond()); - out.push('.'); - out.push_str(&fraction[..precision]); - } - Ok(out) -} - -fn millis_to_naive_datetime(millis: i64, nano_of_milli: i32) -> Result { - let nanos = timestamp_nanos(millis, nano_of_milli)?; - let days = millis.div_euclid(MILLIS_PER_DAY); - let millis_of_day = millis.rem_euclid(MILLIS_PER_DAY) as u64; - let nano_of_day = millis_of_day * 1_000_000 + u64::from(nanos % 1_000_000); - let ce_days = days.checked_add(719_163).ok_or_else(|| { - data_invalid(format!( - "Timestamp row string cast millis {millis} is outside supported range" - )) - })?; - let ce_days = i32::try_from(ce_days).map_err(|e| Error::DataInvalid { - message: format!("Timestamp row string cast millis {millis} is outside supported range"), - source: Some(Box::new(e)), - })?; - let date = NaiveDate::from_num_days_from_ce_opt(ce_days).ok_or_else(|| { - data_invalid(format!( - "Timestamp row string cast millis {millis} is outside supported range" - )) - })?; - let time = chrono::NaiveTime::from_num_seconds_from_midnight_opt( - (nano_of_day / 1_000_000_000) as u32, - (nano_of_day % 1_000_000_000) as u32, - ) - .ok_or_else(|| data_invalid(format!("Invalid timestamp millis {millis}")))?; - Ok(NaiveDateTime::new(date, time)) -} - -fn timestamp_nanos(millis: i64, nano_of_milli: i32) -> Result { - if !(0..=999_999).contains(&nano_of_milli) { - return Err(data_invalid(format!( - "Timestamp nano-of-millisecond {nano_of_milli} is outside 0..=999999" - ))); - } - Ok(millis.rem_euclid(1000) as u32 * 1_000_000 + nano_of_milli as u32) -} - fn data_invalid(message: String) -> Error { Error::DataInvalid { message, @@ -344,11 +302,12 @@ fn data_invalid(message: String) -> Error { #[cfg(test)] mod tests { use super::*; - use paimon::spec::{ + use crate::spec::{ BigIntType, BinaryType, BlobType, BooleanType, CharType, DateType, DecimalType, DoubleType, FloatType, IntType, LocalZonedTimestampType, SmallIntType, TimeType, TimestampType, TinyIntType, VarBinaryType, VarCharType, }; + use chrono::{Local, TimeZone}; fn field(id: i32, data_type: DataType) -> DataField { DataField::new(id, format!("f{id}"), data_type) @@ -540,6 +499,51 @@ mod tests { ); } + #[test] + fn metadata_objects_use_internal_values_and_iso_timestamps() { + let timestamp_type = DataType::Timestamp(TimestampType::new(3).unwrap()); + let date_type = DataType::Date(DateType::new()); + let time_type = DataType::Time(TimeType::new(3).unwrap()); + let values = [ + (Some(Datum::Date(1)), date_type), + (Some(Datum::Time(42)), time_type), + ( + Some(Datum::Timestamp { + millis: 123, + nanos: 0, + }), + timestamp_type, + ), + ( + Some(Datum::LocalZonedTimestamp { + millis: 0, + nanos: 0, + }), + DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()), + ), + ]; + let datums: Vec<_> = values + .iter() + .map(|(value, typ)| (value.as_ref(), typ)) + .collect(); + let row = BinaryRow::from_datums(&datums); + let fields: Vec<_> = values + .iter() + .enumerate() + .map(|(i, (_, typ))| DataField::new(i as i32, i.to_string(), typ.clone())) + .collect(); + assert_eq!( + format_row_as_java_array_string(&row, &fields).unwrap(), + "[1, 42, 1970-01-01T00:00:00.123, 1970-01-01T00:00]" + ); + assert_eq!( + format_row_field_as_java_object_string(&row, 2, fields[2].data_type()) + .unwrap() + .as_deref(), + Some("1970-01-01T00:00:00.123") + ); + } + #[test] fn test_format_unsupported_type_returns_error() { let data_type = DataType::Blob(BlobType::new()); diff --git a/crates/paimon/src/table/system_table/schemas.rs b/crates/paimon/src/table/system_table/schemas.rs new file mode 100644 index 000000000..638ab6a9d --- /dev/null +++ b/crates/paimon/src/table/system_table/schemas.rs @@ -0,0 +1,98 @@ +// 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. + +//! Mirrors Java [SchemasTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java). + +use std::sync::{Arc, OnceLock}; + +use crate::table::Table; +use arrow_array::{Int64Array, RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; +use serde::Serialize; + +use crate::Result; + +pub(super) fn schemas_schema() -> SchemaRef { + static SCHEMA: OnceLock = OnceLock::new(); + SCHEMA + .get_or_init(|| { + Arc::new(Schema::new(vec![ + Field::new("schema_id", DataType::Int64, false), + Field::new("fields", DataType::Utf8, false), + Field::new("partition_keys", DataType::Utf8, false), + Field::new("primary_keys", DataType::Utf8, false), + Field::new("options", DataType::Utf8, false), + Field::new("comment", DataType::Utf8, true), + Field::new( + "update_time", + DataType::Timestamp(TimeUnit::Millisecond, None), + false, + ), + ])) + }) + .clone() +} + +pub(super) async fn read( + table: &Table, + _projection: Option<&[usize]>, + _predicate: Option<&crate::spec::Predicate>, +) -> Result { + let schemas = table.schema_manager().list_all().await?; + + let n = schemas.len(); + let mut schema_ids: Vec = Vec::with_capacity(n); + let mut fields_json: Vec = Vec::with_capacity(n); + let mut partition_keys_json: Vec = Vec::with_capacity(n); + let mut primary_keys_json: Vec = Vec::with_capacity(n); + let mut options_json: Vec = Vec::with_capacity(n); + let mut comments: Vec> = Vec::with_capacity(n); + let mut update_times: Vec = Vec::with_capacity(n); + + for schema in &schemas { + schema_ids.push(schema.id()); + fields_json.push(to_json(schema.fields())?); + partition_keys_json.push(to_json(schema.partition_keys())?); + primary_keys_json.push(to_json(schema.primary_keys())?); + options_json.push(to_json(schema.options())?); + comments.push(schema.comment().map(str::to_string)); + update_times.push(schema.time_millis()); + } + + let schema = schemas_schema(); + let batch = super::record_batch( + schema.clone(), + vec![ + Arc::new(Int64Array::from(schema_ids)), + Arc::new(StringArray::from(fields_json)), + Arc::new(StringArray::from(partition_keys_json)), + Arc::new(StringArray::from(primary_keys_json)), + Arc::new(StringArray::from(options_json)), + Arc::new(StringArray::from(comments)), + super::local_timestamp_column(update_times)?, + ], + )?; + + Ok(batch) +} + +fn to_json(value: &T) -> Result { + serde_json::to_string(value).map_err(|e| crate::Error::DataInvalid { + message: e.to_string(), + source: Some(Box::new(e)), + }) +} diff --git a/crates/paimon/src/table/system_table/snapshots.rs b/crates/paimon/src/table/system_table/snapshots.rs new file mode 100644 index 000000000..90d85e5ef --- /dev/null +++ b/crates/paimon/src/table/system_table/snapshots.rs @@ -0,0 +1,123 @@ +// 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. + +//! Mirrors Java [SnapshotsTable](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java). + +use std::sync::{Arc, OnceLock}; + +use crate::table::Table; +use arrow_array::{Int64Array, RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; + +use crate::Result; + +pub(super) fn snapshots_schema() -> SchemaRef { + static SCHEMA: OnceLock = OnceLock::new(); + SCHEMA + .get_or_init(|| { + Arc::new(Schema::new(vec![ + Field::new("snapshot_id", DataType::Int64, false), + Field::new("schema_id", DataType::Int64, false), + Field::new("commit_user", DataType::Utf8, false), + Field::new("commit_identifier", DataType::Int64, false), + Field::new("commit_kind", DataType::Utf8, false), + Field::new( + "commit_time", + DataType::Timestamp(TimeUnit::Millisecond, None), + false, + ), + Field::new("base_manifest_list", DataType::Utf8, false), + Field::new("delta_manifest_list", DataType::Utf8, false), + Field::new("changelog_manifest_list", DataType::Utf8, true), + Field::new("total_record_count", DataType::Int64, true), + Field::new("delta_record_count", DataType::Int64, true), + Field::new("changelog_record_count", DataType::Int64, true), + Field::new("watermark", DataType::Int64, true), + Field::new("next_row_id", DataType::Int64, true), + Field::new("writer_version", DataType::Utf8, true), + ])) + }) + .clone() +} + +pub(super) async fn read( + table: &Table, + _projection: Option<&[usize]>, + _predicate: Option<&crate::spec::Predicate>, +) -> Result { + let sm = table.snapshot_manager(); + let snapshots = sm.list_all().await?; + + let n = snapshots.len(); + let mut snapshot_ids = Vec::with_capacity(n); + let mut schema_ids = Vec::with_capacity(n); + let mut commit_users: Vec = Vec::with_capacity(n); + let mut commit_identifiers = Vec::with_capacity(n); + let mut commit_kinds: Vec = Vec::with_capacity(n); + let mut commit_times = Vec::with_capacity(n); + let mut base_manifest_lists: Vec = Vec::with_capacity(n); + let mut delta_manifest_lists: Vec = Vec::with_capacity(n); + let mut changelog_manifest_lists: Vec> = Vec::with_capacity(n); + let mut total_record_counts: Vec> = Vec::with_capacity(n); + let mut delta_record_counts: Vec> = Vec::with_capacity(n); + let mut changelog_record_counts: Vec> = Vec::with_capacity(n); + let mut watermarks: Vec> = Vec::with_capacity(n); + let mut next_row_ids: Vec> = Vec::with_capacity(n); + + let mut writer_versions = Vec::with_capacity(n); + for snap in &snapshots { + snapshot_ids.push(snap.id()); + schema_ids.push(snap.schema_id()); + commit_users.push(snap.commit_user().to_string()); + commit_identifiers.push(snap.commit_identifier()); + commit_kinds.push(snap.commit_kind().to_string()); + commit_times.push(snap.time_millis() as i64); + base_manifest_lists.push(snap.base_manifest_list().to_string()); + delta_manifest_lists.push(snap.delta_manifest_list().to_string()); + changelog_manifest_lists.push(snap.changelog_manifest_list().map(str::to_string)); + total_record_counts.push(snap.total_record_count()); + delta_record_counts.push(snap.delta_record_count()); + changelog_record_counts.push(snap.changelog_record_count()); + watermarks.push(snap.watermark()); + next_row_ids.push(snap.next_row_id()); + writer_versions.push(snap.writer_version()); + } + + let schema = snapshots_schema(); + let batch = super::record_batch( + schema.clone(), + vec![ + Arc::new(Int64Array::from(snapshot_ids)), + Arc::new(Int64Array::from(schema_ids)), + Arc::new(StringArray::from(commit_users)), + Arc::new(Int64Array::from(commit_identifiers)), + Arc::new(StringArray::from(commit_kinds)), + super::local_timestamp_column(commit_times)?, + Arc::new(StringArray::from(base_manifest_lists)), + Arc::new(StringArray::from(delta_manifest_lists)), + Arc::new(StringArray::from(changelog_manifest_lists)), + Arc::new(Int64Array::from(total_record_counts)), + Arc::new(Int64Array::from(delta_record_counts)), + Arc::new(Int64Array::from(changelog_record_counts)), + Arc::new(Int64Array::from(watermarks)), + Arc::new(Int64Array::from(next_row_ids)), + Arc::new(StringArray::from(writer_versions)), + ], + )?; + + Ok(batch) +} diff --git a/crates/integrations/datafusion/src/system_tables/table_indexes.rs b/crates/paimon/src/table/system_table/table_indexes.rs similarity index 55% rename from crates/integrations/datafusion/src/system_tables/table_indexes.rs rename to crates/paimon/src/table/system_table/table_indexes.rs index c51d3e55b..44d1d088e 100644 --- a/crates/integrations/datafusion/src/system_tables/table_indexes.rs +++ b/crates/paimon/src/table/system_table/table_indexes.rs @@ -19,31 +19,21 @@ use std::sync::{Arc, OnceLock}; -use async_trait::async_trait; -use datafusion::arrow::array::builder::{ - ArrayBuilder, Int32Builder, Int64Builder, ListBuilder, StringBuilder, StructBuilder, -}; -use datafusion::arrow::array::{ArrayRef, Int32Array, Int64Array, RecordBatch, StringArray}; -use datafusion::arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef}; -use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::Result as DFResult; -use datafusion::logical_expr::Expr; -use datafusion::physical_plan::ExecutionPlan; -use paimon::spec::{ +use crate::spec::{ BinaryRow, DataField, DeletionVectorMeta, FileKind, IndexManifest, IndexManifestEntry, }; -use paimon::table::Table; +use crate::table::Table; +use arrow_array::builder::{ + ArrayBuilder, Int32Builder, Int64Builder, ListBuilder, StringBuilder, StructBuilder, +}; +use arrow_array::{ArrayRef, Int32Array, Int64Array, RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Fields, Schema, SchemaRef}; use super::row_string_cast::format_row_as_java_cast_string; -use crate::error::to_datafusion_error; -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(TableIndexesTable { table })) -} +use crate::Result; -fn table_indexes_schema() -> SchemaRef { +pub(super) fn table_indexes_schema() -> SchemaRef { static SCHEMA: OnceLock = OnceLock::new(); SCHEMA .get_or_init(|| { @@ -82,113 +72,82 @@ fn dv_meta_fields() -> Fields { .into() } -#[derive(Debug)] -pub(super) struct TableIndexesTable { - table: Table, -} - -#[async_trait] -impl TableProvider for TableIndexesTable { - fn schema(&self) -> SchemaRef { - table_indexes_schema() - } - - fn table_type(&self) -> TableType { - TableType::View - } - - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - _filters: &[Expr], - _limit: Option, - ) -> DFResult> { - let table = self.table.clone(); - let entries = - crate::runtime::await_with_runtime(async move { collect_index_entries(&table).await }) - .await - .map_err(to_datafusion_error)?; +pub(super) async fn read( + table: &Table, + _projection: Option<&[usize]>, + _predicate: Option<&crate::spec::Predicate>, +) -> Result { + let entries = collect_index_entries(table).await?; - let partition_fields = self.table.schema().partition_fields(); - let fields = self.table.schema().fields(); - let n = entries.len(); - let mut partitions: Vec> = Vec::with_capacity(n); - let mut buckets = Vec::with_capacity(n); - let mut index_types = Vec::with_capacity(n); - let mut file_names = Vec::with_capacity(n); - let mut file_sizes = Vec::with_capacity(n); - let mut row_counts = Vec::with_capacity(n); - let mut dv_ranges = dv_ranges_builder(); - let mut row_range_starts: Vec> = Vec::with_capacity(n); - let mut row_range_ends: Vec> = Vec::with_capacity(n); - let mut index_field_ids: Vec> = Vec::with_capacity(n); - let mut index_field_names: Vec> = Vec::with_capacity(n); + let partition_fields = table.schema().partition_fields(); + let fields = table.schema().fields(); + let n = entries.len(); + let mut partitions: Vec> = Vec::with_capacity(n); + let mut buckets = Vec::with_capacity(n); + let mut index_types = Vec::with_capacity(n); + let mut file_names = Vec::with_capacity(n); + let mut file_sizes = Vec::with_capacity(n); + let mut row_counts = Vec::with_capacity(n); + let mut dv_ranges = dv_ranges_builder(); + let mut row_range_starts: Vec> = Vec::with_capacity(n); + let mut row_range_ends: Vec> = Vec::with_capacity(n); + let mut index_field_ids: Vec> = Vec::with_capacity(n); + let mut index_field_names: Vec> = Vec::with_capacity(n); - for entry in &entries { - let index_file = &entry.index_file; - partitions.push(Some(format_partition(&entry.partition, &partition_fields)?)); - buckets.push(entry.bucket); - index_types.push(index_file.index_type.as_str()); - file_names.push(index_file.file_name.as_str()); - file_sizes.push(index_file.file_size); - row_counts.push(index_file.row_count); - append_dv_ranges( - &mut dv_ranges, - index_file - .deletion_vectors_ranges - .as_ref() - .map(|ranges| ranges.iter()), - ); + for entry in &entries { + let index_file = &entry.index_file; + partitions.push(Some(format_partition(&entry.partition, &partition_fields)?)); + buckets.push(entry.bucket); + index_types.push(index_file.index_type.as_str()); + file_names.push(index_file.file_name.as_str()); + file_sizes.push(index_file.file_size); + row_counts.push(index_file.row_count); + append_dv_ranges( + &mut dv_ranges, + index_file + .deletion_vectors_ranges + .as_ref() + .map(|ranges| ranges.iter()), + ); - if let Some(global_meta) = &index_file.global_index_meta { - row_range_starts.push(Some(global_meta.row_range_start)); - row_range_ends.push(Some(global_meta.row_range_end)); - index_field_ids.push(Some(global_meta.index_field_id)); - index_field_names.push( - fields - .iter() - .find(|field| field.id() == global_meta.index_field_id) - .map(|field| field.name().to_string()), - ); - } else { - row_range_starts.push(None); - row_range_ends.push(None); - index_field_ids.push(None); - index_field_names.push(None); - } + if let Some(global_meta) = &index_file.global_index_meta { + row_range_starts.push(Some(global_meta.row_range_start)); + row_range_ends.push(Some(global_meta.row_range_end)); + index_field_ids.push(Some(global_meta.index_field_id)); + index_field_names.push(global_index_field_names(global_meta, fields)); + } else { + row_range_starts.push(None); + row_range_ends.push(None); + index_field_ids.push(None); + index_field_names.push(None); } + } - let schema = table_indexes_schema(); - let batch = RecordBatch::try_new( - schema.clone(), - vec![ - Arc::new(StringArray::from(partitions)), - Arc::new(Int32Array::from(buckets)), - Arc::new(StringArray::from(index_types)), - Arc::new(StringArray::from(file_names)), - Arc::new(Int64Array::from(file_sizes)), - Arc::new(Int64Array::from(row_counts)), - Arc::new(dv_ranges.finish()) as ArrayRef, - Arc::new(Int64Array::from(row_range_starts)), - Arc::new(Int64Array::from(row_range_ends)), - Arc::new(Int32Array::from(index_field_ids)), - Arc::new(StringArray::from(index_field_names)), - ], - )?; + let schema = table_indexes_schema(); + let batch = super::record_batch( + schema.clone(), + vec![ + Arc::new(StringArray::from(partitions)), + Arc::new(Int32Array::from(buckets)), + Arc::new(StringArray::from(index_types)), + Arc::new(StringArray::from(file_names)), + Arc::new(Int64Array::from(file_sizes)), + Arc::new(Int64Array::from(row_counts)), + Arc::new(dv_ranges.finish()) as ArrayRef, + Arc::new(Int64Array::from(row_range_starts)), + Arc::new(Int64Array::from(row_range_ends)), + Arc::new(Int32Array::from(index_field_ids)), + Arc::new(StringArray::from(index_field_names)), + ], + )?; - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema, - projection.cloned(), - )?) - } + Ok(batch) } -async fn collect_index_entries(table: &Table) -> paimon::Result> { +async fn collect_index_entries(table: &Table) -> crate::Result> { let file_io = table.file_io(); let manifest_sm = - paimon::table::SnapshotManager::new(file_io.clone(), table.location().to_string()); + crate::table::SnapshotManager::new(file_io.clone(), table.location().to_string()); let Some(snapshot) = table.resolve_read_snapshot().await? else { return Ok(Vec::new()); }; @@ -212,9 +171,25 @@ fn visible_index_entries(entries: Vec) -> Vec DFResult { - let row = BinaryRow::from_serialized_bytes(partition).map_err(to_datafusion_error)?; - format_row_as_java_cast_string(&row, partition_fields).map_err(to_datafusion_error) +fn global_index_field_names( + meta: &crate::spec::GlobalIndexMeta, + fields: &[DataField], +) -> Option { + let names: Option> = std::iter::once(&meta.index_field_id) + .chain(meta.extra_field_ids.iter().flatten()) + .map(|id| { + fields + .iter() + .find(|field| field.id() == *id) + .map(DataField::name) + }) + .collect(); + names.map(|names| names.join(",")) +} + +fn format_partition(partition: &[u8], partition_fields: &[DataField]) -> Result { + let row = BinaryRow::from_serialized_bytes(partition)?; + format_row_as_java_cast_string(&row, partition_fields) } fn dv_ranges_builder() -> ListBuilder { @@ -271,7 +246,7 @@ where #[cfg(test)] mod tests { use super::*; - use paimon::spec::IndexFileMeta; + use crate::spec::IndexFileMeta; fn index_entry(kind: FileKind, file_name: &str) -> IndexManifestEntry { IndexManifestEntry { @@ -291,6 +266,38 @@ mod tests { } } + #[test] + fn composite_index_names_preserve_order_and_require_every_field() { + use crate::spec::{GlobalIndexMeta, IntType}; + let fields = vec![ + DataField::new(3, "a".into(), crate::spec::DataType::Int(IntType::new())), + DataField::new( + 7, + "renamed".into(), + crate::spec::DataType::Int(IntType::new()), + ), + ]; + let mut meta = GlobalIndexMeta { + row_range_start: 0, + row_range_end: 1, + index_field_id: 7, + extra_field_ids: Some(vec![3]), + index_meta: None, + source_meta: None, + }; + assert_eq!( + global_index_field_names(&meta, &fields).as_deref(), + Some("renamed,a") + ); + meta.extra_field_ids = Some(vec![99]); + assert_eq!(global_index_field_names(&meta, &fields), None); + meta.extra_field_ids = None; + assert_eq!( + global_index_field_names(&meta, &fields).as_deref(), + Some("renamed") + ); + } + #[test] fn visible_index_entries_skips_delete_entries() { let entries = vec![ diff --git a/crates/integrations/datafusion/src/system_tables/tags.rs b/crates/paimon/src/table/system_table/tags.rs similarity index 58% rename from crates/integrations/datafusion/src/system_tables/tags.rs rename to crates/paimon/src/table/system_table/tags.rs index 433d59f17..2aacbdc34 100644 --- a/crates/integrations/datafusion/src/system_tables/tags.rs +++ b/crates/paimon/src/table/system_table/tags.rs @@ -19,24 +19,13 @@ use std::sync::{Arc, OnceLock}; -use async_trait::async_trait; -use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray, TimestampMillisecondArray}; -use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit}; -use datafusion::catalog::Session; -use datafusion::datasource::memory::MemorySourceConfig; -use datafusion::datasource::{TableProvider, TableType}; -use datafusion::error::Result as DFResult; -use datafusion::logical_expr::Expr; -use datafusion::physical_plan::ExecutionPlan; -use paimon::table::Table; +use crate::table::Table; +use arrow_array::{Int64Array, RecordBatch, StringArray, TimestampMillisecondArray}; +use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; -use crate::error::to_datafusion_error; +use crate::Result; -pub(super) fn build(table: Table) -> DFResult> { - Ok(Arc::new(TagsTable { table })) -} - -fn tags_schema() -> SchemaRef { +pub(super) fn tags_schema() -> SchemaRef { static SCHEMA: OnceLock = OnceLock::new(); SCHEMA .get_or_init(|| { @@ -61,73 +50,48 @@ fn tags_schema() -> SchemaRef { .clone() } -#[derive(Debug)] -pub(super) struct TagsTable { - table: Table, -} - -#[async_trait] -impl TableProvider for TagsTable { - fn schema(&self) -> SchemaRef { - tags_schema() - } - - fn table_type(&self) -> TableType { - TableType::View +pub(super) async fn read( + table: &Table, + _projection: Option<&[usize]>, + _predicate: Option<&crate::spec::Predicate>, +) -> Result { + let tm = table.tag_manager(); + let tags = tm.list_all_with_metadata().await?; + + let n = tags.len(); + let mut tag_names: Vec = Vec::with_capacity(n); + let mut snapshot_ids = Vec::with_capacity(n); + let mut schema_ids = Vec::with_capacity(n); + let mut commit_times = Vec::with_capacity(n); + let mut record_counts: Vec> = Vec::with_capacity(n); + let mut create_times: Vec> = Vec::with_capacity(n); + let mut time_retained: Vec> = Vec::with_capacity(n); + + for (name, snap, created, retained) in tags { + tag_names.push(name); + snapshot_ids.push(snap.id()); + schema_ids.push(snap.schema_id()); + commit_times.push(snap.time_millis() as i64); + record_counts.push(snap.total_record_count()); + create_times.push(created); + time_retained.push(retained.map(format_duration_iso8601)); } - async fn scan( - &self, - _state: &dyn Session, - projection: Option<&Vec>, - _filters: &[Expr], - _limit: Option, - ) -> DFResult> { - let tm = self.table.tag_manager(); - let tags = - crate::runtime::await_with_runtime(async move { tm.list_all_with_metadata().await }) - .await - .map_err(to_datafusion_error)?; - - let n = tags.len(); - let mut tag_names: Vec = Vec::with_capacity(n); - let mut snapshot_ids = Vec::with_capacity(n); - let mut schema_ids = Vec::with_capacity(n); - let mut commit_times = Vec::with_capacity(n); - let mut record_counts: Vec> = Vec::with_capacity(n); - let mut create_times: Vec> = Vec::with_capacity(n); - let mut time_retained: Vec> = Vec::with_capacity(n); - - for (name, snap, created, retained) in tags { - tag_names.push(name); - snapshot_ids.push(snap.id()); - schema_ids.push(snap.schema_id()); - commit_times.push(snap.time_millis() as i64); - record_counts.push(snap.total_record_count()); - create_times.push(created); - time_retained.push(retained.map(format_duration_iso8601)); - } - - let schema = tags_schema(); - let batch = RecordBatch::try_new( - schema.clone(), - vec![ - Arc::new(StringArray::from(tag_names)), - Arc::new(Int64Array::from(snapshot_ids)), - Arc::new(Int64Array::from(schema_ids)), - Arc::new(TimestampMillisecondArray::from(commit_times)), - Arc::new(Int64Array::from(record_counts)), - Arc::new(TimestampMillisecondArray::from(create_times)), - Arc::new(StringArray::from(time_retained)), - ], - )?; - - Ok(MemorySourceConfig::try_new_exec( - &[vec![batch]], - schema, - projection.cloned(), - )?) - } + let schema = tags_schema(); + let batch = super::record_batch( + schema.clone(), + vec![ + Arc::new(StringArray::from(tag_names)), + Arc::new(Int64Array::from(snapshot_ids)), + Arc::new(Int64Array::from(schema_ids)), + super::local_timestamp_column(commit_times)?, + Arc::new(Int64Array::from(record_counts)), + Arc::new(TimestampMillisecondArray::from(create_times)), + Arc::new(StringArray::from(time_retained)), + ], + )?; + + Ok(batch) } /// Render a retention in seconds the way Java's `Duration.toString()` does, so