feat: Honor write.metadata.path for metadata file locations

crates/iceberg/public-api.txt

@@ -2591,6 +2591,7 @@ pub fn iceberg::spec::TableMetadata::current_schema(&self) -> &iceberg::spec::Sc
 pub fn iceberg::spec::TableMetadata::current_schema_id(&self) -> iceberg::spec::SchemaId
 pub fn iceberg::spec::TableMetadata::current_snapshot(&self) -> core::option::Option<&iceberg::spec::SnapshotRef>
 pub fn iceberg::spec::TableMetadata::current_snapshot_id(&self) -> core::option::Option<i64>
+pub fn iceberg::spec::TableMetadata::default_metadata_dir(table_location: &str) -> alloc::string::String
 pub fn iceberg::spec::TableMetadata::default_partition_spec(&self) -> &iceberg::spec::PartitionSpecRef
 pub fn iceberg::spec::TableMetadata::default_partition_spec_id(&self) -> i32
 pub fn iceberg::spec::TableMetadata::default_partition_type(&self) -> &iceberg::spec::StructType
@@ -2608,6 +2609,7 @@ pub fn iceberg::spec::TableMetadata::last_updated_ms(&self) -> i64
 pub fn iceberg::spec::TableMetadata::last_updated_timestamp(&self) -> iceberg::Result<chrono::datetime::DateTime<chrono::offset::utc::Utc>>
 pub fn iceberg::spec::TableMetadata::location(&self) -> &str
 pub fn iceberg::spec::TableMetadata::metadata_compression_codec(&self) -> iceberg::Result<iceberg::compression::CompressionCodec>
+pub fn iceberg::spec::TableMetadata::metadata_location_root(&self) -> iceberg::Result<alloc::string::String>
 pub fn iceberg::spec::TableMetadata::metadata_log(&self) -> &[iceberg::spec::MetadataLog]
 pub fn iceberg::spec::TableMetadata::next_row_id(&self) -> u64
 pub fn iceberg::spec::TableMetadata::next_sequence_number(&self) -> i64
@@ -2711,6 +2713,7 @@ pub iceberg::spec::TableProperties::metadata_compression_codec: iceberg::compres
 pub iceberg::spec::TableProperties::min_snapshots_to_keep: usize
 pub iceberg::spec::TableProperties::write_datafusion_fanout_enabled: bool
 pub iceberg::spec::TableProperties::write_format_default: alloc::string::String
+pub iceberg::spec::TableProperties::write_metadata_path: core::option::Option<alloc::string::String>
 pub iceberg::spec::TableProperties::write_target_file_size_bytes: usize
 impl iceberg::spec::TableProperties
 pub const iceberg::spec::TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS: &str
@@ -2758,6 +2761,8 @@ pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL: &str
 pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT: i32
 pub const iceberg::spec::TableProperties::PROPERTY_SNAPSHOT_COUNT: &str
 pub const iceberg::spec::TableProperties::PROPERTY_UUID: &str
+pub const iceberg::spec::TableProperties::PROPERTY_WRITE_METADATA_PATH: &str
+pub const iceberg::spec::TableProperties::PROPERTY_WRITE_METADATA_PATH_DEFAULT_DIR: &str
 pub const iceberg::spec::TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT: &str
 pub const iceberg::spec::TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT: u64
 pub const iceberg::spec::TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES: &str

crates/iceberg/src/catalog/metadata_location.rs

@@ -25,11 +25,14 @@ use crate::compression::CompressionCodec;
 use crate::spec::{TableMetadata, parse_metadata_file_compression};
 use crate::{Error, ErrorKind, Result};
 
