diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index 340cf07c2..06f34c3c4 100644 --- a/crates/integrations/datafusion/src/catalog.rs +++ b/crates/integrations/datafusion/src/catalog.rs @@ -693,11 +693,17 @@ impl SchemaProvider for PaimonSchemaProvider { let object = system_tables::parse_object_name_for_datafusion(name)?; if let Some(system_name) = object.system_table().map(str::to_string) { + let dynamic_options = self + .dynamic_options + .read() + .unwrap_or_else(|e| e.into_inner()) + .clone(); return await_with_runtime(system_tables::load( Arc::clone(&self.catalog), self.database.clone(), object, system_name, + dynamic_options, )) .await; } diff --git a/crates/integrations/datafusion/src/physical_plan/audit_log.rs b/crates/integrations/datafusion/src/physical_plan/audit_log.rs new file mode 100644 index 000000000..5fdfeccb1 --- /dev/null +++ b/crates/integrations/datafusion/src/physical_plan/audit_log.rs @@ -0,0 +1,210 @@ +// 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. + +//! Audit execution policy layered over the shared scan mechanics. + +use std::sync::Arc; + +use datafusion::common::{stats::Precision, Statistics}; +use datafusion::error::Result as DFResult; +use datafusion::execution::{SendableRecordBatchStream, TaskContext}; +use datafusion::physical_plan::{DisplayAs, ExecutionPlan, PlanProperties}; +use paimon::table::AuditLogRead; + +use super::PaimonTableScan; + +/// Retains retract rows; physical filters remain above this scan. +#[derive(Debug, Clone)] +pub(crate) struct PaimonAuditLogScan { + inner: PaimonTableScan, +} + +impl PaimonAuditLogScan { + pub(crate) fn new(inner: PaimonTableScan) -> Self { + Self { inner } + } +} + +impl ExecutionPlan for PaimonAuditLogScan { + fn name(&self) -> &str { + "PaimonAuditLogScan" + } + + fn properties(&self) -> &Arc { + self.inner.properties() + } + + fn children(&self) -> Vec<&Arc> { + vec![] + } + + fn with_new_children( + self: Arc, + _children: Vec>, + ) -> DFResult> { + Ok(self) + } + + fn execute( + &self, + partition: usize, + _context: Arc, + ) -> DFResult { + self.inner.execute_with(partition, |read, splits| { + AuditLogRead::new(read)?.to_arrow(splits) + }) + } + + fn partition_statistics(&self, partition: Option) -> DFResult> { + let mut statistics = self.inner.partition_statistics(partition)?; + Arc::make_mut(&mut statistics).num_rows = Precision::Absent; + Ok(statistics) + } +} + +impl DisplayAs for PaimonAuditLogScan { + fn fmt_as( + &self, + t: datafusion::physical_plan::DisplayFormatType, + f: &mut std::fmt::Formatter, + ) -> std::fmt::Result { + write!(f, "{}: ", self.name())?; + self.inner.fmt_as(t, f) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::table::{datafusion_arrow_schema, PaimonScanBuilder}; + use datafusion::config::ConfigOptions; + use datafusion::logical_expr::Operator; + use datafusion::physical_expr::expressions::{lit, BinaryExpr, Column}; + use datafusion::physical_expr::PhysicalExpr; + use datafusion::physical_plan::filter_pushdown::{ + ChildFilterPushdownResult, ChildPushdownResult, FilterPushdownPhase, PushedDown, + }; + use paimon::catalog::Identifier; + use paimon::table::Table; + use paimon::DataSplitBuilder; + + fn first_row_audit_scan() -> PaimonAuditLogScan { + let file_io = paimon::io::FileIOBuilder::new("memory").build().unwrap(); + let schema = paimon::spec::Schema::builder() + .column( + "id", + paimon::spec::DataType::Int(paimon::spec::IntType::new()), + ) + .primary_key(["id"]) + .option("bucket", "1") + .option("merge-engine", "first-row") + .build() + .unwrap(); + let table = Table::new( + file_io, + Identifier::new("default", "first_row_audit"), + "memory:/first-row-audit".to_string(), + paimon::spec::TableSchema::new(0, &schema), + None, + ); + let split = |snapshot| { + DataSplitBuilder::new() + .with_snapshot(snapshot) + .with_partition(paimon::spec::BinaryRow::new(0)) + .with_bucket(0) + .with_bucket_path("memory:/first-row-audit/bucket-0".to_string()) + .with_total_buckets(1) + .with_data_files(vec![]) + .build() + .unwrap() + }; + let read_fields = paimon::table::AuditLogTable::new(table.clone()) + .fields() + .unwrap(); + let arrow_schema = datafusion_arrow_schema(&read_fields, true).unwrap(); + let plan = PaimonScanBuilder { + table: &table, + schema: &arrow_schema, + plan: paimon::table::Plan::new(vec![split(1), split(2)]), + scan_trace: None, + projection: None, + pushed_predicate: None, + limit: None, + target_partitions: 8, + filter_exact: false, + case_sensitive: true, + } + .build_scan(read_fields) + .unwrap(); + PaimonAuditLogScan::new(plan) + } + + #[test] + fn test_first_row_audit_distributes_independent_splits() { + let scan = first_row_audit_scan(); + + assert_eq!(scan.inner.planned_partitions().len(), 2); + assert!(scan + .inner + .planned_partitions() + .iter() + .all(|splits| splits.len() == 1)); + } + + #[test] + fn test_audit_physical_filters_remain_above_scan() { + let scan = first_row_audit_scan(); + let filters: Vec> = vec![ + Arc::new(BinaryExpr::new( + Arc::new(Column::new("id", 1)), + Operator::Gt, + lit(1_i32), + )), + Arc::new(BinaryExpr::new( + Arc::new(Column::new("rowkind", 0)), + Operator::Eq, + lit("-D"), + )), + ]; + let result = scan + .handle_child_pushdown_result( + FilterPushdownPhase::Post, + ChildPushdownResult { + parent_filters: filters + .into_iter() + .map(|filter| ChildFilterPushdownResult { + filter, + child_results: Vec::new(), + }) + .collect(), + self_filters: Vec::new(), + }, + &ConfigOptions::default(), + ) + .unwrap(); + + assert!(matches!( + result.filters.as_slice(), + [PushedDown::No, PushedDown::No] + )); + assert!(result.updated_node.is_none()); + assert_eq!( + scan.partition_statistics(None).unwrap().num_rows, + Precision::Absent + ); + } +} diff --git a/crates/integrations/datafusion/src/physical_plan/mod.rs b/crates/integrations/datafusion/src/physical_plan/mod.rs index 2d1905035..e0e3c8878 100644 --- a/crates/integrations/datafusion/src/physical_plan/mod.rs +++ b/crates/integrations/datafusion/src/physical_plan/mod.rs @@ -15,10 +15,12 @@ // specific language governing permissions and limitations // under the License. +mod audit_log; pub(crate) mod scan; mod search_score; pub(crate) mod sink; +pub(crate) use audit_log::PaimonAuditLogScan; pub use scan::PaimonTableScan; pub(crate) use search_score::{SearchScoreExec, SearchScoreOutputColumn}; pub use sink::PaimonDataSink; diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs b/crates/integrations/datafusion/src/physical_plan/scan.rs index d2fc56c38..432e3a2a0 100644 --- a/crates/integrations/datafusion/src/physical_plan/scan.rs +++ b/crates/integrations/datafusion/src/physical_plan/scan.rs @@ -53,7 +53,7 @@ use paimon::arrow::ParquetReadBudget; use paimon::spec::{ CoreOptions, DataField, Datum, MergeEngine, Predicate, PredicateBuilder, PredicateOperator, }; -use paimon::table::{ScanTrace, Table}; +use paimon::table::{ArrowRecordBatchStream, ScanTrace, Table, TableRead}; use paimon::DataSplit; use crate::error::to_datafusion_error; @@ -996,6 +996,92 @@ impl PaimonTableScan { .map(|(accumulator, field)| accumulator.finish(field.data_type(), exact_null_counts)) .collect() } + + pub(crate) fn execute_with( + &self, + partition: usize, + read_splits: impl FnOnce(TableRead<'_>, &[DataSplit]) -> paimon::Result + + Send + + 'static, + ) -> DFResult { + let splits = Arc::clone(self.planned_partitions.get(partition).ok_or_else(|| { + datafusion::error::DataFusionError::Internal(format!( + "PaimonTableScan: partition index {partition} out of range (total {})", + self.planned_partitions.len() + )) + })?); + + let table = self.table.clone(); + let schema = self.schema(); + let read_type = self.read_type.clone(); + let pushed_predicate = self.pushed_predicate.clone(); + let case_sensitive = self.case_sensitive; + let runtime_filters = self.runtime_filters.clone(); + let decoder_filters = self.decoder_filters.clone(); + let parquet_read_budget = Arc::clone(&self.parquet_read_budget); + + let fut = async move { + let mut read_builder = table.new_read_builder(); + let runtime_filter_plan = partition_runtime_decoder_filters( + &decoder_filters, + table.schema().fields(), + case_sensitive, + ); + let mut paimon_predicates = pushed_predicate.into_iter().collect::>(); + paimon_predicates.extend(runtime_filter_plan.paimon_predicates); + + read_builder.with_case_sensitive(case_sensitive); + read_builder.with_read_type(read_type); + if !paimon_predicates.is_empty() { + read_builder.with_filter(Predicate::and(paimon_predicates)); + } + read_builder.with_parquet_read_budget(parquet_read_budget); + + let mut read = read_builder.new_read().map_err(to_datafusion_error)?; + if !runtime_filter_plan.datafusion_filters.is_empty() { + let predicate = conjunction(runtime_filter_plan.datafusion_filters); + read = read.with_row_filter_factory(Arc::new(DataFusionRowFilterFactory::new( + predicate, + Arc::clone(&schema), + ))); + } + let stream = read_splits(read, &splits).map_err(to_datafusion_error)?; + let batch_schema = Arc::clone(&schema); + let stream = stream.map(move |result| { + let mut batch = result + .map_err(to_datafusion_error) + .and_then(|batch| to_datafusion_batch(batch, &batch_schema))?; + // The decoder hook is an optimization and may be unavailable + // for a file/path. Retain every original live expression as + // the exact fallback; evaluating it on decoder survivors is + // idempotent. + for filter in &runtime_filters { + let predicate = filter.evaluate(&batch)?.into_array(batch.num_rows())?; + let predicate = predicate + .as_any() + .downcast_ref::() + .ok_or_else(|| { + datafusion::error::DataFusionError::Execution(format!( + "Paimon runtime filter must return Boolean, got {}", + predicate.data_type() + )) + })?; + batch = filter_record_batch(&batch, predicate)?; + } + Ok(batch) + }); + + Ok::<_, datafusion::error::DataFusionError>(RecordBatchStreamAdapter::new( + schema, + Box::pin(stream), + )) + }; + + Ok(Box::pin(RecordBatchStreamAdapter::new( + self.schema(), + futures::stream::once(fut).try_flatten(), + ))) + } } impl ExecutionPlan for PaimonTableScan { @@ -1088,83 +1174,7 @@ impl ExecutionPlan for PaimonTableScan { partition: usize, _context: Arc, ) -> DFResult { - let splits = Arc::clone(self.planned_partitions.get(partition).ok_or_else(|| { - datafusion::error::DataFusionError::Internal(format!( - "PaimonTableScan: partition index {partition} out of range (total {})", - self.planned_partitions.len() - )) - })?); - - let table = self.table.clone(); - let schema = self.schema(); - let read_type = self.read_type.clone(); - let pushed_predicate = self.pushed_predicate.clone(); - let case_sensitive = self.case_sensitive; - let runtime_filters = self.runtime_filters.clone(); - let decoder_filters = self.decoder_filters.clone(); - let parquet_read_budget = Arc::clone(&self.parquet_read_budget); - - let fut = async move { - let mut read_builder = table.new_read_builder(); - let runtime_filter_plan = partition_runtime_decoder_filters( - &decoder_filters, - table.schema().fields(), - case_sensitive, - ); - let mut paimon_predicates = pushed_predicate.into_iter().collect::>(); - paimon_predicates.extend(runtime_filter_plan.paimon_predicates); - - read_builder.with_case_sensitive(case_sensitive); - read_builder.with_read_type(read_type); - if !paimon_predicates.is_empty() { - read_builder.with_filter(Predicate::and(paimon_predicates)); - } - read_builder.with_parquet_read_budget(parquet_read_budget); - - let mut read = read_builder.new_read().map_err(to_datafusion_error)?; - if !runtime_filter_plan.datafusion_filters.is_empty() { - let predicate = conjunction(runtime_filter_plan.datafusion_filters); - read = read.with_row_filter_factory(Arc::new(DataFusionRowFilterFactory::new( - predicate, - Arc::clone(&schema), - ))); - } - let stream = read.to_arrow(&splits).map_err(to_datafusion_error)?; - let batch_schema = Arc::clone(&schema); - let stream = stream.map(move |result| { - let mut batch = result - .map_err(to_datafusion_error) - .and_then(|batch| to_datafusion_batch(batch, &batch_schema))?; - // The decoder hook is an optimization and may be unavailable - // for a file/path. Retain every original live expression as - // the exact fallback; evaluating it on decoder survivors is - // idempotent. - for filter in &runtime_filters { - let predicate = filter.evaluate(&batch)?.into_array(batch.num_rows())?; - let predicate = predicate - .as_any() - .downcast_ref::() - .ok_or_else(|| { - datafusion::error::DataFusionError::Execution(format!( - "Paimon runtime filter must return Boolean, got {}", - predicate.data_type() - )) - })?; - batch = filter_record_batch(&batch, predicate)?; - } - Ok(batch) - }); - - Ok::<_, datafusion::error::DataFusionError>(RecordBatchStreamAdapter::new( - schema, - Box::pin(stream), - )) - }; - - Ok(Box::pin(RecordBatchStreamAdapter::new( - self.schema(), - futures::stream::once(fut).try_flatten(), - ))) + self.execute_with(partition, |read, splits| read.to_arrow(splits)) } fn partition_statistics(&self, partition: Option) -> DFResult> { diff --git a/crates/integrations/datafusion/src/system_tables/audit_log.rs b/crates/integrations/datafusion/src/system_tables/audit_log.rs new file mode 100644 index 000000000..b464afd4a --- /dev/null +++ b/crates/integrations/datafusion/src/system_tables/audit_log.rs @@ -0,0 +1,132 @@ +// 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 [AuditLogTable](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java). + +use std::sync::Arc; + +use async_trait::async_trait; +use datafusion::arrow::datatypes::SchemaRef; +use datafusion::catalog::Session; +use datafusion::datasource::{TableProvider, TableType}; +use datafusion::error::Result as DFResult; +use datafusion::logical_expr::{Expr, TableProviderFilterPushDown}; +use datafusion::physical_plan::ExecutionPlan; +use paimon::spec::DataField; +use paimon::table::{AuditLogTable as PaimonAuditLogTable, Table}; + +use crate::error::to_datafusion_error; +use crate::filter_pushdown::{analyze_filters, classify_filter_pushdown}; +use crate::physical_plan::PaimonAuditLogScan; +use crate::runtime::await_with_runtime; +use crate::table::{datafusion_arrow_schema, PaimonScanBuilder}; + +pub(super) fn build(table: Table) -> DFResult> { + let fields = PaimonAuditLogTable::new(table.clone()) + .fields() + .map_err(to_datafusion_error)?; + let schema = datafusion_arrow_schema(&fields, true)?; + Ok(Arc::new(AuditLogTable { + table, + fields, + schema, + })) +} + +#[derive(Debug)] +struct AuditLogTable { + table: Table, + fields: Vec, + schema: SchemaRef, +} + +#[async_trait] +impl TableProvider for AuditLogTable { + fn schema(&self) -> SchemaRef { + self.schema.clone() + } + + fn table_type(&self) -> TableType { + TableType::View + } + + async fn scan( + &self, + state: &dyn Session, + projection: Option<&Vec>, + filters: &[Expr], + limit: Option, + ) -> DFResult> { + let filter_analysis = analyze_filters(filters, self.table.schema().fields(), true); + let pushed_limit = limit.filter(|_| !filter_analysis.requires_residual); + let mut read_builder = self.table.new_read_builder(); + if let Some(indices) = projection { + read_builder.with_read_type( + indices + .iter() + .map(|&index| self.fields[index].clone()) + .filter(|field| { + !matches!( + field.id(), + paimon::spec::ROW_KIND_FIELD_ID + | paimon::spec::SEQUENCE_NUMBER_FIELD_ID + ) + }) + .collect(), + ); + } + if let Some(predicate) = filter_analysis.pushed_predicate.clone() { + read_builder.with_filter(predicate); + } + if let Some(limit) = pushed_limit { + read_builder.with_limit(limit); + } + let (plan, trace) = await_with_runtime(read_builder.new_audit_scan().plan_with_trace()) + .await + .map_err(to_datafusion_error)?; + + let scan = PaimonScanBuilder { + table: &self.table, + schema: &self.schema, + plan, + scan_trace: Some(trace), + projection, + pushed_predicate: filter_analysis.pushed_predicate, + limit: pushed_limit, + target_partitions: state.config_options().execution.target_partitions, + filter_exact: false, + case_sensitive: true, + } + .build_scan(self.fields.clone())?; + Ok(Arc::new(PaimonAuditLogScan::new(scan))) + } + + fn supports_filters_pushdown( + &self, + filters: &[&Expr], + ) -> DFResult> { + let read_builder = self.table.new_read_builder(); + Ok(filters + .iter() + .map(|filter| { + classify_filter_pushdown(filter, self.table.schema().fields(), true, |predicate| { + read_builder.is_exact_filter_pushdown(predicate) + }) + }) + .collect()) + } +} diff --git a/crates/integrations/datafusion/src/system_tables/mod.rs b/crates/integrations/datafusion/src/system_tables/mod.rs index c81b8403d..22fbc5089 100644 --- a/crates/integrations/datafusion/src/system_tables/mod.rs +++ b/crates/integrations/datafusion/src/system_tables/mod.rs @@ -20,6 +20,7 @@ //! Mirrors Java [SystemTableLoader](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java): //! `TABLES` maps each system-table name to its builder function. +use std::collections::HashMap; use std::sync::Arc; use datafusion::datasource::TableProvider; @@ -29,6 +30,7 @@ use paimon::table::Table; use crate::error::to_datafusion_error; +mod audit_log; mod branches; mod consumers; mod files; @@ -49,6 +51,7 @@ type Builder = fn(Table) -> DFResult>; // in `load` because it needs the catalog handle (for metastore-tracked audit // metadata via `Catalog::list_partitions`). const TABLES: &[(&str, Builder)] = &[ + ("audit_log", audit_log::build), ("branches", branches::build), ("consumers", consumers::build), ("files", files::build), @@ -63,6 +66,7 @@ const TABLES: &[(&str, Builder)] = &[ ]; const SYSTEM_TABLE_NAMES: &[&str] = &[ + "audit_log", "branches", "consumers", "files", @@ -133,6 +137,7 @@ pub(crate) fn provider_for_table( if !is_registered(system_name) { return Ok(None); } + crate::table_loader::ensure_paimon_served(&table, &identifier)?; // Fail closed: system tables expose file metadata the client can't authorize. paimon::spec::CoreOptions::new(table.schema().options()) .ensure_read_authorized() @@ -156,13 +161,29 @@ pub(crate) async fn load( database: String, object: ParsedObjectName, system_name: String, + dynamic_options: HashMap, ) -> DFResult>> { if !is_registered(&system_name) { return Ok(None); } + if system_name.eq_ignore_ascii_case("audit_log") + && paimon::spec::CoreOptions::new(&dynamic_options).table_read_sequence_number_enabled() + { + return Err(DataFusionError::Plan( + "table-read.sequence-number.enabled is not supported by dynamic options for $audit_log" + .to_string(), + )); + } + paimon::spec::CoreOptions::new(&dynamic_options) + .ensure_read_authorized() + .map_err(to_datafusion_error)?; let identifier = Identifier::new(database, object.table().to_string()); match catalog.get_table(&identifier).await { Ok(mut table) => { + crate::table_loader::ensure_paimon_served(&table, &identifier)?; + paimon::spec::CoreOptions::new(table.schema().options()) + .ensure_read_authorized() + .map_err(to_datafusion_error)?; if let Some(branch) = object.branch() { if !system_name.eq_ignore_ascii_case("branches") { table = table @@ -171,6 +192,12 @@ pub(crate) async fn load( .map_err(to_datafusion_error)?; } } + if system_name.eq_ignore_ascii_case("audit_log") && !dynamic_options.is_empty() { + table = table + .copy_with_time_travel(dynamic_options) + .await + .map_err(to_datafusion_error)?; + } provider_for_table(catalog, identifier, table, &system_name) } Err(paimon::Error::TableNotExist { .. }) => Err(DataFusionError::Plan(format!( @@ -209,6 +236,9 @@ mod tests { #[test] fn is_registered_is_case_insensitive() { + assert!(is_registered("audit_log")); + assert!(is_registered("Audit_Log")); + assert!(is_registered("AUDIT_LOG")); assert!(is_registered("options")); assert!(is_registered("Options")); assert!(is_registered("OPTIONS")); diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index 9683bf588..f7db6c3e2 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -356,6 +356,10 @@ impl PaimonScanBuilder<'_> { self, read_fields: Vec, ) -> DFResult> { + Ok(Arc::new(self.build_scan(read_fields)?)) + } + + pub(crate) fn build_scan(self, read_fields: Vec) -> DFResult { let (projected_schema, read_type) = if let Some(indices) = self.projection { let fields: Vec = indices .iter() @@ -381,7 +385,7 @@ impl PaimonScanBuilder<'_> { .collect() }; - Ok(Arc::new(PaimonTableScan::try_new( + PaimonTableScan::try_new( projected_schema, self.table.clone(), read_type, @@ -392,7 +396,7 @@ impl PaimonScanBuilder<'_> { self.scan_trace, None, self.case_sensitive, - )?)) + ) } } diff --git a/crates/integrations/datafusion/tests/system_tables.rs b/crates/integrations/datafusion/tests/system_tables.rs index 1a64ccb77..dae555f74 100644 --- a/crates/integrations/datafusion/tests/system_tables.rs +++ b/crates/integrations/datafusion/tests/system_tables.rs @@ -22,7 +22,8 @@ mod common; use std::sync::Arc; use datafusion::arrow::array::{ - Array, BooleanArray, Int32Array, Int64Array, ListArray, StringArray, TimestampMillisecondArray, + Array, BooleanArray, Int32Array, Int64Array, Int8Array, ListArray, StringArray, + TimestampMillisecondArray, }; use datafusion::arrow::datatypes::{DataType, Field, TimeUnit}; use datafusion::arrow::record_batch::RecordBatch; @@ -30,6 +31,8 @@ use paimon::catalog::Identifier; use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options}; use paimon_datafusion::SQLContext; +use common::string_value; + const FIXTURE_TABLE: &str = "test_tantivy_fulltext"; fn extract_test_warehouse() -> (tempfile::TempDir, String) { @@ -85,17 +88,26 @@ async fn query_error(ctx: &SQLContext, sql: &str) -> String { } #[tokio::test] -async fn test_query_auth_table_fails_closed() { +async fn test_query_auth_system_tables_fail_closed() { let (ctx, _catalog, _tmp) = create_context().await; run_sql( &ctx, - "CREATE TABLE paimon.default.qa (id INT) WITH ('query-auth.enabled' = 'true')", + "CREATE TABLE paimon.default.qa (id INT) WITH ( + 'query-auth.enabled' = 'true', + 's3.secret-key' = 'persisted-secret' + )", ) .await; - // Data reads and data-derived system tables must all fail closed. + // Rust cannot yet apply query-auth filters or masks to table and system-table + // data. Metadata paths, including persisted options, must fail closed. for sql in [ "SELECT * FROM paimon.default.qa", + "SELECT * FROM paimon.default.qa$audit_log", + "SELECT * FROM paimon.default.qa$files", + "SELECT value FROM paimon.default.qa$options WHERE key = 's3.secret-key'", + "SELECT * FROM paimon.default.qa$schemas", + "SELECT * FROM paimon.default.qa$partitions", "SELECT * FROM paimon.default.qa$manifests", "SELECT * FROM paimon.default.qa$table_indexes", ] { @@ -105,6 +117,308 @@ async fn test_query_auth_table_fails_closed() { "`{sql}` should fail closed, got: {err}" ); } + + run_sql(&ctx, "CREATE TABLE paimon.default.qa_dynamic (id INT)").await; + run_sql(&ctx, "SET 'paimon.query-auth.enabled' = 'true'").await; + for sql in [ + "SELECT * FROM paimon.default.qa_dynamic", + "SELECT * FROM paimon.default.qa_dynamic$audit_log", + "SELECT * FROM paimon.default.qa_dynamic$files", + "SELECT * FROM paimon.default.qa_dynamic$options", + "SELECT * FROM paimon.default.qa_dynamic$schemas", + "SELECT * FROM paimon.default.qa_dynamic$partitions", + "SELECT * FROM paimon.default.qa_dynamic$manifests", + "SELECT * FROM paimon.default.qa_dynamic$table_indexes", + ] { + let err = query_error(&ctx, sql).await; + assert!( + err.contains("query-auth.enabled"), + "dynamic auth should make `{sql}` fail closed, got: {err}" + ); + } + run_sql(&ctx, "RESET 'paimon.query-auth.enabled'").await; + + run_sql(&ctx, "SET 'paimon.s3.secret-key' = 'session-secret'").await; + let batches = run_sql( + &ctx, + "SELECT value FROM paimon.default.qa_dynamic$options \ + WHERE key = 's3.secret-key'", + ) + .await; + assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 0); + run_sql(&ctx, "RESET 'paimon.s3.secret-key'").await; +} + +#[tokio::test] +async fn test_audit_log_rejects_dynamic_sequence_number_option() { + let (ctx, _catalog, _tmp) = create_context().await; + run_sql( + &ctx, + "CREATE TABLE paimon.default.audit_dynamic_sequence ( + id INT NOT NULL, + PRIMARY KEY (id) + ) WITH ('bucket' = '1')", + ) + .await; + + run_sql( + &ctx, + "SET 'paimon.table-read.sequence-number.enabled' = 'true'", + ) + .await; + let err = query_error( + &ctx, + "SELECT * FROM paimon.default.audit_dynamic_sequence$audit_log", + ) + .await; + assert!( + err.contains("table-read.sequence-number.enabled") + && err.contains("not supported by dynamic options"), + "unexpected error: {err}" + ); + run_sql(&ctx, "RESET 'paimon.table-read.sequence-number.enabled'").await; +} + +#[tokio::test] +async fn test_audit_log_respects_dynamic_time_travel() { + let (ctx, _catalog, _tmp) = create_context().await; + run_sql(&ctx, "CREATE TABLE paimon.default.audit_tt (id INT)").await; + run_sql(&ctx, "INSERT INTO paimon.default.audit_tt VALUES (1)").await; + run_sql(&ctx, "INSERT INTO paimon.default.audit_tt VALUES (2)").await; + + run_sql(&ctx, "SET 'paimon.scan.version' = '1'").await; + let batches = run_sql( + &ctx, + "SELECT COUNT(*) FROM paimon.default.audit_tt$audit_log", + ) + .await; + assert_eq!( + batches[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 1 + ); + run_sql(&ctx, "RESET 'paimon.scan.version'").await; +} + +#[tokio::test] +async fn test_audit_log_system_table_keeps_row_kinds_and_sequence_numbers() { + let (ctx, catalog, _tmp) = create_context().await; + run_sql( + &ctx, + "CREATE TABLE paimon.default.audit_rows ( + id INT NOT NULL, + value INT, + PRIMARY KEY (id) + ) WITH ( + 'bucket' = '1', + 'merge-engine' = 'deduplicate', + 'changelog-producer' = 'input', + 'table-read.sequence-number.enabled' = 'true' + )", + ) + .await; + + let table = catalog + .get_table(&Identifier::new("default", "audit_rows")) + .await + .unwrap(); + let batch = RecordBatch::try_new( + Arc::new(datafusion::arrow::datatypes::Schema::new(vec![ + Field::new("id", DataType::Int32, false), + Field::new("value", DataType::Int32, true), + Field::new("_VALUE_KIND", DataType::Int8, false), + ])), + vec![ + Arc::new(Int32Array::from(vec![1, 2, 1, 2])), + Arc::new(Int32Array::from(vec![10, 20, 10, 25])), + Arc::new(Int8Array::from(vec![0, 0, 3, 2])), + ], + ) + .unwrap(); + let builder = table.new_write_builder(); + let mut write = builder.new_write().unwrap(); + write.write_arrow_batch(&batch).await.unwrap(); + let messages = write.prepare_commit().await.unwrap(); + builder.new_commit().commit(messages).await.unwrap(); + + let batches = run_sql( + &ctx, + "SELECT \"_SEQUENCE_NUMBER\", rowkind, id, value + FROM paimon.default.audit_rows$audit_log + WHERE rowkind = '-D' OR id = 2 + ORDER BY id", + ) + .await; + let mut rows = Vec::new(); + for batch in &batches { + let sequence = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let rowkind = batch.column(1); + let id = batch + .column(2) + .as_any() + .downcast_ref::() + .unwrap(); + let value = batch + .column(3) + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + rows.push(( + sequence.value(row), + string_value(rowkind.as_ref(), row).to_string(), + id.value(row), + value.value(row), + )); + } + } + assert_eq!( + rows, + vec![(2, "-D".to_string(), 1, 10), (3, "+U".to_string(), 2, 25)] + ); + + let batches = run_sql( + &ctx, + "SELECT id FROM paimon.default.audit_rows$audit_log WHERE value = 20", + ) + .await; + assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 0); + + let batches = run_sql( + &ctx, + "SELECT COUNT(*) FROM paimon.default.audit_rows$audit_log", + ) + .await; + assert_eq!( + batches[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 2 + ); + + run_sql(&ctx, "CREATE TABLE paimon.default.append_rows (id INT)").await; + run_sql( + &ctx, + "INSERT INTO paimon.default.append_rows VALUES (1), (2)", + ) + .await; + let batches = run_sql( + &ctx, + "SELECT rowkind FROM paimon.default.append_rows$audit_log", + ) + .await; + assert!(batches.iter().all(|batch| { + (0..batch.num_rows()).all(|row| string_value(batch.column(0).as_ref(), row) == "+I") + })); + assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 2); + + let explain = run_sql( + &ctx, + "EXPLAIN SELECT id FROM paimon.default.audit_rows$audit_log WHERE id = 2", + ) + .await; + assert!(explain.iter().any(|batch| { + (0..batch.num_rows()).any(|row| { + let plan = string_value(batch.column(1).as_ref(), row); + plan.contains("PaimonAuditLogScan") && plan.contains("predicate=") + }) + })); +} + +#[tokio::test] +async fn test_first_row_audit_log_merges_level_zero_before_filtering() { + let (ctx, _catalog, _tmp) = create_context().await; + run_sql( + &ctx, + "CREATE TABLE paimon.default.first_row_audit ( + id INT NOT NULL, + value INT, + PRIMARY KEY (id) + ) WITH ( + 'bucket' = '1', + 'merge-engine' = 'first-row' + )", + ) + .await; + run_sql( + &ctx, + "INSERT INTO paimon.default.first_row_audit VALUES (1, 10)", + ) + .await; + run_sql( + &ctx, + "INSERT INTO paimon.default.first_row_audit VALUES (1, 20)", + ) + .await; + + let batches = run_sql( + &ctx, + "SELECT value FROM paimon.default.first_row_audit$audit_log", + ) + .await; + assert_eq!( + batches[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values(), + &[10] + ); + + let batches = run_sql( + &ctx, + "SELECT value FROM paimon.default.first_row_audit$audit_log WHERE value = 20", + ) + .await; + assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 0); +} + +#[tokio::test] +async fn test_audit_log_system_table_matches_deletion_vector_visibility() { + let (ctx, _catalog, _tmp) = create_context().await; + run_sql( + &ctx, + "CREATE TABLE paimon.default.dv_audit (id INT NOT NULL) WITH ( + 'row-tracking.enabled' = 'true', + 'data-evolution.enabled' = 'true', + 'deletion-vectors.enabled' = 'true' + )", + ) + .await; + run_sql( + &ctx, + "INSERT INTO paimon.default.dv_audit (id) VALUES (1), (2)", + ) + .await; + run_sql(&ctx, "DELETE FROM paimon.default.dv_audit WHERE id = 1").await; + + let batches = run_sql( + &ctx, + "SELECT rowkind, id FROM paimon.default.dv_audit$audit_log", + ) + .await; + assert_eq!(string_value(batches[0].column(0).as_ref(), 0), "+I"); + assert_eq!( + batches[0] + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .values(), + &[2] + ); } #[tokio::test] diff --git a/crates/paimon/src/table/audit_log_table.rs b/crates/paimon/src/table/audit_log_table.rs index a6b6e5fe0..f3d604df3 100644 --- a/crates/paimon/src/table/audit_log_table.rs +++ b/crates/paimon/src/table/audit_log_table.rs @@ -16,12 +16,14 @@ // under the License. use super::incremental_scan::{IncrementalPlan, IncrementalScan, IncrementalScanMode}; -use super::{ArrowRecordBatchStream, Table}; +use super::{ArrowRecordBatchStream, AuditLogRead, AuditLogScan, Table}; use crate::spec::{ BigIntType, DataField, DataType, VarCharType, ROW_KIND_FIELD_ID, ROW_KIND_FIELD_NAME, SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, }; +pub(super) mod merge; + /// Wrapper that exposes table rows with a leading `rowkind` audit column. /// /// Incremental reads produce: @@ -80,6 +82,21 @@ impl AuditLogTable { IncrementalScan::for_table(&self.wrapped, mode, start_exclusive, end_inclusive) } + /// Plan a current-state audit read for [`Self::new_read`]. + pub fn new_scan(&self) -> AuditLogScan<'_> { + self.wrapped.new_read_builder().new_audit_scan() + } + + /// Creates an audit reader using the table's configured fields and options. + pub fn new_read(&self) -> crate::Result> { + AuditLogRead::new( + self.wrapped + .new_read_builder() + .with_read_type(self.fields()?) + .new_read()?, + ) + } + pub fn to_arrow(&self, plan: &IncrementalPlan) -> crate::Result { plan.validate()?; let read = self.wrapped.new_read_builder().new_read()?; diff --git a/crates/paimon/src/table/audit_log_table/merge.rs b/crates/paimon/src/table/audit_log_table/merge.rs new file mode 100644 index 000000000..801fa5cc7 --- /dev/null +++ b/crates/paimon/src/table/audit_log_table/merge.rs @@ -0,0 +1,189 @@ +// 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. + +//! Merge policies for current-state audit reads. + +use super::super::sort_merge::{ + compare_sequence_order, AggregateMergeFunction, BufferedBatch, MergeFunction, MergeResult, + MergeRow, PartialUpdateMergeFunction, +}; +use crate::arrow::build_target_arrow_schema; +use crate::spec::{ + CoreOptions, DataField, MergeEngine, RowKind, SEQUENCE_NUMBER_FIELD_ID, VALUE_KIND_FIELD_ID, +}; +use crate::table::kv_file_reader::KeyValueReadConfig; +use crate::Error; +use arrow_array::{Int64Array, Int8Array, RecordBatch}; +use arrow_schema::SchemaRef; +use std::sync::Arc; + +pub(in crate::table) fn new_merge_function( + config: &KeyValueReadConfig, + fields: &[DataField], +) -> crate::Result> { + let options = &config.table_options; + let engine = config.merge_engine; + if matches!(engine, MergeEngine::Deduplicate | MergeEngine::FirstRow) { + return Ok(Box::new(AuditKeyMergeFunction { + first_row: engine == MergeEngine::FirstRow, + ignore_delete: CoreOptions::new(options).ignore_delete(), + })); + } + let value_projection: Vec<_> = fields + .iter() + .enumerate() + .filter(|(_, field)| !matches!(field.id(), SEQUENCE_NUMBER_FIELD_ID | VALUE_KIND_FIELD_ID)) + .map(|(index, _)| index) + .collect(); + let value_fields: Vec<_> = value_projection + .iter() + .map(|&index| fields[index].clone()) + .collect(); + let inner: Box = match engine { + MergeEngine::PartialUpdate => Box::new(PartialUpdateMergeFunction::new_with_schema( + options, + &config.table_name, + &config.table_fields, + &value_fields, + &config.primary_keys, + )?), + MergeEngine::Aggregation => Box::new(AggregateMergeFunction::new( + options, + &config.table_name, + &value_fields, + &config.primary_keys, + &config.sequence_fields, + )?), + _ => unreachable!(), + }; + if value_projection.len() == fields.len() { + return Ok(inner); + } + Ok(Box::new(AuditValueMergeFunction { + inner, + value_projection, + value_schema: build_target_arrow_schema(&value_fields)?, + fields: fields.to_vec(), + })) +} + +// Reuse the ordinary value merge, then expose its logical INSERT and latest add sequence. +struct AuditValueMergeFunction { + inner: Box, + value_projection: Vec, + value_schema: SchemaRef, + fields: Vec, +} + +impl MergeFunction for AuditValueMergeFunction { + fn merge( + &self, + rows: &[MergeRow], + batch_buffer: &[BufferedBatch], + source_output_col_indices: &[usize], + output_schema: &SchemaRef, + ) -> crate::Result { + let indices: Vec<_> = self + .value_projection + .iter() + .map(|&index| source_output_col_indices[index]) + .collect(); + match self + .inner + .merge(rows, batch_buffer, &indices, &self.value_schema)? + { + MergeResult::MaterializedRow(batch) => { + let mut columns = batch.columns().to_vec(); + for (index, field) in self.fields.iter().enumerate() { + match field.id() { + SEQUENCE_NUMBER_FIELD_ID => { + let winner = rows + .iter() + .filter(|row| matches!(row.value_kind, 0 | 2)) + .max_by(|left, right| compare_sequence_order(left, right)) + .expect("materialized merge must contain an add row"); + columns.insert( + index, + Arc::new(Int64Array::from(vec![winner.sequence_number])), + ); + } + VALUE_KIND_FIELD_ID => { + columns.insert(index, Arc::new(Int8Array::from(vec![0]))) + } + _ => {} + } + } + Ok(MergeResult::MaterializedRow( + RecordBatch::try_new(output_schema.clone(), columns).map_err(|error| { + Error::UnexpectedError { + message: format!("Failed to build audit merge row: {error}"), + source: Some(Box::new(error)), + } + })?, + )) + } + result => Ok(result), + } + } +} + +/// Keep the winning physical row, including retracts for deduplicate tables. +struct AuditKeyMergeFunction { + first_row: bool, + ignore_delete: bool, +} + +impl MergeFunction for AuditKeyMergeFunction { + fn merge( + &self, + rows: &[MergeRow], + _batch_buffer: &[BufferedBatch], + _source_output_col_indices: &[usize], + _output_schema: &SchemaRef, + ) -> crate::Result { + let mut winner = None; + for row in rows { + if (self.first_row || self.ignore_delete) + && !RowKind::from_value(row.value_kind)?.is_add() + { + if self.ignore_delete { + continue; + } + return Err(Error::Unsupported { + message: "merge-engine=first-row does not support DELETE or UPDATE_BEFORE rows; set ignore-delete=true to ignore them".to_string(), + }); + } + if winner.is_none_or(|best| { + let order = compare_sequence_order(row, best); + if self.first_row { + order.is_lt() + } else { + order.is_ge() + } + }) { + winner = Some(row); + } + } + Ok(match winner { + Some(row) => MergeResult::SourceRow { + batch_idx: row.batch_idx, + row_idx: row.row_idx, + }, + None => MergeResult::Omit, + }) + } +} diff --git a/crates/paimon/src/table/audit_log_table/read.rs b/crates/paimon/src/table/audit_log_table/read.rs new file mode 100644 index 000000000..625d31700 --- /dev/null +++ b/crates/paimon/src/table/audit_log_table/read.rs @@ -0,0 +1,210 @@ +// 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. + +//! Current-state audit reads, including winning retract rows. + +use super::{ + audit_sequence_number_enabled, rowkind_array_from_column, PaimonTableRead, TableRead, + TableReadKind, MAX_MERGE_INPUT_STREAMS, +}; +use crate::arrow::build_target_arrow_schema; +use crate::spec::{ + DataField, DataType, MergeEngine, TinyIntType, ROW_KIND_FIELD_ID, SEQUENCE_NUMBER_FIELD_ID, + VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME, +}; +use crate::table::data_file_reader::DataFileReader; +use crate::table::kv_file_reader::{KeyValueFileReader, KeyValueReadConfig}; +use crate::table::ArrowRecordBatchStream; +use crate::DataSplit; +use arrow_array::{ArrayRef, RecordBatch, RecordBatchOptions, StringArray}; +use futures::{stream, StreamExt}; +use std::sync::Arc; + +/// Reads current-state audit rows using the supplied read's exact projection, +/// predicates and Parquet budget. Include `rowkind` in the projection to expose it. +#[derive(Debug, Clone)] +pub struct AuditLogRead<'a> { + read: PaimonTableRead<'a>, +} + +impl<'a> AuditLogRead<'a> { + pub fn new(read: TableRead<'a>) -> crate::Result { + read.ensure_query_auth_allowed()?; + match read.0 { + TableReadKind::Paimon(read) => Ok(Self { read }), + TableReadKind::Format(_) => Err(crate::Error::Unsupported { + message: "Format tables do not support audit log batch read".to_string(), + }), + } + } + + /// Reads splits planned by an audit scan, retaining winning retract rows. + pub fn to_arrow(&self, data_splits: &[DataSplit]) -> crate::Result { + let output_read_type = self.read.read_type.clone(); + if output_read_type + .iter() + .any(|field| field.id() == SEQUENCE_NUMBER_FIELD_ID) + && !audit_sequence_number_enabled(self.read.table) + { + return Err(crate::Error::DataInvalid { + message: "Audit read requested _SEQUENCE_NUMBER but table-read.sequence-number.enabled is false".to_string(), + source: None, + }); + } + let audit_schema = build_target_arrow_schema(&output_read_type)?; + let has_primary_keys = !self.read.table.schema().primary_keys().is_empty(); + let mut read_type: Vec<_> = output_read_type + .iter() + .filter(|field| field.id() != ROW_KIND_FIELD_ID) + .cloned() + .collect(); + if has_primary_keys + && output_read_type + .iter() + .any(|field| field.id() == ROW_KIND_FIELD_ID) + { + read_type.push(DataField::new( + VALUE_KIND_FIELD_ID, + VALUE_KIND_FIELD_NAME.to_string(), + DataType::TinyInt(TinyIntType::new()), + )); + } + + let physical_stream = if has_primary_keys { + let core_options = self.read.table.schema().core_options(); + let merge_engine = core_options.merge_engine()?; + let (raw_splits, merge_splits): (Vec<_>, Vec<_>) = data_splits + .iter() + .cloned() + .partition(|split| audit_raw_convertible(split, merge_engine)); + let parquet_read_budget = self.read.parquet_read_budget()?; + let raw_stream = DataFileReader::new( + self.read.table.file_io.clone(), + self.read.table.schema_manager().clone(), + self.read.table.schema().id(), + self.read.table.schema.fields().to_vec(), + read_type.clone(), + self.read.data_predicates.clone(), + ) + .with_file_index_read_enabled(core_options.file_index_read_enabled()) + .with_batch_size(Some(core_options.read_batch_size()?)) + .with_parquet_read_budget(Some(Arc::clone(&parquet_read_budget))) + .read(&raw_splits)?; + let merge_stream = KeyValueFileReader::new( + self.read.table.file_io.clone(), + KeyValueReadConfig { + table_name: self.read.table.identifier().full_name(), + table_options: self.read.table.schema().options().clone(), + schema_manager: self.read.table.schema_manager().clone(), + table_schema_id: self.read.table.schema().id(), + table_fields: self.read.table.schema.fields().to_vec(), + read_type, + predicates: self.read.data_predicates.clone(), + primary_keys: self.read.table.schema.trimmed_primary_keys(), + merge_engine, + sequence_fields: core_options + .sequence_fields() + .iter() + .map(|field| field.to_string()) + .collect(), + read_batch_size: core_options.read_batch_size()?, + merge_splits: false, + max_merge_input_streams: Some(MAX_MERGE_INPUT_STREAMS), + parquet_read_budget: Some(parquet_read_budget), + }, + ) + .read_with_merge_function( + &merge_splits, + crate::table::audit_log_table::merge::new_merge_function, + )?; + Box::pin(stream::select_all([raw_stream, merge_stream])) + } else { + let mut read = self.read.clone(); + read.read_type = read_type; + read.to_arrow(data_splits)? + }; + + Ok(Box::pin(async_stream::try_stream! { + futures::pin_mut!(physical_stream); + let mut projection = None; + while let Some(batch) = physical_stream.next().await { + let batch = batch?; + if projection.is_none() { + projection = Some(output_read_type.iter().map(|field| { + let name = if field.id() == ROW_KIND_FIELD_ID { + if !has_primary_keys { + return Ok(None); + } + VALUE_KIND_FIELD_NAME + } else { + field.name() + }; + batch.schema().index_of(name).map(Some).map_err(|error| crate::Error::DataInvalid { + message: format!("Audit read missing column '{name}': {error}"), + source: None, + }) + }).collect::>>()?); + } + let columns = output_read_type.iter().zip(projection.as_ref().unwrap()) + .map(|(field, index)| { + let column: ArrayRef = match index { + None => Arc::new(StringArray::from(vec!["+I"; batch.num_rows()])), + Some(index) if field.id() == ROW_KIND_FIELD_ID => + Arc::new(rowkind_array_from_column(batch.column(*index).as_ref())?), + Some(index) => batch.column(*index).clone(), + }; + Ok(column) + }).collect::>>()?; + let options = RecordBatchOptions::new().with_row_count(Some(batch.num_rows())); + yield RecordBatch::try_new_with_options(audit_schema.clone(), columns, &options) + .map_err(|error| crate::Error::UnexpectedError { + message: format!("Failed to build audit log batch: {error}"), + source: Some(Box::new(error)), + })?; + } + })) + } +} + +// Legacy unknown delete counts and first-row level-0 files stay on the merge path. +fn audit_raw_convertible(split: &DataSplit, merge_engine: MergeEngine) -> bool { + split.raw_convertible() + && split.data_files().iter().all(|file| { + file.delete_row_count == Some(0) + && (merge_engine != MergeEngine::FirstRow || file.level != 0) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::table::table_read::tests::{file, split}; + + #[test] + fn test_audit_split_routing() { + let raw = split(vec![file("a", 5, Some(0))], true); + let merge = split(vec![file("a", 5, Some(0))], false); + let legacy = split(vec![file("a", 5, None)], true); + assert!(audit_raw_convertible(&raw, MergeEngine::Deduplicate)); + assert!(audit_raw_convertible(&raw, MergeEngine::FirstRow)); + assert!(!audit_raw_convertible(&merge, MergeEngine::Deduplicate)); + assert!(!audit_raw_convertible(&legacy, MergeEngine::Deduplicate)); + let level_zero = split(vec![file("a", 0, Some(0))], true); + assert!(audit_raw_convertible(&level_zero, MergeEngine::Deduplicate)); + assert!(!audit_raw_convertible(&level_zero, MergeEngine::FirstRow)); + } +} diff --git a/crates/paimon/src/table/audit_log_table/scan.rs b/crates/paimon/src/table/audit_log_table/scan.rs new file mode 100644 index 000000000..fa14c0f09 --- /dev/null +++ b/crates/paimon/src/table/audit_log_table/scan.rs @@ -0,0 +1,265 @@ +// 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. + +//! Audit scan wrapper: retain row versions and only forward safe pruning. + +use super::{TableScan, TableScanKind}; +use crate::spec::MergeEngine; +use crate::table::kv_file_reader::retain_primary_key_conjuncts; +use crate::table::merge_tree_split_generator::{merge_tree_split_for_batch, KeyComparator}; +use crate::table::{DataSplit, Plan, ReadBuilder, ScanTrace}; +use std::collections::HashMap; + +/// Plans current-state audit reads over the wrapped table's files. +#[derive(Debug, Clone)] +pub struct AuditLogScan<'a> { + scan: TableScan<'a>, +} + +impl<'a> AuditLogScan<'a> { + fn new(mut scan: TableScan<'a>) -> Self { + if let TableScanKind::Paimon(inner) = &mut scan.0 { + // Preserve the read projection, including data-evolution column pruning. + inner.scan_all_files = true; + inner.limit = None; + inner.row_range_optimization_disabled = true; + inner.row_ranges = None; + if !inner.table.schema().primary_keys().is_empty() { + inner.data_predicates = retain_primary_key_conjuncts( + &inner.data_predicates, + inner.table.schema().fields(), + &inner.table.schema().trimmed_primary_keys(), + ); + } + } + Self { scan } + } + + pub async fn plan(&self) -> crate::Result { + self.plan_with_trace().await.map(|(plan, _)| plan) + } + + pub async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> { + let TableScanKind::Paimon(inner) = &self.scan.0 else { + return Err(crate::Error::Unsupported { + message: "Format tables do not support audit log batch scan".to_string(), + }); + }; + let (plan, mut trace) = self.scan.plan_with_trace().await?; + let options = inner.table.schema().core_options(); + let engine = options.merge_engine()?; + // Ordinary first-row and DV scans pack files without a key merge. Audit + // reads need the existing merge-tree planner to retain overlapping versions. + let plan = if engine == MergeEngine::FirstRow + || (options.deletion_vectors_enabled() && !options.deletion_vectors_merge_on_read()) + { + if let Some(comparator) = KeyComparator::from_table_schema(inner.table.schema()) { + let mut buckets: HashMap<(Vec, i32), Vec> = HashMap::new(); + for split in plan.into_splits() { + buckets + .entry((split.partition().to_serialized_bytes(), split.bucket())) + .or_default() + .push(split); + } + let mut splits = Vec::new(); + for bucket in buckets.into_values() { + let first = &bucket[0]; + let files = bucket + .iter() + .flat_map(|split| split.data_files().iter().cloned()) + .collect(); + let deletion_files: HashMap<_, _> = bucket + .iter() + .flat_map(|split| { + split.data_files().iter().filter_map(move |file| { + split + .deletion_file_for_data_file(file) + .map(|deletion| (file.file_name.clone(), deletion.clone())) + }) + }) + .collect(); + for group in merge_tree_split_for_batch( + files, + &comparator, + options.source_split_target_size(), + options.source_split_open_file_cost(), + matches!(engine, MergeEngine::Deduplicate | MergeEngine::FirstRow), + ) { + let mut builder = DataSplit::builder() + .with_snapshot(first.snapshot_id()) + .with_partition(first.partition().clone()) + .with_bucket(first.bucket()) + .with_bucket_path(first.bucket_path().to_string()) + .with_total_buckets(first.total_buckets()) + .with_raw_convertible(group.raw_convertible); + if !deletion_files.is_empty() { + builder = builder.with_data_deletion_files( + group + .files + .iter() + .map(|file| deletion_files.get(&file.file_name).cloned()) + .collect(), + ); + } + splits.push(builder.with_data_files(group.files).build()?); + } + } + Plan::new(splits) + } else { + plan + } + } else { + plan + }; + trace.record_final_plan( + plan.splits().len(), + plan.splits().len(), + plan.splits() + .iter() + .map(|split| split.data_files().len()) + .sum(), + ); + trace.planned_data_file_bytes = plan.planned_data_file_bytes(); + Ok((plan, trace)) + } +} + +impl<'a> ReadBuilder<'a> { + /// Create an audit scan that retains every visible row version. + pub fn new_audit_scan(&self) -> AuditLogScan<'a> { + AuditLogScan::new(self.new_scan()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::{ + BinaryRowBuilder, CoreOptions, DataField, DataType, Datum, IntType, PredicateBuilder, + }; + use crate::table::table_scan::{ + tests::{ + data_evolution_test_table, pk_stats_file, pk_stats_gate_table, setup_scan_trace_dirs, + two_column_schema, + }, + PaimonTableScan, + }; + use crate::table::{CommitMessage, TableCommit}; + use std::collections::HashSet; + #[test] + fn test_audit_scan_all_files_preserves_data_evolution_projection() { + let table = data_evolution_test_table( + "memory:/de_audit_scan_projection", + two_column_schema(0, "id", "name"), + ) + .copy_with_options(HashMap::from([( + "global-index.enabled".to_string(), + "true".to_string(), + )])); + let projected = HashSet::from([1]); + let predicate = PredicateBuilder::new(table.schema().fields()) + .equal("id", Datum::Int(1)) + .unwrap(); + let scan = PaimonTableScan::new(&table, None, vec![predicate], None, None, None) + .with_projected_read_field_ids(Some(projected.clone())); + let wrapped = AuditLogScan::new(TableScan(TableScanKind::Paimon(scan))); + let TableScanKind::Paimon(scan) = wrapped.scan.0 else { + panic!("expected Paimon scan") + }; + + assert!(scan.scan_all_files); + assert_eq!(scan.projected_read_field_ids, Some(projected)); + assert!( + scan.global_index_scan_settings(&CoreOptions::new(table.schema().options()), true,) + .unwrap() + .is_none(), + "audit scans must not prune physical row versions via global indexes" + ); + } + + #[tokio::test] + async fn test_audit_stats_pruning_keeps_overlapping_versions() { + for option in ["deletion-vectors.enabled", "merge-engine"] { + let table_path = format!("memory:/audit_stats_{option}"); + let value = if option == "merge-engine" { + "first-row" + } else { + "true" + }; + let table = pk_stats_gate_table(&table_path).copy_with_options(HashMap::from([ + (option.to_string(), value.to_string()), + ("source.split.target-size".to_string(), "1b".to_string()), + ])); + setup_scan_trace_dirs(&table).await; + + let mut old = pk_stats_file("old-version.parquet", (1, 5), (100, 200)); + old.level = 1; + let mut new = pk_stats_file("new-version.parquet", (1, 5), (10, 60)); + // Compacted files on the same level have disjoint key ranges. + new.level = 2; + TableCommit::new(table.clone(), "dv-audit-gate-test".to_string()) + .commit(vec![CommitMessage::new( + BinaryRowBuilder::new(0).build_serialized(), + 0, + vec![old, new], + )]) + .await + .unwrap(); + + let fields = vec![ + DataField::new(0, "id".to_string(), DataType::Int(IntType::new())), + DataField::new(1, "value".to_string(), DataType::Int(IntType::new())), + ]; + let value_filter = PredicateBuilder::new(&fields) + .greater_than("value", Datum::Int(90)) + .unwrap(); + let mut reader = table.new_read_builder(); + reader.with_filter(value_filter); + + let (ordinary_plan, ordinary_trace) = + reader.new_scan().plan_with_trace().await.unwrap(); + assert!(ordinary_trace.manifest_entries_pruned_by_data_stats >= 1); + assert_eq!( + ordinary_plan + .splits() + .iter() + .map(|split| split.data_files().len()) + .sum::(), + 1 + ); + + let (audit_plan, audit_trace) = + reader.new_audit_scan().plan_with_trace().await.unwrap(); + assert_eq!(audit_trace.manifest_entries_pruned_by_data_stats, 0); + assert_eq!( + audit_plan + .splits() + .iter() + .map(|split| split.data_files().len()) + .sum::(), + 2, + "both key versions must reach the audit merge path" + ); + assert_eq!( + audit_plan.splits().len(), + 1, + "overlapping versions must be planned together" + ); + assert!(!audit_plan.splits()[0].raw_convertible()); + } + } +} diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index 88b1bf806..3b0a31ce6 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -27,7 +27,7 @@ use super::data_file_reader::DataFileReader; use super::sort_merge::{ - AggregateMergeFunction, DeduplicateMergeFunction, PartialUpdateMergeFunction, + AggregateMergeFunction, DeduplicateMergeFunction, MergeFunction, PartialUpdateMergeFunction, SortMergeReaderBuilder, }; use crate::arrow::{build_target_arrow_schema, ParquetReadBudget}; @@ -283,39 +283,44 @@ impl KeyValueFileReader { } fn new_merge_function( - merge_engine: MergeEngine, - table_options: &HashMap, - table_name: &str, - table_fields: &[DataField], + config: &KeyValueReadConfig, merge_output_fields: &[DataField], - primary_keys: &[String], - sequence_fields: &[String], - ) -> crate::Result> { - match merge_engine { + ) -> crate::Result> { + match config.merge_engine { MergeEngine::Deduplicate => Ok(Box::new(DeduplicateMergeFunction)), - MergeEngine::PartialUpdate => Ok(Box::new( - PartialUpdateMergeFunction::new_with_schema( - table_options, - table_name, - table_fields, + MergeEngine::PartialUpdate => { + Ok(Box::new(PartialUpdateMergeFunction::new_with_schema( + &config.table_options, + &config.table_name, + &config.table_fields, merge_output_fields, - primary_keys, - )?, - )), + &config.primary_keys, + )?)) + } MergeEngine::FirstRow => Err(Error::Unsupported { message: "KeyValueFileReader does not support merge-engine=first-row; first-row reads should use the non-KV path".to_string(), }), MergeEngine::Aggregation => Ok(Box::new(AggregateMergeFunction::new( - table_options, - table_name, + &config.table_options, + &config.table_name, merge_output_fields, - primary_keys, - sequence_fields, + &config.primary_keys, + &config.sequence_fields, )?)), } } pub fn read(self, data_splits: &[DataSplit]) -> crate::Result { + self.read_with_merge_function(data_splits, Self::new_merge_function) + } + + pub(super) fn read_with_merge_function( + self, + data_splits: &[DataSplit], + merge_function: impl Fn(&KeyValueReadConfig, &[DataField]) -> crate::Result> + + Send + + 'static, + ) -> crate::Result { // A projected `_ROW_ID` is synthesized as all-nulls here, so the residual // would silently drop every row rather than hit its missing-column guard. super::row_id_predicate::reject_row_id_filter( @@ -371,10 +376,17 @@ impl KeyValueFileReader { )) }; // User columns = read_type fields + any key fields not already in read_type - // + any sequence fields not already included. + // + any sequence fields not already included. Physical system + // fields are already the first two columns of every KV file. let read_type_names: std::collections::HashSet<&str> = self.config.read_type.iter().map(|f| f.name()).collect(); - let mut user_fields: Vec = self.config.read_type.clone(); + let mut user_fields: Vec = self + .config + .read_type + .iter() + .filter(|field| !matches!(field.id(), SEQUENCE_NUMBER_FIELD_ID | VALUE_KIND_FIELD_ID)) + .cloned() + .collect(); for kf in &key_fields { if !read_type_names.contains(kf.name()) { user_fields.push(kf.clone()); @@ -447,17 +459,20 @@ impl KeyValueFileReader { .unwrap() }) .collect(); - let value_fields: Vec = user_fields - .iter() - .filter(|f| !key_names.contains(f.name())) - .cloned() - .collect(); - let value_indices: Vec = user_fields + let (value_indices, value_fields): (Vec<_>, Vec<_>) = internal_read_type .iter() .enumerate() - .filter(|(_, f)| !key_names.contains(f.name())) - .map(|(i, _)| i + 2) - .collect(); + .filter(|(index, field)| { + !key_names.contains(field.name()) + && (*index >= 2 + || self + .config + .read_type + .iter() + .any(|requested| requested.id() == field.id())) + }) + .map(|(index, field)| (index, field.clone())) + .unzip(); // If sequence.field is configured, find each field's index in the internal schema. let user_sequence_indices: Vec = self @@ -506,23 +521,13 @@ impl KeyValueFileReader { data_splits.into_iter().map(|split| vec![split]).collect() }; let file_io = self.file_io; - let merge_engine = self.config.merge_engine; - let schema_manager = self.config.schema_manager; - let table_schema_id = self.config.table_schema_id; - let table_fields = self.config.table_fields; - let table_name = self.config.table_name; - let table_options = self.config.table_options; + let config = self.config; + let table_schema_id = config.table_schema_id; let pushdown_predicates = self.pushdown_predicates; - let residual_predicates = self.config.predicates; - let primary_keys = self.config.primary_keys; - let sequence_fields = self.config.sequence_fields; - let read_batch_size = self.config.read_batch_size; - let max_merge_input_streams = self.config.max_merge_input_streams; - let parquet_read_budget = self.config.parquet_read_budget; #[cfg(test)] let input_batch_sizes = self.input_batch_sizes; - // Build the merge output schema (keys + values, no system columns). + // Build the merge output schema (keys + projected values). let mut merge_output_fields: Vec = Vec::new(); merge_output_fields.extend(key_fields); merge_output_fields.extend(value_fields); @@ -561,13 +566,13 @@ impl KeyValueFileReader { merge_splits, ) { let input_stream_count = merge_group.len(); - ensure_merge_input_limit(input_stream_count, max_merge_input_streams)?; + ensure_merge_input_limit(input_stream_count, config.max_merge_input_streams)?; // Sort-merge must first obtain one batch from every input // stream. Keep concurrent row-group reads disabled whenever // multiple runs advance in lockstep; one run may still use // the shared budget because its files are opened serially. let group_parquet_read_budget = if input_stream_count == 1 { - parquet_read_budget.clone() + config.parquet_read_budget.clone() } else { None }; @@ -576,15 +581,15 @@ impl KeyValueFileReader { for MergeRun { files } in merge_group { let reader = DataFileReader::new( file_io.clone(), - schema_manager.clone(), + config.schema_manager.clone(), table_schema_id, - table_fields.clone(), + config.table_fields.clone(), internal_read_type.clone(), pushdown_predicates.clone(), ) - .with_batch_size(Some(read_batch_size)) + .with_batch_size(Some(config.read_batch_size)) .with_parquet_read_budget(group_parquet_read_budget.clone()); - let run_schema_manager = schema_manager.clone(); + let run_schema_manager = config.schema_manager.clone(); let run_file_io = file_io.clone(); let deletion_files_by_split = deletion_files_by_split.clone(); let run_stream: ArrowRecordBatchStream = Box::pin(try_stream! { @@ -646,15 +651,7 @@ impl KeyValueFileReader { user_sequence_indices.clone(), value_indices.clone(), merge_output_schema.clone(), - Self::new_merge_function( - merge_engine, - &table_options, - &table_name, - &table_fields, - &merge_output_fields, - &primary_keys, - &sequence_fields, - )?, + merge_function(&config, &merge_output_fields)?, ) .build()?; @@ -668,13 +665,13 @@ impl KeyValueFileReader { // the merge-output batch (keys + values, including widened // predicate columns); the reorder below projects the output // back to read_type. - let batch = if residual_predicates.is_empty() { + let batch = if config.predicates.is_empty() { batch } else { match crate::arrow::residual::evaluate_predicates_mask( &batch, - &residual_predicates, - &table_fields, + &config.predicates, + &config.table_fields, &merge_output_fields, )? { Some(mask) => arrow_select::filter::filter_record_batch( diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index cc2e0c629..370e600df 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -171,8 +171,8 @@ pub use source::{ merge_row_ranges, DataSplit, DataSplitBuilder, DeletionFile, PartitionBucket, Plan, RowRange, }; pub use table_commit::TableCommit; -pub use table_read::TableRead; -pub use table_scan::TableScan; +pub use table_read::{AuditLogRead, TableRead}; +pub use table_scan::{AuditLogScan, TableScan}; pub use table_update::TableUpdate; pub use table_write::TableWrite; pub use tag_manager::TagManager; diff --git a/crates/paimon/src/table/sort_merge.rs b/crates/paimon/src/table/sort_merge.rs index a26197009..d1f2bdee8 100644 --- a/crates/paimon/src/table/sort_merge.rs +++ b/crates/paimon/src/table/sort_merge.rs @@ -141,7 +141,7 @@ pub(crate) trait MergeFunction: Send + Sync { /// Filters out DELETE and UPDATE_BEFORE rows. pub(crate) struct DeduplicateMergeFunction; -fn compare_sequence_order(lhs: &MergeRow, rhs: &MergeRow) -> Ordering { +pub(super) fn compare_sequence_order(lhs: &MergeRow, rhs: &MergeRow) -> Ordering { match (lhs.user_sequences.is_empty(), rhs.user_sequences.is_empty()) { (false, false) => lhs .user_sequences diff --git a/crates/paimon/src/table/table_read.rs b/crates/paimon/src/table/table_read.rs index acaebeeb1..b1926cb56 100644 --- a/crates/paimon/src/table/table_read.rs +++ b/crates/paimon/src/table/table_read.rs @@ -41,6 +41,10 @@ use futures::{stream, StreamExt}; use std::cmp::Ordering; use std::sync::Arc; +#[path = "audit_log_table/read.rs"] +mod audit; +pub use audit::AuditLogRead; + const MAX_MERGE_INPUT_STREAMS: usize = 256; /// Table read: reads data from splits (e.g. produced by [TableScan::plan]). @@ -1497,7 +1501,7 @@ mod tests { use crate::table::source::DataSplitBuilder; use futures::TryStreamExt; - fn file(name: &str, level: i32, delete_row_count: Option) -> DataFileMeta { + pub(super) fn file(name: &str, level: i32, delete_row_count: Option) -> DataFileMeta { DataFileMeta { file_name: name.to_string(), file_size: 128, @@ -1523,7 +1527,7 @@ mod tests { } } - fn split(files: Vec, raw_convertible: bool) -> DataSplit { + pub(super) fn split(files: Vec, raw_convertible: bool) -> DataSplit { DataSplitBuilder::new() .with_snapshot(1) .with_partition(BinaryRow::new(0)) @@ -1577,11 +1581,14 @@ mod tests { .to_vec() } - fn file_index_table(path: &str, enabled: Option) -> Table { + fn file_index_table(path: &str, enabled: Option, primary_key: bool) -> Table { let mut builder = Schema::builder().column("id", DataType::Int(IntType::new())); if let Some(enabled) = enabled { builder = builder.option("file-index.read.enabled", enabled.to_string()); } + if primary_key { + builder = builder.primary_key(["id"]).option("bucket", "1"); + } Table::new( FileIOBuilder::new("memory").build().unwrap(), Identifier::new("default", "file_index_t"), @@ -1597,7 +1604,7 @@ mod tests { indexed_file.row_count = 1; indexed_file.embedded_index = Some(embedded_bitmap_index().await); let split = split(vec![indexed_file], true); - let table = file_index_table("memory:/table_read_file_index", None); + let table = file_index_table("memory:/table_read_file_index", None, false); let fields = table.schema().fields().to_vec(); let predicate = PredicateBuilder::new(&fields) .equal("id", Datum::Int(99)) @@ -1631,8 +1638,24 @@ mod tests { .unwrap(); assert!(audit.is_empty()); + let pk_table = file_index_table("memory:/table_read_audit_file_index", None, true); + let pk_fields = pk_table.schema().fields().to_vec(); + let pk_predicate = PredicateBuilder::new(&pk_fields) + .equal("id", Datum::Int(99)) + .unwrap(); + let pk_read = TableRead::new(&pk_table, pk_fields, vec![pk_predicate]); + let splits = vec![split.clone()]; + let current_audit = AuditLogRead::new(pk_read) + .unwrap() + .to_arrow(&splits) + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert!(current_audit.is_empty()); + let disabled_table = - file_index_table("memory:/table_read_file_index_disabled", Some(false)); + file_index_table("memory:/table_read_file_index_disabled", Some(false), false); let disabled_fields = disabled_table.schema().fields().to_vec(); let disabled_predicate = PredicateBuilder::new(&disabled_fields) .equal("id", Datum::Int(99)) diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index 37139e58a..3835e4b8e 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -56,6 +56,10 @@ use indexmap::IndexMap; use std::collections::{HashMap, HashSet}; use std::sync::Arc; +#[path = "audit_log_table/scan.rs"] +mod audit; +pub use audit::AuditLogScan; + /// Path segment for manifest directory under table. const MANIFEST_DIR: &str = "manifest"; /// Path segment for index directory under table. @@ -2416,7 +2420,7 @@ mod tests { ); } - fn data_evolution_test_table(table_path: &str, schema: TableSchema) -> Table { + pub(super) fn data_evolution_test_table(table_path: &str, schema: TableSchema) -> Table { let file_io = FileIOBuilder::new("memory").build().unwrap(); let schema = schema.copy_with_options(HashMap::from([( "data-evolution.enabled".to_string(), @@ -2431,7 +2435,7 @@ mod tests { ) } - fn two_column_schema(id: i64, left: &str, right: &str) -> TableSchema { + pub(super) fn two_column_schema(id: i64, left: &str, right: &str) -> TableSchema { TableSchema::new( id, &PaimonSchema::builder() @@ -2632,7 +2636,7 @@ mod tests { ])) } - async fn setup_scan_trace_dirs(table: &Table) { + pub(super) async fn setup_scan_trace_dirs(table: &Table) { table .file_io() .mkdirs(&format!("{}/snapshot/", table.location())) @@ -3530,7 +3534,7 @@ mod tests { ); } - fn pk_stats_gate_table(table_path: &str) -> Table { + pub(super) fn pk_stats_gate_table(table_path: &str) -> Table { let file_io = FileIOBuilder::new("memory").build().unwrap(); let schema = PaimonSchema::builder() .column("id", DataType::Int(IntType::new())) @@ -3599,7 +3603,11 @@ mod tests { builder.build_serialized() } - fn pk_stats_file(name: &str, id_range: (i32, i32), value_range: (i32, i32)) -> DataFileMeta { + pub(super) fn pk_stats_file( + name: &str, + id_range: (i32, i32), + value_range: (i32, i32), + ) -> DataFileMeta { let mut file = test_data_file_meta( two_int_stats_row(Some(id_range.0), Some(value_range.0)), two_int_stats_row(Some(id_range.1), Some(value_range.1)), diff --git a/crates/paimon/tests/audit_log_table_test.rs b/crates/paimon/tests/audit_log_table_test.rs index 662ccaa4f..94a31d43b 100644 --- a/crates/paimon/tests/audit_log_table_test.rs +++ b/crates/paimon/tests/audit_log_table_test.rs @@ -20,8 +20,8 @@ mod common; use arrow_array::{Array, Int32Array, Int64Array, RecordBatch, StringArray}; use futures::TryStreamExt; use paimon::spec::{ - DataType, IntType, Schema, TableSchema, VarCharType, ROW_KIND_FIELD_ID, ROW_KIND_FIELD_NAME, - SEQUENCE_NUMBER_FIELD_NAME, + BigIntType, DataField, DataType, IntType, Schema, TableSchema, VarCharType, ROW_KIND_FIELD_ID, + ROW_KIND_FIELD_NAME, SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, }; use paimon::table::{AuditLogTable, IncrementalPlan, IncrementalScanMode, IncrementalSplit}; @@ -320,6 +320,101 @@ async fn audit_log_exposes_sequence_number_when_enabled() { assert!(rows.iter().all(|(_, seq, _, _)| *seq >= 0)); } +#[tokio::test] +async fn ordinary_read_projection_keeps_sequence_number() { + let table_path = "memory:/audit_log/ordinary_sequence_projection"; + let (file_io, table) = memory_table( + table_path, + pk_schema(&[ + ("merge-engine", "deduplicate"), + ("bucket", "1"), + ("table-read.sequence-number.enabled", "true"), + ]), + ); + setup_dirs(&file_io, table_path).await; + persist_table_schema(&file_io, table_path, table.schema()).await; + write_batch(&table, &make_batch(vec![1], vec![10])).await; + + let mut builder = table.new_read_builder(); + builder.with_read_type(vec![ + DataField::new( + SEQUENCE_NUMBER_FIELD_ID, + SEQUENCE_NUMBER_FIELD_NAME.to_string(), + DataType::BigInt(BigIntType::new()), + ), + table.schema().fields()[0].clone(), + ]); + let plan = builder.new_scan().plan().await.unwrap(); + let batches: Vec = builder + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + + assert_eq!( + batches[0] + .schema() + .fields() + .iter() + .map(|field| field.name().as_str()) + .collect::>(), + vec![SEQUENCE_NUMBER_FIELD_NAME, "id"] + ); + assert_eq!( + batches[0] + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 1 + ); +} + +#[tokio::test] +async fn audit_log_current_scan_keeps_delete_and_sequence_number() { + let table_path = "memory:/audit_log/current_state"; + let (file_io, table) = memory_table( + table_path, + pk_schema(&[ + ("changelog-producer", "input"), + ("merge-engine", "deduplicate"), + ("bucket", "1"), + ("table-read.sequence-number.enabled", "true"), + ]), + ); + setup_dirs(&file_io, table_path).await; + persist_table_schema(&file_io, table_path, table.schema()).await; + + write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await; + let builder = table.new_write_builder(); + let mut write = builder.new_write().unwrap(); + write + .write_arrow_batch(&make_batch_with_kinds(vec![1, 2], vec![10, 25], vec![3, 2])) + .await + .unwrap(); + let messages = write.prepare_commit().await.unwrap(); + builder.new_commit().commit(messages).await.unwrap(); + + let plan = table.new_read_builder().new_scan().plan().await.unwrap(); + let batches: Vec = AuditLogTable::new(table) + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + + assert_eq!( + collect_audit_rows_with_sequence(&batches), + vec![("+U".to_string(), 3, 2, 25), ("-D".to_string(), 2, 1, 10),] + ); +} + async fn audit_diff_rows( table: &paimon::table::Table, start: i64, @@ -356,6 +451,148 @@ fn assert_rows_exclude(rows: &[(String, i32, i32)], excluded: &[(&str, i32, i32) } } +#[tokio::test] +async fn audit_log_current_scan_uses_merged_rowkind() { + for merge_engine in ["partial-update", "aggregation"] { + let table_path = format!("memory:/audit_log/current_{merge_engine}"); + let (file_io, table) = memory_table( + &table_path, + pk_schema(&[ + ("merge-engine", merge_engine), + ("bucket", "1"), + ("table-read.sequence-number.enabled", "true"), + ]), + ); + setup_dirs(&file_io, &table_path).await; + persist_table_schema(&file_io, &table_path, table.schema()).await; + + write_batch(&table, &make_batch(vec![1], vec![10])).await; + let builder = table.new_write_builder(); + let mut write = builder.new_write().unwrap(); + write + .write_arrow_batch(&make_batch_with_kinds(vec![1], vec![20], vec![2])) + .await + .unwrap(); + let messages = write.prepare_commit().await.unwrap(); + builder.new_commit().commit(messages).await.unwrap(); + + let plan = table.new_read_builder().new_scan().plan().await.unwrap(); + let batches: Vec = AuditLogTable::new(table) + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!( + collect_audit_rows_with_sequence(&batches), + vec![("+I".to_string(), 1, 1, 20)], + "merge-engine={merge_engine}" + ); + } +} + +#[tokio::test] +async fn audit_log_current_scan_respects_ignore_delete() { + let table_path = "memory:/audit_log/current_ignore_delete"; + let (file_io, table) = memory_table( + table_path, + pk_schema(&[ + ("merge-engine", "deduplicate"), + ("ignore-delete", "true"), + ("bucket", "1"), + ]), + ); + setup_dirs(&file_io, table_path).await; + persist_table_schema(&file_io, table_path, table.schema()).await; + + write_batch(&table, &make_batch(vec![1], vec![10])).await; + let builder = table.new_write_builder(); + let mut write = builder.new_write().unwrap(); + write + .write_arrow_batch(&make_batch_with_kinds(vec![1], vec![10], vec![3])) + .await + .unwrap(); + let messages = write.prepare_commit().await.unwrap(); + builder.new_commit().commit(messages).await.unwrap(); + + let plan = table.new_read_builder().new_scan().plan().await.unwrap(); + let batches: Vec = AuditLogTable::new(table) + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + + assert_eq!( + collect_audit_rows(&batches), + vec![("+I".to_string(), 1, 10)] + ); +} + +#[tokio::test] +async fn audit_log_current_scan_supports_first_row() { + let table_path = "memory:/audit_log/current_first_row"; + let (file_io, table) = memory_table( + table_path, + pk_schema(&[ + ("merge-engine", "first-row"), + ("bucket", "1"), + ("source.split.target-size", "1b"), + ("source.split.open-file-cost", "1b"), + ]), + ); + setup_dirs(&file_io, table_path).await; + persist_table_schema(&file_io, table_path, table.schema()).await; + + write_batch(&table, &make_batch(vec![1], vec![10])).await; + write_batch(&table, &make_batch(vec![3], vec![30])).await; + write_batch(&table, &make_batch(vec![1], vec![20])).await; + + let mut limited_reader = table.new_read_builder(); + limited_reader.with_limit(1); + let limited_plan = limited_reader.new_audit_scan().plan().await.unwrap(); + assert_eq!( + limited_plan.splits().len(), + 2, + "audit LIMIT must not discard files needed to resolve row versions" + ); + assert_eq!( + limited_plan + .splits() + .iter() + .map(|split| split.data_files().len()) + .sum::(), + 3, + "audit LIMIT must retain every physical row version" + ); + + let audit = AuditLogTable::new(table); + let plan = audit.new_scan().plan().await.unwrap(); + assert_eq!(plan.splits().len(), 2); + let mut rows = Vec::new(); + for split in plan.splits() { + let batches: Vec = audit + .new_read() + .unwrap() + .to_arrow(std::slice::from_ref(split)) + .unwrap() + .try_collect() + .await + .unwrap(); + rows.extend(collect_audit_rows(&batches)); + } + rows.sort_unstable(); + + assert_eq!( + rows, + vec![("+I".to_string(), 1, 10), ("+I".to_string(), 3, 30)] + ); +} + #[tokio::test] async fn audit_log_diff_scan_emits_row_level_delete_insert_and_updates() { let table_path = "memory:/audit_log/diff_range"; diff --git a/docs/src/sql.md b/docs/src/sql.md index 8dc81360f..276e6c23d 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -1986,7 +1986,22 @@ let df = ctx.sql("SELECT * FROM paimon.my_db.table_a JOIN paimon.my_db.table_b O ## System Tables -Access table metadata via the `$` syntax. +Access table metadata and audit rows via the `$` syntax. + +### $audit_log + +Read the current table state with each row's Paimon row kind (`+I`, `-U`, `+U`, or `-D`): + +```sql +SELECT * FROM paimon.default.my_table$audit_log; +``` + +`rowkind` is the first column, followed by the table columns. Append-only rows are +reported as `+I`; deduplicate primary-key reads retain the latest physical retract row +instead of dropping it. Other primary-key merge engines retain or reject retracts +according to their merge-engine options. As in Paimon Java, rows masked by deletion +vectors are not reconstructed as retract records. When +`table-read.sequence-number.enabled=true`, `_SEQUENCE_NUMBER` appears after `rowkind`. ### $options