You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by gi...@apache.org on 2026/09/22 07:24:02 UTC
(iceberg-rust) 01/01: Introduce invalid data error macro (#2928)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-2928-bb1e4a4861f02377489eff818b75138f414c4cb0
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit b2a042bd2ca4d332ff437773423ba3179531497a
Author: Xander <za...@googlemail.com>
AuthorDate: Tue Sep 22 07:23:37 2026 +0000
Introduce invalid data error macro (#2928)
* Introduce invalid data error macro
* fix
* comments
---
.../src/arrow/caching_delete_file_loader.rs | 78 ++----
.../src/arrow/partition_value_calculator.rs | 20 +-
crates/iceberg/src/arrow/reader/pipeline.rs | 11 +-
.../iceberg/src/arrow/reader/predicate_visitor.rs | 19 +-
crates/iceberg/src/arrow/reader/projection.rs | 9 +-
crates/iceberg/src/arrow/reader/row_lineage.rs | 7 +-
.../src/arrow/record_batch_partition_splitter.rs | 27 +--
crates/iceberg/src/arrow/record_batch_projector.rs | 13 +-
.../iceberg/src/arrow/record_batch_transformer.rs | 32 +--
crates/iceberg/src/arrow/schema.rs | 118 +++-------
crates/iceberg/src/arrow/value.rs | 262 +++++----------------
crates/iceberg/src/avro/schema.rs | 53 +++--
crates/iceberg/src/catalog/memory/catalog.rs | 11 +-
.../iceberg/src/catalog/memory/namespace_state.rs | 6 +-
crates/iceberg/src/catalog/mod.rs | 12 +-
crates/iceberg/src/delete_file_index.rs | 61 ++---
crates/iceberg/src/delete_vector.rs | 65 ++---
crates/iceberg/src/encryption/crypto.rs | 32 +--
crates/iceberg/src/encryption/key_metadata.rs | 26 +-
crates/iceberg/src/encryption/kms/memory.rs | 16 +-
crates/iceberg/src/encryption/manager.rs | 44 +---
crates/iceberg/src/encryption/stream.rs | 72 +++---
crates/iceberg/src/error.rs | 71 ++++++
crates/iceberg/src/expr/predicate.rs | 11 +-
crates/iceberg/src/expr/term.rs | 21 +-
.../src/expr/visitors/strict_metrics_evaluator.rs | 8 +-
crates/iceberg/src/inspect/manifests.rs | 13 +-
crates/iceberg/src/io/storage/config/s3.rs | 10 +-
crates/iceberg/src/io/storage/local_fs.rs | 45 +---
crates/iceberg/src/io/storage/memory.rs | 36 +--
crates/iceberg/src/partitioning.rs | 20 +-
crates/iceberg/src/puffin/metadata.rs | 56 ++---
crates/iceberg/src/puffin/mod.rs | 18 +-
crates/iceberg/src/scan/mod.rs | 13 +-
crates/iceberg/src/scan/task.rs | 32 +--
crates/iceberg/src/spec/manifest/_serde.rs | 15 +-
crates/iceberg/src/spec/manifest/data_file.rs | 14 +-
crates/iceberg/src/spec/manifest/entry.rs | 9 +-
crates/iceberg/src/spec/manifest/metadata.rs | 45 +---
crates/iceberg/src/spec/manifest/mod.rs | 21 +-
crates/iceberg/src/spec/manifest/reader.rs | 18 +-
crates/iceberg/src/spec/manifest/writer.rs | 44 ++--
crates/iceberg/src/spec/manifest_list/_serde.rs | 68 +-----
.../src/spec/manifest_list/manifest_file.rs | 14 +-
crates/iceberg/src/spec/manifest_list/writer.rs | 41 +---
crates/iceberg/src/spec/name_mapping/mod.rs | 9 +-
crates/iceberg/src/spec/partition.rs | 98 +++-----
crates/iceberg/src/spec/schema/id_reassigner.rs | 28 +--
crates/iceberg/src/spec/schema/index.rs | 8 +-
crates/iceberg/src/spec/schema/mod.rs | 23 +-
crates/iceberg/src/spec/schema/prune_columns.rs | 25 +-
crates/iceberg/src/spec/schema/utils.rs | 8 +-
crates/iceberg/src/spec/snapshot.rs | 23 +-
crates/iceberg/src/spec/snapshot_summary.rs | 6 +-
crates/iceberg/src/spec/table_metadata.rs | 190 +++++----------
crates/iceberg/src/spec/table_metadata_builder.rs | 242 +++++++------------
crates/iceberg/src/spec/table_properties.rs | 47 ++--
crates/iceberg/src/spec/transform.rs | 97 +++-----
crates/iceberg/src/spec/values/datum.rs | 122 ++++------
crates/iceberg/src/spec/values/decimal_utils.rs | 6 +-
crates/iceberg/src/spec/values/literal.rs | 185 ++++++---------
crates/iceberg/src/spec/values/serde.rs | 49 ++--
crates/iceberg/src/spec/view_metadata.rs | 17 +-
crates/iceberg/src/spec/view_metadata_builder.rs | 76 ++----
crates/iceberg/src/spec/view_version.rs | 10 +-
crates/iceberg/src/table.rs | 33 ++-
crates/iceberg/src/transaction/expire_snapshots.rs | 20 +-
crates/iceberg/src/transaction/snapshot.rs | 31 +--
crates/iceberg/src/transaction/sort_order.rs | 13 +-
crates/iceberg/src/transaction/update_location.rs | 8 +-
.../src/transaction/upgrade_format_version.rs | 8 +-
crates/iceberg/src/transform/temporal.rs | 26 +-
crates/iceberg/src/transform/truncate.rs | 11 +-
.../src/writer/base_writer/data_file_writer.rs | 9 +-
.../writer/base_writer/equality_delete_writer.rs | 28 +--
.../writer/base_writer/position_delete_writer.rs | 80 +++----
.../src/writer/file_writer/parquet_writer.rs | 45 ++--
77 files changed, 1130 insertions(+), 2088 deletions(-)
diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs
index 905fa86d8..0208a678e 100644
--- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs
+++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs
@@ -29,6 +29,7 @@ use crate::arrow::scan_metrics::ScanMetrics;
use crate::arrow::{arrow_primitive_to_literal, arrow_schema_to_schema};
use crate::delete_vector::DeleteVector;
use crate::encryption::{EncryptedInputFile, StandardKeyMetadata};
+use crate::error::invalid_data;
use crate::expr::Predicate::AlwaysTrue;
use crate::expr::{Predicate, Reference};
use crate::io::FileIO;
@@ -348,55 +349,37 @@ impl CachingDeleteFileLoader {
task: &FileScanTaskDeleteFile,
) -> Result<(u64, u64, String, u64)> {
let content_offset = task.content_offset.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {} is missing content_offset",
- task.file_path
- ),
+ invalid_data!(
+ "deletion vector {} is missing content_offset",
+ task.file_path
)
})?;
let content_size = task.content_size_in_bytes.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {} is missing content_size_in_bytes",
- task.file_path
- ),
+ invalid_data!(
+ "deletion vector {} is missing content_size_in_bytes",
+ task.file_path
)
})?;
let data_file_path = task.referenced_data_file.clone().ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {} is missing referenced_data_file",
- task.file_path
- ),
+ invalid_data!(
+ "deletion vector {} is missing referenced_data_file",
+ task.file_path
)
})?;
let record_count = task.record_count.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("deletion vector {} is missing record_count", task.file_path),
- )
+ invalid_data!("deletion vector {} is missing record_count", task.file_path)
})?;
let start = u64::try_from(content_offset).map_err(|_| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {} has negative content_offset {content_offset}",
- task.file_path
- ),
+ invalid_data!(
+ "deletion vector {} has negative content_offset {content_offset}",
+ task.file_path
)
})?;
let len = u64::try_from(content_size).map_err(|_| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {} has negative content_size_in_bytes {content_size}",
- task.file_path
- ),
+ invalid_data!(
+ "deletion vector {} has negative content_size_in_bytes {content_size}",
+ task.file_path
)
})?;
@@ -412,11 +395,8 @@ impl CachingDeleteFileLoader {
) -> Result<()> {
let actual = delete_vector.len();
if actual != expected {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {dv_path} decoded to {actual} positions, expected {expected} from record_count"
- ),
+ return Err(invalid_data!(
+ "deletion vector {dv_path} decoded to {actual} positions, expected {expected} from record_count"
));
}
Ok(())
@@ -528,15 +508,13 @@ impl CachingDeleteFileLoader {
let columns = batch.columns();
let Some(file_paths) = columns[0].as_any().downcast_ref::<StringArray>() else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Could not downcast file paths array to StringArray",
+ return Err(invalid_data!(
+ "Could not downcast file paths array to StringArray"
));
};
let Some(positions) = columns[1].as_any().downcast_ref::<Int64Array>() else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Could not downcast positions array to Int64Array",
+ return Err(invalid_data!(
+ "Could not downcast positions array to Int64Array"
));
};
@@ -551,15 +529,11 @@ impl CachingDeleteFileLoader {
for (file_path, pos) in file_paths.iter().zip(positions.iter()) {
let (Some(file_path), Some(pos)) = (file_path, pos) else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "null values in delete file",
- ));
+ return Err(invalid_data!("null values in delete file"));
};
if pos < 0 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("negative position in delete file {file_path}: {pos}"),
+ return Err(invalid_data!(
+ "negative position in delete file {file_path}: {pos}"
));
}
diff --git a/crates/iceberg/src/arrow/partition_value_calculator.rs b/crates/iceberg/src/arrow/partition_value_calculator.rs
index 3520f75ac..c8c917779 100644
--- a/crates/iceberg/src/arrow/partition_value_calculator.rs
+++ b/crates/iceberg/src/arrow/partition_value_calculator.rs
@@ -27,9 +27,10 @@ use arrow_schema::DataType;
use super::record_batch_projector::RecordBatchProjector;
use super::type_to_arrow_type;
+use crate::Result;
+use crate::error::invalid_data;
use crate::spec::{PartitionSpec, Schema, StructType, Type};
use crate::transform::{BoxedTransformFunction, create_transform_function};
-use crate::{Error, ErrorKind, Result};
/// Calculator for partition values in Iceberg tables.
///
@@ -63,9 +64,8 @@ impl PartitionValueCalculator {
/// - Projector initialization fails
pub fn try_new(partition_spec: &PartitionSpec, table_schema: &Schema) -> Result<Self> {
if partition_spec.is_unpartitioned() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot create partition calculator for unpartitioned table",
+ return Err(invalid_data!(
+ "Cannot create partition calculator for unpartitioned table"
));
}
@@ -140,10 +140,7 @@ impl PartitionValueCalculator {
let expected_struct_fields = match &self.partition_arrow_type {
DataType::Struct(fields) => fields.clone(),
_ => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Expected partition type must be a struct",
- ));
+ return Err(invalid_data!("Expected partition type must be a struct"));
}
};
@@ -156,12 +153,7 @@ impl PartitionValueCalculator {
// Construct the StructArray
let struct_array = StructArray::try_new(expected_struct_fields, partition_values, None)
- .map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to create partition struct array: {e}"),
- )
- })?;
+ .map_err(|e| invalid_data!("Failed to create partition struct array: {e}"))?;
Ok(Arc::new(struct_array))
}
diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs
index 7ec9bf4d5..ee027c5e7 100644
--- a/crates/iceberg/src/arrow/reader/pipeline.rs
+++ b/crates/iceberg/src/arrow/reader/pipeline.rs
@@ -43,7 +43,7 @@ use crate::arrow::int96::coerce_int96_timestamps;
use crate::arrow::record_batch_transformer::RecordBatchTransformerBuilder;
use crate::arrow::scan_metrics::{CountingFileRead, ScanMetrics, ScanResult};
use crate::encryption::StandardKeyMetadata;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::expr::BoundPredicate;
use crate::expr::visitors::bloom_filter_evaluator::{
BloomFilterEvaluator, ColumnBloomFilter, collect_bloom_filter_field_ids,
@@ -475,12 +475,9 @@ impl FileScanTaskReader {
// inheritance a committed entry always has one, so this is a malformed
// manifest rather than a legitimate null.
(Some(_), None) => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Data file {} has a first_row_id but no data sequence number",
- task.data_file_path()
- ),
+ return Err(invalid_data!(
+ "Data file {} has a first_row_id but no data sequence number",
+ task.data_file_path()
));
}
};
diff --git a/crates/iceberg/src/arrow/reader/predicate_visitor.rs b/crates/iceberg/src/arrow/reader/predicate_visitor.rs
index 2b05e3ebc..264c34a9a 100644
--- a/crates/iceberg/src/arrow/reader/predicate_visitor.rs
+++ b/crates/iceberg/src/arrow/reader/predicate_visitor.rs
@@ -35,11 +35,10 @@ use fnv::FnvHashSet;
use parquet::schema::types::SchemaDescriptor;
use crate::arrow::get_arrow_datum;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::expr::visitors::bound_predicate_visitor::BoundPredicateVisitor;
use crate::expr::{BoundPredicate, BoundReference};
use crate::spec::Datum;
-use crate::{Error, ErrorKind};
/// A visitor to collect field ids from bound predicates.
pub(super) struct CollectFieldIdVisitor {
@@ -215,12 +214,9 @@ impl PredicateConverter<'_> {
// The leaf column's index in Parquet schema.
if let Some(column_idx) = self.column_map.get(&reference.field().id) {
if self.parquet_schema.get_column_root(*column_idx).is_group() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Leaf column `{}` in predicates isn't a root column in Parquet schema.",
- reference.field().name
- ),
+ return Err(invalid_data!(
+ "Leaf column `{}` in predicates isn't a root column in Parquet schema.",
+ reference.field().name
));
}
@@ -229,13 +225,10 @@ impl PredicateConverter<'_> {
.column_indices
.iter()
.position(|&idx| idx == *column_idx)
- .ok_or(Error::new(
- ErrorKind::DataInvalid,
- format!(
+ .ok_or(invalid_data!(
"Leaf column `{}` in predicates cannot be found in the required column indices.",
reference.field().name
- ),
- ))?;
+ ))?;
Ok(Some(index))
} else {
diff --git a/crates/iceberg/src/arrow/reader/projection.rs b/crates/iceberg/src/arrow/reader/projection.rs
index 873adb05c..0f34f00b3 100644
--- a/crates/iceberg/src/arrow/reader/projection.rs
+++ b/crates/iceberg/src/arrow/reader/projection.rs
@@ -29,7 +29,7 @@ use parquet::schema::types::{SchemaDescriptor, Type as ParquetType};
use super::{ArrowReader, CollectFieldIdVisitor};
use crate::arrow::arrow_schema_to_schema;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::expr::BoundPredicate;
use crate::expr::visitors::bound_predicate_visitor::visit;
use crate::spec::{NameMapping, NestedField, PrimitiveType, Schema, Type};
@@ -299,11 +299,8 @@ pub(super) fn build_field_id_map(
column_map.insert(basic_info.id(), idx);
}
ParquetType::GroupType { .. } => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Leaf column in schema should be primitive type but got {field_type:?}"
- ),
+ return Err(invalid_data!(
+ "Leaf column in schema should be primitive type but got {field_type:?}"
));
}
};
diff --git a/crates/iceberg/src/arrow/reader/row_lineage.rs b/crates/iceberg/src/arrow/reader/row_lineage.rs
index ee8dda4e6..9510f71c5 100644
--- a/crates/iceberg/src/arrow/reader/row_lineage.rs
+++ b/crates/iceberg/src/arrow/reader/row_lineage.rs
@@ -32,6 +32,7 @@ use arrow_schema::{DataType, Field, Schema};
use arrow_select::zip::zip;
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
+use crate::error::invalid_data;
use crate::metadata_columns::{
RESERVED_COL_NAME_ROW_ID, RESERVED_FIELD_ID_POS, RESERVED_FIELD_ID_ROW_ID,
};
@@ -75,9 +76,9 @@ pub(crate) fn synthesize_row_id_column(
match column_by_field_id(&batch, RESERVED_FIELD_ID_ROW_ID) {
Some(id) => {
if id.data_type() != &DataType::Int64 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("_row_id source must be Int64, got {}", id.data_type()),
+ return Err(invalid_data!(
+ "_row_id source must be Int64, got {}",
+ id.data_type()
));
}
diff --git a/crates/iceberg/src/arrow/record_batch_partition_splitter.rs b/crates/iceberg/src/arrow/record_batch_partition_splitter.rs
index 598bbf09f..6b3aa73d2 100644
--- a/crates/iceberg/src/arrow/record_batch_partition_splitter.rs
+++ b/crates/iceberg/src/arrow/record_batch_partition_splitter.rs
@@ -24,8 +24,9 @@ use arrow_select::filter::filter_record_batch;
use super::arrow_struct_to_literal;
use super::partition_value_calculator::PartitionValueCalculator;
+use crate::Result;
+use crate::error::invalid_data;
use crate::spec::{Literal, PartitionKey, PartitionSpecRef, SchemaRef, StructType};
-use crate::{Error, ErrorKind, Result};
/// Column name for the projected partition values struct
pub const PROJECTED_PARTITION_VALUE_COLUMN: &str = "_partition";
@@ -129,9 +130,8 @@ impl RecordBatchPartitionSplitter {
if let Some(Literal::Struct(s)) = s {
Ok(s)
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "Partition value is not a struct literal or is null",
+ Err(invalid_data!(
+ "Partition value is not a struct literal or is null"
))
}
})
@@ -141,23 +141,15 @@ impl RecordBatchPartitionSplitter {
let partition_column = batch
.column_by_name(PROJECTED_PARTITION_VALUE_COLUMN)
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Partition column '{PROJECTED_PARTITION_VALUE_COLUMN}' not found in batch"
- ),
+ invalid_data!(
+ "Partition column '{PROJECTED_PARTITION_VALUE_COLUMN}' not found in batch"
)
})?;
let partition_struct_array = partition_column
.as_any()
.downcast_ref::<StructArray>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Partition column is not a StructArray",
- )
- })?;
+ .ok_or_else(|| invalid_data!("Partition column is not a StructArray"))?;
let arrow_struct_array = Arc::new(partition_struct_array.clone()) as ArrayRef;
let struct_array = arrow_struct_to_literal(&arrow_struct_array, &self.partition_type)?;
@@ -168,9 +160,8 @@ impl RecordBatchPartitionSplitter {
if let Some(Literal::Struct(s)) = s {
Ok(s)
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "Partition value is not a struct literal or is null",
+ Err(invalid_data!(
+ "Partition value is not a struct literal or is null"
))
}
})
diff --git a/crates/iceberg/src/arrow/record_batch_projector.rs b/crates/iceberg/src/arrow/record_batch_projector.rs
index 7028eee96..d0802ee53 100644
--- a/crates/iceberg/src/arrow/record_batch_projector.rs
+++ b/crates/iceberg/src/arrow/record_batch_projector.rs
@@ -23,7 +23,7 @@ use arrow_schema::{DataType, Field, FieldRef, Fields, Schema, SchemaRef};
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use crate::arrow::schema::schema_to_arrow_schema;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::spec::Schema as IcebergSchema;
use crate::{Error, ErrorKind};
@@ -97,12 +97,9 @@ impl RecordBatchProjector {
let field_id_fetch_func = |field: &Field| -> Result<Option<i64>> {
if let Some(value) = field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
let field_id = value.parse::<i32>().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "Failed to parse field id".to_string(),
- )
- .with_context("value", value)
- .with_source(e)
+ invalid_data!("Failed to parse field id")
+ .with_context("value", value)
+ .with_source(e)
})?;
Ok(Some(field_id as i64))
} else {
@@ -167,7 +164,7 @@ impl RecordBatchProjector {
self.projected_schema.clone(),
self.project_column(batch.columns())?,
)
- .map_err(|err| Error::new(ErrorKind::DataInvalid, format!("{err}")))
+ .map_err(|err| invalid_data!("{err}"))
}
/// Do projection with columns
diff --git a/crates/iceberg/src/arrow/record_batch_transformer.rs b/crates/iceberg/src/arrow/record_batch_transformer.rs
index 800dc6da4..19976a81c 100644
--- a/crates/iceberg/src/arrow/record_batch_transformer.rs
+++ b/crates/iceberg/src/arrow/record_batch_transformer.rs
@@ -36,6 +36,7 @@ use crate::arrow::{
datum_to_arrow_type_with_ree, primitive_type_to_arrow_type_with_ree, schema_to_arrow_schema,
type_to_arrow_type,
};
+use crate::error::invalid_data;
use crate::metadata_columns::{
RESERVED_COL_NAME_PARTITION, RESERVED_FIELD_ID_PARTITION, get_metadata_field,
};
@@ -279,13 +280,10 @@ impl StructConstant {
/// the same length.
pub(crate) fn new(fields: Fields, child_values: Vec<Option<PrimitiveLiteral>>) -> Result<Self> {
if fields.len() != child_values.len() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "StructConstant: fields length ({}) != child_values length ({})",
- fields.len(),
- child_values.len()
- ),
+ return Err(invalid_data!(
+ "StructConstant: fields length ({}) != child_values length ({})",
+ fields.len(),
+ child_values.len()
));
}
Ok(Self {
@@ -863,10 +861,7 @@ impl RecordBatchTransformer {
// Iceberg-Java's Parquet readers (BaseParquetReaders / SparkParquetReaders),
// which raise "Missing required field: <name>".
if iceberg_field.initial_default.is_none() && iceberg_field.required {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Missing required field: {}", iceberg_field.name),
- ));
+ return Err(invalid_data!("Missing required field: {}", iceberg_field.name));
}
let default_value = iceberg_field.initial_default.as_ref().and_then(|lit| {
@@ -895,12 +890,9 @@ impl RecordBatchTransformer {
for (source_field_idx, source_field) in source_schema.fields.iter().enumerate() {
// Check if field has a field ID in metadata
if let Some(field_id_str) = source_field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
- let this_field_id = field_id_str.parse().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("field id not parseable as an i32: {e}"),
- )
- })?;
+ let this_field_id = field_id_str
+ .parse()
+ .map_err(|e| invalid_data!("field id not parseable as an i32: {e}"))?;
field_id_to_source_schema
.insert(this_field_id, (source_field.clone(), source_field_idx));
@@ -998,9 +990,9 @@ impl RecordBatchTransformer {
target_type: &DataType,
) -> Result<ArrayRef> {
if source.data_type() != &DataType::Int64 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("coalesce source must be Int64, got {}", source.data_type()),
+ return Err(invalid_data!(
+ "coalesce source must be Int64, got {}",
+ source.data_type()
));
}
let PrimitiveLiteral::Long(seq) = fallback else {
diff --git a/crates/iceberg/src/arrow/schema.rs b/crates/iceberg/src/arrow/schema.rs
index 4ef7b2943..e88c24d1a 100644
--- a/crates/iceberg/src/arrow/schema.rs
+++ b/crates/iceberg/src/arrow/schema.rs
@@ -34,7 +34,7 @@ use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use parquet::file::statistics::Statistics;
use uuid::Uuid;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::spec::decimal_utils::i128_from_be_bytes;
use crate::spec::{
Datum, FIRST_FIELD_ID, ListType, MapType, NestedField, NestedFieldRef, PrimitiveLiteral,
@@ -206,10 +206,7 @@ fn visit_type<V: ArrowSchemaVisitor>(r#type: &DataType, visitor: &mut V) -> Resu
DataType::Map(field, _) => match field.data_type() {
DataType::Struct(fields) => {
if fields.len() != 2 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Map field must have exactly 2 fields",
- ));
+ return Err(invalid_data!("Map field must have exactly 2 fields"));
}
let key_field = &fields[0];
@@ -231,17 +228,11 @@ fn visit_type<V: ArrowSchemaVisitor>(r#type: &DataType, visitor: &mut V) -> Resu
visitor.map(r#type, key_result, value_result)
}
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- "Map field must have struct type",
- )),
+ _ => Err(invalid_data!("Map field must have struct type")),
},
DataType::Struct(fields) => visit_struct(fields, visitor),
DataType::Dictionary(_key_type, value_type) => visit_type(value_type, visitor),
- other => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Cannot visit Arrow data type: {other}"),
- )),
+ other => Err(invalid_data!("Cannot visit Arrow data type: {other}")),
}
}
@@ -329,18 +320,12 @@ const ARROW_FIELD_DOC_KEY: &str = "doc";
pub(super) fn get_field_id_from_metadata(field: &FieldRef) -> Result<i32> {
if let Some(value) = field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
return value.parse::<i32>().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "Failed to parse field id".to_string(),
- )
- .with_context("value", value)
- .with_source(e)
+ invalid_data!("Failed to parse field id")
+ .with_context("value", value)
+ .with_source(e)
});
}
- Err(Error::new(
- ErrorKind::DataInvalid,
- "Field id not found in metadata",
- ))
+ Err(invalid_data!("Field id not found in metadata"))
}
fn get_field_doc(field: &FieldRef) -> Option<String> {
@@ -439,10 +424,7 @@ impl ArrowSchemaVisitor for ArrowSchemaConverter {
DataType::LargeList(element_field) => element_field,
DataType::FixedSizeList(element_field, _) => element_field,
_ => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "List type must have list data type",
- ));
+ return Err(invalid_data!("List type must have list data type"));
}
};
@@ -462,10 +444,7 @@ impl ArrowSchemaVisitor for ArrowSchemaConverter {
DataType::Map(field, _) => match field.data_type() {
DataType::Struct(fields) => {
if fields.len() != 2 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Map field must have exactly 2 fields",
- ));
+ return Err(invalid_data!("Map field must have exactly 2 fields"));
}
let key_field = &fields[0];
@@ -496,15 +475,9 @@ impl ArrowSchemaVisitor for ArrowSchemaConverter {
value_field,
}))
}
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- "Map field must have struct type",
- )),
+ _ => Err(invalid_data!("Map field must have struct type")),
},
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- "Map type must have map data type",
- )),
+ _ => Err(invalid_data!("Map type must have map data type")),
}
}
@@ -520,20 +493,14 @@ impl ArrowSchemaVisitor for ArrowSchemaConverter {
DataType::Int64 => Ok(Type::Primitive(PrimitiveType::Long)),
DataType::UInt64 => {
// Block uint64 - no safe casting option
- Err(Error::new(
- ErrorKind::DataInvalid,
- "UInt64 is not supported. Use Int64 for values ≤ 9,223,372,036,854,775,807 or Decimal(20,0) for full uint64 range.",
+ Err(invalid_data!(
+ "UInt64 is not supported. Use Int64 for values ≤ 9,223,372,036,854,775,807 or Decimal(20,0) for full uint64 range."
))
}
DataType::Float32 => Ok(Type::Primitive(PrimitiveType::Float)),
DataType::Float64 => Ok(Type::Primitive(PrimitiveType::Double)),
- DataType::Decimal128(p, s) => Type::decimal(*p as u32, *s as u32).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "Failed to create decimal type".to_string(),
- )
- .with_source(e)
- }),
+ DataType::Decimal128(p, s) => Type::decimal(*p as u32, *s as u32)
+ .map_err(|e| invalid_data!("Failed to create decimal type").with_source(e)),
DataType::Date32 => Ok(Type::Primitive(PrimitiveType::Date)),
DataType::Time64(unit) if unit == &TimeUnit::Microsecond => {
Ok(Type::Primitive(PrimitiveType::Time))
@@ -565,10 +532,7 @@ impl ArrowSchemaVisitor for ArrowSchemaConverter {
DataType::Utf8View | DataType::Utf8 | DataType::LargeUtf8 => {
Ok(Type::Primitive(PrimitiveType::String))
}
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Unsupported Arrow data type: {p}"),
- )),
+ _ => Err(invalid_data!("Unsupported Arrow data type: {p}")),
}
}
@@ -576,9 +540,8 @@ impl ArrowSchemaVisitor for ArrowSchemaConverter {
// The extension may only sit on struct storage (mirrors
// `VariantExtensionType::supports_data_type`).
if !matches!(field.data_type(), DataType::Struct(_)) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "arrow.parquet.variant extension requires Struct storage",
+ return Err(invalid_data!(
+ "arrow.parquet.variant extension requires Struct storage"
));
}
// Fold the whole struct into a single logical variant without descending:
@@ -708,28 +671,19 @@ impl SchemaVisitor for ToArrowSchemaConverter {
PrimitiveType::Decimal { precision, scale } => {
let (precision, scale) = {
let precision: u8 = precision.to_owned().try_into().map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- "incompatible precision for decimal type convert",
- )
- .with_source(err)
+ invalid_data!("incompatible precision for decimal type convert")
+ .with_source(err)
})?;
let scale = scale.to_owned().try_into().map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- "incompatible scale for decimal type convert",
- )
- .with_source(err)
+ invalid_data!("incompatible scale for decimal type convert")
+ .with_source(err)
})?;
(precision, scale)
};
validate_decimal_precision_and_scale::<Decimal128Type>(precision, scale).map_err(
|err| {
- Error::new(
- ErrorKind::DataInvalid,
- "incompatible precision and scale for decimal type convert",
- )
- .with_source(err)
+ invalid_data!("incompatible precision and scale for decimal type convert")
+ .with_source(err)
},
)?;
Ok(ArrowSchemaOrFieldOrType::Type(DataType::Decimal128(
@@ -854,7 +808,7 @@ pub(crate) fn get_arrow_datum(datum: &Datum) -> Result<Arc<dyn ArrowDatum + Send
}
(PrimitiveType::Fixed(_), PrimitiveLiteral::Binary(value)) => {
let array = FixedSizeBinaryArray::try_from_iter(std::iter::once(value.as_slice()))
- .map_err(|e| Error::new(ErrorKind::DataInvalid, e.to_string()))?;
+ .map_err(|e| invalid_data!("FixedSizeBinary conversion failed").with_source(e))?;
Ok(Arc::new(Scalar::new(array)))
}
@@ -941,12 +895,10 @@ pub(crate) fn get_parquet_stat_min_as_datum(
};
Some(Datum::new(
primitive_type.clone(),
- PrimitiveLiteral::Int128(i128_from_be_bytes(bytes).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't convert bytes to i128: {bytes:?}"),
- )
- })?),
+ PrimitiveLiteral::Int128(
+ i128_from_be_bytes(bytes)
+ .ok_or_else(|| invalid_data!("Can't convert bytes to i128: {bytes:?}"))?,
+ ),
))
}
(
@@ -1087,12 +1039,10 @@ pub(crate) fn get_parquet_stat_max_as_datum(
};
Some(Datum::new(
primitive_type.clone(),
- PrimitiveLiteral::Int128(i128_from_be_bytes(bytes).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't convert bytes to i128: {bytes:?}"),
- )
- })?),
+ PrimitiveLiteral::Int128(
+ i128_from_be_bytes(bytes)
+ .ok_or_else(|| invalid_data!("Can't convert bytes to i128: {bytes:?}"))?,
+ ),
))
}
(
diff --git a/crates/iceberg/src/arrow/value.rs b/crates/iceberg/src/arrow/value.rs
index ad2a7f647..080d1b99a 100644
--- a/crates/iceberg/src/arrow/value.rs
+++ b/crates/iceberg/src/arrow/value.rs
@@ -28,6 +28,7 @@ use arrow_schema::{DataType, FieldRef, TimeUnit};
use uuid::Uuid;
use super::get_field_id_from_metadata;
+use crate::error::invalid_data;
use crate::spec::{
ListType, Literal, Map, MapType, NestedField, PartnerAccessor, PrimitiveLiteral, PrimitiveType,
SchemaWithPartnerVisitor, Struct, StructType, Type, VariantType, visit_struct_with_partner,
@@ -57,12 +58,9 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
) -> Result<Vec<Option<Literal>>> {
// Make there is no null value if the field is required
if field.required && value.iter().any(Option::is_none) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "The field is required but has null value",
- )
- .with_context("field_id", field.id.to_string())
- .with_context("field_name", &field.name));
+ return Err(invalid_data!("The field is required but has null value")
+ .with_context("field_id", field.id.to_string())
+ .with_context("field_name", &field.name));
}
Ok(value)
}
@@ -75,12 +73,11 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
) -> Result<Vec<Option<Literal>>> {
let row_len = results.first().map(|column| column.len()).unwrap_or(0);
if let Some(col) = results.iter().find(|col| col.len() != row_len) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "The struct columns have different row length",
- )
- .with_context("first col length", row_len.to_string())
- .with_context("actual col length", col.len().to_string()));
+ return Err(
+ invalid_data!("The struct columns have different row length")
+ .with_context("first col length", row_len.to_string())
+ .with_context("actual col length", col.len().to_string()),
+ );
}
let mut struct_literals = Vec::with_capacity(row_len);
@@ -111,19 +108,14 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
elements: Vec<Option<Literal>>,
) -> Result<Vec<Option<Literal>>> {
if list.element_field.required && elements.iter().any(Option::is_none) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "The list should not have null value",
- ));
+ return Err(invalid_data!("The list should not have null value"));
}
match array.data_type() {
DataType::List(_) => {
let offset = array
.as_any()
.downcast_ref::<ListArray>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The partner is not a list array")
- })?
+ .ok_or_else(|| invalid_data!("The partner is not a list array"))?
.offsets();
// combine the result according to the offset
let mut result = Vec::with_capacity(offset.len() - 1);
@@ -138,12 +130,7 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
let offset = array
.as_any()
.downcast_ref::<LargeListArray>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a large list array",
- )
- })?
+ .ok_or_else(|| invalid_data!("The partner is not a large list array"))?
.offsets();
// combine the result according to the offset
let mut result = Vec::with_capacity(offset.len() - 1);
@@ -163,10 +150,7 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
}
Ok(result)
}
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a list type",
- )),
+ _ => Err(invalid_data!("The partner is not a list type")),
}
}
@@ -179,16 +163,15 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
) -> Result<Vec<Option<Literal>>> {
// Make sure key_value and value have the same row length
if key_values.len() != values.len() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "The key value and value of map should have the same row length",
+ return Err(invalid_data!(
+ "The key value and value of map should have the same row length"
));
}
let offsets = partner
.as_any()
.downcast_ref::<MapArray>()
- .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "The partner is not a map array"))?
+ .ok_or_else(|| invalid_data!("The partner is not a map array"))?
.offsets();
// combine the result according to the offset
let mut result = Vec::with_capacity(offsets.len() - 1);
@@ -211,65 +194,47 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
let array = partner
.as_any()
.downcast_ref::<BooleanArray>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The partner is not a boolean array")
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a boolean array"))?;
Ok(array.iter().map(|v| v.map(Literal::bool)).collect())
}
PrimitiveType::Int => {
let array = partner
.as_any()
.downcast_ref::<Int32Array>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The partner is not a int32 array")
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a int32 array"))?;
Ok(array.iter().map(|v| v.map(Literal::int)).collect())
}
PrimitiveType::Long => {
let array = partner
.as_any()
.downcast_ref::<Int64Array>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The partner is not a int64 array")
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a int64 array"))?;
Ok(array.iter().map(|v| v.map(Literal::long)).collect())
}
PrimitiveType::Float => {
let array = partner
.as_any()
.downcast_ref::<Float32Array>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The partner is not a float32 array")
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a float32 array"))?;
Ok(array.iter().map(|v| v.map(Literal::float)).collect())
}
PrimitiveType::Double => {
let array = partner
.as_any()
.downcast_ref::<Float64Array>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The partner is not a float64 array")
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a float64 array"))?;
Ok(array.iter().map(|v| v.map(Literal::double)).collect())
}
PrimitiveType::Decimal { precision, scale } => {
let array = partner
.as_any()
.downcast_ref::<Decimal128Array>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a decimal128 array",
- )
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a decimal128 array"))?;
if let DataType::Decimal128(arrow_precision, arrow_scale) = array.data_type()
&& (*arrow_precision as u32 != *precision || *arrow_scale as u32 != *scale)
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "The precision or scale ({arrow_precision},{arrow_scale}) of arrow decimal128 array is not compatible with iceberg decimal type ({precision},{scale})"
- ),
+ return Err(invalid_data!(
+ "The precision or scale ({arrow_precision},{arrow_scale}) of arrow decimal128 array is not compatible with iceberg decimal type ({precision},{scale})"
));
}
Ok(array.iter().map(|v| v.map(Literal::decimal)).collect())
@@ -278,54 +243,35 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
let array = partner
.as_any()
.downcast_ref::<Date32Array>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The partner is not a date32 array")
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a date32 array"))?;
Ok(array.iter().map(|v| v.map(Literal::date)).collect())
}
PrimitiveType::Time => {
let array = partner
.as_any()
.downcast_ref::<Time64MicrosecondArray>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The partner is not a time64 array")
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a time64 array"))?;
Ok(array.iter().map(|v| v.map(Literal::time)).collect())
}
PrimitiveType::Timestamp => {
let array = partner
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a timestamp array",
- )
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a timestamp array"))?;
Ok(array.iter().map(|v| v.map(Literal::timestamp)).collect())
}
PrimitiveType::Timestamptz => {
let array = partner
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a timestamptz array",
- )
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a timestamptz array"))?;
Ok(array.iter().map(|v| v.map(Literal::timestamptz)).collect())
}
PrimitiveType::TimestampNs => {
let array = partner
.as_any()
.downcast_ref::<TimestampNanosecondArray>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a timestamp_ns array",
- )
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a timestamp_ns array"))?;
Ok(array
.iter()
.map(|v| v.map(Literal::timestamp_nano))
@@ -335,12 +281,7 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
let array = partner
.as_any()
.downcast_ref::<TimestampNanosecondArray>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a timestamptz_ns array",
- )
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a timestamptz_ns array"))?;
Ok(array
.iter()
.map(|v| v.map(Literal::timestamptz_nano))
@@ -352,54 +293,37 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
} else if let Some(array) = partner.as_any().downcast_ref::<StringArray>() {
Ok(array.iter().map(|v| v.map(Literal::string)).collect())
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a string array",
- ))
+ Err(invalid_data!("The partner is not a string array"))
}
}
PrimitiveType::Uuid => {
if let Some(array) = partner.as_any().downcast_ref::<FixedSizeBinaryArray>() {
if array.value_length() != 16 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a uuid array",
- ));
+ return Err(invalid_data!("The partner is not a uuid array"));
}
Ok(array
.iter()
.map(|v| {
v.map(|v| {
Ok(Literal::uuid(Uuid::from_bytes(v.try_into().map_err(
- |_| {
- Error::new(
- ErrorKind::DataInvalid,
- "Failed to convert binary to uuid",
- )
- },
+ |_| invalid_data!("Failed to convert binary to uuid"),
)?)))
})
.transpose()
})
.collect::<Result<Vec<_>>>()?)
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a uuid array",
- ))
+ Err(invalid_data!("The partner is not a uuid array"))
}
}
PrimitiveType::Fixed(len) => {
let array = partner
.as_any()
.downcast_ref::<FixedSizeBinaryArray>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The partner is not a fixed array")
- })?;
+ .ok_or_else(|| invalid_data!("The partner is not a fixed array"))?;
if array.value_length() != *len as i32 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "The length of fixed size binary array is not compatible with iceberg fixed type",
+ return Err(invalid_data!(
+ "The length of fixed size binary array is not compatible with iceberg fixed type"
));
}
Ok(array
@@ -419,10 +343,7 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
.map(|v| v.map(|v| Literal::binary(v.to_vec())))
.collect())
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "The partner is not a binary array",
- ))
+ Err(invalid_data!("The partner is not a binary array"))
}
}
}
@@ -494,10 +415,7 @@ impl Default for ArrowArrayAccessor {
impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
fn struct_partner<'a>(&self, schema_partner: &'a ArrayRef) -> Result<&'a ArrayRef> {
if !matches!(schema_partner.data_type(), DataType::Struct(_)) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "The schema partner is not a struct type",
- ));
+ return Err(invalid_data!("The schema partner is not a struct type"));
}
Ok(schema_partner)
@@ -512,11 +430,8 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
.as_any()
.downcast_ref::<StructArray>()
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "The struct partner is not a struct array, partner: {struct_partner:?}"
- ),
+ invalid_data!(
+ "The struct partner is not a struct array, partner: {struct_partner:?}"
)
})?;
@@ -524,12 +439,7 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
.fields()
.iter()
.position(|arrow_field| self.match_mode.match_field(arrow_field, field))
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Field id {} not found in struct array", field.id),
- )
- })?;
+ .ok_or_else(|| invalid_data!("Field id {} not found in struct array", field.id))?;
Ok(struct_array.column(field_pos))
}
@@ -540,24 +450,14 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
let list_array = list_partner
.as_any()
.downcast_ref::<ListArray>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "The list partner is not a list array",
- )
- })?;
+ .ok_or_else(|| invalid_data!("The list partner is not a list array"))?;
Ok(list_array.values())
}
DataType::LargeList(_) => {
let list_array = list_partner
.as_any()
.downcast_ref::<LargeListArray>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "The list partner is not a large list array",
- )
- })?;
+ .ok_or_else(|| invalid_data!("The list partner is not a large list array"))?;
Ok(list_array.values())
}
DataType::FixedSizeList(_, _) => {
@@ -565,17 +465,11 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
.as_any()
.downcast_ref::<FixedSizeListArray>()
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "The list partner is not a fixed size list array",
- )
+ invalid_data!("The list partner is not a fixed size list array")
})?;
Ok(list_array.values())
}
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- "The list partner is not a list type",
- )),
+ _ => Err(invalid_data!("The list partner is not a list type")),
}
}
@@ -583,9 +477,7 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
let map_array = map_partner
.as_any()
.downcast_ref::<MapArray>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The map partner is not a map array")
- })?;
+ .ok_or_else(|| invalid_data!("The map partner is not a map array"))?;
Ok(map_array.keys())
}
@@ -593,9 +485,7 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
let map_array = map_partner
.as_any()
.downcast_ref::<MapArray>()
- .ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "The map partner is not a map array")
- })?;
+ .ok_or_else(|| invalid_data!("The map partner is not a map array"))?;
Ok(map_array.values())
}
}
@@ -708,12 +598,9 @@ pub(crate) fn create_primitive_array_single_element(
let array = Decimal128Array::from(vec![{ *v }])
.with_precision_and_scale(*precision, *scale)
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ invalid_data!(
"Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
- ),
- )
+ )
})?;
Ok(Arc::new(array))
}
@@ -721,12 +608,9 @@ pub(crate) fn create_primitive_array_single_element(
let array = Decimal128Array::from(vec![*v as i128])
.with_precision_and_scale(*precision, *scale)
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ invalid_data!(
"Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
- ),
- )
+ )
})?;
Ok(Arc::new(array))
}
@@ -734,12 +618,9 @@ pub(crate) fn create_primitive_array_single_element(
let array = Decimal128Array::from(vec![Option::<i128>::None])
.with_precision_and_scale(*precision, *scale)
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ invalid_data!(
"Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
- ),
- )
+ )
})?;
Ok(Arc::new(array))
}
@@ -885,22 +766,16 @@ pub(crate) fn create_primitive_array_repeated(
// wrong-width literal would otherwise produce a FixedSizeBinary of the
// wrong width instead of erroring.
if value.len() != *len as usize {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "FixedSizeBinary literal length {} does not match declared width {len}",
- value.len()
- ),
+ return Err(invalid_data!(
+ "FixedSizeBinary literal length {} does not match declared width {len}",
+ value.len()
));
}
Arc::new(
FixedSizeBinaryArray::try_from_iter(std::iter::repeat_n(value.as_slice(), num_rows))
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to create FixedSizeBinary({len}) array: {e}"),
- )
+ invalid_data!("Failed to create FixedSizeBinary({len}) array: {e}")
})?,
)
}
@@ -912,12 +787,9 @@ pub(crate) fn create_primitive_array_repeated(
Decimal128Array::from(vec![*value; num_rows])
.with_precision_and_scale(*precision, *scale)
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ invalid_data!(
"Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
- ),
- )
+ )
})?,
)
}
@@ -926,12 +798,9 @@ pub(crate) fn create_primitive_array_repeated(
Decimal128Array::from(vec![*value as i128; num_rows])
.with_precision_and_scale(*precision, *scale)
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ invalid_data!(
"Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
- ),
- )
+ )
})?,
)
}
@@ -943,12 +812,9 @@ pub(crate) fn create_primitive_array_repeated(
Decimal128Array::from(vals)
.with_precision_and_scale(*precision, *scale)
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ invalid_data!(
"Failed to create Decimal128Array with precision {precision} and scale {scale}: {e}"
- ),
- )
+ )
})?,
)
}
diff --git a/crates/iceberg/src/avro/schema.rs b/crates/iceberg/src/avro/schema.rs
index a67e4ed71..908dff5c9 100644
--- a/crates/iceberg/src/avro/schema.rs
+++ b/crates/iceberg/src/avro/schema.rs
@@ -26,6 +26,7 @@ use apache_avro::schema::{
use itertools::{Either, Itertools};
use serde_json::{Number, Value};
+use crate::error::invalid_data;
use crate::spec::{
ListType, MapType, NestedField, NestedFieldRef, PrimitiveType, Schema, SchemaVisitor,
StructType, Type, VariantType, visit_schema,
@@ -401,9 +402,8 @@ pub(crate) fn visit<V: AvroSchemaVisitor>(schema: &AvroSchema, visitor: &mut V)
let value = visit(&record_schema.fields[1].schema, visitor)?;
return visitor.map_array(record_schema, key, value);
} else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Can't convert avro map schema, item is not a record.",
+ return Err(invalid_data!(
+ "Can't convert avro map schema, item is not a record."
));
}
} else {
@@ -437,25 +437,16 @@ impl AvroSchemaToSchema {
) -> Result<i32> {
attributes
.get(name)
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Can't convert avro array schema, missing element id.",
- )
- })?
+ .ok_or_else(|| invalid_data!("Can't convert avro array schema, missing element id."))?
.as_i64()
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Can't convert avro array schema, element id is not a valid i64 number.",
+ invalid_data!(
+ "Can't convert avro array schema, element id is not a valid i64 number."
)
})?
.try_into()
.map_err(|_| {
- Error::new(
- ErrorKind::DataInvalid,
- "Can't convert avro array schema, element id is not a valid i32.",
- )
+ invalid_data!("Can't convert avro array schema, element id is not a valid i32.")
})
}
}
@@ -580,12 +571,8 @@ impl AvroSchemaVisitor for AvroSchemaToSchema {
key: Option<Type>,
value: Option<Type>,
) -> Result<Self::T> {
- let key = key.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Can't convert avro map schema, missing key schema.",
- )
- })?;
+ let key =
+ key.ok_or_else(|| invalid_data!("Can't convert avro map schema, missing key schema."))?;
let value = unwrap_or_unknown(value);
let key_id = Self::get_element_id_from_attributes(
&array.fields[0].custom_attributes,
@@ -628,9 +615,8 @@ pub(crate) fn avro_schema_to_schema(avro_schema: &AvroSchema) -> Result<Schema>
))
}
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "Can't convert non record avro schema to iceberg schema: {avro_schema}",
+ Err(invalid_data!(
+ "Can't convert non record avro schema to iceberg schema: {avro_schema}"
))
}
}
@@ -1242,6 +1228,23 @@ mod tests {
assert!(avro_schema_to_schema(&avro_schema).is_err());
}
+ #[test]
+ fn test_non_record_schema_error_interpolates_schema() {
+ let err = avro_schema_to_schema(&AvroSchema::String).unwrap_err();
+
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(
+ !err.message().contains("{avro_schema}"),
+ "message should interpolate the schema, got: {}",
+ err.message()
+ );
+ assert!(
+ err.message().ends_with(&AvroSchema::String.to_string()),
+ "message should end with the rendered schema, got: {}",
+ err.message()
+ );
+ }
+
#[test]
fn test_decimal_type() {
let avro_schema = {
diff --git a/crates/iceberg/src/catalog/memory/catalog.rs b/crates/iceberg/src/catalog/memory/catalog.rs
index aae0f7d5a..dc5079a1a 100644
--- a/crates/iceberg/src/catalog/memory/catalog.rs
+++ b/crates/iceberg/src/catalog/memory/catalog.rs
@@ -28,13 +28,14 @@ use itertools::Itertools;
use super::namespace_state::NamespaceState;
use crate::encryption::kms::{KeyManagementClient, KmsClientFactory};
+use crate::error::invalid_data;
use crate::io::{FileIO, FileIOBuilder, MemoryStorageFactory, StorageFactory};
use crate::runtime::Runtime;
use crate::spec::{TableMetadata, TableMetadataBuilder};
use crate::table::Table;
use crate::{
- Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result,
- TableCommit, TableCreation, TableIdent,
+ Catalog, CatalogBuilder, MetadataLocation, Namespace, NamespaceIdent, Result, TableCommit,
+ TableCreation, TableIdent,
};
/// Memory catalog warehouse location
@@ -79,10 +80,7 @@ impl CatalogBuilder for MemoryCatalogBuilder {
async move {
let catalog_properties = MemoryCatalogProperties::from_properties(&props)?;
if catalog_properties.warehouse.is_empty() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Catalog warehouse is required",
- ));
+ return Err(invalid_data!("Catalog warehouse is required"));
}
let runtime = self.runtime.unwrap_or_else(Runtime::current);
@@ -438,6 +436,7 @@ pub(crate) mod tests {
use tempfile::TempDir;
use super::*;
+ use crate::ErrorKind;
use crate::encryption::kms::MemoryKmsClientFactory;
use crate::io::{FileIO, LocalFsStorageFactory};
use crate::spec::{NestedField, PartitionSpec, PrimitiveType, Schema, SortOrder, Type};
diff --git a/crates/iceberg/src/catalog/memory/namespace_state.rs b/crates/iceberg/src/catalog/memory/namespace_state.rs
index d7dd6c4b2..d8361dd24 100644
--- a/crates/iceberg/src/catalog/memory/namespace_state.rs
+++ b/crates/iceberg/src/catalog/memory/namespace_state.rs
@@ -19,6 +19,7 @@ use std::collections::{HashMap, hash_map};
use itertools::Itertools;
+use crate::error::invalid_data;
use crate::table::Table;
use crate::{Error, ErrorKind, NamespaceIdent, Result, TableIdent};
@@ -111,10 +112,7 @@ impl NamespaceState {
namespace_ident: &NamespaceIdent,
) -> Result<(&mut NamespaceState, String)> {
match namespace_ident.split_last() {
- None => Err(Error::new(
- ErrorKind::DataInvalid,
- "Namespace identifier can't be empty!",
- )),
+ None => Err(invalid_data!("Namespace identifier can't be empty!")),
Some((child_namespace_name, parent_name_parts)) => {
let parent_namespace_state = if parent_name_parts.is_empty() {
Ok(self)
diff --git a/crates/iceberg/src/catalog/mod.rs b/crates/iceberg/src/catalog/mod.rs
index d2ecc7d51..cf7a94d20 100644
--- a/crates/iceberg/src/catalog/mod.rs
+++ b/crates/iceberg/src/catalog/mod.rs
@@ -42,6 +42,7 @@ use typed_builder::TypedBuilder;
use uuid::Uuid;
use crate::encryption::kms::KmsClientFactory;
+use crate::error::invalid_data;
use crate::io::StorageFactory;
use crate::runtime::Runtime;
use crate::spec::{
@@ -209,10 +210,7 @@ impl NamespaceIdent {
/// Create a multi-level namespace identifier from vector.
pub fn from_vec(names: Vec<String>) -> Result<Self> {
if names.is_empty() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Namespace identifier can't be empty!",
- ));
+ return Err(invalid_data!("Namespace identifier can't be empty!"));
}
Ok(Self(names))
}
@@ -322,9 +320,9 @@ impl TableIdent {
/// Try to create table identifier from an iterator of string.
pub fn from_strs(iter: impl IntoIterator<Item = impl ToString>) -> Result<Self> {
let mut vec: Vec<String> = iter.into_iter().map(|s| s.to_string()).collect();
- let table_name = vec.pop().ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "Table identifier can't be empty!")
- })?;
+ let table_name = vec
+ .pop()
+ .ok_or_else(|| invalid_data!("Table identifier can't be empty!"))?;
let namespace_ident = NamespaceIdent::from_vec(vec)?;
Ok(Self {
diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs
index 2e749cab1..692ab9315 100644
--- a/crates/iceberg/src/delete_file_index.rs
+++ b/crates/iceberg/src/delete_file_index.rs
@@ -23,11 +23,12 @@ use futures::StreamExt;
use futures::channel::mpsc::{Sender, channel};
use tokio::sync::Notify;
+use crate::error::invalid_data;
use crate::metadata_columns::RESERVED_FIELD_ID_DELETE_FILE_PATH;
use crate::runtime::Runtime;
use crate::scan::{DeleteFileContext, FileScanTaskDeleteFile};
use crate::spec::{DataContentType, DataFile, DataFileFormat, PrimitiveLiteral, Struct};
-use crate::{Error, ErrorKind, Result};
+use crate::{Error, Result};
/// Index of delete files
#[derive(Debug, Clone)]
@@ -219,24 +220,18 @@ impl PopulatedDeleteFileIndex {
// malformed manifest entry, not an ordinary position delete to fall back
// on.
let Some(path) = data_file.referenced_data_file() else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {} is missing referenced_data_file",
- arc_ctx.manifest_entry.file_path()
- ),
+ return Err(invalid_data!(
+ "deletion vector {} is missing referenced_data_file",
+ arc_ctx.manifest_entry.file_path()
));
};
if data_file.content_offset().is_none()
|| data_file.content_size_in_bytes().is_none()
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {} is missing content_offset or content_size_in_bytes",
- arc_ctx.manifest_entry.file_path()
- ),
+ return Err(invalid_data!(
+ "deletion vector {} is missing content_offset or content_size_in_bytes",
+ arc_ctx.manifest_entry.file_path()
));
}
@@ -244,13 +239,10 @@ impl PopulatedDeleteFileIndex {
dvs_by_referenced_data_file.insert(path.clone(), arc_ctx)
{
let inserted = &dvs_by_referenced_data_file[&path];
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "found multiple deletion vectors for data file {path}: {} and {}",
- existing.manifest_entry.file_path(),
- inserted.manifest_entry.file_path()
- ),
+ return Err(invalid_data!(
+ "found multiple deletion vectors for data file {path}: {} and {}",
+ existing.manifest_entry.file_path(),
+ inserted.manifest_entry.file_path()
));
}
continue;
@@ -339,29 +331,23 @@ impl PopulatedDeleteFileIndex {
if data_file.partition() != dv_data_file.partition()
|| data_file.partition_spec_id != dv.partition_spec_id
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {} references data file {} but its partition (spec {}, {:?}) does not match the data file's partition (spec {}, {:?})",
- dv.manifest_entry.file_path(),
- data_file.file_path(),
- dv.partition_spec_id,
- dv_data_file.partition(),
- data_file.partition_spec_id,
- data_file.partition()
- ),
+ return Err(invalid_data!(
+ "deletion vector {} references data file {} but its partition (spec {}, {:?}) does not match the data file's partition (spec {}, {:?})",
+ dv.manifest_entry.file_path(),
+ data_file.file_path(),
+ dv.partition_spec_id,
+ dv_data_file.partition(),
+ data_file.partition_spec_id,
+ data_file.partition()
));
}
if let Some(seq_num) = seq_num {
let dv_seq = dv.manifest_entry.sequence_number();
if dv_seq < Some(seq_num) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion vector {} has data sequence number {dv_seq:?}, which must be >= the data file's sequence number {seq_num}",
- dv.manifest_entry.file_path()
- ),
+ return Err(invalid_data!(
+ "deletion vector {} has data sequence number {dv_seq:?}, which must be >= the data file's sequence number {seq_num}",
+ dv.manifest_entry.file_path()
));
}
}
@@ -406,6 +392,7 @@ mod tests {
use uuid::Uuid;
use super::*;
+ use crate::ErrorKind;
use crate::spec::{
DataContentType, DataFileBuilder, DataFileFormat, Datum, Literal, ManifestEntry,
ManifestStatus, Struct,
diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs
index 4ab13d52b..9b2d48d91 100644
--- a/crates/iceberg/src/delete_vector.rs
+++ b/crates/iceberg/src/delete_vector.rs
@@ -22,6 +22,7 @@ use roaring::bitmap::Iter;
use roaring::treemap::BitmapIter;
use roaring::{RoaringBitmap, RoaringTreemap};
+use crate::error::invalid_data;
use crate::{Error, ErrorKind, Result};
/// Magic bytes prefixing a serialized `deletion-vector-v1` bitmap, per the Iceberg Puffin spec.
@@ -98,12 +99,9 @@ impl DeleteVector {
/// comparison, or the roaring payload fails to decode.
pub fn deserialize(blob: &[u8]) -> Result<Self> {
if blob.len() < DV_MIN_BLOB_BYTES {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion-vector-v1 blob is {} bytes, shorter than the {DV_MIN_BLOB_BYTES}-byte minimum",
- blob.len()
- ),
+ return Err(invalid_data!(
+ "deletion-vector-v1 blob is {} bytes, shorter than the {DV_MIN_BLOB_BYTES}-byte minimum",
+ blob.len()
));
}
@@ -133,12 +131,9 @@ fn verify_length_prefix(mut prefix: &[u8], body: &[u8]) -> Result<()> {
.with_source(e)
})? as usize;
if declared_len != body.len() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion-vector-v1 length prefix is {declared_len}, expected {}",
- body.len()
- ),
+ return Err(invalid_data!(
+ "deletion-vector-v1 length prefix is {declared_len}, expected {}",
+ body.len()
));
}
Ok(())
@@ -154,11 +149,8 @@ fn verify_crc(body: &[u8], mut crc_bytes: &[u8]) -> Result<()> {
})?;
let computed_crc = crc32fast::hash(body);
if computed_crc != stored_crc {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion-vector-v1 CRC mismatch: computed {computed_crc:#010x}, stored {stored_crc:#010x}"
- ),
+ return Err(invalid_data!(
+ "deletion-vector-v1 CRC mismatch: computed {computed_crc:#010x}, stored {stored_crc:#010x}"
));
}
Ok(())
@@ -166,9 +158,8 @@ fn verify_crc(body: &[u8], mut crc_bytes: &[u8]) -> Result<()> {
fn verify_magic(magic: &[u8]) -> Result<()> {
if magic != DV_MAGIC {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("deletion-vector-v1 magic mismatch: {magic:02x?}, expected {DV_MAGIC:02x?}"),
+ return Err(invalid_data!(
+ "deletion-vector-v1 magic mismatch: {magic:02x?}, expected {DV_MAGIC:02x?}"
));
}
Ok(())
@@ -182,21 +173,14 @@ fn verify_magic(magic: &[u8]) -> Result<()> {
// that doesn't match what was actually written.
fn decode_roaring_directory(mut reader: &[u8]) -> Result<RoaringTreemap> {
let bitmap_count = reader.try_get_u64_le().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "failed to decode deletion-vector-v1 roaring payload",
- )
- .with_source(e)
+ invalid_data!("failed to decode deletion-vector-v1 roaring payload").with_source(e)
})?;
// The roaring portable format restricts the bitmap count to [0, 2^32 - 1] (it is stored as a
// u64 with the upper 32 bits reserved as zero padding).
if bitmap_count > u32::MAX as u64 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion-vector-v1 roaring bitmap count {bitmap_count} exceeds the {}-key maximum",
- u32::MAX
- ),
+ return Err(invalid_data!(
+ "deletion-vector-v1 roaring bitmap count {bitmap_count} exceeds the {}-key maximum",
+ u32::MAX
));
}
@@ -204,30 +188,19 @@ fn decode_roaring_directory(mut reader: &[u8]) -> Result<RoaringTreemap> {
let mut last_key: Option<u32> = None;
for _ in 0..bitmap_count {
let key = reader.try_get_u32_le().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "failed to decode deletion-vector-v1 roaring payload",
- )
- .with_source(e)
+ invalid_data!("failed to decode deletion-vector-v1 roaring payload").with_source(e)
})?;
if let Some(last) = last_key
&& key <= last
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "deletion-vector-v1 roaring keys must be ordered by unsigned comparison, got key {key} after {last}"
- ),
+ return Err(invalid_data!(
+ "deletion-vector-v1 roaring keys must be ordered by unsigned comparison, got key {key} after {last}"
));
}
last_key = Some(key);
let bitmap = RoaringBitmap::deserialize_from(&mut reader).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "failed to decode deletion-vector-v1 roaring payload",
- )
- .with_source(e)
+ invalid_data!("failed to decode deletion-vector-v1 roaring payload").with_source(e)
})?;
bitmaps.push((key, bitmap));
}
diff --git a/crates/iceberg/src/encryption/crypto.rs b/crates/iceberg/src/encryption/crypto.rs
index babdb33bc..27ae9a336 100644
--- a/crates/iceberg/src/encryption/crypto.rs
+++ b/crates/iceberg/src/encryption/crypto.rs
@@ -28,6 +28,7 @@ use aes_gcm::{Aes128Gcm, Aes256Gcm, AesGcm, Nonce};
/// from the underlying primitives, same as `Aes128Gcm` and `Aes256Gcm`.
type Aes192Gcm = AesGcm<aes_gcm::aes::Aes192, U12>;
+use crate::error::invalid_data;
use crate::sensitive::SensitiveBytes;
use crate::{Error, ErrorKind, Result};
@@ -65,9 +66,8 @@ impl AesKeySize {
16 => Ok(Self::Bits128),
24 => Ok(Self::Bits192),
32 => Ok(Self::Bits256),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid data key length: {len} (must be 16, 24, or 32)"),
+ _ => Err(invalid_data!(
+ "Invalid data key length: {len} (must be 16, 24, or 32)"
)),
}
}
@@ -81,10 +81,7 @@ impl FromStr for AesKeySize {
"128" | "AES_GCM_128" | "AES128_GCM" => Ok(Self::Bits128),
"192" | "AES_GCM_192" | "AES192_GCM" => Ok(Self::Bits192),
"256" | "AES_GCM_256" | "AES256_GCM" => Ok(Self::Bits256),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid AES key size: {s}"),
- )),
+ _ => Err(invalid_data!("Invalid AES key size: {s}")),
}
}
}
@@ -195,13 +192,10 @@ impl AesGcmCipher {
/// The decrypted plaintext.
pub fn decrypt(&self, ciphertext: &[u8], aad: Option<&[u8]>) -> Result<Vec<u8>> {
if ciphertext.len() < Self::NONCE_LEN + Self::TAG_LEN {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Ciphertext too short: expected at least {} bytes, got {}",
- Self::NONCE_LEN + Self::TAG_LEN,
- ciphertext.len()
- ),
+ return Err(invalid_data!(
+ "Ciphertext too short: expected at least {} bytes, got {}",
+ Self::NONCE_LEN + Self::TAG_LEN,
+ ciphertext.len()
));
}
@@ -221,9 +215,8 @@ impl AesGcmCipher {
fn encrypt_aes_gcm<C>(key_bytes: &[u8], plaintext: &[u8], aad: Option<&[u8]>) -> Result<Vec<u8>>
where C: Aead + AeadCore + KeyInit {
- let cipher = C::new_from_slice(key_bytes).map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Invalid AES key").with_source(anyhow::anyhow!(e))
- })?;
+ let cipher = C::new_from_slice(key_bytes)
+ .map_err(|e| invalid_data!("Invalid AES key").with_source(anyhow::anyhow!(e)))?;
let nonce = C::generate_nonce(&mut OsRng);
let ciphertext = if let Some(aad) = aad {
@@ -248,9 +241,8 @@ where C: Aead + AeadCore + KeyInit {
fn decrypt_aes_gcm<C>(key_bytes: &[u8], ciphertext: &[u8], aad: Option<&[u8]>) -> Result<Vec<u8>>
where C: Aead + AeadCore + KeyInit {
- let cipher = C::new_from_slice(key_bytes).map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Invalid AES key").with_source(anyhow::anyhow!(e))
- })?;
+ let cipher = C::new_from_slice(key_bytes)
+ .map_err(|e| invalid_data!("Invalid AES key").with_source(anyhow::anyhow!(e)))?;
let nonce = Nonce::from_slice(&ciphertext[..AesGcmCipher::NONCE_LEN]);
let encrypted_data = &ciphertext[AesGcmCipher::NONCE_LEN..];
diff --git a/crates/iceberg/src/encryption/key_metadata.rs b/crates/iceberg/src/encryption/key_metadata.rs
index 765d600b5..4f8445a9a 100644
--- a/crates/iceberg/src/encryption/key_metadata.rs
+++ b/crates/iceberg/src/encryption/key_metadata.rs
@@ -24,6 +24,7 @@ use aes_gcm::aead::OsRng;
use aes_gcm::aead::rand_core::RngCore;
use super::{AesKeySize, SecureKey};
+use crate::error::invalid_data;
use crate::{Error, ErrorKind, Result};
/// Standard key metadata for Iceberg table encryption.
@@ -196,10 +197,7 @@ mod _serde {
pub(super) fn decode(bytes: &[u8]) -> Result<Self> {
if bytes.is_empty() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Empty key metadata buffer",
- ));
+ return Err(invalid_data!("Empty key metadata buffer"));
}
let version = bytes[0];
@@ -211,17 +209,11 @@ mod _serde {
}
let mut reader = Cursor::new(&bytes[1..]);
- let value = from_avro_datum(&AVRO_SCHEMA_V1, &mut reader, None).map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Failed to decode key metadata").with_source(e)
- })?;
+ let value = from_avro_datum(&AVRO_SCHEMA_V1, &mut reader, None)
+ .map_err(|e| invalid_data!("Failed to decode key metadata").with_source(e))?;
- from_value(&value).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "Failed to decode key metadata fields",
- )
- .with_source(e)
- })
+ from_value(&value)
+ .map_err(|e| invalid_data!("Failed to decode key metadata fields").with_source(e))
}
}
@@ -243,11 +235,7 @@ mod _serde {
fn try_from(v1: StandardKeyMetadataV1) -> Result<Self> {
let encryption_key = SecureKey::new(&v1.encryption_key).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "Invalid encryption key in key metadata",
- )
- .with_source(e)
+ invalid_data!("Invalid encryption key in key metadata").with_source(e)
})?;
Ok(Self {
encryption_key,
diff --git a/crates/iceberg/src/encryption/kms/memory.rs b/crates/iceberg/src/encryption/kms/memory.rs
index 4c95b8421..3950b11fb 100644
--- a/crates/iceberg/src/encryption/kms/memory.rs
+++ b/crates/iceberg/src/encryption/kms/memory.rs
@@ -29,7 +29,7 @@ use async_trait::async_trait;
use super::KeyManagementClient;
use super::factory::KmsClientFactory;
use crate::encryption::{AesGcmCipher, AesKeySize, SecureKey, SensitiveBytes};
-use crate::error::lock_error;
+use crate::error::{invalid_data, lock_error};
use crate::{Error, ErrorKind, Result};
/// In-memory KMS for testing. Not suitable for production use.
@@ -108,10 +108,7 @@ impl MemoryKeyManagementClient {
let mut keys = self.master_keys.write().map_err(lock_error)?;
if keys.contains_key(&key_id) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Master key already exists: {key_id}"),
- ));
+ return Err(invalid_data!("Master key already exists: {key_id}"));
}
keys.insert(key_id, key);
@@ -121,12 +118,9 @@ impl MemoryKeyManagementClient {
fn get_master_key(&self, key_id: &str) -> Result<SensitiveBytes> {
let keys = self.master_keys.read().map_err(lock_error)?;
- keys.get(key_id).cloned().ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Master key not found: {key_id}"),
- )
- })
+ keys.get(key_id)
+ .cloned()
+ .ok_or_else(|| invalid_data!("Master key not found: {key_id}"))
}
/// Number of master keys stored.
diff --git a/crates/iceberg/src/encryption/manager.rs b/crates/iceberg/src/encryption/manager.rs
index f54ef464d..d6f4e4af6 100644
--- a/crates/iceberg/src/encryption/manager.rs
+++ b/crates/iceberg/src/encryption/manager.rs
@@ -38,6 +38,7 @@ use super::crypto::{AesGcmCipher, AesKeySize, SecureKey};
use super::io::EncryptedOutputFile;
use super::key_metadata::StandardKeyMetadata;
use super::kms::KeyManagementClient;
+use crate::error::invalid_data;
use crate::io::OutputFile;
use crate::sensitive::SensitiveBytes;
use crate::spec::{EncryptedKey, FormatVersion, TableMetadataRef};
@@ -204,20 +205,12 @@ impl EncryptionManager {
.expect("encryption_keys lock poisoned")
.get(encryption_key_id)
.cloned()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Encryption key '{encryption_key_id}' not found"),
- )
- })?;
+ .ok_or_else(|| invalid_data!("Encryption key '{encryption_key_id}' not found"))?;
let kek_key_id = encrypted_key.encrypted_by_id().ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "EncryptedKey '{}' has no encrypted_by_id",
- encrypted_key.key_id()
- ),
+ invalid_data!(
+ "EncryptedKey '{}' has no encrypted_by_id",
+ encrypted_key.key_id()
)
})?;
@@ -330,12 +323,9 @@ impl EncryptionManager {
return Ok(cached);
}
- let master_key_id = kek.encrypted_by_id().ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("KEK '{}' has no encrypted_by_id", kek.key_id()),
- )
- })?;
+ let master_key_id = kek
+ .encrypted_by_id()
+ .ok_or_else(|| invalid_data!("KEK '{}' has no encrypted_by_id", kek.key_id()))?;
let plaintext = self
.kms_client
@@ -356,12 +346,7 @@ impl EncryptionManager {
.expect("encryption_keys lock poisoned")
.get(kek_key_id)
.cloned()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("KEK not found in encryption keys: {kek_key_id}"),
- )
- })?;
+ .ok_or_else(|| invalid_data!("KEK not found in encryption keys: {kek_key_id}"))?;
// KEK timestamp as AAD prevents timestamp tampering.
let aad = Self::kek_timestamp_aad(&kek)?;
@@ -383,13 +368,10 @@ impl EncryptionManager {
.get(KEK_CREATED_AT_PROPERTY)
.map(|ts| ts.as_bytes())
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "KEK '{}' is missing required '{}' property",
- kek.key_id(),
- KEK_CREATED_AT_PROPERTY
- ),
+ invalid_data!(
+ "KEK '{}' is missing required '{}' property",
+ kek.key_id(),
+ KEK_CREATED_AT_PROPERTY
)
})
}
diff --git a/crates/iceberg/src/encryption/stream.rs b/crates/iceberg/src/encryption/stream.rs
index 7971a3df5..c14905f70 100644
--- a/crates/iceberg/src/encryption/stream.rs
+++ b/crates/iceberg/src/encryption/stream.rs
@@ -49,6 +49,7 @@ use std::sync::Arc;
use bytes::{Bytes, BytesMut};
use super::AesGcmCipher;
+use crate::error::invalid_data;
use crate::io::{FileRead, FileWrite};
use crate::{Error, ErrorKind, Result};
@@ -171,12 +172,9 @@ impl AesGcmFileRead {
};
if num_blocks > u32::MAX as u64 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "AGS1 format supports at most {} blocks (~4 TiB per file), but file requires {num_blocks} blocks",
- u32::MAX
- ),
+ return Err(invalid_data!(
+ "AGS1 format supports at most {} blocks (~4 TiB per file), but file requires {num_blocks} blocks",
+ u32::MAX
));
}
@@ -207,11 +205,8 @@ impl AesGcmFileRead {
/// `AesGcmInputStream.calculatePlaintextLength()`.
pub fn calculate_plaintext_length(encrypted_file_length: u64) -> Result<u64> {
if encrypted_file_length < GCM_STREAM_HEADER_LENGTH as u64 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Encrypted file too short: {encrypted_file_length} bytes (minimum {GCM_STREAM_HEADER_LENGTH})"
- ),
+ return Err(invalid_data!(
+ "Encrypted file too short: {encrypted_file_length} bytes (minimum {GCM_STREAM_HEADER_LENGTH})"
));
}
@@ -229,13 +224,10 @@ impl AesGcmFileRead {
0
} else {
if cipher_bytes_in_last_block < (NONCE_LENGTH + GCM_TAG_LENGTH) as u64 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Truncated encrypted file: last block is {} bytes (minimum {})",
- cipher_bytes_in_last_block,
- NONCE_LENGTH + GCM_TAG_LENGTH
- ),
+ return Err(invalid_data!(
+ "Truncated encrypted file: last block is {} bytes (minimum {})",
+ cipher_bytes_in_last_block,
+ NONCE_LENGTH + GCM_TAG_LENGTH
));
}
cipher_bytes_in_last_block - NONCE_LENGTH as u64 - GCM_TAG_LENGTH as u64
@@ -293,22 +285,19 @@ impl FileRead for AesGcmFileRead {
}
if range.start > range.end {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid read range: start ({}) is greater than end ({})",
- range.start, range.end
- ),
+ return Err(invalid_data!(
+ "Invalid read range: start ({}) is greater than end ({})",
+ range.start,
+ range.end
));
}
if range.end > self.plain_stream_size {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Read range {}..{} exceeds plaintext size {}",
- range.start, range.end, self.plain_stream_size
- ),
+ return Err(invalid_data!(
+ "Read range {}..{} exceeds plaintext size {}",
+ range.start,
+ range.end,
+ self.plain_stream_size
));
}
@@ -442,9 +431,8 @@ impl AesGcmFileWrite {
return Err(e);
}
self.block_index = self.block_index.checked_add(1).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "AGS1 block index overflow: file exceeds the maximum supported size (~4 TiB)",
+ invalid_data!(
+ "AGS1 block index overflow: file exceeds the maximum supported size (~4 TiB)"
)
})?;
Ok(())
@@ -462,9 +450,8 @@ impl AesGcmFileWrite {
return Err(e);
}
self.block_index = self.block_index.checked_add(1).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "AGS1 block index overflow: file exceeds the maximum supported size (~4 TiB)",
+ invalid_data!(
+ "AGS1 block index overflow: file exceeds the maximum supported size (~4 TiB)"
)
})?;
self.buffer.drain(..PLAIN_BLOCK_SIZE as usize);
@@ -600,14 +587,11 @@ mod tests {
let start = range.start as usize;
let end = range.end as usize;
if end > self.0.len() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Range {}..{} out of bounds for {} bytes",
- start,
- end,
- self.0.len()
- ),
+ return Err(invalid_data!(
+ "Range {}..{} out of bounds for {} bytes",
+ start,
+ end,
+ self.0.len()
));
}
Ok(self.0.slice(start..end))
diff --git a/crates/iceberg/src/error.rs b/crates/iceberg/src/error.rs
index 02c3eee8f..2f5c92894 100644
--- a/crates/iceberg/src/error.rs
+++ b/crates/iceberg/src/error.rs
@@ -469,6 +469,47 @@ macro_rules! ensure_data_valid {
};
}
+/// Helper macro to construct an [`ErrorKind::DataInvalid`] error.
+///
+/// This is a shorthand for `Error::new(ErrorKind::DataInvalid, ...)`, the most
+/// common error constructed in this crate. It returns the [`Error`] value (it
+/// does *not* return from the enclosing function), so it composes with `?`,
+/// `.map_err(...)`, `.ok_or_else(...)`, and explicit `return Err(...)`.
+///
+/// The message may be a plain expression or a format string with arguments.
+///
+/// Unlike the public [`ensure_data_valid!`], this macro is deliberately
+/// crate-internal — adding `#[macro_export]` would commit it to the public API.
+///
+/// # Examples
+///
+/// The `use` path below is crate-internal and only resolves inside this crate.
+///
+/// ```ignore
+/// use crate::error::invalid_data;
+///
+/// // As an expression
+/// let err = invalid_data!("unexpected value: {value}");
+///
+/// // With `.ok_or_else`
+/// let field = fields.get(id).ok_or_else(|| invalid_data!("missing field {id}"))?;
+///
+/// // Attaching a source error
+/// let n: i32 = s.parse().map_err(|e| invalid_data!("not an int: {s}").with_source(e))?;
+/// ```
+macro_rules! invalid_data {
+ // Bare literals stay in `format!` so inline captures like `{id}` still interpolate.
+ ($fmt: literal $(, $($arg:tt)*)?) => {
+ $crate::error::Error::new($crate::error::ErrorKind::DataInvalid, format!($fmt $(, $($arg)*)?))
+ };
+ ($msg: expr $(,)?) => {
+ $crate::error::Error::new($crate::error::ErrorKind::DataInvalid, $msg)
+ };
+}
+
+// Crate-internal macro: re-exported so other modules can `use crate::error::invalid_data;`.
+pub(crate) use invalid_data;
+
#[cfg(test)]
mod tests {
use anyhow::anyhow;
@@ -532,6 +573,36 @@ Source: networking error
)
}
+ #[test]
+ fn test_invalid_data_macro() {
+ // Plain literal.
+ let err = invalid_data!("something is wrong");
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert_eq!(err.message(), "something is wrong");
+
+ // Literal with an inline capture.
+ let value = 42;
+ let err = invalid_data!("unexpected value: {value}");
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert_eq!(err.message(), "unexpected value: 42");
+
+ // Literal with trailing arguments.
+ let err = invalid_data!("unexpected value: {}", value);
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert_eq!(err.message(), "unexpected value: 42");
+
+ // Non-literal message expression.
+ let computed = String::from("computed");
+ let err = invalid_data!(computed);
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert_eq!(err.message(), "computed");
+
+ // Composes with builder methods like `.with_source`.
+ let err = invalid_data!("wrapping").with_source(anyhow!("root cause"));
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.source.is_some());
+ }
+
/// Backtrace contains build information, so we just assert the header of error content.
#[test]
fn test_error_debug_with_backtrace() {
diff --git a/crates/iceberg/src/expr/predicate.rs b/crates/iceberg/src/expr/predicate.rs
index 5a7330c8c..f8ea89639 100644
--- a/crates/iceberg/src/expr/predicate.rs
+++ b/crates/iceberg/src/expr/predicate.rs
@@ -27,7 +27,7 @@ use fnv::FnvHashSet;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::expr::visitors::bound_predicate_visitor::visit as visit_bound;
use crate::expr::visitors::predicate_visitor::visit;
use crate::expr::visitors::rewrite_not::RewriteNotVisitor;
@@ -398,12 +398,9 @@ impl Bind for Predicate {
}
&PredicateOperator::IsNan | &PredicateOperator::NotNan => {
if !bound_expr.term.field().field_type.is_floating_type() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Expecting floating point type, but found {}",
- bound_expr.term.field().field_type
- ),
+ return Err(invalid_data!(
+ "Expecting floating point type, but found {}",
+ bound_expr.term.field().field_type
));
}
}
diff --git a/crates/iceberg/src/expr/term.rs b/crates/iceberg/src/expr/term.rs
index ecdb88912..4a5aedf74 100644
--- a/crates/iceberg/src/expr/term.rs
+++ b/crates/iceberg/src/expr/term.rs
@@ -22,12 +22,12 @@ use std::fmt::{Display, Formatter};
use fnv::FnvHashSet;
use serde::{Deserialize, Serialize};
+use crate::error::invalid_data;
use crate::expr::accessor::{StructAccessor, StructAccessorRef};
use crate::expr::{
BinaryExpression, Bind, Predicate, PredicateOperator, SetExpression, UnaryExpression,
};
use crate::spec::{Datum, NestedField, NestedFieldRef, SchemaRef};
-use crate::{Error, ErrorKind};
/// Unbound term before binding to a schema.
pub type Term = Reference;
@@ -316,19 +316,12 @@ impl Bind for Reference {
schema.field_by_name_case_insensitive(&self.name)
};
- let field = field.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Field {} not found in schema", self.name),
- )
- })?;
-
- let accessor = schema.accessor_by_field_id(field.id).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Accessor for Field {} not found", self.name),
- )
- })?;
+ let field =
+ field.ok_or_else(|| invalid_data!("Field {} not found in schema", self.name))?;
+
+ let accessor = schema
+ .accessor_by_field_id(field.id)
+ .ok_or_else(|| invalid_data!("Accessor for Field {} not found", self.name))?;
Ok(BoundReference::new(
self.name.clone(),
diff --git a/crates/iceberg/src/expr/visitors/strict_metrics_evaluator.rs b/crates/iceberg/src/expr/visitors/strict_metrics_evaluator.rs
index 9fe8f216c..6cb9f01c5 100644
--- a/crates/iceberg/src/expr/visitors/strict_metrics_evaluator.rs
+++ b/crates/iceberg/src/expr/visitors/strict_metrics_evaluator.rs
@@ -17,10 +17,11 @@
use fnv::FnvHashSet;
+use crate::Result;
+use crate::error::invalid_data;
use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor, visit};
use crate::expr::{BoundPredicate, BoundReference};
use crate::spec::{DataFile, Datum};
-use crate::{Error, ErrorKind, Result};
#[allow(dead_code)]
const ROWS_MUST_MATCH: Result<bool> = Ok(true);
@@ -159,10 +160,7 @@ impl BoundPredicateVisitor for StrictMetricsEvaluator<'_> {
}
fn not(&mut self, _inner: bool) -> Result<bool> {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "NOT should be rewritten",
- ))
+ Err(invalid_data!("NOT should be rewritten"))
}
fn is_null(&mut self, reference: &BoundReference, _predicate: &BoundPredicate) -> Result<bool> {
diff --git a/crates/iceberg/src/inspect/manifests.rs b/crates/iceberg/src/inspect/manifests.rs
index 9e0f7f224..e26037576 100644
--- a/crates/iceberg/src/inspect/manifests.rs
+++ b/crates/iceberg/src/inspect/manifests.rs
@@ -26,11 +26,12 @@ use arrow_array::types::{Int32Type, Int64Type};
use arrow_schema::{DataType, Field, Fields};
use futures::{StreamExt, stream};
+use crate::Result;
use crate::arrow::schema_to_arrow_schema;
+use crate::error::invalid_data;
use crate::scan::ArrowRecordBatchStream;
use crate::spec::{Datum, FieldSummary, ListType, NestedField, PrimitiveType, StructType, Type};
use crate::table::Table;
-use crate::{Error, ErrorKind, Result};
/// Manifests table.
pub struct ManifestsTable<'a> {
@@ -185,12 +186,10 @@ impl<'a> ManifestsTable<'a> {
.metadata()
.partition_spec_by_id(manifest.partition_spec_id)
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Partition spec {} for manifest {} is not in table metadata",
- manifest.partition_spec_id, manifest.manifest_path
- ),
+ invalid_data!(
+ "Partition spec {} for manifest {} is not in table metadata",
+ manifest.partition_spec_id,
+ manifest.manifest_path
)
})?;
let spec_struct = spec.partition_type(self.table.metadata().current_schema())?;
diff --git a/crates/iceberg/src/io/storage/config/s3.rs b/crates/iceberg/src/io/storage/config/s3.rs
index 664e8637b..ae3bc4ba3 100644
--- a/crates/iceberg/src/io/storage/config/s3.rs
+++ b/crates/iceberg/src/io/storage/config/s3.rs
@@ -24,8 +24,9 @@ use serde::{Deserialize, Serialize};
use typed_builder::TypedBuilder;
use super::StorageConfig;
+use crate::error::invalid_data;
use crate::io::is_truthy;
-use crate::{Error, ErrorKind, Result};
+use crate::{Error, Result};
/// S3 endpoint URL.
pub const S3_ENDPOINT: &str = "s3.endpoint";
@@ -201,11 +202,8 @@ impl TryFrom<&StorageConfig> for S3Config {
cfg.server_side_encryption_customer_key_md5 = props.get(S3_SSE_MD5).cloned();
}
_ => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid {S3_SSE_TYPE}: {sse_type}. Expected one of (custom, kms, s3, none)"
- ),
+ return Err(invalid_data!(
+ "Invalid {S3_SSE_TYPE}: {sse_type}. Expected one of (custom, kms, s3, none)"
));
}
}
diff --git a/crates/iceberg/src/io/storage/local_fs.rs b/crates/iceberg/src/io/storage/local_fs.rs
index 836ea74d4..d5ef0a473 100644
--- a/crates/iceberg/src/io/storage/local_fs.rs
+++ b/crates/iceberg/src/io/storage/local_fs.rs
@@ -33,6 +33,7 @@ use futures::StreamExt;
use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
+use crate::error::invalid_data;
use crate::io::{
FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig,
StorageFactory,
@@ -98,12 +99,8 @@ impl Storage for LocalFsStorage {
async fn metadata(&self, path: &str) -> Result<FileMetadata> {
let path = Self::normalize_path(path);
- let metadata = fs::metadata(&path).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to get metadata for {}: {}", path.display(), e),
- )
- })?;
+ let metadata = fs::metadata(&path)
+ .map_err(|e| invalid_data!("Failed to get metadata for {}: {}", path.display(), e))?;
Ok(FileMetadata {
size: metadata.len(),
})
@@ -111,23 +108,15 @@ impl Storage for LocalFsStorage {
async fn read(&self, path: &str) -> Result<Bytes> {
let path = Self::normalize_path(path);
- let content = fs::read(&path).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to read file {}: {}", path.display(), e),
- )
- })?;
+ let content = fs::read(&path)
+ .map_err(|e| invalid_data!("Failed to read file {}: {}", path.display(), e))?;
Ok(Bytes::from(content))
}
async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> {
let path = Self::normalize_path(path);
- let file = fs::File::open(&path).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to open file {}: {}", path.display(), e),
- )
- })?;
+ let file = fs::File::open(&path)
+ .map_err(|e| invalid_data!("Failed to open file {}: {}", path.display(), e))?;
Ok(Box::new(LocalFsFileRead::new(file)))
}
@@ -242,21 +231,13 @@ impl FileRead for LocalFsFileRead {
)
})?;
- file.seek(SeekFrom::Start(range.start)).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to seek to position {}: {}", range.start, e),
- )
- })?;
+ file.seek(SeekFrom::Start(range.start))
+ .map_err(|e| invalid_data!("Failed to seek to position {}: {}", range.start, e))?;
let len = (range.end - range.start) as usize;
let mut buffer = vec![0u8; len];
- file.read_exact(&mut buffer).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to read {len} bytes: {e}"),
- )
- })?;
+ file.read_exact(&mut buffer)
+ .map_err(|e| invalid_data!("Failed to read {len} bytes: {e}"))?;
Ok(Bytes::from(buffer))
}
@@ -283,7 +264,7 @@ impl FileWrite for LocalFsFileWrite {
let file = self
.file
.as_mut()
- .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "Cannot write to closed file"))?;
+ .ok_or_else(|| invalid_data!("Cannot write to closed file"))?;
file.write_all(&bs).map_err(|e| {
Error::new(
@@ -299,7 +280,7 @@ impl FileWrite for LocalFsFileWrite {
let file = self
.file
.take()
- .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "File already closed"))?;
+ .ok_or_else(|| invalid_data!("File already closed"))?;
file.sync_all()
.map_err(|e| Error::new(ErrorKind::Unexpected, format!("Failed to sync file: {e}")))?;
diff --git a/crates/iceberg/src/io/storage/memory.rs b/crates/iceberg/src/io/storage/memory.rs
index dc9ca10d6..b39c741ea 100644
--- a/crates/iceberg/src/io/storage/memory.rs
+++ b/crates/iceberg/src/io/storage/memory.rs
@@ -32,6 +32,7 @@ use futures::StreamExt;
use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
+use crate::error::invalid_data;
use crate::io::{
FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig,
StorageFactory,
@@ -119,10 +120,7 @@ impl Storage for MemoryStorage {
Some(bytes) => Ok(FileMetadata {
size: bytes.len() as u64,
}),
- None => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("File not found: {path}"),
- )),
+ None => Err(invalid_data!("File not found: {path}")),
}
}
@@ -136,10 +134,7 @@ impl Storage for MemoryStorage {
})?;
match data.get(&normalized) {
Some(bytes) => Ok(bytes.clone()),
- None => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("File not found: {path}"),
- )),
+ None => Err(invalid_data!("File not found: {path}")),
}
}
@@ -153,10 +148,7 @@ impl Storage for MemoryStorage {
})?;
match data.get(&normalized) {
Some(bytes) => Ok(Box::new(MemoryFileRead::new(bytes.clone()))),
- None => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("File not found: {path}"),
- )),
+ None => Err(invalid_data!("File not found: {path}")),
}
}
@@ -272,14 +264,11 @@ impl FileRead for MemoryFileRead {
let end = range.end as usize;
if start > self.data.len() || end > self.data.len() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Range {}..{} is out of bounds for data of length {}",
- start,
- end,
- self.data.len()
- ),
+ return Err(invalid_data!(
+ "Range {}..{} is out of bounds for data of length {}",
+ start,
+ end,
+ self.data.len()
));
}
@@ -316,10 +305,7 @@ impl MemoryFileWrite {
impl FileWrite for MemoryFileWrite {
async fn write(&mut self, bs: Bytes) -> Result<()> {
if self.closed {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot write to closed file",
- ));
+ return Err(invalid_data!("Cannot write to closed file"));
}
self.buffer.extend_from_slice(&bs);
Ok(())
@@ -327,7 +313,7 @@ impl FileWrite for MemoryFileWrite {
async fn close(&mut self) -> Result<()> {
if self.closed {
- return Err(Error::new(ErrorKind::DataInvalid, "File already closed"));
+ return Err(invalid_data!("File already closed"));
}
let mut data = self.data.write().map_err(|e| {
diff --git a/crates/iceberg/src/partitioning.rs b/crates/iceberg/src/partitioning.rs
index ae8a58c6a..ef230ef9e 100644
--- a/crates/iceberg/src/partitioning.rs
+++ b/crates/iceberg/src/partitioning.rs
@@ -20,6 +20,7 @@
use std::cmp::Reverse;
use std::collections::{HashMap, HashSet};
+use crate::error::invalid_data;
use crate::spec::{
NestedField, NestedFieldRef, PartitionField, PartitionSpec, Schema, StructType, Transform, Type,
};
@@ -68,13 +69,10 @@ pub fn compute_unified_partition_type<'a>(
// active_field_ids filter below, otherwise an unknown transform could be
// silently skipped.
if matches!(field.transform, Transform::Unknown) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Partition field '{}' uses an unknown transform whose result type \
+ return Err(invalid_data!(
+ "Partition field '{}' uses an unknown transform whose result type \
cannot be determined",
- field.name
- ),
+ field.name
));
}
@@ -98,13 +96,11 @@ pub fn compute_unified_partition_type<'a>(
// V1 tables do not guarantee field ids are unique across specs, so two
// specs may define the same field id. They must be compatible.
if !equivalent_ignoring_names(field, existing) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Conflicting partition fields for field id {field_id}: \
+ return Err(invalid_data!(
+ "Conflicting partition fields for field id {field_id}: \
'{}' and '{}'",
- field.name, existing.name
- ),
+ field.name,
+ existing.name
));
}
diff --git a/crates/iceberg/src/puffin/metadata.rs b/crates/iceberg/src/puffin/metadata.rs
index 8791cb4d8..e237edf51 100644
--- a/crates/iceberg/src/puffin/metadata.rs
+++ b/crates/iceberg/src/puffin/metadata.rs
@@ -20,9 +20,10 @@ use std::collections::{HashMap, HashSet};
use bytes::Bytes;
use serde::{Deserialize, Serialize};
+use crate::Result;
use crate::compression::CompressionCodec;
+use crate::error::invalid_data;
use crate::io::FileRead;
-use crate::{Error, ErrorKind, Result};
/// Human-readable identification of the application writing the file, along with its version.
/// Example: "Trino version 381"
@@ -119,9 +120,8 @@ impl Flag {
if Flag::FooterPayloadCompressed.matches(byte_idx, bit_idx) {
Ok(Flag::FooterPayloadCompressed)
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Unknown flag byte {byte_idx} and bit {bit_idx} combination"),
+ Err(invalid_data!(
+ "Unknown flag byte {byte_idx} and bit {bit_idx} combination"
))
}
}
@@ -188,13 +188,10 @@ impl FileMetadata {
if bytes == FileMetadata::MAGIC {
Ok(())
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Bad magic value: {:?} should be {:?}",
- bytes,
- FileMetadata::MAGIC
- ),
+ Err(invalid_data!(
+ "Bad magic value: {:?} should be {:?}",
+ bytes,
+ FileMetadata::MAGIC
))
}
}
@@ -223,11 +220,8 @@ impl FileMetadata {
let start = input_file_length
.checked_sub(footer_length)
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Footer length {footer_length} exceeds file length {input_file_length}"
- ),
+ invalid_data!(
+ "Footer length {footer_length} exceeds file length {input_file_length}"
)
})?;
let end = input_file_length;
@@ -243,9 +237,9 @@ impl FileMetadata {
- FileMetadata::FOOTER_STRUCT_FLAGS_LENGTH as usize
+ byte_idx as usize;
- let flag_byte = *footer_bytes.get(byte_offset).ok_or_else(|| {
- Error::new(ErrorKind::DataInvalid, "Index range is out of bounds.")
- })?;
+ let flag_byte = *footer_bytes
+ .get(byte_offset)
+ .ok_or_else(|| invalid_data!("Index range is out of bounds."))?;
for bit_idx in 0..8 {
if ((flag_byte >> bit_idx) & 1) != 0 {
@@ -274,32 +268,26 @@ impl FileMetadata {
FileMetadata::MAGIC_LENGTH as usize + usize::try_from(footer_payload_length)?;
let footer_payload_bytes = footer_bytes
.get(start_offset..end_offset)
- .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "Index range is out of bounds."))?;
+ .ok_or_else(|| invalid_data!("Index range is out of bounds."))?;
let decompressed_footer_payload_bytes =
footer_compression_codec.decompress(footer_payload_bytes.into())?;
- String::from_utf8(decompressed_footer_payload_bytes).map_err(|src| {
- Error::new(ErrorKind::DataInvalid, "Footer is not a valid UTF-8 string")
- .with_source(src)
- })
+ String::from_utf8(decompressed_footer_payload_bytes)
+ .map_err(|src| invalid_data!("Footer is not a valid UTF-8 string").with_source(src))
}
fn from_json_str(string: &str) -> Result<FileMetadata> {
- serde_json::from_str::<FileMetadata>(string).map_err(|src| {
- Error::new(ErrorKind::DataInvalid, "Given string is not valid JSON").with_source(src)
- })
+ serde_json::from_str::<FileMetadata>(string)
+ .map_err(|src| invalid_data!("Given string is not valid JSON").with_source(src))
}
/// Returns the file metadata about a Puffin file
pub(crate) async fn read(file_read: &dyn FileRead, file_length: u64) -> Result<FileMetadata> {
if file_length < FileMetadata::MIN_FILE_LENGTH {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "File length {} is too short to be a Puffin file, expected at least {} bytes",
- file_length,
- FileMetadata::MIN_FILE_LENGTH
- ),
+ return Err(invalid_data!(
+ "File length {} is too short to be a Puffin file, expected at least {} bytes",
+ file_length,
+ FileMetadata::MIN_FILE_LENGTH
));
}
diff --git a/crates/iceberg/src/puffin/mod.rs b/crates/iceberg/src/puffin/mod.rs
index 0e054cac5..998520cb8 100644
--- a/crates/iceberg/src/puffin/mod.rs
+++ b/crates/iceberg/src/puffin/mod.rs
@@ -19,7 +19,8 @@
#![deny(missing_docs)]
-use crate::{Error, ErrorKind, Result};
+use crate::Result;
+use crate::error::invalid_data;
mod blob;
pub use blob::{APACHE_DATASKETCHES_THETA_V1, Blob, DELETION_VECTOR_V1};
@@ -31,15 +32,12 @@ pub use crate::compression::CompressionCodec;
fn validate_puffin_compression(codec: CompressionCodec) -> Result<()> {
match codec {
CompressionCodec::None | CompressionCodec::Lz4 | CompressionCodec::Zstd(_) => Ok(()),
- other => Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Compression codec {} is not supported for Puffin files. Only {}, {}, and {} are supported.",
- other.name(),
- CompressionCodec::None.name(),
- CompressionCodec::Lz4.name(),
- CompressionCodec::zstd_default().name()
- ),
+ other => Err(invalid_data!(
+ "Compression codec {} is not supported for Puffin files. Only {}, {}, and {} are supported.",
+ other.name(),
+ CompressionCodec::None.name(),
+ CompressionCodec::Lz4.name(),
+ CompressionCodec::zstd_default().name()
)),
}
}
diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs
index fd833d9b3..7b39189bf 100644
--- a/crates/iceberg/src/scan/mod.rs
+++ b/crates/iceberg/src/scan/mod.rs
@@ -35,6 +35,7 @@ pub use task::*;
use crate::arrow::ArrowReaderBuilder;
pub use crate::arrow::{ScanMetrics, ScanResult};
use crate::delete_file_index::DeleteFileIndex;
+use crate::error::invalid_data;
use crate::expr::visitors::inclusive_metrics_evaluator::InclusiveMetricsEvaluator;
use crate::expr::{Bind, BoundPredicate, Predicate};
use crate::io::FileIO;
@@ -78,10 +79,7 @@ fn collect_scan_field_ids(
}
let field_id = resolve_field_id(schema, column_name, case_sensitive).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Column {column_name} not found in table. Schema: {schema}"),
- )
+ invalid_data!("Column {column_name} not found in table. Schema: {schema}")
})?;
schema
@@ -289,12 +287,7 @@ impl<'a> TableScanBuilder<'a> {
.table
.metadata()
.snapshot_by_id(snapshot_id)
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Snapshot with id {snapshot_id} not found"),
- )
- })?
+ .ok_or_else(|| invalid_data!("Snapshot with id {snapshot_id} not found"))?
.clone(),
None => {
let Some(current_snapshot_id) = self.table.metadata().current_snapshot() else {
diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs
index 531f40556..96a322e3d 100644
--- a/crates/iceberg/src/scan/task.rs
+++ b/crates/iceberg/src/scan/task.rs
@@ -21,12 +21,13 @@ use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
use typed_builder::TypedBuilder;
+use crate::Result;
+use crate::error::invalid_data;
use crate::expr::BoundPredicate;
use crate::spec::{
DataContentType, DataFileFormat, ManifestEntryRef, NameMapping, PartitionSpec, Schema,
SchemaRef, SortOrderRef, Struct, StructType,
};
-use crate::{Error, ErrorKind, Result};
/// A stream of [`FileScanTask`].
pub type FileScanTaskStream = BoxStream<'static, Result<FileScanTask>>;
@@ -261,25 +262,20 @@ impl FileScanTask {
match (self.partition.as_ref(), self.partition_spec.as_deref()) {
(None, None) => Ok(()),
(None, Some(partition_spec)) if partition_spec.is_unpartitioned() => Ok(()),
- (None, Some(_)) => Err(Error::new(
- ErrorKind::DataInvalid,
- "FileScanTask with a partitioned spec requires partition values",
+ (None, Some(_)) => Err(invalid_data!(
+ "FileScanTask with a partitioned spec requires partition values"
)),
(Some(partition), None) if partition.fields().is_empty() => Ok(()),
- (Some(_), None) => Err(Error::new(
- ErrorKind::DataInvalid,
- "Non-empty FileScanTask partition requires a partition spec",
+ (Some(_), None) => Err(invalid_data!(
+ "Non-empty FileScanTask partition requires a partition spec"
)),
(Some(partition), Some(partition_spec))
if partition.fields().len() != partition_spec.fields().len() =>
{
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "FileScanTask partition has {} fields but partition spec has {} fields",
- partition.fields().len(),
- partition_spec.fields().len()
- ),
+ Err(invalid_data!(
+ "FileScanTask partition has {} fields but partition spec has {} fields",
+ partition.fields().len(),
+ partition_spec.fields().len()
))
}
(Some(_), Some(partition_spec)) => {
@@ -400,12 +396,13 @@ mod _serde {
use serde::{Deserialize, Serialize};
use super::{FileScanTask, FileScanTaskDeleteFile};
+ use crate::error::invalid_data;
use crate::expr::BoundPredicate;
use crate::spec::{
DataFileFormat, Literal, NameMapping, PartitionSpec, RawLiteral, SchemaRef, SortOrderRef,
StructType, Type,
};
- use crate::{Error, ErrorKind, Result};
+ use crate::{Error, Result};
#[derive(Deserialize)]
pub(super) struct FileScanTaskSerde {
@@ -543,10 +540,7 @@ mod _serde {
partition_type(value.partition_spec.as_deref(), &value.schema)?;
match partition.try_into(&partition_type)? {
Some(Literal::Struct(partition)) => Ok(partition),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- "FileScanTask partition must be a struct",
- )),
+ _ => Err(invalid_data!("FileScanTask partition must be a struct")),
}
})
.transpose()?;
diff --git a/crates/iceberg/src/spec/manifest/_serde.rs b/crates/iceberg/src/spec/manifest/_serde.rs
index 5bc0bf937..336c883f1 100644
--- a/crates/iceberg/src/spec/manifest/_serde.rs
+++ b/crates/iceberg/src/spec/manifest/_serde.rs
@@ -21,8 +21,9 @@ use serde_derive::{Deserialize, Serialize};
use serde_with::serde_as;
use super::{Datum, ManifestEntry, Schema, Struct};
+use crate::error::invalid_data;
use crate::spec::{FormatVersion, Literal, RawLiteral, Type};
-use crate::{Error, ErrorKind, metadata_columns};
+use crate::{Error, metadata_columns};
#[derive(Serialize, Deserialize)]
pub(super) struct ManifestEntryV2 {
@@ -173,10 +174,7 @@ impl DataFileSerde {
if let Literal::Struct(v) = v {
Ok(v)
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "partition value is not a struct",
- ))
+ Err(invalid_data!("partition value is not a struct"))
}
})
.transpose()?
@@ -251,12 +249,7 @@ fn parse_bytes_entry(v: Vec<BytesEntry>, schema: &Schema) -> Result<HashMap<i32,
let data_type = field
.field_type
.as_primitive_type()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("field {} is not a primitive type", field.name),
- )
- })?
+ .ok_or_else(|| invalid_data!("field {} is not a primitive type", field.name))?
.clone();
m.insert(entry.key, Datum::try_from_bytes(&entry.value, data_type)?);
}
diff --git a/crates/iceberg/src/spec/manifest/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs
index 325e4de81..1fcdaa3a8 100644
--- a/crates/iceberg/src/spec/manifest/data_file.rs
+++ b/crates/iceberg/src/spec/manifest/data_file.rs
@@ -27,9 +27,9 @@ use super::_serde::DataFileSerde;
use super::{
Datum, FormatVersion, Schema, data_file_schema_v1, data_file_schema_v2, data_file_schema_v3,
};
-use crate::error::Result;
+use crate::Error;
+use crate::error::{Result, invalid_data};
use crate::spec::{DEFAULT_PARTITION_SPEC_ID, Struct, StructType, Type};
-use crate::{Error, ErrorKind};
/// Data file carries data file path, partition tuple, metrics, …
#[derive(Debug, PartialEq, Clone, Eq, Builder)]
@@ -370,10 +370,7 @@ impl TryFrom<i32> for DataContentType {
0 => Ok(DataContentType::Data),
1 => Ok(DataContentType::PositionDeletes),
2 => Ok(DataContentType::EqualityDeletes),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("data content type {v} is invalid"),
- )),
+ _ => Err(invalid_data!("data content type {v} is invalid")),
}
}
}
@@ -400,10 +397,7 @@ impl FromStr for DataFileFormat {
"orc" => Ok(Self::Orc),
"parquet" => Ok(Self::Parquet),
"puffin" => Ok(Self::Puffin),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Unsupported data file format: {s}"),
- )),
+ _ => Err(invalid_data!("Unsupported data file format: {s}")),
}
}
}
diff --git a/crates/iceberg/src/spec/manifest/entry.rs b/crates/iceberg/src/spec/manifest/entry.rs
index e8fe0f223..555d86a25 100644
--- a/crates/iceberg/src/spec/manifest/entry.rs
+++ b/crates/iceberg/src/spec/manifest/entry.rs
@@ -21,13 +21,13 @@ use apache_avro::Schema as AvroSchema;
use once_cell::sync::Lazy;
use typed_builder::TypedBuilder;
+use crate::Error;
use crate::avro::schema_to_avro_schema;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::spec::{
DataContentType, DataFile, INITIAL_SEQUENCE_NUMBER, ListType, Literal, ManifestFile, MapType,
NestedField, NestedFieldRef, PrimitiveLiteral, PrimitiveType, Schema, StructType, Type,
};
-use crate::{Error, ErrorKind};
/// Reference to [`ManifestEntry`].
pub type ManifestEntryRef = Arc<ManifestEntry>;
@@ -170,10 +170,7 @@ impl TryFrom<i32> for ManifestStatus {
0 => Ok(ManifestStatus::Existing),
1 => Ok(ManifestStatus::Added),
2 => Ok(ManifestStatus::Deleted),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("manifest status {v} is invalid"),
- )),
+ _ => Err(invalid_data!("manifest status {v} is invalid")),
}
}
}
diff --git a/crates/iceberg/src/spec/manifest/metadata.rs b/crates/iceberg/src/spec/manifest/metadata.rs
index 25e4ae7e0..b06122215 100644
--- a/crates/iceberg/src/spec/manifest/metadata.rs
+++ b/crates/iceberg/src/spec/manifest/metadata.rs
@@ -21,9 +21,8 @@ use std::sync::Arc;
use typed_builder::TypedBuilder;
use super::{FormatVersion, ManifestContentType, PartitionSpec, Schema};
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::spec::{PartitionField, SchemaId, SchemaRef};
-use crate::{Error, ErrorKind};
/// Meta data of a manifest that is stored in the key-value metadata of the Avro file
#[derive(Debug, PartialEq, Clone, Eq, TypedBuilder)]
@@ -46,28 +45,17 @@ impl ManifestMetadata {
pub fn parse(meta: &HashMap<String, Vec<u8>>) -> Result<Self> {
let schema = Arc::new({
let bs = meta.get("schema").ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "schema is required in manifest metadata but not found",
- )
+ invalid_data!("schema is required in manifest metadata but not found")
})?;
serde_json::from_slice::<Schema>(bs).map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- "Fail to parse schema in manifest metadata",
- )
- .with_source(err)
+ invalid_data!("Fail to parse schema in manifest metadata").with_source(err)
})?
});
let schema_id: i32 = meta
.get("schema-id")
.map(|bs| {
String::from_utf8_lossy(bs).parse().map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- "Fail to parse schema id in manifest metadata",
- )
- .with_source(err)
+ invalid_data!("Fail to parse schema id in manifest metadata").with_source(err)
})
})
.transpose()?
@@ -75,28 +63,19 @@ impl ManifestMetadata {
let partition_spec = {
let fields = {
let bs = meta.get("partition-spec").ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "partition-spec is required in manifest metadata but not found",
- )
+ invalid_data!("partition-spec is required in manifest metadata but not found")
})?;
serde_json::from_slice::<Vec<PartitionField>>(bs).map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- "Fail to parse partition spec in manifest metadata",
- )
- .with_source(err)
+ invalid_data!("Fail to parse partition spec in manifest metadata")
+ .with_source(err)
})?
};
let spec_id = meta
.get("partition-spec-id")
.map(|bs| {
String::from_utf8_lossy(bs).parse().map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- "Fail to parse partition spec id in manifest metadata",
- )
- .with_source(err)
+ invalid_data!("Fail to parse partition spec id in manifest metadata")
+ .with_source(err)
})
})
.transpose()?
@@ -108,11 +87,7 @@ impl ManifestMetadata {
};
let format_version = if let Some(bs) = meta.get("format-version") {
serde_json::from_slice::<FormatVersion>(bs).map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- "Fail to parse format version in manifest metadata",
- )
- .with_source(err)
+ invalid_data!("Fail to parse format version in manifest metadata").with_source(err)
})?
} else {
FormatVersion::V1
diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs
index f05eeae77..3bd4dc8ad 100644
--- a/crates/iceberg/src/spec/manifest/mod.rs
+++ b/crates/iceberg/src/spec/manifest/mod.rs
@@ -35,8 +35,7 @@ use super::{
Datum, FormatVersion, ManifestContentType, PartitionSpec, PrimitiveType, Schema, Struct, Type,
UNASSIGNED_SEQUENCE_NUMBER,
};
-use crate::error::Result;
-use crate::{Error, ErrorKind};
+use crate::error::{Result, invalid_data};
/// A manifest contains metadata and a list of entries.
#[derive(Debug, PartialEq, Eq, Clone)]
@@ -135,13 +134,8 @@ pub fn serialize_data_file_to_json(
) -> Result<String> {
let partition_struct_type = Type::Struct(partition_type.clone());
let serde = _serde::DataFileSerde::try_from(data_file, &partition_struct_type, format_version)?;
- serde_json::to_string(&serde).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "Failed to serialize DataFile to JSON!".to_string(),
- )
- .with_source(e)
- })
+ serde_json::to_string(&serde)
+ .map_err(|e| invalid_data!("Failed to serialize DataFile to JSON!").with_source(e))
}
/// Deserialize a DataFile from a JSON string.
@@ -151,13 +145,8 @@ pub fn deserialize_data_file_from_json(
partition_type: &super::StructType,
schema: &Schema,
) -> Result<DataFile> {
- let serde = serde_json::from_str::<_serde::DataFileSerde>(json).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "Failed to deserialize JSON to DataFile!".to_string(),
- )
- .with_source(e)
- })?;
+ let serde = serde_json::from_str::<_serde::DataFileSerde>(json)
+ .map_err(|e| invalid_data!("Failed to deserialize JSON to DataFile!").with_source(e))?;
let partition_struct_type = Type::Struct(partition_type.clone());
serde.try_into(partition_spec_id, &partition_struct_type, schema)
diff --git a/crates/iceberg/src/spec/manifest/reader.rs b/crates/iceberg/src/spec/manifest/reader.rs
index 499ff8fc2..e7044a56b 100644
--- a/crates/iceberg/src/spec/manifest/reader.rs
+++ b/crates/iceberg/src/spec/manifest/reader.rs
@@ -17,10 +17,9 @@
use super::Manifest;
use crate::encryption::{EncryptedInputFile, StandardKeyMetadata};
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::io::FileIO;
use crate::spec::{ManifestContentType, ManifestEntry, ManifestFile};
-use crate::{Error, ErrorKind};
/// Reads a manifest file referenced by a manifest list entry, transparently
/// decrypting it when the entry records key metadata.
@@ -103,10 +102,7 @@ impl ManifestReader {
};
let mut next_row_id = i64::try_from(manifest_first_row_id).map_err(|_| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid first_row_id: {manifest_first_row_id} (exceeds i64::MAX)"),
- )
+ invalid_data!("Invalid first_row_id: {manifest_first_row_id} (exceeds i64::MAX)")
})?;
for entry in entries {
@@ -119,13 +115,8 @@ impl ManifestReader {
entry.data_file.first_row_id = Some(file_first_row_id);
let record_count = entry.data_file.record_count;
next_row_id = file_first_row_id.checked_add_unsigned(record_count).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Row ID overflow assigning first_row_id in {}. File first_row_id: {file_first_row_id}, record count: {record_count}",
- manifest_file.manifest_path
- ),
- )
+ invalid_data!("Row ID overflow assigning first_row_id in {}. File first_row_id: {file_first_row_id}, record count: {record_count}",
+ manifest_file.manifest_path)
})?;
}
}
@@ -140,6 +131,7 @@ mod tests {
use std::sync::Arc;
use super::*;
+ use crate::ErrorKind;
use crate::encryption::{EncryptedOutputFile, StandardKeyMetadata};
use crate::io::FileIO;
use crate::spec::{
diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs
index 725669295..b80567a68 100644
--- a/crates/iceberg/src/spec/manifest/writer.rs
+++ b/crates/iceberg/src/spec/manifest/writer.rs
@@ -29,7 +29,7 @@ use super::{
UNASSIGNED_SEQUENCE_NUMBER,
};
use crate::encryption::EncryptedOutputFile;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::io::{FileWrite, OutputFile};
use crate::spec::manifest::_serde::{ManifestEntryV1, ManifestEntryV2};
use crate::spec::manifest::{manifest_schema_v1, manifest_schema_v2};
@@ -37,7 +37,6 @@ use crate::spec::{
DataContentType, DataFile, FieldSummary, ManifestEntry, ManifestFile, ManifestMetadata,
ManifestStatus, PrimitiveLiteral, SchemaRef, StructType, Type,
};
-use crate::{Error, ErrorKind};
/// Placeholder for snapshot ID. The field with this value must be replaced
/// with the actual snapshot ID before it is committed.
@@ -268,13 +267,10 @@ impl ManifestWriter {
match self.metadata.content {
ManifestContentType::Data => {
if data_file.content != DataContentType::Data {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Date file at path {} with manifest content type `data`, should have DataContentType `Data`, but has `{:?}`",
- data_file.file_path(),
- data_file.content
- ),
+ return Err(invalid_data!(
+ "Date file at path {} with manifest content type `data`, should have DataContentType `Data`, but has `{:?}`",
+ data_file.file_path(),
+ data_file.content
));
}
}
@@ -282,13 +278,10 @@ impl ManifestWriter {
if data_file.content != DataContentType::EqualityDeletes
&& data_file.content != DataContentType::PositionDeletes
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Date file at path {} with manifest content type `deletes`, should have DataContentType `Data`, but has `{:?}`",
- data_file.file_path(),
- data_file.content
- ),
+ return Err(invalid_data!(
+ "Date file at path {} with manifest content type `deletes`, should have DataContentType `Data`, but has `{:?}`",
+ data_file.file_path(),
+ data_file.content
));
}
}
@@ -408,9 +401,8 @@ impl ManifestWriter {
if (entry.status == ManifestStatus::Deleted || entry.status == ManifestStatus::Existing)
&& (entry.sequence_number.is_none() || entry.file_sequence_number.is_none())
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Manifest entry with status Existing or Deleted should have sequence number",
+ return Err(invalid_data!(
+ "Manifest entry with status Existing or Deleted should have sequence number"
));
}
@@ -457,10 +449,8 @@ impl ManifestWriter {
let mut avro_writer = AvroWriter::new(&avro_schema, Vec::new());
avro_writer.add_user_metadata(
"schema".to_string(),
- to_vec(table_schema).map_err(|err| {
- Error::new(ErrorKind::DataInvalid, "Fail to serialize table schema")
- .with_source(err)
- })?,
+ to_vec(table_schema)
+ .map_err(|err| invalid_data!("Fail to serialize table schema").with_source(err))?,
)?;
avro_writer.add_user_metadata(
"schema-id".to_string(),
@@ -469,8 +459,7 @@ impl ManifestWriter {
avro_writer.add_user_metadata(
"partition-spec".to_string(),
to_vec(&self.metadata.partition_spec.fields()).map_err(|err| {
- Error::new(ErrorKind::DataInvalid, "Fail to serialize partition spec")
- .with_source(err)
+ invalid_data!("Fail to serialize partition spec").with_source(err)
})?,
)?;
avro_writer.add_user_metadata(
@@ -562,10 +551,7 @@ impl PartitionFieldStats {
return Ok(());
};
if !self.partition_type.compatible(&value) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "value is not compatible with type",
- ));
+ return Err(invalid_data!("value is not compatible with type"));
}
let value = Datum::new(self.partition_type.clone(), value);
diff --git a/crates/iceberg/src/spec/manifest_list/_serde.rs b/crates/iceberg/src/spec/manifest_list/_serde.rs
index 2a10c39ea..3528dec7c 100644
--- a/crates/iceberg/src/spec/manifest_list/_serde.rs
+++ b/crates/iceberg/src/spec/manifest_list/_serde.rs
@@ -24,7 +24,7 @@ use serde_derive::{Deserialize, Serialize};
use super::ManifestFile;
use crate::Error;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::spec::FieldSummary;
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
@@ -319,56 +319,32 @@ impl TryFrom<ManifestFile> for ManifestFileV3 {
added_files_count: value
.added_files_count
.ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "added_data_files_count in ManifestFileV3 is required",
- )
+ invalid_data!("added_data_files_count in ManifestFileV3 is required")
})?
.try_into()?,
existing_files_count: value
.existing_files_count
.ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "existing_data_files_count in ManifestFileV3 is required",
- )
+ invalid_data!("existing_data_files_count in ManifestFileV3 is required")
})?
.try_into()?,
deleted_files_count: value
.deleted_files_count
.ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "deleted_data_files_count in ManifestFileV3 is required",
- )
+ invalid_data!("deleted_data_files_count in ManifestFileV3 is required")
})?
.try_into()?,
added_rows_count: value
.added_rows_count
- .ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "added_rows_count in ManifestFileV3 is required",
- )
- })?
+ .ok_or_else(|| invalid_data!("added_rows_count in ManifestFileV3 is required"))?
.try_into()?,
existing_rows_count: value
.existing_rows_count
- .ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "existing_rows_count in ManifestFileV3 is required",
- )
- })?
+ .ok_or_else(|| invalid_data!("existing_rows_count in ManifestFileV3 is required"))?
.try_into()?,
deleted_rows_count: value
.deleted_rows_count
- .ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "deleted_rows_count in ManifestFileV3 is required",
- )
- })?
+ .ok_or_else(|| invalid_data!("deleted_rows_count in ManifestFileV3 is required"))?
.try_into()?,
partitions: value.partitions,
key_metadata,
@@ -393,55 +369,37 @@ impl TryFrom<ManifestFile> for ManifestFileV2 {
added_files_count: value
.added_files_count
.ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "added_data_files_count in ManifestFileV2 should be require",
- )
+ invalid_data!("added_data_files_count in ManifestFileV2 should be require")
})?
.try_into()?,
existing_files_count: value
.existing_files_count
.ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "existing_data_files_count in ManifestFileV2 should be require",
- )
+ invalid_data!("existing_data_files_count in ManifestFileV2 should be require")
})?
.try_into()?,
deleted_files_count: value
.deleted_files_count
.ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "deleted_data_files_count in ManifestFileV2 should be require",
- )
+ invalid_data!("deleted_data_files_count in ManifestFileV2 should be require")
})?
.try_into()?,
added_rows_count: value
.added_rows_count
.ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "added_rows_count in ManifestFileV2 should be require",
- )
+ invalid_data!("added_rows_count in ManifestFileV2 should be require")
})?
.try_into()?,
existing_rows_count: value
.existing_rows_count
.ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "existing_rows_count in ManifestFileV2 should be require",
- )
+ invalid_data!("existing_rows_count in ManifestFileV2 should be require")
})?
.try_into()?,
deleted_rows_count: value
.deleted_rows_count
.ok_or_else(|| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "deleted_rows_count in ManifestFileV2 should be require",
- )
+ invalid_data!("deleted_rows_count in ManifestFileV2 should be require")
})?
.try_into()?,
partitions: value.partitions,
diff --git a/crates/iceberg/src/spec/manifest_list/manifest_file.rs b/crates/iceberg/src/spec/manifest_list/manifest_file.rs
index 567e78e3a..341748a41 100644
--- a/crates/iceberg/src/spec/manifest_list/manifest_file.rs
+++ b/crates/iceberg/src/spec/manifest_list/manifest_file.rs
@@ -20,8 +20,8 @@ use std::str::FromStr;
use serde_derive::{Deserialize, Serialize};
use super::ByteBuf;
-use crate::error::Result;
-use crate::{Error, ErrorKind};
+use crate::Error;
+use crate::error::{Result, invalid_data};
/// Entry in a manifest list.
#[derive(Debug, PartialEq, Clone, Eq, Hash)]
@@ -139,10 +139,7 @@ impl FromStr for ManifestContentType {
match s {
"data" => Ok(ManifestContentType::Data),
"deletes" => Ok(ManifestContentType::Deletes),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid manifest content type: {s}"),
- )),
+ _ => Err(invalid_data!("Invalid manifest content type: {s}")),
}
}
}
@@ -163,9 +160,8 @@ impl TryFrom<i32> for ManifestContentType {
match value {
0 => Ok(ManifestContentType::Data),
1 => Ok(ManifestContentType::Deletes),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid manifest content type. Expected 0 or 1, got {value}"),
+ _ => Err(invalid_data!(
+ "Invalid manifest content type. Expected 0 or 1, got {value}"
)),
}
}
diff --git a/crates/iceberg/src/spec/manifest_list/writer.rs b/crates/iceberg/src/spec/manifest_list/writer.rs
index 710471780..24613770d 100644
--- a/crates/iceberg/src/spec/manifest_list/writer.rs
+++ b/crates/iceberg/src/spec/manifest_list/writer.rs
@@ -25,7 +25,7 @@ use super::_const_schema::{
};
use super::_serde::{ManifestFileV1, ManifestFileV2, ManifestFileV3};
use super::{FormatVersion, ManifestContentType, ManifestFile, UNASSIGNED_SEQUENCE_NUMBER};
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::io::FileWrite;
use crate::{Error, ErrorKind};
@@ -208,12 +208,9 @@ impl ManifestListWriter {
fn assign_sequence_numbers(&self, manifest: &mut ManifestFile) -> Result<()> {
if manifest.sequence_number == UNASSIGNED_SEQUENCE_NUMBER {
if manifest.added_snapshot_id != self.snapshot_id {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Found unassigned sequence number for a manifest from snapshot {}.",
- manifest.added_snapshot_id
- ),
+ return Err(invalid_data!(
+ "Found unassigned sequence number for a manifest from snapshot {}.",
+ manifest.added_snapshot_id
));
}
manifest.sequence_number = self.sequence_number;
@@ -221,12 +218,9 @@ impl ManifestListWriter {
if manifest.min_sequence_number == UNASSIGNED_SEQUENCE_NUMBER {
if manifest.added_snapshot_id != self.snapshot_id {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Found unassigned sequence number for a manifest from snapshot {}.",
- manifest.added_snapshot_id
- ),
+ return Err(invalid_data!(
+ "Found unassigned sequence number for a manifest from snapshot {}.",
+ manifest.added_snapshot_id
));
}
manifest.min_sequence_number = self.sequence_number;
@@ -265,13 +259,10 @@ impl ManifestListWriter {
.checked_add(existing_rows_count)
.and_then(|sum| sum.checked_add(added_rows_count))
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ invalid_data!(
"Row ID overflow when computing next row ID for Manifest {}. Next Row ID: {writer_next_row_id}, Existing Rows Count: {existing_rows_count}, Added Rows Count: {added_rows_count}",
manifest.manifest_path
- ),
- )
+ )
}).map(Some)?;
}
(None, None) => {
@@ -291,22 +282,16 @@ impl ManifestListWriter {
fn require_row_counts_in_manifest(manifest: &ManifestFile) -> Result<(u64, u64)> {
let existing_rows_count = manifest.existing_rows_count.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ invalid_data!(
"Cannot include a Manifest without existing-rows-count to a table with row lineage enabled. Manifest path: {}",
manifest.manifest_path,
- ),
- )
+ )
})?;
let added_rows_count = manifest.added_rows_count.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ invalid_data!(
"Cannot include a Manifest without added-rows-count to a table with row lineage enabled. Manifest path: {}",
manifest.manifest_path,
- ),
- )
+ )
})?;
Ok((existing_rows_count, added_rows_count))
}
diff --git a/crates/iceberg/src/spec/name_mapping/mod.rs b/crates/iceberg/src/spec/name_mapping/mod.rs
index a87e829ad..95b8642fa 100644
--- a/crates/iceberg/src/spec/name_mapping/mod.rs
+++ b/crates/iceberg/src/spec/name_mapping/mod.rs
@@ -23,7 +23,8 @@ use std::sync::Arc;
use serde::{Deserialize, Serialize};
use serde_with::{DefaultOnNull, serde_as};
-use crate::{Error, ErrorKind, Result};
+use crate::error::invalid_data;
+use crate::{Error, Result};
/// Iceberg fallback field name to ID mapping.
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
@@ -50,11 +51,7 @@ impl FromStr for NameMapping {
/// Parses a [`NameMapping`] from its JSON representation.
fn from_str(value: &str) -> Result<Self> {
serde_json::from_str(value).map_err(|error| {
- Error::new(
- ErrorKind::DataInvalid,
- "Failed to parse value as a NameMapping",
- )
- .with_source(error)
+ invalid_data!("Failed to parse value as a NameMapping").with_source(error)
})
}
}
diff --git a/crates/iceberg/src/spec/partition.rs b/crates/iceberg/src/spec/partition.rs
index c79eaae63..3999c0d01 100644
--- a/crates/iceberg/src/spec/partition.rs
+++ b/crates/iceberg/src/spec/partition.rs
@@ -26,8 +26,9 @@ use typed_builder::TypedBuilder;
use super::transform::Transform;
use super::{NestedField, PrimitiveType, Schema, SchemaRef, StructType};
+use crate::Result;
+use crate::error::invalid_data;
use crate::spec::Struct;
-use crate::{Error, ErrorKind, Result};
pub(crate) const UNPARTITIONED_LAST_ASSIGNED_ID: i32 = 999;
pub(crate) const DEFAULT_PARTITION_SPEC_ID: i32 = 0;
@@ -487,12 +488,9 @@ impl PartitionSpecBuilder {
.schema
.field_by_name(source_name.as_ref())
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot find source column with name: {} in schema",
- source_name.as_ref()
- ),
+ invalid_data!(
+ "Cannot find source column with name: {} in schema",
+ source_name.as_ref()
)
})?
.id;
@@ -558,12 +556,8 @@ impl PartitionSpecBuilder {
.collect::<std::collections::HashSet<_>>();
fn _check_add_1(prev: i32) -> Result<i32> {
- prev.checked_add(1).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Cannot assign more partition ids. Overflow.",
- )
- })
+ prev.checked_add(1)
+ .ok_or_else(|| invalid_data!("Cannot assign more partition ids. Overflow."))
}
let mut bound_fields = Vec::with_capacity(fields.len());
@@ -604,21 +598,17 @@ impl PartitionSpecBuilder {
if schema_collision.id == field.source_id {
Ok(())
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot create identity partition sourced from different field in schema. Field name '{}' has id `{}` in schema but partition source id is `{}`",
- field.name, schema_collision.id, field.source_id
- ),
+ Err(invalid_data!(
+ "Cannot create identity partition sourced from different field in schema. Field name '{}' has id `{}` in schema but partition source id is `{}`",
+ field.name,
+ schema_collision.id,
+ field.source_id
))
}
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot create partition with name: '{}' that conflicts with schema field and is not an identity transform.",
- field.name
- ),
+ Err(invalid_data!(
+ "Cannot create partition with name: '{}' that conflicts with schema field and is not an identity transform.",
+ field.name
))
}
}
@@ -630,23 +620,17 @@ impl PartitionSpecBuilder {
/// in the schema. Implicitly also checks if the source field exists in the schema.
fn check_transform_compatibility(field: &UnboundPartitionField, schema: &Schema) -> Result<()> {
let schema_field = schema.field_by_id(field.source_id).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot find partition source field with id `{}` in schema",
- field.source_id
- ),
+ invalid_data!(
+ "Cannot find partition source field with id `{}` in schema",
+ field.source_id
)
})?;
if field.transform != Transform::Void {
if !schema_field.field_type.is_primitive() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot partition by non-primitive source field: '{}'.",
- schema_field.field_type
- ),
+ return Err(invalid_data!(
+ "Cannot partition by non-primitive source field: '{}'.",
+ schema_field.field_type
));
}
@@ -655,13 +639,10 @@ impl PartitionSpecBuilder {
.result_type(&schema_field.field_type)
.is_err()
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid source type: '{}' for transform: '{}'.",
- schema_field.field_type,
- field.transform.dedup_name()
- ),
+ return Err(invalid_data!(
+ "Invalid source type: '{}' for transform: '{}'.",
+ schema_field.field_type,
+ field.transform.dedup_name()
));
}
}
@@ -675,16 +656,12 @@ trait CorePartitionSpecValidator {
/// Ensure that the partition name is unique among the partition fields and is not empty.
fn check_name_set_and_unique(&self, name: &str) -> Result<()> {
if name.is_empty() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot use empty partition name",
- ));
+ return Err(invalid_data!("Cannot use empty partition name"));
}
if self.fields().iter().any(|f| f.name == name) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Cannot use partition name more than once: {name}"),
+ return Err(invalid_data!(
+ "Cannot use partition name more than once: {name}"
));
}
Ok(())
@@ -697,14 +674,11 @@ trait CorePartitionSpecValidator {
});
if let Some(collision) = collision {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot add redundant partition with source id `{}` and transform `{}`. A partition with the same source id and transform already exists with name `{}`",
- source_id,
- transform.dedup_name(),
- collision.name
- ),
+ Err(invalid_data!(
+ "Cannot add redundant partition with source id `{}` and transform `{}`. A partition with the same source id and transform already exists with name `{}`",
+ source_id,
+ transform.dedup_name(),
+ collision.name
))
} else {
Ok(())
@@ -714,9 +688,8 @@ trait CorePartitionSpecValidator {
/// Check field / partition_id unique within the partition spec if set
fn check_partition_id_unique(&self, field_id: i32) -> Result<()> {
if self.fields().iter().any(|f| f.field_id == Some(field_id)) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Cannot use field id more than once in one PartitionSpec: {field_id}"),
+ return Err(invalid_data!(
+ "Cannot use field id more than once in one PartitionSpec: {field_id}"
));
}
@@ -741,6 +714,7 @@ impl CorePartitionSpecValidator for UnboundPartitionSpecBuilder {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::ErrorKind;
use crate::spec::{Literal, PrimitiveType, Type};
#[test]
diff --git a/crates/iceberg/src/spec/schema/id_reassigner.rs b/crates/iceberg/src/spec/schema/id_reassigner.rs
index d817ccbd1..0c9b4de3f 100644
--- a/crates/iceberg/src/spec/schema/id_reassigner.rs
+++ b/crates/iceberg/src/spec/schema/id_reassigner.rs
@@ -17,6 +17,7 @@
use super::utils::try_insert_field;
use super::*;
+use crate::error::invalid_data;
pub struct ReassignFieldIds {
next_field_id: i32,
@@ -107,12 +108,10 @@ impl ReassignFieldIds {
}
fn increase_next_field_id(&mut self) -> Result<()> {
- self.next_field_id = self.next_field_id.checked_add(1).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Field ID overflowed, cannot add more fields",
- )
- })?;
+ self.next_field_id = self
+ .next_field_id
+ .checked_add(1)
+ .ok_or_else(|| invalid_data!("Field ID overflowed, cannot add more fields"))?;
Ok(())
}
@@ -120,12 +119,10 @@ impl ReassignFieldIds {
field_ids
.into_iter()
.map(|id| {
- self.old_to_new_id.get(&id).copied().ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Identifier Field ID {id} not found"),
- )
- })
+ self.old_to_new_id
+ .get(&id)
+ .copied()
+ .ok_or_else(|| invalid_data!("Identifier Field ID {id} not found"))
})
.collect()
}
@@ -140,12 +137,7 @@ impl ReassignFieldIds {
self.old_to_new_id
.get(&id)
.copied()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Field with id {id} for alias {name} not found"),
- )
- })
+ .ok_or_else(|| invalid_data!("Field with id {id} for alias {name} not found"))
.map(|new_id| (name, new_id))
})
.collect()
diff --git a/crates/iceberg/src/spec/schema/index.rs b/crates/iceberg/src/spec/schema/index.rs
index e4358e9ef..eb2f990a3 100644
--- a/crates/iceberg/src/spec/schema/index.rs
+++ b/crates/iceberg/src/spec/schema/index.rs
@@ -17,6 +17,7 @@
use super::utils::try_insert_field;
use super::*;
+use crate::error::invalid_data;
use crate::spec::VariantType;
/// Creates a field id to field map.
@@ -183,11 +184,8 @@ impl IndexByName {
.chain(vec![name])
.join(".");
if let Some(existing_field_id) = self.name_to_id.get(full_name.as_str()) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid schema: multiple fields for name {full_name}: {field_id} and {existing_field_id}"
- ),
+ return Err(invalid_data!(
+ "Invalid schema: multiple fields for name {full_name}: {field_id} and {existing_field_id}"
));
} else {
self.name_to_id.insert(full_name, field_id);
diff --git a/crates/iceberg/src/spec/schema/mod.rs b/crates/iceberg/src/spec/schema/mod.rs
index a0197b82c..c6dcf0707 100644
--- a/crates/iceberg/src/spec/schema/mod.rs
+++ b/crates/iceberg/src/spec/schema/mod.rs
@@ -37,7 +37,7 @@ use self::id_reassigner::ReassignFieldIds;
use self::index::{IndexByName, index_by_id, index_parents};
pub use self::prune_columns::prune_columns;
use super::NestedField;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::expr::accessor::StructAccessor;
use crate::spec::FormatVersion;
use crate::spec::datatypes::{
@@ -258,11 +258,8 @@ impl SchemaBuilder {
let id_to_parent = index_parents(r#struct)?;
for identifier_field_id in identifier_field_ids {
let field = id_to_field.get(&identifier_field_id).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot add identifier field {identifier_field_id}: field does not exist"
- ),
+ invalid_data!(
+ "Cannot add identifier field {identifier_field_id}: field does not exist"
)
})?;
ensure_data_valid!(
@@ -277,12 +274,9 @@ impl SchemaBuilder {
field.name
);
} else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot add field {} as an identifier field: not a primitive type field",
- field.name
- ),
+ return Err(invalid_data!(
+ "Cannot add field {} as an identifier field: not a primitive type field",
+ field.name
));
}
@@ -504,9 +498,8 @@ impl Schema {
.sorted_by_key(|(id, _)| *id)
.map(|(_, msg)| msg)
.join("\n- ");
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid schema for {format_version}:\n- {message}"),
+ Err(invalid_data!(
+ "Invalid schema for {format_version}:\n- {message}"
))
}
}
diff --git a/crates/iceberg/src/spec/schema/prune_columns.rs b/crates/iceberg/src/spec/schema/prune_columns.rs
index e80d86221..edfbb5fa1 100644
--- a/crates/iceberg/src/spec/schema/prune_columns.rs
+++ b/crates/iceberg/src/spec/schema/prune_columns.rs
@@ -16,6 +16,7 @@
// under the License.
use super::*;
+use crate::error::invalid_data;
use crate::spec::VariantType;
struct PruneColumn {
@@ -117,10 +118,8 @@ impl SchemaVisitor for PruneColumn {
} else if !field.field_type.is_nested() {
Ok(Some(*field.field_type.clone()))
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
+ Err(invalid_data!(
"Can't project list or map field directly when not selecting full type."
- .to_string(),
)
.with_context("field_id", field.id.to_string())
.with_context("field_type", field.field_type.to_string()))
@@ -182,12 +181,10 @@ impl SchemaVisitor for PruneColumn {
} else if list.element_field.field_type.is_primitive() {
Ok(Some(Type::List(list.clone())))
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot explicitly project List or Map types, List element {} of type {} was selected",
- list.element_field.id, list.element_field.field_type
- ),
+ Err(invalid_data!(
+ "Cannot explicitly project List or Map types, List element {} of type {} was selected",
+ list.element_field.id,
+ list.element_field.field_type
))
}
} else if let Some(result) = value {
@@ -216,12 +213,10 @@ impl SchemaVisitor for PruneColumn {
} else if map.value_field.field_type.is_primitive() {
Ok(Some(Type::Map(map.clone())))
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot explicitly project List or Map types, Map value {} of type {} was selected",
- map.value_field.id, map.value_field.field_type
- ),
+ Err(invalid_data!(
+ "Cannot explicitly project List or Map types, Map value {} of type {} was selected",
+ map.value_field.id,
+ map.value_field.field_type
))
}
} else if let Some(value_result) = value {
diff --git a/crates/iceberg/src/spec/schema/utils.rs b/crates/iceberg/src/spec/schema/utils.rs
index 34ee39b09..26f49b23d 100644
--- a/crates/iceberg/src/spec/schema/utils.rs
+++ b/crates/iceberg/src/spec/schema/utils.rs
@@ -17,15 +17,15 @@
use std::collections::HashMap;
-use crate::{Error, ErrorKind, Result};
+use crate::Result;
+use crate::error::invalid_data;
pub fn try_insert_field<V>(map: &mut HashMap<i32, V>, field_id: i32, value: V) -> Result<()> {
map.insert(field_id, value).map_or_else(
|| Ok(()),
|_| {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Found duplicate 'field.id' {field_id}. Field ids must be unique."),
+ Err(invalid_data!(
+ "Found duplicate 'field.id' {field_id}. Field ids must be unique."
))
},
)
diff --git a/crates/iceberg/src/spec/snapshot.rs b/crates/iceberg/src/spec/snapshot.rs
index 9b0e11cf1..bbd53a0c3 100644
--- a/crates/iceberg/src/spec/snapshot.rs
+++ b/crates/iceberg/src/spec/snapshot.rs
@@ -25,9 +25,8 @@ use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use typed_builder::TypedBuilder;
-use crate::error::{Result, timestamp_ms_to_utc};
+use crate::error::{Result, invalid_data, timestamp_ms_to_utc};
use crate::spec::{SchemaId, SchemaRef, TableMetadata};
-use crate::{Error, ErrorKind};
/// The ref name of the main branch of the table.
pub const MAIN_BRANCH: &str = "main";
@@ -172,12 +171,7 @@ impl Snapshot {
Ok(match self.schema_id() {
Some(schema_id) => table_metadata
.schema_by_id(schema_id)
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Schema with id {schema_id} not found"),
- )
- })?
+ .ok_or_else(|| invalid_data!("Schema with id {schema_id} not found"))?
.clone(),
None => table_metadata.current_schema().clone(),
})
@@ -234,9 +228,10 @@ pub(super) mod _serde {
use serde::{Deserialize, Serialize};
use super::{Operation, Snapshot, Summary};
+ use crate::Error;
+ use crate::error::invalid_data;
use crate::spec::SchemaId;
use crate::spec::snapshot::SnapshotRowRange;
- use crate::{Error, ErrorKind};
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
@@ -381,15 +376,13 @@ pub(super) mod _serde {
manifest_list: match (v1.manifest_list, v1.manifests) {
(Some(file), None) => file,
(Some(_), Some(_)) => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Invalid v1 snapshot, when manifest list provided, manifest files should be omitted",
+ return Err(invalid_data!(
+ "Invalid v1 snapshot, when manifest list provided, manifest files should be omitted"
));
}
(None, _) => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Unsupported v1 snapshot, only manifest list is supported",
+ return Err(invalid_data!(
+ "Unsupported v1 snapshot, only manifest list is supported"
));
}
},
diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs
index c9fd02272..a48a5abdd 100644
--- a/crates/iceberg/src/spec/snapshot_summary.rs
+++ b/crates/iceberg/src/spec/snapshot_summary.rs
@@ -20,6 +20,7 @@ use std::collections::HashMap;
use itertools::Itertools;
use super::{DataContentType, DataFile, PartitionSpecRef};
+use crate::error::invalid_data;
use crate::spec::{ManifestContentType, ManifestFile, Operation, SchemaRef, Summary};
use crate::{Error, ErrorKind, Result};
@@ -339,10 +340,7 @@ pub(crate) fn update_snapshot_summaries(
&& summary.operation != Operation::Overwrite
&& summary.operation != Operation::Delete
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Operation is not supported.",
- ));
+ return Err(invalid_data!("Operation is not supported."));
}
let mut summary = match previous_summary {
diff --git a/crates/iceberg/src/spec/table_metadata.rs b/crates/iceberg/src/spec/table_metadata.rs
index 84bb97a32..24f10355b 100644
--- a/crates/iceberg/src/spec/table_metadata.rs
+++ b/crates/iceberg/src/spec/table_metadata.rs
@@ -39,7 +39,7 @@ use super::{
};
use crate::catalog::{METADATA_FOLDER_NAME, MetadataLocation};
use crate::compression::CompressionCodec;
-use crate::error::{Result, timestamp_ms_to_utc};
+use crate::error::{Result, invalid_data, timestamp_ms_to_utc};
use crate::io::FileIO;
use crate::partitioning::compute_unified_partition_type;
use crate::spec::EncryptedKey;
@@ -487,12 +487,9 @@ impl TableMetadata {
let decompressed_data = CompressionCodec::gzip_default()
.decompress(metadata_content.to_vec())
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- "Trying to read compressed metadata file",
- )
- .with_context("file_path", metadata_location)
- .with_source(e)
+ invalid_data!("Trying to read compressed metadata file")
+ .with_context("file_path", metadata_location)
+ .with_source(e)
})?;
serde_json::from_slice(&decompressed_data)?
} else {
@@ -514,13 +511,10 @@ impl TableMetadata {
let codec = self.table_properties().metadata_compression_codec()?;
if codec != metadata_location.compression_codec() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Compression codec mismatch: metadata_location has {:?}, but table properties specify {:?}",
- metadata_location.compression_codec(),
- codec
- ),
+ return Err(invalid_data!(
+ "Compression codec mismatch: metadata_location has {:?}, but table properties specify {:?}",
+ metadata_location.compression_codec(),
+ codec
));
}
@@ -529,9 +523,8 @@ impl TableMetadata {
CompressionCodec::Gzip(_) => codec.compress(json_data)?,
CompressionCodec::None => json_data,
_ => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Unsupported metadata compression codec: {codec:?}"),
+ return Err(invalid_data!(
+ "Unsupported metadata compression codec: {codec:?}"
));
}
};
@@ -573,14 +566,11 @@ impl TableMetadata {
if field.transform != Transform::Void
&& self.current_schema().field_by_id(field.source_id).is_none()
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Default partition spec {} references missing source field {} in current schema {}",
- self.default_spec.spec_id(),
- field.source_id,
- self.current_schema_id
- ),
+ return Err(invalid_data!(
+ "Default partition spec {} references missing source field {} in current schema {}",
+ self.default_spec.spec_id(),
+ field.source_id,
+ self.current_schema_id
));
}
}
@@ -618,12 +608,9 @@ impl TableMetadata {
}
if self.default_sort_order_id != SortOrder::UNSORTED_ORDER_ID {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "No sort order exists with the default sort order id {}.",
- self.default_sort_order_id
- ),
+ return Err(invalid_data!(
+ "No sort order exists with the default sort order id {}.",
+ self.default_sort_order_id
));
}
@@ -636,12 +623,9 @@ impl TableMetadata {
/// Validate the current schema is set and exists.
fn validate_current_schema(&self) -> Result<()> {
if self.schema_by_id(self.current_schema_id).is_none() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "No schema exists with the current schema id {}.",
- self.current_schema_id
- ),
+ return Err(invalid_data!(
+ "No schema exists with the current schema id {}.",
+ self.current_schema_id
));
}
Ok(())
@@ -653,11 +637,8 @@ impl TableMetadata {
if current_snapshot_id == EMPTY_SNAPSHOT_ID {
self.current_snapshot_id = None;
} else if self.snapshot_by_id(current_snapshot_id).is_none() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Snapshot for current snapshot id {current_snapshot_id} does not exist in the existing snapshots list"
- ),
+ return Err(invalid_data!(
+ "Snapshot for current snapshot id {current_snapshot_id} does not exist in the existing snapshots list"
));
}
}
@@ -668,11 +649,8 @@ impl TableMetadata {
fn validate_refs(&self) -> Result<()> {
for (name, snapshot_ref) in self.refs.iter() {
if self.snapshot_by_id(snapshot_ref.snapshot_id).is_none() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Snapshot for reference {name} does not exist in the existing snapshots list"
- ),
+ return Err(invalid_data!(
+ "Snapshot for reference {name} does not exist in the existing snapshots list"
));
}
}
@@ -682,19 +660,15 @@ impl TableMetadata {
if let Some(main_ref) = main_ref
&& main_ref.snapshot_id != self.current_snapshot_id.unwrap_or_default()
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Current snapshot id does not match main branch ({:?} != {:?})",
- self.current_snapshot_id.unwrap_or_default(),
- main_ref.snapshot_id
- ),
+ return Err(invalid_data!(
+ "Current snapshot id does not match main branch ({:?} != {:?})",
+ self.current_snapshot_id.unwrap_or_default(),
+ main_ref.snapshot_id
));
}
} else if main_ref.is_some() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Current snapshot is not set, but main branch exists",
+ return Err(invalid_data!(
+ "Current snapshot is not set, but main branch exists"
));
}
@@ -704,12 +678,9 @@ impl TableMetadata {
/// Validate that for V1 Metadata the last_sequence_number is 0
fn validate_snapshot_sequence_number(&self) -> Result<()> {
if self.format_version < FormatVersion::V2 && self.last_sequence_number != 0 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Last sequence number must be 0 in v1. Found {}",
- self.last_sequence_number
- ),
+ return Err(invalid_data!(
+ "Last sequence number must be 0 in v1. Found {}",
+ self.last_sequence_number
));
}
@@ -719,14 +690,11 @@ impl TableMetadata {
.values()
.find(|snapshot| snapshot.sequence_number() > self.last_sequence_number)
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid snapshot with id {} and sequence number {} greater than last sequence number {}",
- snapshot.snapshot_id(),
- snapshot.sequence_number(),
- self.last_sequence_number
- ),
+ return Err(invalid_data!(
+ "Invalid snapshot with id {} and sequence number {} greater than last sequence number {}",
+ snapshot.snapshot_id(),
+ snapshot.sequence_number(),
+ self.last_sequence_number
));
}
@@ -740,10 +708,7 @@ impl TableMetadata {
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if curr.timestamp_ms - prev.timestamp_ms < -ONE_MINUTE_MS {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Expected sorted snapshot log entries",
- ));
+ return Err(invalid_data!("Expected sorted snapshot log entries"));
}
}
@@ -751,12 +716,10 @@ impl TableMetadata {
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if self.last_updated_ms - last.timestamp_ms < -ONE_MINUTE_MS {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid update timestamp {}: before last snapshot log entry at {}",
- self.last_updated_ms, last.timestamp_ms
- ),
+ return Err(invalid_data!(
+ "Invalid update timestamp {}: before last snapshot log entry at {}",
+ self.last_updated_ms,
+ last.timestamp_ms
));
}
}
@@ -769,10 +732,7 @@ impl TableMetadata {
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if curr.timestamp_ms - prev.timestamp_ms < -ONE_MINUTE_MS {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Expected sorted metadata log entries",
- ));
+ return Err(invalid_data!("Expected sorted metadata log entries"));
}
}
@@ -780,12 +740,10 @@ impl TableMetadata {
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if self.last_updated_ms - last.timestamp_ms < -ONE_MINUTE_MS {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid update timestamp {}: before last metadata log entry at {}",
- self.last_updated_ms, last.timestamp_ms
- ),
+ return Err(invalid_data!(
+ "Invalid update timestamp {}: before last metadata log entry at {}",
+ self.last_updated_ms,
+ last.timestamp_ms
));
}
}
@@ -821,6 +779,7 @@ pub(super) mod _serde {
DEFAULT_PARTITION_SPEC_ID, EMPTY_SNAPSHOT_ID, FormatVersion, MAIN_BRANCH, MetadataLog,
SnapshotLog, TableMetadata,
};
+ use crate::error::invalid_data;
use crate::spec::schema::_serde::{SchemaV1, SchemaV2};
use crate::spec::snapshot::_serde::{SnapshotV1, SnapshotV2, SnapshotV3};
use crate::spec::{
@@ -1018,12 +977,9 @@ pub(super) mod _serde {
let current_schema: &SchemaRef =
schemas.get(&value.current_schema_id).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "No schema exists with the current schema id {}.",
- value.current_schema_id
- ),
+ invalid_data!(
+ "No schema exists with the current schema id {}.",
+ value.current_schema_id
)
})?;
let partition_specs = HashMap::from_iter(
@@ -1040,12 +996,7 @@ pub(super) mod _serde {
(DEFAULT_PARTITION_SPEC_ID == default_spec_id)
.then(PartitionSpec::unpartition_spec)
})
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Default partition spec {default_spec_id} not found"),
- )
- })?
+ .ok_or_else(|| invalid_data!("Default partition spec {default_spec_id} not found"))?
.into();
let default_partition_type = default_spec.partition_type(current_schema)?;
@@ -1131,12 +1082,9 @@ pub(super) mod _serde {
let current_schema: &SchemaRef =
schemas.get(&value.current_schema_id).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "No schema exists with the current schema id {}.",
- value.current_schema_id
- ),
+ invalid_data!(
+ "No schema exists with the current schema id {}.",
+ value.current_schema_id
)
})?;
let partition_specs = HashMap::from_iter(
@@ -1153,12 +1101,7 @@ pub(super) mod _serde {
(DEFAULT_PARTITION_SPEC_ID == default_spec_id)
.then(PartitionSpec::unpartition_spec)
})
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Default partition spec {default_spec_id} not found"),
- )
- })?
+ .ok_or_else(|| invalid_data!("Default partition spec {default_spec_id} not found"))?
.into();
let default_partition_type = default_spec.partition_type(current_schema)?;
@@ -1248,9 +1191,8 @@ pub(super) mod _serde {
let schema = schema_map
.get(&schema_id)
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("No schema exists with the current schema id {schema_id}."),
+ invalid_data!(
+ "No schema exists with the current schema id {schema_id}."
)
})?
.clone();
@@ -1264,9 +1206,8 @@ pub(super) mod _serde {
(schema_map, schema_id, schema_arc)
} else {
// Option 3: No valid schema configuration found
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "No valid schema configuration found in table metadata",
+ return Err(invalid_data!(
+ "No valid schema configuration found in table metadata"
));
};
@@ -1303,12 +1244,7 @@ pub(super) mod _serde {
let default_spec: PartitionSpecRef = partition_specs
.get(&default_spec_id)
.map(|x| Arc::unwrap_or_clone(x.clone()))
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Default partition spec {default_spec_id} not found"),
- )
- })?
+ .ok_or_else(|| invalid_data!("Default partition spec {default_spec_id} not found"))?
.into();
let default_partition_type = default_spec.partition_type(¤t_schema)?;
diff --git a/crates/iceberg/src/spec/table_metadata_builder.rs b/crates/iceberg/src/spec/table_metadata_builder.rs
index 6467014ab..3ed848d2d 100644
--- a/crates/iceberg/src/spec/table_metadata_builder.rs
+++ b/crates/iceberg/src/spec/table_metadata_builder.rs
@@ -27,7 +27,7 @@ use super::{
StatisticsFile, StructType, TableMetadata, TableProperties, UNPARTITIONED_LAST_ASSIGNED_ID,
UnboundPartitionSpec,
};
-use crate::error::{Error, ErrorKind, Result};
+use crate::error::{Error, ErrorKind, Result, invalid_data};
use crate::spec::{EncryptedKey, INITIAL_ROW_ID, MIN_FORMAT_VERSION_ROW_LINEAGE};
use crate::{TableCreation, TableUpdate};
@@ -175,12 +175,8 @@ impl TableMetadataBuilder {
format_version,
} = table_creation;
- let location = location.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Can't create table without location",
- )
- })?;
+ let location =
+ location.ok_or_else(|| invalid_data!("Can't create table without location"))?;
let partition_spec = partition_spec.unwrap_or(UnboundPartitionSpec {
spec_id: None,
fields: vec![],
@@ -212,12 +208,10 @@ impl TableMetadataBuilder {
/// - Cannot downgrade to older format versions.
pub fn upgrade_format_version(mut self, format_version: FormatVersion) -> Result<Self> {
if format_version < self.metadata.format_version {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot downgrade FormatVersion from {} to {}",
- self.metadata.format_version, format_version
- ),
+ return Err(invalid_data!(
+ "Cannot downgrade FormatVersion from {} to {}",
+ self.metadata.format_version,
+ format_version
));
}
@@ -261,12 +255,9 @@ impl TableMetadataBuilder {
.collect::<Vec<_>>();
if !reserved_properties.is_empty() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Table properties should not contain reserved properties, but got: [{}]",
- reserved_properties.join(", ")
- ),
+ return Err(invalid_data!(
+ "Table properties should not contain reserved properties, but got: [{}]",
+ reserved_properties.join(", ")
));
}
@@ -299,12 +290,9 @@ impl TableMetadataBuilder {
.collect::<Vec<_>>();
if !reserved_properties.is_empty() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Table properties to remove contain reserved properties: [{}]",
- reserved_properties.join(", ")
- ),
+ return Err(invalid_data!(
+ "Table properties to remove contain reserved properties: [{}]",
+ reserved_properties.join(", ")
));
}
@@ -348,9 +336,9 @@ impl TableMetadataBuilder {
.snapshots
.contains_key(&snapshot.snapshot_id())
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Snapshot already exists for: '{}'", snapshot.snapshot_id()),
+ return Err(invalid_data!(
+ "Snapshot already exists for: '{}'",
+ snapshot.snapshot_id()
));
}
@@ -358,13 +346,10 @@ impl TableMetadataBuilder {
&& snapshot.sequence_number() <= self.metadata.last_sequence_number
&& snapshot.parent_snapshot_id().is_some()
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot add snapshot with sequence number {} older than last sequence number {}",
- snapshot.sequence_number(),
- self.metadata.last_sequence_number
- ),
+ return Err(invalid_data!(
+ "Cannot add snapshot with sequence number {} older than last sequence number {}",
+ snapshot.sequence_number(),
+ self.metadata.last_sequence_number
));
}
@@ -372,13 +357,10 @@ impl TableMetadataBuilder {
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if snapshot.timestamp_ms() - last.timestamp_ms < -ONE_MINUTE_MS {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid snapshot timestamp {}: before last snapshot timestamp {}",
- snapshot.timestamp_ms(),
- last.timestamp_ms
- ),
+ return Err(invalid_data!(
+ "Invalid snapshot timestamp {}: before last snapshot timestamp {}",
+ snapshot.timestamp_ms(),
+ last.timestamp_ms
));
}
}
@@ -388,13 +370,10 @@ impl TableMetadataBuilder {
.unwrap_or_default()
.max(self.metadata.last_updated_ms);
if snapshot.timestamp_ms() - max_last_updated < -ONE_MINUTE_MS {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid snapshot timestamp {}: before last updated timestamp {}",
- snapshot.timestamp_ms(),
- max_last_updated
- ),
+ return Err(invalid_data!(
+ "Invalid snapshot timestamp {}: before last updated timestamp {}",
+ snapshot.timestamp_ms(),
+ max_last_updated
));
}
@@ -402,22 +381,16 @@ impl TableMetadataBuilder {
if self.metadata.format_version >= MIN_FORMAT_VERSION_ROW_LINEAGE {
if let Some((first_row_id, added_rows_count)) = snapshot.row_range() {
if first_row_id < self.metadata.next_row_id {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot add a snapshot, first-row-id is behind table next-row-id: {first_row_id} < {}",
- self.metadata.next_row_id
- ),
+ return Err(invalid_data!(
+ "Cannot add a snapshot, first-row-id is behind table next-row-id: {first_row_id} < {}",
+ self.metadata.next_row_id
));
}
added_rows = Some(added_rows_count);
} else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot add a snapshot: first-row-id is null. first-row-id must be set for format version >= {MIN_FORMAT_VERSION_ROW_LINEAGE}",
- ),
+ return Err(invalid_data!(
+ "Cannot add a snapshot: first-row-id is null. first-row-id must be set for format version >= {MIN_FORMAT_VERSION_ROW_LINEAGE}",
));
}
}
@@ -428,9 +401,8 @@ impl TableMetadataBuilder {
.next_row_id
.checked_add(added_rows)
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Cannot add snapshot: next-row-id overflowed when adding added-rows",
+ invalid_data!(
+ "Cannot add snapshot: next-row-id overflowed when adding added-rows"
)
})?;
}
@@ -459,9 +431,8 @@ impl TableMetadataBuilder {
let reference = if let Some(mut reference) = reference {
if !reference.is_branch() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Cannot append snapshot to non-branch reference '{branch}'",),
+ return Err(invalid_data!(
+ "Cannot append snapshot to non-branch reference '{branch}'",
));
}
@@ -525,12 +496,9 @@ impl TableMetadataBuilder {
}
let Some(snapshot) = self.metadata.snapshots.get(&reference.snapshot_id) else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot set '{ref_name}' to unknown snapshot: '{}'",
- reference.snapshot_id
- ),
+ return Err(invalid_data!(
+ "Cannot set '{ref_name}' to unknown snapshot: '{}'",
+ reference.snapshot_id
));
};
@@ -687,9 +655,8 @@ impl TableMetadataBuilder {
pub fn set_current_schema(mut self, mut schema_id: i32) -> Result<Self> {
if schema_id == Self::LAST_ADDED {
schema_id = self.last_added_schema_id.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Cannot set current schema to last added schema: no schema has been added.",
+ invalid_data!(
+ "Cannot set current schema to last added schema: no schema has been added."
)
})?;
};
@@ -700,10 +667,7 @@ impl TableMetadataBuilder {
}
let _schema = self.metadata.schemas.get(&schema_id).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Cannot set current schema to unknown schema with id: '{schema_id}'"),
- )
+ invalid_data!("Cannot set current schema to unknown schema with id: '{schema_id}'")
})?;
// Old partition specs and sort-orders should be preserved even if they are not compatible with the new schema,
@@ -749,12 +713,9 @@ impl TableMetadataBuilder {
let is_new_field = !self.metadata.name_exists_in_any_schema(field_name);
if has_partition_conflict && is_new_field {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot add schema field '{field_name}' because it conflicts with existing partition field name. \
+ return Err(invalid_data!(
+ "Cannot add schema field '{field_name}' because it conflicts with existing partition field name. \
Schema evolution cannot introduce field names that match existing partition field names."
- ),
));
}
}
@@ -794,23 +755,19 @@ impl TableMetadataBuilder {
let has_matching_source_id = schema_field.id == partition_field.source_id;
if !is_identity_transform {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot create partition with name '{}' that conflicts with schema field and is not an identity transform.",
- partition_field.name
- ),
+ return Err(invalid_data!(
+ "Cannot create partition with name '{}' that conflicts with schema field and is not an identity transform.",
+ partition_field.name
));
}
if !has_matching_source_id {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot create identity partition sourced from different field in schema. \
+ return Err(invalid_data!(
+ "Cannot create identity partition sourced from different field in schema. \
Field name '{}' has id `{}` in schema but partition source id is `{}`",
- partition_field.name, schema_field.id, partition_field.source_id
- ),
+ partition_field.name,
+ schema_field.id,
+ partition_field.source_id
));
}
}
@@ -858,9 +815,8 @@ impl TableMetadataBuilder {
}
if self.metadata.format_version <= FormatVersion::V1 && !spec.has_sequential_ids() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot add partition spec with non-sequential field ids to format version 1 table",
+ return Err(invalid_data!(
+ "Cannot add partition spec with non-sequential field ids to format version 1 table"
));
}
@@ -926,9 +882,8 @@ impl TableMetadataBuilder {
pub fn set_default_partition_spec(mut self, mut spec_id: i32) -> Result<Self> {
if spec_id == Self::LAST_ADDED {
spec_id = self.last_added_spec_id.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Cannot set default partition spec to last added spec: no spec has been added.",
+ invalid_data!(
+ "Cannot set default partition spec to last added spec: no spec has been added."
)
})?;
}
@@ -938,9 +893,8 @@ impl TableMetadataBuilder {
}
if !self.metadata.partition_specs.contains_key(&spec_id) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Cannot set default partition spec to unknown spec with id: '{spec_id}'",),
+ return Err(invalid_data!(
+ "Cannot set default partition spec to unknown spec with id: '{spec_id}'",
));
}
@@ -949,11 +903,8 @@ impl TableMetadataBuilder {
.partition_specs
.get(&spec_id)
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot set default partition spec to unknown spec with id: '{spec_id}'",
- ),
+ invalid_data!(
+ "Cannot set default partition spec to unknown spec with id: '{spec_id}'",
)
})?
.clone();
@@ -987,10 +938,7 @@ impl TableMetadataBuilder {
/// - Cannot remove the default partition spec.
pub fn remove_partition_specs(mut self, spec_ids: &[i32]) -> Result<Self> {
if spec_ids.contains(&self.metadata.default_spec.spec_id()) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot remove default partition spec",
- ));
+ return Err(invalid_data!("Cannot remove default partition spec"));
}
let mut removed_specs = Vec::with_capacity(spec_ids.len());
@@ -1040,11 +988,8 @@ impl TableMetadataBuilder {
.with_fields(sort_order.fields)
.build(&schema)
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Sort order to add is incompatible with current schema: {e}"),
- )
- .with_source(e)
+ invalid_data!("Sort order to add is incompatible with current schema: {e}")
+ .with_source(e)
})?;
self.last_added_order_id = Some(new_order_id);
@@ -1064,9 +1009,8 @@ impl TableMetadataBuilder {
pub fn set_default_sort_order(mut self, mut sort_order_id: i64) -> Result<Self> {
if sort_order_id == Self::LAST_ADDED as i64 {
sort_order_id = self.last_added_order_id.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Cannot set default sort order to last added order: no order has been added.",
+ invalid_data!(
+ "Cannot set default sort order to last added order: no order has been added."
)
})?;
}
@@ -1076,11 +1020,8 @@ impl TableMetadataBuilder {
}
if !self.metadata.sort_orders.contains_key(&sort_order_id) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot set default sort order to unknown order with id: '{sort_order_id}'"
- ),
+ return Err(invalid_data!(
+ "Cannot set default sort order to unknown order with id: '{sort_order_id}'"
));
}
@@ -1217,9 +1158,8 @@ impl TableMetadataBuilder {
if let Some(current_snapshot_id) = self.metadata.current_snapshot_id {
let last_id = new_snapshot_log.last().map(|entry| entry.snapshot_id);
if last_id != Some(current_snapshot_id) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot set invalid snapshot log: latest entry is not the current snapshot",
+ return Err(invalid_data!(
+ "Cannot set invalid snapshot log: latest entry is not the current snapshot"
));
}
};
@@ -1289,12 +1229,10 @@ impl TableMetadataBuilder {
let mut fresh_spec = PartitionSpecBuilder::new(fresh_schema.clone());
for field in spec.fields() {
let source_field_name = previous_id_to_name.get(&field.source_id).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot find source column with id {} for partition column {} in schema.",
- field.source_id, field.name
- ),
+ invalid_data!(
+ "Cannot find source column with id {} for partition column {} in schema.",
+ field.source_id,
+ field.name
)
})?;
fresh_spec =
@@ -1306,12 +1244,9 @@ impl TableMetadataBuilder {
let mut fresh_order = SortOrder::builder();
for mut field in sort_order.fields {
let source_field_name = previous_id_to_name.get(&field.source_id).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot find source column with id {} for sort column in schema.",
- field.source_id
- ),
+ invalid_data!(
+ "Cannot find source column with id {} for sort column in schema.",
+ field.source_id
)
})?;
let new_field_id = fresh_schema
@@ -1354,12 +1289,9 @@ impl TableMetadataBuilder {
.schemas
.get(&self.metadata.current_schema_id)
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Current schema with id '{}' not found in table metadata.",
- self.metadata.current_schema_id
- ),
+ invalid_data!(
+ "Current schema with id '{}' not found in table metadata.",
+ self.metadata.current_schema_id
)
})
}
@@ -1370,12 +1302,9 @@ impl TableMetadataBuilder {
.get(&self.metadata.default_sort_order_id)
.cloned()
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Default sort order with id '{}' not found in table metadata.",
- self.metadata.default_sort_order_id
- ),
+ invalid_data!(
+ "Default sort order with id '{}' not found in table metadata.",
+ self.metadata.default_sort_order_id
)
})
}
@@ -1424,10 +1353,7 @@ impl TableMetadataBuilder {
/// Does nothing if a schema id is not present. Active schemas should not be removed.
pub fn remove_schemas(mut self, schema_id_to_remove: &[i32]) -> Result<Self> {
if schema_id_to_remove.contains(&self.metadata.current_schema_id) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot remove current schema",
- ));
+ return Err(invalid_data!("Cannot remove current schema"));
}
if schema_id_to_remove.is_empty() {
diff --git a/crates/iceberg/src/spec/table_properties.rs b/crates/iceberg/src/spec/table_properties.rs
index 01a330e28..09bf71c19 100644
--- a/crates/iceberg/src/spec/table_properties.rs
+++ b/crates/iceberg/src/spec/table_properties.rs
@@ -21,13 +21,13 @@ use iceberg_property_macro::properties_view;
use crate::compression::CompressionCodec;
use crate::encryption::AesKeySize;
-use crate::error::{Error, ErrorKind, Result};
+use crate::error::{Result, invalid_data};
use crate::spec::NameMapping;
use crate::util::location::strip_trailing_slash;
fn parse_location_property(path: &str) -> Result<String> {
if path.is_empty() {
- return Err(Error::new(ErrorKind::DataInvalid, "path must not be empty"));
+ return Err(invalid_data!("path must not be empty"));
}
Ok(strip_trailing_slash(path).to_string())
@@ -43,30 +43,22 @@ fn parse_metadata_compression(value: &str) -> Result<CompressionCodec> {
let lowercase_value = value.to_lowercase();
// Use serde to parse the codec (which has rename_all = "lowercase")
- let codec: CompressionCodec = serde_json::from_value(serde_json::Value::String(
- lowercase_value,
- ))
- .map_err(|_| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
+ let codec: CompressionCodec =
+ serde_json::from_value(serde_json::Value::String(lowercase_value)).map_err(|_| {
+ invalid_data!(
"Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported.",
CompressionCodec::None.name(),
CompressionCodec::gzip_default().name()
- ),
- )
- })?;
+ )
+ })?;
// Validate that only None and Gzip are used for metadata
match codec {
CompressionCodec::None | CompressionCodec::Gzip(_) => Ok(codec),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.",
- CompressionCodec::None.name(),
- CompressionCodec::gzip_default().name()
- ),
+ _ => Err(invalid_data!(
+ "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.",
+ CompressionCodec::None.name(),
+ CompressionCodec::gzip_default().name()
)),
}
}
@@ -85,12 +77,9 @@ fn parse_parquet_compression(
.get(codec_key)
.map(|value| {
serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid Parquet compression codec: {value}. Supported codecs: \
+ invalid_data!(
+ "Invalid Parquet compression codec: {value}. Supported codecs: \
uncompressed, snappy, gzip, lzo, brotli, lz4, lz4_raw, zstd"
- ),
)
})
})
@@ -100,12 +89,9 @@ fn parse_parquet_compression(
let level = properties
.get(level_key)
.map(|value| {
- value.parse::<u8>().map_err(|error| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid value for {level_key}: {error}"),
- )
- })
+ value
+ .parse::<u8>()
+ .map_err(|error| invalid_data!("Invalid value for {level_key}: {error}"))
})
.transpose()?;
@@ -560,6 +546,7 @@ impl TableProperties<'_> {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::ErrorKind;
use crate::compression::CompressionCodec;
#[test]
diff --git a/crates/iceberg/src/spec/transform.rs b/crates/iceberg/src/spec/transform.rs
index af539b32a..c4c915c08 100644
--- a/crates/iceberg/src/spec/transform.rs
+++ b/crates/iceberg/src/spec/transform.rs
@@ -29,7 +29,7 @@ use super::values::decimal_utils::decimal_from_i128_with_scale;
use super::values::temporal::date;
use super::{Datum, PrimitiveLiteral};
use crate::ErrorKind;
-use crate::error::{Error, Result};
+use crate::error::{Error, Result, invalid_data};
use crate::expr::{
BinaryExpression, BoundPredicate, BoundReference, Predicate, PredicateOperator, Reference,
SetExpression, UnaryExpression,
@@ -282,9 +282,8 @@ impl Transform {
if matches!(input_type, Type::Primitive(_)) {
Ok(input_type.clone())
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of identity transform",),
+ Err(invalid_data!(
+ "{input_type} is not a valid input type of identity transform",
))
}
}
@@ -306,15 +305,13 @@ impl Transform {
| PrimitiveType::Uuid
| PrimitiveType::Fixed(_)
| PrimitiveType::Binary => Ok(Type::Primitive(PrimitiveType::Int)),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of bucket transform",),
+ _ => Err(invalid_data!(
+ "{input_type} is not a valid input type of bucket transform",
)),
}
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of bucket transform",),
+ Err(invalid_data!(
+ "{input_type} is not a valid input type of bucket transform",
))
}
}
@@ -326,15 +323,13 @@ impl Transform {
| PrimitiveType::String
| PrimitiveType::Binary
| PrimitiveType::Decimal { .. } => Ok(input_type.clone()),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of truncate transform",),
+ _ => Err(invalid_data!(
+ "{input_type} is not a valid input type of truncate transform",
)),
}
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of truncate transform",),
+ Err(invalid_data!(
+ "{input_type} is not a valid input type of truncate transform",
))
}
}
@@ -346,15 +341,13 @@ impl Transform {
| PrimitiveType::TimestampNs
| PrimitiveType::TimestamptzNs
| PrimitiveType::Date => Ok(Type::Primitive(PrimitiveType::Int)),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of {self} transform",),
+ _ => Err(invalid_data!(
+ "{input_type} is not a valid input type of {self} transform",
)),
}
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of {self} transform",),
+ Err(invalid_data!(
+ "{input_type} is not a valid input type of {self} transform",
))
}
}
@@ -366,15 +359,13 @@ impl Transform {
| PrimitiveType::TimestampNs
| PrimitiveType::TimestamptzNs
| PrimitiveType::Date => Ok(Type::Primitive(PrimitiveType::Date)),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of {self} transform",),
+ _ => Err(invalid_data!(
+ "{input_type} is not a valid input type of {self} transform",
)),
}
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of {self} transform",),
+ Err(invalid_data!(
+ "{input_type} is not a valid input type of {self} transform",
))
}
}
@@ -385,15 +376,13 @@ impl Transform {
| PrimitiveType::Timestamptz
| PrimitiveType::TimestampNs
| PrimitiveType::TimestamptzNs => Ok(Type::Primitive(PrimitiveType::Int)),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of {self} transform",),
+ _ => Err(invalid_data!(
+ "{input_type} is not a valid input type of {self} transform",
)),
}
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("{input_type} is not a valid input type of {self} transform",),
+ Err(invalid_data!(
+ "{input_type} is not a valid input type of {self} transform",
))
}
}
@@ -496,12 +485,9 @@ impl Transform {
PrimitiveLiteral::String(s) => s.len(),
PrimitiveLiteral::Binary(b) => b.len(),
_ => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Expected a string or binary literal, got: {:?}",
- expr.literal()
- ),
+ return Err(invalid_data!(
+ "Expected a string or binary literal, got: {:?}",
+ expr.literal()
));
}
};
@@ -523,12 +509,9 @@ impl Transform {
PrimitiveLiteral::String(s) => s.len(),
PrimitiveLiteral::Binary(b) => b.len(),
_ => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Expected a string or binary literal, got: {:?}",
- expr.literal()
- ),
+ return Err(invalid_data!(
+ "Expected a string or binary literal, got: {:?}",
+ expr.literal()
));
}
};
@@ -1032,9 +1015,8 @@ impl Transform {
| &PrimitiveType::TimestampNs
| &PrimitiveType::TimestamptzNs
) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Expected a numeric literal, got: {boundary:?}"),
+ return Err(invalid_data!(
+ "Expected a numeric literal, got: {boundary:?}"
));
}
@@ -1139,11 +1121,7 @@ impl FromStr for Transform {
.trim_end_matches(']')
.parse()
.map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("transform bucket type {v:?} is invalid"),
- )
- .with_source(err)
+ invalid_data!("transform bucket type {v:?} is invalid").with_source(err)
})?;
Transform::Bucket(length)
@@ -1156,20 +1134,13 @@ impl FromStr for Transform {
.trim_end_matches(']')
.parse()
.map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("transform truncate type {v:?} is invalid"),
- )
- .with_source(err)
+ invalid_data!("transform truncate type {v:?} is invalid").with_source(err)
})?;
Transform::Truncate(width)
}
v => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("transform {v:?} is invalid"),
- ));
+ return Err(invalid_data!("transform {v:?} is invalid"));
}
};
diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs
index 289002906..57d967573 100644
--- a/crates/iceberg/src/spec/values/datum.rs
+++ b/crates/iceberg/src/spec/values/datum.rs
@@ -36,10 +36,10 @@ use super::literal::Literal;
use super::primitive::PrimitiveLiteral;
use super::serde::_serde::RawLiteral;
use super::temporal::{date, time, timestamp, timestamptz};
-use crate::error::Result;
+use crate::ensure_data_valid;
+use crate::error::{Error, ErrorKind, Result, invalid_data};
use crate::spec::MAX_DECIMAL_PRECISION;
use crate::spec::datatypes::{PrimitiveType, Type};
-use crate::{Error, ErrorKind, ensure_data_valid};
/// Maximum value for [`PrimitiveType::Time`] type in microseconds, e.g. 23 hours 59 minutes 59 seconds 999999 microseconds.
pub(crate) const MAX_TIME_VALUE: i64 = 24 * 60 * 60 * 1_000_000i64 - 1;
@@ -425,14 +425,10 @@ impl Datum {
}
PrimitiveType::Fixed(_) => PrimitiveLiteral::Binary(Vec::from(bytes)),
PrimitiveType::Binary => PrimitiveLiteral::Binary(Vec::from(bytes)),
- PrimitiveType::Decimal { .. } => {
- PrimitiveLiteral::Int128(i128_from_be_bytes(bytes).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't convert bytes to i128: {bytes:?}"),
- )
- })?)
- }
+ PrimitiveType::Decimal { .. } => PrimitiveLiteral::Int128(
+ i128_from_be_bytes(bytes)
+ .ok_or_else(|| invalid_data!("Can't convert bytes to i128: {bytes:?}"))?,
+ ),
};
Ok(Datum::new(data_type, literal))
}
@@ -458,23 +454,17 @@ impl Datum {
PrimitiveLiteral::Binary(val) => ByteBuf::from(val.as_slice()),
PrimitiveLiteral::Int128(val) => {
let PrimitiveType::Decimal { precision, .. } = self.r#type else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "PrimitiveLiteral Int128 must be PrimitiveType Decimal but got {}",
- self.r#type
- ),
+ return Err(invalid_data!(
+ "PrimitiveLiteral Int128 must be PrimitiveType Decimal but got {}",
+ self.r#type
));
};
// It's required by iceberg spec that we must keep the minimum
// number of bytes for the value
let Ok(required_bytes) = Type::decimal_required_bytes(precision) else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "PrimitiveType Decimal must has valid precision but got {precision}"
- ),
+ return Err(invalid_data!(
+ "PrimitiveType Decimal must has valid precision but got {precision}"
));
};
@@ -487,9 +477,8 @@ impl Datum {
ByteBuf::from(bytes)
}
PrimitiveLiteral::AboveMax | PrimitiveLiteral::BelowMin => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot convert AboveMax or BelowMin to bytes".to_string(),
+ return Err(invalid_data!(
+ "Cannot convert AboveMax or BelowMin to bytes"
));
}
};
@@ -532,9 +521,10 @@ impl Datum {
/// );
/// ```
pub fn bool_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
- let v = s.as_ref().parse::<bool>().map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Can't parse string to bool.").with_source(e)
- })?;
+ let v = s
+ .as_ref()
+ .parse::<bool>()
+ .map_err(|e| invalid_data!("Can't parse string to bool.").with_source(e))?;
Ok(Self::bool(v))
}
@@ -646,11 +636,7 @@ impl Datum {
/// ```
pub fn date_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
let t = s.as_ref().parse::<NaiveDate>().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't parse date from string: {}", s.as_ref()),
- )
- .with_source(e)
+ invalid_data!("Can't parse date from string: {}", s.as_ref()).with_source(e)
})?;
Ok(Self::date(date::date_from_naive_date(t)))
@@ -671,10 +657,7 @@ impl Datum {
/// ```
pub fn date_from_ymd(year: i32, month: u32, day: u32) -> Result<Self> {
let t = NaiveDate::from_ymd_opt(year, month, day).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't create date from year: {year}, month: {month}, day: {day}"),
- )
+ invalid_data!("Can't create date from year: {year}, month: {month}, day: {day}")
})?;
Ok(Self::date(date::date_from_naive_date(t)))
@@ -744,11 +727,7 @@ impl Datum {
/// ```
pub fn time_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
let t = s.as_ref().parse::<NaiveTime>().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't parse time from string: {}", s.as_ref()),
- )
- .with_source(e)
+ invalid_data!("Can't parse time from string: {}", s.as_ref()).with_source(e)
})?;
Ok(Self::time_from_naive_time(t))
@@ -767,10 +746,7 @@ impl Datum {
/// ```
pub fn time_from_hms_micro(hour: u32, min: u32, sec: u32, micro: u32) -> Result<Self> {
let t = NaiveTime::from_hms_micro_opt(hour, min, sec, micro)
- .ok_or_else(|| Error::new(
- ErrorKind::DataInvalid,
- format!("Can't create time from hour: {hour}, min: {min}, second: {sec}, microsecond: {micro}"),
- ))?;
+ .ok_or_else(|| invalid_data!("Can't create time from hour: {hour}, min: {min}, second: {sec}, microsecond: {micro}"))?;
Ok(Self::time_from_naive_time(t))
}
@@ -842,9 +818,10 @@ impl Datum {
/// assert_eq!(&format!("{t}"), "1992-03-01 01:02:03.000088");
/// ```
pub fn timestamp_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
- let dt = s.as_ref().parse::<NaiveDateTime>().map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Can't parse timestamp.").with_source(e)
- })?;
+ let dt = s
+ .as_ref()
+ .parse::<NaiveDateTime>()
+ .map_err(|e| invalid_data!("Can't parse timestamp.").with_source(e))?;
Ok(Self::timestamp_from_datetime(dt))
}
@@ -911,9 +888,8 @@ impl Datum {
/// assert_eq!(&format!("{t}"), "1992-02-29 17:02:03.000088 UTC");
/// ```
pub fn timestamptz_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
- let dt = DateTime::<Utc>::from_str(s.as_ref()).map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Can't parse datetime.").with_source(e)
- })?;
+ let dt = DateTime::<Utc>::from_str(s.as_ref())
+ .map_err(|e| invalid_data!("Can't parse datetime.").with_source(e))?;
Ok(Self::timestamptz_from_datetime(dt))
}
@@ -965,11 +941,7 @@ impl Datum {
/// ```
pub fn uuid_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
let uuid = uuid::Uuid::parse_str(s.as_ref()).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't parse uuid from string: {}", s.as_ref()),
- )
- .with_source(e)
+ invalid_data!("Can't parse uuid from string: {}", s.as_ref()).with_source(e)
})?;
Ok(Self::uuid(uuid))
}
@@ -1103,18 +1075,17 @@ impl Datum {
}
fn string_to_i128<S: AsRef<str>>(s: S) -> Result<i128> {
- s.as_ref().parse::<i128>().map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Can't parse string to i128.").with_source(e)
- })
+ s.as_ref()
+ .parse::<i128>()
+ .map_err(|e| invalid_data!("Can't parse string to i128.").with_source(e))
}
fn decimal_from_mantissa(mantissa: i128, precision: u32, scale: u32) -> Result<Self> {
let r#type = Type::decimal(precision, scale)?;
if decimal_precision(mantissa) > precision {
let value = decimal_from_i128_with_scale(mantissa, scale);
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Decimal value {value} is too large for precision {precision}"),
+ return Err(invalid_data!(
+ "Decimal value {value} is too large for precision {precision}"
));
}
@@ -1180,11 +1151,8 @@ impl Datum {
scale: target_scale,
..
},
- ) => Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Decimal scale conversion is not supported: source scale {self_scale}, target scale {target_scale}"
- ),
+ ) => Err(invalid_data!(
+ "Decimal scale conversion is not supported: source scale {self_scale}, target scale {target_scale}"
)),
(PrimitiveLiteral::String(val), _, PrimitiveType::Boolean) => {
Datum::bool_from_str(val)
@@ -1204,21 +1172,17 @@ impl Datum {
// TODO: implement more type conversions
(_, self_type, target_type) if self_type == target_type => Ok(self),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Can't convert datum from {} type to {} type.",
- self.r#type, target_primitive_type
- ),
+ _ => Err(invalid_data!(
+ "Can't convert datum from {} type to {} type.",
+ self.r#type,
+ target_primitive_type
)),
}
}
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Can't convert datum from {} type to {} type.",
- self.r#type, target_type
- ),
+ _ => Err(invalid_data!(
+ "Can't convert datum from {} type to {} type.",
+ self.r#type,
+ target_type
)),
}
}
diff --git a/crates/iceberg/src/spec/values/decimal_utils.rs b/crates/iceberg/src/spec/values/decimal_utils.rs
index 3d09d6f25..8bb90c973 100644
--- a/crates/iceberg/src/spec/values/decimal_utils.rs
+++ b/crates/iceberg/src/spec/values/decimal_utils.rs
@@ -23,7 +23,8 @@
use fastnum::D128;
use fastnum::decimal::Context;
-use crate::{Error, ErrorKind, Result};
+use crate::Result;
+use crate::error::invalid_data;
/// Re-export D128 as the Decimal type for use throughout the crate.
pub type Decimal = D128;
@@ -92,8 +93,7 @@ pub fn decimal_new(mantissa: i64, scale: u32) -> Decimal {
///
/// This is equivalent to rust_decimal's `Decimal::from_str_exact`.
pub fn decimal_from_str_exact(s: &str) -> Result<Decimal> {
- D128::from_str(s, Context::default())
- .map_err(|e| Error::new(ErrorKind::DataInvalid, format!("Can't parse decimal: {e}")))
+ D128::from_str(s, Context::default()).map_err(|e| invalid_data!("Can't parse decimal: {e}"))
}
/// Get the mantissa (unscaled coefficient) as i128.
diff --git a/crates/iceberg/src/spec/values/literal.rs b/crates/iceberg/src/spec/values/literal.rs
index 63d6c1aee..1d27ae995 100644
--- a/crates/iceberg/src/spec/values/literal.rs
+++ b/crates/iceberg/src/spec/values/literal.rs
@@ -32,9 +32,8 @@ use super::decimal_utils::{
use super::primitive::PrimitiveLiteral;
use super::struct_value::Struct;
use super::temporal::{date, time, timestamp, timestamptz};
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::spec::datatypes::{PrimitiveType, Type};
-use crate::{Error, ErrorKind};
/// Values present in iceberg type
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -80,9 +79,10 @@ impl Literal {
/// assert_eq!(Literal::Primitive(PrimitiveLiteral::Boolean(false)), t);
/// ```
pub fn bool_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
- let v = s.as_ref().parse::<bool>().map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Can't parse string to bool.").with_source(e)
- })?;
+ let v = s
+ .as_ref()
+ .parse::<bool>()
+ .map_err(|e| invalid_data!("Can't parse string to bool.").with_source(e))?;
Ok(Self::Primitive(PrimitiveLiteral::Boolean(v)))
}
@@ -164,11 +164,7 @@ impl Literal {
/// ```
pub fn date_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
let t = s.as_ref().parse::<NaiveDate>().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't parse date from string: {}", s.as_ref()),
- )
- .with_source(e)
+ invalid_data!("Can't parse date from string: {}", s.as_ref()).with_source(e)
})?;
Ok(Self::date(date::date_from_naive_date(t)))
@@ -188,10 +184,7 @@ impl Literal {
/// ```
pub fn date_from_ymd(year: i32, month: u32, day: u32) -> Result<Self> {
let t = NaiveDate::from_ymd_opt(year, month, day).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't create date from year: {year}, month: {month}, day: {day}"),
- )
+ invalid_data!("Can't create date from year: {year}, month: {month}, day: {day}")
})?;
Ok(Self::date(date::date_from_naive_date(t)))
@@ -230,11 +223,7 @@ impl Literal {
/// ```
pub fn time_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
let t = s.as_ref().parse::<NaiveTime>().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't parse time from string: {}", s.as_ref()),
- )
- .with_source(e)
+ invalid_data!("Can't parse time from string: {}", s.as_ref()).with_source(e)
})?;
Ok(Self::time_from_naive_time(t))
@@ -253,10 +242,7 @@ impl Literal {
/// ```
pub fn time_from_hms_micro(hour: u32, min: u32, sec: u32, micro: u32) -> Result<Self> {
let t = NaiveTime::from_hms_micro_opt(hour, min, sec, micro)
- .ok_or_else(|| Error::new(
- ErrorKind::DataInvalid,
- format!("Can't create time from hour: {hour}, min: {min}, second: {sec}, microsecond: {micro}"),
- ))?;
+ .ok_or_else(|| invalid_data!("Can't create time from hour: {hour}, min: {min}, second: {sec}, microsecond: {micro}"))?;
Ok(Self::time_from_naive_time(t))
}
@@ -314,18 +300,16 @@ impl Literal {
/// assert_eq!(t, t2);
/// ```
pub fn timestamp_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
- let dt = DateTime::<Utc>::from_str(s.as_ref()).map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Can't parse datetime.").with_source(e)
- })?;
+ let dt = DateTime::<Utc>::from_str(s.as_ref())
+ .map_err(|e| invalid_data!("Can't parse datetime.").with_source(e))?;
Ok(Self::timestamp_from_datetime(dt))
}
/// Similar to [`Literal::timestamp_from_str`], but return timestamp with timezone literal.
pub fn timestamptz_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
- let dt = DateTime::<Utc>::from_str(s.as_ref()).map_err(|e| {
- Error::new(ErrorKind::DataInvalid, "Can't parse datetime.").with_source(e)
- })?;
+ let dt = DateTime::<Utc>::from_str(s.as_ref())
+ .map_err(|e| invalid_data!("Can't parse datetime.").with_source(e))?;
Ok(Self::timestamptz_from_datetime(dt))
}
@@ -354,11 +338,7 @@ impl Literal {
/// ```
pub fn uuid_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
let uuid = Uuid::parse_str(s.as_ref()).map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Can't parse uuid from string: {}", s.as_ref()),
- )
- .with_source(e)
+ invalid_data!("Can't parse uuid from string: {}", s.as_ref()).with_source(e)
})?;
Ok(Self::uuid(uuid))
}
@@ -435,30 +415,31 @@ impl Literal {
Ok(Some(Literal::Primitive(PrimitiveLiteral::Int(
number
.as_i64()
- .ok_or(Error::new(
- ErrorKind::DataInvalid,
- "Failed to convert json number to int",
- ))?
+ .ok_or(invalid_data!("Failed to convert json number to int"))?
.try_into()?,
))))
}
- (PrimitiveType::Long, JsonValue::Number(number)) => Ok(Some(Literal::Primitive(
- PrimitiveLiteral::Long(number.as_i64().ok_or(Error::new(
- ErrorKind::DataInvalid,
- "Failed to convert json number to long",
- ))?),
- ))),
+ (PrimitiveType::Long, JsonValue::Number(number)) => {
+ Ok(Some(Literal::Primitive(PrimitiveLiteral::Long(
+ number
+ .as_i64()
+ .ok_or(invalid_data!("Failed to convert json number to long"))?,
+ ))))
+ }
(PrimitiveType::Float, JsonValue::Number(number)) => Ok(Some(Literal::Primitive(
- PrimitiveLiteral::Float(OrderedFloat(number.as_f64().ok_or(Error::new(
- ErrorKind::DataInvalid,
- "Failed to convert json number to float",
- ))? as f32)),
+ PrimitiveLiteral::Float(OrderedFloat(
+ number
+ .as_f64()
+ .ok_or(invalid_data!("Failed to convert json number to float"))?
+ as f32,
+ )),
))),
(PrimitiveType::Double, JsonValue::Number(number)) => Ok(Some(Literal::Primitive(
- PrimitiveLiteral::Double(OrderedFloat(number.as_f64().ok_or(Error::new(
- ErrorKind::DataInvalid,
- "Failed to convert json number to double",
- ))?)),
+ PrimitiveLiteral::Double(OrderedFloat(
+ number
+ .as_f64()
+ .ok_or(invalid_data!("Failed to convert json number to double"))?,
+ )),
))),
(PrimitiveType::Date, JsonValue::String(s)) => {
Ok(Some(Literal::Primitive(PrimitiveLiteral::Int(
@@ -469,9 +450,8 @@ impl Literal {
Ok(Some(Literal::Primitive(PrimitiveLiteral::Int(
number
.as_i64()
- .ok_or(Error::new(
- ErrorKind::DataInvalid,
- "Failed to convert json number to date (days since epoch)",
+ .ok_or(invalid_data!(
+ "Failed to convert json number to date (days since epoch)"
))?
.try_into()?,
))))
@@ -496,11 +476,8 @@ impl Literal {
(PrimitiveType::TimestampNs, JsonValue::String(s)) => {
let ndt = NaiveDateTime::parse_from_str(&s, "%Y-%m-%dT%H:%M:%S%.f")?;
let nanos = timestamp::datetime_to_nanoseconds(&ndt).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Timestamp is outside the representable nanosecond range: {ndt}"
- ),
+ invalid_data!(
+ "Timestamp is outside the representable nanosecond range: {ndt}"
)
})?;
Ok(Some(Literal::Primitive(PrimitiveLiteral::Long(nanos))))
@@ -511,11 +488,8 @@ impl Literal {
"%Y-%m-%dT%H:%M:%S%.f+00:00",
)?);
let nanos = timestamptz::datetimetz_to_nanoseconds(&dt).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Timestamptz is outside the representable nanosecond range: {dt}"
- ),
+ invalid_data!(
+ "Timestamptz is outside the representable nanosecond range: {dt}"
)
})?;
Ok(Some(Literal::Primitive(PrimitiveLiteral::Long(nanos))))
@@ -547,14 +521,12 @@ impl Literal {
decimal_mantissa(&rescaled),
))))
}
- (PrimitiveType::Unknown, value) if !value.is_null() => Err(Error::new(
- ErrorKind::DataInvalid,
- "Unknown type only supports null default values",
+ (PrimitiveType::Unknown, value) if !value.is_null() => Err(invalid_data!(
+ "Unknown type only supports null default values"
)),
(_, JsonValue::Null) => Ok(None),
- (i, j) => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("The json value {j} doesn't fit to the iceberg type {i}."),
+ (i, j) => Err(invalid_data!(
+ "The json value {j} doesn't fit to the iceberg type {i}."
)),
},
Type::Struct(schema) => {
@@ -569,9 +541,8 @@ impl Literal {
.collect::<Result<Vec<_>>>()?;
Ok(Some(Literal::Struct(Struct::from_iter(values))))
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "The json value for a struct type must be an object.",
+ Err(invalid_data!(
+ "The json value for a struct type must be an object."
))
}
}
@@ -586,9 +557,8 @@ impl Literal {
.collect::<Result<Vec<_>>>()?,
)))
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "The json value for a list type must be an array.",
+ Err(invalid_data!(
+ "The json value for a list type must be an array."
))
}
}
@@ -598,13 +568,10 @@ impl Literal {
(object.remove("keys"), object.remove("values"))
{
if keys.len() != values.len() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Map keys and values must have the same length, got {} keys and {} values",
- keys.len(),
- values.len()
- ),
+ return Err(invalid_data!(
+ "Map keys and values must have the same length, got {} keys and {} values",
+ keys.len(),
+ values.len()
));
}
@@ -615,9 +582,8 @@ impl Literal {
Ok((
Literal::try_from_json(key, &map.key_field.field_type)
.and_then(|value| {
- value.ok_or(Error::new(
- ErrorKind::DataInvalid,
- "Key of map cannot be null",
+ value.ok_or(invalid_data!(
+ "Key of map cannot be null"
))
})?,
Literal::try_from_json(value, &map.value_field.field_type)?,
@@ -626,21 +592,18 @@ impl Literal {
.collect::<Result<Vec<_>>>()?,
))))
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "The json value for a list type must be an array.",
+ Err(invalid_data!(
+ "The json value for a list type must be an array."
))
}
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- "The json value for a list type must be an array.",
+ Err(invalid_data!(
+ "The json value for a list type must be an array."
))
}
}
- Type::Variant(_) => Err(Error::new(
- ErrorKind::DataInvalid,
- "Variant type is not supported for single-value JSON serialization",
+ Type::Variant(_) => Err(invalid_data!(
+ "Variant type is not supported for single-value JSON serialization"
)),
}
}
@@ -721,14 +684,12 @@ impl Literal {
let decimal = try_decimal_from_i128_with_scale(val, *scale)?;
Ok(JsonValue::String(decimal.to_string()))
}
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- "The iceberg type for decimal literal must be decimal.",
+ _ => Err(invalid_data!(
+ "The iceberg type for decimal literal must be decimal."
))?,
},
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- "The iceberg value doesn't fit to the iceberg type.",
+ _ => Err(invalid_data!(
+ "The iceberg value doesn't fit to the iceberg type."
)),
},
(Literal::Struct(s), Type::Struct(struct_type)) => {
@@ -765,9 +726,8 @@ impl Literal {
object.insert("values".to_string(), JsonValue::Array(json_values));
Ok(JsonValue::Object(object))
}
- (value, r#type) => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("The iceberg value {value:?} doesn't fit to the iceberg type {type}."),
+ (value, r#type) => Err(invalid_data!(
+ "The iceberg value {value:?} doesn't fit to the iceberg type {type}."
)),
}
}
@@ -794,9 +754,8 @@ impl Literal {
fn decode_hex_bytes(value: &str) -> Result<Vec<u8>> {
if !value.len().is_multiple_of(2) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Hex string must have an even number of characters: {value:?}"),
+ return Err(invalid_data!(
+ "Hex string must have an even number of characters: {value:?}"
));
}
@@ -816,9 +775,8 @@ fn decode_hex_digit(digit: u8, value: &str) -> Result<u8> {
b'0'..=b'9' => Ok(digit - b'0'),
b'a'..=b'f' => Ok(digit - b'a' + 10),
b'A'..=b'F' => Ok(digit - b'A' + 10),
- _ => Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Hex string contains invalid character: {value:?}"),
+ _ => Err(invalid_data!(
+ "Hex string contains invalid character: {value:?}"
)),
}
}
@@ -838,9 +796,8 @@ fn validate_fixed_size(actual: usize, expected: u64) -> Result<()> {
if actual as u64 == expected {
Ok(())
} else {
- Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Fixed type must be exactly {expected} bytes, got {actual}"),
+ Err(invalid_data!(
+ "Fixed type must be exactly {expected} bytes, got {actual}"
))
}
}
diff --git a/crates/iceberg/src/spec/values/serde.rs b/crates/iceberg/src/spec/values/serde.rs
index 8a00a79de..896235b1e 100644
--- a/crates/iceberg/src/spec/values/serde.rs
+++ b/crates/iceberg/src/spec/values/serde.rs
@@ -24,9 +24,10 @@ pub(crate) mod _serde {
use serde_bytes::ByteBuf;
use serde_derive::{Deserialize as DeserializeDerive, Serialize as SerializeDerive};
+ use crate::Error;
+ use crate::error::invalid_data;
use crate::spec::values::{Literal, Map, PrimitiveLiteral, Struct};
use crate::spec::{MAP_KEY_FIELD_NAME, MAP_VALUE_FIELD_NAME, PrimitiveType, Type};
- use crate::{Error, ErrorKind};
#[derive(SerializeDerive, DeserializeDerive, Debug)]
#[serde(transparent)]
@@ -245,10 +246,7 @@ pub(crate) mod _serde {
RawLiteralEnum::Bytes(ByteBuf::from(v.to_be_bytes()))
}
PrimitiveLiteral::AboveMax | PrimitiveLiteral::BelowMin => {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Can't convert AboveMax or BelowMax",
- ));
+ return Err(invalid_data!("Can't convert AboveMax or BelowMax"));
}
},
Literal::Struct(r#struct) => {
@@ -263,9 +261,8 @@ pub(crate) mod _serde {
RawLiteralEnum::try_from(value, &field.field_type)?,
));
} else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Can't convert null to required field",
+ return Err(invalid_data!(
+ "Can't convert null to required field"
));
}
} else if let Some(value) = value {
@@ -278,10 +275,7 @@ pub(crate) mod _serde {
}
}
} else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Type {ty} should be a struct"),
- ));
+ return Err(invalid_data!("Type {ty} should be a struct"));
}
RawLiteralEnum::Record(Record { required, optional })
}
@@ -301,10 +295,7 @@ pub(crate) mod _serde {
required: list_ty.element_field.required,
})
} else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Type {ty} should be a list"),
- ));
+ return Err(invalid_data!("Type {ty} should be a list"));
}
}
Literal::Map(map) => {
@@ -325,9 +316,8 @@ pub(crate) mod _serde {
.transpose()?,
));
} else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "literal type is inconsistent with type",
+ return Err(invalid_data!(
+ "literal type is inconsistent with type"
));
}
}
@@ -348,7 +338,7 @@ pub(crate) mod _serde {
Ok(Some(RawLiteralEnum::Record(Record {
required: vec![
(MAP_KEY_FIELD_NAME.to_string(), raw_k),
- (MAP_VALUE_FIELD_NAME.to_string(), raw_v.ok_or_else(||Error::new(ErrorKind::DataInvalid, "Map value is required, value cannot be null"))?),
+ (MAP_VALUE_FIELD_NAME.to_string(), raw_v.ok_or_else(||invalid_data!("Map value is required, value cannot be null"))?),
],
optional: vec![],
})))
@@ -369,10 +359,7 @@ pub(crate) mod _serde {
})
}
} else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Type {ty} should be a map"),
- ));
+ return Err(invalid_data!("Type {ty} should be a map"));
}
}
};
@@ -381,19 +368,13 @@ pub(crate) mod _serde {
pub fn try_into(self, ty: &Type) -> Result<Option<Literal>, Error> {
let invalid_err = |v: &str| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Unable to convert raw literal ({v}) fail convert to type {ty} for: type mismatch"
- ),
+ invalid_data!(
+ "Unable to convert raw literal ({v}) fail convert to type {ty} for: type mismatch"
)
};
let invalid_err_with_reason = |v: &str, reason: &str| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Unable to convert raw literal ({v}) fail convert to type {ty} for: {reason}"
- ),
+ invalid_data!(
+ "Unable to convert raw literal ({v}) fail convert to type {ty} for: {reason}"
)
};
match self {
diff --git a/crates/iceberg/src/spec/view_metadata.rs b/crates/iceberg/src/spec/view_metadata.rs
index cb691e208..4a816140c 100644
--- a/crates/iceberg/src/spec/view_metadata.rs
+++ b/crates/iceberg/src/spec/view_metadata.rs
@@ -32,8 +32,7 @@ use uuid::Uuid;
pub use super::view_metadata_builder::ViewMetadataBuilder;
use super::view_version::{ViewVersionId, ViewVersionRef};
use super::{SchemaId, SchemaRef};
-use crate::error::{Result, timestamp_ms_to_utc};
-use crate::{Error, ErrorKind};
+use crate::error::{Result, invalid_data, timestamp_ms_to_utc};
/// Reference to [`ViewMetadata`].
pub type ViewMetadataRef = Arc<ViewMetadata>;
@@ -170,12 +169,9 @@ impl ViewMetadata {
fn validate_current_version_id(&self) -> Result<()> {
if !self.versions.contains_key(&self.current_version_id) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "No version exists with the current version id {}.",
- self.current_version_id
- ),
+ return Err(invalid_data!(
+ "No version exists with the current version id {}.",
+ self.current_version_id
));
}
Ok(())
@@ -184,9 +180,8 @@ impl ViewMetadata {
fn validate_current_schema_id(&self) -> Result<()> {
let schema_id = self.current_version().schema_id();
if !self.schemas.contains_key(&schema_id) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("No schema exists with the schema id {schema_id}."),
+ return Err(invalid_data!(
+ "No schema exists with the schema id {schema_id}."
));
}
Ok(())
diff --git a/crates/iceberg/src/spec/view_metadata_builder.rs b/crates/iceberg/src/spec/view_metadata_builder.rs
index 38041ca62..a59822e70 100644
--- a/crates/iceberg/src/spec/view_metadata_builder.rs
+++ b/crates/iceberg/src/spec/view_metadata_builder.rs
@@ -31,7 +31,7 @@ use super::{
};
use crate::ViewCreation;
use crate::catalog::ViewUpdate;
-use crate::error::{Error, ErrorKind, Result};
+use crate::error::{Result, invalid_data};
use crate::io::is_truthy;
/// Manipulating view metadata.
@@ -142,12 +142,10 @@ impl ViewMetadataBuilder {
/// - Cannot downgrade to older format versions.
pub fn upgrade_format_version(self, format_version: ViewFormatVersion) -> Result<Self> {
if format_version < self.metadata.format_version {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot downgrade ViewFormatVersion from {} to {}",
- self.metadata.format_version, format_version
- ),
+ return Err(invalid_data!(
+ "Cannot downgrade ViewFormatVersion from {} to {}",
+ self.metadata.format_version,
+ format_version
));
}
@@ -183,9 +181,8 @@ impl ViewMetadataBuilder {
pub fn set_current_version_id(mut self, mut version_id: i32) -> Result<Self> {
if version_id == Self::LAST_ADDED {
let Some(last_added_id) = self.last_added_version_id else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot set current version id to last added version: no version has been added.",
+ return Err(invalid_data!(
+ "Cannot set current version id to last added version: no version has been added."
));
};
version_id = last_added_id;
@@ -198,10 +195,7 @@ impl ViewMetadataBuilder {
}
let version = self.metadata.versions.get(&version_id).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Cannot set current version to unknown version with id: {version_id}"),
- )
+ invalid_data!("Cannot set current version to unknown version with id: {version_id}")
})?;
self.metadata.current_version_id = version_id;
@@ -276,10 +270,7 @@ impl ViewMetadataBuilder {
let view_version = if view_version.schema_id() == Self::LAST_ADDED {
let last_added_schema_id = self.last_added_schema_id.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Cannot set last added schema: no schema has been added",
- )
+ invalid_data!("Cannot set last added schema: no schema has been added")
})?;
view_version.with_schema_id(last_added_schema_id)
} else {
@@ -291,12 +282,9 @@ impl ViewMetadataBuilder {
.schemas
.contains_key(&view_version.schema_id())
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot add version with unknown schema: {}",
- view_version.schema_id()
- ),
+ return Err(invalid_data!(
+ "Cannot add version with unknown schema: {}",
+ view_version.schema_id()
));
}
@@ -308,13 +296,10 @@ impl ViewMetadataBuilder {
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if view_version.timestamp_ms() - last.timestamp_ms() < -ONE_MINUTE_MS {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid snapshot timestamp {}: before last snapshot timestamp {}",
- view_version.timestamp_ms(),
- last.timestamp_ms()
- ),
+ return Err(invalid_data!(
+ "Invalid snapshot timestamp {}: before last snapshot timestamp {}",
+ view_version.timestamp_ms(),
+ last.timestamp_ms()
));
}
}
@@ -427,11 +412,8 @@ impl ViewMetadataBuilder {
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(1);
if num_versions_to_keep < 0 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "{VIEW_PROPERTY_VERSION_HISTORY_SIZE} must be positive but was {num_versions_to_keep}"
- ),
+ return Err(invalid_data!(
+ "{VIEW_PROPERTY_VERSION_HISTORY_SIZE} must be positive but was {num_versions_to_keep}"
));
}
@@ -587,14 +569,11 @@ fn require_no_dialect_dropped(previous: &ViewVersion, current: &ViewVersion) ->
let updated_dialects = lowercase_sql_dialects_for(current);
if !updated_dialects.is_superset(&base_dialects) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot replace view due to loss of view dialects: \nPrevious dialects: {:?}\nNew dialects: {:?}\nSet {} to true to allow dropping dialects.",
- Vec::from_iter(base_dialects),
- Vec::from_iter(updated_dialects),
- VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED
- ),
+ return Err(invalid_data!(
+ "Cannot replace view due to loss of view dialects: \nPrevious dialects: {:?}\nNew dialects: {:?}\nSet {} to true to allow dropping dialects.",
+ Vec::from_iter(base_dialects),
+ Vec::from_iter(updated_dialects),
+ VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED
));
}
@@ -617,12 +596,9 @@ pub(super) fn require_unique_dialects(view_version: &ViewVersion) -> Result<()>
match repr {
ViewRepresentation::Sql(sql_repr) => {
if !seen_dialects.insert(sql_repr.dialect.to_lowercase()) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid view version: Cannot add multiple queries for dialect {}",
- sql_repr.dialect
- ),
+ return Err(invalid_data!(
+ "Invalid view version: Cannot add multiple queries for dialect {}",
+ sql_repr.dialect
));
}
}
diff --git a/crates/iceberg/src/spec/view_version.rs b/crates/iceberg/src/spec/view_version.rs
index a02d58c4e..af95f74e6 100644
--- a/crates/iceberg/src/spec/view_version.rs
+++ b/crates/iceberg/src/spec/view_version.rs
@@ -29,9 +29,8 @@ use typed_builder::TypedBuilder;
use super::INITIAL_VIEW_VERSION_ID;
use super::view_metadata::ViewVersionLog;
use crate::catalog::NamespaceIdent;
-use crate::error::{Result, timestamp_ms_to_utc};
+use crate::error::{Result, invalid_data, timestamp_ms_to_utc};
use crate::spec::{SchemaId, SchemaRef, ViewMetadata};
-use crate::{Error, ErrorKind};
/// Reference to [`ViewVersion`].
pub type ViewVersionRef = Arc<ViewVersion>;
@@ -116,12 +115,7 @@ impl ViewVersion {
pub fn schema(&self, view_metadata: &ViewMetadata) -> Result<SchemaRef> {
view_metadata
.schema_by_id(self.schema_id())
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Schema with id {} not found", self.schema_id()),
- )
- })
+ .ok_or_else(|| invalid_data!("Schema with id {} not found", self.schema_id()))
.cloned()
}
diff --git a/crates/iceberg/src/table.rs b/crates/iceberg/src/table.rs
index 21cd8681a..78c2dff05 100644
--- a/crates/iceberg/src/table.rs
+++ b/crates/iceberg/src/table.rs
@@ -22,6 +22,7 @@ use std::sync::Arc;
use crate::arrow::ArrowReaderBuilder;
use crate::encryption::EncryptionManager;
use crate::encryption::kms::KeyManagementClient;
+use crate::error::invalid_data;
use crate::inspect::MetadataTable;
use crate::io::FileIO;
use crate::io::object_cache::ObjectCache;
@@ -30,7 +31,7 @@ use crate::scan::TableScanBuilder;
use crate::spec::{
ManifestListReader, ManifestReader, SchemaRef, SnapshotRef, TableMetadata, TableMetadataRef,
};
-use crate::{Error, ErrorKind, Result, TableIdent};
+use crate::{Result, TableIdent};
/// Builder to create table scan.
pub struct TableBuilder {
@@ -135,30 +136,26 @@ impl TableBuilder {
} = self;
let Some(file_io) = file_io else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "FileIO must be provided with TableBuilder.file_io()",
+ return Err(invalid_data!(
+ "FileIO must be provided with TableBuilder.file_io()"
));
};
let Some(metadata) = metadata else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "TableMetadataRef must be provided with TableBuilder.metadata()",
+ return Err(invalid_data!(
+ "TableMetadataRef must be provided with TableBuilder.metadata()"
));
};
let Some(identifier) = identifier else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "TableIdent must be provided with TableBuilder.identifier()",
+ return Err(invalid_data!(
+ "TableIdent must be provided with TableBuilder.identifier()"
));
};
let Some(runtime) = runtime else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Runtime must be provided with TableBuilder.runtime()",
+ return Err(invalid_data!(
+ "Runtime must be provided with TableBuilder.runtime()"
));
};
@@ -248,12 +245,9 @@ impl Table {
/// Returns current metadata location in a result.
pub fn metadata_location_result(&self) -> Result<&str> {
- self.metadata_location.as_deref().ok_or(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Metadata location does not exist for table: {}",
- self.identifier
- ),
+ self.metadata_location.as_deref().ok_or(invalid_data!(
+ "Metadata location does not exist for table: {}",
+ self.identifier
))
}
@@ -415,6 +409,7 @@ mod tests {
use std::fs;
use super::*;
+ use crate::ErrorKind;
use crate::encryption::SensitiveBytes;
use crate::encryption::kms::MemoryKeyManagementClient;
use crate::spec::TableProperties;
diff --git a/crates/iceberg/src/transaction/expire_snapshots.rs b/crates/iceberg/src/transaction/expire_snapshots.rs
index 0a390d982..0e08c6b24 100644
--- a/crates/iceberg/src/transaction/expire_snapshots.rs
+++ b/crates/iceberg/src/transaction/expire_snapshots.rs
@@ -21,12 +21,13 @@ use std::sync::Arc;
use async_trait::async_trait;
use chrono::Utc;
+use crate::error::invalid_data;
use crate::spec::{
MAIN_BRANCH, SnapshotReference, SnapshotRetention, TableMetadata, TableProperties,
};
use crate::table::Table;
use crate::transaction::action::{ActionCommit, TransactionAction};
-use crate::{Error, ErrorKind, Result, TableRequirement, TableUpdate};
+use crate::{Error, Result, TableRequirement, TableUpdate};
/// A transaction action that removes snapshots from table metadata.
///
@@ -100,9 +101,8 @@ impl ExpireSnapshotsAction {
fn plan(&self, table: &Table, properties: &TableProperties<'_>) -> Result<ExpirePlan> {
// Matches Java `RemoveSnapshots.retainLast`, which requires at least one snapshot.
if self.retain_last == Some(0) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Number of snapshots to retain must be at least 1",
+ return Err(invalid_data!(
+ "Number of snapshots to retain must be at least 1"
));
}
@@ -277,7 +277,7 @@ impl ExpireSnapshotsAction {
fn reference_error(metadata: &TableMetadata, snapshot_id: i64) -> Error {
if metadata.current_snapshot_id() == Some(snapshot_id) {
- return Error::new(ErrorKind::DataInvalid, "Cannot expire the current snapshot");
+ return invalid_data!("Cannot expire the current snapshot");
}
let ref_names: Vec<&str> = metadata
.refs
@@ -285,10 +285,7 @@ impl ExpireSnapshotsAction {
.filter(|(_, snapshot_ref)| snapshot_ref.snapshot_id == snapshot_id)
.map(|(ref_name, _)| ref_name.as_str())
.collect();
- Error::new(
- ErrorKind::DataInvalid,
- format!("Cannot expire snapshot {snapshot_id}: still referenced by {ref_names:?}"),
- )
+ invalid_data!("Cannot expire snapshot {snapshot_id}: still referenced by {ref_names:?}")
}
}
@@ -306,9 +303,8 @@ impl TransactionAction for ExpireSnapshotsAction {
// Expiring metadata defeats a user's explicit decision to disable GC (Java refuses too).
if !properties.gc_enabled()? {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Cannot expire snapshots: gc.enabled is false",
+ return Err(invalid_data!(
+ "Cannot expire snapshots: gc.enabled is false"
));
}
diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs
index 1f09a79a0..43e8b25d5 100644
--- a/crates/iceberg/src/transaction/snapshot.rs
+++ b/crates/iceberg/src/transaction/snapshot.rs
@@ -23,7 +23,7 @@ use futures::TryStreamExt;
use futures::stream::FuturesUnordered;
use uuid::Uuid;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::spec::{
DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType, ManifestEntry,
ManifestFile, ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation, Snapshot,
@@ -139,16 +139,14 @@ impl<'a> SnapshotProducer<'a> {
pub(crate) fn validate_added_data_files(&self) -> Result<()> {
for data_file in &self.added_data_files {
if data_file.content_type() != crate::spec::DataContentType::Data {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Only data content type is allowed for fast append",
+ return Err(invalid_data!(
+ "Only data content type is allowed for fast append"
));
}
// Check if the data file partition spec id matches the table default partition spec id.
if self.table.metadata().default_partition_spec_id() != data_file.partition_spec_id {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Data file partition spec id does not match table default partition spec id",
+ return Err(invalid_data!(
+ "Data file partition spec id does not match table default partition spec id"
));
}
Self::validate_partition_value(
@@ -200,12 +198,9 @@ impl<'a> SnapshotProducer<'a> {
.await?;
if !referenced_files.is_empty() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Cannot add files that are already referenced by table, files: {}",
- referenced_files.join(", ")
- ),
+ return Err(invalid_data!(
+ "Cannot add files that are already referenced by table, files: {}",
+ referenced_files.join(", ")
));
}
@@ -281,9 +276,8 @@ impl<'a> SnapshotProducer<'a> {
partition_type: &StructType,
) -> Result<()> {
if partition_value.fields().len() != partition_type.fields().len() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Partition value is not compatible with partition type",
+ return Err(invalid_data!(
+ "Partition value is not compatible with partition type"
));
}
@@ -297,9 +291,8 @@ impl<'a> SnapshotProducer<'a> {
if let Some(value) = value
&& !field.compatible(&value.as_primitive_literal().unwrap())
{
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Partition value is not compatible partition type",
+ return Err(invalid_data!(
+ "Partition value is not compatible partition type"
));
}
}
diff --git a/crates/iceberg/src/transaction/sort_order.rs b/crates/iceberg/src/transaction/sort_order.rs
index 17655fe23..45c281703 100644
--- a/crates/iceberg/src/transaction/sort_order.rs
+++ b/crates/iceberg/src/transaction/sort_order.rs
@@ -19,11 +19,11 @@ use std::sync::Arc;
use async_trait::async_trait;
-use crate::error::Result;
+use crate::error::{Result, invalid_data};
use crate::spec::{NullOrder, SchemaRef, SortDirection, SortField, SortOrder, Transform};
use crate::table::Table;
use crate::transaction::{ActionCommit, TransactionAction};
-use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
+use crate::{TableRequirement, TableUpdate};
/// Represents a sort field whose construction and validation are deferred until commit time.
/// This avoids the need to pass a `Table` reference into methods like `asc` or `desc` when
@@ -38,12 +38,9 @@ struct PendingSortField {
impl PendingSortField {
fn to_sort_field(&self, schema: &SchemaRef) -> Result<SortField> {
- let field_id = schema.field_id_by_name(self.name.as_str()).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Cannot find field {} in table schema", self.name),
- )
- })?;
+ let field_id = schema
+ .field_id_by_name(self.name.as_str())
+ .ok_or_else(|| invalid_data!("Cannot find field {} in table schema", self.name))?;
// This action intentionally refuses to author all-null (void) sort keys;
// this policy does not restrict reading or preserving existing metadata.
diff --git a/crates/iceberg/src/transaction/update_location.rs b/crates/iceberg/src/transaction/update_location.rs
index 4842bcfdf..7d0e2f954 100644
--- a/crates/iceberg/src/transaction/update_location.rs
+++ b/crates/iceberg/src/transaction/update_location.rs
@@ -19,9 +19,10 @@ use std::sync::Arc;
use async_trait::async_trait;
+use crate::error::invalid_data;
use crate::table::Table;
use crate::transaction::action::{ActionCommit, TransactionAction};
-use crate::{Error, ErrorKind, Result, TableUpdate};
+use crate::{Result, TableUpdate};
/// A transaction action that sets or updates the location of a table.
///
@@ -60,9 +61,8 @@ impl TransactionAction for UpdateLocationAction {
if let Some(location) = self.location.clone() {
updates = vec![TableUpdate::SetLocation { location }];
} else {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Location is not set for UpdateLocationAction!",
+ return Err(invalid_data!(
+ "Location is not set for UpdateLocationAction!"
));
}
diff --git a/crates/iceberg/src/transaction/upgrade_format_version.rs b/crates/iceberg/src/transaction/upgrade_format_version.rs
index bc0f79901..b833b5722 100644
--- a/crates/iceberg/src/transaction/upgrade_format_version.rs
+++ b/crates/iceberg/src/transaction/upgrade_format_version.rs
@@ -19,11 +19,12 @@ use std::sync::Arc;
use async_trait::async_trait;
+use crate::Result;
use crate::TableUpdate::UpgradeFormatVersion;
+use crate::error::invalid_data;
use crate::spec::FormatVersion;
use crate::table::Table;
use crate::transaction::action::{ActionCommit, TransactionAction};
-use crate::{Error, ErrorKind, Result};
/// A transaction action to upgrade a table's format version.
///
@@ -61,10 +62,7 @@ impl UpgradeFormatVersionAction {
impl TransactionAction for UpgradeFormatVersionAction {
async fn commit(self: Arc<Self>, _table: &Table) -> Result<ActionCommit> {
let format_version = self.format_version.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "FormatVersion is not set for UpgradeFormatVersionAction!",
- )
+ invalid_data!("FormatVersion is not set for UpgradeFormatVersionAction!")
})?;
Ok(ActionCommit::new(
diff --git a/crates/iceberg/src/transform/temporal.rs b/crates/iceberg/src/transform/temporal.rs
index e9831b542..04a8256f5 100644
--- a/crates/iceberg/src/transform/temporal.rs
+++ b/crates/iceberg/src/transform/temporal.rs
@@ -27,6 +27,7 @@ use arrow_schema::{DataType, TimeUnit};
use chrono::{DateTime, Datelike, Duration};
use super::TransformFunction;
+use crate::error::invalid_data;
use crate::spec::{Datum, PrimitiveLiteral, PrimitiveType};
use crate::{Error, ErrorKind, Result};
@@ -49,12 +50,7 @@ impl Year {
#[inline]
fn timestamp_to_year_micros(timestamp: i64) -> Result<i32> {
Ok(DateTime::from_timestamp_micros(timestamp)
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Fail to convert timestamp to date in year transform",
- )
- })?
+ .ok_or_else(|| invalid_data!("Fail to convert timestamp to date in year transform"))?
.year()
- UNIX_EPOCH_YEAR)
}
@@ -120,12 +116,8 @@ impl Month {
// unix epoch date: 1970-01-01
// if date > unix epoch date, delta month = (aa - 1) + 12 * (aaaa-1970)
// if date < unix epoch date, delta month = (12 - (aa - 1)) + 12 * (1970-aaaa-1)
- let date = DateTime::from_timestamp_micros(timestamp).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Fail to convert timestamp to date in month transform",
- )
- })?;
+ let date = DateTime::from_timestamp_micros(timestamp)
+ .ok_or_else(|| invalid_data!("Fail to convert timestamp to date in month transform"))?;
let unix_epoch_date = DateTime::from_timestamp_micros(0)
.expect("0 timestamp from unix epoch should be valid");
if date > unix_epoch_date {
@@ -228,10 +220,7 @@ impl Day {
};
let delta = Duration::new(secs, nanos).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to create 'TimeDelta' from seconds {secs} and nanos {nanos}"),
- )
+ invalid_data!("Failed to create 'TimeDelta' from seconds {secs} and nanos {nanos}")
})?;
let days = (delta.num_days() - offset) as i32;
@@ -254,10 +243,7 @@ impl Day {
};
let delta = Duration::new(secs, nanos).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to create 'TimeDelta' from seconds {secs} and nanos {nanos}"),
- )
+ invalid_data!("Failed to create 'TimeDelta' from seconds {secs} and nanos {nanos}")
})?;
let days = (delta.num_days() - offset) as i32;
diff --git a/crates/iceberg/src/transform/truncate.rs b/crates/iceberg/src/transform/truncate.rs
index f60f8e493..0f95854ee 100644
--- a/crates/iceberg/src/transform/truncate.rs
+++ b/crates/iceberg/src/transform/truncate.rs
@@ -22,6 +22,7 @@ use arrow_schema::DataType;
use super::TransformFunction;
use crate::Error;
+use crate::error::invalid_data;
use crate::spec::decimal_utils::decimal_from_i128_with_scale;
use crate::spec::{Datum, PrimitiveLiteral};
@@ -69,10 +70,7 @@ impl TransformFunction for Truncate {
match input.data_type() {
DataType::Int32 => {
let width: i32 = self.width.try_into().map_err(|_| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "width is failed to convert to i32 when truncate Int32Array",
- )
+ invalid_data!("width is failed to convert to i32 when truncate Int32Array")
})?;
let res: arrow_array::Int32Array = input
.as_any()
@@ -151,10 +149,7 @@ impl TransformFunction for Truncate {
match input.literal() {
PrimitiveLiteral::Int(v) => Ok(Some({
let width: i32 = self.width.try_into().map_err(|_| {
- Error::new(
- crate::ErrorKind::DataInvalid,
- "width is failed to convert to i32 when truncate Int32Array",
- )
+ invalid_data!("width is failed to convert to i32 when truncate Int32Array")
})?;
Datum::int(Self::truncate_i32(*v, width))
})),
diff --git a/crates/iceberg/src/writer/base_writer/data_file_writer.rs b/crates/iceberg/src/writer/base_writer/data_file_writer.rs
index 02dcda416..c7a988374 100644
--- a/crates/iceberg/src/writer/base_writer/data_file_writer.rs
+++ b/crates/iceberg/src/writer/base_writer/data_file_writer.rs
@@ -19,6 +19,7 @@
use arrow_array::RecordBatch;
+use crate::error::invalid_data;
use crate::spec::{DataContentType, DataFile, PartitionKey};
use crate::writer::file_writer::FileWriterBuilder;
use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
@@ -98,12 +99,8 @@ where
res.partition(pk.data().clone());
res.partition_spec_id(pk.spec().spec_id());
}
- res.build().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to build data file: {e}"),
- )
- })
+ res.build()
+ .map_err(|e| invalid_data!("Failed to build data file: {e}"))
})
.collect()
} else {
diff --git a/crates/iceberg/src/writer/base_writer/equality_delete_writer.rs b/crates/iceberg/src/writer/base_writer/equality_delete_writer.rs
index e961d4678..84c10b176 100644
--- a/crates/iceberg/src/writer/base_writer/equality_delete_writer.rs
+++ b/crates/iceberg/src/writer/base_writer/equality_delete_writer.rs
@@ -27,6 +27,7 @@ use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use crate::arrow::record_batch_projector::RecordBatchProjector;
use crate::arrow::schema_to_arrow_schema;
+use crate::error::invalid_data;
use crate::spec::{DataFile, PartitionKey, Schema, SchemaRef};
use crate::writer::file_writer::FileWriterBuilder;
use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
@@ -78,28 +79,21 @@ pub struct EqualityDeleteWriterConfig {
/// here produces clearer errors and avoids generating meaningless delete files.
fn validate_equality_ids(equality_ids: &[i32], original_schema: &Schema) -> Result<()> {
if equality_ids.is_empty() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "Equality delete field ids must not be empty.",
+ return Err(invalid_data!(
+ "Equality delete field ids must not be empty."
));
}
let mut seen = HashSet::with_capacity(equality_ids.len());
for id in equality_ids {
if !seen.insert(*id) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Duplicate equality delete field id: {id}"),
- )
- .with_context("field_id", id.to_string()));
+ return Err(invalid_data!("Duplicate equality delete field id: {id}")
+ .with_context("field_id", id.to_string()));
}
if original_schema.field_by_id(*id).is_none() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid equality delete field id: {id}"),
- )
- .with_context("field_id", id.to_string()));
+ return Err(invalid_data!("Invalid equality delete field id: {id}")
+ .with_context("field_id", id.to_string()));
}
}
@@ -218,12 +212,8 @@ where
res.partition(pk.data().clone());
res.partition_spec_id(pk.spec().spec_id());
}
- res.build().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to build data file: {e}"),
- )
- })
+ res.build()
+ .map_err(|e| invalid_data!("Failed to build data file: {e}"))
})
.collect()
} else {
diff --git a/crates/iceberg/src/writer/base_writer/position_delete_writer.rs b/crates/iceberg/src/writer/base_writer/position_delete_writer.rs
index ebe3e9588..29b65b74a 100644
--- a/crates/iceberg/src/writer/base_writer/position_delete_writer.rs
+++ b/crates/iceberg/src/writer/base_writer/position_delete_writer.rs
@@ -38,6 +38,7 @@ use arrow_schema::{DataType, Field};
use once_cell::sync::Lazy;
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
+use crate::error::invalid_data;
use crate::metadata_columns::{
RESERVED_FIELD_ID_DELETE_FILE_PATH, RESERVED_FIELD_ID_DELETE_FILE_POS, delete_file_path_field,
delete_file_pos_field,
@@ -95,22 +96,16 @@ fn field_id(field: &Field) -> Result<i32> {
.metadata()
.get(PARQUET_FIELD_ID_META_KEY)
.ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Position delete column `{}` is missing its Iceberg field id metadata.",
- field.name()
- ),
+ invalid_data!(
+ "Position delete column `{}` is missing its Iceberg field id metadata.",
+ field.name()
)
})?
.parse::<i32>()
.map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Position delete column `{}` has an invalid field id: {e}",
- field.name()
- ),
+ invalid_data!(
+ "Position delete column `{}` has an invalid field id: {e}",
+ field.name()
)
})
}
@@ -121,68 +116,51 @@ fn field_id(field: &Field) -> Result<i32> {
fn validate_position_delete_batch(batch: &RecordBatch) -> Result<()> {
let fields = batch.schema_ref().fields();
if fields.len() != 2 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "This writer supports only the two required position delete columns (`file_path`, `pos`); \
- batches with a different column count (e.g. including the optional `row` column) are not supported. Got {} columns.",
- fields.len()
- ),
+ return Err(invalid_data!(
+ "This writer supports only the two required position delete columns (`file_path`, `pos`); \
+ batches with a different column count (e.g. including the optional `row` column) are not supported. Got {} columns.",
+ fields.len()
));
}
let path = &fields[0];
let path_id = field_id(path)?;
if path_id != RESERVED_FIELD_ID_DELETE_FILE_PATH {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "The first position delete column must be `file_path` (field id {RESERVED_FIELD_ID_DELETE_FILE_PATH}), but got field id {path_id}."
- ),
+ return Err(invalid_data!(
+ "The first position delete column must be `file_path` (field id {RESERVED_FIELD_ID_DELETE_FILE_PATH}), but got field id {path_id}."
));
}
// The canonical schema maps Iceberg `string` to `Utf8` and the file writer is
// configured with it, so a `LargeUtf8` column has to be cast to `Utf8` first.
if path.data_type() != &DataType::Utf8 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "The position delete `file_path` column must be Utf8 (cast it first); got {:?}.",
- path.data_type()
- ),
+ return Err(invalid_data!(
+ "The position delete `file_path` column must be Utf8 (cast it first); got {:?}.",
+ path.data_type()
));
}
// Required column: a nullable field could write nulls under a required schema.
if path.is_nullable() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "The position delete `file_path` column must be required (non-nullable).",
+ return Err(invalid_data!(
+ "The position delete `file_path` column must be required (non-nullable)."
));
}
let pos = &fields[1];
let pos_id = field_id(pos)?;
if pos_id != RESERVED_FIELD_ID_DELETE_FILE_POS {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "The second position delete column must be `pos` (field id {RESERVED_FIELD_ID_DELETE_FILE_POS}), but got field id {pos_id}."
- ),
+ return Err(invalid_data!(
+ "The second position delete column must be `pos` (field id {RESERVED_FIELD_ID_DELETE_FILE_POS}), but got field id {pos_id}."
));
}
if pos.data_type() != &DataType::Int64 {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "The position delete `pos` column must be Int64, but got {:?}.",
- pos.data_type()
- ),
+ return Err(invalid_data!(
+ "The position delete `pos` column must be Int64, but got {:?}.",
+ pos.data_type()
));
}
if pos.is_nullable() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- "The position delete `pos` column must be required (non-nullable).",
+ return Err(invalid_data!(
+ "The position delete `pos` column must be required (non-nullable)."
));
}
@@ -282,12 +260,8 @@ where
res.partition(pk.data().clone());
res.partition_spec_id(pk.spec().spec_id());
}
- res.build().map_err(|e| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Failed to build position delete file: {e}"),
- )
- })
+ res.build()
+ .map_err(|e| invalid_data!("Failed to build position delete file: {e}"))
})
.collect()
} else {
diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs
index fbf333c7b..b874cb95a 100644
--- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs
+++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs
@@ -40,6 +40,7 @@ use crate::arrow::{
};
use crate::compression::CompressionCodec;
use crate::encryption::{EncryptionManager, StandardKeyMetadata};
+use crate::error::invalid_data;
use crate::io::{FileIO, FileWrite, OutputFile};
use crate::spec::{
DataContentType, DataFileBuilder, DataFileFormat, Datum, ListType, Literal, MapType,
@@ -155,11 +156,7 @@ fn parquet_compression(codec: CompressionCodec) -> Result<Compression> {
}
fn invalid_level_error(codec: &str, source: impl Into<anyhow::Error>) -> Error {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid {codec} compression level"),
- )
- .with_source(source)
+ invalid_data!("Invalid {codec} compression level").with_source(source)
}
impl FileWriterBuilder for ParquetWriterBuilder {
@@ -288,11 +285,8 @@ impl SchemaVisitor for IndexByParquetPathName {
let full_name = self.field_names.iter().map(String::as_str).join(".");
let field_id = self.field_id;
if let Some(existing_field_id) = self.name_to_id.get(full_name.as_str()) {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Invalid schema: multiple fields for name {full_name}: {field_id} and {existing_field_id}"
- ),
+ return Err(invalid_data!(
+ "Invalid schema: multiple fields for name {full_name}: {field_id} and {existing_field_id}"
));
} else {
self.name_to_id.insert(full_name, field_id);
@@ -426,12 +420,10 @@ impl ParquetWriter {
let reader = input_file.reader().await?;
let mut parquet_reader = ArrowFileReader::new(file_metadata, reader);
- let parquet_metadata = parquet_reader.get_metadata(None).await.map_err(|err| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Error reading Parquet metadata: {err}"),
- )
- })?;
+ let parquet_metadata = parquet_reader
+ .get_metadata(None)
+ .await
+ .map_err(|err| invalid_data!("Error reading Parquet metadata: {err}"))?;
let mut builder = ParquetWriter::parquet_to_data_file_builder(
table_metadata.current_schema().clone(),
parquet_metadata,
@@ -547,22 +539,19 @@ impl ParquetWriter {
upper_bounds.get(&field.source_id),
) {
if !field.transform.preserves_order() {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "cannot infer partition value for non linear partition field (needs to preserve order): {} with transform {}",
- field.name, field.transform
- ),
+ return Err(invalid_data!(
+ "cannot infer partition value for non linear partition field (needs to preserve order): {} with transform {}",
+ field.name,
+ field.transform
));
}
if lower != upper {
- return Err(Error::new(
- ErrorKind::DataInvalid,
- format!(
- "multiple partition values for field {}: lower: {:?}, upper: {:?}",
- field.name, lower, upper
- ),
+ return Err(invalid_data!(
+ "multiple partition values for field {}: lower: {:?}, upper: {:?}",
+ field.name,
+ lower,
+ upper
));
}