You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by gi...@apache.org on 2026/09/30 20:37:47 UTC
(iceberg-rust) branch gh-readonly-queue/main/pr-3107-71e571e531618371499eaf3c995e30c69910e2d9 created (now 36eae2969)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-3107-71e571e531618371499eaf3c995e30c69910e2d9
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
at 36eae2969 refactor(scan) move test utils to test_utils.rs (#3107)
This branch includes the following new commits:
new 36eae2969 refactor(scan) move test utils to test_utils.rs (#3107)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iceberg-rust) 01/01: refactor(scan) move test utils to test_utils.rs (#3107)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3107-71e571e531618371499eaf3c995e30c69910e2d9
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 36eae2969b0773ee5b6f23d9be230fd0a17ffe92
Author: Xander <za...@googlemail.com>
AuthorDate: Wed Sep 30 19:25:21 2026 +0000
refactor(scan) move test utils to test_utils.rs (#3107)
* refactor(scan) move test utils to test_utils.rs
* move things around
* move
* fix
* comments
---
Cargo.lock | 1 +
crates/catalog/rest/src/catalog.rs | 2 +-
crates/iceberg/Cargo.toml | 1 -
crates/iceberg/public-api.txt | 4 -
crates/iceberg/src/inspect/history.rs | 2 +-
crates/iceberg/src/inspect/manifests.rs | 2 +-
crates/iceberg/src/inspect/snapshots.rs | 2 +-
crates/iceberg/src/lib.rs | 3 +-
crates/iceberg/src/scan/mod.rs | 1157 +----------------------
crates/iceberg/src/test_utils.rs | 190 ----
crates/iceberg/src/test_utils/delete_vector.rs | 32 +
crates/iceberg/src/test_utils/encryption.rs | 94 ++
crates/iceberg/src/test_utils/mod.rs | 33 +
crates/iceberg/src/test_utils/record_batch.rs | 76 ++
crates/iceberg/src/test_utils/runtime.rs | 45 +
crates/iceberg/src/test_utils/scan.rs | 1160 ++++++++++++++++++++++++
crates/iceberg/src/util/snapshot.rs | 2 +-
crates/test_utils/Cargo.toml | 1 +
crates/test_utils/src/lib.rs | 5 +
crates/test_utils/src/runtime.rs | 40 +
20 files changed, 1503 insertions(+), 1349 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 64c5c6187..fcf5dd449 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2952,6 +2952,7 @@ name = "iceberg_test_utils"
version = "0.10.1"
dependencies = [
"iceberg",
+ "tokio",
"tracing-subscriber",
]
diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs
index b7d3b3163..cc5973316 100644
--- a/crates/catalog/rest/src/catalog.rs
+++ b/crates/catalog/rest/src/catalog.rs
@@ -1673,8 +1673,8 @@ mod tests {
SnapshotLog, SortDirection, SortField, SortOrder, Summary, Transform, Type,
UnboundPartitionField, UnboundPartitionSpec,
};
- use iceberg::test_utils::test_runtime;
use iceberg::transaction::{ApplyTransactionAction, Transaction};
+ use iceberg_test_utils::test_runtime;
use mockito::{Mock, Server, ServerGuard};
use serde_json::json;
use uuid::uuid;
diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml
index 18256d362..8783251fd 100644
--- a/crates/iceberg/Cargo.toml
+++ b/crates/iceberg/Cargo.toml
@@ -54,7 +54,6 @@ bytes = { workspace = true }
chrono = { workspace = true }
crc32fast = { workspace = true }
derive_builder = { workspace = true }
-expect-test = { workspace = true }
fastnum = { workspace = true }
flate2 = { workspace = true }
fnv = { workspace = true }
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index 58c7db7dd..da0b6b191 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -3229,10 +3229,6 @@ pub fn iceberg::table::TableBuilder::metadata<T: core::convert::Into<iceberg::sp
pub fn iceberg::table::TableBuilder::metadata_location<T: core::convert::Into<alloc::string::String>>(self, metadata_location: T) -> Self
pub fn iceberg::table::TableBuilder::readonly(self, readonly: bool) -> Self
pub fn iceberg::table::TableBuilder::runtime(self, runtime: iceberg::Runtime) -> Self
-pub mod iceberg::test_utils
-pub fn iceberg::test_utils::check_record_batches(record_batches: alloc::vec::Vec<arrow_array::record_batch::RecordBatch>, expected_schema: expect_test::Expect, expected_data: expect_test::Expect, ignore_check_columns: &[&str], sort_column: core::option::Option<&str>)
-pub async fn iceberg::test_utils::make_encrypted_table() -> iceberg::table::Table
-pub fn iceberg::test_utils::test_runtime() -> iceberg::Runtime
pub mod iceberg::transaction
pub struct iceberg::transaction::ActionCommit
impl iceberg::transaction::ActionCommit
diff --git a/crates/iceberg/src/inspect/history.rs b/crates/iceberg/src/inspect/history.rs
index fef9b5d6a..8fec7d577 100644
--- a/crates/iceberg/src/inspect/history.rs
+++ b/crates/iceberg/src/inspect/history.rs
@@ -126,9 +126,9 @@ mod tests {
use crate::TableIdent;
use crate::io::FileIO;
- use crate::scan::tests::TableTestFixture;
use crate::spec::TableMetadata;
use crate::table::Table;
+ use crate::test_utils::scan::TableTestFixture;
use crate::test_utils::{check_record_batches, test_runtime};
fn table_from_metadata_json(metadata_json: &str) -> Table {
diff --git a/crates/iceberg/src/inspect/manifests.rs b/crates/iceberg/src/inspect/manifests.rs
index e26037576..c9f4c3f40 100644
--- a/crates/iceberg/src/inspect/manifests.rs
+++ b/crates/iceberg/src/inspect/manifests.rs
@@ -289,9 +289,9 @@ mod tests {
use expect_test::expect;
use futures::TryStreamExt;
- use crate::scan::tests::TableTestFixture;
use crate::spec::TableMetadata;
use crate::test_utils::check_record_batches;
+ use crate::test_utils::scan::TableTestFixture;
#[tokio::test]
async fn test_manifests_table() {
diff --git a/crates/iceberg/src/inspect/snapshots.rs b/crates/iceberg/src/inspect/snapshots.rs
index fbed7ec11..22046daba 100644
--- a/crates/iceberg/src/inspect/snapshots.rs
+++ b/crates/iceberg/src/inspect/snapshots.rs
@@ -139,8 +139,8 @@ mod tests {
use expect_test::expect;
use futures::TryStreamExt;
- use crate::scan::tests::TableTestFixture;
use crate::test_utils::check_record_batches;
+ use crate::test_utils::scan::TableTestFixture;
#[tokio::test]
async fn test_snapshots_table() {
diff --git a/crates/iceberg/src/lib.rs b/crates/iceberg/src/lib.rs
index 94641b665..bccf82e1f 100644
--- a/crates/iceberg/src/lib.rs
+++ b/crates/iceberg/src/lib.rs
@@ -99,7 +99,8 @@ pub use runtime::{JoinHandle, Runtime, RuntimeHandle};
pub mod arrow;
pub(crate) mod delete_file_index;
pub mod encryption;
-pub mod test_utils;
+#[cfg(test)]
+mod test_utils;
pub mod writer;
mod delete_vector;
diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs
index 7b39189bf..7d8140396 100644
--- a/crates/iceberg/src/scan/mod.rs
+++ b/crates/iceberg/src/scan/mod.rs
@@ -652,1178 +652,39 @@ pub(crate) struct BoundPredicates {
}
#[cfg(test)]
-pub mod tests {
+mod tests {
//! shared tests for the table scan API
- #![allow(missing_docs)]
use std::collections::HashMap;
- use std::fs;
- use std::fs::File;
use std::sync::Arc;
use arrow_array::cast::AsArray;
use arrow_array::types::Int32Type;
use arrow_array::{
- Array, ArrayRef, BooleanArray, Float64Array, Int32Array, Int64Array, RecordBatch, RunArray,
- StringArray,
+ Array, ArrayRef, BooleanArray, Float64Array, Int32Array, Int64Array, RunArray, StringArray,
};
use futures::{TryStreamExt, stream};
- use minijinja::value::Value;
- use minijinja::{AutoEscape, Environment, context};
- use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY};
- use parquet::basic::Compression;
- use parquet::file::properties::WriterProperties;
- use tempfile::TempDir;
use uuid::Uuid;
use crate::arrow::ArrowReaderBuilder;
use crate::expr::{BoundPredicate, Reference};
- use crate::io::{FileIO, OutputFile};
+ use crate::io::FileIO;
use crate::metadata_columns::{
- RESERVED_COL_NAME_DELETE_FILE_PATH, RESERVED_COL_NAME_DELETE_FILE_POS,
RESERVED_COL_NAME_FILE, RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER,
- RESERVED_COL_NAME_POS, RESERVED_COL_NAME_SPEC_ID, RESERVED_FIELD_ID_DELETE_FILE_PATH,
- RESERVED_FIELD_ID_DELETE_FILE_POS, RESERVED_FIELD_ID_POS,
+ RESERVED_COL_NAME_POS, RESERVED_COL_NAME_SPEC_ID, RESERVED_FIELD_ID_POS,
};
use crate::scan::{FileScanTask, FileScanTaskDeleteFile};
use crate::spec::{
- DataContentType, DataFileBuilder, DataFileFormat, Datum, FormatVersion, Literal,
- MAIN_BRANCH, ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder,
- MappedField, NameMapping, NestedField, NullOrder, Operation, PartitionSpec, PrimitiveType,
- Schema, Snapshot, SortDirection, SortField, SortOrder, Struct, StructType, Summary,
- TableMetadata, TableMetadataBuilder, TableProperties, Transform, Type,
- UnboundPartitionSpec,
+ DataContentType, DataFileFormat, Datum, Literal, MAIN_BRANCH, MappedField, NameMapping,
+ NestedField, NullOrder, Operation, PartitionSpec, PrimitiveType, Schema, Snapshot,
+ SortDirection, SortField, SortOrder, Struct, Summary, TableMetadataBuilder,
+ TableProperties, Transform, Type, UnboundPartitionSpec,
};
use crate::table::Table;
+ use crate::test_utils::scan::{TableTestFixture, assert_last_updated_seq_all};
use crate::test_utils::test_runtime;
use crate::{ErrorKind, TableIdent};
- fn render_template(template: &str, ctx: Value) -> String {
- let mut env = Environment::new();
- env.set_auto_escape_callback(|_| AutoEscape::None);
- env.render_str(template, ctx).unwrap()
- }
-
- /// Asserts every row of the `_last_updated_sequence_number` column across all
- /// batches equals `expected` (or is null when `expected` is `None`), decoding
- /// the logical value independent of the physical (run-end) encoding.
- fn assert_last_updated_seq_all(batches: &[RecordBatch], expected: Option<i64>) {
- use arrow_cast::cast;
- use arrow_schema::DataType;
- for batch in batches {
- let col = batch
- .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
- .expect("_last_updated_sequence_number column should be present");
- let logical = cast(col, &DataType::Int64).unwrap();
- let values = logical.as_primitive::<arrow_array::types::Int64Type>();
- for i in 0..values.len() {
- let actual = (!values.is_null(i)).then(|| values.value(i));
- assert_eq!(actual, expected, "row {i}");
- }
- }
- }
-
- pub struct TableTestFixture {
- pub table_location: String,
- pub table: Table,
- }
-
- impl TableTestFixture {
- #[allow(clippy::new_without_default)]
- pub fn new() -> Self {
- let tmp_dir = TempDir::new().unwrap();
- let table_location = tmp_dir.path().join("table1");
- let manifest_list1_location = table_location.join("metadata/manifests_list_1.avro");
- let manifest_list2_location = table_location.join("metadata/manifests_list_2.avro");
- let table_metadata1_location = table_location.join("metadata/v1.json");
-
- let file_io = FileIO::new_with_fs();
-
- let table_metadata = {
- let template_json_str = fs::read_to_string(format!(
- "{}/testdata/example_table_metadata_v2.json",
- env!("CARGO_MANIFEST_DIR")
- ))
- .unwrap();
- let metadata_json = render_template(&template_json_str, context! {
- table_location => &table_location,
- manifest_list_1_location => &manifest_list1_location,
- manifest_list_2_location => &manifest_list2_location,
- table_metadata_1_location => &table_metadata1_location,
- });
- serde_json::from_str::<TableMetadata>(&metadata_json).unwrap()
- };
-
- let table = Table::builder()
- .metadata(table_metadata)
- .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
- .file_io(file_io.clone())
- .metadata_location(table_metadata1_location.as_os_str().to_str().unwrap())
- .runtime(test_runtime())
- .build()
- .unwrap();
-
- Self {
- table_location: table_location.to_str().unwrap().to_string(),
- table,
- }
- }
-
- #[allow(clippy::new_without_default)]
- pub fn new_empty() -> Self {
- let tmp_dir = TempDir::new().unwrap();
- let table_location = tmp_dir.path().join("table1");
- let table_metadata1_location = table_location.join("metadata/v1.json");
-
- let file_io = FileIO::new_with_fs();
-
- let table_metadata = {
- let template_json_str = fs::read_to_string(format!(
- "{}/testdata/example_empty_table_metadata_v2.json",
- env!("CARGO_MANIFEST_DIR")
- ))
- .unwrap();
- let metadata_json = render_template(&template_json_str, context! {
- table_location => &table_location,
- table_metadata_1_location => &table_metadata1_location,
- });
- serde_json::from_str::<TableMetadata>(&metadata_json).unwrap()
- };
-
- let table = Table::builder()
- .metadata(table_metadata)
- .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
- .file_io(file_io.clone())
- .metadata_location(table_metadata1_location.as_os_str().to_str().unwrap())
- .runtime(test_runtime())
- .build()
- .unwrap();
-
- Self {
- table_location: table_location.to_str().unwrap().to_string(),
- table,
- }
- }
-
- /// Creates a fixture with 5 snapshots chained as:
- /// S1 (root) -> S2 -> S3 -> S4 -> S5 (current)
- /// Useful for testing snapshot history traversal.
- pub fn new_with_deep_history() -> Self {
- let tmp_dir = TempDir::new().unwrap();
- let table_location = tmp_dir.path().join("table1");
- let table_metadata1_location = table_location.join("metadata/v1.json");
-
- let file_io = FileIO::new_with_fs();
-
- let table_metadata = {
- let json_str = fs::read_to_string(format!(
- "{}/testdata/example_table_metadata_v2_deep_history.json",
- env!("CARGO_MANIFEST_DIR")
- ))
- .unwrap();
- serde_json::from_str::<TableMetadata>(&json_str).unwrap()
- };
-
- let table = Table::builder()
- .metadata(table_metadata)
- .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
- .file_io(file_io.clone())
- .metadata_location(table_metadata1_location.as_os_str().to_str().unwrap())
- .runtime(test_runtime())
- .build()
- .unwrap();
-
- Self {
- table_location: table_location.to_str().unwrap().to_string(),
- table,
- }
- }
-
- pub fn new_unpartitioned() -> Self {
- let tmp_dir = TempDir::new().unwrap();
- let table_location = tmp_dir.path().join("table1");
- let manifest_list1_location = table_location.join("metadata/manifests_list_1.avro");
- let manifest_list2_location = table_location.join("metadata/manifests_list_2.avro");
- let table_metadata1_location = table_location.join("metadata/v1.json");
-
- let file_io = FileIO::new_with_fs();
-
- let mut table_metadata = {
- let template_json_str = fs::read_to_string(format!(
- "{}/testdata/example_table_metadata_v2.json",
- env!("CARGO_MANIFEST_DIR")
- ))
- .unwrap();
- let metadata_json = render_template(&template_json_str, context! {
- table_location => &table_location,
- manifest_list_1_location => &manifest_list1_location,
- manifest_list_2_location => &manifest_list2_location,
- table_metadata_1_location => &table_metadata1_location,
- });
- serde_json::from_str::<TableMetadata>(&metadata_json).unwrap()
- };
-
- table_metadata.default_spec = Arc::new(PartitionSpec::unpartition_spec());
- table_metadata.partition_specs.clear();
- table_metadata.default_partition_type = StructType::new(vec![]);
- table_metadata
- .partition_specs
- .insert(0, table_metadata.default_spec.clone());
-
- let table = Table::builder()
- .metadata(table_metadata)
- .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
- .file_io(file_io.clone())
- .metadata_location(table_metadata1_location.to_str().unwrap())
- .runtime(test_runtime())
- .build()
- .unwrap();
-
- Self {
- table_location: table_location.to_str().unwrap().to_string(),
- table,
- }
- }
-
- pub fn new_with_partition_evolution() -> Self {
- let table = Self::new().table;
- let table_location = table.metadata().location.clone();
-
- let manifest_list1_location =
- format!("{}/metadata/manifests_list_1.avro", table_location);
- let manifest_list2_location =
- format!("{}/metadata/manifests_list_2.avro", table_location);
- let manifest_list3_location =
- format!("{}/metadata/manifests_list_3.avro", table_location);
- let table_metadata1_location = format!("{}/metadata/v1.json", table_location);
-
- let new_table_metadata = {
- let template_json_str = fs::read_to_string(format!(
- "{}/testdata/example_table_metadata_v2_partition_evolution.json",
- env!("CARGO_MANIFEST_DIR")
- ))
- .unwrap();
- let metadata_json = render_template(&template_json_str, context! {
- table_location => &table_location,
- manifest_list_1_location => &manifest_list1_location,
- manifest_list_2_location => &manifest_list2_location,
- manifest_list_3_location => &manifest_list3_location,
- table_metadata_1_location => &table_metadata1_location,
- });
- Arc::new(serde_json::from_str::<TableMetadata>(&metadata_json).unwrap())
- };
-
- Self {
- table_location,
- table: table.with_metadata(new_table_metadata),
- }
- }
-
- fn next_manifest_file(&self) -> OutputFile {
- self.table
- .file_io()
- .new_output(format!(
- "{}/metadata/manifest_{}.avro",
- self.table_location,
- Uuid::new_v4()
- ))
- .unwrap()
- }
-
- pub async fn setup_manifest_files(&mut self) {
- let current_snapshot = self.table.metadata().current_snapshot().unwrap();
- let parent_snapshot = current_snapshot
- .parent_snapshot(self.table.metadata())
- .unwrap();
- let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
- let current_partition_spec = self.table.metadata().default_partition_spec();
-
- // Write the data files first, then use the file size in the manifest entries
- let parquet_file_size = self.write_parquet_data_files();
-
- let mut writer = ManifestWriterBuilder::new(
- self.next_manifest_file(),
- Some(current_snapshot.snapshot_id()),
- current_schema.clone(),
- current_partition_spec.as_ref().clone(),
- )
- .build_v2_data();
- writer
- .add_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Added)
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(format!("{}/1.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(Struct::from_iter([Some(Literal::long(100))]))
- .key_metadata(None)
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- writer
- .add_delete_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Deleted)
- .snapshot_id(parent_snapshot.snapshot_id())
- .sequence_number(parent_snapshot.sequence_number())
- .file_sequence_number(parent_snapshot.sequence_number())
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(format!("{}/2.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(Struct::from_iter([Some(Literal::long(200))]))
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- writer
- .add_existing_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Existing)
- .snapshot_id(parent_snapshot.snapshot_id())
- .sequence_number(parent_snapshot.sequence_number())
- .file_sequence_number(parent_snapshot.sequence_number())
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(format!("{}/3.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(Struct::from_iter([Some(Literal::long(300))]))
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- let data_file_manifest = writer.write_manifest_file().await.unwrap();
-
- // Write to manifest list
- let manifest_list_writer = self
- .table
- .file_io()
- .new_output(current_snapshot.manifest_list())
- .unwrap()
- .writer()
- .await
- .unwrap();
- let mut manifest_list_write = ManifestListWriter::v2(
- manifest_list_writer,
- current_snapshot.snapshot_id(),
- current_snapshot.parent_snapshot_id(),
- current_snapshot.sequence_number(),
- );
- manifest_list_write
- .add_manifests(vec![data_file_manifest].into_iter())
- .unwrap();
- manifest_list_write.close().await.unwrap();
- }
-
- /// Writes a v3 data manifest with a manifest-level `first_row_id` of 42,
- /// so live entries inherit a per-file `first_row_id` on read. Upgrades the
- /// table to v3 first, so the manifest list is read as v3.
- pub async fn setup_v3_manifest_files(&mut self) {
- let metadata = TableMetadataBuilder::new_from_metadata(
- self.table.metadata().clone(),
- self.table.metadata_location().map(str::to_string),
- )
- .upgrade_format_version(FormatVersion::V3)
- .unwrap()
- .build()
- .unwrap()
- .metadata;
- self.table = Table::builder()
- .metadata(metadata)
- .identifier(self.table.identifier().clone())
- .file_io(self.table.file_io().clone())
- .metadata_location(self.table.metadata_location().unwrap().to_string())
- .runtime(test_runtime())
- .build()
- .unwrap();
-
- let current_snapshot = self.table.metadata().current_snapshot().unwrap();
- let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
- let current_partition_spec = self.table.metadata().default_partition_spec();
-
- let parquet_file_size = self.write_parquet_data_files();
-
- let mut writer = ManifestWriterBuilder::new(
- self.next_manifest_file(),
- Some(current_snapshot.snapshot_id()),
- current_schema.clone(),
- current_partition_spec.as_ref().clone(),
- )
- .build_v3_data();
- writer
- .add_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Added)
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(format!("{}/1.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(Struct::from_iter([Some(Literal::long(100))]))
- .key_metadata(None)
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- let data_file_manifest = writer.write_manifest_file().await.unwrap();
-
- let manifest_list_writer = self
- .table
- .file_io()
- .new_output(current_snapshot.manifest_list())
- .unwrap()
- .writer()
- .await
- .unwrap();
- let mut manifest_list_write = ManifestListWriter::v3(
- manifest_list_writer,
- current_snapshot.snapshot_id(),
- current_snapshot.parent_snapshot_id(),
- current_snapshot.sequence_number(),
- Some(42),
- );
- manifest_list_write
- .add_manifests(vec![data_file_manifest].into_iter())
- .unwrap();
- manifest_list_write.close().await.unwrap();
- }
-
- pub async fn setup_manifest_files_with_partition_evolution(&mut self) {
- let current_snapshot = self.table.metadata().current_snapshot().unwrap();
- let parent_snapshot = current_snapshot
- .parent_snapshot(self.table.metadata())
- .unwrap();
- let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
- let current_partition_spec = self.table.metadata().default_partition_spec();
-
- // Write the data files first, then use the file size in the manifest entries
- let parquet_file_size = self.write_parquet_data_files();
-
- let mut writer = ManifestWriterBuilder::new(
- self.next_manifest_file(),
- Some(current_snapshot.snapshot_id()),
- current_schema.clone(),
- current_partition_spec.as_ref().clone(),
- )
- .build_v2_data();
- writer
- .add_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Added)
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(1)
- .content(DataContentType::Data)
- .file_path(format!("{}/1.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(Struct::from_iter([
- Some(Literal::long(100)),
- Some(Literal::string("apa")),
- Some(Literal::int(27)),
- ]))
- .key_metadata(None)
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- writer
- .add_delete_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Deleted)
- .snapshot_id(parent_snapshot.snapshot_id())
- .sequence_number(parent_snapshot.sequence_number())
- .file_sequence_number(parent_snapshot.sequence_number())
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(1)
- .content(DataContentType::Data)
- .file_path(format!("{}/2.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(Struct::from_iter([
- Some(Literal::long(200)),
- Some(Literal::string("ice")),
- Some(Literal::int(5)),
- ]))
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- writer
- .add_existing_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Existing)
- .snapshot_id(parent_snapshot.snapshot_id())
- .sequence_number(parent_snapshot.sequence_number())
- .file_sequence_number(parent_snapshot.sequence_number())
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(1)
- .content(DataContentType::Data)
- .file_path(format!("{}/3.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(Struct::from_iter([
- Some(Literal::long(300)),
- Some(Literal::string("apa")),
- Some(Literal::int(19)),
- ]))
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- let data_file_manifest = writer.write_manifest_file().await.unwrap();
-
- // Write to manifest list
- let manifest_list_writer = self
- .table
- .file_io()
- .new_output(current_snapshot.manifest_list())
- .unwrap()
- .writer()
- .await
- .unwrap();
- let mut manifest_list_write = ManifestListWriter::v2(
- manifest_list_writer,
- current_snapshot.snapshot_id(),
- current_snapshot.parent_snapshot_id(),
- current_snapshot.sequence_number(),
- );
- manifest_list_write
- .add_manifests(vec![data_file_manifest].into_iter())
- .unwrap();
- manifest_list_write.close().await.unwrap();
- }
-
- /// Writes identical Parquet data files (1.parquet, 2.parquet, 3.parquet)
- /// and returns the file size in bytes.
- fn write_parquet_data_files(&self) -> u64 {
- fs::create_dir_all(&self.table_location).unwrap();
-
- let schema = {
- let fields = vec![
- arrow_schema::Field::new("x", arrow_schema::DataType::Int64, false)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "1".to_string(),
- )])),
- arrow_schema::Field::new("y", arrow_schema::DataType::Int64, false)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "2".to_string(),
- )])),
- arrow_schema::Field::new("z", arrow_schema::DataType::Int64, false)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "3".to_string(),
- )])),
- arrow_schema::Field::new("a", arrow_schema::DataType::Utf8, false)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "4".to_string(),
- )])),
- arrow_schema::Field::new("dbl", arrow_schema::DataType::Float64, false)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "5".to_string(),
- )])),
- arrow_schema::Field::new("i32", arrow_schema::DataType::Int32, false)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "6".to_string(),
- )])),
- arrow_schema::Field::new("i64", arrow_schema::DataType::Int64, false)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "7".to_string(),
- )])),
- arrow_schema::Field::new("bool", arrow_schema::DataType::Boolean, false)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "8".to_string(),
- )])),
- ];
- Arc::new(arrow_schema::Schema::new(fields))
- };
- // x: [1, 1, 1, 1, ...]
- let col1 = Arc::new(Int64Array::from_iter_values(vec![1; 1024])) as ArrayRef;
-
- let mut values = vec![2; 512];
- values.append(vec![3; 200].as_mut());
- values.append(vec![4; 300].as_mut());
- values.append(vec![5; 12].as_mut());
-
- // y: [2, 2, 2, 2, ..., 3, 3, 3, 3, ..., 4, 4, 4, 4, ..., 5, 5, 5, 5]
- let col2 = Arc::new(Int64Array::from_iter_values(values)) as ArrayRef;
-
- let mut values = vec![3; 512];
- values.append(vec![4; 512].as_mut());
-
- // z: [3, 3, 3, 3, ..., 4, 4, 4, 4]
- let col3 = Arc::new(Int64Array::from_iter_values(values)) as ArrayRef;
-
- // a: ["Apache", "Apache", "Apache", ..., "Iceberg", "Iceberg", "Iceberg"]
- let mut values = vec!["Apache"; 512];
- values.append(vec!["Iceberg"; 512].as_mut());
- let col4 = Arc::new(StringArray::from_iter_values(values)) as ArrayRef;
-
- // dbl:
- let mut values = vec![100.0f64; 512];
- values.append(vec![150.0f64; 12].as_mut());
- values.append(vec![200.0f64; 500].as_mut());
- let col5 = Arc::new(Float64Array::from_iter_values(values)) as ArrayRef;
-
- // i32:
- let mut values = vec![100i32; 512];
- values.append(vec![150i32; 12].as_mut());
- values.append(vec![200i32; 500].as_mut());
- let col6 = Arc::new(Int32Array::from_iter_values(values)) as ArrayRef;
-
- // i64:
- let mut values = vec![100i64; 512];
- values.append(vec![150i64; 12].as_mut());
- values.append(vec![200i64; 500].as_mut());
- let col7 = Arc::new(Int64Array::from_iter_values(values)) as ArrayRef;
-
- // bool:
- let mut values = vec![false; 512];
- values.append(vec![true; 512].as_mut());
- let values: BooleanArray = values.into();
- let col8 = Arc::new(values) as ArrayRef;
-
- let to_write = RecordBatch::try_new(schema.clone(), vec![
- col1, col2, col3, col4, col5, col6, col7, col8,
- ])
- .unwrap();
-
- // Write the Parquet files
- let props = WriterProperties::builder()
- .set_compression(Compression::SNAPPY)
- .build();
-
- for n in 1..=3 {
- let file = File::create(format!("{}/{}.parquet", &self.table_location, n)).unwrap();
- let mut writer =
- ArrowWriter::try_new(file, to_write.schema(), Some(props.clone())).unwrap();
-
- writer.write(&to_write).expect("Writing batch");
-
- // writer must be closed to write footer
- writer.close().unwrap();
- }
-
- fs::metadata(format!("{}/1.parquet", &self.table_location))
- .unwrap()
- .len()
- }
-
- pub async fn setup_unpartitioned_manifest_files(&mut self) {
- let current_snapshot = self.table.metadata().current_snapshot().unwrap();
- let parent_snapshot = current_snapshot
- .parent_snapshot(self.table.metadata())
- .unwrap();
- let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
- let current_partition_spec = Arc::new(PartitionSpec::unpartition_spec());
-
- // Write the data files first, then use the file size in the manifest entries
- let parquet_file_size = self.write_parquet_data_files();
-
- // Write data files using an empty partition for unpartitioned tables.
- let mut writer = ManifestWriterBuilder::new(
- self.next_manifest_file(),
- Some(current_snapshot.snapshot_id()),
- current_schema.clone(),
- current_partition_spec.as_ref().clone(),
- )
- .build_v2_data();
-
- // Create an empty partition value.
- let empty_partition = Struct::empty();
-
- writer
- .add_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Added)
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(format!("{}/1.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(empty_partition.clone())
- .key_metadata(None)
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
-
- writer
- .add_delete_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Deleted)
- .snapshot_id(parent_snapshot.snapshot_id())
- .sequence_number(parent_snapshot.sequence_number())
- .file_sequence_number(parent_snapshot.sequence_number())
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(format!("{}/2.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(empty_partition.clone())
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
-
- writer
- .add_existing_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Existing)
- .snapshot_id(parent_snapshot.snapshot_id())
- .sequence_number(parent_snapshot.sequence_number())
- .file_sequence_number(parent_snapshot.sequence_number())
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(format!("{}/3.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(empty_partition.clone())
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
-
- let data_file_manifest = writer.write_manifest_file().await.unwrap();
-
- // Write to manifest list
- let manifest_list_writer = self
- .table
- .file_io()
- .new_output(current_snapshot.manifest_list())
- .unwrap()
- .writer()
- .await
- .unwrap();
- let mut manifest_list_write = ManifestListWriter::v2(
- manifest_list_writer,
- current_snapshot.snapshot_id(),
- current_snapshot.parent_snapshot_id(),
- current_snapshot.sequence_number(),
- );
- manifest_list_write
- .add_manifests(vec![data_file_manifest].into_iter())
- .unwrap();
- manifest_list_write.close().await.unwrap();
- }
-
- pub async fn setup_deadlock_manifests(&mut self) {
- let current_snapshot = self.table.metadata().current_snapshot().unwrap();
- let _parent_snapshot = current_snapshot
- .parent_snapshot(self.table.metadata())
- .unwrap();
- let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
- let current_partition_spec = self.table.metadata().default_partition_spec();
-
- // 1. Write DATA manifest with MULTIPLE entries to fill buffer
- let mut writer = ManifestWriterBuilder::new(
- self.next_manifest_file(),
- Some(current_snapshot.snapshot_id()),
- current_schema.clone(),
- current_partition_spec.as_ref().clone(),
- )
- .build_v2_data();
-
- // Add 10 data entries
- for i in 0..10 {
- writer
- .add_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Added)
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(format!("{}/{}.parquet", &self.table_location, i))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(100)
- .record_count(1)
- .partition(Struct::from_iter([Some(Literal::long(100))]))
- .key_metadata(None)
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- }
- let data_manifest = writer.write_manifest_file().await.unwrap();
-
- // 2. Write DELETE manifest
- let mut writer = ManifestWriterBuilder::new(
- self.next_manifest_file(),
- Some(current_snapshot.snapshot_id()),
- current_schema.clone(),
- current_partition_spec.as_ref().clone(),
- )
- .build_v2_deletes();
-
- writer
- .add_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Added)
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::PositionDeletes)
- .file_path(format!("{}/del.parquet", &self.table_location))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(100)
- .record_count(1)
- .partition(Struct::from_iter([Some(Literal::long(100))]))
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- let delete_manifest = writer.write_manifest_file().await.unwrap();
-
- // Write to manifest list - DATA FIRST then DELETE
- // This order is crucial for reproduction
- let manifest_list_writer = self
- .table
- .file_io()
- .new_output(current_snapshot.manifest_list())
- .unwrap()
- .writer()
- .await
- .unwrap();
- let mut manifest_list_write = ManifestListWriter::v2(
- manifest_list_writer,
- current_snapshot.snapshot_id(),
- current_snapshot.parent_snapshot_id(),
- current_snapshot.sequence_number(),
- );
- manifest_list_write
- .add_manifests(vec![data_manifest, delete_manifest].into_iter())
- .unwrap();
- manifest_list_write.close().await.unwrap();
- }
-
- /// Sets up a single data file `mrg.parquet` with three 100-row row groups
- /// (column `x` = 1000..1300, so row position `p` carries `x = 1000 + p`) and
- /// registers it in the current snapshot. When `delete_positions` is non-empty,
- /// also writes a positional delete file targeting those file-absolute positions
- /// and registers it in a delete manifest.
- ///
- /// Used to exercise the `_pos` metadata column through the real `TableScan`
- /// planning path across row-group boundaries and (optionally) positional deletes.
- pub async fn setup_multi_row_group_manifest(&mut self, delete_positions: &[i64]) {
- let current_snapshot = self.table.metadata().current_snapshot().unwrap();
- let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
- let current_partition_spec = self.table.metadata().default_partition_spec();
-
- // The table's spec 0 is identity on `x`, so give the data and delete files a
- // fixed partition value. Filter tests deliberately filter on `y` (a
- // non-partition column) so pruning is driven by Parquet row-group statistics
- // rather than partition values.
- let partition = Struct::from_iter([Some(Literal::long(1000))]);
-
- let (data_file_path, data_file_size) = self.write_multi_row_group_data_file();
-
- let mut data_writer = ManifestWriterBuilder::new(
- self.next_manifest_file(),
- Some(current_snapshot.snapshot_id()),
- current_schema.clone(),
- current_partition_spec.as_ref().clone(),
- )
- .build_v2_data();
- data_writer
- .add_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Added)
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(data_file_path.clone())
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(data_file_size)
- .record_count(300)
- .partition(partition.clone())
- .key_metadata(None)
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- let data_manifest = data_writer.write_manifest_file().await.unwrap();
-
- let mut manifests = vec![data_manifest];
-
- if !delete_positions.is_empty() {
- let (del_path, del_size) =
- self.write_positional_delete_file(&data_file_path, delete_positions);
-
- let mut delete_writer = ManifestWriterBuilder::new(
- self.next_manifest_file(),
- Some(current_snapshot.snapshot_id()),
- current_schema.clone(),
- current_partition_spec.as_ref().clone(),
- )
- .build_v2_deletes();
- delete_writer
- .add_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Added)
- .data_file(
- DataFileBuilder::default()
- .partition_spec_id(0)
- .content(DataContentType::PositionDeletes)
- .file_path(del_path)
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(del_size)
- .record_count(delete_positions.len() as u64)
- .partition(partition.clone())
- .build()
- .unwrap(),
- )
- .build(),
- )
- .unwrap();
- manifests.push(delete_writer.write_manifest_file().await.unwrap());
- }
-
- let manifest_list_writer = self
- .table
- .file_io()
- .new_output(current_snapshot.manifest_list())
- .unwrap()
- .writer()
- .await
- .unwrap();
- let mut manifest_list_write = ManifestListWriter::v2(
- manifest_list_writer,
- current_snapshot.snapshot_id(),
- current_snapshot.parent_snapshot_id(),
- current_snapshot.sequence_number(),
- );
- manifest_list_write
- .add_manifests(manifests.into_iter())
- .unwrap();
- manifest_list_write.close().await.unwrap();
- }
-
- /// Writes a manifest with three live "Added" data-file entries (partitioned on `x`
- /// = 100, 200, 300), each with the given `sort_order_id` set on its `DataFile`
- /// (`None` leaves the field unset). Used to test how `sort_order_id` resolution
- /// against the table's sort orders flows into each entry's `FileScanTask`.
- pub async fn setup_manifest_files_with_sort_order_ids(
- &mut self,
- sort_order_ids: [Option<i32>; 4],
- ) {
- let current_snapshot = self.table.metadata().current_snapshot().unwrap();
- let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
- let current_partition_spec = self.table.metadata().default_partition_spec();
- let parquet_file_size = self.write_parquet_data_files();
-
- let mut writer = ManifestWriterBuilder::new(
- self.next_manifest_file(),
- Some(current_snapshot.snapshot_id()),
- current_schema.clone(),
- current_partition_spec.as_ref().clone(),
- )
- .build_v2_data();
-
- for (i, sort_order_id) in sort_order_ids.into_iter().enumerate() {
- let mut data_file_builder = DataFileBuilder::default();
- data_file_builder
- .partition_spec_id(0)
- .content(DataContentType::Data)
- .file_path(format!("{}/{}.parquet", &self.table_location, i + 1))
- .file_format(DataFileFormat::Parquet)
- .file_size_in_bytes(parquet_file_size)
- .record_count(1)
- .partition(Struct::from_iter([Some(Literal::long(
- 100 * (i as i64 + 1),
- ))]));
- if let Some(id) = sort_order_id {
- data_file_builder.sort_order_id(id);
- }
- let data_file = data_file_builder.build().unwrap();
-
- writer
- .add_entry(
- ManifestEntry::builder()
- .status(ManifestStatus::Added)
- .data_file(data_file)
- .build(),
- )
- .unwrap();
- }
-
- let data_file_manifest = writer.write_manifest_file().await.unwrap();
-
- let manifest_list_writer = self
- .table
- .file_io()
- .new_output(current_snapshot.manifest_list())
- .unwrap()
- .writer()
- .await
- .unwrap();
- let mut manifest_list_write = ManifestListWriter::v2(
- manifest_list_writer,
- current_snapshot.snapshot_id(),
- current_snapshot.parent_snapshot_id(),
- current_snapshot.sequence_number(),
- );
- manifest_list_write
- .add_manifests(std::iter::once(data_file_manifest))
- .unwrap();
- manifest_list_write.close().await.unwrap();
- }
-
- /// Writes `mrg.parquet` with three 100-row row groups. Columns `x` (field
- /// id `1`) and `y` (field id `2`) both run 1000..1300, so row position `p`
- /// carries `x = y = 1000 + p`. Returns `(path, file_size_in_bytes)`.
- fn write_multi_row_group_data_file(&self) -> (String, u64) {
- fs::create_dir_all(&self.table_location).unwrap();
-
- let arrow_schema = Arc::new(arrow_schema::Schema::new(vec![
- arrow_schema::Field::new("x", arrow_schema::DataType::Int64, false).with_metadata(
- HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string())]),
- ),
- arrow_schema::Field::new("y", arrow_schema::DataType::Int64, false).with_metadata(
- HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "2".to_string())]),
- ),
- ]));
-
- let path = format!("{}/mrg.parquet", &self.table_location);
- let max_row_group_row_count = 100;
- let props = WriterProperties::builder()
- .set_compression(Compression::SNAPPY)
- .set_max_row_group_row_count(Some(max_row_group_row_count))
- .build();
-
- let file = File::create(&path).unwrap();
- let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), Some(props)).unwrap();
- for group in 0..3i64 {
- let base = 1000 + group * max_row_group_row_count as i64;
- let col = Arc::new(Int64Array::from_iter_values(
- base..base + max_row_group_row_count as i64,
- )) as ArrayRef;
- let batch =
- RecordBatch::try_new(arrow_schema.clone(), vec![col.clone(), col]).unwrap();
- writer.write(&batch).unwrap();
- }
- writer.close().unwrap();
-
- let size = fs::metadata(&path).unwrap().len();
- (path, size)
- }
-
- /// Writes a positional delete file targeting `positions` in `data_path`.
- /// Returns `(path, file_size_in_bytes)`.
- fn write_positional_delete_file(
- &self,
- data_path: &str,
- positions: &[i64],
- ) -> (String, u64) {
- let del_schema = Arc::new(arrow_schema::Schema::new(vec![
- arrow_schema::Field::new(
- RESERVED_COL_NAME_DELETE_FILE_PATH,
- arrow_schema::DataType::Utf8,
- false,
- )
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- RESERVED_FIELD_ID_DELETE_FILE_PATH.to_string(), // 2147483546
- )])),
- arrow_schema::Field::new(
- RESERVED_COL_NAME_DELETE_FILE_POS,
- arrow_schema::DataType::Int64,
- false,
- )
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- RESERVED_FIELD_ID_DELETE_FILE_POS.to_string(), // 2147483545
- )])),
- ]));
-
- let batch = RecordBatch::try_new(del_schema.clone(), vec![
- Arc::new(StringArray::from_iter_values(std::iter::repeat_n(
- data_path.to_string(),
- positions.len(),
- ))) as ArrayRef,
- Arc::new(Int64Array::from_iter_values(positions.iter().copied())) as ArrayRef,
- ])
- .unwrap();
-
- let path = format!("{}/pos-del.parquet", &self.table_location);
- let props = WriterProperties::builder()
- .set_compression(Compression::SNAPPY)
- .build();
- let file = File::create(&path).unwrap();
- let mut writer = ArrowWriter::try_new(file, del_schema, Some(props)).unwrap();
- writer.write(&batch).unwrap();
- writer.close().unwrap();
-
- let size = fs::metadata(&path).unwrap().len();
- (path, size)
- }
- }
-
#[tokio::test]
async fn test_table_scan_columns() {
let table = TableTestFixture::new().table;
diff --git a/crates/iceberg/src/test_utils.rs b/crates/iceberg/src/test_utils.rs
deleted file mode 100644
index e80a51d98..000000000
--- a/crates/iceberg/src/test_utils.rs
+++ /dev/null
@@ -1,190 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-//! Test utilities.
-//! This module is pub just for internal testing.
-//! It is subject to change and is not intended to be used by external users.
-
-use std::sync::{Arc, OnceLock};
-
-use arrow_array::RecordBatch;
-use expect_test::Expect;
-use itertools::Itertools;
-#[cfg(test)]
-use roaring::RoaringTreemap;
-
-use crate::TableIdent;
-#[cfg(test)]
-use crate::encryption::EncryptionManager;
-use crate::encryption::SensitiveBytes;
-use crate::encryption::kms::{KeyManagementClient, MemoryKeyManagementClient};
-use crate::io::FileIO;
-use crate::runtime::Runtime;
-use crate::spec::TableMetadata;
-use crate::table::Table;
-
-/// Returns a process-wide [`Runtime`] suitable for tests that need to construct
-/// a [`Table`] outside a tokio context.
-///
-/// The returned [`Runtime`] wraps a single shared multi-thread tokio runtime
-/// that is lazily built on first call and lives until process exit. Cloning is
-/// cheap, so test code can call this every time it needs a runtime to feed
-/// into [`TableBuilder::runtime`](crate::table::TableBuilder::runtime).
-pub fn test_runtime() -> Runtime {
- static TOKIO_RT: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
- let tokio_rt = TOKIO_RT.get_or_init(|| {
- tokio::runtime::Builder::new_multi_thread()
- .enable_all()
- .build()
- .expect("failed to build test tokio runtime")
- });
- Runtime::new(tokio_rt)
-}
-
-/// Snapshot testing to check the resulting record batch.
-///
-/// - `expected_schema/data`: put `expect![[""]]` as a placeholder,
-/// and then run test with `UPDATE_EXPECT=1 cargo test` to automatically update the result,
-/// or use rust-analyzer (see [video](https://github.com/rust-analyzer/expect-test)).
-/// Check the doc of [`expect_test`] for more details.
-/// - `ignore_check_columns`: Some columns are not stable, so we can skip them.
-/// - `sort_column`: The order of the data might be non-deterministic, so we can sort it by a column.
-pub fn check_record_batches(
- record_batches: Vec<RecordBatch>,
- expected_schema: Expect,
- expected_data: Expect,
- ignore_check_columns: &[&str],
- sort_column: Option<&str>,
-) {
- assert!(!record_batches.is_empty(), "Empty record batches");
-
- // Combine record batches using the first batch's schema
- let first_batch = record_batches.first().unwrap();
- let record_batch =
- arrow_select::concat::concat_batches(&first_batch.schema(), &record_batches).unwrap();
-
- let mut columns = record_batch.columns().to_vec();
- if let Some(sort_column) = sort_column {
- let column = record_batch.column_by_name(sort_column).unwrap();
- let indices = arrow_ord::sort::sort_to_indices(column, None, None).unwrap();
- columns = columns
- .iter()
- .map(|column| arrow_select::take::take(column.as_ref(), &indices, None).unwrap())
- .collect_vec();
- }
-
- expected_schema.assert_eq(&format!(
- "{}",
- record_batch.schema().fields().iter().format(",\n")
- ));
- expected_data.assert_eq(&format!(
- "{}",
- record_batch
- .schema()
- .fields()
- .iter()
- .zip_eq(columns)
- .map(|(field, column)| {
- if ignore_check_columns.contains(&field.name().as_str()) {
- format!("{}: (skipped)", field.name())
- } else {
- format!("{}: {:?}", field.name(), column)
- }
- })
- .format(",\n")
- ));
-}
-
-/// An [`EncryptionManager`] backed by an in-memory KMS holding `table_key_id`.
-#[cfg(test)]
-pub(crate) fn make_encryption_manager(table_key_id: &str) -> Arc<EncryptionManager> {
- let kms = MemoryKeyManagementClient::new();
- kms.add_master_key(table_key_id).unwrap();
- Arc::new(
- EncryptionManager::builder()
- .kms_client(Arc::new(kms) as Arc<dyn KeyManagementClient>)
- .table_key_id(table_key_id)
- .build(),
- )
-}
-
-/// Encodes a `deletion-vector-v1` Puffin blob for the given positions, matching the framing in
-/// [`DeleteVector::deserialize`](crate::delete_vector::DeleteVector::deserialize).
-#[cfg(test)]
-pub(crate) fn encode_dv_blob(positions: impl IntoIterator<Item = u64>) -> Vec<u8> {
- let mut bitmap = RoaringTreemap::new();
- for pos in positions {
- bitmap.insert(pos);
- }
- let mut vector = Vec::new();
- bitmap.serialize_into(&mut vector).unwrap();
- crate::delete_vector::frame_dv_blob(&vector)
-}
-
-/// Build a table backed by the V3 encryption fixture and an in-memory KMS,
-/// so it has an [`EncryptionManager`](crate::encryption::EncryptionManager).
-///
-/// The fixture's snapshot references an encrypted manifest list; its bytes
-/// (the `manifest-list-v3-encrypted.avro` testdata, an encrypted empty list)
-/// are seeded into the in-memory `FileIO` at that path so callers can read
-/// the current snapshot's manifest list.
-pub async fn make_encrypted_table() -> Table {
- let metadata_json = std::fs::read_to_string(format!(
- "{}/testdata/table_metadata/TableMetadataV3ValidEncryption.json",
- env!("CARGO_MANIFEST_DIR"),
- ))
- .unwrap();
- let metadata: TableMetadata = serde_json::from_str(&metadata_json).unwrap();
-
- let kms: Arc<dyn KeyManagementClient> = {
- let k = MemoryKeyManagementClient::new();
- k.add_master_key_bytes(
- "master-1",
- SensitiveBytes::new([
- 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d,
- 0x0e, 0x0f,
- ]),
- )
- .unwrap();
- Arc::new(k)
- };
-
- let file_io = FileIO::new_with_memory();
-
- // Seed the encrypted (empty) manifest list at the path the snapshot references.
- let manifest_list_bytes = std::fs::read(format!(
- "{}/testdata/manifests_lists/manifest-list-v3-encrypted.avro",
- env!("CARGO_MANIFEST_DIR"),
- ))
- .unwrap();
- file_io
- .new_output(metadata.current_snapshot().unwrap().manifest_list())
- .unwrap()
- .write(manifest_list_bytes.into())
- .await
- .unwrap();
-
- Table::builder()
- .metadata(metadata)
- .metadata_location("memory:///table/metadata/v1.json")
- .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
- .file_io(file_io)
- .kms_client(kms)
- .runtime(test_runtime())
- .build()
- .unwrap()
-}
diff --git a/crates/iceberg/src/test_utils/delete_vector.rs b/crates/iceberg/src/test_utils/delete_vector.rs
new file mode 100644
index 000000000..b06bd7498
--- /dev/null
+++ b/crates/iceberg/src/test_utils/delete_vector.rs
@@ -0,0 +1,32 @@
+// 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.
+
+//! Fixtures for tests that need encoded deletion vectors.
+
+use roaring::RoaringTreemap;
+
+/// Encodes a `deletion-vector-v1` Puffin blob for the given positions, matching the framing in
+/// [`DeleteVector::deserialize`](crate::delete_vector::DeleteVector::deserialize).
+pub(crate) fn encode_dv_blob(positions: impl IntoIterator<Item = u64>) -> Vec<u8> {
+ let mut bitmap = RoaringTreemap::new();
+ for pos in positions {
+ bitmap.insert(pos);
+ }
+ let mut vector = Vec::new();
+ bitmap.serialize_into(&mut vector).unwrap();
+ crate::delete_vector::frame_dv_blob(&vector)
+}
diff --git a/crates/iceberg/src/test_utils/encryption.rs b/crates/iceberg/src/test_utils/encryption.rs
new file mode 100644
index 000000000..2accd5255
--- /dev/null
+++ b/crates/iceberg/src/test_utils/encryption.rs
@@ -0,0 +1,94 @@
+// 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.
+
+//! Fixtures for tests that need an [`EncryptionManager`] or an encrypted table.
+
+use std::sync::Arc;
+
+use super::test_runtime;
+use crate::TableIdent;
+use crate::encryption::kms::{KeyManagementClient, MemoryKeyManagementClient};
+use crate::encryption::{EncryptionManager, SensitiveBytes};
+use crate::io::FileIO;
+use crate::spec::TableMetadata;
+use crate::table::Table;
+
+/// An [`EncryptionManager`] backed by an in-memory KMS holding `table_key_id`.
+pub(crate) fn make_encryption_manager(table_key_id: &str) -> Arc<EncryptionManager> {
+ let kms = MemoryKeyManagementClient::new();
+ kms.add_master_key(table_key_id).unwrap();
+ Arc::new(
+ EncryptionManager::builder()
+ .kms_client(Arc::new(kms) as Arc<dyn KeyManagementClient>)
+ .table_key_id(table_key_id)
+ .build(),
+ )
+}
+
+/// Build a table backed by the V3 encryption fixture and an in-memory KMS,
+/// so it has an [`EncryptionManager`](crate::encryption::EncryptionManager).
+///
+/// The fixture's snapshot references an encrypted manifest list; its bytes
+/// (the `manifest-list-v3-encrypted.avro` testdata, an encrypted empty list)
+/// are seeded into the in-memory `FileIO` at that path so callers can read
+/// the current snapshot's manifest list.
+pub(crate) async fn make_encrypted_table() -> Table {
+ let metadata_json = std::fs::read_to_string(format!(
+ "{}/testdata/table_metadata/TableMetadataV3ValidEncryption.json",
+ env!("CARGO_MANIFEST_DIR"),
+ ))
+ .unwrap();
+ let metadata: TableMetadata = serde_json::from_str(&metadata_json).unwrap();
+
+ let kms: Arc<dyn KeyManagementClient> = {
+ let k = MemoryKeyManagementClient::new();
+ k.add_master_key_bytes(
+ "master-1",
+ SensitiveBytes::new([
+ 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d,
+ 0x0e, 0x0f,
+ ]),
+ )
+ .unwrap();
+ Arc::new(k)
+ };
+
+ let file_io = FileIO::new_with_memory();
+
+ // Seed the encrypted (empty) manifest list at the path the snapshot references.
+ let manifest_list_bytes = std::fs::read(format!(
+ "{}/testdata/manifests_lists/manifest-list-v3-encrypted.avro",
+ env!("CARGO_MANIFEST_DIR"),
+ ))
+ .unwrap();
+ file_io
+ .new_output(metadata.current_snapshot().unwrap().manifest_list())
+ .unwrap()
+ .write(manifest_list_bytes.into())
+ .await
+ .unwrap();
+
+ Table::builder()
+ .metadata(metadata)
+ .metadata_location("memory:///table/metadata/v1.json")
+ .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
+ .file_io(file_io)
+ .kms_client(kms)
+ .runtime(test_runtime())
+ .build()
+ .unwrap()
+}
diff --git a/crates/iceberg/src/test_utils/mod.rs b/crates/iceberg/src/test_utils/mod.rs
new file mode 100644
index 000000000..753a17e53
--- /dev/null
+++ b/crates/iceberg/src/test_utils/mod.rs
@@ -0,0 +1,33 @@
+// 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.
+
+//! Test utilities for this crate's own tests.
+//!
+//! Compiled under `cfg(test)` only, so nothing here is public API. Fixtures
+//! that other crates in the workspace need from their own tests live in the
+//! `iceberg_test_utils` crate instead.
+
+mod delete_vector;
+mod encryption;
+mod record_batch;
+mod runtime;
+pub(crate) mod scan;
+
+pub(crate) use delete_vector::encode_dv_blob;
+pub(crate) use encryption::{make_encrypted_table, make_encryption_manager};
+pub(crate) use record_batch::check_record_batches;
+pub(crate) use runtime::test_runtime;
diff --git a/crates/iceberg/src/test_utils/record_batch.rs b/crates/iceberg/src/test_utils/record_batch.rs
new file mode 100644
index 000000000..7e37f260d
--- /dev/null
+++ b/crates/iceberg/src/test_utils/record_batch.rs
@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Snapshot assertions over [`RecordBatch`] results.
+
+use arrow_array::RecordBatch;
+use expect_test::Expect;
+use itertools::Itertools;
+
+/// Snapshot testing to check the resulting record batch.
+///
+/// - `expected_schema/data`: put `expect![[""]]` as a placeholder,
+/// and then run test with `UPDATE_EXPECT=1 cargo test` to automatically update the result,
+/// or use rust-analyzer (see [video](https://github.com/rust-analyzer/expect-test)).
+/// Check the doc of [`expect_test`] for more details.
+/// - `ignore_check_columns`: Some columns are not stable, so we can skip them.
+/// - `sort_column`: The order of the data might be non-deterministic, so we can sort it by a column.
+pub(crate) fn check_record_batches(
+ record_batches: Vec<RecordBatch>,
+ expected_schema: Expect,
+ expected_data: Expect,
+ ignore_check_columns: &[&str],
+ sort_column: Option<&str>,
+) {
+ assert!(!record_batches.is_empty(), "Empty record batches");
+
+ // Combine record batches using the first batch's schema
+ let first_batch = record_batches.first().unwrap();
+ let record_batch =
+ arrow_select::concat::concat_batches(&first_batch.schema(), &record_batches).unwrap();
+
+ let mut columns = record_batch.columns().to_vec();
+ if let Some(sort_column) = sort_column {
+ let column = record_batch.column_by_name(sort_column).unwrap();
+ let indices = arrow_ord::sort::sort_to_indices(column, None, None).unwrap();
+ columns = columns
+ .iter()
+ .map(|column| arrow_select::take::take(column.as_ref(), &indices, None).unwrap())
+ .collect_vec();
+ }
+
+ expected_schema.assert_eq(&format!(
+ "{}",
+ record_batch.schema().fields().iter().format(",\n")
+ ));
+ expected_data.assert_eq(&format!(
+ "{}",
+ record_batch
+ .schema()
+ .fields()
+ .iter()
+ .zip_eq(columns)
+ .map(|(field, column)| {
+ if ignore_check_columns.contains(&field.name().as_str()) {
+ format!("{}: (skipped)", field.name())
+ } else {
+ format!("{}: {:?}", field.name(), column)
+ }
+ })
+ .format(",\n")
+ ));
+}
diff --git a/crates/iceberg/src/test_utils/runtime.rs b/crates/iceberg/src/test_utils/runtime.rs
new file mode 100644
index 000000000..2070e6cb2
--- /dev/null
+++ b/crates/iceberg/src/test_utils/runtime.rs
@@ -0,0 +1,45 @@
+// 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.
+
+//! A shared tokio runtime for tests that build a [`Table`](crate::table::Table).
+//!
+//! Twin of `iceberg_test_utils::test_runtime`, which serves other crates in the
+//! workspace. The two cannot be merged: a dev-dependency links a separate
+//! instance of this crate, so a `Runtime` built there is a distinct type from
+//! `crate::Runtime` here.
+
+use std::sync::OnceLock;
+
+use crate::runtime::Runtime;
+
+/// Returns a process-wide [`Runtime`] suitable for tests that need to construct
+/// a [`Table`](crate::table::Table) outside a tokio context.
+///
+/// The returned [`Runtime`] wraps a single shared multi-thread tokio runtime
+/// that is lazily built on first call and lives until process exit. Cloning is
+/// cheap, so test code can call this every time it needs a runtime to feed
+/// into [`TableBuilder::runtime`](crate::table::TableBuilder::runtime).
+pub(crate) fn test_runtime() -> Runtime {
+ static TOKIO_RT: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
+ let tokio_rt = TOKIO_RT.get_or_init(|| {
+ tokio::runtime::Builder::new_multi_thread()
+ .enable_all()
+ .build()
+ .expect("failed to build test tokio runtime")
+ });
+ Runtime::new(tokio_rt)
+}
diff --git a/crates/iceberg/src/test_utils/scan.rs b/crates/iceberg/src/test_utils/scan.rs
new file mode 100644
index 000000000..0c4a7e9dd
--- /dev/null
+++ b/crates/iceberg/src/test_utils/scan.rs
@@ -0,0 +1,1160 @@
+// 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.
+
+//! Shared test fixtures for the table scan API.
+
+use std::collections::HashMap;
+use std::fs;
+use std::fs::File;
+use std::sync::Arc;
+
+use arrow_array::cast::AsArray;
+use arrow_array::{
+ Array, ArrayRef, BooleanArray, Float64Array, Int32Array, Int64Array, RecordBatch, StringArray,
+};
+use minijinja::value::Value;
+use minijinja::{AutoEscape, Environment, context};
+use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY};
+use parquet::basic::Compression;
+use parquet::file::properties::WriterProperties;
+use tempfile::TempDir;
+use uuid::Uuid;
+
+use crate::TableIdent;
+use crate::io::{FileIO, OutputFile};
+use crate::metadata_columns::{
+ RESERVED_COL_NAME_DELETE_FILE_PATH, RESERVED_COL_NAME_DELETE_FILE_POS,
+ RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER, RESERVED_FIELD_ID_DELETE_FILE_PATH,
+ RESERVED_FIELD_ID_DELETE_FILE_POS,
+};
+use crate::spec::{
+ DataContentType, DataFileBuilder, DataFileFormat, FormatVersion, Literal, ManifestEntry,
+ ManifestListWriter, ManifestStatus, ManifestWriterBuilder, PartitionSpec, Struct, StructType,
+ TableMetadata, TableMetadataBuilder,
+};
+use crate::table::Table;
+use crate::test_utils::test_runtime;
+
+fn render_template(template: &str, ctx: Value) -> String {
+ let mut env = Environment::new();
+ env.set_auto_escape_callback(|_| AutoEscape::None);
+ env.render_str(template, ctx).unwrap()
+}
+
+/// Asserts every row of the `_last_updated_sequence_number` column across all
+/// batches equals `expected` (or is null when `expected` is `None`), decoding
+/// the logical value independent of the physical (run-end) encoding.
+pub(crate) fn assert_last_updated_seq_all(batches: &[RecordBatch], expected: Option<i64>) {
+ use arrow_cast::cast;
+ use arrow_schema::DataType;
+ for batch in batches {
+ let col = batch
+ .column_by_name(RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER)
+ .expect("_last_updated_sequence_number column should be present");
+ let logical = cast(col, &DataType::Int64).unwrap();
+ let values = logical.as_primitive::<arrow_array::types::Int64Type>();
+ for i in 0..values.len() {
+ let actual = (!values.is_null(i)).then(|| values.value(i));
+ assert_eq!(actual, expected, "row {i}");
+ }
+ }
+}
+
+pub(crate) struct TableTestFixture {
+ pub(crate) table_location: String,
+ pub(crate) table: Table,
+}
+
+impl TableTestFixture {
+ #[allow(clippy::new_without_default)]
+ pub(crate) fn new() -> Self {
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().join("table1");
+ let manifest_list1_location = table_location.join("metadata/manifests_list_1.avro");
+ let manifest_list2_location = table_location.join("metadata/manifests_list_2.avro");
+ let table_metadata1_location = table_location.join("metadata/v1.json");
+
+ let file_io = FileIO::new_with_fs();
+
+ let table_metadata = {
+ let template_json_str = fs::read_to_string(format!(
+ "{}/testdata/example_table_metadata_v2.json",
+ env!("CARGO_MANIFEST_DIR")
+ ))
+ .unwrap();
+ let metadata_json = render_template(&template_json_str, context! {
+ table_location => &table_location,
+ manifest_list_1_location => &manifest_list1_location,
+ manifest_list_2_location => &manifest_list2_location,
+ table_metadata_1_location => &table_metadata1_location,
+ });
+ serde_json::from_str::<TableMetadata>(&metadata_json).unwrap()
+ };
+
+ let table = Table::builder()
+ .metadata(table_metadata)
+ .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
+ .file_io(file_io.clone())
+ .metadata_location(table_metadata1_location.as_os_str().to_str().unwrap())
+ .runtime(test_runtime())
+ .build()
+ .unwrap();
+
+ Self {
+ table_location: table_location.to_str().unwrap().to_string(),
+ table,
+ }
+ }
+
+ #[allow(clippy::new_without_default)]
+ pub(crate) fn new_empty() -> Self {
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().join("table1");
+ let table_metadata1_location = table_location.join("metadata/v1.json");
+
+ let file_io = FileIO::new_with_fs();
+
+ let table_metadata = {
+ let template_json_str = fs::read_to_string(format!(
+ "{}/testdata/example_empty_table_metadata_v2.json",
+ env!("CARGO_MANIFEST_DIR")
+ ))
+ .unwrap();
+ let metadata_json = render_template(&template_json_str, context! {
+ table_location => &table_location,
+ table_metadata_1_location => &table_metadata1_location,
+ });
+ serde_json::from_str::<TableMetadata>(&metadata_json).unwrap()
+ };
+
+ let table = Table::builder()
+ .metadata(table_metadata)
+ .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
+ .file_io(file_io.clone())
+ .metadata_location(table_metadata1_location.as_os_str().to_str().unwrap())
+ .runtime(test_runtime())
+ .build()
+ .unwrap();
+
+ Self {
+ table_location: table_location.to_str().unwrap().to_string(),
+ table,
+ }
+ }
+
+ /// Creates a fixture with 5 snapshots chained as:
+ /// S1 (root) -> S2 -> S3 -> S4 -> S5 (current)
+ /// Useful for testing snapshot history traversal.
+ pub(crate) fn new_with_deep_history() -> Self {
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().join("table1");
+ let table_metadata1_location = table_location.join("metadata/v1.json");
+
+ let file_io = FileIO::new_with_fs();
+
+ let table_metadata = {
+ let json_str = fs::read_to_string(format!(
+ "{}/testdata/example_table_metadata_v2_deep_history.json",
+ env!("CARGO_MANIFEST_DIR")
+ ))
+ .unwrap();
+ serde_json::from_str::<TableMetadata>(&json_str).unwrap()
+ };
+
+ let table = Table::builder()
+ .metadata(table_metadata)
+ .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
+ .file_io(file_io.clone())
+ .metadata_location(table_metadata1_location.as_os_str().to_str().unwrap())
+ .runtime(test_runtime())
+ .build()
+ .unwrap();
+
+ Self {
+ table_location: table_location.to_str().unwrap().to_string(),
+ table,
+ }
+ }
+
+ pub(crate) fn new_unpartitioned() -> Self {
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().join("table1");
+ let manifest_list1_location = table_location.join("metadata/manifests_list_1.avro");
+ let manifest_list2_location = table_location.join("metadata/manifests_list_2.avro");
+ let table_metadata1_location = table_location.join("metadata/v1.json");
+
+ let file_io = FileIO::new_with_fs();
+
+ let mut table_metadata = {
+ let template_json_str = fs::read_to_string(format!(
+ "{}/testdata/example_table_metadata_v2.json",
+ env!("CARGO_MANIFEST_DIR")
+ ))
+ .unwrap();
+ let metadata_json = render_template(&template_json_str, context! {
+ table_location => &table_location,
+ manifest_list_1_location => &manifest_list1_location,
+ manifest_list_2_location => &manifest_list2_location,
+ table_metadata_1_location => &table_metadata1_location,
+ });
+ serde_json::from_str::<TableMetadata>(&metadata_json).unwrap()
+ };
+
+ table_metadata.default_spec = Arc::new(PartitionSpec::unpartition_spec());
+ table_metadata.partition_specs.clear();
+ table_metadata.default_partition_type = StructType::new(vec![]);
+ table_metadata
+ .partition_specs
+ .insert(0, table_metadata.default_spec.clone());
+
+ let table = Table::builder()
+ .metadata(table_metadata)
+ .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
+ .file_io(file_io.clone())
+ .metadata_location(table_metadata1_location.to_str().unwrap())
+ .runtime(test_runtime())
+ .build()
+ .unwrap();
+
+ Self {
+ table_location: table_location.to_str().unwrap().to_string(),
+ table,
+ }
+ }
+
+ pub(crate) fn new_with_partition_evolution() -> Self {
+ let table = Self::new().table;
+ let table_location = table.metadata().location.clone();
+
+ let manifest_list1_location = format!("{}/metadata/manifests_list_1.avro", table_location);
+ let manifest_list2_location = format!("{}/metadata/manifests_list_2.avro", table_location);
+ let manifest_list3_location = format!("{}/metadata/manifests_list_3.avro", table_location);
+ let table_metadata1_location = format!("{}/metadata/v1.json", table_location);
+
+ let new_table_metadata = {
+ let template_json_str = fs::read_to_string(format!(
+ "{}/testdata/example_table_metadata_v2_partition_evolution.json",
+ env!("CARGO_MANIFEST_DIR")
+ ))
+ .unwrap();
+ let metadata_json = render_template(&template_json_str, context! {
+ table_location => &table_location,
+ manifest_list_1_location => &manifest_list1_location,
+ manifest_list_2_location => &manifest_list2_location,
+ manifest_list_3_location => &manifest_list3_location,
+ table_metadata_1_location => &table_metadata1_location,
+ });
+ Arc::new(serde_json::from_str::<TableMetadata>(&metadata_json).unwrap())
+ };
+
+ Self {
+ table_location,
+ table: table.with_metadata(new_table_metadata),
+ }
+ }
+
+ fn next_manifest_file(&self) -> OutputFile {
+ self.table
+ .file_io()
+ .new_output(format!(
+ "{}/metadata/manifest_{}.avro",
+ self.table_location,
+ Uuid::new_v4()
+ ))
+ .unwrap()
+ }
+
+ pub(crate) async fn setup_manifest_files(&mut self) {
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let parent_snapshot = current_snapshot
+ .parent_snapshot(self.table.metadata())
+ .unwrap();
+ let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
+ let current_partition_spec = self.table.metadata().default_partition_spec();
+
+ // Write the data files first, then use the file size in the manifest entries
+ let parquet_file_size = self.write_parquet_data_files();
+
+ let mut writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_data();
+ writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/1.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(Struct::from_iter([Some(Literal::long(100))]))
+ .key_metadata(None)
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ writer
+ .add_delete_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Deleted)
+ .snapshot_id(parent_snapshot.snapshot_id())
+ .sequence_number(parent_snapshot.sequence_number())
+ .file_sequence_number(parent_snapshot.sequence_number())
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/2.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(Struct::from_iter([Some(Literal::long(200))]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ writer
+ .add_existing_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Existing)
+ .snapshot_id(parent_snapshot.snapshot_id())
+ .sequence_number(parent_snapshot.sequence_number())
+ .file_sequence_number(parent_snapshot.sequence_number())
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/3.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(Struct::from_iter([Some(Literal::long(300))]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let data_file_manifest = writer.write_manifest_file().await.unwrap();
+
+ // Write to manifest list
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v2(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ );
+ manifest_list_write
+ .add_manifests(vec![data_file_manifest].into_iter())
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+ }
+
+ /// Writes a v3 data manifest with a manifest-level `first_row_id` of 42,
+ /// so live entries inherit a per-file `first_row_id` on read. Upgrades the
+ /// table to v3 first, so the manifest list is read as v3.
+ pub(crate) async fn setup_v3_manifest_files(&mut self) {
+ let metadata = TableMetadataBuilder::new_from_metadata(
+ self.table.metadata().clone(),
+ self.table.metadata_location().map(str::to_string),
+ )
+ .upgrade_format_version(FormatVersion::V3)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ self.table = Table::builder()
+ .metadata(metadata)
+ .identifier(self.table.identifier().clone())
+ .file_io(self.table.file_io().clone())
+ .metadata_location(self.table.metadata_location().unwrap().to_string())
+ .runtime(test_runtime())
+ .build()
+ .unwrap();
+
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
+ let current_partition_spec = self.table.metadata().default_partition_spec();
+
+ let parquet_file_size = self.write_parquet_data_files();
+
+ let mut writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v3_data();
+ writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/1.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(Struct::from_iter([Some(Literal::long(100))]))
+ .key_metadata(None)
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let data_file_manifest = writer.write_manifest_file().await.unwrap();
+
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v3(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ Some(42),
+ );
+ manifest_list_write
+ .add_manifests(vec![data_file_manifest].into_iter())
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+ }
+
+ pub(crate) async fn setup_manifest_files_with_partition_evolution(&mut self) {
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let parent_snapshot = current_snapshot
+ .parent_snapshot(self.table.metadata())
+ .unwrap();
+ let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
+ let current_partition_spec = self.table.metadata().default_partition_spec();
+
+ // Write the data files first, then use the file size in the manifest entries
+ let parquet_file_size = self.write_parquet_data_files();
+
+ let mut writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_data();
+ writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(1)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/1.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(Struct::from_iter([
+ Some(Literal::long(100)),
+ Some(Literal::string("apa")),
+ Some(Literal::int(27)),
+ ]))
+ .key_metadata(None)
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ writer
+ .add_delete_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Deleted)
+ .snapshot_id(parent_snapshot.snapshot_id())
+ .sequence_number(parent_snapshot.sequence_number())
+ .file_sequence_number(parent_snapshot.sequence_number())
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(1)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/2.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(Struct::from_iter([
+ Some(Literal::long(200)),
+ Some(Literal::string("ice")),
+ Some(Literal::int(5)),
+ ]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ writer
+ .add_existing_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Existing)
+ .snapshot_id(parent_snapshot.snapshot_id())
+ .sequence_number(parent_snapshot.sequence_number())
+ .file_sequence_number(parent_snapshot.sequence_number())
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(1)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/3.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(Struct::from_iter([
+ Some(Literal::long(300)),
+ Some(Literal::string("apa")),
+ Some(Literal::int(19)),
+ ]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let data_file_manifest = writer.write_manifest_file().await.unwrap();
+
+ // Write to manifest list
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v2(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ );
+ manifest_list_write
+ .add_manifests(vec![data_file_manifest].into_iter())
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+ }
+
+ /// Writes identical Parquet data files (1.parquet, 2.parquet, 3.parquet)
+ /// and returns the file size in bytes.
+ fn write_parquet_data_files(&self) -> u64 {
+ fs::create_dir_all(&self.table_location).unwrap();
+
+ let schema = {
+ let fields = vec![
+ arrow_schema::Field::new("x", arrow_schema::DataType::Int64, false).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string())]),
+ ),
+ arrow_schema::Field::new("y", arrow_schema::DataType::Int64, false).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "2".to_string())]),
+ ),
+ arrow_schema::Field::new("z", arrow_schema::DataType::Int64, false).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "3".to_string())]),
+ ),
+ arrow_schema::Field::new("a", arrow_schema::DataType::Utf8, false).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "4".to_string())]),
+ ),
+ arrow_schema::Field::new("dbl", arrow_schema::DataType::Float64, false)
+ .with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "5".to_string(),
+ )])),
+ arrow_schema::Field::new("i32", arrow_schema::DataType::Int32, false)
+ .with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "6".to_string(),
+ )])),
+ arrow_schema::Field::new("i64", arrow_schema::DataType::Int64, false)
+ .with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "7".to_string(),
+ )])),
+ arrow_schema::Field::new("bool", arrow_schema::DataType::Boolean, false)
+ .with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "8".to_string(),
+ )])),
+ ];
+ Arc::new(arrow_schema::Schema::new(fields))
+ };
+ // x: [1, 1, 1, 1, ...]
+ let col1 = Arc::new(Int64Array::from_iter_values(vec![1; 1024])) as ArrayRef;
+
+ let mut values = vec![2; 512];
+ values.append(vec![3; 200].as_mut());
+ values.append(vec![4; 300].as_mut());
+ values.append(vec![5; 12].as_mut());
+
+ // y: [2, 2, 2, 2, ..., 3, 3, 3, 3, ..., 4, 4, 4, 4, ..., 5, 5, 5, 5]
+ let col2 = Arc::new(Int64Array::from_iter_values(values)) as ArrayRef;
+
+ let mut values = vec![3; 512];
+ values.append(vec![4; 512].as_mut());
+
+ // z: [3, 3, 3, 3, ..., 4, 4, 4, 4]
+ let col3 = Arc::new(Int64Array::from_iter_values(values)) as ArrayRef;
+
+ // a: ["Apache", "Apache", "Apache", ..., "Iceberg", "Iceberg", "Iceberg"]
+ let mut values = vec!["Apache"; 512];
+ values.append(vec!["Iceberg"; 512].as_mut());
+ let col4 = Arc::new(StringArray::from_iter_values(values)) as ArrayRef;
+
+ // dbl:
+ let mut values = vec![100.0f64; 512];
+ values.append(vec![150.0f64; 12].as_mut());
+ values.append(vec![200.0f64; 500].as_mut());
+ let col5 = Arc::new(Float64Array::from_iter_values(values)) as ArrayRef;
+
+ // i32:
+ let mut values = vec![100i32; 512];
+ values.append(vec![150i32; 12].as_mut());
+ values.append(vec![200i32; 500].as_mut());
+ let col6 = Arc::new(Int32Array::from_iter_values(values)) as ArrayRef;
+
+ // i64:
+ let mut values = vec![100i64; 512];
+ values.append(vec![150i64; 12].as_mut());
+ values.append(vec![200i64; 500].as_mut());
+ let col7 = Arc::new(Int64Array::from_iter_values(values)) as ArrayRef;
+
+ // bool:
+ let mut values = vec![false; 512];
+ values.append(vec![true; 512].as_mut());
+ let values: BooleanArray = values.into();
+ let col8 = Arc::new(values) as ArrayRef;
+
+ let to_write = RecordBatch::try_new(schema.clone(), vec![
+ col1, col2, col3, col4, col5, col6, col7, col8,
+ ])
+ .unwrap();
+
+ // Write the Parquet files
+ let props = WriterProperties::builder()
+ .set_compression(Compression::SNAPPY)
+ .build();
+
+ for n in 1..=3 {
+ let file = File::create(format!("{}/{}.parquet", &self.table_location, n)).unwrap();
+ let mut writer =
+ ArrowWriter::try_new(file, to_write.schema(), Some(props.clone())).unwrap();
+
+ writer.write(&to_write).expect("Writing batch");
+
+ // writer must be closed to write footer
+ writer.close().unwrap();
+ }
+
+ fs::metadata(format!("{}/1.parquet", &self.table_location))
+ .unwrap()
+ .len()
+ }
+
+ pub(crate) async fn setup_unpartitioned_manifest_files(&mut self) {
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let parent_snapshot = current_snapshot
+ .parent_snapshot(self.table.metadata())
+ .unwrap();
+ let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
+ let current_partition_spec = Arc::new(PartitionSpec::unpartition_spec());
+
+ // Write the data files first, then use the file size in the manifest entries
+ let parquet_file_size = self.write_parquet_data_files();
+
+ // Write data files using an empty partition for unpartitioned tables.
+ let mut writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_data();
+
+ // Create an empty partition value.
+ let empty_partition = Struct::empty();
+
+ writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/1.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(empty_partition.clone())
+ .key_metadata(None)
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+
+ writer
+ .add_delete_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Deleted)
+ .snapshot_id(parent_snapshot.snapshot_id())
+ .sequence_number(parent_snapshot.sequence_number())
+ .file_sequence_number(parent_snapshot.sequence_number())
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/2.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(empty_partition.clone())
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+
+ writer
+ .add_existing_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Existing)
+ .snapshot_id(parent_snapshot.snapshot_id())
+ .sequence_number(parent_snapshot.sequence_number())
+ .file_sequence_number(parent_snapshot.sequence_number())
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/3.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(empty_partition.clone())
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+
+ let data_file_manifest = writer.write_manifest_file().await.unwrap();
+
+ // Write to manifest list
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v2(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ );
+ manifest_list_write
+ .add_manifests(vec![data_file_manifest].into_iter())
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+ }
+
+ pub(crate) async fn setup_deadlock_manifests(&mut self) {
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let _parent_snapshot = current_snapshot
+ .parent_snapshot(self.table.metadata())
+ .unwrap();
+ let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
+ let current_partition_spec = self.table.metadata().default_partition_spec();
+
+ // 1. Write DATA manifest with MULTIPLE entries to fill buffer
+ let mut writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_data();
+
+ // Add 10 data entries
+ for i in 0..10 {
+ writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/{}.parquet", &self.table_location, i))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(1)
+ .partition(Struct::from_iter([Some(Literal::long(100))]))
+ .key_metadata(None)
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ }
+ let data_manifest = writer.write_manifest_file().await.unwrap();
+
+ // 2. Write DELETE manifest
+ let mut writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_deletes();
+
+ writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::PositionDeletes)
+ .file_path(format!("{}/del.parquet", &self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(1)
+ .partition(Struct::from_iter([Some(Literal::long(100))]))
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let delete_manifest = writer.write_manifest_file().await.unwrap();
+
+ // Write to manifest list - DATA FIRST then DELETE
+ // This order is crucial for reproduction
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v2(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ );
+ manifest_list_write
+ .add_manifests(vec![data_manifest, delete_manifest].into_iter())
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+ }
+
+ /// Sets up a single data file `mrg.parquet` with three 100-row row groups
+ /// (column `x` = 1000..1300, so row position `p` carries `x = 1000 + p`) and
+ /// registers it in the current snapshot. When `delete_positions` is non-empty,
+ /// also writes a positional delete file targeting those file-absolute positions
+ /// and registers it in a delete manifest.
+ ///
+ /// Used to exercise the `_pos` metadata column through the real `TableScan`
+ /// planning path across row-group boundaries and (optionally) positional deletes.
+ pub(crate) async fn setup_multi_row_group_manifest(&mut self, delete_positions: &[i64]) {
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
+ let current_partition_spec = self.table.metadata().default_partition_spec();
+
+ // The table's spec 0 is identity on `x`, so give the data and delete files a
+ // fixed partition value. Filter tests deliberately filter on `y` (a
+ // non-partition column) so pruning is driven by Parquet row-group statistics
+ // rather than partition values.
+ let partition = Struct::from_iter([Some(Literal::long(1000))]);
+
+ let (data_file_path, data_file_size) = self.write_multi_row_group_data_file();
+
+ let mut data_writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_data();
+ data_writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(data_file_path.clone())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(data_file_size)
+ .record_count(300)
+ .partition(partition.clone())
+ .key_metadata(None)
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let data_manifest = data_writer.write_manifest_file().await.unwrap();
+
+ let mut manifests = vec![data_manifest];
+
+ if !delete_positions.is_empty() {
+ let (del_path, del_size) =
+ self.write_positional_delete_file(&data_file_path, delete_positions);
+
+ let mut delete_writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_deletes();
+ delete_writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::PositionDeletes)
+ .file_path(del_path)
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(del_size)
+ .record_count(delete_positions.len() as u64)
+ .partition(partition.clone())
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ manifests.push(delete_writer.write_manifest_file().await.unwrap());
+ }
+
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v2(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ );
+ manifest_list_write
+ .add_manifests(manifests.into_iter())
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+ }
+
+ /// Writes a manifest with three live "Added" data-file entries (partitioned on `x`
+ /// = 100, 200, 300), each with the given `sort_order_id` set on its `DataFile`
+ /// (`None` leaves the field unset). Used to test how `sort_order_id` resolution
+ /// against the table's sort orders flows into each entry's `FileScanTask`.
+ pub(crate) async fn setup_manifest_files_with_sort_order_ids(
+ &mut self,
+ sort_order_ids: [Option<i32>; 4],
+ ) {
+ let current_snapshot = self.table.metadata().current_snapshot().unwrap();
+ let current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
+ let current_partition_spec = self.table.metadata().default_partition_spec();
+ let parquet_file_size = self.write_parquet_data_files();
+
+ let mut writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v2_data();
+
+ for (i, sort_order_id) in sort_order_ids.into_iter().enumerate() {
+ let mut data_file_builder = DataFileBuilder::default();
+ data_file_builder
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/{}.parquet", &self.table_location, i + 1))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+ .partition(Struct::from_iter([Some(Literal::long(
+ 100 * (i as i64 + 1),
+ ))]));
+ if let Some(id) = sort_order_id {
+ data_file_builder.sort_order_id(id);
+ }
+ let data_file = data_file_builder.build().unwrap();
+
+ writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(data_file)
+ .build(),
+ )
+ .unwrap();
+ }
+
+ let data_file_manifest = writer.write_manifest_file().await.unwrap();
+
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v2(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ );
+ manifest_list_write
+ .add_manifests(std::iter::once(data_file_manifest))
+ .unwrap();
+ manifest_list_write.close().await.unwrap();
+ }
+
+ /// Writes `mrg.parquet` with three 100-row row groups. Columns `x` (field
+ /// id `1`) and `y` (field id `2`) both run 1000..1300, so row position `p`
+ /// carries `x = y = 1000 + p`. Returns `(path, file_size_in_bytes)`.
+ fn write_multi_row_group_data_file(&self) -> (String, u64) {
+ fs::create_dir_all(&self.table_location).unwrap();
+
+ let arrow_schema = Arc::new(arrow_schema::Schema::new(vec![
+ arrow_schema::Field::new("x", arrow_schema::DataType::Int64, false).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string())]),
+ ),
+ arrow_schema::Field::new("y", arrow_schema::DataType::Int64, false).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "2".to_string())]),
+ ),
+ ]));
+
+ let path = format!("{}/mrg.parquet", &self.table_location);
+ let max_row_group_row_count = 100;
+ let props = WriterProperties::builder()
+ .set_compression(Compression::SNAPPY)
+ .set_max_row_group_row_count(Some(max_row_group_row_count))
+ .build();
+
+ let file = File::create(&path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, arrow_schema.clone(), Some(props)).unwrap();
+ for group in 0..3i64 {
+ let base = 1000 + group * max_row_group_row_count as i64;
+ let col = Arc::new(Int64Array::from_iter_values(
+ base..base + max_row_group_row_count as i64,
+ )) as ArrayRef;
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![col.clone(), col]).unwrap();
+ writer.write(&batch).unwrap();
+ }
+ writer.close().unwrap();
+
+ let size = fs::metadata(&path).unwrap().len();
+ (path, size)
+ }
+
+ /// Writes a positional delete file targeting `positions` in `data_path`.
+ /// Returns `(path, file_size_in_bytes)`.
+ fn write_positional_delete_file(&self, data_path: &str, positions: &[i64]) -> (String, u64) {
+ let del_schema = Arc::new(arrow_schema::Schema::new(vec![
+ arrow_schema::Field::new(
+ RESERVED_COL_NAME_DELETE_FILE_PATH,
+ arrow_schema::DataType::Utf8,
+ false,
+ )
+ .with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ RESERVED_FIELD_ID_DELETE_FILE_PATH.to_string(), // 2147483546
+ )])),
+ arrow_schema::Field::new(
+ RESERVED_COL_NAME_DELETE_FILE_POS,
+ arrow_schema::DataType::Int64,
+ false,
+ )
+ .with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ RESERVED_FIELD_ID_DELETE_FILE_POS.to_string(), // 2147483545
+ )])),
+ ]));
+
+ let batch = RecordBatch::try_new(del_schema.clone(), vec![
+ Arc::new(StringArray::from_iter_values(std::iter::repeat_n(
+ data_path.to_string(),
+ positions.len(),
+ ))) as ArrayRef,
+ Arc::new(Int64Array::from_iter_values(positions.iter().copied())) as ArrayRef,
+ ])
+ .unwrap();
+
+ let path = format!("{}/pos-del.parquet", &self.table_location);
+ let props = WriterProperties::builder()
+ .set_compression(Compression::SNAPPY)
+ .build();
+ let file = File::create(&path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, del_schema, Some(props)).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+
+ let size = fs::metadata(&path).unwrap().len();
+ (path, size)
+ }
+}
diff --git a/crates/iceberg/src/util/snapshot.rs b/crates/iceberg/src/util/snapshot.rs
index d5b87a340..deac53ef8 100644
--- a/crates/iceberg/src/util/snapshot.rs
+++ b/crates/iceberg/src/util/snapshot.rs
@@ -79,7 +79,7 @@ pub fn ancestors_between(
#[cfg(test)]
mod tests {
use super::*;
- use crate::scan::tests::TableTestFixture;
+ use crate::test_utils::scan::TableTestFixture;
// Five snapshots chained as: S1 (root) -> S2 -> S3 -> S4 -> S5 (current)
const S1: i64 = 3051729675574597004;
diff --git a/crates/test_utils/Cargo.toml b/crates/test_utils/Cargo.toml
index eae0b32a4..41f4e38dd 100644
--- a/crates/test_utils/Cargo.toml
+++ b/crates/test_utils/Cargo.toml
@@ -28,6 +28,7 @@ repository = { workspace = true }
[dependencies]
iceberg = { workspace = true }
+tokio = { workspace = true }
tracing-subscriber = { workspace = true }
[features]
diff --git a/crates/test_utils/src/lib.rs b/crates/test_utils/src/lib.rs
index b64e46269..bf2dad0ed 100644
--- a/crates/test_utils/src/lib.rs
+++ b/crates/test_utils/src/lib.rs
@@ -19,8 +19,13 @@
//!
//! It's not intended for use outside of `iceberg-rust`.
+#[cfg(feature = "tests")]
+mod runtime;
+
#[cfg(feature = "tests")]
pub use common::*;
+#[cfg(feature = "tests")]
+pub use runtime::test_runtime;
#[cfg(feature = "tests")]
mod common {
diff --git a/crates/test_utils/src/runtime.rs b/crates/test_utils/src/runtime.rs
new file mode 100644
index 000000000..28568b67f
--- /dev/null
+++ b/crates/test_utils/src/runtime.rs
@@ -0,0 +1,40 @@
+// 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.
+
+//! A shared tokio runtime for tests that build an [`iceberg::table::Table`].
+
+use std::sync::OnceLock;
+
+use iceberg::Runtime;
+
+/// Returns a process-wide [`Runtime`] suitable for tests that need to construct
+/// a [`Table`](iceberg::table::Table) outside a tokio context.
+///
+/// The returned [`Runtime`] wraps a single shared multi-thread tokio runtime
+/// that is lazily built on first call and lives until process exit. Cloning is
+/// cheap, so test code can call this every time it needs a runtime to feed
+/// into [`TableBuilder::runtime`](iceberg::table::TableBuilder::runtime).
+pub fn test_runtime() -> Runtime {
+ static TOKIO_RT: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
+ let tokio_rt = TOKIO_RT.get_or_init(|| {
+ tokio::runtime::Builder::new_multi_thread()
+ .enable_all()
+ .build()
+ .expect("failed to build test tokio runtime")
+ });
+ Runtime::new(tokio_rt)
+}