refactor: trim metadata PK tests and simplify sync helper

- Keep only "set + sync" and "merge insert" tests in dataset::metadata per review; parse-error coverage lives in lance-core field unit tests.
- Extract parse_unenforced_primary_key_position helper so TryFrom<&ArrowField> computes the value before the struct literal instead of the None+update pattern.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Vova Kolmakov

rust/lance-core/src/datatypes/field.rs

@@ -1025,23 +1025,30 @@ impl Field {
     /// to keep structured fields like `unenforced_primary_key_position` in sync.
     pub fn sync_embedded_metadata(&mut self) -> Result<()> {
         self.unenforced_primary_key_position =
-            if let Some(s) = self.metadata.get(LANCE_UNENFORCED_PRIMARY_KEY_POSITION) {
-                Some(s.parse::<u32>().map_err(|e| {
-                    Error::invalid_input(format!(
-                        "Invalid value '{}' for {}: {}",
-                        s, LANCE_UNENFORCED_PRIMARY_KEY_POSITION, e
-                    ))
-                })?)
-            } else {
-                self.metadata
-                    .get(LANCE_UNENFORCED_PRIMARY_KEY)
-                    .filter(|s| matches!(s.to_lowercase().as_str(), "true" | "1" | "yes"))
-                    .map(|_| 0)
-            };
+            parse_unenforced_primary_key_position(&self.metadata)?;
         Ok(())
     }
 }
 
+fn parse_unenforced_primary_key_position(
+    metadata: &HashMap<String, String>,
+) -> Result<Option<u32>> {
+    if let Some(s) = metadata.get(LANCE_UNENFORCED_PRIMARY_KEY_POSITION) {
+        let parsed = s.parse::<u32>().map_err(|e| {
+            Error::invalid_input(format!(
+                "Invalid value '{}' for {}: {}",
+                s, LANCE_UNENFORCED_PRIMARY_KEY_POSITION, e
+            ))
+        })?;
+        Ok(Some(parsed))
+    } else {
+        Ok(metadata
+            .get(LANCE_UNENFORCED_PRIMARY_KEY)
+            .filter(|s| matches!(s.to_lowercase().as_str(), "true" | "1" | "yes"))
+            .map(|_| 0))
+    }
+}
+
 impl fmt::Display for Field {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
@@ -1137,7 +1144,9 @@ impl TryFrom<&ArrowField> for Field {
             LogicalType::try_from(field.data_type())?
         };
 
-        let mut result = Self {
+        let unenforced_primary_key_position = parse_unenforced_primary_key_position(&metadata)?;
+
+        Ok(Self {
             id,
             parent_id: -1,
             name: field.name().clone(),
@@ -1156,10 +1165,8 @@ impl TryFrom<&ArrowField> for Field {
             nullable: field.is_nullable(),
             children,
             dictionary: None,
-            unenforced_primary_key_position: None,
-        };
-        result.sync_embedded_metadata()?;
-        Ok(result)
+            unenforced_primary_key_position,
+        })
     }
 }
 

rust/lance/src/dataset/metadata.rs

