Merge remote-tracking branch 'upstream/main' into feat/metrics

Author: gotocoding-DB

ffi/examples/read-table/CMakeLists.txt

@@ -16,6 +16,7 @@ set(DatPath "../../../../acceptance/tests/dat/out/reader_tests/generated")
 set(ExpectedPath "../../../tests/read-table-testing/expected-data")
 set(KernelTestPath "../../../../kernel/tests/data")
 add_test(NAME read_and_print_all_prim COMMAND ${TestRunner} ${DatPath}/all_primitive_types/delta/ ${ExpectedPath}/all-prim-types.expected)
+add_test(NAME read_and_print_nested COMMAND ${TestRunner} ${DatPath}/nested_types/delta/ ${ExpectedPath}/nested-types.expected)
 add_test(NAME read_and_print_basic_partitioned COMMAND ${TestRunner} ${DatPath}/basic_partitioned/delta/ ${ExpectedPath}/basic-partitioned.expected)
 add_test(NAME read_and_print_with_dv_small COMMAND ${TestRunner} ${KernelTestPath}/table-with-dv-small/ ${ExpectedPath}/table-with-dv-small.expected)
 

ffi/examples/read-table/schema.h

@@ -148,8 +148,10 @@ void visit_array(
 {
   SchemaBuilder* builder = data;
   char* name_ptr = malloc(sizeof(char) * (name.len + 22));
-  snprintf(name_ptr, name.len + 1, "%s", name.ptr);
-  snprintf(name_ptr + name.len, 22, " (is nullable: %s)", is_nullable ? "true" : "false");
+  // NOTE: we truncate to the max int size because the format specifier "%.*s" requires an int length specifier
+  int name_chars = name.len > INT_MAX ? INT_MAX : (int)name.len; // handle _REALLY_ long names by truncation
+  int wrote = snprintf(name_ptr, name.len + 1, "%.*s", name_chars, name.ptr);
+  snprintf(name_ptr + wrote, 22, " (is nullable: %s)", is_nullable ? "true" : "false");
   print_physical_name(name_ptr, metadata);
   PRINT_CHILD_VISIT("array", name_ptr, sibling_list_id, "Types", child_list_id);
   SchemaItem* array_item = add_to_list(&builder->lists[sibling_list_id], name_ptr, "array", is_nullable);
@@ -166,8 +168,10 @@ void visit_map(
 {
   SchemaBuilder* builder = data;
   char* name_ptr = malloc(sizeof(char) * (name.len + 22));
-  snprintf(name_ptr, name.len + 1, "%s", name.ptr);
-  snprintf(name_ptr + name.len, 22, " (is nullable: %s)", is_nullable ? "true" : "false");
+  // NOTE: we truncate to the max int size because the format specifier "%.*s" requires an int length specifier
+  int name_chars = name.len > INT_MAX ? INT_MAX : (int)name.len; // handle _REALLY_ long names by truncation
+  int wrote = snprintf(name_ptr, name.len + 1, "%.*s", name_chars, name.ptr);
+  snprintf(name_ptr + wrote, 22, " (is nullable: %s)", is_nullable ? "true" : "false");
   print_physical_name(name_ptr, metadata);
   PRINT_CHILD_VISIT("map", name_ptr, sibling_list_id, "Types", child_list_id);
   SchemaItem* map_item = add_to_list(&builder->lists[sibling_list_id], name_ptr, "map", is_nullable);

ffi/src/expressions/engine_visitor.rs

@@ -317,7 +317,6 @@ fn visit_expression_array(
     array: &ArrayData,
     sibling_list_id: usize,
 ) {
-    #[allow(deprecated)]
     let elements = array.array_elements();
     let child_list_id = call!(visitor, make_field_list, elements.len());
     for scalar in elements {

ffi/tests/read-table-testing/expected-data/nested-types.expected

@@ -0,0 +1,115 @@
+Reading table at ../../../../acceptance/tests/dat/out/reader_tests/generated/nested_types/delta/
+version: 0
+
+Schema:
+├─ pk: integer
+├─ struct: struct
+│  ├─ float64: double
+│  └─ bool: boolean
+├─ array (is nullable: true): array
+│  └─ array_element: short
+└─ map (is nullable: true): map
+   ├─ map_key: string
+   └─ map_value: integer
+
+pk:  [
+  0,
+  1,
+  2,
+  3,
+  4
+]
+struct:  -- is_valid: all not null
+-- child 0 type: double
+  [
+    0,
+    1,
+    2,
+    3,
+    4
+  ]
+-- child 1 type: bool
+  [
+    true,
+    false,
+    true,
+    false,
+    true
+  ]
+array:  [
+  [
+    0
+  ],
+  [
+    0,
+    1
+  ],
+  [
+    0,
+    1,
+    2
+  ],
+  [
+    0,
+    1,
+    2,
+    3
+  ],
+  [
+    0,
+    1,
+    2,
+    3,
+    4
+  ]
+]
+map:  [
+  keys:
+  []
+  values:
+  [],
+  keys:
+  [
+    "0"
+  ]
+  values:
+  [
+    0
+  ],
+  keys:
+  [
+    "0",
+    "1"
+  ]
+  values:
+  [
+    0,
+    1
+  ],
+  keys:
+  [
+    "0",
+    "1",
+    "2"
+  ]
+  values:
+  [
+    0,
+    1,
+    2
+  ],
+  keys:
+  [
+    "0",
+    "1",
+    "2",
+    "3"
+  ]
+  values:
+  [
+    0,
+    1,
+    2,
+    3
+  ]
+]

kernel/src/actions/mod.rs

@@ -590,11 +590,6 @@ impl Protocol {
     /// Check if writing to a table with this protocol is supported. That is: does the kernel
     /// support the specified protocol writer version and all enabled writer features?
     pub(crate) fn ensure_write_supported(&self) -> DeltaResult<()> {
-        #[cfg(feature = "catalog-managed")]
-        require!(
-            !self.is_catalog_managed(),
-            Error::unsupported("Writes are not yet supported for catalog-managed tables")
-        );
         match &self.writer_features {
             Some(writer_features) if self.min_writer_version == 7 => {
                 // if we're on version 7, make sure we support all the specified features
@@ -1610,7 +1605,7 @@ mod tests {
         .unwrap();
         assert_result_error_with_message(
             protocol.ensure_write_supported(),
-            r#"Unsupported: Found unsupported TableFeatures: "identityColumns". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#,
+            r#"Unsupported: Found unsupported TableFeatures: "identityColumns". Supported TableFeatures: "changeDataFeed", "appendOnly", "catalogManaged", "catalogOwned-preview", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck", "variantType", "variantType-preview", "variantShredding-preview""#,
         );
 
         // Unknown writer features are allowed during creation for forward compatibility,
@@ -1624,7 +1619,7 @@ mod tests {
         .unwrap();
         assert_result_error_with_message(
             protocol.ensure_write_supported(),
-            r#"Unsupported: Found unsupported TableFeatures: "unsupported writer". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#,
+            r#"Unsupported: Found unsupported TableFeatures: "unsupported writer". Supported TableFeatures: "changeDataFeed", "appendOnly", "catalogManaged", "catalogOwned-preview", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck", "variantType", "variantType-preview", "variantShredding-preview""#,
         );
     }
 
@@ -1680,24 +1675,25 @@ mod tests {
         assert_eq!(parse_features::<TableFeature>(features), expected);
     }
 
+    #[cfg(feature = "catalog-managed")]
     #[test]
-    fn test_no_catalog_managed_writes() {
+    fn test_catalog_managed_writes() {
         let protocol = Protocol::try_new(
             3,
             7,
             Some([TableFeature::CatalogManaged]),
             Some([TableFeature::CatalogManaged]),
         )
         .unwrap();
-        assert!(protocol.ensure_write_supported().is_err());
+        assert!(protocol.ensure_write_supported().is_ok());
         let protocol = Protocol::try_new(
             3,
             7,
             Some([TableFeature::CatalogOwnedPreview]),
             Some([TableFeature::CatalogOwnedPreview]),
         )
         .unwrap();
-        assert!(protocol.ensure_write_supported().is_err());
+        assert!(protocol.ensure_write_supported().is_ok());
     }
 
     #[test]

kernel/src/committer.rs

@@ -45,12 +45,17 @@ use url::Url;
 pub struct CommitMetadata {
     pub(crate) log_root: LogRoot,
     pub(crate) version: Version,
+    pub(crate) in_commit_timestamp: i64,
     // in the future this will include Protocol, Metadata, CommitInfo, Domain Metadata, etc.
 }
 
 impl CommitMetadata {
-    pub(crate) fn new(log_root: LogRoot, version: Version) -> Self {
-        Self { log_root, version }
+    pub(crate) fn new(log_root: LogRoot, version: Version, in_commit_timestamp: i64) -> Self {
+        Self {
+            log_root,
+            version,
+            in_commit_timestamp,
+        }
     }
 
     /// The commit path is the absolute path (e.g. s3://bucket/table/_delta_log/{version}.json) to
@@ -73,6 +78,16 @@ impl CommitMetadata {
     pub fn version(&self) -> Version {
         self.version
     }
+
+    /// The in-commit timestamp for the commit. Note that this may differ from the actual commit
+    /// file modification time.
+    pub fn in_commit_timestamp(&self) -> i64 {
+        self.in_commit_timestamp
+    }
+
+    pub fn table_root(&self) -> &Url {
+        self.log_root.table_root()
+    }
 }
 
 /// `CommitResponse` is the result of committing a transaction via a catalog. The committer uses
@@ -171,11 +186,14 @@ mod tests {
         let table_root = Url::parse("s3://my-bucket/path/to/table/").unwrap();
         let log_root = LogRoot::new(table_root).unwrap();
         let version = 42;
+        let ts = 1234;
 
-        let commit_metadata = CommitMetadata::new(log_root, version);
+        let commit_metadata = CommitMetadata::new(log_root, version, ts);
 
         // version
         assert_eq!(commit_metadata.version(), 42);
+        // in_commit_timestamp
+        assert_eq!(commit_metadata.in_commit_timestamp(), 1234);
 
         // published commit path
         let published_path = commit_metadata.published_commit_path().unwrap();
@@ -189,8 +207,9 @@ mod tests {
         let staged_path_str = staged_path.as_str();
 
         assert!(
-            staged_path_str
-                .starts_with("s3://my-bucket/path/to/table/_delta_log/00000000000000000042."),
+            staged_path_str.starts_with(
+                "s3://my-bucket/path/to/table/_delta_log/_staged_commits/00000000000000000042."
+            ),
             "Staged path should start with the correct prefix, got: {}",
             staged_path_str
         );
@@ -200,15 +219,17 @@ mod tests {
             staged_path_str
         );
         let uuid_str = staged_path_str
-            .strip_prefix("s3://my-bucket/path/to/table/_delta_log/00000000000000000042.")
+            .strip_prefix(
+                "s3://my-bucket/path/to/table/_delta_log/_staged_commits/00000000000000000042.",
+            )
             .and_then(|s| s.strip_suffix(".json"))
             .expect("Staged path should have expected format");
         uuid::Uuid::parse_str(uuid_str).expect("Staged path should contain a valid UUID");
     }
 
     #[cfg(feature = "catalog-managed")]
     #[tokio::test]
-    async fn catalog_managed_tables_block_transactions() {
+    async fn disallow_filesystem_committer_for_catalog_managed_tables() {
         let storage = Arc::new(InMemory::new());
         let table_root = Url::parse("memory:///").unwrap();
         let engine = DefaultEngine::new(storage.clone());
@@ -225,18 +246,16 @@ mod tests {
         let snapshot = crate::snapshot::SnapshotBuilder::new_for(table_root)
             .build(&engine)
             .unwrap();
-        // Try to create a transaction with FileSystemCommitter
+        // Try to commit a transaction with FileSystemCommitter
         let committer = Box::new(FileSystemCommitter::new());
-        let err = snapshot.transaction(committer).unwrap_err();
+        let err = snapshot
+            .transaction(committer)
+            .unwrap()
+            .commit(&engine)
+            .unwrap_err();
         assert!(matches!(
             err,
-            crate::Error::Unsupported(e) if e.contains("Writes are not yet supported for catalog-managed tables")
+            crate::Error::Generic(e) if e.contains("The FileSystemCommitter cannot be used to commit to catalog-managed tables. Please provide a committer for your catalog via Transaction::with_committer().")
         ));
-        // after allowing writes, we will check that this disallows default committer for
-        // catalog-managed tables.
-        // assert!(matches!(
-        //     err,
-        //     crate::Error::Generic(e) if e.contains("Cannot use the default committer for a catalog-managed table")
-        // ));
     }
 }

kernel/src/engine/arrow_expression/evaluate_expression.rs

@@ -380,7 +380,6 @@ pub fn evaluate_predicate(
                     }
                 }
                 (Expression::Literal(lit), Expression::Literal(Scalar::Array(ad))) => {
-                    #[allow(deprecated)]
                     let exists = ad.array_elements().contains(lit);
                     Ok(BooleanArray::from(vec![exists]))
                 }

kernel/src/engine/arrow_expression/mod.rs

@@ -113,7 +113,6 @@ impl Scalar {
             Array(data) => {
                 let builder = builder_as!(array::ListBuilder<Box<dyn ArrayBuilder>>);
                 for _ in 0..num_rows {
-                    #[allow(deprecated)]
                     for value in data.array_elements() {
                         value.append_to(builder.values(), 1)?;
                     }
@@ -218,7 +217,6 @@ impl ArrayData {
     pub fn to_arrow(&self) -> DeltaResult<ArrayRef> {
         let arrow_data_type = ArrowDataType::try_from_kernel(self.array_type().element_type())?;
 
-        #[allow(deprecated)]
         let elements = self.array_elements();
         let mut builder = array::make_builder(&arrow_data_type, elements.len());
         for element in elements {

kernel/src/expressions/scalars.rs

@@ -94,9 +94,6 @@ impl ArrayData {
         &self.tpe
     }
 
-    #[deprecated(
-        note = "These fields will be removed eventually and are unstable. See https://github.com/delta-io/delta-kernel-rs/issues/291"
-    )]
     pub fn array_elements(&self) -> &[Scalar] {
         &self.elements
     }
@@ -923,7 +920,6 @@ mod tests {
 
     #[test]
     fn test_arrays() {
-        #[allow(deprecated)]
         let array = Scalar::Array(ArrayData {
             tpe: ArrayType::new(DataType::INTEGER, false),
             elements: vec![Scalar::Integer(1), Scalar::Integer(2), Scalar::Integer(3)],
@@ -1173,7 +1169,6 @@ mod tests {
         let Scalar::Array(array_data) = scalar else {
             panic!("Expected Array scalar");
         };
-        #[allow(deprecated)]
         let elements = array_data.array_elements();
         assert_eq!(elements.len(), 3);
         assert!(!array_data.array_type().contains_null());
@@ -1206,7 +1201,6 @@ mod tests {
             panic!("Expected Array scalar");
         };
 
-        #[allow(deprecated)]
         let elements = array_data.array_elements();
         assert_eq!(elements.len(), 3);
         assert!(array_data.array_type().contains_null());