Merge branch 'main' into fix/name-validation

Author: zfarrell

README.md

@@ -4,6 +4,8 @@ A [DataFusion](https://datafusion.apache.org/) extension for querying [DuckLake]
 
 The goal of this project is to make DuckLake a first-class, Arrow-native lakehouse format inside DataFusion.
 
+[Join the Discord](https://discord.gg/FefVb3u9sH)
+
 ---
 
 ## Currently Supported

src/metadata_writer.rs

@@ -140,7 +140,18 @@ pub struct DataFileInfo {
 
 impl DataFileInfo {
     /// Create a new data file info with relative path.
+    ///
+    /// # Panics
+    ///
+    /// Panics if `record_count` is negative. Record counts originate from
+    /// `RecordBatch::num_rows()` (always non-negative), so a negative value
+    /// indicates a programming error.
     pub fn new(path: impl Into<String>, file_size_bytes: i64, record_count: i64) -> Self {
+        assert!(
+            record_count >= 0,
+            "record_count must be non-negative, got {}",
+            record_count
+        );
         Self {
             path: path.into(),
             path_is_relative: true,
@@ -443,4 +454,16 @@ mod tests {
             other => panic!("Expected InvalidConfig, got {:?}", other),
         }
     }
+
+    #[test]
+    fn test_data_file_info_zero_record_count() {
+        let file = DataFileInfo::new("empty.parquet", 0, 0);
+        assert_eq!(file.record_count, 0);
+    }
+
+    #[test]
+    #[should_panic(expected = "record_count must be non-negative")]
+    fn test_data_file_info_negative_record_count_panics() {
+        DataFileInfo::new("test.parquet", 1024, -1);
+    }
 }

src/metadata_writer_sqlite.rs

@@ -441,8 +441,8 @@ impl MetadataWriter for SqliteMetadataWriter {
                     if let Some((existing_type, _existing_nullable)) =
                         existing_map.get(new_col.name.as_str())
                     {
-                        // Column exists - check type matches
-                        if *existing_type != new_col.ducklake_type {
+                        // Column exists - check type compatibility (normalize aliases + allow promotions)
+                        if !crate::types::types_compatible(existing_type, &new_col.ducklake_type) {
                             return Err(crate::error::DuckLakeError::InvalidConfig(format!(
                                 "Schema evolution error: column '{}' has type '{}' in existing table but '{}' in new schema. Type changes are not allowed.",
                                 new_col.name, existing_type, new_col.ducklake_type

src/table.rs

@@ -63,6 +63,20 @@ pub(crate) fn validated_file_size(file_size_bytes: i64, file_path: &str) -> Data
     })
 }
 
+/// Validate and convert record_count from i64 (as stored in DuckLake metadata) to u64.
+///
+/// DuckLake stores record counts as signed integers in SQL. A negative value indicates
+/// corrupt or invalid metadata. Without this check, a negative record_count would cause
+/// incorrect behavior (e.g., empty ranges in full-file deletes, or incorrect row filtering).
+pub(crate) fn validated_record_count(record_count: i64, file_path: &str) -> DataFusionResult<u64> {
+    u64::try_from(record_count).map_err(|_| {
+        DataFusionError::Execution(format!(
+            "Invalid record_count ({}) for file '{}': value must be non-negative",
+            record_count, file_path
+        ))
+    })
+}
+
 /// Returns the expected schema for DuckLake delete files
 ///
 /// Delete files have a standard schema: (file_path: VARCHAR, pos: INT64)
@@ -731,4 +745,43 @@ mod tests {
         assert!(msg.contains("bad.parquet"));
         assert!(msg.contains(&i64::MIN.to_string()));
     }
+
+    #[test]
+    fn test_validated_record_count_positive() {
+        assert_eq!(validated_record_count(0, "test.parquet").unwrap(), 0);
+        assert_eq!(validated_record_count(100, "test.parquet").unwrap(), 100);
+        assert_eq!(
+            validated_record_count(i64::MAX, "test.parquet").unwrap(),
+            i64::MAX as u64
+        );
+    }
+
+    #[test]
+    fn test_validated_record_count_negative() {
+        let err = validated_record_count(-1, "data/test.parquet").unwrap_err();
+        let msg = err.to_string();
+        assert!(
+            msg.contains("-1"),
+            "Error should contain the negative value: {}",
+            msg
+        );
+        assert!(
+            msg.contains("data/test.parquet"),
+            "Error should contain the file path: {}",
+            msg
+        );
+        assert!(
+            msg.contains("record_count"),
+            "Error should mention record_count: {}",
+            msg
+        );
+    }
+
+    #[test]
+    fn test_validated_record_count_large_negative() {
+        let err = validated_record_count(i64::MIN, "bad.parquet").unwrap_err();
+        let msg = err.to_string();
+        assert!(msg.contains("bad.parquet"));
+        assert!(msg.contains(&i64::MIN.to_string()));
+    }
 }

src/table_deletions.rs

@@ -36,7 +36,7 @@ use futures::Stream;
 
 use crate::metadata_provider::{DeleteFileChange, MetadataProvider};
 use crate::path_resolver::resolve_path;
-use crate::table::validated_file_size;
+use crate::table::{validated_file_size, validated_record_count};
 
 /// Delete file schema: (file_path: VARCHAR, pos: INT64)
 fn delete_file_schema() -> SchemaRef {
@@ -143,6 +143,10 @@ impl TableDeletionsTable {
             delete_file.data_file_footer_size,
         )?;
 
+        // Validate record_count before use — a negative value from corrupt metadata
+        // would cause incorrect behavior (e.g., empty ranges in full-file deletes).
+        validated_record_count(delete_file.data_record_count, &delete_file.data_file_path)?;
+
         Ok(Arc::new(DeletedRowsExec::new(
             current_delete_exec,
             previous_delete_exec,