@@ -587,7 +587,7 @@ mod tests {
     async fn test_update_field_metadata_sets_unenforced_primary_key() {
         let mut dataset = test_dataset_for_pk().await;
 
-        // Set the boolean primary key metadata on the "id" field
+        // Legacy boolean flag should map to position 0.
         dataset
             .update_field_metadata()
             .update("id", [("lance-schema:unenforced-primary-key", "true")])
@@ -605,21 +605,13 @@ mod tests {
             Some(0),
             "Legacy boolean flag should map to position 0"
         );
-    }
 
-    #[tokio::test]
-    async fn test_update_field_metadata_sets_unenforced_primary_key_position() {
-        let mut dataset = test_dataset_for_pk().await;
-
-        // Set both the boolean flag and explicit position
+        // Explicit position should override the legacy flag.
         dataset
             .update_field_metadata()
             .update(
                 "id",
-                [
-                    ("lance-schema:unenforced-primary-key", "true"),
-                    ("lance-schema:unenforced-primary-key:position", "2"),
-                ],
+                [("lance-schema:unenforced-primary-key:position", "2")],
             )
             .unwrap()
             .await
@@ -630,137 +622,7 @@ mod tests {
         assert_eq!(
             field.unenforced_primary_key_position,
             Some(2),
-            "Explicit position should take precedence over boolean flag"
-        );
-    }
-
-    #[tokio::test]
-    async fn test_update_field_metadata_removes_unenforced_primary_key() {
-        let mut dataset = test_dataset_for_pk().await;
-
-        // First, set the primary key
-        dataset
-            .update_field_metadata()
-            .update("id", [("lance-schema:unenforced-primary-key", "true")])
-            .unwrap()
-            .await
-            .unwrap();
-        assert!(
-            dataset
-                .schema()
-                .field("id")
-                .unwrap()
-                .is_unenforced_primary_key()
-        );
-
-        // Now remove it by setting value to None (delete the key)
-        dataset
-            .update_field_metadata()
-            .update(
-                "id",
-                [("lance-schema:unenforced-primary-key", Option::<&str>::None)],
-            )
-            .unwrap()
-            .await
-            .unwrap();
-
-        let field = dataset.schema().field("id").unwrap();
-        assert!(
-            !field.is_unenforced_primary_key(),
-            "Field should no longer be a primary key after removing the metadata key"
-        );
-        assert_eq!(field.unenforced_primary_key_position, None);
-    }
-
-    #[tokio::test]
-    async fn test_update_field_metadata_replace_clears_unenforced_primary_key() {
-        let mut dataset = test_dataset_for_pk().await;
-
-        // First, set the primary key
-        dataset
-            .update_field_metadata()
-            .update("id", [("lance-schema:unenforced-primary-key", "true")])
-            .unwrap()
-            .await
-            .unwrap();
-        assert!(
-            dataset
-                .schema()
-                .field("id")
-                .unwrap()
-                .is_unenforced_primary_key()
-        );
-
-        // Replace all metadata with unrelated keys — PK metadata should be cleared
-        dataset
-            .update_field_metadata()
-            .replace("id", [("some-other-key", "some-value")])
-            .unwrap()
-            .await
-            .unwrap();
-
-        let field = dataset.schema().field("id").unwrap();
-        assert!(
-            !field.is_unenforced_primary_key(),
-            "Primary key status should be cleared after replacing metadata without PK keys"
-        );
-        assert_eq!(field.unenforced_primary_key_position, None);
-    }
-
-    #[tokio::test]
-    async fn test_update_field_metadata_primary_key_roundtrip() {
-        use lance_core::utils::tempfile::TempDir;
-
-        let dir = TempDir::default();
-        let uri = dir.path_str();
-
-        let schema = Arc::new(ArrowSchema::new(vec![
-            ArrowField::new("id", DataType::Int32, false),
-            ArrowField::new("value", DataType::Utf8, true),
-        ]));
-
-        let batch = RecordBatch::try_new(
-            schema.clone(),
-            vec![
-                Arc::new(Int32Array::from(vec![1, 2, 3])),
-                Arc::new(arrow_array::StringArray::from(vec!["a", "b", "c"])),
-            ],
-        )
-        .unwrap();
-
-        let mut dataset = Dataset::write(
-            RecordBatchIterator::new(vec![Ok(batch)], schema.clone()),
-            &uri,
-            None,
-        )
-        .await
-        .unwrap();
-
-        // Set PK metadata via update
-        dataset
-            .update_field_metadata()
-            .update(
-                "id",
-                [
-                    ("lance-schema:unenforced-primary-key", "true"),
-                    ("lance-schema:unenforced-primary-key:position", "1"),
-                ],
-            )
-            .unwrap()
-            .await
-            .unwrap();
-
-        // Reload the dataset from storage to verify protobuf round-trip
-        let reloaded = Dataset::open(&uri).await.unwrap();
-        let field = reloaded.schema().field("id").unwrap();
-        assert!(
-            field.is_unenforced_primary_key(),
-            "Primary key should survive protobuf round-trip after metadata update"
-        );
-        assert_eq!(
-            field.unenforced_primary_key_position,
-            Some(1),
-            "Primary key position should survive protobuf round-trip"
+            "Explicit position should take precedence over the legacy boolean flag"
         );
     }
 
@@ -838,61 +700,4 @@ mod tests {
             ]
         );
     }
-
-    #[tokio::test]
-    async fn test_update_field_metadata_invalid_pk_position_returns_error() {
-        let mut dataset = test_dataset_for_pk().await;
-
-        let result = dataset
-            .update_field_metadata()
-            .update(
-                "id",
-                [("lance-schema:unenforced-primary-key:position", "abc")],
-            )
-            .unwrap()
-            .await;
-
-        assert!(result.is_err(), "Non-numeric position should return error");
-        assert!(matches!(result.unwrap_err(), Error::InvalidInput { .. }));
-    }
-
-    #[tokio::test]
-    async fn test_update_field_metadata_negative_pk_position_returns_error() {
-        let mut dataset = test_dataset_for_pk().await;
-
-        let result = dataset
-            .update_field_metadata()
-            .update(
-                "id",
-                [("lance-schema:unenforced-primary-key:position", "-1")],
-            )
-            .unwrap()
-            .await;
-
-        assert!(result.is_err(), "Negative position should return error");
-        assert!(matches!(result.unwrap_err(), Error::InvalidInput { .. }));
-    }
-
-    #[tokio::test]
-    async fn test_update_field_metadata_overflow_pk_position_returns_error() {
-        let mut dataset = test_dataset_for_pk().await;
-
-        let result = dataset
-            .update_field_metadata()
-            .update(
-                "id",
-                [(
-                    "lance-schema:unenforced-primary-key:position",
-                    "99999999999",
-                )],
-            )
-            .unwrap()
-            .await;
-
-        assert!(
-            result.is_err(),
-            "Overflowing u32 position should return error"
-        );
-        assert!(matches!(result.unwrap_err(), Error::InvalidInput { .. }));
-    }
 }