Add location propertiesfrom TableMetadata into Table entity internalProperties

Author: singhpk234

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java

@@ -1657,7 +1657,18 @@ private String newTableMetadataFilePath(TableMetadata meta, int newVersion) {
 
   private static Map<String, String> buildTableMetadataPropertiesMap(TableMetadata metadata) {
     Map<String, String> storedProperties = new HashMap<>();
+    // Location specific properties
     storedProperties.put(IcebergTableLikeEntity.LOCATION, metadata.location());
+    if (metadata.properties().containsKey(TableProperties.WRITE_DATA_LOCATION)) {
+      storedProperties.put(
+          IcebergTableLikeEntity.USER_SPECIFIED_WRITE_DATA_LOCATION_KEY,
+          metadata.properties().get(TableProperties.WRITE_DATA_LOCATION));
+    }
+    if (metadata.properties().containsKey(TableProperties.WRITE_METADATA_LOCATION)) {
+      storedProperties.put(
+          IcebergTableLikeEntity.USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY,
+          metadata.properties().get(TableProperties.WRITE_METADATA_LOCATION));
+    }
     storedProperties.put(
         IcebergTableLikeEntity.FORMAT_VERSION, String.valueOf(metadata.formatVersion()));
     storedProperties.put(IcebergTableLikeEntity.TABLE_UUID, metadata.uuid());

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogTest.java

@@ -72,6 +72,7 @@
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableMetadataParser;
 import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.UpdateRequirement;
 import org.apache.iceberg.UpdateRequirements;
 import org.apache.iceberg.UpdateSchema;
@@ -2269,6 +2270,16 @@ public void testTableInternalPropertiesStoredOnCommit() {
     catalog.createNamespace(NS);
     catalog.buildTable(TABLE, SCHEMA).create();
     catalog.loadTable(TABLE).newFastAppend().appendFile(FILE_A).commit();
+    catalog
+        .loadTable(TABLE)
+        .updateProperties()
+        .set(
+            TableProperties.WRITE_DATA_LOCATION,
+            "s3://my-bucket/path/to/data/newdb/newtable/my-data")
+        .set(
+            TableProperties.WRITE_METADATA_LOCATION,
+            "s3://my-bucket/path/to/data/newdb/newtable/my-metadata")
+        .commit();
     Table afterAppend = catalog.loadTable(TABLE);
     EntityResult schemaResult =
         metaStoreManager.readEntityByName(
@@ -2295,6 +2306,12 @@ public void testTableInternalPropertiesStoredOnCommit() {
             IcebergTableLikeEntity.CURRENT_SNAPSHOT_ID,
             String.valueOf(afterAppend.currentSnapshot().snapshotId()))
         .containsEntry(IcebergTableLikeEntity.LOCATION, afterAppend.location())
+        .containsEntry(
+            IcebergTableLikeEntity.USER_SPECIFIED_WRITE_DATA_LOCATION_KEY,
+            afterAppend.properties().get(TableProperties.WRITE_DATA_LOCATION))
+        .containsEntry(
+            IcebergTableLikeEntity.USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY,
+            afterAppend.properties().get(TableProperties.WRITE_METADATA_LOCATION))
         .containsEntry(IcebergTableLikeEntity.TABLE_UUID, afterAppend.uuid().toString())
         .containsEntry(
             IcebergTableLikeEntity.CURRENT_SCHEMA_ID,