Change error type for change data feed

Author: OussamaSaoudi

kernel/src/error.rs

@@ -6,7 +6,7 @@ use std::{
     str::Utf8Error,
 };
 
-use crate::schema::{DataType, StructType};
+use crate::schema::{compare::SchemaComparisonError, DataType};
 use crate::table_properties::ParseIntervalError;
 use crate::Version;
 
@@ -189,8 +189,8 @@ pub enum Error {
     #[error("Change data feed is unsupported for the table at version {0}")]
     ChangeDataFeedUnsupported(Version),
 
-    #[error("Change data feed encountered incompatible schema. Expected {0}, got {1}")]
-    ChangeDataFeedIncompatibleSchema(String, String),
+    #[error("Change data feed encountered incompatible schema at version {0}: {1}")]
+    ChangeDataFeedIncompatibleSchema(Version, #[source] SchemaComparisonError),
 
     /// Invalid checkpoint files
     #[error("Invalid Checkpoint: {0}")]
@@ -262,10 +262,10 @@ impl Error {
         Self::ChangeDataFeedUnsupported(version.into())
     }
     pub(crate) fn change_data_feed_incompatible_schema(
-        expected: &StructType,
-        actual: &StructType,
+        version: Version,
+        schema_error: SchemaComparisonError,
     ) -> Self {
-        Self::ChangeDataFeedIncompatibleSchema(format!("{expected:?}"), format!("{actual:?}"))
+        Self::ChangeDataFeedIncompatibleSchema(version, schema_error)
     }
 
     pub fn invalid_checkpoint(msg: impl ToString) -> Self {

kernel/src/schema/compare.rs

@@ -33,7 +33,7 @@ pub(crate) struct Nullable(bool);
 
 /// Represents the ways a schema comparison can fail.
 #[derive(Debug, thiserror::Error)]
-pub(crate) enum Error {
+pub enum SchemaComparisonError {
     #[error("The nullability was tightened for a field")]
     NullabilityTightening,
     #[error("Field names do not match")]
@@ -48,8 +48,8 @@ pub(crate) enum Error {
     TypeMismatch,
 }
 
-/// A [`std::result::Result`] that has the schema comparison [`Error`] as the error variant.
-pub(crate) type SchemaComparisonResult = Result<(), Error>;
+/// A [`std::result::Result`] that has the schema comparison [`SchemaComparisonError`] as the error variant.
+pub(crate) type SchemaComparisonResult = Result<(), SchemaComparisonError>;
 
 /// Represents a schema compatibility check for the type. If `self` can be read as `read_type`,
 /// this function returns `Ok(())`. Otherwise, this function returns `Err`.
@@ -65,7 +65,10 @@ impl SchemaComparison for Nullable {
         // column as non-nullable. So we avoid the case where !read_nullable && nullable
         // Hence we check that !(!read_nullable && existing_nullable)
         // == read_nullable || !existing_nullable
-        require!(read_nullable.0 || !self.0, Error::NullabilityTightening);
+        require!(
+            read_nullable.0 || !self.0,
+            SchemaComparisonError::NullabilityTightening
+        );
         Ok(())
     }
 }
@@ -78,7 +81,10 @@ impl SchemaComparison for StructField {
     ///     3. You can read this data type as the `read_field`'s data type.
     fn can_read_as(&self, read_field: &Self) -> SchemaComparisonResult {
         Nullable(self.nullable).can_read_as(&Nullable(read_field.nullable))?;
-        require!(self.name() == read_field.name(), Error::FieldNameMismatch);
+        require!(
+            self.name() == read_field.name(),
+            SchemaComparisonError::FieldNameMismatch
+        );
         self.data_type().can_read_as(read_field.data_type())?;
         Ok(())
     }
@@ -100,30 +106,33 @@ impl SchemaComparison for StructType {
             .collect();
         require!(
             lowercase_field_map.len() == self.fields.len(),
-            Error::InvalidSchema
+            SchemaComparisonError::InvalidSchema
         );
 
         let lowercase_read_field_names: HashSet<String> =
             read_type.fields.keys().map(|x| x.to_lowercase()).collect();
         require!(
             lowercase_read_field_names.len() == read_type.fields.len(),
-            Error::InvalidSchema
+            SchemaComparisonError::InvalidSchema
         );
 
         // Check that the field names are a subset of the read fields.
         if lowercase_field_map
             .keys()
             .any(|name| !lowercase_read_field_names.contains(name))
         {
-            return Err(Error::MissingColumn);
+            return Err(SchemaComparisonError::MissingColumn);
         }
         for read_field in read_type.fields() {
             match lowercase_field_map.get(&read_field.name().to_lowercase()) {
                 Some(existing_field) => existing_field.can_read_as(read_field)?,
                 None => {
                     // Note: Delta spark does not perform the following check. Hence it ignores
                     // non-null fields that exist in the read schema that aren't in this schema.
-                    require!(read_field.is_nullable(), Error::NewNonNullableColumn);
+                    require!(
+                        read_field.is_nullable(),
+                        SchemaComparisonError::NewNonNullableColumn
+                    );
                 }
             }
         }
@@ -161,7 +170,7 @@ impl SchemaComparison for DataType {
             (a, b) => {
                 // TODO: In the future, we will change this to support type widening.
                 // See: #623
-                require!(a == b, Error::TypeMismatch);
+                require!(a == b, SchemaComparisonError::TypeMismatch);
             }
         };
         Ok(())
@@ -170,7 +179,7 @@ impl SchemaComparison for DataType {
 
 #[cfg(test)]
 mod tests {
-    use crate::schema::compare::{Error, SchemaComparison};
+    use crate::schema::compare::{SchemaComparison, SchemaComparisonError};
     use crate::schema::{ArrayType, DataType, MapType, StructField, StructType};
 
     #[test]
@@ -252,7 +261,7 @@ mod tests {
 
         assert!(matches!(
             existing_schema.can_read_as(&read_schema),
-            Err(Error::NullabilityTightening)
+            Err(SchemaComparisonError::NullabilityTightening)
         ));
     }
     #[test]
@@ -270,7 +279,7 @@ mod tests {
         ]);
         assert!(matches!(
             existing_schema.can_read_as(&read_schema),
-            Err(Error::FieldNameMismatch)
+            Err(SchemaComparisonError::FieldNameMismatch)
         ));
     }
     #[test]
@@ -287,7 +296,7 @@ mod tests {
         ]);
         assert!(matches!(
             existing_schema.can_read_as(&read_schema),
-            Err(Error::TypeMismatch)
+            Err(SchemaComparisonError::TypeMismatch)
         ));
     }
     #[test]
@@ -318,7 +327,7 @@ mod tests {
         ]);
         assert!(matches!(
             existing_schema.can_read_as(&read_schema),
-            Err(Error::NullabilityTightening)
+            Err(SchemaComparisonError::NullabilityTightening)
         ));
     }
     #[test]
@@ -340,7 +349,10 @@ mod tests {
         assert!(a.can_read_as(&b).is_ok());
 
         // Read `b` as `a`. `a` is missing a column that is present in `b`.
-        assert!(matches!(b.can_read_as(&a), Err(Error::MissingColumn)));
+        assert!(matches!(
+            b.can_read_as(&a),
+            Err(SchemaComparisonError::MissingColumn)
+        ));
     }
     #[test]
     fn differ_by_non_nullable_column() {
@@ -360,11 +372,14 @@ mod tests {
         // Read `a` as `b`. `b` has an extra non-nullable column.
         assert!(matches!(
             a.can_read_as(&b),
-            Err(Error::NewNonNullableColumn)
+            Err(SchemaComparisonError::NewNonNullableColumn)
         ));
 
         // Read `b` as `a`. `a` is missing a column that is present in `b`.
-        assert!(matches!(b.can_read_as(&a), Err(Error::MissingColumn)));
+        assert!(matches!(
+            b.can_read_as(&a),
+            Err(SchemaComparisonError::MissingColumn)
+        ));
     }
 
     #[test]
@@ -384,13 +399,13 @@ mod tests {
         ]);
         assert!(matches!(
             existing_schema.can_read_as(&read_schema),
-            Err(Error::InvalidSchema)
+            Err(SchemaComparisonError::InvalidSchema)
         ));
 
         // Checks in the inverse order
         assert!(matches!(
             read_schema.can_read_as(&existing_schema),
-            Err(Error::InvalidSchema)
+            Err(SchemaComparisonError::InvalidSchema)
         ));
     }
 }

