move metadata file location function to Table

Author: geruh

pyiceberg/catalog/__init__.py

@@ -857,7 +857,7 @@ def _create_staged_table(
         database_name, table_name = self.identifier_to_database_and_table(identifier)
 
         location = self._resolve_table_location(location, database_name, table_name)
-        metadata_location = self._get_metadata_location(table_location=location, properties=properties)
+        metadata_location = Table.new_table_metadata_file_location(table_location=location, properties=properties)
         metadata = new_table_metadata(
             location=location, schema=schema, partition_spec=partition_spec, sort_order=sort_order, properties=properties
         )
@@ -888,7 +888,7 @@ def _update_and_stage_table(
         )
 
         new_metadata_version = self._parse_metadata_version(current_table.metadata_location) + 1 if current_table else 0
-        new_metadata_location = self._get_metadata_location(
+        new_metadata_location = Table.new_table_metadata_file_location(
             updated_metadata.location, new_metadata_version, updated_metadata.properties
         )
 
@@ -947,14 +947,6 @@ def _get_default_warehouse_location(self, database_name: str, table_name: str) -
     def _write_metadata(metadata: TableMetadata, io: FileIO, metadata_path: str) -> None:
         ToOutputFile.table_metadata(metadata, io.new_output(metadata_path))
 
-    @staticmethod
-    def _get_metadata_location(table_location: str, new_version: int = 0, properties: Properties = EMPTY_DICT) -> str:
-        if new_version < 0:
-            raise ValueError(f"Table metadata version: `{new_version}` must be a non-negative integer")
-
-        file_name = f"{new_version:05d}-{uuid.uuid4()}.metadata.json"
-        return Table.metadata_file_location(table_location, file_name, properties)
-
     @staticmethod
     def _parse_metadata_version(metadata_location: str) -> int:
         """Parse the version from the metadata location.

pyiceberg/catalog/dynamodb.py

@@ -173,7 +173,7 @@ def create_table(
         database_name, table_name = self.identifier_to_database_and_table(identifier)
 
         location = self._resolve_table_location(location, database_name, table_name)
-        metadata_location = self._get_metadata_location(table_location=location, properties=properties)
+        metadata_location = Table.new_table_metadata_file_location(table_location=location, properties=properties)
         metadata = new_table_metadata(
             location=location, schema=schema, partition_spec=partition_spec, sort_order=sort_order, properties=properties
         )

pyiceberg/catalog/sql.py

@@ -207,7 +207,7 @@ def create_table(
 
         namespace = Catalog.namespace_to_string(namespace_identifier)
         location = self._resolve_table_location(location, namespace, table_name)
-        metadata_location = self._get_metadata_location(table_location=location, properties=properties)
+        metadata_location = Table.new_table_metadata_file_location(table_location=location, properties=properties)
         metadata = new_table_metadata(
             location=location, schema=schema, partition_spec=partition_spec, sort_order=sort_order, properties=properties
         )

pyiceberg/table/__init__.py

@@ -1238,19 +1238,40 @@ def to_polars(self) -> pl.LazyFrame:
         return pl.scan_iceberg(self)
 
     @staticmethod
-    def metadata_file_location(table_location: str, file_name: str, properties: Properties = EMPTY_DICT) -> str:
-        """Get the full path for a metadata file.
+    def new_table_metadata_file_location(table_location: str, new_version: int = 0, properties: Properties = EMPTY_DICT) -> str:
+        """Return a fully-qualified metadata file location for a new table version.
+
+        Args:
+            table_location (str): the base table location.
+            new_version (int): Version number of the metadata file.
+            properties (Properties): Table properties that may contain a custom metadata path.
+
+        Returns:
+            str: fully-qualified URI for the new table metadata file.
+
+        Raises:
+            ValueError: If the version is negative.
+        """
+        if new_version < 0:
+            raise ValueError(f"Table metadata version: `{new_version}` must be a non-negative integer")
+
+        file_name = f"{new_version:05d}-{uuid.uuid4()}.metadata.json"
+        return Table.new_metadata_location(table_location, file_name, properties)
+
+    @staticmethod
+    def new_metadata_location(table_location: str, file_name: str, properties: Properties = EMPTY_DICT) -> str:
+        """Return a fully-qualified metadata file location for the given filename.
 
         Args:
             table_location (str): The base table location
             file_name (str): Name of the metadata file
-            properties (Properties): Table properties that may contain custom metadata path
+            properties (Properties): Table properties that may contain a custom metadata path
 
         Returns:
-            str: Full path where the metadata file should be stored
+            str: A fully-qualified location URI for the metadata file.
         """
         if metadata_path := properties.get(TableProperties.WRITE_METADATA_PATH):
-            return f"{metadata_path.rstrip("/")}/{file_name}"
+            return f"{metadata_path.rstrip('/')}/{file_name}"
 
         return f"{table_location}/metadata/{file_name}"
 

pyiceberg/table/update/snapshot.py

@@ -88,7 +88,7 @@ def _new_manifest_file_name(num: int, commit_uuid: uuid.UUID) -> str:
     return f"{commit_uuid}-m{num}.avro"
 
 
-def _generate_manifest_list_file_name(snapshot_id: int, attempt: int, commit_uuid: uuid.UUID) -> str:
+def _new_manifest_list_file_name(snapshot_id: int, attempt: int, commit_uuid: uuid.UUID) -> str:
     # Mimics the behavior in Java:
     # https://github.com/apache/iceberg/blob/c862b9177af8e2d83122220764a056f3b96fd00c/core/src/main/java/org/apache/iceberg/SnapshotProducer.java#L491
     return f"snap-{snapshot_id}-{attempt}-{commit_uuid}.avro"
@@ -245,12 +245,12 @@ def _commit(self) -> UpdatesAndRequirements:
         summary = self._summary(self.snapshot_properties)
         table_location = self._transaction.table_metadata.location
         properties = self._transaction.table_metadata.properties
-        file_name = _generate_manifest_list_file_name(
+        file_name = _new_manifest_list_file_name(
             snapshot_id=self._snapshot_id,
             attempt=0,
             commit_uuid=self.commit_uuid,
         )
-        manifest_list_file_path = self._transaction._table.metadata_file_location(table_location, file_name, properties)
+        manifest_list_file_path = self._transaction._table.new_metadata_location(table_location, file_name, properties)
         with write_manifest_list(
             format_version=self._transaction.table_metadata.format_version,
             output_file=self._io.new_output(manifest_list_file_path),
@@ -299,7 +299,7 @@ def new_manifest_output(self) -> OutputFile:
         table_location = self._transaction.table_metadata.location
         properties = self._transaction.table_metadata.properties
         file_name = _new_manifest_file_name(num=next(self._manifest_num_counter), commit_uuid=self.commit_uuid)
-        file_path = self._transaction._table.metadata_file_location(table_location, file_name, properties)
+        file_path = self._transaction._table.new_metadata_location(table_location, file_name, properties)
         return self._io.new_output(file_path)
 
     def fetch_manifest_entry(self, manifest: ManifestFile, discard_deleted: bool = True) -> List[ManifestEntry]:

tests/catalog/test_base.py

@@ -580,7 +580,7 @@ def test_table_writes_metadata_to_custom_location(catalog: InMemoryCatalog) -> N
     table.append(df)
     manifests = table.current_snapshot().manifests(table.io)  # type: ignore
 
-    assert table.metadata_file_location(table.location(), "", table.properties).startswith(metadata_path)
+    assert table.new_metadata_location(table.location(), "", table.properties).startswith(metadata_path)
     assert manifests[0].manifest_path.startswith(metadata_path)
     assert table.location() != metadata_path
     assert table.metadata_location.startswith(metadata_path)
@@ -599,6 +599,6 @@ def test_table_writes_metadata_to_default_path(catalog: InMemoryCatalog) -> None
     table.append(df)
     manifests = table.current_snapshot().manifests(table.io)  # type: ignore
 
-    assert table.metadata_file_location(table.location(), "", table.properties).startswith(metadata_path)
+    assert table.new_metadata_location(table.location(), "", table.properties).startswith(metadata_path)
     assert manifests[0].manifest_path.startswith(metadata_path)
     assert table.metadata_location.startswith(metadata_path)