Copilot commented on code in PR #12967:
URL: https://github.com/apache/gluten/pull/12967#discussion_r4074765394
##########
cpp/velox/substrait/VeloxToSubstraitType.cc:
##########
@@ -31,6 +31,14 @@ const ::substrait::Type&
VeloxToSubstraitTypeConvertor::toSubstraitType(
substraitType->set_allocated_date(substraitDate);
return *substraitType;
}
+ if (type->equivalent(*velox::TIMESTAMP_UTC())) {
+ auto substraitPrecisionTimestamp =
+
google::protobuf::Arena::CreateMessage<::substrait::Type_PrecisionTimestamp>(&arena);
+ substraitPrecisionTimestamp->set_precision(6);
+
substraitPrecisionTimestamp->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
+
substraitType->set_allocated_precision_timestamp(substraitPrecisionTimestamp);
+ return *substraitType;
+ }
Review Comment:
Nullability is being hard-coded to `NULLABILITY_NULLABLE` for
`TIMESTAMP_UTC`. This can misrepresent NOT NULL schemas in the emitted
Substrait plan. Please set nullability consistently with how other types in
this method handle the `nullable` flag (or equivalent nullability input) rather
than forcing nullable.
##########
backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -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
+ }
+ }
Review Comment:
This introduces a JVM-global, mutable observer that can be accessed from
multiple task threads. Even though it’s gated behind `Utils.isTesting`, it can
still create cross-test interference if tests run concurrently in the same JVM
(one suite registering while another triggers observation). A more isolated
approach is to store the observer in a per-test dynamic scope (e.g.,
`scala.util.DynamicVariable`) or propagate an identifier via Spark local
properties and keep observers keyed per test, so concurrent tests don’t contend
on a single global slot.
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenTimestampNtzAggregateSuite.scala:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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
+
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.{HashAggregateExecBaseTransformer,
ProjectExecTransformer}
+
+import org.apache.spark.sql.execution.ProjectExec
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.functions.{max, min}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.TimestampNTZType
+
+import java.time.LocalDateTime
+
+class GlutenTimestampNtzAggregateSuite extends GlutenSQLTestsTrait {
+
+ import testImplicits._
+
+ testGluten("min and max") {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "false",
+ GlutenConfig.GLUTEN_ANSI_FALLBACK_ENABLED.key -> "false") {
+ withTempPath {
+ path =>
+ 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")
+ .write
+ .parquet(path.getCanonicalPath)
+
+ val result =
spark.read.parquet(path.getCanonicalPath).agg(min($"ts"), max($"ts"))
+ checkAnswer(
+ result,
+ Row(
+ LocalDateTime.parse("1969-12-31T23:59:59.999999"),
+ LocalDateTime.parse("2024-01-01T00:00:00.123456")))
+ assert(
+
getExecutedPlan(result).exists(_.isInstanceOf[HashAggregateExecBaseTransformer]),
+ result.queryExecution.executedPlan.treeString)
+ }
+ }
+ }
+
+ testGluten("min and max grouped by timestamp_ntz") {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2",
+ GlutenConfig.GLUTEN_ANSI_FALLBACK_ENABLED.key -> "false",
+ "spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation"
-> "false"
+ ) {
+ withTempPath {
+ path =>
+ val beforeEpoch = LocalDateTime.parse("1969-12-31T23:59:59.999999")
+ val afterEpoch = LocalDateTime.parse("1970-01-01T00:00:00.000001")
+ val firstKey = LocalDateTime.parse("2024-01-01T00:00:00.123456")
+ val secondKey = LocalDateTime.parse("2024-01-01T00:00:00.123457")
+ Seq(
+ (firstKey, beforeEpoch),
+ (firstKey, secondKey),
+ (firstKey, null),
+ (secondKey, afterEpoch),
+ (secondKey, firstKey),
+ (null, beforeEpoch),
+ (null, afterEpoch),
+ (beforeEpoch, null)
+ ).toDF("key", "ts")
+ .write
+ .parquet(path.getCanonicalPath)
+
+ val result = spark.read
+ .parquet(path.getCanonicalPath)
+ .groupBy($"key")
+ .agg(min($"ts"), max($"ts"))
+ checkAnswer(
+ result,
+ Seq(
+ Row(firstKey, beforeEpoch, secondKey),
+ Row(secondKey, afterEpoch, firstKey),
+ Row(null, beforeEpoch, afterEpoch),
+ Row(beforeEpoch, null, null)))
+ val aggregates = getExecutedPlan(result).collect {
+ case aggregate: BaseAggregateExec => aggregate
+ }
+ assert(aggregates.nonEmpty,
result.queryExecution.executedPlan.treeString)
+ assert(
+ aggregates.forall {
+ case aggregate: HashAggregateExecBaseTransformer =>
+ aggregate.groupingExpressions.map(_.dataType) ==
Seq(TimestampNTZType)
+ case _ => false
+ },
+ result.queryExecution.executedPlan.treeString
+ )
Review Comment:
This assertion is brittle because it requires *every* `BaseAggregateExec` in
the executed plan to be a `HashAggregateExecBaseTransformer`. If Spark inserts
additional aggregates (e.g., multiple stages) or a non-hash aggregate appears
alongside the transformer, the test will fail even if the intended native
aggregate is present. Prefer filtering/collecting
`HashAggregateExecBaseTransformer` nodes directly and asserting on those
(and/or separately asserting that at least one such node exists).
##########
backends-velox/src-delta40/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")
Review Comment:
The Delta 3.3 and Delta 4.0 versions of `GlutenDeltaStatsSuite` appear to be
identical in this PR. To reduce duplication and keep future changes consistent,
consider extracting the shared test logic into a common trait/helper (e.g.,
under a shared test source directory) and keeping only minimal version-specific
wiring in each module.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
