feat: check schema capability when adding parquet files

crates/iceberg/src/arrow/mod.rs

@@ -28,6 +28,7 @@ pub mod delete_file_loader;
 pub(crate) mod delete_filter;
 
 mod reader;
+pub(crate) use reader::apply_name_mapping_to_arrow_schema;
 /// RecordBatch projection utilities
 pub mod record_batch_projector;
 pub(crate) mod record_batch_transformer;

crates/iceberg/src/arrow/reader.rs

@@ -1025,7 +1025,7 @@ fn build_fallback_field_id_map(parquet_schema: &SchemaDescriptor) -> HashMap<i32
 ///
 /// # Returns
 /// Arrow schema with field IDs assigned based on name mapping
-fn apply_name_mapping_to_arrow_schema(
+pub(crate) fn apply_name_mapping_to_arrow_schema(
     arrow_schema: ArrowSchemaRef,
     name_mapping: &NameMapping,
 ) -> Result<Arc<ArrowSchema>> {

crates/iceberg/src/spec/datatypes.rs

@@ -250,6 +250,34 @@ pub enum PrimitiveType {
 }
 
 impl PrimitiveType {
+    /// Returns `true` if `self` can be promoted to `target` per the Iceberg spec's
+    /// valid type promotion rules.
+    ///
+    /// See <https://iceberg.apache.org/spec/#schema-evolution>
+    pub fn is_promotable_to(&self, target: &PrimitiveType) -> bool {
+        matches!(
+            (self, target),
+            (PrimitiveType::Int, PrimitiveType::Long)
+                | (PrimitiveType::Float, PrimitiveType::Double)
+                | (PrimitiveType::Date, PrimitiveType::Timestamp)
+                | (PrimitiveType::Date, PrimitiveType::TimestampNs)
+                | (PrimitiveType::String, PrimitiveType::Binary)
+                | (PrimitiveType::Binary, PrimitiveType::String)
+        ) || matches!(
+            (self, target),
+            (
+                PrimitiveType::Decimal {
+                    precision: p1,
+                    scale: s1,
+                },
+                PrimitiveType::Decimal {
+                    precision: p2,
+                    scale: s2,
+                },
+            ) if s1 == s2 && p1 < p2
+        )
+    }
+
     /// Check whether literal is compatible with the type.
     pub fn compatible(&self, literal: &PrimitiveLiteral) -> bool {
         matches!(
@@ -1260,4 +1288,48 @@ mod tests {
                 .contains("expected type 'struct'")
         );
     }
+
+    #[test]
+    fn test_is_promotable_to() {
+        assert!(PrimitiveType::Int.is_promotable_to(&PrimitiveType::Long));
+        assert!(PrimitiveType::Float.is_promotable_to(&PrimitiveType::Double));
+        assert!(PrimitiveType::Date.is_promotable_to(&PrimitiveType::Timestamp));
+        assert!(PrimitiveType::Date.is_promotable_to(&PrimitiveType::TimestampNs));
+        assert!(PrimitiveType::String.is_promotable_to(&PrimitiveType::Binary));
+        assert!(PrimitiveType::Binary.is_promotable_to(&PrimitiveType::String));
+        assert!(
+            PrimitiveType::Decimal {
+                precision: 10,
+                scale: 2,
+            }
+            .is_promotable_to(&PrimitiveType::Decimal {
+                precision: 20,
+                scale: 2,
+            })
+        );
+
+        assert!(!PrimitiveType::Long.is_promotable_to(&PrimitiveType::Int));
+        assert!(!PrimitiveType::Double.is_promotable_to(&PrimitiveType::Float));
+        assert!(!PrimitiveType::Int.is_promotable_to(&PrimitiveType::String));
+        assert!(
+            !PrimitiveType::Decimal {
+                precision: 10,
+                scale: 2,
+            }
+            .is_promotable_to(&PrimitiveType::Decimal {
+                precision: 20,
+                scale: 3,
+            })
+        );
+        assert!(
+            !PrimitiveType::Decimal {
+                precision: 20,
+                scale: 2,
+            }
+            .is_promotable_to(&PrimitiveType::Decimal {
+                precision: 10,
+                scale: 2,
+            })
+        );
+    }
 }