feat: add enforcement for invariants to commit checker

crates/deltalake-core/src/kernel/actions/types.rs

@@ -274,6 +274,26 @@ pub enum WriterFeatures {
     Other(String),
 }
 
+#[allow(clippy::from_over_into)]
+impl Into<usize> for WriterFeatures {
+    fn into(self) -> usize {
+        match self {
+            WriterFeatures::Other(_) => 0,
+            WriterFeatures::AppendOnly | WriterFeatures::Invariants => 2,
+            WriterFeatures::CheckConstraints => 3,
+            WriterFeatures::ChangeDataFeed | WriterFeatures::GeneratedColumns => 4,
+            WriterFeatures::ColumnMapping => 5,
+            WriterFeatures::IdentityColumns
+            | WriterFeatures::DeletionVectors
+            | WriterFeatures::RowTracking
+            | WriterFeatures::TimestampWithoutTimezone
+            | WriterFeatures::DomainMetadata
+            | WriterFeatures::V2Checkpoint
+            | WriterFeatures::IcebergCompatV1 => 7,
+        }
+    }
+}
+
 impl From<String> for WriterFeatures {
     fn from(value: String) -> Self {
         value.as_str().into()

crates/deltalake-core/src/operations/transaction/protocol.rs

@@ -142,6 +142,25 @@ impl ProtocolChecker {
             })?;
         }
 
+        // Does the table have invariants?
+        let has_invariants = !snapshot
+            .current_metadata()
+            .and_then(|meta| meta.schema.get_invariants().ok())
+            .unwrap_or_default()
+            .is_empty();
+
+        // The table is at least version 7, has actual invariants and has the proper writer feature
+        // enabled
+        if snapshot.min_writer_version() >= 7
+            && has_invariants
+            && !snapshot
+                .writer_features()
+                .unwrap()
+                .contains(&WriterFeatures::Invariants)
+        {
+            return Err(TransactionError::WriterFeaturesRequired);
+        }
+
         Ok(())
     }
 }