This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new a089a631 feat(datafusion): support the $audit_log system table (#787)
a089a631 is described below
commit a089a63198b99e9663feed5a78abe1119a9efb57
Author: jerry <[email protected]>
AuthorDate: Tue Sep 15 13:37:59 2026 +0800
feat(datafusion): support the $audit_log system table (#787)
---
crates/integrations/datafusion/src/catalog.rs | 6 +
.../datafusion/src/physical_plan/audit_log.rs | 210 ++++++++++++++
.../datafusion/src/physical_plan/mod.rs | 2 +
.../datafusion/src/physical_plan/scan.rs | 166 ++++++-----
.../datafusion/src/system_tables/audit_log.rs | 132 +++++++++
.../datafusion/src/system_tables/mod.rs | 30 ++
crates/integrations/datafusion/src/table/mod.rs | 8 +-
.../integrations/datafusion/tests/system_tables.rs | 322 ++++++++++++++++++++-
crates/paimon/src/table/audit_log_table.rs | 19 +-
crates/paimon/src/table/audit_log_table/merge.rs | 189 ++++++++++++
crates/paimon/src/table/audit_log_table/read.rs | 210 ++++++++++++++
crates/paimon/src/table/audit_log_table/scan.rs | 265 +++++++++++++++++
crates/paimon/src/table/kv_file_reader.rs | 123 ++++----
crates/paimon/src/table/mod.rs | 4 +-
crates/paimon/src/table/sort_merge.rs | 2 +-
crates/paimon/src/table/table_read.rs | 33 ++-
crates/paimon/src/table/table_scan.rs | 18 +-
crates/paimon/tests/audit_log_table_test.rs | 241 ++++++++++++++-
docs/src/sql.md | 17 +-
19 files changed, 1833 insertions(+), 164 deletions(-)
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index 340cf07c..06f34c3c 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 00000000..5fdfeccb
--- /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<PlanProperties> {
+ self.inner.properties()
+ }
+
+ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan + 'static>> {
+ vec![]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ _children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> DFResult<Arc<dyn ExecutionPlan>> {
+ Ok(self)
+ }
+
+ fn execute(
+ &self,
+ partition: usize,
+ _context: Arc<TaskContext>,
+ ) -> DFResult<SendableRecordBatchStream> {
+ self.inner.execute_with(partition, |read, splits| {
+ AuditLogRead::new(read)?.to_arrow(splits)
+ })
+ }
+
+ fn partition_statistics(&self, partition: Option<usize>) ->
DFResult<Arc<Statistics>> {
+ 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<Arc<dyn PhysicalExpr>> = 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 2d190503..e0e3c887 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 d2fc56c3..432e3a2a 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<ArrowRecordBatchStream>
+ + Send
+ + 'static,
+ ) -> DFResult<SendableRecordBatchStream> {
+ 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::<Vec<_>>();
+ 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::<BooleanArray>()
+ .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<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- 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::<Vec<_>>();
- 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::<BooleanArray>()
- .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<usize>) ->
DFResult<Arc<Statistics>> {
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 00000000..b464afd4
--- /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<Arc<dyn TableProvider>> {
+ 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<DataField>,
+ 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<usize>>,
+ filters: &[Expr],
+ limit: Option<usize>,
+ ) -> DFResult<Arc<dyn ExecutionPlan>> {
+ 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<Vec<TableProviderFilterPushDown>> {
+ 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 c81b8403..22fbc508 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<Arc<dyn TableProvider>>;
// 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<String, String>,
) -> DFResult<Option<Arc<dyn TableProvider>>> {
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 9683bf58..f7db6c3e 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<DataField>,
) -> DFResult<Arc<dyn ExecutionPlan>> {
+ Ok(Arc::new(self.build_scan(read_fields)?))
+ }
+
+ pub(crate) fn build_scan(self, read_fields: Vec<DataField>) ->
DFResult<PaimonTableScan> {
let (projected_schema, read_type) = if let Some(indices) =
self.projection {
let fields: Vec<Field> = 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 1a64ccb7..dae555f7 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::<usize>(), 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::<Int64Array>()
+ .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::<Int64Array>()
+ .unwrap();
+ let rowkind = batch.column(1);
+ let id = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let value = batch
+ .column(3)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .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::<usize>(), 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::<Int64Array>()
+ .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::<usize>(), 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::<Int32Array>()
+ .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::<usize>(), 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::<Int32Array>()
+ .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 a6b6e5fe..f3d604df 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<'_>> {
+ AuditLogRead::new(
+ self.wrapped
+ .new_read_builder()
+ .with_read_type(self.fields()?)
+ .new_read()?,
+ )
+ }
+
pub fn to_arrow(&self, plan: &IncrementalPlan) ->
crate::Result<ArrowRecordBatchStream> {
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 00000000..801fa5cc
--- /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<Box<dyn MergeFunction>> {
+ 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<dyn MergeFunction> = 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<dyn MergeFunction>,
+ value_projection: Vec<usize>,
+ value_schema: SchemaRef,
+ fields: Vec<DataField>,
+}
+
+impl MergeFunction for AuditValueMergeFunction {
+ fn merge(
+ &self,
+ rows: &[MergeRow],
+ batch_buffer: &[BufferedBatch],
+ source_output_col_indices: &[usize],
+ output_schema: &SchemaRef,
+ ) -> crate::Result<MergeResult> {
+ 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<MergeResult> {
+ 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 00000000..625d3170
--- /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<Self> {
+ 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<ArrowRecordBatchStream> {
+ 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::<crate::Result<Vec<_>>>()?);
+ }
+ 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::<crate::Result<Vec<_>>>()?;
+ 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 00000000..fa14c0f0
--- /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<Plan> {
+ 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<u8>, i32), Vec<DataSplit>> =
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::<usize>(),
+ 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::<usize>(),
+ 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 88b1bf80..3b0a31ce 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<String, String>,
- table_name: &str,
- table_fields: &[DataField],
+ config: &KeyValueReadConfig,
merge_output_fields: &[DataField],
- primary_keys: &[String],
- sequence_fields: &[String],
- ) -> crate::Result<Box<dyn super::sort_merge::MergeFunction>> {
- match merge_engine {
+ ) -> crate::Result<Box<dyn MergeFunction>> {
+ 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<ArrowRecordBatchStream> {
+ 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<Box<dyn MergeFunction>>
+ + Send
+ + 'static,
+ ) -> crate::Result<ArrowRecordBatchStream> {
// 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<DataField> = self.config.read_type.clone();
+ let mut user_fields: Vec<DataField> = 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<DataField> = user_fields
- .iter()
- .filter(|f| !key_names.contains(f.name()))
- .cloned()
- .collect();
- let value_indices: Vec<usize> = 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<usize> = 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<DataField> = 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 88d257e6..7d682ced 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -172,8 +172,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 a2619700..d1f2bdee 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 acaebeeb..b1926cb5 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<i64>) ->
DataFileMeta {
+ pub(super) fn file(name: &str, level: i32, delete_row_count: Option<i64>)
-> DataFileMeta {
DataFileMeta {
file_name: name.to_string(),
file_size: 128,
@@ -1523,7 +1527,7 @@ mod tests {
}
}
- fn split(files: Vec<DataFileMeta>, raw_convertible: bool) -> DataSplit {
+ pub(super) fn split(files: Vec<DataFileMeta>, 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<bool>) -> Table {
+ fn file_index_table(path: &str, enabled: Option<bool>, 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::<Vec<_>>()
+ .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 221c5920..502202d2 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -57,6 +57,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.
@@ -2585,7 +2589,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(),
@@ -2600,7 +2604,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()
@@ -2801,7 +2805,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()))
@@ -3879,7 +3883,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()))
@@ -3948,7 +3952,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 662ccaa4..94a31d43 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<RecordBatch> = 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<_>>(),
+ vec![SEQUENCE_NUMBER_FIELD_NAME, "id"]
+ );
+ assert_eq!(
+ batches[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .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<RecordBatch> = 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<RecordBatch> = 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<RecordBatch> = 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::<usize>(),
+ 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<RecordBatch> = 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 8dc81360..276e6c23 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
