| // 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. |
| |
| //! Paimon system tables end-to-end via DataFusion SQL. |
| |
| mod common; |
| |
| use std::sync::Arc; |
| |
| use datafusion::arrow::array::{ |
| Array, BooleanArray, Int32Array, Int64Array, ListArray, StringArray, |
| }; |
| use datafusion::arrow::datatypes::{DataType, Field, TimeUnit}; |
| use datafusion::arrow::record_batch::RecordBatch; |
| use paimon::catalog::Identifier; |
| use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options}; |
| use paimon_datafusion::SQLContext; |
| |
| const FIXTURE_TABLE: &str = "test_tantivy_fulltext"; |
| |
| fn extract_test_warehouse() -> (tempfile::TempDir, String) { |
| let archive_path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) |
| .join("testdata/test_tantivy_fulltext.tar.gz"); |
| let file = std::fs::File::open(&archive_path) |
| .unwrap_or_else(|e| panic!("Failed to open {}: {e}", archive_path.display())); |
| let decoder = flate2::read::GzDecoder::new(file); |
| let mut archive = tar::Archive::new(decoder); |
| |
| let tmp = tempfile::tempdir().expect("Failed to create temp dir"); |
| let db_dir = tmp.path().join("default.db"); |
| std::fs::create_dir_all(&db_dir).unwrap(); |
| archive.unpack(&db_dir).unwrap(); |
| |
| let warehouse = format!("file://{}", tmp.path().display()); |
| (tmp, warehouse) |
| } |
| |
| async fn create_context() -> (SQLContext, Arc<dyn Catalog>, tempfile::TempDir) { |
| let (tmp, warehouse) = extract_test_warehouse(); |
| let mut options = Options::new(); |
| options.set(CatalogOptions::WAREHOUSE, warehouse); |
| let catalog = FileSystemCatalog::new(options).expect("Failed to create catalog"); |
| let catalog: Arc<dyn Catalog> = Arc::new(catalog); |
| |
| let mut ctx = SQLContext::new(); |
| ctx.register_catalog("paimon", catalog.clone()) |
| .await |
| .expect("Failed to register catalog"); |
| (ctx, catalog, tmp) |
| } |
| |
| async fn run_sql(ctx: &SQLContext, sql: &str) -> Vec<RecordBatch> { |
| ctx.sql(sql) |
| .await |
| .unwrap_or_else(|e| panic!("Failed to plan `{sql}`: {e}")) |
| .collect() |
| .await |
| .unwrap_or_else(|e| panic!("Failed to execute `{sql}`: {e}")) |
| } |
| |
| #[tokio::test] |
| async fn test_options_system_table() { |
| let (ctx, catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT key, value FROM paimon.default.{FIXTURE_TABLE}$options"); |
| let batches = run_sql(&ctx, &sql).await; |
| |
| assert!(!batches.is_empty(), "$options should return ≥1 batch"); |
| let schema = batches[0].schema(); |
| assert_eq!(schema.field(0).name(), "key"); |
| assert_eq!(schema.field(1).name(), "value"); |
| |
| let mut actual: Vec<(String, String)> = Vec::new(); |
| for batch in &batches { |
| let keys = batch |
| .column(0) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("key column is Utf8"); |
| let values = batch |
| .column(1) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("value column is Utf8"); |
| for i in 0..batch.num_rows() { |
| actual.push((keys.value(i).to_string(), values.value(i).to_string())); |
| } |
| } |
| actual.sort(); |
| |
| let identifier = Identifier::new("default".to_string(), FIXTURE_TABLE.to_string()); |
| let table = catalog |
| .get_table(&identifier) |
| .await |
| .expect("fixture table should load"); |
| let mut expected: Vec<(String, String)> = table |
| .schema() |
| .options() |
| .iter() |
| .map(|(k, v)| (k.clone(), v.clone())) |
| .collect(); |
| expected.sort(); |
| |
| assert_eq!(actual, expected, "$options rows should match table options"); |
| } |
| |
| #[tokio::test] |
| async fn test_table_indexes_system_table() { |
| let (ctx, catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$table_indexes"); |
| let batches = run_sql(&ctx, &sql).await; |
| |
| assert!(!batches.is_empty(), "$table_indexes should return ≥1 batch"); |
| |
| let dv_meta_fields = vec![ |
| Arc::new(Field::new("f0", DataType::Utf8, false)), |
| Arc::new(Field::new("f1", DataType::Int32, false)), |
| Arc::new(Field::new("f2", DataType::Int32, false)), |
| Arc::new(Field::new("_CARDINALITY", DataType::Int64, true)), |
| ] |
| .into(); |
| let expected_columns = [ |
| ("partition", DataType::Utf8), |
| ("bucket", DataType::Int32), |
| ("index_type", DataType::Utf8), |
| ("file_name", DataType::Utf8), |
| ("file_size", DataType::Int64), |
| ("row_count", DataType::Int64), |
| ( |
| "dv_ranges", |
| DataType::List(Arc::new(Field::new( |
| "item", |
| DataType::Struct(dv_meta_fields), |
| true, |
| ))), |
| ), |
| ("row_range_start", DataType::Int64), |
| ("row_range_end", DataType::Int64), |
| ("index_field_id", DataType::Int32), |
| ("index_field_name", DataType::Utf8), |
| ]; |
| let arrow_schema = batches[0].schema(); |
| for (i, (name, dtype)) in expected_columns.iter().enumerate() { |
| let field = arrow_schema.field(i); |
| assert_eq!(field.name(), name, "column {i} name"); |
| assert_eq!(field.data_type(), dtype, "column {i} type"); |
| } |
| |
| let identifier = Identifier::new("default".to_string(), FIXTURE_TABLE.to_string()); |
| let table = catalog |
| .get_table(&identifier) |
| .await |
| .expect("fixture table should load"); |
| let sm = |
| paimon::table::SnapshotManager::new(table.file_io().clone(), table.location().to_string()); |
| let latest = sm |
| .get_latest_snapshot() |
| .await |
| .unwrap() |
| .expect("fixture has snapshots"); |
| let index_manifest = latest |
| .index_manifest() |
| .expect("fixture should have an index manifest"); |
| let expected_entries = |
| paimon::spec::IndexManifest::read(table.file_io(), &sm.manifest_path(index_manifest)) |
| .await |
| .unwrap() |
| .into_iter() |
| .filter(|entry| entry.kind == paimon::spec::FileKind::Add) |
| .collect::<Vec<_>>(); |
| |
| let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); |
| assert_eq!( |
| total_rows, |
| expected_entries.len(), |
| "$table_indexes rows should match the latest index manifest entries" |
| ); |
| |
| let batch = &batches[0]; |
| let partitions = batch |
| .column(0) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("partition is Utf8"); |
| let buckets = batch |
| .column(1) |
| .as_any() |
| .downcast_ref::<Int32Array>() |
| .expect("bucket is Int32"); |
| let index_types = batch |
| .column(2) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("index_type is Utf8"); |
| let file_names = batch |
| .column(3) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("file_name is Utf8"); |
| let file_sizes = batch |
| .column(4) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("file_size is Int64"); |
| let row_counts = batch |
| .column(5) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("row_count is Int64"); |
| let dv_ranges = batch |
| .column(6) |
| .as_any() |
| .downcast_ref::<ListArray>() |
| .expect("dv_ranges is ListArray"); |
| let row_range_starts = batch |
| .column(7) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("row_range_start is Int64"); |
| let row_range_ends = batch |
| .column(8) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("row_range_end is Int64"); |
| let index_field_ids = batch |
| .column(9) |
| .as_any() |
| .downcast_ref::<Int32Array>() |
| .expect("index_field_id is Int32"); |
| let index_field_names = batch |
| .column(10) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("index_field_name is Utf8"); |
| |
| for (row, expected) in expected_entries.iter().enumerate() { |
| assert!(!partitions.is_null(row), "partition should be non-null"); |
| assert_eq!(buckets.value(row), expected.bucket); |
| assert_eq!(index_types.value(row), expected.index_file.index_type); |
| assert_eq!(file_names.value(row), expected.index_file.file_name); |
| assert_eq!( |
| file_sizes.value(row), |
| i64::from(expected.index_file.file_size) |
| ); |
| assert_eq!( |
| row_counts.value(row), |
| i64::from(expected.index_file.row_count) |
| ); |
| assert_eq!( |
| dv_ranges.is_null(row), |
| expected.index_file.deletion_vectors_ranges.is_none() |
| ); |
| |
| if let Some(global_meta) = &expected.index_file.global_index_meta { |
| assert_eq!(row_range_starts.value(row), global_meta.row_range_start); |
| assert_eq!(row_range_ends.value(row), global_meta.row_range_end); |
| assert_eq!(index_field_ids.value(row), global_meta.index_field_id); |
| let expected_field_name = table |
| .schema() |
| .fields() |
| .iter() |
| .find(|field| field.id() == global_meta.index_field_id) |
| .map(|field| field.name()); |
| assert_eq!( |
| (!index_field_names.is_null(row)).then(|| index_field_names.value(row)), |
| expected_field_name |
| ); |
| } else { |
| assert!(row_range_starts.is_null(row)); |
| assert!(row_range_ends.is_null(row)); |
| assert!(index_field_ids.is_null(row)); |
| assert!(index_field_names.is_null(row)); |
| } |
| } |
| } |
| |
| #[tokio::test] |
| async fn test_unknown_system_table_name_returns_not_found() { |
| let (ctx, _catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$nonsense"); |
| let err = ctx |
| .sql(&sql) |
| .await |
| .expect_err("unknown system table should not resolve"); |
| let msg = err.to_string(); |
| assert!( |
| msg.contains("nonsense") || msg.to_lowercase().contains("not found"), |
| "unexpected error for unknown system table: {msg}" |
| ); |
| } |
| |
| #[tokio::test] |
| async fn test_schemas_system_table() { |
| let (ctx, catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$schemas"); |
| let batches = run_sql(&ctx, &sql).await; |
| |
| assert!(!batches.is_empty(), "$schemas should return ≥1 batch"); |
| |
| let arrow_schema = batches[0].schema(); |
| let expected_columns = [ |
| ("schema_id", DataType::Int64), |
| ("fields", DataType::Utf8), |
| ("partition_keys", DataType::Utf8), |
| ("primary_keys", DataType::Utf8), |
| ("options", DataType::Utf8), |
| ("comment", DataType::Utf8), |
| ( |
| "update_time", |
| DataType::Timestamp(TimeUnit::Millisecond, None), |
| ), |
| ]; |
| for (i, (name, dtype)) in expected_columns.iter().enumerate() { |
| let field = arrow_schema.field(i); |
| assert_eq!(field.name(), name, "column {i} name"); |
| assert_eq!(field.data_type(), dtype, "column {i} type"); |
| } |
| |
| let identifier = Identifier::new("default".to_string(), FIXTURE_TABLE.to_string()); |
| let table = catalog |
| .get_table(&identifier) |
| .await |
| .expect("fixture table should load"); |
| let all_schemas = table |
| .schema_manager() |
| .list_all() |
| .await |
| .expect("list_all should succeed"); |
| let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); |
| assert_eq!( |
| total_rows, |
| all_schemas.len(), |
| "$schemas rows should match list_all() length" |
| ); |
| |
| let mut ids: Vec<i64> = Vec::new(); |
| for batch in &batches { |
| let col = batch |
| .column(0) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("schema_id is Int64"); |
| for i in 0..batch.num_rows() { |
| ids.push(col.value(i)); |
| } |
| } |
| let mut sorted = ids.clone(); |
| sorted.sort_unstable(); |
| assert_eq!(ids, sorted, "schema_id column should be ascending"); |
| |
| // The last row's JSON columns must round-trip to the current schema. |
| let last_batch = batches.last().unwrap(); |
| let last_idx = last_batch.num_rows() - 1; |
| let latest = table.schema(); |
| let json_columns: [(usize, &str, String); 4] = [ |
| ( |
| 1, |
| "fields", |
| serde_json::to_string(latest.fields()).expect("serialize fields"), |
| ), |
| ( |
| 2, |
| "partition_keys", |
| serde_json::to_string(latest.partition_keys()).expect("serialize partition_keys"), |
| ), |
| ( |
| 3, |
| "primary_keys", |
| serde_json::to_string(latest.primary_keys()).expect("serialize primary_keys"), |
| ), |
| ( |
| 4, |
| "options", |
| serde_json::to_string(latest.options()).expect("serialize options"), |
| ), |
| ]; |
| for (col_idx, col_name, expected) in &json_columns { |
| let col = last_batch |
| .column(*col_idx) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .unwrap_or_else(|| panic!("column {col_name} is not Utf8")); |
| // Parse both sides before comparing: `options` is a HashMap whose |
| // JSON key order is non-deterministic across `HashMap` instances. |
| let actual: serde_json::Value = serde_json::from_str(col.value(last_idx)) |
| .unwrap_or_else(|e| panic!("parse actual `{col_name}`: {e}")); |
| let expected: serde_json::Value = serde_json::from_str(expected) |
| .unwrap_or_else(|e| panic!("parse expected `{col_name}`: {e}")); |
| assert_eq!( |
| actual, expected, |
| "latest-row `{col_name}` JSON should round-trip" |
| ); |
| } |
| } |
| |
| #[tokio::test] |
| async fn test_missing_base_table_for_system_table_errors() { |
| let (ctx, _catalog, _tmp) = create_context().await; |
| let sql = "SELECT * FROM paimon.default.does_not_exist$options"; |
| let err = ctx |
| .sql(sql) |
| .await |
| .expect_err("missing base table should error"); |
| let msg = err.to_string(); |
| assert!( |
| msg.contains("does_not_exist") && msg.contains("$options"), |
| "expected error to mention both base table and system name, got: {msg}" |
| ); |
| } |
| |
| #[tokio::test] |
| async fn test_snapshots_system_table() { |
| let (ctx, catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$snapshots"); |
| let batches = run_sql(&ctx, &sql).await; |
| |
| assert!(!batches.is_empty(), "$snapshots should return ≥1 batch"); |
| |
| let arrow_schema = batches[0].schema(); |
| let expected_columns = [ |
| ("snapshot_id", DataType::Int64), |
| ("schema_id", DataType::Int64), |
| ("commit_user", DataType::Utf8), |
| ("commit_identifier", DataType::Int64), |
| ("commit_kind", DataType::Utf8), |
| ( |
| "commit_time", |
| DataType::Timestamp(TimeUnit::Millisecond, None), |
| ), |
| ("base_manifest_list", DataType::Utf8), |
| ("delta_manifest_list", DataType::Utf8), |
| ("changelog_manifest_list", DataType::Utf8), |
| ("total_record_count", DataType::Int64), |
| ("delta_record_count", DataType::Int64), |
| ("changelog_record_count", DataType::Int64), |
| ("watermark", DataType::Int64), |
| ("next_row_id", DataType::Int64), |
| ]; |
| for (i, (name, dtype)) in expected_columns.iter().enumerate() { |
| let field = arrow_schema.field(i); |
| assert_eq!(field.name(), name, "column {i} name"); |
| assert_eq!(field.data_type(), dtype, "column {i} type"); |
| } |
| |
| // Row count must match the snapshot directory listing. |
| let identifier = Identifier::new("default".to_string(), FIXTURE_TABLE.to_string()); |
| let table = catalog |
| .get_table(&identifier) |
| .await |
| .expect("fixture table should load"); |
| let sm = |
| paimon::table::SnapshotManager::new(table.file_io().clone(), table.location().to_string()); |
| let all = sm.list_all().await.expect("list_all should succeed"); |
| let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); |
| assert_eq!( |
| total_rows, |
| all.len(), |
| "$snapshots rows should match list_all() length" |
| ); |
| |
| // snapshot_id column must be ascending. |
| let mut ids: Vec<i64> = Vec::new(); |
| for batch in &batches { |
| let col = batch |
| .column(0) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("snapshot_id is Int64"); |
| for i in 0..batch.num_rows() { |
| ids.push(col.value(i)); |
| } |
| } |
| let mut sorted = ids.clone(); |
| sorted.sort_unstable(); |
| assert_eq!(ids, sorted, "snapshot_id should be ascending"); |
| |
| // commit_kind column must contain a known variant. |
| let last_batch = batches.last().unwrap(); |
| let kind_col = last_batch |
| .column(4) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("commit_kind is Utf8"); |
| let kind = kind_col.value(last_batch.num_rows() - 1); |
| assert!( |
| ["APPEND", "COMPACT", "OVERWRITE", "ANALYZE"].contains(&kind), |
| "unexpected commit_kind: {kind}" |
| ); |
| } |
| |
| #[tokio::test] |
| async fn test_branches_system_table_empty_when_no_branch_dir() { |
| let (ctx, _catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$branches"); |
| let batches = run_sql(&ctx, &sql).await; |
| |
| // Schema must be present even with zero rows. |
| assert!(!batches.is_empty(), "$branches should return ≥1 batch"); |
| let arrow_schema = batches[0].schema(); |
| let expected_columns = [ |
| ("branch_name", DataType::Utf8), |
| ( |
| "create_time", |
| DataType::Timestamp(TimeUnit::Millisecond, None), |
| ), |
| ]; |
| for (i, (name, dtype)) in expected_columns.iter().enumerate() { |
| let field = arrow_schema.field(i); |
| assert_eq!(field.name(), name, "column {i} name"); |
| assert_eq!(field.data_type(), dtype, "column {i} type"); |
| } |
| let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); |
| assert_eq!(total_rows, 0, "fixture has no branch dir, expected 0 rows"); |
| } |
| |
| #[tokio::test] |
| async fn test_branches_system_table_with_seeded_branches() { |
| let (ctx, _catalog, tmp) = create_context().await; |
| |
| let table_dir = tmp.path().join("default.db").join(FIXTURE_TABLE); |
| let branch_dir = table_dir.join("branch"); |
| std::fs::create_dir_all(&branch_dir).expect("create branch dir"); |
| std::fs::create_dir_all(branch_dir.join("branch-b1")).unwrap(); |
| std::fs::create_dir_all(branch_dir.join("branch-b2")).unwrap(); |
| |
| let sql = format!("SELECT branch_name FROM paimon.default.{FIXTURE_TABLE}$branches"); |
| let batches = run_sql(&ctx, &sql).await; |
| let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); |
| assert_eq!(total_rows, 2, "expected two seeded branches"); |
| |
| let mut names: Vec<String> = Vec::new(); |
| for batch in &batches { |
| let names_col = batch |
| .column(0) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("branch_name is Utf8"); |
| for i in 0..batch.num_rows() { |
| names.push(names_col.value(i).to_string()); |
| } |
| } |
| let mut sorted_names = names.clone(); |
| sorted_names.sort(); |
| assert_eq!(names, sorted_names, "branch_name should be ascending"); |
| assert_eq!(names, vec!["b1".to_string(), "b2".to_string()]); |
| } |
| |
| #[tokio::test] |
| async fn test_tags_system_table_empty_when_no_tag_dir() { |
| let (ctx, _catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$tags"); |
| let batches = run_sql(&ctx, &sql).await; |
| |
| // Schema must be present even with zero rows. |
| assert!(!batches.is_empty(), "$tags should return ≥1 batch"); |
| let arrow_schema = batches[0].schema(); |
| let expected_columns = [ |
| ("tag_name", DataType::Utf8), |
| ("snapshot_id", DataType::Int64), |
| ("schema_id", DataType::Int64), |
| ( |
| "commit_time", |
| DataType::Timestamp(TimeUnit::Millisecond, None), |
| ), |
| ("record_count", DataType::Int64), |
| ( |
| "create_time", |
| DataType::Timestamp(TimeUnit::Millisecond, None), |
| ), |
| ("time_retained", DataType::Utf8), |
| ]; |
| for (i, (name, dtype)) in expected_columns.iter().enumerate() { |
| let field = arrow_schema.field(i); |
| assert_eq!(field.name(), name, "column {i} name"); |
| assert_eq!(field.data_type(), dtype, "column {i} type"); |
| } |
| let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); |
| assert_eq!(total_rows, 0, "fixture has no tag dir, expected 0 rows"); |
| } |
| |
| #[tokio::test] |
| async fn test_tags_system_table_with_seeded_tags() { |
| let (ctx, catalog, tmp) = create_context().await; |
| |
| let identifier = Identifier::new("default".to_string(), FIXTURE_TABLE.to_string()); |
| let table = catalog.get_table(&identifier).await.unwrap(); |
| let sm = |
| paimon::table::SnapshotManager::new(table.file_io().clone(), table.location().to_string()); |
| let earliest = sm.list_all().await.unwrap().into_iter().next().unwrap(); |
| |
| let table_dir = tmp.path().join("default.db").join(FIXTURE_TABLE); |
| let tag_dir = table_dir.join("tag"); |
| std::fs::create_dir_all(&tag_dir).expect("create tag dir"); |
| let src = table_dir |
| .join("snapshot") |
| .join(format!("snapshot-{}", earliest.id())); |
| std::fs::copy(&src, tag_dir.join("tag-v1")).unwrap(); |
| std::fs::copy(&src, tag_dir.join("tag-v2")).unwrap(); |
| |
| let sql = format!("SELECT tag_name, snapshot_id FROM paimon.default.{FIXTURE_TABLE}$tags"); |
| let batches = run_sql(&ctx, &sql).await; |
| let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); |
| assert_eq!(total_rows, 2, "expected two seeded tags"); |
| |
| let mut names: Vec<String> = Vec::new(); |
| let mut snap_ids: Vec<i64> = Vec::new(); |
| for batch in &batches { |
| let names_col = batch |
| .column(0) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("tag_name is Utf8"); |
| let snap_col = batch |
| .column(1) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("snapshot_id is Int64"); |
| for i in 0..batch.num_rows() { |
| names.push(names_col.value(i).to_string()); |
| snap_ids.push(snap_col.value(i)); |
| } |
| } |
| let mut sorted_names = names.clone(); |
| sorted_names.sort(); |
| assert_eq!(names, sorted_names, "tag_name should be ascending"); |
| assert_eq!(names, vec!["v1".to_string(), "v2".to_string()]); |
| assert_eq!(snap_ids, vec![earliest.id(), earliest.id()]); |
| } |
| |
| #[tokio::test] |
| async fn test_manifests_system_table() { |
| let (ctx, catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$manifests"); |
| let batches = run_sql(&ctx, &sql).await; |
| |
| assert!(!batches.is_empty(), "$manifests should return ≥1 batch"); |
| let arrow_schema = batches[0].schema(); |
| let expected_columns = [ |
| ("file_name", DataType::Utf8), |
| ("file_size", DataType::Int64), |
| ("num_added_files", DataType::Int64), |
| ("num_deleted_files", DataType::Int64), |
| ("schema_id", DataType::Int64), |
| ("min_partition_stats", DataType::Utf8), |
| ("max_partition_stats", DataType::Utf8), |
| ("min_row_id", DataType::Int64), |
| ("max_row_id", DataType::Int64), |
| ]; |
| for (i, (name, dtype)) in expected_columns.iter().enumerate() { |
| let field = arrow_schema.field(i); |
| assert_eq!(field.name(), name, "column {i} name"); |
| assert_eq!(field.data_type(), dtype, "column {i} type"); |
| } |
| |
| let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); |
| assert!(total_rows > 0, "fixture should have at least one manifest"); |
| |
| for batch in &batches { |
| let names = batch |
| .column(0) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("file_name is Utf8"); |
| let sizes = batch |
| .column(1) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("file_size is Int64"); |
| for i in 0..batch.num_rows() { |
| assert!(!names.value(i).is_empty(), "file_name must be non-empty"); |
| assert!(sizes.value(i) >= 0, "file_size must be non-negative"); |
| } |
| } |
| |
| let identifier = Identifier::new("default".to_string(), FIXTURE_TABLE.to_string()); |
| let table = catalog.get_table(&identifier).await.unwrap(); |
| let sm = |
| paimon::table::SnapshotManager::new(table.file_io().clone(), table.location().to_string()); |
| let latest = sm |
| .get_latest_snapshot() |
| .await |
| .unwrap() |
| .expect("fixture has snapshots"); |
| let mut expected = paimon::spec::ManifestList::read( |
| table.file_io(), |
| &sm.manifest_path(latest.base_manifest_list()), |
| ) |
| .await |
| .unwrap() |
| .len(); |
| expected += paimon::spec::ManifestList::read( |
| table.file_io(), |
| &sm.manifest_path(latest.delta_manifest_list()), |
| ) |
| .await |
| .unwrap() |
| .len(); |
| if let Some(changelog) = latest.changelog_manifest_list() { |
| expected += paimon::spec::ManifestList::read(table.file_io(), &sm.manifest_path(changelog)) |
| .await |
| .unwrap() |
| .len(); |
| } |
| assert_eq!( |
| total_rows, expected, |
| "$manifests rows should match base + delta + changelog manifest entries of the latest snapshot" |
| ); |
| } |
| |
| #[tokio::test] |
| async fn test_manifests_system_table_partition_stats() { |
| let (_tmp, sql_context) = common::setup_sql_context().await; |
| common::exec( |
| &sql_context, |
| "CREATE TABLE paimon.test_db.manifest_stats (id INT, pt INT) PARTITIONED BY (pt)", |
| ) |
| .await; |
| common::exec( |
| &sql_context, |
| "INSERT INTO paimon.test_db.manifest_stats VALUES (1, 1), (2, 2)", |
| ) |
| .await; |
| |
| let batches = sql_context |
| .sql( |
| "SELECT min_partition_stats, max_partition_stats \ |
| FROM paimon.test_db.manifest_stats$manifests", |
| ) |
| .await |
| .expect("$manifests query should plan") |
| .collect() |
| .await |
| .expect("$manifests query should execute"); |
| |
| let mut stats = Vec::new(); |
| for batch in &batches { |
| let mins = batch |
| .column(0) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("min_partition_stats is Utf8"); |
| let maxs = batch |
| .column(1) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("max_partition_stats is Utf8"); |
| for row in 0..batch.num_rows() { |
| stats.push(( |
| (!mins.is_null(row)).then(|| mins.value(row).to_string()), |
| (!maxs.is_null(row)).then(|| maxs.value(row).to_string()), |
| )); |
| } |
| } |
| stats.sort(); |
| |
| assert!( |
| !stats.is_empty(), |
| "$manifests should return partition stats" |
| ); |
| |
| let min_partition = stats |
| .iter() |
| .filter_map(|(min, _)| min.as_deref()) |
| .map(single_int_partition_stat) |
| .min(); |
| let max_partition = stats |
| .iter() |
| .filter_map(|(_, max)| max.as_deref()) |
| .map(single_int_partition_stat) |
| .max(); |
| |
| assert_eq!(min_partition, Some(1)); |
| assert_eq!(max_partition, Some(2)); |
| } |
| |
| #[tokio::test] |
| async fn test_files_system_table() { |
| let (ctx, catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$files"); |
| let batches = run_sql(&ctx, &sql).await; |
| |
| assert!( |
| !batches.is_empty(), |
| "$files should return at least one batch" |
| ); |
| let arrow_schema = batches[0].schema(); |
| let expected_columns = [ |
| ("partition", DataType::Utf8), |
| ("bucket", DataType::Int32), |
| ("file_path", DataType::Utf8), |
| ("file_format", DataType::Utf8), |
| ("schema_id", DataType::Int64), |
| ("level", DataType::Int32), |
| ("record_count", DataType::Int64), |
| ("file_size_in_bytes", DataType::Int64), |
| ("min_key", DataType::Utf8), |
| ("max_key", DataType::Utf8), |
| ("null_value_counts", DataType::Utf8), |
| ("min_value_stats", DataType::Utf8), |
| ("max_value_stats", DataType::Utf8), |
| ("min_sequence_number", DataType::Int64), |
| ("max_sequence_number", DataType::Int64), |
| ( |
| "creation_time", |
| DataType::Timestamp(TimeUnit::Millisecond, None), |
| ), |
| ("delete_row_count", DataType::Int64), |
| ("file_source", DataType::Utf8), |
| ("first_row_id", DataType::Int64), |
| ( |
| "write_cols", |
| DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), |
| ), |
| ]; |
| for (i, (name, dtype)) in expected_columns.iter().enumerate() { |
| let field = arrow_schema.field(i); |
| assert_eq!(field.name(), name, "column {i} name"); |
| assert_eq!(field.data_type(), dtype, "column {i} type"); |
| } |
| |
| let identifier = Identifier::new("default".to_string(), FIXTURE_TABLE.to_string()); |
| let table = catalog.get_table(&identifier).await.unwrap(); |
| let plan = table |
| .new_read_builder() |
| .new_scan() |
| .with_scan_all_files() |
| .plan() |
| .await |
| .unwrap(); |
| let expected_rows: usize = plan |
| .splits() |
| .iter() |
| .map(|split| split.data_files().len()) |
| .sum(); |
| let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); |
| assert_eq!(total_rows, expected_rows); |
| assert!(total_rows > 0, "fixture should contain data files"); |
| |
| for batch in &batches { |
| let paths = batch |
| .column(2) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("file_path is Utf8"); |
| let formats = batch |
| .column(3) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("file_format is Utf8"); |
| let null_counts = batch |
| .column(10) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("null_value_counts is Utf8"); |
| for i in 0..batch.num_rows() { |
| assert!(!paths.value(i).is_empty(), "file_path must be non-empty"); |
| assert!( |
| !formats.value(i).is_empty(), |
| "file_format must be non-empty" |
| ); |
| assert!( |
| null_counts.value(i).starts_with('{') && null_counts.value(i).ends_with('}'), |
| "null_value_counts should be Java-map-like" |
| ); |
| } |
| } |
| } |
| |
| fn single_int_partition_stat(value: &str) -> i32 { |
| value |
| .strip_prefix('{') |
| .and_then(|s| s.strip_suffix('}')) |
| .expect("partition stats should use row cast braces") |
| .parse() |
| .expect("partition stats should contain one int partition value") |
| } |
| |
| #[tokio::test] |
| async fn test_partitions_system_table() { |
| let (ctx, _catalog, _tmp) = create_context().await; |
| let sql = format!("SELECT * FROM paimon.default.{FIXTURE_TABLE}$partitions"); |
| let batches = run_sql(&ctx, &sql).await; |
| |
| assert!(!batches.is_empty(), "$partitions should return ≥1 batch"); |
| |
| let arrow_schema = batches[0].schema(); |
| let expected_columns = [ |
| ("partition", DataType::Utf8), |
| ("record_count", DataType::Int64), |
| ("file_size_in_bytes", DataType::Int64), |
| ("file_count", DataType::Int64), |
| ( |
| "last_update_time", |
| DataType::Timestamp(TimeUnit::Millisecond, None), |
| ), |
| ( |
| "created_at", |
| DataType::Timestamp(TimeUnit::Millisecond, None), |
| ), |
| ("created_by", DataType::Utf8), |
| ("updated_by", DataType::Utf8), |
| ("options", DataType::Utf8), |
| ("total_buckets", DataType::Int32), |
| ("done", DataType::Boolean), |
| ]; |
| for (i, (name, dtype)) in expected_columns.iter().enumerate() { |
| let field = arrow_schema.field(i); |
| assert_eq!(field.name(), name, "column {i} name"); |
| assert_eq!(field.data_type(), dtype, "column {i} type"); |
| } |
| |
| let mut partition_strings: Vec<Option<String>> = Vec::new(); |
| let mut record_counts: Vec<i64> = Vec::new(); |
| let mut file_counts: Vec<i64> = Vec::new(); |
| let mut file_sizes: Vec<i64> = Vec::new(); |
| for batch in &batches { |
| let parts = batch |
| .column(0) |
| .as_any() |
| .downcast_ref::<StringArray>() |
| .expect("partition is Utf8"); |
| let rc = batch |
| .column(1) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("record_count is Int64"); |
| let sz = batch |
| .column(2) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("file_size_in_bytes is Int64"); |
| let fc = batch |
| .column(3) |
| .as_any() |
| .downcast_ref::<Int64Array>() |
| .expect("file_count is Int64"); |
| for i in 0..batch.num_rows() { |
| partition_strings.push(if parts.is_null(i) { |
| None |
| } else { |
| Some(parts.value(i).to_string()) |
| }); |
| record_counts.push(rc.value(i)); |
| file_sizes.push(sz.value(i)); |
| file_counts.push(fc.value(i)); |
| } |
| } |
| |
| for fc in &file_counts { |
| assert!(*fc >= 1, "file_count must be ≥ 1 per partition"); |
| } |
| for sz in &file_sizes { |
| assert!(*sz >= 0, "file_size_in_bytes must be non-negative"); |
| } |
| for rc in &record_counts { |
| assert!(*rc >= 0, "record_count must be non-negative"); |
| } |
| |
| let mut sorted = partition_strings.clone(); |
| sorted.sort(); |
| assert_eq!( |
| partition_strings, sorted, |
| "rows should be sorted by partition string" |
| ); |
| |
| // Cols 5-8 stay null with FileSystemCatalog (no metastore audit metadata). |
| for batch in &batches { |
| for col_idx in 5..=8 { |
| for i in 0..batch.num_rows() { |
| assert!( |
| batch.column(col_idx).is_null(i), |
| "column {col_idx} row {i} should be null" |
| ); |
| } |
| } |
| } |
| // total_buckets defaults to 0 and done to false on FileSystemCatalog. |
| for batch in &batches { |
| let tb = batch |
| .column(9) |
| .as_any() |
| .downcast_ref::<Int32Array>() |
| .expect("total_buckets is Int32"); |
| let done = batch |
| .column(10) |
| .as_any() |
| .downcast_ref::<BooleanArray>() |
| .expect("done is Boolean"); |
| for i in 0..batch.num_rows() { |
| assert_eq!(tb.value(i), 0, "total_buckets default for FS catalog"); |
| assert!(!done.value(i), "done default for FS catalog"); |
| } |
| } |
| } |
