Log the full TableMetadata

polaris-service/src/main/java/org/apache/polaris/service/catalog/BasePolarisCatalog.java

@@ -88,6 +88,7 @@
 import org.apache.polaris.core.entity.TableLikeEntity;
 import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.core.persistence.PolarisObjectMapperUtil;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
 import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
 import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifestCatalogView;
@@ -1254,6 +1255,12 @@ public void doRefresh() {
     public void doCommit(TableMetadata base, TableMetadata metadata) {
       LOGGER.debug(
           "doCommit for table {} with base {}, metadata {}", tableIdentifier, base, metadata);
+      LOGGER
+          .atDebug()
+          .setMessage("doCommit full new metadata: {}")
+          .addArgument(
+              () -> PolarisObjectMapperUtil.serialize(getCurrentPolarisContext(), metadata))
+          .log();
       // TODO: Maybe avoid writing metadata if there's definitely a transaction conflict
       if (null == base && !namespaceExists(tableIdentifier.namespace())) {
         throw new NoSuchNamespaceException(

polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogTest.java

@@ -24,6 +24,9 @@
 import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.Mockito.when;
 
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterators;
 import java.io.IOException;
@@ -106,6 +109,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
+import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.sts.StsClient;
 import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
 import software.amazon.awssdk.services.sts.model.AssumeRoleResponse;
@@ -1524,4 +1528,46 @@ public void testFileIOWrapper() {
                 .getFirst()));
     Assertions.assertThat(measured.getNumDeletedFiles()).as("A table was deleted").isGreaterThan(0);
   }
+
+  @Test
+  public void testCommitLogsMetadata() {
+    Logger logger = (Logger) LoggerFactory.getLogger(BasePolarisCatalog.class);
+    ListAppender<ILoggingEvent> listAppender = new ListAppender<>();
+    listAppender.start();
+    logger.addAppender(listAppender);
+
+    RealmContext realmContext = () -> "realm";
+    CallContext callContext = CallContext.of(realmContext, polarisContext);
+    CallContext.setCurrentContext(callContext);
+    PolarisPassthroughResolutionView passthroughView =
+        new PolarisPassthroughResolutionView(
+            callContext, entityManager, authenticatedRoot, CATALOG_NAME);
+
+    BasePolarisCatalog catalog =
+        new BasePolarisCatalog(
+            entityManager,
+            metaStoreManager,
+            callContext,
+            passthroughView,
+            authenticatedRoot,
+            Mockito.mock(),
+            new TestFileIOFactory());
+    catalog.initialize(
+        CATALOG_NAME,
+        ImmutableMap.of(
+            CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO"));
+
+    catalog.createNamespace(NS);
+    catalog.buildTable(TABLE, SCHEMA).create();
+
+    Assertions.assertThat(
+            listAppender.list.stream()
+                .anyMatch(
+                    log ->
+                        log.getMessage().contains("doCommit full new metadata")
+                            && log.getFormattedMessage()
+                                .contains(SCHEMA.columns().getFirst().doc())))
+        .as("The full metadata log line should be emitted")
+        .isTrue();
+  }
 }