kernel/src/table_changes/log_replay.rs

@@ -181,10 +181,10 @@ impl LogReplayScanner {
             }
             if let Some((schema, configuration)) = visitor.metadata_info {
                 let schema: StructType = serde_json::from_str(&schema)?;
-                require!(
-                    schema.can_read_as(table_schema).is_ok(),
-                    Error::change_data_feed_incompatible_schema(table_schema, &schema)
-                );
+
+                schema.can_read_as(table_schema).map_err(|err| {
+                    Error::change_data_feed_incompatible_schema(commit_file.version, err)
+                })?;
                 let table_properties = TableProperties::from(configuration);
                 check_cdf_table_properties(&table_properties)
                     .map_err(|_| Error::change_data_feed_unsupported(commit_file.version))?;

kernel/src/table_changes/mod.rs

@@ -174,11 +174,11 @@ impl TableChanges {
 
         // Verify that the start and end schemas are compatible. We must still check schema
         // compatibility for each schema update in the CDF range.
-        if let Err(err) = start_snapshot.schema().can_read_as(end_snapshot.schema()) {
-            return Err(Error::generic(format!(
+        start_snapshot.schema().can_read_as(end_snapshot.schema()).map_err(|err|{
+            Error::generic(format!(
                 "Failed to build TableChanges: {}\n Found start version schema {:?} and end version schema {:?}", err, start_snapshot.schema(), end_snapshot.schema(),
-            )));
-        }
+            ))
+        })?;
 
         let schema = StructType::new(
             end_snapshot