[GLUTEN-12538][VL] Unblock TIMESTAMP_NTZ min/max in Delta statistics

.github/workflows/util/delta-spark-ut/known-failures.txt

@@ -628,78 +628,36 @@ org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#Data s
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#Data skipping handles aliasing for _metadata fields - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping flags
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping flags - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping on TIMESTAMP_NTZ
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping on TIMESTAMP_NTZ - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping stats before and after optimize
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1JsonCheckpointV2Suite#data skipping stats before and after optimize - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping flags - column mapping name mode
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping flags - column mapping name mode - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping on TIMESTAMP_NTZ - column mapping name mode
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping on TIMESTAMP_NTZ - column mapping name mode - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping on TIMESTAMP_NTZ near Long.MaxValue - column mapping name mode
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping on TIMESTAMP_NTZ near Long.MaxValue - column mapping name mode - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping on TIMESTAMP_NTZ with Long.MaxValue - column mapping name mode
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping on TIMESTAMP_NTZ with Long.MaxValue - column mapping name mode - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping stats before and after optimize - column mapping name mode
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1NameColumnMappingSuite#data skipping stats before and after optimize - column mapping name mode - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#Data skipping handles aliasing for _metadata fields
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#Data skipping handles aliasing for _metadata fields - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping flags
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping flags - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping on TIMESTAMP_NTZ
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping on TIMESTAMP_NTZ - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping stats before and after optimize
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1ParquetCheckpointV2Suite#data skipping stats before and after optimize - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#Data skipping handles aliasing for _metadata fields
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#Data skipping handles aliasing for _metadata fields - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping flags
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping flags - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping on TIMESTAMP_NTZ
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping on TIMESTAMP_NTZ - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping stats before and after optimize
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1Suite#data skipping stats before and after optimize - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#Data skipping handles aliasing for _metadata fields
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#Data skipping handles aliasing for _metadata fields - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data skipping flags
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data skipping flags - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data skipping on TIMESTAMP_NTZ
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data skipping on TIMESTAMP_NTZ - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#Data skipping handles aliasing for _metadata fields
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#Data skipping handles aliasing for _metadata fields - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data skipping flags
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data skipping flags - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data skipping on TIMESTAMP_NTZ
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data skipping on TIMESTAMP_NTZ - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#Data skipping handles aliasing for _metadata fields
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#Data skipping handles aliasing for _metadata fields - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data skipping flags
 org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data skipping flags - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data skipping on TIMESTAMP_NTZ
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data skipping on TIMESTAMP_NTZ - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data skipping on TIMESTAMP_NTZ near Long.MaxValue - old behavior with DataFrame schema
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue
-org.apache.spark.sql.delta.stats.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data skipping on TIMESTAMP_NTZ with Long.MaxValue - old behavior with DataFrame schema
 org.apache.spark.sql.delta.stats.PartitionLikeDataSkippingColumnMappingSuite#partition-like data skipping for expression COALESCE: COALESCE(TO_DATE(S.b), c) = '1976-07-03' - column mapping id mode
 org.apache.spark.sql.delta.stats.StatsCollectionSuite#recompute stats multiple columns and files
 org.apache.spark.sql.delta.typewidening.TypeWideningAlterTableSuite#type widening DATE -> TIMESTAMP_NTZ, partitioned=false

backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala

@@ -39,12 +39,12 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Attribute, EmptyRow, Expression, Projection, SortOrder, SpecificInternalRow, UnsafeProjection}
 import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Complete, DeclarativeAggregate}
 import org.apache.spark.sql.catalyst.expressions.codegen.GenerateMutableProjection
-import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, LeafExecNode, ProjectExec}
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, LeafExecNode, ProjectExec, SparkPlan}
 import org.apache.spark.sql.execution.aggregate.SortAggregateExec
 import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, WriteJobStatsTracker, WriteTaskStats, WriteTaskStatsTracker}
 import org.apache.spark.sql.execution.metric.SQLMetric
 import org.apache.spark.sql.vectorized.ColumnarBatch
-import org.apache.spark.util.{SerializableConfiguration, SparkDirectoryUtil}
+import org.apache.spark.util.{SerializableConfiguration, SparkDirectoryUtil, Utils}
 
 import com.google.common.collect.Lists
 import org.apache.hadoop.conf.Configuration
