This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 05e277e Add CI (#6)
05e277e is described below
commit 05e277e8a8bf342db53cbf2d53ed6f2098b44a88
Author: Gabriel <[email protected]>
AuthorDate: Tue Sep 22 13:09:11 2026 +0200
Add CI (#6)
* Add basic CI for lint, test and format
* Format the codebase
* Fix rust toolchain version to 1.98.1
* Better concurrency control for pipelines
* Commit rust-toolchain.toml
* Commit rustfmt.toml
* Format code
* Improve CARGO_INCREMENTAL comment
* Add checks as required to asf.yaml
---
.asf.yaml | 7 +-
.github/actions/setup-rust/action.yml | 44 +++++
.github/dependabot.yml | 55 ++++++
.github/workflows/asf-allowlist-check.yml | 43 +++++
.github/workflows/ci.yml | 84 +++++++++
crates/datafusion/src/physical_plan/commit.rs | 103 +++++++----
.../src/physical_plan/expr_to_predicate.rs | 136 ++++++++------
.../datafusion/src/physical_plan/metadata_scan.rs | 4 +-
crates/datafusion/src/physical_plan/project.rs | 111 ++++++++----
crates/datafusion/src/physical_plan/repartition.rs | 42 +++--
crates/datafusion/src/physical_plan/sort.rs | 35 ++--
crates/datafusion/src/physical_plan/write.rs | 135 ++++++++------
crates/datafusion/src/schema.rs | 70 +++++---
crates/datafusion/src/table/mod.rs | 195 +++++++++++++--------
.../datafusion/src/table/table_provider_factory.rs | 4 +-
crates/datafusion/src/task_writer.rs | 104 +++++++----
.../tests/integration_datafusion_test.rs | 62 +++++--
crates/playground/src/catalog.rs | 10 +-
crates/playground/src/main.rs | 3 +-
crates/sqllogictest/src/engine/datafusion.rs | 30 +++-
crates/sqllogictest/src/engine/mod.rs | 7 +-
crates/sqllogictest/src/schedule.rs | 7 +-
rust-toolchain.toml | 21 +++
rustfmt.toml | 19 ++
24 files changed, 970 insertions(+), 361 deletions(-)
diff --git a/.asf.yaml b/.asf.yaml
index 2200ca8..426ac8d 100644
--- a/.asf.yaml
+++ b/.asf.yaml
@@ -46,5 +46,8 @@ github:
required_status_checks:
# Require branches to be up-to-date before merging.
strict: true
- # Do not require any jobs to pass.
- contexts: []
+ contexts:
+ - "Format"
+ - "Clippy"
+ - "Test"
+ - "asf-allowlist-check"
diff --git a/.github/actions/setup-rust/action.yml
b/.github/actions/setup-rust/action.yml
new file mode 100644
index 0000000..84a54e2
--- /dev/null
+++ b/.github/actions/setup-rust/action.yml
@@ -0,0 +1,44 @@
+# 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.
+
+name: Setup Rust
+description: Install the Rust toolchain and optional components used by CI.
+
+inputs:
+ cache:
+ description: Restore Rust dependencies and save them on main-branch pushes.
+ required: false
+ default: "false"
+ components:
+ description: Comma-separated Rust components to install.
+ required: false
+
+runs:
+ using: composite
+ steps:
+ - name: Install Rust toolchain
+ uses: dtolnay/rust-toolchain@4cda84d5c5c54efe2404f9d843567869ab1699d4
+ with:
+ toolchain: 1.98.1
+ components: ${{ inputs.components }}
+ - name: Cache Rust artifacts
+ if: inputs.cache == 'true'
+ uses: swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 #
v2.9.2
+ with:
+ # Pull requests may restore the main-branch cache, but only trusted
+ # main-branch runs may create or update cache entries.
+ save-if: ${{ github.event_name == 'push' && github.ref ==
'refs/heads/main' }}
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
new file mode 100644
index 0000000..3216549
--- /dev/null
+++ b/.github/dependabot.yml
@@ -0,0 +1,55 @@
+# 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.
+
+version: 2
+updates:
+ # Keep routine Cargo updates current while reserving ecosystem-coupled major
+ # upgrades for an intentional compatibility review, as DataFusion does.
+ - package-ecosystem: cargo
+ directory: "/"
+ schedule:
+ interval: weekly
+ target-branch: main
+ open-pull-requests-limit: 15
+ ignore:
+ - dependency-name: "datafusion*"
+ update-types: ["version-update:semver-major"]
+ - dependency-name: "parquet"
+ update-types: ["version-update:semver-major"]
+ groups:
+ all-other-cargo-deps:
+ applies-to: version-updates
+ patterns:
+ - "*"
+ exclude-patterns:
+ - "datafusion*"
+ - "parquet"
+ update-types:
+ - minor
+ - patch
+
+ # The glob covers local composite actions such as setup-rust as well as
+ # workflow files in the repository root.
+ - package-ecosystem: github-actions
+ directories: ["/", "/.github/actions/*"]
+ schedule:
+ interval: weekly
+ open-pull-requests-limit: 10
+ groups:
+ github-actions:
+ patterns:
+ - "*"
diff --git a/.github/workflows/asf-allowlist-check.yml
b/.github/workflows/asf-allowlist-check.yml
new file mode 100644
index 0000000..9a75b1b
--- /dev/null
+++ b/.github/workflows/asf-allowlist-check.yml
@@ -0,0 +1,43 @@
+# 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.
+
+# Taken from
https://github.com/apache/infrastructure-actions/tree/main/allowlist-check
+
+name: "ASF Allowlist Check"
+
+on:
+ workflow_dispatch:
+ pull_request:
+ paths:
+ - ".github/**"
+ push:
+ branches:
+ - main
+ paths:
+ - ".github/**"
+
+permissions:
+ contents: read
+
+jobs:
+ asf-allowlist-check:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v4
+ with:
+ persist-credentials: false
+ - uses: apache/infrastructure-actions/allowlist-check@main
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..89c0545
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,84 @@
+# 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.
+
+name: CI
+
+on:
+ pull_request:
+ push:
+ branches:
+ - main
+ merge_group:
+
+# Cancel obsolete runs for the same pull request or branch. The three jobs
below
+# keep this workflow well below ASF's 15-job recommended concurrency limit.
+concurrency:
+ group: ${{ github.workflow }}-${{ github.event.pull_request.number ||
github.sha }}
+ cancel-in-progress: true
+
+# CI needs no credentials beyond reading the source tree.
+permissions:
+ contents: read
+
+env:
+ CARGO_TERM_COLOR: always
+
+jobs:
+ format:
+ name: Format
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 #
v7.0.1
+ with:
+ persist-credentials: false
+ - name: Setup Rust
+ uses: ./.github/actions/setup-rust
+ with:
+ components: rustfmt
+ - run: cargo fmt --all -- --check
+
+ clippy:
+ name: Clippy
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 #
v7.0.1
+ with:
+ persist-credentials: false
+ - name: Setup Rust
+ uses: ./.github/actions/setup-rust
+ with:
+ components: clippy
+ cache: "true"
+ - run: cargo clippy --workspace --locked --all-targets -- -D warnings
+
+ test:
+ name: Test
+ runs-on: ubuntu-latest
+ env:
+ # Incremental artifacts and debug info go unused in CI and only
+ # add cache size and build time.
+ CARGO_INCREMENTAL: "0"
+ CARGO_PROFILE_DEV_DEBUG: "0"
+ steps:
+ - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 #
v7.0.1
+ with:
+ persist-credentials: false
+ - name: Setup Rust
+ uses: ./.github/actions/setup-rust
+ with:
+ cache: "true"
+ - run: cargo test --workspace --locked
diff --git a/crates/datafusion/src/physical_plan/commit.rs
b/crates/datafusion/src/physical_plan/commit.rs
index bb50721..d080f61 100644
--- a/crates/datafusion/src/physical_plan/commit.rs
+++ b/crates/datafusion/src/physical_plan/commit.rs
@@ -28,7 +28,9 @@ use datafusion::execution::{SendableRecordBatchStream,
TaskContext};
use datafusion::physical_expr::{EquivalenceProperties, Partitioning,
PhysicalExpr};
use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
-use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan,
PlanProperties};
+use datafusion::physical_plan::{
+ DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+};
use futures::StreamExt;
use iceberg::Catalog;
use iceberg::spec::{DataFile, deserialize_data_file_from_json};
@@ -85,12 +87,13 @@ impl IcebergCommitExec {
fn make_count_batch(count: u64) -> DFResult<RecordBatch> {
let count_array = Arc::new(UInt64Array::from(vec![count])) as ArrayRef;
- RecordBatch::try_from_iter_with_nullable(vec![("count", count_array,
false)]).map_err(|e| {
- DataFusionError::ArrowError(
- Box::new(e),
- Some("Failed to make count batch!".to_string()),
- )
- })
+ RecordBatch::try_from_iter_with_nullable(vec![("count", count_array,
false)])
+ .map_err(|e| {
+ DataFusionError::ArrowError(
+ Box::new(e),
+ Some("Failed to make count batch!".to_string()),
+ )
+ })
}
fn make_count_schema() -> ArrowSchemaRef {
@@ -144,8 +147,13 @@ impl ExecutionPlan for IcebergCommitExec {
Ok(TreeNodeRecursion::Continue)
}
- fn required_input_distribution(&self) ->
Vec<datafusion::physical_plan::Distribution> {
- vec![datafusion::physical_plan::Distribution::SinglePartition;
self.children().len()]
+ fn required_input_distribution(
+ &self,
+ ) -> Vec<datafusion::physical_plan::Distribution> {
+ vec![
+ datafusion::physical_plan::Distribution::SinglePartition;
+ self.children().len()
+ ]
}
fn benefits_from_input_partitioning(&self) -> Vec<bool> {
@@ -236,7 +244,8 @@ impl ExecutionPlan for IcebergCommitExec {
.collect::<datafusion::common::Result<_>>()?;
// add record_counts from the current batch to total record
count
- total_record_count += batch_files.iter().map(|f|
f.record_count()).sum::<u64>();
+ total_record_count +=
+ batch_files.iter().map(|f| f.record_count()).sum::<u64>();
// Add all deserialized files to our collection
data_files.extend(batch_files);
@@ -276,7 +285,9 @@ mod tests {
use std::fmt;
use std::sync::Arc;
- use datafusion::arrow::array::{ArrayRef, Int32Array, RecordBatch,
StringArray, UInt64Array};
+ use datafusion::arrow::array::{
+ ArrayRef, Int32Array, RecordBatch, StringArray, UInt64Array,
+ };
use datafusion::arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
use datafusion::datasource::MemTable;
use datafusion::execution::context::TaskContext;
@@ -284,13 +295,15 @@ mod tests {
use datafusion::physical_plan::common::collect;
use datafusion::physical_plan::execution_plan::Boundedness;
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
- use datafusion::physical_plan::{DisplayAs, DisplayFormatType,
ExecutionPlan, PlanProperties};
+ use datafusion::physical_plan::{
+ DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+ };
use datafusion::prelude::*;
use futures::StreamExt;
use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
use iceberg::spec::{
- DataContentType, DataFileBuilder, DataFileFormat, NestedField,
PrimitiveType, Schema,
- Struct, Type,
+ DataContentType, DataFileBuilder, DataFileFormat, NestedField,
PrimitiveType,
+ Schema, Struct, Type,
};
use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation,
TableIdent};
@@ -366,7 +379,8 @@ mod tests {
_context: Arc<TaskContext>,
) -> datafusion::common::Result<SendableRecordBatchStream> {
// Create a record batch with the serialized data files
- let array =
Arc::new(StringArray::from(self.data_files_json.clone())) as ArrayRef;
+ let array =
+ Arc::new(StringArray::from(self.data_files_json.clone())) as
ArrayRef;
let batch = RecordBatch::try_new(self.schema.clone(),
vec![array])?;
// Create a stream that returns this batch
@@ -415,8 +429,10 @@ mod tests {
let schema = Schema::builder()
.with_schema_id(1)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()?;
@@ -466,7 +482,8 @@ mod tests {
)?;
// Create a mock execution plan that returns the serialized data files
- let input_exec = Arc::new(MockWriteExec::new(vec![data_file1_json,
data_file2_json]));
+ let input_exec =
+ Arc::new(MockWriteExec::new(vec![data_file1_json,
data_file2_json]));
// Create the IcebergCommitExec
let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new(
@@ -475,8 +492,12 @@ mod tests {
false,
)]));
- let commit_exec =
- IcebergCommitExec::new(table.clone(), catalog.clone(), input_exec,
arrow_schema);
+ let commit_exec = IcebergCommitExec::new(
+ table.clone(),
+ catalog.clone(),
+ input_exec,
+ arrow_schema,
+ );
// Verify Execution Plan schema matches the count schema
assert_eq!(commit_exec.schema(),
IcebergCommitExec::make_count_schema());
@@ -537,7 +558,8 @@ mod tests {
}
#[tokio::test]
- async fn test_iceberg_commit_exec_empty_insert() -> Result<(), Box<dyn
std::error::Error>> {
+ async fn test_iceberg_commit_exec_empty_insert()
+ -> Result<(), Box<dyn std::error::Error>> {
let catalog = Arc::new(
MemoryCatalogBuilder::default()
.load(
@@ -557,7 +579,8 @@ mod tests {
let schema = Schema::builder()
.with_schema_id(1)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
])
.build()?;
@@ -578,8 +601,12 @@ mod tests {
DataType::Utf8,
false,
)]));
- let commit_exec =
- IcebergCommitExec::new(table.clone(), catalog.clone(), input_exec,
arrow_schema);
+ let commit_exec = IcebergCommitExec::new(
+ table.clone(),
+ catalog.clone(),
+ input_exec,
+ arrow_schema,
+ );
let task_ctx = Arc::new(TaskContext::default());
let stream = commit_exec.execute(0, task_ctx)?;
@@ -600,7 +627,8 @@ mod tests {
// No new snapshot should be created for an empty insert
let updated_table = catalog
.load_table(
- &TableIdent::from_strs(["test_empty_insert",
"empty_insert_table"]).unwrap(),
+ &TableIdent::from_strs(["test_empty_insert",
"empty_insert_table"])
+ .unwrap(),
)
.await?;
let snapshot_count_after = updated_table.metadata().snapshots().len();
@@ -614,8 +642,8 @@ mod tests {
}
#[tokio::test]
- async fn test_datafusion_execution_partitioned_source() -> Result<(),
Box<dyn std::error::Error>>
- {
+ async fn test_datafusion_execution_partitioned_source()
+ -> Result<(), Box<dyn std::error::Error>> {
let catalog = Arc::new(
MemoryCatalogBuilder::default()
.load(
@@ -634,8 +662,10 @@ mod tests {
let schema = Schema::builder()
.with_schema_id(1)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()?;
@@ -655,10 +685,14 @@ mod tests {
let batches: Vec<RecordBatch> = (1..4)
.map(|idx| {
- RecordBatch::try_new(arrow_schema.clone(), vec![
- Arc::new(Int32Array::from(vec![idx])) as ArrayRef,
- Arc::new(StringArray::from(vec![format!("Name{idx}")])) as
ArrayRef,
- ])
+ RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![idx])) as ArrayRef,
+ Arc::new(StringArray::from(vec![format!("Name{idx}")]))
+ as ArrayRef,
+ ],
+ )
})
.collect::<Result<_, _>>()?;
@@ -670,7 +704,8 @@ mod tests {
// Create multiple partitions - each batch becomes a separate partition
let partitions: Vec<Vec<RecordBatch>> =
batches.into_iter().map(|batch| vec![batch]).collect();
- let source_table =
Arc::new(MemTable::try_new(Arc::clone(&arrow_schema), partitions)?);
+ let source_table =
+ Arc::new(MemTable::try_new(Arc::clone(&arrow_schema),
partitions)?);
ctx.register_table("source_table", source_table)?;
let iceberg_table_provider = IcebergTableProvider::try_new(
diff --git a/crates/datafusion/src/physical_plan/expr_to_predicate.rs
b/crates/datafusion/src/physical_plan/expr_to_predicate.rs
index fb5440a..58b3ee2 100644
--- a/crates/datafusion/src/physical_plan/expr_to_predicate.rs
+++ b/crates/datafusion/src/physical_plan/expr_to_predicate.rs
@@ -21,7 +21,9 @@ use datafusion::arrow::datatypes::DataType;
use datafusion::logical_expr::expr::ScalarFunction;
use datafusion::logical_expr::{BinaryExpr, Expr, Like, Operator};
use datafusion::scalar::ScalarValue;
-use iceberg::expr::{BinaryExpression, Predicate, PredicateOperator, Reference,
UnaryExpression};
+use iceberg::expr::{
+ BinaryExpression, Predicate, PredicateOperator, Reference, UnaryExpression,
+};
use iceberg::spec::{Datum, PrimitiveLiteral};
// A datafusion expression could be an Iceberg predicate, column, or literal.
@@ -76,7 +78,9 @@ fn to_iceberg_predicate(expr: &Expr) -> TransformedResult {
let right = to_iceberg_predicate(&binary.right);
let op = to_iceberg_operation(binary.op);
match op {
- OpTransformedResult::Operator(op) =>
to_iceberg_binary_predicate(left, right, op),
+ OpTransformedResult::Operator(op) => {
+ to_iceberg_binary_predicate(left, right, op)
+ }
OpTransformedResult::And => to_iceberg_and_predicate(left,
right),
OpTransformedResult::Or => to_iceberg_or_predicate(left,
right),
OpTransformedResult::NotTransformed =>
TransformedResult::NotTransformed,
@@ -88,11 +92,13 @@ fn to_iceberg_predicate(expr: &Expr) -> TransformedResult {
TransformedResult::Predicate(p) =>
TransformedResult::Predicate(!p),
TransformedResult::Column(column) => {
// NOT of a bare boolean column: NOT col => col = false
-
TransformedResult::Predicate(Predicate::Binary(BinaryExpression::new(
- PredicateOperator::Eq,
- column,
- Datum::bool(false),
- )))
+ TransformedResult::Predicate(Predicate::Binary(
+ BinaryExpression::new(
+ PredicateOperator::Eq,
+ column,
+ Datum::bool(false),
+ ),
+ ))
}
_ => TransformedResult::NotTransformed,
}
@@ -124,23 +130,24 @@ fn to_iceberg_predicate(expr: &Expr) -> TransformedResult
{
Expr::IsNull(expr) => {
let p = to_iceberg_predicate(expr);
match p {
- TransformedResult::Column(r) =>
TransformedResult::Predicate(Predicate::Unary(
- UnaryExpression::new(PredicateOperator::IsNull, r),
- )),
+ TransformedResult::Column(r) => TransformedResult::Predicate(
+
Predicate::Unary(UnaryExpression::new(PredicateOperator::IsNull, r)),
+ ),
_ => TransformedResult::NotTransformed,
}
}
Expr::IsNotNull(expr) => {
let p = to_iceberg_predicate(expr);
match p {
- TransformedResult::Column(r) =>
TransformedResult::Predicate(Predicate::Unary(
- UnaryExpression::new(PredicateOperator::NotNull, r),
- )),
+ TransformedResult::Column(r) => TransformedResult::Predicate(
+
Predicate::Unary(UnaryExpression::new(PredicateOperator::NotNull, r)),
+ ),
_ => TransformedResult::NotTransformed,
}
}
Expr::Cast(c) => {
- if *c.field.data_type() == DataType::Date32 ||
*c.field.data_type() == DataType::Date64
+ if *c.field.data_type() == DataType::Date32
+ || *c.field.data_type() == DataType::Date64
{
// Casts to date truncate the expression, we cannot simply
extract it as it
// can create erroneous predicates.
@@ -212,7 +219,9 @@ fn to_iceberg_operation(op: Operator) ->
OpTransformedResult {
Operator::Lt =>
OpTransformedResult::Operator(PredicateOperator::LessThan),
Operator::LtEq =>
OpTransformedResult::Operator(PredicateOperator::LessThanOrEq),
Operator::Gt =>
OpTransformedResult::Operator(PredicateOperator::GreaterThan),
- Operator::GtEq =>
OpTransformedResult::Operator(PredicateOperator::GreaterThanOrEq),
+ Operator::GtEq => {
+ OpTransformedResult::Operator(PredicateOperator::GreaterThanOrEq)
+ }
// AND OR
Operator::And => OpTransformedResult::And,
Operator::Or => OpTransformedResult::Or,
@@ -224,7 +233,10 @@ fn to_iceberg_operation(op: Operator) ->
OpTransformedResult {
/// Translates a DataFusion scalar function into an Iceberg predicate.
/// Unlike dedicated Expr variants (e.g. `Expr::IsNull`), scalar functions are
/// identified by name at runtime, so we need to handle them here.
-fn scalar_function_to_iceberg_predicate(func_name: &str, args: &[Expr]) ->
TransformedResult {
+fn scalar_function_to_iceberg_predicate(
+ func_name: &str,
+ args: &[Expr],
+) -> TransformedResult {
match func_name {
"isnan" if args.len() == 1 => match
resolve_nan_preserving_reference(&args[0]) {
Some(r) => TransformedResult::Predicate(r.is_nan()),
@@ -369,7 +381,10 @@ fn to_iceberg_and_predicate(
}
}
-fn to_iceberg_or_predicate(left: TransformedResult, right: TransformedResult)
-> TransformedResult {
+fn to_iceberg_or_predicate(
+ left: TransformedResult,
+ right: TransformedResult,
+) -> TransformedResult {
match (left, right) {
(TransformedResult::Predicate(left),
TransformedResult::Predicate(right)) => {
TransformedResult::Predicate(left.or(right))
@@ -384,8 +399,12 @@ fn to_iceberg_binary_predicate(
op: PredicateOperator,
) -> TransformedResult {
let (r, d, op) = match (left, right) {
- (TransformedResult::NotTransformed, _) => return
TransformedResult::NotTransformed,
- (_, TransformedResult::NotTransformed) => return
TransformedResult::NotTransformed,
+ (TransformedResult::NotTransformed, _) => {
+ return TransformedResult::NotTransformed;
+ }
+ (_, TransformedResult::NotTransformed) => {
+ return TransformedResult::NotTransformed;
+ }
(TransformedResult::Column(r), TransformedResult::Literal(d)) => (r,
d, op),
(TransformedResult::Literal(d), TransformedResult::Column(r)) => {
(r, d, reverse_predicate_operator(op))
@@ -430,7 +449,9 @@ fn scalar_value_to_datum(value: &ScalarValue) ->
Option<Datum> {
// DataFusion's type coercion always converts them to match the column
type
// (either TimestampMicrosecond or TimestampNanosecond) before
predicate pushdown.
// See unit tests for how those conversions would work if needed.
- ScalarValue::TimestampMicrosecond(Some(v), _) =>
Some(Datum::timestamp_micros(*v)),
+ ScalarValue::TimestampMicrosecond(Some(v), _) => {
+ Some(Datum::timestamp_micros(*v))
+ }
ScalarValue::TimestampNanosecond(Some(v), _) =>
Some(Datum::timestamp_nanos(*v)),
_ => None,
}
@@ -451,23 +472,24 @@ mod tests {
use super::convert_filters_to_predicate;
fn create_test_schema() -> DFSchema {
- let arrow_schema = Schema::new(vec![
- Field::new("foo", DataType::Int32,
true).with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "1".to_string(),
- )])),
- Field::new("bar", DataType::Utf8,
true).with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "2".to_string(),
- )])),
- Field::new("ts", DataType::Timestamp(TimeUnit::Second, None),
true).with_metadata(
- HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(),
"3".to_string())]),
- ),
- Field::new("qux", DataType::Float64,
true).with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "4".to_string(),
- )])),
- ]);
+ let arrow_schema =
+ Schema::new(vec![
+ Field::new("foo", DataType::Int32,
true).with_metadata(HashMap::from([
+ (PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string()),
+ ])),
+ Field::new("bar", DataType::Utf8,
true).with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "2".to_string(),
+ )])),
+ Field::new("ts", DataType::Timestamp(TimeUnit::Second, None),
true)
+ .with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "3".to_string(),
+ )])),
+ Field::new("qux", DataType::Float64,
true).with_metadata(HashMap::from(
+ [(PARQUET_FIELD_ID_META_KEY.to_string(), "4".to_string())],
+ )),
+ ]);
DFSchema::try_from_qualified_schema("my_table", &arrow_schema).unwrap()
}
@@ -640,8 +662,8 @@ mod tests {
fn test_predicate_conversion_with_cast() {
let sql = "ts >= timestamp '2023-01-05T00:00:00'";
let predicate = convert_to_iceberg_predicate(sql).unwrap();
- let expected_predicate =
-
Reference::new("ts").greater_than_or_equal_to(Datum::string("2023-01-05T00:00:00"));
+ let expected_predicate = Reference::new("ts")
+ .greater_than_or_equal_to(Datum::string("2023-01-05T00:00:00"));
assert_eq!(predicate, expected_predicate);
}
@@ -658,18 +680,23 @@ mod tests {
// Test TimestampMicrosecond - maps directly to Datum::timestamp_micros
let ts_micros = 1672876800000000i64; // 2023-01-05 00:00:00 UTC in
microseconds
- let datum =
-
super::scalar_value_to_datum(&ScalarValue::TimestampMicrosecond(Some(ts_micros),
None));
+ let datum =
super::scalar_value_to_datum(&ScalarValue::TimestampMicrosecond(
+ Some(ts_micros),
+ None,
+ ));
assert_eq!(datum, Some(Datum::timestamp_micros(ts_micros)));
// Test TimestampNanosecond - maps to Datum::timestamp_nanos to
preserve precision
let ts_nanos = 1672876800000000500i64; // 2023-01-05
00:00:00.000000500 UTC in nanoseconds
- let datum =
-
super::scalar_value_to_datum(&ScalarValue::TimestampNanosecond(Some(ts_nanos),
None));
+ let datum =
super::scalar_value_to_datum(&ScalarValue::TimestampNanosecond(
+ Some(ts_nanos),
+ None,
+ ));
assert_eq!(datum, Some(Datum::timestamp_nanos(ts_nanos)));
// Test None timestamp
- let datum =
super::scalar_value_to_datum(&ScalarValue::TimestampMicrosecond(None, None));
+ let datum =
+
super::scalar_value_to_datum(&ScalarValue::TimestampMicrosecond(None, None));
assert_eq!(datum, None);
// Note: TimestampSecond and TimestampMillisecond are not supported
because
@@ -678,13 +705,17 @@ mod tests {
//
// These return None (not pushed down):
let ts_seconds = 1672876800i64; // 2023-01-05 00:00:00 UTC in seconds
- let datum =
-
super::scalar_value_to_datum(&ScalarValue::TimestampSecond(Some(ts_seconds),
None));
+ let datum = super::scalar_value_to_datum(&ScalarValue::TimestampSecond(
+ Some(ts_seconds),
+ None,
+ ));
assert_eq!(datum, None);
let ts_millis = 1672876800000i64; // 2023-01-05 00:00:00 UTC in
milliseconds
- let datum =
-
super::scalar_value_to_datum(&ScalarValue::TimestampMillisecond(Some(ts_millis),
None));
+ let datum =
super::scalar_value_to_datum(&ScalarValue::TimestampMillisecond(
+ Some(ts_millis),
+ None,
+ ));
assert_eq!(datum, None);
}
@@ -693,10 +724,12 @@ mod tests {
use datafusion::common::ScalarValue;
let bytes = vec![1u8, 2u8, 3u8];
- let datum =
super::scalar_value_to_datum(&ScalarValue::Binary(Some(bytes.clone())));
+ let datum =
+
super::scalar_value_to_datum(&ScalarValue::Binary(Some(bytes.clone())));
assert_eq!(datum, Some(Datum::binary(bytes.clone())));
- let datum =
super::scalar_value_to_datum(&ScalarValue::LargeBinary(Some(bytes.clone())));
+ let datum =
+
super::scalar_value_to_datum(&ScalarValue::LargeBinary(Some(bytes.clone())));
assert_eq!(datum, Some(Datum::binary(bytes)));
let datum = super::scalar_value_to_datum(&ScalarValue::Binary(None));
@@ -892,7 +925,8 @@ mod tests {
#[test]
fn test_predicate_conversion_with_isnan_nested_expr() {
// Nested NaN-preserving transformations resolve to the inner column
- let predicate = convert_to_iceberg_predicate("isnan(-(abs(qux) + 1) *
3)").unwrap();
+ let predicate =
+ convert_to_iceberg_predicate("isnan(-(abs(qux) + 1) *
3)").unwrap();
assert_eq!(predicate, Reference::new("qux").is_nan());
}
diff --git a/crates/datafusion/src/physical_plan/metadata_scan.rs
b/crates/datafusion/src/physical_plan/metadata_scan.rs
index 33c1143..92e0671 100644
--- a/crates/datafusion/src/physical_plan/metadata_scan.rs
+++ b/crates/datafusion/src/physical_plan/metadata_scan.rs
@@ -73,7 +73,9 @@ impl ExecutionPlan for IcebergMetadataScan {
fn apply_expressions(
&self,
- _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) ->
datafusion::error::Result<TreeNodeRecursion>,
+ _f: &mut dyn FnMut(
+ &Arc<dyn PhysicalExpr>,
+ ) -> datafusion::error::Result<TreeNodeRecursion>,
) -> datafusion::error::Result<TreeNodeRecursion> {
Ok(TreeNodeRecursion::Continue)
}
diff --git a/crates/datafusion/src/physical_plan/project.rs
b/crates/datafusion/src/physical_plan/project.rs
index a77abd0..d8ab9b3 100644
--- a/crates/datafusion/src/physical_plan/project.rs
+++ b/crates/datafusion/src/physical_plan/project.rs
@@ -68,8 +68,8 @@ pub fn project_with_partition(
schema_to_arrow_schema(table_schema.as_ref()).map_err(to_datafusion_error)?;
let input_schema_cleaned =
strip_metadata_from_schema(&input_schema).map_err(to_datafusion_error)?;
- let expected_schema_cleaned =
-
strip_metadata_from_schema(&expected_arrow_schema).map_err(to_datafusion_error)?;
+ let expected_schema_cleaned =
strip_metadata_from_schema(&expected_arrow_schema)
+ .map_err(to_datafusion_error)?;
if input_schema_cleaned != expected_schema_cleaned {
return Err(DataFusionError::Plan(format!(
@@ -106,7 +106,10 @@ struct PartitionExpr {
}
impl PartitionExpr {
- fn new(calculator: PartitionValueCalculator, partition_spec:
Arc<PartitionSpec>) -> Self {
+ fn new(
+ calculator: PartitionValueCalculator,
+ partition_spec: Arc<PartitionSpec>,
+ ) -> Self {
Self {
calculator: Arc::new(calculator),
partition_spec,
@@ -189,7 +192,9 @@ mod tests {
use datafusion::arrow::array::{ArrayRef, Int32Array, StructArray};
use datafusion::arrow::datatypes::{DataType, Field, Fields};
use datafusion::physical_plan::empty::EmptyExec;
- use iceberg::spec::{NestedField, PrimitiveType, Schema, StructType,
Transform, Type};
+ use iceberg::spec::{
+ NestedField, PrimitiveType, Schema, StructType, Transform, Type,
+ };
use iceberg::test_utils::test_runtime;
use super::*;
@@ -199,8 +204,10 @@ mod tests {
let table_schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()
.unwrap();
@@ -211,7 +218,8 @@ mod tests {
.build()
.unwrap();
- let calculator = PartitionValueCalculator::try_new(&partition_spec,
&table_schema).unwrap();
+ let calculator =
+ PartitionValueCalculator::try_new(&partition_spec,
&table_schema).unwrap();
// Verify partition type
assert_eq!(calculator.partition_type().fields().len(), 1);
@@ -223,8 +231,10 @@ mod tests {
let table_schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()
.unwrap();
@@ -244,7 +254,8 @@ mod tests {
let input = Arc::new(EmptyExec::new(arrow_schema.clone()));
- let calculator = PartitionValueCalculator::try_new(&partition_spec,
&table_schema).unwrap();
+ let calculator =
+ PartitionValueCalculator::try_new(&partition_spec,
&table_schema).unwrap();
let mut projection_exprs: Vec<(Arc<dyn PhysicalExpr>, String)> =
Vec::with_capacity(arrow_schema.fields().len() + 1);
@@ -254,7 +265,8 @@ mod tests {
}
let partition_expr = Arc::new(PartitionExpr::new(calculator,
partition_spec));
- projection_exprs.push((partition_expr,
PROJECTED_PARTITION_VALUE_COLUMN.to_string()));
+ projection_exprs
+ .push((partition_expr,
PROJECTED_PARTITION_VALUE_COLUMN.to_string()));
let projection = ProjectionExec::try_new(projection_exprs,
input).unwrap();
let result = Arc::new(projection);
@@ -270,8 +282,10 @@ mod tests {
let table_schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "data",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(2, "data",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()
.unwrap();
@@ -287,16 +301,20 @@ mod tests {
Field::new("data", DataType::Utf8, false),
]));
- let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
- Arc::new(Int32Array::from(vec![10, 20, 30])),
- Arc::new(datafusion::arrow::array::StringArray::from(vec![
- "a", "b", "c",
- ])),
- ])
+ let batch = RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![10, 20, 30])),
+ Arc::new(datafusion::arrow::array::StringArray::from(vec![
+ "a", "b", "c",
+ ])),
+ ],
+ )
.unwrap();
let partition_spec = Arc::new(partition_spec);
- let calculator = PartitionValueCalculator::try_new(&partition_spec,
&table_schema).unwrap();
+ let calculator =
+ PartitionValueCalculator::try_new(&partition_spec,
&table_schema).unwrap();
let partition_type = calculator.partition_arrow_type().clone();
let expr = PartitionExpr::new(calculator, partition_spec);
@@ -324,14 +342,17 @@ mod tests {
#[test]
fn test_nested_partition() {
let address_struct = StructType::new(vec![
- NestedField::required(3, "street",
Type::Primitive(PrimitiveType::String)).into(),
- NestedField::required(4, "city",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(3, "street",
Type::Primitive(PrimitiveType::String))
+ .into(),
+ NestedField::required(4, "city",
Type::Primitive(PrimitiveType::String))
+ .into(),
]);
let table_schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
NestedField::required(2, "address",
Type::Struct(address_struct)).into(),
])
.build()
@@ -373,13 +394,17 @@ mod tests {
),
]);
- let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
- Arc::new(Int32Array::from(vec![1, 2])),
- Arc::new(struct_array),
- ])
+ let batch = RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(struct_array),
+ ],
+ )
.unwrap();
- let calculator = PartitionValueCalculator::try_new(&partition_spec,
&table_schema).unwrap();
+ let calculator =
+ PartitionValueCalculator::try_new(&partition_spec,
&table_schema).unwrap();
let array = calculator.calculate(&batch).unwrap();
let struct_array =
array.as_any().downcast_ref::<StructArray>().unwrap();
@@ -403,8 +428,14 @@ mod tests {
let table_schema = Arc::new(
Schema::builder()
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(
+ 2,
+ "name",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
])
.build()
.unwrap(),
@@ -462,8 +493,14 @@ mod tests {
let table_schema = Arc::new(
Schema::builder()
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(
+ 2,
+ "name",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
])
.build()
.unwrap(),
@@ -532,8 +569,14 @@ mod tests {
let table_schema = Arc::new(
Schema::builder()
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(
+ 2,
+ "name",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
])
.build()
.unwrap(),
diff --git a/crates/datafusion/src/physical_plan/repartition.rs
b/crates/datafusion/src/physical_plan/repartition.rs
index efa0f3c..e6cbc13 100644
--- a/crates/datafusion/src/physical_plan/repartition.rs
+++ b/crates/datafusion/src/physical_plan/repartition.rs
@@ -171,15 +171,16 @@ fn determine_partitioning_strategy(
#[cfg(test)]
mod tests {
use datafusion::arrow::datatypes::{
- DataType as ArrowDataType, Field as ArrowField, Fields, Schema as
ArrowSchema, TimeUnit,
+ DataType as ArrowDataType, Field as ArrowField, Fields, Schema as
ArrowSchema,
+ TimeUnit,
};
use datafusion::execution::TaskContext;
use datafusion::physical_plan::empty::EmptyExec;
use iceberg::TableIdent;
use iceberg::io::FileIO;
use iceberg::spec::{
- NestedField, NullOrder, PrimitiveType, Schema, SortDirection,
SortField, SortOrder,
- Transform, Type,
+ NestedField, NullOrder, PrimitiveType, Schema, SortDirection,
SortField,
+ SortOrder, Transform, Type,
};
use iceberg::table::Table;
use iceberg::test_utils::test_runtime;
@@ -258,7 +259,8 @@ mod tests {
let input = Arc::new(EmptyExec::new(create_test_arrow_schema()));
let repartitioned_plan =
- repartition(input, table.metadata_ref(),
NonZeroUsize::new(8).unwrap()).unwrap();
+ repartition(input, table.metadata_ref(),
NonZeroUsize::new(8).unwrap())
+ .unwrap();
let partitioning =
repartitioned_plan.properties().output_partitioning();
match partitioning {
@@ -310,7 +312,8 @@ mod tests {
let input = Arc::new(EmptyExec::new(create_test_arrow_schema()));
let repartitioned_plan =
- repartition(input, table.metadata_ref(),
NonZeroUsize::new(3).unwrap()).unwrap();
+ repartition(input, table.metadata_ref(),
NonZeroUsize::new(3).unwrap())
+ .unwrap();
let partitioning =
repartitioned_plan.properties().output_partitioning();
match partitioning {
@@ -384,7 +387,8 @@ mod tests {
]));
let input = Arc::new(EmptyExec::new(arrow_schema));
let repartitioned_plan =
- repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap()).unwrap();
+ repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap())
+ .unwrap();
let partitioning =
repartitioned_plan.properties().output_partitioning();
// For bucketed tables without _partition column, should use
round-robin
@@ -453,7 +457,9 @@ mod tests {
let table_metadata = table_metadata_builder.build().unwrap();
let table = Table::builder()
.metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test",
"partitioned_bucketed_table"]).unwrap())
+ .identifier(
+ TableIdent::from_strs(["test",
"partitioned_bucketed_table"]).unwrap(),
+ )
.file_io(FileIO::new_with_fs())
.metadata_location("/test/partitioned_bucketed_metadata.json")
.runtime(test_runtime())
@@ -472,7 +478,8 @@ mod tests {
]));
let input = Arc::new(EmptyExec::new(arrow_schema));
let repartitioned_plan =
- repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap()).unwrap();
+ repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap())
+ .unwrap();
let partitioning =
repartitioned_plan.properties().output_partitioning();
match partitioning {
@@ -542,7 +549,8 @@ mod tests {
let input = Arc::new(EmptyExec::new(create_test_arrow_schema()));
let repartitioned_plan =
- repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap()).unwrap();
+ repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap())
+ .unwrap();
let partitioning =
repartitioned_plan.properties().output_partitioning();
assert!(
@@ -620,7 +628,8 @@ mod tests {
]));
let input = Arc::new(EmptyExec::new(arrow_schema));
let repartitioned_plan =
- repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap()).unwrap();
+ repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap())
+ .unwrap();
let partitioning =
repartitioned_plan.properties().output_partitioning();
assert!(
@@ -674,7 +683,9 @@ mod tests {
let table_metadata = table_metadata_builder.build().unwrap();
let table = Table::builder()
.metadata(table_metadata.metadata)
- .identifier(TableIdent::from_strs(["test",
"mixed_transforms_table"]).unwrap())
+ .identifier(
+ TableIdent::from_strs(["test",
"mixed_transforms_table"]).unwrap(),
+ )
.file_io(FileIO::new_with_fs())
.metadata_location("/test/mixed_transforms_metadata.json")
.runtime(test_runtime())
@@ -693,7 +704,8 @@ mod tests {
]));
let input = Arc::new(EmptyExec::new(arrow_schema));
let repartitioned_plan =
- repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap()).unwrap();
+ repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap())
+ .unwrap();
let partitioning =
repartitioned_plan.properties().output_partitioning();
match partitioning {
@@ -776,7 +788,8 @@ mod tests {
let input = Arc::new(EmptyExec::new(arrow_schema));
let repartitioned_plan =
- repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap()).unwrap();
+ repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap())
+ .unwrap();
let partitioning =
repartitioned_plan.properties().output_partitioning();
assert!(
@@ -842,7 +855,8 @@ mod tests {
let input = Arc::new(EmptyExec::new(arrow_schema));
let repartitioned_plan =
- repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap()).unwrap();
+ repartition(input, table.metadata_ref(),
NonZeroUsize::new(4).unwrap())
+ .unwrap();
let partitioning =
repartitioned_plan.properties().output_partitioning();
assert!(
diff --git a/crates/datafusion/src/physical_plan/sort.rs
b/crates/datafusion/src/physical_plan/sort.rs
index 587ab12..fce9210 100644
--- a/crates/datafusion/src/physical_plan/sort.rs
+++ b/crates/datafusion/src/physical_plan/sort.rs
@@ -42,7 +42,9 @@ use iceberg::arrow::PROJECTED_PARTITION_VALUE_COLUMN;
/// # Returns
/// * `Ok(Arc<dyn ExecutionPlan>)` - A SortExec that sorts by partition values
/// * `Err` - If the partition column is not found
-pub(crate) fn sort_by_partition(input: Arc<dyn ExecutionPlan>) ->
DFResult<Arc<dyn ExecutionPlan>> {
+pub(crate) fn sort_by_partition(
+ input: Arc<dyn ExecutionPlan>,
+) -> DFResult<Arc<dyn ExecutionPlan>> {
let schema = input.schema();
// Find the partition column in the schema
@@ -68,7 +70,9 @@ pub(crate) fn sort_by_partition(input: Arc<dyn
ExecutionPlan>) -> DFResult<Arc<d
// Create a SortExec with preserve_partitioning=true to ensure the output
partitioning
// is the same as the input partitioning, and the data is sorted within
each partition
let lex_ordering = LexOrdering::new(vec![sort_expr]).ok_or_else(|| {
- DataFusionError::Plan("Failed to create LexOrdering from sort
expression".to_string())
+ DataFusionError::Plan(
+ "Failed to create LexOrdering from sort expression".to_string(),
+ )
})?;
let sort_exec = SortExec::new(lex_ordering,
input).with_preserve_partitioning(true);
@@ -109,9 +113,11 @@ mod tests {
Arc::new(Int32Array::from(vec![3, 1, 2])) as _,
)]));
- let batch =
- RecordBatch::try_new(schema.clone(), vec![id_array, name_array,
partition_array])
- .unwrap();
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![id_array, name_array, partition_array],
+ )
+ .unwrap();
let ctx = SessionContext::new();
let mem_table = MemTable::try_new(schema.clone(),
vec![vec![batch]]).unwrap();
@@ -147,10 +153,13 @@ mod tests {
Field::new("name", DataType::Utf8, false),
]));
- let batch = RecordBatch::try_new(schema.clone(), vec![
- Arc::new(Int32Array::from(vec![1, 2, 3])),
- Arc::new(StringArray::from(vec!["a", "b", "c"])),
- ])
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(StringArray::from(vec!["a", "b", "c"])),
+ ],
+ )
.unwrap();
let ctx = SessionContext::new();
@@ -204,9 +213,11 @@ mod tests {
),
]));
- let batch =
- RecordBatch::try_new(schema.clone(), vec![id_array, data_array,
partition_array])
- .unwrap();
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![id_array, data_array, partition_array],
+ )
+ .unwrap();
let ctx = SessionContext::new();
let mem_table = MemTable::try_new(schema.clone(),
vec![vec![batch]]).unwrap();
diff --git a/crates/datafusion/src/physical_plan/write.rs
b/crates/datafusion/src/physical_plan/write.rs
index 40ad9fa..c5a4206 100644
--- a/crates/datafusion/src/physical_plan/write.rs
+++ b/crates/datafusion/src/physical_plan/write.rs
@@ -68,7 +68,8 @@ pub(crate) struct IcebergWriteExec {
impl IcebergWriteExec {
pub fn new(table: Table, input: Arc<dyn ExecutionPlan>) -> Self {
- let plan_properties = Self::compute_properties(&input,
Self::make_result_schema());
+ let plan_properties =
+ Self::compute_properties(&input, Self::make_result_schema());
Self {
table,
@@ -84,7 +85,9 @@ impl IcebergWriteExec {
) -> Arc<PlanProperties> {
Arc::new(PlanProperties::new(
EquivalenceProperties::new(schema),
-
Partitioning::UnknownPartitioning(input.output_partitioning().partition_count()),
+ Partitioning::UnknownPartitioning(
+ input.output_partitioning().partition_count(),
+ ),
EmissionType::Final,
Boundedness::Bounded,
))
@@ -216,27 +219,30 @@ impl ExecutionPlan for IcebergWriteExec {
let write_format_default = table_props
.write_format_default()
.map_err(to_datafusion_error)?;
- let file_format =
-
DataFileFormat::from_str(&write_format_default).map_err(to_datafusion_error)?;
+ let file_format = DataFileFormat::from_str(&write_format_default)
+ .map_err(to_datafusion_error)?;
if file_format != DataFileFormat::Parquet {
return Err(to_datafusion_error(Error::new(
ErrorKind::FeatureUnsupported,
- format!("File format {file_format} is not supported for
insert_into yet!"),
+ format!(
+ "File format {file_format} is not supported for
insert_into yet!"
+ ),
)));
}
// Build the writer from the already-parsed table properties so it
honors
// `write.parquet.*` settings (e.g. CDC). Arrow batches flowing through
// DataFusion carry no field-id metadata, so match fields by name.
- let mut parquet_file_writer_builder =
ParquetWriterBuilder::from_table_properties(
- &table_props,
- self.table.metadata().current_schema().clone(),
- )
- .map_err(to_datafusion_error)?
- .with_match_mode(FieldMatchMode::Name);
+ let mut parquet_file_writer_builder =
+ ParquetWriterBuilder::from_table_properties(
+ &table_props,
+ self.table.metadata().current_schema().clone(),
+ )
+ .map_err(to_datafusion_error)?
+ .with_match_mode(FieldMatchMode::Name);
if let Some(encryption_manager) = self.table.encryption_manager() {
- parquet_file_writer_builder =
-
parquet_file_writer_builder.with_encryption_manager(encryption_manager.clone());
+ parquet_file_writer_builder = parquet_file_writer_builder
+ .with_encryption_manager(encryption_manager.clone());
}
let target_file_size = table_props
.write_target_file_size_bytes()
@@ -244,8 +250,8 @@ impl ExecutionPlan for IcebergWriteExec {
let file_io = self.table.file_io().clone();
// todo location_gen and file_name_gen should be configurable
- let location_generator =
-
DefaultLocationGenerator::new(self.table.metadata()).map_err(to_datafusion_error)?;
+ let location_generator =
DefaultLocationGenerator::new(self.table.metadata())
+ .map_err(to_datafusion_error)?;
// todo filename prefix/suffix should be configurable
let file_name_generator =
DefaultFileNameGenerator::new(Uuid::now_v7().to_string(), None,
file_format);
@@ -299,8 +305,12 @@ impl ExecutionPlan for IcebergWriteExec {
let data_files_strs: Vec<String> = data_files
.into_iter()
.map(|data_file| {
- serialize_data_file_to_json(data_file, &partition_type,
format_version)
- .map_err(to_datafusion_error)
+ serialize_data_file_to_json(
+ data_file,
+ &partition_type,
+ format_version,
+ )
+ .map_err(to_datafusion_error)
})
.collect::<DFResult<Vec<String>>>()?;
@@ -330,13 +340,18 @@ mod tests {
use datafusion::physical_expr::{EquivalenceProperties, Partitioning};
use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
- use datafusion::physical_plan::{DisplayAs, DisplayFormatType,
ExecutionPlan, PlanProperties};
+ use datafusion::physical_plan::{
+ DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+ };
use futures::{StreamExt, stream};
use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
use iceberg::spec::{
- DataFileFormat, NestedField, PrimitiveType, Schema, Type,
deserialize_data_file_from_json,
+ DataFileFormat, NestedField, PrimitiveType, Schema, Type,
+ deserialize_data_file_from_json,
+ };
+ use iceberg::{
+ Catalog, CatalogBuilder, MemoryCatalog, NamespaceIdent, Result,
TableCreation,
};
- use iceberg::{Catalog, CatalogBuilder, MemoryCatalog, NamespaceIdent,
Result, TableCreation};
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use tempfile::TempDir;
@@ -447,8 +462,10 @@ mod tests {
Schema::builder()
.with_schema_id(0)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()
}
@@ -488,32 +505,34 @@ mod tests {
let table = iceberg_catalog.create_table(&namespace, creation).await?;
// 2. Create test data
- let arrow_schema = Arc::new(ArrowSchema::new(vec![
- Field::new("id", DataType::Int32,
false).with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "1".to_string(),
- )])),
- Field::new("name", DataType::Utf8,
false).with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- "2".to_string(),
- )])),
- ]));
+ let arrow_schema =
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("id", DataType::Int32,
false).with_metadata(HashMap::from([
+ (PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string()),
+ ])),
+ Field::new("name", DataType::Utf8,
false).with_metadata(HashMap::from([
+ (PARQUET_FIELD_ID_META_KEY.to_string(), "2".to_string()),
+ ])),
+ ]));
let id_array = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef;
- let name_array = Arc::new(StringArray::from(vec!["Alice", "Bob",
"Charlie"])) as ArrayRef;
-
- let batch = RecordBatch::try_new(arrow_schema.clone(), vec![id_array,
name_array])
- .map_err(|e| {
- Error::new(
- ErrorKind::Unexpected,
- format!("Failed to create record batch: {e}"),
- )
- })?;
+ let name_array =
+ Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie"])) as
ArrayRef;
+
+ let batch =
+ RecordBatch::try_new(arrow_schema.clone(), vec![id_array,
name_array])
+ .map_err(|e| {
+ Error::new(
+ ErrorKind::Unexpected,
+ format!("Failed to create record batch: {e}"),
+ )
+ })?;
// 3. Create mock input execution plan
- let input_plan = Arc::new(MockExecutionPlan::new(arrow_schema.clone(),
vec![
- batch.clone(),
- ]));
+ let input_plan = Arc::new(MockExecutionPlan::new(
+ arrow_schema.clone(),
+ vec![batch.clone()],
+ ));
// 4. Create IcebergWriteExec
let write_exec = IcebergWriteExec::new(table.clone(), input_plan);
@@ -543,7 +562,11 @@ mod tests {
// Check schema
assert_eq!(
result_batch.schema().as_ref(),
- &ArrowSchema::new(vec![Field::new(DATA_FILES_COL_NAME,
DataType::Utf8, false)])
+ &ArrowSchema::new(vec![Field::new(
+ DATA_FILES_COL_NAME,
+ DataType::Utf8,
+ false
+ )])
);
// Check data
@@ -562,9 +585,13 @@ mod tests {
let spec_id = table.metadata().default_partition_spec_id();
let schema = table.metadata().current_schema();
- let data_file =
- deserialize_data_file_from_json(data_file_json, spec_id,
partition_type, schema)
- .expect("Failed to deserialize data file JSON");
+ let data_file = deserialize_data_file_from_json(
+ data_file_json,
+ spec_id,
+ partition_type,
+ schema,
+ )
+ .expect("Failed to deserialize data file JSON");
// Verify data file properties
assert_eq!(
@@ -605,11 +632,13 @@ mod tests {
// Verify lower and upper bounds
assert!(
- data_file.lower_bounds().contains_key(&1) ||
data_file.lower_bounds().contains_key(&2),
+ data_file.lower_bounds().contains_key(&1)
+ || data_file.lower_bounds().contains_key(&2),
"Expected lower bounds to contain at least one column"
);
assert!(
- data_file.upper_bounds().contains_key(&1) ||
data_file.upper_bounds().contains_key(&2),
+ data_file.upper_bounds().contains_key(&1)
+ || data_file.upper_bounds().contains_key(&2),
"Expected upper bounds to contain at least one column"
);
@@ -644,7 +673,11 @@ mod tests {
assert_eq!(
write_exec.schema().as_ref(),
- &ArrowSchema::new(vec![Field::new(DATA_FILES_COL_NAME,
DataType::Utf8, false)]),
+ &ArrowSchema::new(vec![Field::new(
+ DATA_FILES_COL_NAME,
+ DataType::Utf8,
+ false
+ )]),
"IcebergWriteExec should advertise the data_files schema, not the
table schema"
);
diff --git a/crates/datafusion/src/schema.rs b/crates/datafusion/src/schema.rs
index 545863f..04bea0b 100644
--- a/crates/datafusion/src/schema.rs
+++ b/crates/datafusion/src/schema.rs
@@ -29,7 +29,9 @@ use futures::future::try_join_all;
use iceberg::arrow::arrow_schema_to_schema_auto_assign_ids;
use iceberg::inspect::MetadataTableType;
use iceberg::spec::FormatVersion;
-use iceberg::{Catalog, Error, ErrorKind, NamespaceIdent, Result,
TableCreation, TableIdent};
+use iceberg::{
+ Catalog, Error, ErrorKind, NamespaceIdent, Result, TableCreation,
TableIdent,
+};
use crate::table::IcebergTableProvider;
use crate::to_datafusion_error;
@@ -75,7 +77,9 @@ impl IcebergSchemaProvider {
let providers = try_join_all(
table_names
.iter()
- .map(|name| IcebergTableProvider::try_new(client.clone(),
namespace.clone(), name))
+ .map(|name| {
+ IcebergTableProvider::try_new(client.clone(),
namespace.clone(), name)
+ })
.collect::<Vec<_>>(),
)
.await?;
@@ -100,13 +104,11 @@ impl SchemaProvider for IcebergSchemaProvider {
.iter()
.flat_map(|entry| {
let table_name = entry.key().clone();
- [table_name.clone()]
- .into_iter()
- .chain(
- MetadataTableType::all_types().map(move
|metadata_table_name| {
- format!("{}${}", table_name,
metadata_table_name.as_str())
- }),
- )
+ [table_name.clone()].into_iter().chain(
+ MetadataTableType::all_types().map(move
|metadata_table_name| {
+ format!("{}${}", table_name,
metadata_table_name.as_str())
+ }),
+ )
})
.collect()
}
@@ -122,8 +124,8 @@ impl SchemaProvider for IcebergSchemaProvider {
async fn table(&self, name: &str) -> DFResult<Option<Arc<dyn
TableProvider>>> {
if let Some((table_name, metadata_table_name)) = name.split_once('$') {
- let metadata_table_type =
-
MetadataTableType::try_from(metadata_table_name).map_err(DataFusionError::Plan)?;
+ let metadata_table_type =
MetadataTableType::try_from(metadata_table_name)
+ .map_err(DataFusionError::Plan)?;
if let Some(table) = self.tables.get(table_name) {
let metadata_table = table
.metadata_table(metadata_table_type)
@@ -246,8 +248,9 @@ impl SchemaProvider for IcebergSchemaProvider {
})
});
- futures::executor::block_on(result)
- .map_err(|e| DataFusionError::Execution(format!("Failed to drop
Iceberg table: {e}")))?
+ futures::executor::block_on(result).map_err(|e| {
+ DataFusionError::Execution(format!("Failed to drop Iceberg table:
{e}"))
+ })?
}
}
@@ -258,7 +261,9 @@ async fn ensure_table_is_empty(table: &Arc<dyn
TableProvider>) -> Result<()> {
let exec_plan = table
.scan(&session_ctx.state(), None, &[], Some(1))
.await
- .map_err(|e| Error::new(ErrorKind::Unexpected, format!("Failed to scan
table: {e}")))?;
+ .map_err(|e| {
+ Error::new(ErrorKind::Unexpected, format!("Failed to scan table:
{e}"))
+ })?;
let task_ctx = Arc::new(TaskContext::default());
let stream = exec_plan.execute(0, task_ctx).map_err(|e| {
@@ -306,7 +311,10 @@ mod tests {
let catalog = MemoryCatalogBuilder::default()
.load(
"memory",
- HashMap::from([(MEMORY_CATALOG_WAREHOUSE.to_string(),
warehouse_path.clone())]),
+ HashMap::from([(
+ MEMORY_CATALOG_WAREHOUSE.to_string(),
+ warehouse_path.clone(),
+ )]),
)
.await
.unwrap();
@@ -334,16 +342,20 @@ mod tests {
Field::new("name", DataType::Utf8, true),
]));
- let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
- Arc::new(Int32Array::from(vec![1, 2, 3])),
- Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie"])),
- ])
+ let batch = RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie"])),
+ ],
+ )
.unwrap();
let mem_table = MemTable::try_new(arrow_schema,
vec![vec![batch]]).unwrap();
// Attempt to register the table with data - should fail
- let result = schema_provider.register_table("test_table".to_string(),
Arc::new(mem_table));
+ let result =
+ schema_provider.register_table("test_table".to_string(),
Arc::new(mem_table));
assert!(result.is_err());
let err = result.unwrap_err();
@@ -369,7 +381,8 @@ mod tests {
let mem_table = MemTable::try_new(arrow_schema,
vec![vec![empty_batch]]).unwrap();
// Attempt to register the empty table - should succeed
- let result = schema_provider.register_table("empty_table".to_string(),
Arc::new(mem_table));
+ let result = schema_provider
+ .register_table("empty_table".to_string(), Arc::new(mem_table));
assert!(result.is_ok(), "Expected success, got: {result:?}");
@@ -390,15 +403,19 @@ mod tests {
let empty_batch1 = RecordBatch::new_empty(arrow_schema.clone());
let empty_batch2 = RecordBatch::new_empty(arrow_schema.clone());
- let mem_table1 = MemTable::try_new(arrow_schema.clone(),
vec![vec![empty_batch1]]).unwrap();
- let mem_table2 = MemTable::try_new(arrow_schema,
vec![vec![empty_batch2]]).unwrap();
+ let mem_table1 =
+ MemTable::try_new(arrow_schema.clone(),
vec![vec![empty_batch1]]).unwrap();
+ let mem_table2 =
+ MemTable::try_new(arrow_schema, vec![vec![empty_batch2]]).unwrap();
// Register first table - should succeed
- let result1 = schema_provider.register_table("dup_table".to_string(),
Arc::new(mem_table1));
+ let result1 =
+ schema_provider.register_table("dup_table".to_string(),
Arc::new(mem_table1));
assert!(result1.is_ok());
// Register second table with same name - should fail
- let result2 = schema_provider.register_table("dup_table".to_string(),
Arc::new(mem_table2));
+ let result2 =
+ schema_provider.register_table("dup_table".to_string(),
Arc::new(mem_table2));
assert!(result2.is_err());
let err = result2.unwrap_err();
assert!(
@@ -422,7 +439,8 @@ mod tests {
let mem_table = MemTable::try_new(arrow_schema,
vec![vec![empty_batch]]).unwrap();
// Register the table
- let result = schema_provider.register_table("drop_me".to_string(),
Arc::new(mem_table));
+ let result =
+ schema_provider.register_table("drop_me".to_string(),
Arc::new(mem_table));
assert!(result.is_ok());
assert!(schema_provider.table_exist("drop_me"));
diff --git a/crates/datafusion/src/table/mod.rs
b/crates/datafusion/src/table/mod.rs
index 9de7bcb..a1db3c8 100644
--- a/crates/datafusion/src/table/mod.rs
+++ b/crates/datafusion/src/table/mod.rs
@@ -179,8 +179,8 @@ impl TableProvider for IcebergTableProvider {
};
// Step 2: Repartition for parallel processing
- let target_partitions =
-
NonZeroUsize::new(state.config().target_partitions()).ok_or_else(|| {
+ let target_partitions =
NonZeroUsize::new(state.config().target_partitions())
+ .ok_or_else(|| {
DataFusionError::Configuration(
"target_partitions must be greater than 0".to_string(),
)
@@ -267,7 +267,10 @@ impl IcebergStaticTableProvider {
///
/// Queries the specified snapshot for all operations. Useful for
time-travel queries.
/// Does not support write operations.
- pub async fn try_new_from_table_snapshot(table: Table, snapshot_id: i64)
-> Result<Self> {
+ pub async fn try_new_from_table_snapshot(
+ table: Table,
+ snapshot_id: i64,
+ ) -> Result<Self> {
let snapshot = table
.metadata()
.snapshot_by_id(snapshot_id)
@@ -366,22 +369,30 @@ mod tests {
metadata_file_name
);
let file_io = FileIO::new_with_fs();
- let static_identifier = TableIdent::from_strs(["static_ns",
"static_table"]).unwrap();
- let static_table =
- StaticTable::from_metadata_file(&metadata_file_path,
static_identifier, file_io)
- .await
- .unwrap();
+ let static_identifier =
+ TableIdent::from_strs(["static_ns", "static_table"]).unwrap();
+ let static_table = StaticTable::from_metadata_file(
+ &metadata_file_path,
+ static_identifier,
+ file_io,
+ )
+ .await
+ .unwrap();
static_table.into_table()
}
- async fn get_test_catalog_and_table() -> (Arc<dyn Catalog>,
NamespaceIdent, String, TempDir) {
+ async fn get_test_catalog_and_table()
+ -> (Arc<dyn Catalog>, NamespaceIdent, String, TempDir) {
let temp_dir = TempDir::new().unwrap();
let warehouse_path = temp_dir.path().to_str().unwrap().to_string();
let catalog = MemoryCatalogBuilder::default()
.load(
"memory",
- HashMap::from([(MEMORY_CATALOG_WAREHOUSE.to_string(),
warehouse_path.clone())]),
+ HashMap::from([(
+ MEMORY_CATALOG_WAREHOUSE.to_string(),
+ warehouse_path.clone(),
+ )]),
)
.await
.unwrap();
@@ -395,8 +406,10 @@ mod tests {
let schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()
.unwrap();
@@ -426,9 +439,10 @@ mod tests {
#[tokio::test]
async fn test_static_provider_from_table() {
let table = get_test_table_from_metadata_file().await;
- let table_provider =
IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
let ctx = SessionContext::new();
ctx.register_table("mytable", Arc::new(table_provider))
.unwrap();
@@ -451,10 +465,12 @@ mod tests {
async fn test_static_provider_from_snapshot() {
let table = get_test_table_from_metadata_file().await;
let snapshot_id =
table.metadata().snapshots().next().unwrap().snapshot_id();
- let table_provider =
-
IcebergStaticTableProvider::try_new_from_table_snapshot(table.clone(),
snapshot_id)
- .await
- .unwrap();
+ let table_provider =
IcebergStaticTableProvider::try_new_from_table_snapshot(
+ table.clone(),
+ snapshot_id,
+ )
+ .await
+ .unwrap();
let ctx = SessionContext::new();
ctx.register_table("mytable", Arc::new(table_provider))
.unwrap();
@@ -476,9 +492,10 @@ mod tests {
#[tokio::test]
async fn test_static_provider_rejects_writes() {
let table = get_test_table_from_metadata_file().await;
- let table_provider =
IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
let ctx = SessionContext::new();
ctx.register_table("mytable", Arc::new(table_provider))
.unwrap();
@@ -499,9 +516,10 @@ mod tests {
#[tokio::test]
async fn test_static_provider_scan() {
let table = get_test_table_from_metadata_file().await;
- let table_provider =
IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
let ctx = SessionContext::new();
ctx.register_table("mytable", Arc::new(table_provider))
.unwrap();
@@ -516,13 +534,17 @@ mod tests {
#[tokio::test]
async fn test_catalog_backed_provider_creation() {
- let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
+ let (catalog, namespace, table_name, _temp_dir) =
+ get_test_catalog_and_table().await;
// Test creating a catalog-backed provider
- let provider =
- IcebergTableProvider::try_new(catalog.clone(), namespace.clone(),
table_name.clone())
- .await
- .unwrap();
+ let provider = IcebergTableProvider::try_new(
+ catalog.clone(),
+ namespace.clone(),
+ table_name.clone(),
+ )
+ .await
+ .unwrap();
// Verify the schema is loaded correctly
let schema = provider.schema();
@@ -533,12 +555,16 @@ mod tests {
#[tokio::test]
async fn test_catalog_backed_provider_scan() {
- let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
+ let (catalog, namespace, table_name, _temp_dir) =
+ get_test_catalog_and_table().await;
- let provider =
- IcebergTableProvider::try_new(catalog.clone(), namespace.clone(),
table_name.clone())
- .await
- .unwrap();
+ let provider = IcebergTableProvider::try_new(
+ catalog.clone(),
+ namespace.clone(),
+ table_name.clone(),
+ )
+ .await
+ .unwrap();
let ctx = SessionContext::new();
ctx.register_table("test_table", Arc::new(provider))
@@ -559,12 +585,16 @@ mod tests {
#[tokio::test]
async fn test_catalog_backed_provider_insert() {
- let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
+ let (catalog, namespace, table_name, _temp_dir) =
+ get_test_catalog_and_table().await;
- let provider =
- IcebergTableProvider::try_new(catalog.clone(), namespace.clone(),
table_name.clone())
- .await
- .unwrap();
+ let provider = IcebergTableProvider::try_new(
+ catalog.clone(),
+ namespace.clone(),
+ table_name.clone(),
+ )
+ .await
+ .unwrap();
let ctx = SessionContext::new();
ctx.register_table("test_table", Arc::new(provider))
@@ -586,12 +616,16 @@ mod tests {
#[tokio::test]
async fn test_physical_input_schema_consistent_with_logical_input_schema()
{
- let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
+ let (catalog, namespace, table_name, _temp_dir) =
+ get_test_catalog_and_table().await;
- let provider =
- IcebergTableProvider::try_new(catalog.clone(), namespace.clone(),
table_name.clone())
- .await
- .unwrap();
+ let provider = IcebergTableProvider::try_new(
+ catalog.clone(),
+ namespace.clone(),
+ table_name.clone(),
+ )
+ .await
+ .unwrap();
let ctx = SessionContext::new();
ctx.register_table("test_table", Arc::new(provider))
@@ -634,7 +668,10 @@ mod tests {
let catalog = MemoryCatalogBuilder::default()
.load(
"memory",
- HashMap::from([(MEMORY_CATALOG_WAREHOUSE.to_string(),
warehouse_path.clone())]),
+ HashMap::from([(
+ MEMORY_CATALOG_WAREHOUSE.to_string(),
+ warehouse_path.clone(),
+ )]),
)
.await
.unwrap();
@@ -648,8 +685,14 @@ mod tests {
let schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "category",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(
+ 2,
+ "category",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
])
.build()
.unwrap();
@@ -706,7 +749,8 @@ mod tests {
async fn test_catalog_backed_provider_rejects_non_append_op() {
use datafusion::physical_plan::empty::EmptyExec;
- let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
+ let (catalog, namespace, table_name, _temp_dir) =
+ get_test_catalog_and_table().await;
let provider = IcebergTableProvider::try_new(catalog, namespace,
table_name)
.await
.unwrap();
@@ -722,7 +766,8 @@ mod tests {
"IcebergTableProvider supports only append inserts, got
Replace Into",
),
] {
- let input = Arc::new(EmptyExec::new(provider.schema())) as Arc<dyn
ExecutionPlan>;
+ let input =
+ Arc::new(EmptyExec::new(provider.schema())) as Arc<dyn
ExecutionPlan>;
let error = provider
.insert_into(&ctx.state(), input, insert_op)
.await
@@ -748,10 +793,13 @@ mod tests {
let (catalog, namespace, table_name, _temp_dir) =
get_partitioned_test_catalog_and_table(Some(true)).await;
- let provider =
- IcebergTableProvider::try_new(catalog.clone(), namespace.clone(),
table_name.clone())
- .await
- .unwrap();
+ let provider = IcebergTableProvider::try_new(
+ catalog.clone(),
+ namespace.clone(),
+ table_name.clone(),
+ )
+ .await
+ .unwrap();
let ctx = SessionContext::new();
let input_schema = provider.schema();
@@ -780,10 +828,13 @@ mod tests {
let (catalog, namespace, table_name, _temp_dir) =
get_partitioned_test_catalog_and_table(Some(false)).await;
- let provider =
- IcebergTableProvider::try_new(catalog.clone(), namespace.clone(),
table_name.clone())
- .await
- .unwrap();
+ let provider = IcebergTableProvider::try_new(
+ catalog.clone(),
+ namespace.clone(),
+ table_name.clone(),
+ )
+ .await
+ .unwrap();
let ctx = SessionContext::new();
let input_schema = provider.schema();
@@ -807,9 +858,10 @@ mod tests {
use datafusion::datasource::TableProvider;
let table = get_test_table_from_metadata_file().await;
- let table_provider =
IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
let ctx = SessionContext::new();
let state = ctx.state();
@@ -837,12 +889,16 @@ mod tests {
async fn test_limit_pushdown_catalog_backed_provider() {
use datafusion::datasource::TableProvider;
- let (catalog, namespace, table_name, _temp_dir) =
get_test_catalog_and_table().await;
+ let (catalog, namespace, table_name, _temp_dir) =
+ get_test_catalog_and_table().await;
- let provider =
- IcebergTableProvider::try_new(catalog.clone(), namespace.clone(),
table_name.clone())
- .await
- .unwrap();
+ let provider = IcebergTableProvider::try_new(
+ catalog.clone(),
+ namespace.clone(),
+ table_name.clone(),
+ )
+ .await
+ .unwrap();
let ctx = SessionContext::new();
let state = ctx.state();
@@ -868,9 +924,10 @@ mod tests {
use datafusion::datasource::TableProvider;
let table = get_test_table_from_metadata_file().await;
- let table_provider =
IcebergStaticTableProvider::try_new_from_table(table.clone())
- .await
- .unwrap();
+ let table_provider =
+ IcebergStaticTableProvider::try_new_from_table(table.clone())
+ .await
+ .unwrap();
let ctx = SessionContext::new();
let state = ctx.state();
diff --git a/crates/datafusion/src/table/table_provider_factory.rs
b/crates/datafusion/src/table/table_provider_factory.rs
index 6d81700..656854d 100644
--- a/crates/datafusion/src/table/table_provider_factory.rs
+++ b/crates/datafusion/src/table/table_provider_factory.rs
@@ -196,7 +196,9 @@ fn check_cmd(cmd: &CreateExternalTable) -> Result<&str> {
/// # See also
/// - [`iceberg::NamespaceIdent`]
/// - [`datafusion::sql::planner::SqlToRel::external_table_to_plan`]
-fn complement_namespace_if_necessary(table_name: &TableReference) -> Cow<'_,
TableReference> {
+fn complement_namespace_if_necessary(
+ table_name: &TableReference,
+) -> Cow<'_, TableReference> {
match table_name {
TableReference::Bare { table } => {
Cow::Owned(TableReference::partial("default", table.as_ref()))
diff --git a/crates/datafusion/src/task_writer.rs
b/crates/datafusion/src/task_writer.rs
index 99ba5c0..3e7c713 100644
--- a/crates/datafusion/src/task_writer.rs
+++ b/crates/datafusion/src/task_writer.rs
@@ -186,10 +186,12 @@ impl<B: IcebergWriterBuilder> TaskWriter<B> {
writer.write(batch).await
}
SupportedWriter::Fanout(writer) => {
- Self::write_partitioned_batches(writer,
&self.partition_splitter, &batch).await
+ Self::write_partitioned_batches(writer,
&self.partition_splitter, &batch)
+ .await
}
SupportedWriter::Clustered(writer) => {
- Self::write_partitioned_batches(writer,
&self.partition_splitter, &batch).await
+ Self::write_partitioned_batches(writer,
&self.partition_splitter, &batch)
+ .await
}
}
}
@@ -264,11 +266,15 @@ mod tests {
use std::collections::HashMap;
use std::sync::Arc;
- use datafusion::arrow::array::{ArrayRef, Int32Array, RecordBatch,
StringArray, StructArray};
+ use datafusion::arrow::array::{
+ ArrayRef, Int32Array, RecordBatch, StringArray, StructArray,
+ };
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use iceberg::arrow::PROJECTED_PARTITION_VALUE_COLUMN;
use iceberg::io::FileIO;
- use iceberg::spec::{DataFileFormat, NestedField, PartitionSpec,
PrimitiveType, Type};
+ use iceberg::spec::{
+ DataFileFormat, NestedField, PartitionSpec, PrimitiveType, Type,
+ };
use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
use iceberg::writer::file_writer::ParquetWriterBuilder;
use iceberg::writer::file_writer::location_generator::{
@@ -286,10 +292,20 @@ mod tests {
iceberg::spec::Schema::builder()
.with_schema_id(1)
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
- NestedField::required(3, "region",
Type::Primitive(PrimitiveType::String))
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
.into(),
+ NestedField::required(
+ 2,
+ "name",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
+ NestedField::required(
+ 3,
+ "region",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
])
.build()?,
))
@@ -353,8 +369,11 @@ mod tests {
let location_gen = DefaultLocationGenerator::with_data_location(
temp_dir.path().to_str().unwrap().to_string(),
);
- let file_name_gen =
- DefaultFileNameGenerator::new("test".to_string(), None,
DataFileFormat::Parquet);
+ let file_name_gen = DefaultFileNameGenerator::new(
+ "test".to_string(),
+ None,
+ DataFileFormat::Parquet,
+ );
let parquet_writer_builder =
ParquetWriterBuilder::new(WriterProperties::builder().build(),
schema);
let rolling_writer_builder =
RollingFileWriterBuilder::new_with_default_file_size(
@@ -376,14 +395,18 @@ mod tests {
let partition_spec =
Arc::new(PartitionSpec::builder(schema.clone()).build()?);
let writer_builder = create_writer_builder(&temp_dir, schema.clone())?;
- let mut task_writer = TaskWriter::try_new(writer_builder, false,
schema, partition_spec)?;
+ let mut task_writer =
+ TaskWriter::try_new(writer_builder, false, schema,
partition_spec)?;
// Write data
- let batch = RecordBatch::try_new(arrow_schema, vec![
- Arc::new(Int32Array::from(vec![1, 2, 3])),
- Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie"])),
- Arc::new(StringArray::from(vec!["US", "EU", "US"])),
- ])?;
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie"])),
+ Arc::new(StringArray::from(vec!["US", "EU", "US"])),
+ ],
+ )?;
task_writer.write(batch).await?;
let data_files = task_writer.close().await?;
@@ -417,7 +440,8 @@ mod tests {
_ => panic!("Expected string partition value"),
};
- *partition_counts.entry(region.clone()).or_insert(0) +=
data_file.record_count();
+ *partition_counts.entry(region.clone()).or_insert(0) +=
+ data_file.record_count();
// Verify file path contains partition information
assert!(
@@ -437,12 +461,17 @@ mod tests {
let partition_spec = Arc::new(
PartitionSpec::builder(schema.clone())
.with_spec_id(1)
- .add_partition_field("region", "region",
iceberg::spec::Transform::Identity)?
+ .add_partition_field(
+ "region",
+ "region",
+ iceberg::spec::Transform::Identity,
+ )?
.build()?,
);
let writer_builder = create_writer_builder(&temp_dir, schema.clone())?;
- let mut task_writer = TaskWriter::try_new(writer_builder, true,
schema, partition_spec)?;
+ let mut task_writer =
+ TaskWriter::try_new(writer_builder, true, schema, partition_spec)?;
// Create partition column
let partition_field = Field::new("region", DataType::Utf8,
false).with_metadata(
@@ -454,12 +483,15 @@ mod tests {
Arc::new(partition_values) as ArrayRef,
)]);
- let batch = RecordBatch::try_new(arrow_schema, vec![
- Arc::new(Int32Array::from(vec![1, 2, 3, 4])),
- Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie",
"Dave"])),
- Arc::new(StringArray::from(vec!["US", "EU", "US", "EU"])),
- Arc::new(partition_struct),
- ])?;
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])),
+ Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie",
"Dave"])),
+ Arc::new(StringArray::from(vec!["US", "EU", "US", "EU"])),
+ Arc::new(partition_struct),
+ ],
+ )?;
task_writer.write(batch).await?;
let data_files = task_writer.close().await?;
@@ -480,12 +512,17 @@ mod tests {
let partition_spec = Arc::new(
PartitionSpec::builder(schema.clone())
.with_spec_id(1)
- .add_partition_field("region", "region",
iceberg::spec::Transform::Identity)?
+ .add_partition_field(
+ "region",
+ "region",
+ iceberg::spec::Transform::Identity,
+ )?
.build()?,
);
let writer_builder = create_writer_builder(&temp_dir, schema.clone())?;
- let mut task_writer = TaskWriter::try_new(writer_builder, false,
schema, partition_spec)?;
+ let mut task_writer =
+ TaskWriter::try_new(writer_builder, false, schema,
partition_spec)?;
// Create partition column
let partition_field = Field::new("region", DataType::Utf8,
false).with_metadata(
@@ -498,12 +535,15 @@ mod tests {
)]);
// ClusteredWriter expects data to be pre-sorted by partition
- let batch = RecordBatch::try_new(arrow_schema, vec![
- Arc::new(Int32Array::from(vec![1, 2, 3, 4])),
- Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie",
"Dave"])),
- Arc::new(StringArray::from(vec!["ASIA", "ASIA", "EU", "EU"])),
- Arc::new(partition_struct),
- ])?;
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])),
+ Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie",
"Dave"])),
+ Arc::new(StringArray::from(vec!["ASIA", "ASIA", "EU", "EU"])),
+ Arc::new(partition_struct),
+ ],
+ )?;
task_writer.write(batch).await?;
let data_files = task_writer.close().await?;
diff --git a/crates/datafusion/tests/integration_datafusion_test.rs
b/crates/datafusion/tests/integration_datafusion_test.rs
index 83f780f..41753ab 100644
--- a/crates/datafusion/tests/integration_datafusion_test.rs
+++ b/crates/datafusion/tests/integration_datafusion_test.rs
@@ -34,7 +34,8 @@ use iceberg::spec::{
};
use iceberg::test_utils::check_record_batches;
use iceberg::{
- Catalog, CatalogBuilder, MemoryCatalog, NamespaceIdent, Result,
TableCreation, TableIdent,
+ Catalog, CatalogBuilder, MemoryCatalog, NamespaceIdent, Result,
TableCreation,
+ TableIdent,
};
use tempfile::TempDir;
@@ -61,7 +62,10 @@ fn get_struct_type() -> StructType {
])
}
-async fn set_test_namespace(catalog: &MemoryCatalog, namespace:
&NamespaceIdent) -> Result<()> {
+async fn set_test_namespace(
+ catalog: &MemoryCatalog,
+ namespace: &NamespaceIdent,
+) -> Result<()> {
let properties = HashMap::new();
catalog.create_namespace(namespace, properties).await?;
@@ -78,8 +82,10 @@ fn get_table_creation(
None => Schema::builder()
.with_schema_id(0)
.with_fields(vec![
- NestedField::required(1, "foo1",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "foo2",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "foo1",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(2, "foo2",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()?,
Some(schema) => schema,
@@ -216,7 +222,8 @@ async fn test_table_projection() -> Result<()> {
.with_schema_id(0)
.with_fields(vec![
NestedField::required(1, "foo1",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "foo2",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(2, "foo2",
Type::Primitive(PrimitiveType::String))
+ .into(),
NestedField::optional(3, "foo3",
Type::Struct(get_struct_type())).into(),
])
.build()?;
@@ -284,7 +291,8 @@ async fn test_table_predict_pushdown() -> Result<()> {
.with_schema_id(0)
.with_fields(vec![
NestedField::required(1, "foo",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::optional(2, "bar",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::optional(2, "bar",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()?;
let creation = get_table_creation(temp_path(), "t1", Some(schema))?;
@@ -329,7 +337,8 @@ async fn test_metadata_table() -> Result<()> {
.with_schema_id(0)
.with_fields(vec![
NestedField::required(1, "foo",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::optional(2, "bar",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::optional(2, "bar",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()?;
let creation = get_table_creation(temp_path(), "t1", Some(schema))?;
@@ -536,9 +545,16 @@ fn get_nested_struct_type() -> StructType {
10,
"address",
Type::Struct(StructType::new(vec![
- NestedField::optional(11, "street",
Type::Primitive(PrimitiveType::String)).into(),
- NestedField::optional(12, "city",
Type::Primitive(PrimitiveType::String)).into(),
- NestedField::optional(13, "zip",
Type::Primitive(PrimitiveType::Int)).into(),
+ NestedField::optional(
+ 11,
+ "street",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
+ NestedField::optional(12, "city",
Type::Primitive(PrimitiveType::String))
+ .into(),
+ NestedField::optional(13, "zip",
Type::Primitive(PrimitiveType::Int))
+ .into(),
])),
)
.into(),
@@ -546,8 +562,18 @@ fn get_nested_struct_type() -> StructType {
20,
"contact",
Type::Struct(StructType::new(vec![
- NestedField::optional(21, "email",
Type::Primitive(PrimitiveType::String)).into(),
- NestedField::optional(22, "phone",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::optional(
+ 21,
+ "email",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
+ NestedField::optional(
+ 22,
+ "phone",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
])),
)
.into(),
@@ -566,8 +592,10 @@ async fn test_insert_into_nested() -> Result<()> {
.with_schema_id(0)
.with_fields(vec![
NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
- NestedField::optional(3, "profile",
Type::Struct(get_nested_struct_type())).into(),
+ NestedField::required(2, "name",
Type::Primitive(PrimitiveType::String))
+ .into(),
+ NestedField::optional(3, "profile",
Type::Struct(get_nested_struct_type()))
+ .into(),
])
.build()?;
@@ -821,8 +849,10 @@ async fn test_insert_into_partitioned() -> Result<()> {
.with_schema_id(0)
.with_fields(vec![
NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "category",
Type::Primitive(PrimitiveType::String)).into(),
- NestedField::required(3, "value",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(2, "category",
Type::Primitive(PrimitiveType::String))
+ .into(),
+ NestedField::required(3, "value",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()?;
diff --git a/crates/playground/src/catalog.rs b/crates/playground/src/catalog.rs
index 10a7ac3..3aa19e8 100644
--- a/crates/playground/src/catalog.rs
+++ b/crates/playground/src/catalog.rs
@@ -21,11 +21,11 @@ use std::sync::Arc;
use anyhow::anyhow;
use datafusion::catalog::{CatalogProvider, CatalogProviderList};
+use datafusion_iceberg::IcebergCatalogProvider;
use fs_err::read_to_string;
use iceberg::CatalogBuilder;
use iceberg::memory::MemoryCatalogBuilder;
use iceberg_catalog_rest::RestCatalogBuilder;
-use datafusion_iceberg::IcebergCatalogProvider;
use toml::{Table as TomlTable, Value};
const CONFIG_NAME_CATALOGS: &str = "catalogs";
@@ -44,7 +44,9 @@ impl IcebergCatalogList {
pub async fn parse_table(configs: &TomlTable) -> anyhow::Result<Self> {
if let Value::Array(catalogs_config) =
configs.get(CONFIG_NAME_CATALOGS).ok_or_else(|| {
- anyhow::Error::msg(format!("{CONFIG_NAME_CATALOGS} entry not
found in config"))
+ anyhow::Error::msg(format!(
+ "{CONFIG_NAME_CATALOGS} entry not found in config"
+ ))
})?
{
let mut catalogs = HashMap::with_capacity(catalogs_config.len());
@@ -96,7 +98,9 @@ impl IcebergCatalogList {
// Create catalog based on type using the appropriate builder
let catalog: Arc<dyn iceberg::Catalog> = match r#type {
"rest" => Arc::new(RestCatalogBuilder::default().load(name,
props).await?),
- "memory" => Arc::new(MemoryCatalogBuilder::default().load(name,
props).await?),
+ "memory" => {
+ Arc::new(MemoryCatalogBuilder::default().load(name,
props).await?)
+ }
_ => {
return Err(anyhow::anyhow!(
"Unsupported catalog type: '{type}'. Supported types:
rest, memory"
diff --git a/crates/playground/src/main.rs b/crates/playground/src/main.rs
index 94068bb..853494c 100644
--- a/crates/playground/src/main.rs
+++ b/crates/playground/src/main.rs
@@ -87,7 +87,8 @@ async fn main_inner() -> anyhow::Result<()> {
let runtime_env = rt_builder.build_arc()?;
// enable dynamic file query
- let ctx = SessionContext::new_with_config_rt(session_config,
runtime_env).enable_url_table();
+ let ctx = SessionContext::new_with_config_rt(session_config, runtime_env)
+ .enable_url_table();
ctx.refresh_catalogs().await?;
let mut print_options = PrintOptions {
diff --git a/crates/sqllogictest/src/engine/datafusion.rs
b/crates/sqllogictest/src/engine/datafusion.rs
index 56008a0..e5fc045 100644
--- a/crates/sqllogictest/src/engine/datafusion.rs
+++ b/crates/sqllogictest/src/engine/datafusion.rs
@@ -21,14 +21,15 @@ use std::sync::Arc;
use datafusion::catalog::CatalogProvider;
use datafusion::prelude::{SessionConfig, SessionContext};
+use datafusion_iceberg::IcebergCatalogProvider;
use datafusion_sqllogictest::DataFusion;
use iceberg::encryption::kms::MemoryKmsClientFactory;
use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
use iceberg::spec::{
- NestedField, PrimitiveType, Schema, TableProperties, Transform, Type,
UnboundPartitionSpec,
+ NestedField, PrimitiveType, Schema, TableProperties, Transform, Type,
+ UnboundPartitionSpec,
};
use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation};
-use datafusion_iceberg::IcebergCatalogProvider;
use indicatif::ProgressBar;
use crate::engine::{DatafusionCatalogConfig, EngineRunner,
run_slt_with_runner};
@@ -121,9 +122,16 @@ impl DataFusionEngine {
) -> anyhow::Result<()> {
let schema = Schema::builder()
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "category",
Type::Primitive(PrimitiveType::String)).into(),
- NestedField::optional(3, "value",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::required(
+ 2,
+ "category",
+ Type::Primitive(PrimitiveType::String),
+ )
+ .into(),
+ NestedField::optional(3, "value",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()?;
@@ -155,8 +163,10 @@ impl DataFusionEngine {
) -> anyhow::Result<()> {
let schema = Schema::builder()
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::optional(2, "data",
Type::Primitive(PrimitiveType::Binary)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::optional(2, "data",
Type::Primitive(PrimitiveType::Binary))
+ .into(),
])
.build()?;
@@ -179,8 +189,10 @@ impl DataFusionEngine {
) -> anyhow::Result<()> {
let schema = Schema::builder()
.with_fields(vec![
- NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::optional(2, "name",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int))
+ .into(),
+ NestedField::optional(2, "name",
Type::Primitive(PrimitiveType::String))
+ .into(),
])
.build()?;
diff --git a/crates/sqllogictest/src/engine/mod.rs
b/crates/sqllogictest/src/engine/mod.rs
index a276671..aa9292b 100644
--- a/crates/sqllogictest/src/engine/mod.rs
+++ b/crates/sqllogictest/src/engine/mod.rs
@@ -55,7 +55,9 @@ pub trait EngineRunner: Send {
pub async fn load_engine_runner(config: EngineConfig) -> Result<Box<dyn
EngineRunner>> {
match config {
- EngineConfig::Datafusion { catalog } =>
Ok(Box::new(DataFusionEngine::new(catalog).await?)),
+ EngineConfig::Datafusion { catalog } => {
+ Ok(Box::new(DataFusionEngine::new(catalog).await?))
+ }
}
}
@@ -68,7 +70,8 @@ where
M: MakeConnection<Conn = D> + Send + 'static,
{
let path = step_slt_file.as_ref().canonicalize()?;
- let records = parse_file(&path).map_err(|e| Error(anyhow!("parsing slt
file failed: {e}")))?;
+ let records =
+ parse_file(&path).map_err(|e| Error(anyhow!("parsing slt file failed:
{e}")))?;
for record in records {
if let Err(err) = runner.run_async(record).await {
diff --git a/crates/sqllogictest/src/schedule.rs
b/crates/sqllogictest/src/schedule.rs
index 29e37c5..4dc3da3 100644
--- a/crates/sqllogictest/src/schedule.rs
+++ b/crates/sqllogictest/src/schedule.rs
@@ -147,9 +147,10 @@ mod tests {
assert_eq!(config.engines.len(), 1);
assert!(config.engines.contains_key("df"));
- assert!(matches!(config.engines["df"], EngineConfig::Datafusion {
- catalog: None
- }));
+ assert!(matches!(
+ config.engines["df"],
+ EngineConfig::Datafusion { catalog: None }
+ ));
assert_eq!(config.steps.len(), 1);
assert_eq!(config.steps[0].engine, "df");
assert_eq!(config.steps[0].slt, "test.slt");
diff --git a/rust-toolchain.toml b/rust-toolchain.toml
new file mode 100644
index 0000000..52d0bc7
--- /dev/null
+++ b/rust-toolchain.toml
@@ -0,0 +1,21 @@
+# 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.
+
+# Keep local development and CI on the same reviewed toolchain.
+[toolchain]
+channel = "1.98.1"
+components = ["rustfmt", "clippy"]
diff --git a/rustfmt.toml b/rustfmt.toml
new file mode 100644
index 0000000..5b864b0
--- /dev/null
+++ b/rustfmt.toml
@@ -0,0 +1,19 @@
+# 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.
+
+edition = "2024"
+max_width = 90
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
