[core] Add more commit info logs (apache#5657)

Author: Zouxxyy

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -269,6 +269,10 @@ public void commit(
             ManifestCommittable committable,
             Map<String, String> properties,
             boolean checkAppendFiles) {
+        LOG.info(
+                "Ready to commit to table {}, number of commit messages: {}",
+                tableName,
+                committable.fileCommittables().size());
         if (LOG.isDebugEnabled()) {
             LOG.debug("Ready to commit\n{}", committable.toString());
         }
@@ -368,6 +372,7 @@ public void commit(
             }
         } finally {
             long commitDuration = (System.nanoTime() - started) / 1_000_000;
+            LOG.info("Finished commit to table {}, duration {} ms", tableName, commitDuration);
             if (this.commitMetrics != null) {
                 reportCommit(
                         appendTableFiles,
@@ -406,6 +411,10 @@ public void overwrite(
             Map<String, String> partition,
             ManifestCommittable committable,
             Map<String, String> properties) {
+        LOG.info(
+                "Ready to overwrite to table {}, number of commit messages: {}",
+                tableName,
+                committable.fileCommittables().size());
         if (LOG.isDebugEnabled()) {
             LOG.debug(
                     "Ready to overwrite partition {}\nManifestCommittable: {}\nProperties: {}",
@@ -513,6 +522,7 @@ public void overwrite(
             }
         } finally {
             long commitDuration = (System.nanoTime() - started) / 1_000_000;
+            LOG.info("Finished overwrite to table {}, duration {} ms", tableName, commitDuration);
             if (this.commitMetrics != null) {
                 reportCommit(
                         appendTableFiles,
@@ -701,6 +711,28 @@ private void collectChanges(
                                 }
                             });
         }
+        if (!commitMessages.isEmpty()) {
+            List<String> msg = new ArrayList<>();
+            if (appendTableFiles.size() > 0) {
+                msg.add(appendTableFiles.size() + " append table files");
+            }
+            if (appendChangelog.size() > 0) {
+                msg.add(appendChangelog.size() + " append Changelogs");
+            }
+            if (compactTableFiles.size() > 0) {
+                msg.add(compactTableFiles.size() + " compact table files");
+            }
+            if (compactChangelog.size() > 0) {
+                msg.add(compactChangelog.size() + " compact Changelogs");
+            }
+            if (appendHashIndexFiles.size() > 0) {
+                msg.add(appendHashIndexFiles.size() + " append hash index files");
+            }
+            if (compactDvIndexFiles.size() > 0) {
+                msg.add(compactDvIndexFiles.size() + " compact dv index files");
+            }
+            LOG.info("Finished collecting changes, including: {}", String.join(", ", msg));
+        }
     }
 
     private ManifestEntry makeEntry(FileKind kind, CommitMessage commitMessage, DataFileMeta file) {
@@ -1013,13 +1045,14 @@ CommitResult tryCommitOnce(
         }
 
         if (commitSnapshotImpl(newSnapshot, deltaStatistics)) {
-            if (LOG.isDebugEnabled()) {
-                LOG.debug(
-                        String.format(
-                                "Successfully commit snapshot #%d by user %s "
-                                        + "with identifier %s and kind %s.",
-                                newSnapshotId, commitUser, identifier, commitKind.name()));
-            }
+            LOG.info(
+                    "Successfully commit snapshot {} to table {} by user {} "
+                            + "with identifier {} and kind {}.",
+                    newSnapshotId,
+                    tableName,
+                    commitUser,
+                    identifier,
+                    commitKind.name());
             commitCallbacks.forEach(callback -> callback.call(deltaFiles, newSnapshot));
             return new SuccessResult();
         }

paimon-core/src/main/java/org/apache/paimon/table/ExpireSnapshotsImpl.java

@@ -135,6 +135,8 @@ public int expire() {
 
     @VisibleForTesting
     public int expireUntil(long earliestId, long endExclusiveId) {
+        long startTime = System.currentTimeMillis();
+
         if (endExclusiveId <= earliestId) {
             // No expire happens:
             // write the hint file in order to see the earliest snapshot directly next time
@@ -158,11 +160,6 @@ public int expireUntil(long earliestId, long endExclusiveId) {
             }
         }
 
-        if (LOG.isDebugEnabled()) {
-            LOG.debug(
-                    "Snapshot expire range is [" + beginInclusiveId + ", " + endExclusiveId + ")");
-        }
-
         List<Snapshot> taggedSnapshots = tagManager.taggedSnapshots();
 
         // delete merge tree files
@@ -269,6 +266,12 @@ public int expireUntil(long earliestId, long endExclusiveId) {
         }
 
         writeEarliestHint(endExclusiveId);
+        long duration = System.currentTimeMillis() - startTime;
+        LOG.info(
+                "Finished expire snapshots, duration {} ms, range is [{}, {})",
+                duration,
+                beginInclusiveId,
+                endExclusiveId);
         return (int) (endExclusiveId - beginInclusiveId);
     }