@@ -86,6 +86,22 @@ private[stats] class GlutenDeltaJobStatsTracker(val delegate: DeltaJobStatistics
 }
 
 object GlutenDeltaJobStatsTracker extends Logging {
+  private val statsPlanObserverLock = new Object
+  @volatile private var statsPlanObserver: Option[(Path, SparkPlan) => Unit] = None
+
+  /** Observes task-local statistics plans in local-mode tests; callbacks run on task threads. */
+  private[delta] def withStatsPlanObserver[T](observer: (Path, SparkPlan) => Unit)(f: => T): T =
+    statsPlanObserverLock.synchronized {
+      require(Utils.isTesting, "Statistics plan observation is only available in tests")
+      require(statsPlanObserver.isEmpty, "A statistics plan observer is already registered")
+      statsPlanObserver = Some(observer)
+      try {
+        f
+      } finally {
+        statsPlanObserver = None
+      }
+    }
+
   def apply(tracker: WriteJobStatsTracker): WriteJobStatsTracker = tracker match {
     case tracker: BasicWriteJobStatsTracker =>
       new GlutenDeltaJobStatsRowCountingTracker(tracker)
@@ -171,6 +187,9 @@ object GlutenDeltaJobStatsTracker extends Logging {
         .asInstanceOf[WholeStageTransformer]
         .child
         .asInstanceOf[TransformSupport]
+      if (Utils.isTesting) {
+        statsPlanObserver.foreach(_(rootPath, wholeStageTransformer))
+      }
       val substraitContext = new SubstraitContext
       TransformerState.enterValidation
       val transformedNode =

backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/GlutenDeltaStatsSuite.scala

@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.delta
+
+import org.apache.gluten.execution.HashAggregateExecTransformer
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.stats.GlutenDeltaJobStatsTracker
+import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
+import org.apache.spark.sql.execution.SparkPlan
+
+import java.util.concurrent.ConcurrentLinkedQueue
+
+import scala.collection.JavaConverters._
+
+class GlutenDeltaStatsSuite extends DeltaSQLCommandTest {
+
+  import testImplicits._
+
+  test("collect TIMESTAMP_NTZ statistics natively") {
+    withSQLConf(DeltaSQLConf.DELTA_COLLECT_STATS.key -> "true") {
+      withTempDir {
+        dir =>
+          val path = dir.getCanonicalPath
+          val data = Seq(
+            "1969-12-31 23:59:59.999999",
+            "2024-01-01 00:00:00.123456"
+          ).toDF("input")
+            .selectExpr(
+              "cast(input as timestamp_ntz) as ts",
+              "struct(cast(input as timestamp_ntz) as ts) as nested")
+
+          val statsPlans = new ConcurrentLinkedQueue[SparkPlan]()
+          GlutenDeltaJobStatsTracker.withStatsPlanObserver {
+            (statsPath, plan) =>
+              if (statsPath.toUri.getPath == path) {
+                statsPlans.add(plan)
+              }
+          } {
+            data.coalesce(1).write.format("delta").save(path)
+          }
+          assert(!statsPlans.isEmpty, "The write did not use the native Delta statistics tracker")
+          statsPlans.asScala.foreach {
+            plan =>
+              assert(
+                plan.exists(_.isInstanceOf[HashAggregateExecTransformer]),
+                s"Expected a native statistics aggregate, but got:\n${plan.treeString}")
+          }
+
+          val actual = spark.read.format("delta").load(path)
+          assert(actual.collect().toSet == data.collect().toSet)
+
+          val addFiles = DeltaLog.forTable(spark, path).update().allFiles.collect()
+          assert(addFiles.length == 1)
+          val stats = addFiles.head.stats
+          assert(stats != null)
+          val statsValues = Seq(stats)
+            .toDF("stats")
+            .selectExpr(
+              "get_json_object(stats, '$.minValues.ts')",
+              "get_json_object(stats, '$.minValues.nested.ts')",
+              "get_json_object(stats, '$.maxValues.ts')",
+              "get_json_object(stats, '$.maxValues.nested.ts')"
+            )
+            .head()
+          assert(
+            statsValues == Row(
+              "1969-12-31T23:59:59.999",
+              "1969-12-31T23:59:59.999",
+              "2024-01-01T00:00:00.123",
+              "2024-01-01T00:00:00.123"),
+            stats)
+      }
+    }
+  }
+}

backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala

@@ -39,12 +39,12 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Attribute, BindReferences, EmptyRow, Expression, RuntimeReplaceable, SortOrder, SpecificInternalRow}
 import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Complete, DeclarativeAggregate}
 import org.apache.spark.sql.catalyst.expressions.codegen.GenerateMutableProjection
-import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, LeafExecNode, ProjectExec}
+import org.apache.spark.sql.execution.{ColumnarCollapseTransformStages, LeafExecNode, ProjectExec, SparkPlan}
 import org.apache.spark.sql.execution.aggregate.SortAggregateExec
 import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, WriteJobStatsTracker, WriteTaskStats, WriteTaskStatsTracker}
 import org.apache.spark.sql.execution.metric.SQLMetric
 import org.apache.spark.sql.vectorized.ColumnarBatch
-import org.apache.spark.util.{SerializableConfiguration, SparkDirectoryUtil}
+import org.apache.spark.util.{SerializableConfiguration, SparkDirectoryUtil, Utils}
 
 import com.google.common.collect.Lists
 import org.apache.hadoop.conf.Configuration
@@ -86,6 +86,22 @@ private[stats] class GlutenDeltaJobStatsTracker(val delegate: DeltaJobStatistics
 }
 
 object GlutenDeltaJobStatsTracker extends Logging {
+  private val statsPlanObserverLock = new Object
+  @volatile private var statsPlanObserver: Option[(Path, SparkPlan) => Unit] = None
+
+  /** Observes task-local statistics plans in local-mode tests; callbacks run on task threads. */
+  private[delta] def withStatsPlanObserver[T](observer: (Path, SparkPlan) => Unit)(f: => T): T =
+    statsPlanObserverLock.synchronized {
+      require(Utils.isTesting, "Statistics plan observation is only available in tests")
+      require(statsPlanObserver.isEmpty, "A statistics plan observer is already registered")
+      statsPlanObserver = Some(observer)
+      try {
+        f
+      } finally {
+        statsPlanObserver = None
+      }
+    }
+
   def apply(tracker: WriteJobStatsTracker): WriteJobStatsTracker = tracker match {
     case tracker: BasicWriteJobStatsTracker =>
       new GlutenDeltaJobStatsRowCountingTracker(tracker)
@@ -175,6 +191,9 @@ object GlutenDeltaJobStatsTracker extends Logging {
         .asInstanceOf[WholeStageTransformer]
         .child
         .asInstanceOf[TransformSupport]
+      if (Utils.isTesting) {
+        statsPlanObserver.foreach(_(rootPath, wholeStageTransformer))
+      }
       val substraitContext = new SubstraitContext
       TransformerState.enterValidation
       val transformedNode =