Add schema compatibility check to CDF, update cdf tests

Author: OussamaSaoudi-db

kernel/src/table_changes/log_replay.rs

@@ -79,9 +79,8 @@ pub(crate) fn table_changes_action_iter(
 ///       phase, so we must perform it ahead of time in phase 1.
 ///     - Ensure that reading is supported on any protocol updates.
 ///     - Ensure that Change Data Feed is enabled for any metadata update. See  [`TableProperties`]
-///     - Ensure that any schema update is compatible with the provided `schema`. Currently, schema
-///       compatibility is checked through schema equality. This will be expanded in the future to
-///       allow limited schema evolution.
+///     - Ensure that any schema update is compatible with the provided `schema`. To see how schema
+///       compatibility is defined, see [`StructType::can_read_as`].
 ///
 /// Note: We check the protocol, change data feed enablement, and schema compatibility in phase 1
 /// in order to detect errors and fail early.
@@ -181,11 +180,8 @@ impl LogReplayScanner {
             }
             if let Some((schema, configuration)) = visitor.metadata_info {
                 let schema: StructType = serde_json::from_str(&schema)?;
-                // Currently, schema compatibility is defined as having equal schema types. In the
-                // future, more permisive schema evolution will be supported.
-                // See: https://github.com/delta-io/delta-kernel-rs/issues/523
                 require!(
-                    table_schema.as_ref() == &schema,
+                    schema.can_read_as(table_schema).is_ok(),
                     Error::change_data_feed_incompatible_schema(table_schema, &schema)
                 );
                 let table_properties = TableProperties::from(configuration);

kernel/src/table_changes/log_replay/tests.rs

@@ -185,7 +185,7 @@ async fn column_mapping_should_fail() {
 // Note: This should be removed once type widening support is added for CDF
 #[tokio::test]
 async fn incompatible_schemas_fail() {
-    async fn assert_incompatible_schema(commit_schema: StructType, cdf_schema: StructType) {
+    async fn assert_schema_check(commit_schema: StructType, cdf_schema: StructType, passes: bool) {
         let engine = Arc::new(SyncEngine::new());
         let mut mock_table = LocalMockTable::new();
 
@@ -210,28 +210,32 @@ async fn incompatible_schemas_fail() {
                 .unwrap()
                 .try_collect();
 
-        assert!(matches!(
-            res,
-            Err(Error::ChangeDataFeedIncompatibleSchema(_, _))
-        ));
+        if passes {
+            assert!(res.is_ok());
+        } else {
+            assert!(matches!(
+                res,
+                Err(Error::ChangeDataFeedIncompatibleSchema(_, _))
+            ));
+        }
     }
 
     // The CDF schema has fields: `id: int` and `value: string`.
-    // This commit has schema with fields: `id: long`, `value: string` and `year: int` (nullable).
+    // This commit has schema with fields: `id: int`, `value: string` and `year: int` (nullable).
     let schema = StructType::new([
-        StructField::new("id", DataType::LONG, true),
+        StructField::new("id", DataType::INTEGER, true),
         StructField::new("value", DataType::STRING, true),
         StructField::new("year", DataType::INTEGER, true),
     ]);
-    assert_incompatible_schema(schema, get_schema()).await;
+    assert_schema_check(schema, get_schema(), false).await;
 
     // The CDF schema has fields: `id: int` and `value: string`.
     // This commit has schema with fields: `id: long` and `value: string`.
     let schema = StructType::new([
         StructField::new("id", DataType::LONG, true),
         StructField::new("value", DataType::STRING, true),
     ]);
-    assert_incompatible_schema(schema, get_schema()).await;
+    assert_schema_check(schema, get_schema(), false).await;
 
     // NOTE: Once type widening is supported, this should not return an error.
     //
@@ -245,31 +249,31 @@ async fn incompatible_schemas_fail() {
         StructField::new("id", DataType::INTEGER, true),
         StructField::new("value", DataType::STRING, true),
     ]);
-    assert_incompatible_schema(cdf_schema, commit_schema).await;
+    assert_schema_check(commit_schema, cdf_schema, false).await;
 
-    // Note: Once schema evolution is supported, this should not return an error.
-    //
-    // The CDF schema has fields: nullable `id`  and nullable `value`.
-    // This commit has schema with fields: non-nullable `id` and nullable `value`.
+    // The CDF schema has fields: `id: int` and `value: string`.
+    // This commit has schema with fields:`id: string` and `value: string`.
     let schema = StructType::new([
-        StructField::new("id", DataType::LONG, false),
+        StructField::new("id", DataType::STRING, true),
         StructField::new("value", DataType::STRING, true),
     ]);
-    assert_incompatible_schema(schema, get_schema()).await;
+    assert_schema_check(schema, get_schema(), false).await;
 
-    // The CDF schema has fields: `id: int` and `value: string`.
-    // This commit has schema with fields:`id: string` and `value: string`.
+    // Note: Once schema evolution is supported, this should not return an error.
+    //
+    // The CDF schema has fields: nullable `id`  and nullable `value`.
+    // This commit has schema with fields: non-nullable `id` and nullable `value`.
     let schema = StructType::new([
-        StructField::new("id", DataType::STRING, true),
+        StructField::new("id", DataType::INTEGER, false),
         StructField::new("value", DataType::STRING, true),
     ]);
-    assert_incompatible_schema(schema, get_schema()).await;
+    assert_schema_check(schema, get_schema(), true).await;
 
     // Note: Once schema evolution is supported, this should not return an error.
     // The CDF schema has fields: `id` (nullable) and `value` (nullable).
     // This commit has schema with fields: `id` (nullable).
     let schema = get_schema().project_as_struct(&["id"]).unwrap();
-    assert_incompatible_schema(schema, get_schema()).await;
+    assert_schema_check(schema, get_schema(), true).await;
 }
 
 #[tokio::test]

kernel/src/table_changes/mod.rs

@@ -165,11 +165,9 @@ 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.
-        // Note: Schema compatibility check will be changed in the future to be more flexible.
-        // See issue [#523](https://github.com/delta-io/delta-kernel-rs/issues/523)
-        if start_snapshot.schema() != end_snapshot.schema() {
+        if let Err(err) = start_snapshot.schema().can_read_as(end_snapshot.schema()) {
             return Err(Error::generic(format!(
-                "Failed to build TableChanges: Start and end version schemas are different. Found start version schema {:?} and end version schema {:?}", start_snapshot.schema(), end_snapshot.schema(),
+                "Failed to build TableChanges: {}\n Found start version schema {:?} and end version schema {:?}", err, start_snapshot.schema(), end_snapshot.schema(),
             )));
         }
 
@@ -303,7 +301,7 @@ mod tests {
         let path = "./tests/data/table-with-cdf";
         let engine = Box::new(SyncEngine::new());
         let table = Table::try_from_uri(path).unwrap();
-        let expected_msg = "Failed to build TableChanges: Start and end version schemas are different. Found start version schema StructType { type_name: \"struct\", fields: {\"part\": StructField { name: \"part\", data_type: Primitive(Integer), nullable: true, metadata: {} }, \"id\": StructField { name: \"id\", data_type: Primitive(Integer), nullable: true, metadata: {} }} } and end version schema StructType { type_name: \"struct\", fields: {\"part\": StructField { name: \"part\", data_type: Primitive(Integer), nullable: true, metadata: {} }, \"id\": StructField { name: \"id\", data_type: Primitive(Integer), nullable: false, metadata: {} }} }";
+        let expected_msg = "Failed to build TableChanges: Generic delta kernel error: Read field is non-nullable while this field is nullable\n Found start version schema StructType { type_name: \"struct\", fields: {\"part\": StructField { name: \"part\", data_type: Primitive(Integer), nullable: true, metadata: {} }, \"id\": StructField { name: \"id\", data_type: Primitive(Integer), nullable: true, metadata: {} }} } and end version schema StructType { type_name: \"struct\", fields: {\"part\": StructField { name: \"part\", data_type: Primitive(Integer), nullable: true, metadata: {} }, \"id\": StructField { name: \"id\", data_type: Primitive(Integer), nullable: false, metadata: {} }} }";
 
         // A field in the schema goes from being nullable to non-nullable
         let table_changes_res = table.table_changes(engine.as_ref(), 3, 4);