Skip to content

Introduce invalid data error macro #2928

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
xanderbailey wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Introduce invalid data error macro #2928

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 8 additions & 13 deletions crates/iceberg/src/arrow/caching_delete_file_loader.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ use crate::arrow::delete_file_loader::BasicDeleteFileLoader;
use crate::arrow::scan_metrics::ScanMetrics;
use crate::arrow::{arrow_primitive_to_literal, arrow_schema_to_schema};
use crate::delete_vector::DeleteVector;
use crate::error::invalid_data;
use crate::expr::Predicate::AlwaysTrue;
use crate::expr::{Predicate, Reference};
use crate::io::FileIO;
Expand Down Expand Up @@ -348,29 +349,23 @@ 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"
));
};

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}"
));
}

Expand Down
20 changes: 6 additions & 14 deletions crates/iceberg/src/arrow/partition_value_calculator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
///
Expand Down Expand Up @@ -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"
));
}

Expand Down Expand Up @@ -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"));
}
};

Expand All @@ -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))
}
Expand Down
19 changes: 6 additions & 13 deletions crates/iceberg/src/arrow/reader/predicate_visitor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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
));
}

Expand All @@ -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 {
Expand Down
9 changes: 3 additions & 6 deletions crates/iceberg/src/arrow/reader/projection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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:?}"
));
}
};
Expand Down
27 changes: 9 additions & 18 deletions crates/iceberg/src/arrow/record_batch_partition_splitter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,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";
Expand Down Expand Up @@ -128,9 +129,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"
))
}
})
Expand All @@ -140,23 +140,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)?;
Expand All @@ -167,9 +159,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"
))
}
})
Expand Down
13 changes: 5 additions & 8 deletions crates/iceberg/src/arrow/record_batch_projector.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};

Expand Down Expand Up @@ -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".to_string())
.with_context("value", value)
.with_source(e)
})?;
Ok(Some(field_id as i64))
} else {
Expand Down Expand Up @@ -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
Expand Down
26 changes: 9 additions & 17 deletions crates/iceberg/src/arrow/record_batch_transformer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ use parquet::arrow::PARQUET_FIELD_ID_META_KEY;

use crate::arrow::value::{create_primitive_array_repeated, create_primitive_array_single_element};
use crate::arrow::{datum_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,
};
Expand Down Expand Up @@ -249,13 +250,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 {
Expand Down Expand Up @@ -740,10 +738,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| {
Expand Down Expand Up @@ -772,12 +767,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));
Expand Down
Loading
Loading