[core] Disable PARTITION_IDLE_TIME_TO_REPORT_STATISTIC by default

Author: JingsongLi

docs/layouts/shortcodes/generated/core_configuration.html

@@ -703,7 +703,7 @@
         </tr>
         <tr>
             <td><h5>partition.idle-time-to-report-statistic</h5></td>
-            <td style="word-wrap: break-word;">1 h</td>
+            <td style="word-wrap: break-word;">0 ms</td>
             <td>Duration</td>
             <td>Set a time duration when a partition has no new data after this time duration, start to report the partition statistics to hms.</td>
         </tr>

paimon-common/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1640,7 +1640,7 @@ public class CoreOptions implements Serializable {
     public static final ConfigOption<Duration> PARTITION_IDLE_TIME_TO_REPORT_STATISTIC =
             key("partition.idle-time-to-report-statistic")
                     .durationType()
-                    .defaultValue(Duration.ofHours(1))
+                    .defaultValue(Duration.ofMillis(0))
                     .withDescription(
                             "Set a time duration when a partition has no new data after this time duration, "
                                     + "start to report the partition statistics to hms.");

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala

@@ -27,8 +27,8 @@ import org.apache.paimon.deletionvectors.DeletionVector
 import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer
 import org.apache.paimon.index.{BucketAssigner, SimpleHashBucketAssigner}
 import org.apache.paimon.io.{CompactIncrement, DataIncrement, IndexIncrement}
-import org.apache.paimon.manifest.{FileKind, IndexManifestEntry}
-import org.apache.paimon.spark.{SparkInternalRowWrapper, SparkRow, SparkTableWrite, SparkTypeUtils}
+import org.apache.paimon.manifest.FileKind
+import org.apache.paimon.spark.{SparkRow, SparkTableWrite, SparkTypeUtils}
 import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL, ROW_KIND_COL}
 import org.apache.paimon.spark.util.OptionUtils.paimonExtensionEnabled
 import org.apache.paimon.spark.util.SparkRowUtils
@@ -40,10 +40,8 @@ import org.apache.paimon.utils.{InternalRowPartitionComputer, PartitionPathUtils
 
 import org.apache.spark.{Partitioner, TaskContext}
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{Column, DataFrame, Dataset, Row, SparkSession}
-import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
 import org.apache.spark.sql.functions._
-import org.apache.spark.sql.types.StructType
 import org.slf4j.LoggerFactory
 
 import java.io.IOException
@@ -59,6 +57,15 @@ case class PaimonSparkWriter(table: FileStoreTable) {
 
   private lazy val bucketMode = table.bucketMode
 
+  private lazy val disableReportStats = {
+    val options = table.coreOptions()
+    val config = options.toConfiguration
+    config.get(CoreOptions.PARTITION_IDLE_TIME_TO_REPORT_STATISTIC).toMillis <= 0 ||
+    table.partitionKeys.isEmpty ||
+    !options.partitionedTableInMetastore ||
+    table.catalogEnvironment.partitionHandler() == null
+  }
+
   private lazy val log = LoggerFactory.getLogger(classOf[PaimonSparkWriter])
 
   @transient private lazy val serializer = new CommitMessageSerializer
@@ -326,37 +333,12 @@ case class PaimonSparkWriter(table: FileStoreTable) {
       .map(deserializeCommitMessage(serializer, _))
   }
 
-  def buildCommitMessageFromIndexManifestEntry(
-      indexManifestEntries: Seq[IndexManifestEntry]): Seq[CommitMessage] = {
-    indexManifestEntries
-      .groupBy(entry => (entry.partition(), entry.bucket()))
-      .map {
-        case ((partition, bucket), entries) =>
-          val (added, removed) = entries.partition(_.kind() == FileKind.ADD)
-          new CommitMessageImpl(
-            partition,
-            bucket,
-            null,
-            DataIncrement.emptyIncrement(),
-            CompactIncrement.emptyIncrement(),
-            new IndexIncrement(added.map(_.indexFile()).asJava, removed.map(_.indexFile()).asJava))
-      }
-      .toSeq
-  }
-
   private def reportToHms(messages: Seq[CommitMessage]): Unit = {
-    val options = table.coreOptions()
-    val config = options.toConfiguration
-
-    if (
-      config.get(CoreOptions.PARTITION_IDLE_TIME_TO_REPORT_STATISTIC).toMillis <= 0 ||
-      table.partitionKeys.isEmpty ||
-      !options.partitionedTableInMetastore ||
-      table.catalogEnvironment.partitionHandler() == null
-    ) {
+    if (disableReportStats) {
       return
     }
 
+    val options = table.coreOptions()
     val partitionComputer = new InternalRowPartitionComputer(
       options.partitionDefaultName,
       table.schema.logicalPartitionType,