-/// Helper for parsing a location of the format: `<location>/metadata/<version>-<uuid>.metadata.json`
-/// or with compression: `<location>/metadata/<version>-<uuid>.gz.metadata.json`
+/// Helper for parsing a location of the format: `<metadata-dir>/<version>-<uuid>.metadata.json`
+/// or with compression: `<metadata-dir>/<version>-<uuid>.gz.metadata.json`
+///
+/// `<metadata-dir>` is set to the `write.metadata.path` table property and
+/// it defaults to the `<location>/metadata` when the property is not set.
 #[derive(Clone, Debug, PartialEq)]
 pub struct MetadataLocation {
-    table_location: String,
+    metadata_dir: String,
     version: i32,
     id: Uuid,
     compression_codec: CompressionCodec,
@@ -50,7 +53,7 @@ impl MetadataLocation {
     )]
     pub fn new_with_table_location(table_location: impl ToString) -> Self {
         Self {
-            table_location: table_location.to_string(),
+            metadata_dir: TableMetadata::default_metadata_dir(&table_location.to_string()),
             version: 0,
             id: Uuid::new_v4(),
             compression_codec: CompressionCodec::None,
@@ -60,9 +63,16 @@ impl MetadataLocation {
     /// Creates a completely new metadata location starting at version 0,
     /// with compression settings from the table metadata.
     /// Only used for creating a new table. For updates, see `next_version`.
+    ///
+    /// The metadata directory honors the `write.metadata.path` table property when set,
+    /// otherwise defaults to the `metadata` subdirectory of `table_location`.
     pub fn new_with_metadata(table_location: impl ToString, metadata: &TableMetadata) -> Self {
+        let table_location = table_location.to_string();
+        let metadata_dir = metadata
+            .metadata_location_root_with_base(&table_location)
+            .unwrap_or_else(|_| TableMetadata::default_metadata_dir(&table_location));
         Self {
-            table_location: table_location.to_string(),
+            metadata_dir,
             version: 0,
             id: Uuid::new_v4(),
             compression_codec: Self::compression_from_properties(metadata.properties()),
@@ -73,7 +83,7 @@ impl MetadataLocation {
     /// Increments the version number and generates a new UUID.
     pub fn with_next_version(&self) -> Self {
         Self {
-            table_location: self.table_location.clone(),
+            metadata_dir: self.metadata_dir.clone(),
             version: self.version + 1,
             id: Uuid::new_v4(),
             compression_codec: self.compression_codec,
@@ -83,7 +93,7 @@ impl MetadataLocation {
     /// Updates the metadata location with compression settings from the new metadata.
     pub fn with_new_metadata(&self, new_metadata: &TableMetadata) -> Self {
         Self {
-            table_location: self.table_location.clone(),
+            metadata_dir: self.metadata_dir.clone(),
             version: self.version,
             id: self.id,
             compression_codec: Self::compression_from_properties(new_metadata.properties()),
@@ -95,15 +105,6 @@ impl MetadataLocation {
         self.compression_codec
     }
 
-    fn parse_metadata_path_prefix(path: &str) -> Result<String> {
-        let prefix = path.strip_suffix("/metadata").ok_or(Error::new(
-            ErrorKind::Unexpected,
-            format!("Metadata location not under \"/metadata\" subdirectory: {path}"),
-        ))?;
-
-        Ok(prefix.to_string())
-    }
-
     /// Parses a file name of the format `<version>-<uuid>.metadata.json`
     /// or with compression: `<version>-<uuid>.gz.metadata.json`.
     /// Parse errors for compression codec result in CompressionCodec::None.
@@ -139,8 +140,8 @@ impl Display for MetadataLocation {
         let suffix = self.compression_codec.suffix().unwrap_or("");
         write!(
             f,
-            "{}/metadata/{:0>5}-{}{}.metadata.json",
-            self.table_location, self.version, self.id, suffix
+            "{}/{:0>5}-{}{}.metadata.json",
+            self.metadata_dir, self.version, self.id, suffix
         )
     }
 }
@@ -149,16 +150,15 @@ impl FromStr for MetadataLocation {
     type Err = Error;
 
     fn from_str(s: &str) -> Result<Self> {
-        let (path, file_name) = s.rsplit_once('/').ok_or(Error::new(
+        let (metadata_dir, file_name) = s.rsplit_once('/').ok_or(Error::new(
             ErrorKind::Unexpected,
             format!("Invalid metadata location: {s}"),
         ))?;
 
-        let prefix = Self::parse_metadata_path_prefix(path)?;
         let (version, id, compression_codec) = Self::parse_file_name(file_name)?;
 
         Ok(MetadataLocation {
-            table_location: prefix,
+            metadata_dir: metadata_dir.to_string(),
             version,
             id,
             compression_codec,
@@ -198,7 +198,7 @@ mod test {
             (
                 "/metadata/1234567-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json",
                 Ok(MetadataLocation {
-                    table_location: "".to_string(),
+                    metadata_dir: "/metadata".to_string(),
                     version: 1234567,
                     id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(),
                     compression_codec: CompressionCodec::None,
@@ -208,7 +208,7 @@ mod test {
             (
                 "/abc/metadata/1234567-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json",
                 Ok(MetadataLocation {
-                    table_location: "/abc".to_string(),
+                    metadata_dir: "/abc/metadata".to_string(),
                     version: 1234567,
                     id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(),
                     compression_codec: CompressionCodec::None,
@@ -218,7 +218,7 @@ mod test {
             (
                 "/abc/def/metadata/1234567-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json",
                 Ok(MetadataLocation {
-                    table_location: "/abc/def".to_string(),
+                    metadata_dir: "/abc/def/metadata".to_string(),
                     version: 1234567,
                     id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(),
                     compression_codec: CompressionCodec::None,
@@ -228,7 +228,7 @@ mod test {
             (
                 "https://127.0.0.1/metadata/1234567-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json",
                 Ok(MetadataLocation {
-                    table_location: "https://127.0.0.1".to_string(),
+                    metadata_dir: "https://127.0.0.1/metadata".to_string(),
                     version: 1234567,
                     id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(),
                     compression_codec: CompressionCodec::None,
@@ -238,7 +238,7 @@ mod test {
             (
                 "/abc/metadata/1234567-81056704-ce5b-41c4-bb83-eb6408081af6.metadata.json",
                 Ok(MetadataLocation {
-                    table_location: "/abc".to_string(),
+                    metadata_dir: "/abc/metadata".to_string(),
                     version: 1234567,
                     id: Uuid::from_str("81056704-ce5b-41c4-bb83-eb6408081af6").unwrap(),
                     compression_codec: CompressionCodec::None,
@@ -248,7 +248,7 @@ mod test {
             (
                 "/abc/metadata/00000-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json",
                 Ok(MetadataLocation {
-                    table_location: "/abc".to_string(),
+                    metadata_dir: "/abc/metadata".to_string(),
                     version: 0,
                     id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(),
                     compression_codec: CompressionCodec::None,
@@ -258,7 +258,7 @@ mod test {
             (
                 "/abc/metadata/1234567-2cd22b57-5127-4198-92ba-e4e67c79821b.gz.metadata.json",
                 Ok(MetadataLocation {
-                    table_location: "/abc".to_string(),
+                    metadata_dir: "/abc/metadata".to_string(),
                     version: 1234567,
                     id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(),
                     compression_codec: CompressionCodec::gzip_default(),
@@ -279,10 +279,15 @@ mod test {
                 "/metadata/noversion-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json",
                 Err("".to_string()),
             ),
-            // No /metadata subdirectory
+            // Metadata dir does not need to be named "metadata" (e.g. a `write.metadata.path`` location)
             (
                 "/wrongsubdir/1234567-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json",
-                Err("".to_string()),
+                Ok(MetadataLocation {
+                    metadata_dir: "/wrongsubdir".to_string(),
+                    version: 1234567,
+                    id: Uuid::from_str("2cd22b57-5127-4198-92ba-e4e67c79821b").unwrap(),
+                    compression_codec: CompressionCodec::None,
+                }),
             ),
             // No .metadata.json suffix
             (
@@ -321,7 +326,7 @@ mod test {
             let next = MetadataLocation::from_str(&input.to_string())
                 .unwrap()
                 .with_next_version();
-            assert_eq!(next.table_location, input.table_location);
+            assert_eq!(next.metadata_dir, input.metadata_dir);
             assert_eq!(next.version, input.version + 1);
             assert_ne!(next.id, input.id);
         }
@@ -410,4 +415,32 @@ mod test {
             "/test/table/metadata/00000-2cd22b57-5127-4198-92ba-e4e67c79821b.metadata.json"
         );
     }
+
+    #[test]
+    fn test_new_with_metadata_honors_write_metadata_path() {
+        // Test metadata lives under `<location>/metadata` by default
+        let default_meta = create_test_metadata(HashMap::new());
+        let default_loc = MetadataLocation::new_with_metadata("/test/table", &default_meta);
+        assert!(
+            default_loc
+                .to_string()
+                .starts_with("/test/table/metadata/00000-"),
+            "unexpected location: {default_loc}"
+        );
+
+        // Test a configured `write.metadata.path` is honored
+        let props = HashMap::from([(
+            "write.metadata.path".to_string(),
+            "s3://bucket/custom-meta".to_string(),
+        )]);
+        let custom_meta = create_test_metadata(props);
+        let custom_loc = MetadataLocation::new_with_metadata("/test/table", &custom_meta);
+        assert!(
+            custom_loc
+                .to_string()
+                .starts_with("s3://bucket/custom-meta/00000-"),
+            "unexpected location: {custom_loc}"
+        );
+        assert!(custom_loc.to_string().ends_with(".metadata.json"));
+    }
 }

crates/iceberg/src/spec/table_metadata.rs

@@ -364,6 +364,33 @@ impl TableMetadata {
         &self.properties
     }
 
+    /// Returns the default metadata directory for a table location: the `metadata`
+    /// subdirectory under `table_location`. Used when `write.metadata.path` is not set.
+    pub fn default_metadata_dir(table_location: &str) -> String {
+        format!(
+            "{table_location}/{}",
+            TableProperties::PROPERTY_WRITE_METADATA_PATH_DEFAULT_DIR
+        )
+    }
+
+    /// Returns the base location for metadata files.
+    ///
+    /// Honors the `write.metadata.path` table property when set, otherwise defaults
+    /// to the `metadata` directory under the table location.
+    pub fn metadata_location_root(&self) -> Result<String> {
+        self.metadata_location_root_with_base(self.location())
+    }
+
+    /// Like [`Self::metadata_location_root`], but uses an explicit table location as
+    /// the base for the default `<location>/metadata` when `write.metadata.path` is not
+    /// configured.
+    pub(crate) fn metadata_location_root_with_base(&self, base: &str) -> Result<String> {
+        Ok(self
+            .table_properties()?
+            .write_metadata_path
+            .unwrap_or_else(|| Self::default_metadata_dir(base)))
+    }
+
     /// Returns the metadata compression codec from table properties.
     ///
     /// Returns `CompressionCodec::None` if compression is disabled or not configured.
@@ -4283,4 +4310,52 @@ mod tests {
         assert_eq!(deserialized_first_row_id, 100);
         assert_eq!(deserialized_added_rows, 50);
     }
+
+    #[test]
+    fn test_metadata_location_root_default() {
+        // Verify metadata files go to `<location>/metadata` when `write.metadata.path` is not set
+        let metadata = get_test_table_metadata("TableMetadataV2Valid.json");
+        assert_eq!(metadata.location(), "s3://bucket/test/location");
+        assert_eq!(
+            metadata.metadata_location_root().unwrap(),
+            "s3://bucket/test/location/metadata"
+        );
+    }
+
+    #[test]
+    fn test_metadata_location_root_honors_write_metadata_path() {
+        let metadata = get_test_table_metadata("TableMetadataV2Valid.json")
+            .into_builder(None)
+            .set_properties(HashMap::from([(
+                TableProperties::PROPERTY_WRITE_METADATA_PATH.to_string(),
+                "s3://other-bucket/custom-meta".to_string(),
+            )]))
+            .unwrap()
+            .build()
+            .unwrap()
+            .metadata;
+        assert_eq!(
+            metadata.metadata_location_root().unwrap(),
+            "s3://other-bucket/custom-meta"
+        );
+    }
+
+    #[test]
+    fn test_metadata_location_root_trims_trailing_slash() {
+        // A configured path with a trailing slash must not yield a doubled separator
+        let metadata = get_test_table_metadata("TableMetadataV2Valid.json")
+            .into_builder(None)
+            .set_properties(HashMap::from([(
+                TableProperties::PROPERTY_WRITE_METADATA_PATH.to_string(),
+                "s3://other-bucket/custom-meta/".to_string(),
+            )]))
+            .unwrap()
+            .build()
+            .unwrap()
+            .metadata;
+        assert_eq!(
+            metadata.metadata_location_root().unwrap(),
+            "s3://other-bucket/custom-meta"
+        );
+    }
 }