Address review comments

Author: lbhm

kernel/src/engine/arrow_utils.rs

@@ -568,32 +568,32 @@ fn get_indices(
         // some fields are missing, but they might be nullable or metadata columns, need to insert them into the reorder_indices
         for (requested_position, field) in requested_schema.fields().enumerate() {
             if !found_fields.contains(field.name()) {
-                if let Some(metadata_spec) = field.get_metadata_column_spec() {
-                    match metadata_spec {
-                        MetadataColumnSpec::RowIndex => {
-                            debug!("Inserting a row index column: {}", field.name());
-                            reorder_indices.push(ReorderIndex::row_index(
-                                requested_position,
-                                Arc::new(field.try_into_arrow()?),
-                            ));
-                        }
-                        _ => {
-                            return Err(Error::Generic(format!(
-                                "Metadata column {metadata_spec:?} is not supported by the default parquet reader"
-                            )));
-                        }
+                match field.get_metadata_column_spec() {
+                    Some(MetadataColumnSpec::RowIndex) => {
+                        debug!("Inserting a row index column: {}", field.name());
+                        reorder_indices.push(ReorderIndex::row_index(
+                            requested_position,
+                            Arc::new(field.try_into_arrow()?),
+                        ));
+                    }
+                    Some(metadata_spec) => {
+                        return Err(Error::Generic(format!(
+                            "Metadata column {metadata_spec:?} is not supported by the default parquet reader"
+                        )));
+                    }
+                    None if field.nullable => {
+                        debug!("Inserting missing and nullable field: {}", field.name());
+                        reorder_indices.push(ReorderIndex::missing(
+                            requested_position,
+                            Arc::new(field.try_into_arrow()?),
+                        ));
+                    }
+                    None => {
+                        return Err(Error::Generic(format!(
+                            "Requested field not found in parquet schema, and field is not nullable: {}",
+                            field.name()
+                        )));
                     }
-                } else if field.nullable {
-                    debug!("Inserting missing and nullable field: {}", field.name());
-                    reorder_indices.push(ReorderIndex::missing(
-                        requested_position,
-                        Arc::new(field.try_into_arrow()?),
-                    ));
-                } else {
-                    return Err(Error::Generic(format!(
-                        "Requested field not found in parquet schema, and field is not nullable: {}",
-                        field.name()
-                    )));
                 }
             }
         }
@@ -821,15 +821,12 @@ pub(crate) fn reorder_struct_array(
                         row_indexes.take(num_rows).collect();
                     require!(
                         row_index_array.len() == num_rows,
-                        Error::internal_error(format!(
-                            "Row index iterator exhausted after only {} of {} rows",
-                            row_index_array.len(),
-                            num_rows
-                        ))
+                        Error::internal_error(
+                            "Row index iterator exhausted before reaching the end of the file"
+                        )
                     );
-                    let field = field.clone(); // cheap Arc clone
                     final_fields_cols[reorder_index.index] =
-                        Some((field, Arc::new(row_index_array)));
+                        Some((Arc::clone(field), Arc::new(row_index_array)));
                 }
             }
         }
@@ -856,9 +853,6 @@ fn reorder_list<O: OffsetSizeTrait>(
     let (list_field, offset_buffer, maybe_sa, null_buf) = list_array.into_parts();
     if let Some(struct_array) = maybe_sa.as_struct_opt() {
         let struct_array = struct_array.clone();
-        // WARNING: We cannot naively plumb through our caller's row index iterator, because each
-        // array element of a given row must replicate the row's index and empty arrays must drop
-        // that row's index. For now, just don't support it (Delta doesn't need the capability).
         let result_array = Arc::new(reorder_struct_array(
             struct_array,
             children,

kernel/tests/read.rs

@@ -1487,67 +1487,48 @@ async fn test_unsupported_metadata_columns() -> Result<(), Box<dyn std::error::E
         Arc::new(TokioBackgroundExecutor::new()),
     ));
 
-    // Test reading the RowId metadata column (should fail)
-    let snapshot1 = Snapshot::builder(location.clone()).build(engine.as_ref())?;
-    let schema = Arc::new(StructType::try_new([
-        StructField::nullable("id", DataType::INTEGER),
-        StructField::create_metadata_column("row_id", MetadataColumnSpec::RowId),
-    ])?);
-    let scan = snapshot1.into_scan_builder().with_schema(schema).build()?;
-    let stream = scan.execute(engine.clone())?;
-
-    let mut found_error = false;
-    for scan_result in stream {
-        match scan_result {
-            Err(e) => {
-                let error_msg = e.to_string();
-                println!("error_msg: {}", error_msg);
-                if error_msg.contains("RowId") && error_msg.contains("not supported") {
-                    found_error = true;
-                    break;
-                }
-            }
-            Ok(_) => {
-                panic!("Expected error for RowId metadata column, but scan succeeded");
-            }
-        }
-    }
-    assert!(
-        found_error,
-        "Expected error about RowId not being supported"
-    );
-
-    // Test reading the RowCommitVersion metadata column (should fail)
-    let snapshot2 = Snapshot::builder(location.clone()).build(engine.as_ref())?;
-    let schema = Arc::new(StructType::try_new([
-        StructField::nullable("id", DataType::INTEGER),
-        StructField::create_metadata_column(
+    // Test that unsupported metadata columns fail with appropriate errors
+    let test_cases = [
+        ("row_id", MetadataColumnSpec::RowId, "RowId"),
+        (
             "row_commit_version",
             MetadataColumnSpec::RowCommitVersion,
+            "RowCommitVersion",
         ),
-    ])?);
-    let scan = snapshot2.into_scan_builder().with_schema(schema).build()?;
-    let stream = scan.execute(engine)?;
-
-    let mut found_error = false;
-    for scan_result in stream {
-        match scan_result {
-            Err(e) => {
-                let error_msg = e.to_string();
-                if error_msg.contains("RowCommitVersion") && error_msg.contains("not supported") {
-                    found_error = true;
-                    break;
+    ];
+    for (column_name, metadata_spec, error_text) in test_cases {
+        let snapshot = Snapshot::builder(location.clone()).build(engine.as_ref())?;
+        let schema = Arc::new(StructType::try_new([
+            StructField::nullable("id", DataType::INTEGER),
+            StructField::create_metadata_column(column_name, metadata_spec),
+        ])?);
+        let scan = snapshot.into_scan_builder().with_schema(schema).build()?;
+        let stream = scan.execute(engine.clone())?;
+
+        let mut found_error = false;
+        for scan_result in stream {
+            match scan_result {
+                Err(e) => {
+                    let error_msg = e.to_string();
+                    if error_msg.contains(error_text) && error_msg.contains("not supported") {
+                        found_error = true;
+                        break;
+                    }
+                }
+                Ok(_) => {
+                    panic!(
+                        "Expected error for {} metadata column, but scan succeeded",
+                        error_text
+                    );
                 }
-            }
-            Ok(_) => {
-                panic!("Expected error for RowCommitVersion metadata column, but scan succeeded");
             }
         }
+        assert!(
+            found_error,
+            "Expected error about {} not being supported",
+            error_text
+        );
     }
-    assert!(
-        found_error,
-        "Expected error about RowCommitVersion not being supported"
-    );
 
     Ok(())
 }