From 18a78d1795ce1f553b8f39f53da8ee6ef5dee30d Mon Sep 17 00:00:00 2001
From: Tien-Lung Fu <0050211@gmail.com>
Date: Tue, 3 Nov 2026 00:23:34 -0800
Subject: [PATCH 1/7] fix: fall back TopK whose later sort key can fail under
 ANSI

Spark's TakeOrderedAndProject ordering evaluates a sort key after the
first only for rows whose earlier keys tie, while Comet's native TopK
evaluates every key over the whole batch. Under ANSI mode a later key
such as `b + 1` could raise ARITHMETIC_OVERFLOW on a row Spark never
evaluates it for. Mark such a TopK Incompatible so it runs on Spark.

Part of https://github.com/apache/datafusion-comet/issues/7246

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 3ec1c07f-32b3-4cb1-a6ad-09293da13ab9
---
 .../latest/compatibility/operators.md         |  8 ++
 .../apache/comet/expressions/CometCast.scala  | 24 +++++-
 .../apache/comet/serde/QueryPlanSerde.scala   | 32 ++++++++
 .../CometTakeOrderedAndProjectExec.scala      | 11 ++-
 .../operators/topk_later_key_ansi.sql         | 76 +++++++++++++++++++
 5 files changed, 148 insertions(+), 3 deletions(-)
 create mode 100644 spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql

diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md
index 43f118942d1..3cd85d313c6 100644
--- a/docs/source/user-guide/latest/compatibility/operators.md
+++ b/docs/source/user-guide/latest/compatibility/operators.md
@@ -94,6 +94,14 @@ TopK, and ranking window order keys run natively under both default and nondefau
 searches boundaries over a nullable nested key still runs on Spark
 ([#6477](https://github.com/apache/datafusion-comet/issues/6477)).
 
+Spark's TopK (`TakeOrderedAndProject`, an `ORDER BY` with a `LIMIT`) evaluates a sort key after
+the first only for the rows whose earlier keys tie. Comet evaluates every key over the whole batch,
+so under `spark.sql.ansi.enabled=true` a later key such as `b + 1` could raise
+`ARITHMETIC_OVERFLOW` on a row that Spark never evaluates it for. A TopK whose later sort key can
+fail therefore runs on Spark, unless
+`spark.comet.operator.TakeOrderedAndProjectExec.allowIncompatible=true`
+([#7246](https://github.com/apache/datafusion-comet/issues/7246)).
+
 ## Hash Aggregation
 
 Ungrouped `AVG` and `TRY_AVG` on `DECIMAL(p, s)` fall back to Spark when the intermediate sum
diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index f57b6074a81..0b2d2904c79 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -24,7 +24,7 @@ import scala.util.Try
 import org.apache.spark.SPARK_VERSION
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression, Literal}
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DecimalType, MapType, NullType, NumericType, StructType, TimestampNTZType, TimestampType}
+import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DecimalType, MapType, NullType, NumericType, StringType, StructType, TimestampNTZType, TimestampType}
 
 import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, withFallbackReason}
 import org.apache.comet.DataTypeSupport.{containsDuplicateFieldNames, isComplexType}
@@ -572,6 +572,28 @@ object CometCast
   private[comet] def canProduceNullMapKey(cast: Cast): Boolean =
     evalMode(cast) == CometEvalMode.TRY && mayNullMapKey(cast.child.dataType, cast.dataType)
 
+  /**
+   * Whether `cast` cannot raise an error for any value. A TRY cast returns NULL for a value it
+   * cannot convert, and so does a LEGACY cast between the atomic types listed here. An ANSI cast
+   * cannot fail only where the conversion is always exact, such as a widening one.
+   */
+  private[comet] def cannotFail(cast: Cast): Boolean = {
+    val from = cast.child.dataType
+    val to = cast.dataType
+    evalMode(cast) match {
+      case CometEvalMode.TRY => true
+      case CometEvalMode.LEGACY if isAtomicCastType(from) && isAtomicCastType(to) => true
+      case _ => keyCastCannotFail(from, to)
+    }
+  }
+
+  private def isAtomicCastType(dataType: DataType): Boolean = dataType match {
+    case _: NumericType | _: StringType | DataTypes.BooleanType | DataTypes.BinaryType |
+        DataTypes.DateType | DataTypes.TimestampType | _: TimestampNTZType =>
+      true
+    case _ => false
+  }
+
   private def mayNullMapKey(fromType: DataType, toType: DataType): Boolean =
     (fromType, toType) match {
       case (MapType(fromKey, fromValue, _), MapType(toKey, toValue, _)) =>
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index c8b89b0742c..36944f2e256 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke}
 import org.apache.spark.sql.catalyst.expressions.xml.{XPathBoolean, XPathDouble, XPathFloat, XPathInt, XPathList, XPathLong, XPathShort, XPathString}
+import org.apache.spark.sql.catalyst.optimizer.NormalizeNaNAndZero
 import org.apache.spark.sql.comet.DecimalPrecision
 import org.apache.spark.sql.execution.{ScalarSubquery, SparkPlan}
 import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils
@@ -1394,6 +1395,37 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
     }
   }
 
+  /**
+   * Whether `expr` cannot raise an error for any row, so native execution may evaluate it for
+   * rows that Spark skips. Like the native `cannot_fail`, it is conservative: columns, literals,
+   * and comparisons, boolean logic, null checks, conditionals, numeric arithmetic, negation and
+   * absolute value outside ANSI mode and casts that cannot fail over them. Anything else is
+   * treated as able to fail.
+   */
+  def cannotFail(expr: Expression): Boolean = {
+    def numeric: Boolean = expr.dataType.isInstanceOf[NumericType]
+    def nonAnsiNumeric(evalMode: EvalMode.Value): Boolean = evalMode != EvalMode.ANSI && numeric
+    val nodeCannotFail = expr match {
+      case _: Attribute | _: BoundReference | _: Literal => true
+      case _: BinaryComparison | _: And | _: Or | _: Not | _: IsNull | _: IsNotNull => true
+      case _: If | _: CaseWhen | _: Coalesce => true
+      case _: KnownFloatingPointNormalized | _: NormalizeNaNAndZero => true
+      case c: Cast => CometCast.cannotFail(c)
+      case _: UnaryPositive => true
+      case u: UnaryMinus => !u.failOnError && numeric
+      case a: Abs => !a.failOnError && numeric
+      case a: Add => nonAnsiNumeric(a.evalMode)
+      case s: Subtract => nonAnsiNumeric(s.evalMode)
+      case m: Multiply => nonAnsiNumeric(m.evalMode)
+      case d: Divide => nonAnsiNumeric(d.evalMode)
+      case d: IntegralDivide => nonAnsiNumeric(d.evalMode)
+      case r: Remainder => nonAnsiNumeric(r.evalMode)
+      case p: Pmod => nonAnsiNumeric(p.evalMode)
+      case _ => false
+    }
+    nodeCannotFail && expr.children.forall(cannotFail)
+  }
+
   /**
    * Has native execution evaluate the arguments of `scalarFunc`, the serialized `expr`, the way
    * Spark evaluates those of a null-intolerant `BinaryExpression` or `TernaryExpression`: each
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala
index f9597ca046b..1ff1af4ffb3 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala
@@ -33,8 +33,8 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
 
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometSparkSessionExtensions.isCometShuffleEnabled
-import org.apache.comet.serde.{Compatible, OperatorOuterClass, SupportLevel, Unsupported}
-import org.apache.comet.serde.QueryPlanSerde.{exprToProto, supportedSortType, withSharedSubexpressions}
+import org.apache.comet.serde.{Compatible, Incompatible, OperatorOuterClass, SupportLevel, Unsupported}
+import org.apache.comet.serde.QueryPlanSerde.{cannotFail, exprToProto, supportedSortType, withSharedSubexpressions}
 import org.apache.comet.serde.operator.CometSink
 
 object CometTakeOrderedAndProjectExec extends CometSink[TakeOrderedAndProjectExec] {
@@ -63,6 +63,13 @@ object CometTakeOrderedAndProjectExec extends CometSink[TakeOrderedAndProjectExe
     if (!supportedSortType(op, op.sortOrder)) {
       return Unsupported(Some("unsupported data type in sort order"))
     }
+    // Spark's ordering evaluates a sort key only for the rows whose earlier keys tie, but native
+    // execution evaluates every key over the whole batch, so a later key that can fail may raise
+    // on a row that Spark skips (#7246).
+    op.sortOrder.drop(1).find(s => !cannotFail(s.child)).foreach { s =>
+      return Incompatible(
+        Some(s"sort key ${s.child.sql} can fail on rows that Spark's ordering skips"))
+    }
     Compatible()
   }
 
diff --git a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
new file mode 100644
index 00000000000..8288fad7ecb
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
@@ -0,0 +1,76 @@
+-- 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.
+
+-- Spark's top-K ordering evaluates a later sort key only for rows whose earlier keys tie, so a
+-- later key that fails under ANSI mode does not fail on a row Spark skips. Comet evaluates every
+-- key over the whole batch, so a top-K whose later key can fail runs on Spark (#7246).
+-- Each table is one batch.
+
+-- Config: spark.sql.ansi.enabled=true
+-- Config: spark.sql.adaptive.enabled=false
+
+statement
+CREATE TABLE topk(a int, b int) USING parquet
+
+statement
+INSERT INTO topk SELECT /*+ REPARTITION(1) */ * FROM VALUES
+  (1, 2147483647), (2, 5), (3, 2147483647), (4, 7), (5, 9) AS t(a, b)
+
+-- No two rows tie on a, so Spark never evaluates b + 1
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, b FROM topk ORDER BY a, b + 1 LIMIT 2
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, b FROM topk ORDER BY a DESC, CAST(CAST(b AS STRING) AS INT) * 2 LIMIT 2
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, b FROM topk ORDER BY a, abs(b) LIMIT 2
+
+-- Later keys that cannot fail stay native
+query
+SELECT a, b FROM topk ORDER BY a, b LIMIT 2
+
+query
+SELECT a, b FROM topk ORDER BY a, b IS NULL, CAST(b AS BIGINT), try_add(b, 1) LIMIT 2
+
+-- The first key is evaluated for every row in Spark too
+query
+SELECT a, b FROM topk ORDER BY a + 1, b LIMIT 2
+
+statement
+CREATE TABLE topk_ties(a int, b int) USING parquet
+
+statement
+INSERT INTO topk_ties SELECT /*+ REPARTITION(1) */ * FROM VALUES
+  (1, 2147483647), (1, 5), (2, 7) AS t(a, b)
+
+-- Rows tie on a, so Spark evaluates b + 1 and fails
+query expect_error(ARITHMETIC_OVERFLOW)
+SELECT a, b FROM topk_ties ORDER BY a, b + 1 LIMIT 2
+
+-- Without ANSI mode the key cannot fail and stays native
+statement
+SET spark.sql.ansi.enabled=false
+
+query
+SELECT a, b FROM topk ORDER BY a, b + 1 LIMIT 2
+
+query
+SELECT a, b FROM topk_ties ORDER BY a, b + 1 LIMIT 2
+
+query
+SELECT a, b FROM topk_ties ORDER BY a, -b, abs(b) LIMIT 2

From 307fb82f6ac05b383e86bffc95f71ad7cfe7166c Mon Sep 17 00:00:00 2001
From: Tien-Lung Fu <0050211@gmail.com>
Date: Tue, 3 Nov 2026 01:31:20 -0800
Subject: [PATCH 2/7] fix: keep TopK native for string keys that cannot fail

Whitelist Substring, Upper, Lower and Length, which never raise, so a
TopK such as TPC-DS q79 stays native. Allow the incompatible TopK in
array_null_short_circuit.sql, whose later key holds an ANSI cast that
can fail, to keep testing that TopK keys skip a shared cast.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 3ec1c07f-32b3-4cb1-a6ad-09293da13ab9
---
 .../scala/org/apache/comet/serde/QueryPlanSerde.scala    | 7 ++++---
 .../expressions/array/array_null_short_circuit.sql       | 9 ++++++++-
 .../sql-tests/operators/topk_later_key_ansi.sql          | 4 ++++
 3 files changed, 16 insertions(+), 4 deletions(-)

diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index 36944f2e256..c8fa0da3b41 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -1398,9 +1398,9 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
   /**
    * Whether `expr` cannot raise an error for any row, so native execution may evaluate it for
    * rows that Spark skips. Like the native `cannot_fail`, it is conservative: columns, literals,
-   * and comparisons, boolean logic, null checks, conditionals, numeric arithmetic, negation and
-   * absolute value outside ANSI mode and casts that cannot fail over them. Anything else is
-   * treated as able to fail.
+   * and comparisons, boolean logic, null checks, conditionals, substrings, case conversion and
+   * length, numeric arithmetic, negation and absolute value outside ANSI mode and casts that
+   * cannot fail over them. Anything else is treated as able to fail.
    */
   def cannotFail(expr: Expression): Boolean = {
     def numeric: Boolean = expr.dataType.isInstanceOf[NumericType]
@@ -1410,6 +1410,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       case _: BinaryComparison | _: And | _: Or | _: Not | _: IsNull | _: IsNotNull => true
       case _: If | _: CaseWhen | _: Coalesce => true
       case _: KnownFloatingPointNormalized | _: NormalizeNaNAndZero => true
+      case _: Substring | _: Upper | _: Lower | _: Length => true
       case c: Cast => CometCast.cannotFail(c)
       case _: UnaryPositive => true
       case u: UnaryMinus => !u.failOnError && numeric
diff --git a/spark/src/test/resources/sql-tests/expressions/array/array_null_short_circuit.sql b/spark/src/test/resources/sql-tests/expressions/array/array_null_short_circuit.sql
index 559574fc38f..9eec890d230 100644
--- a/spark/src/test/resources/sql-tests/expressions/array/array_null_short_circuit.sql
+++ b/spark/src/test/resources/sql-tests/expressions/array/array_null_short_circuit.sql
@@ -108,11 +108,18 @@ SELECT array_contains(ai, CAST(s AS INT))
 FROM (SELECT * FROM array_null_short_circuit ORDER BY i LIMIT 10)
 
 -- A TakeOrderedAndProject compares rows one sort key at a time and evaluates its projection apart
--- from the keys, so a cast in two sort keys, or in a sort key and the projection, is skipped too
+-- from the keys, so a cast in two sort keys, or in a sort key and the projection, is skipped too.
+-- A later sort key that can fail runs the top-K on Spark (#7246), so allow it here.
+statement
+SET spark.comet.operator.TakeOrderedAndProjectExec.allowIncompatible=true
+
 query expect_native(array_contains, array_position)
 SELECT i FROM (SELECT * FROM array_null_short_circuit
   ORDER BY array_contains(ai, CAST(s AS INT)), array_position(ai, CAST(s AS INT)), i LIMIT 10)
 
+statement
+RESET spark.comet.operator.TakeOrderedAndProjectExec.allowIncompatible
+
 query expect_native(array_contains, array_position)
 SELECT array_contains(ai, CAST(s AS INT))
 FROM (SELECT * FROM array_null_short_circuit ORDER BY array_position(ai, CAST(s AS INT)), i LIMIT 10)
diff --git a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
index 8288fad7ecb..ca8317ea174 100644
--- a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
+++ b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
@@ -47,6 +47,10 @@ SELECT a, b FROM topk ORDER BY a, b LIMIT 2
 query
 SELECT a, b FROM topk ORDER BY a, b IS NULL, CAST(b AS BIGINT), try_add(b, 1) LIMIT 2
 
+query
+SELECT a, b FROM topk ORDER BY a, substr(CAST(b AS STRING), 1, 3), upper(CAST(b AS STRING)),
+  length(lower(CAST(b AS STRING))) LIMIT 2
+
 -- The first key is evaluated for every row in Spark too
 query
 SELECT a, b FROM topk ORDER BY a + 1, b LIMIT 2

From fa0d55349b8073391f85be93b9142a9842ad08ce Mon Sep 17 00:00:00 2001
From: Tien-Lung Fu <0050211@gmail.com>
Date: Tue, 3 Nov 2026 15:27:22 -0800
Subject: [PATCH 3/7] fix: widen TopK later-key whitelist and address review

Extract laterSortKeyThatCanFail helper, add string, datetime, math and
hash functions that cannot fail in Spark, treat string-to-binary casts as
infallible, extend tests and reword the Sort compatibility notes.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 3ec1c07f-32b3-4cb1-a6ad-09293da13ab9
---
 .../latest/compatibility/operators.md         | 10 +++-
 .../apache/comet/expressions/CometCast.scala  |  4 +-
 .../apache/comet/serde/QueryPlanSerde.scala   | 43 +++++++++++++--
 .../CometTakeOrderedAndProjectExec.scala      |  7 +--
 .../operators/topk_later_key_ansi.sql         | 55 +++++++++++++++++++
 5 files changed, 104 insertions(+), 15 deletions(-)

diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md
index 3cd85d313c6..18b8aec6c9b 100644
--- a/docs/source/user-guide/latest/compatibility/operators.md
+++ b/docs/source/user-guide/latest/compatibility/operators.md
@@ -97,10 +97,14 @@ searches boundaries over a nullable nested key still runs on Spark
 Spark's TopK (`TakeOrderedAndProject`, an `ORDER BY` with a `LIMIT`) evaluates a sort key after
 the first only for the rows whose earlier keys tie. Comet evaluates every key over the whole batch,
 so under `spark.sql.ansi.enabled=true` a later key such as `b + 1` could raise
-`ARITHMETIC_OVERFLOW` on a row that Spark never evaluates it for. A TopK whose later sort key can
-fail therefore runs on Spark, unless
+`ARITHMETIC_OVERFLOW` on a row that Spark never evaluates it for. A TopK therefore runs on Spark
+when Comet cannot show that a later sort key never fails, unless
 `spark.comet.operator.TakeOrderedAndProjectExec.allowIncompatible=true`
-([#7246](https://github.com/apache/datafusion-comet/issues/7246)).
+([#7246](https://github.com/apache/datafusion-comet/issues/7246)). The check does not depend on
+`spark.sql.ansi.enabled`: a later key built from an expression outside the set Comet recognizes as
+unable to fail runs on Spark in either mode. A sort without a `LIMIT` that has no exchange beneath
+it, such as `SORT BY`, still runs natively and can raise such an error
+([#7470](https://github.com/apache/datafusion-comet/issues/7470)).
 
 ## Hash Aggregation
 
diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index 0b2d2904c79..4975fda2fb4 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -575,7 +575,8 @@ object CometCast
   /**
    * Whether `cast` cannot raise an error for any value. A TRY cast returns NULL for a value it
    * cannot convert, and so does a LEGACY cast between the atomic types listed here. An ANSI cast
-   * cannot fail only where the conversion is always exact, such as a widening one.
+   * cannot fail only where the conversion is always exact, such as a widening one or a string to
+   * its bytes.
    */
   private[comet] def cannotFail(cast: Cast): Boolean = {
     val from = cast.child.dataType
@@ -583,6 +584,7 @@ object CometCast
     evalMode(cast) match {
       case CometEvalMode.TRY => true
       case CometEvalMode.LEGACY if isAtomicCastType(from) && isAtomicCastType(to) => true
+      case _ if from.isInstanceOf[StringType] && to == DataTypes.BinaryType => true
       case _ => keyCastCannotFail(from, to)
     }
   }
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index c8fa0da3b41..dc750cdd3e1 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -1397,10 +1397,10 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
 
   /**
    * Whether `expr` cannot raise an error for any row, so native execution may evaluate it for
-   * rows that Spark skips. Like the native `cannot_fail`, it is conservative: columns, literals,
-   * and comparisons, boolean logic, null checks, conditionals, substrings, case conversion and
-   * length, numeric arithmetic, negation and absolute value outside ANSI mode and casts that
-   * cannot fail over them. Anything else is treated as able to fail.
+   * rows that Spark skips. It answers from Spark's semantics, independently of the native
+   * `cannot_fail` in `case_when.rs`, which judges the physical expressions it is given. It is
+   * conservative: an allowed expression whose children also cannot fail. Anything not listed is
+   * treated as able to fail.
    */
   def cannotFail(expr: Expression): Boolean = {
     def numeric: Boolean = expr.dataType.isInstanceOf[NumericType]
@@ -1408,9 +1408,25 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
     val nodeCannotFail = expr match {
       case _: Attribute | _: BoundReference | _: Literal => true
       case _: BinaryComparison | _: And | _: Or | _: Not | _: IsNull | _: IsNotNull => true
-      case _: If | _: CaseWhen | _: Coalesce => true
+      case _: In | _: InSet | _: Greatest | _: Least => true
+      case _: If | _: CaseWhen | _: Coalesce | _: IsNaN | _: NaNvl => true
       case _: KnownFloatingPointNormalized | _: NormalizeNaNAndZero => true
-      case _: Substring | _: Upper | _: Lower | _: Length => true
+      case _: Substring | _: Upper | _: Lower | _: Length | _: OctetLength | _: BitLength => true
+      case _: StringTrim | _: StringTrimLeft | _: StringTrimRight => true
+      case _: StringReplace | _: InitCap => true
+      // These fail only when the result would exceed 2GB, more than a row can hold
+      case _: ConcatWs => true
+      case c: Concat => c.children.nonEmpty && c.dataType == StringType
+      case _: Reverse => expr.dataType == StringType
+      case _: Year | _: Quarter | _: Month | _: DayOfMonth | _: DayOfWeek | _: WeekDay |
+          _: DayOfYear | _: Hour | _: Minute | _: Second =>
+        true
+      case _: Murmur3Hash | _: XxHash64 | _: Md5 | _: Sha1 | _: Sha2 | _: Crc32 => true
+      // Invalid input gives NaN or NULL, and a floating-point floor or ceil saturates to a long
+      case _: Floor | _: Ceil | _: Sqrt | _: Exp | _: Log | _: Signum => true
+      // Rounding an integer fails on overflow in ANSI mode; a decimal is never treated as safe
+      case r: Round => roundCannotFail(r.dataType, r.ansiEnabled)
+      case r: BRound => roundCannotFail(r.dataType, r.ansiEnabled)
       case c: Cast => CometCast.cannotFail(c)
       case _: UnaryPositive => true
       case u: UnaryMinus => !u.failOnError && numeric
@@ -1427,6 +1443,21 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
     nodeCannotFail && expr.children.forall(cannotFail)
   }
 
+  private def roundCannotFail(dataType: DataType, ansiEnabled: Boolean): Boolean =
+    dataType match {
+      case FloatType | DoubleType => true
+      case ByteType | ShortType | IntegerType | LongType => !ansiEnabled
+      case _ => false
+    }
+
+  /**
+   * The first sort key after the first that can fail. Spark's ordering evaluates a sort key only
+   * for the rows whose earlier keys tie, but native execution evaluates every key over the whole
+   * batch, so such a key may raise on a row that Spark skips (#7246).
+   */
+  def laterSortKeyThatCanFail(sortOrder: Seq[SortOrder]): Option[SortOrder] =
+    sortOrder.drop(1).find(s => !cannotFail(s.child))
+
   /**
    * Has native execution evaluate the arguments of `scalarFunc`, the serialized `expr`, the way
    * Spark evaluates those of a null-intolerant `BinaryExpression` or `TernaryExpression`: each
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala
index 1ff1af4ffb3..a68b4b9acc1 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala
@@ -34,7 +34,7 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometSparkSessionExtensions.isCometShuffleEnabled
 import org.apache.comet.serde.{Compatible, Incompatible, OperatorOuterClass, SupportLevel, Unsupported}
-import org.apache.comet.serde.QueryPlanSerde.{cannotFail, exprToProto, supportedSortType, withSharedSubexpressions}
+import org.apache.comet.serde.QueryPlanSerde.{exprToProto, laterSortKeyThatCanFail, supportedSortType, withSharedSubexpressions}
 import org.apache.comet.serde.operator.CometSink
 
 object CometTakeOrderedAndProjectExec extends CometSink[TakeOrderedAndProjectExec] {
@@ -63,10 +63,7 @@ object CometTakeOrderedAndProjectExec extends CometSink[TakeOrderedAndProjectExe
     if (!supportedSortType(op, op.sortOrder)) {
       return Unsupported(Some("unsupported data type in sort order"))
     }
-    // Spark's ordering evaluates a sort key only for the rows whose earlier keys tie, but native
-    // execution evaluates every key over the whole batch, so a later key that can fail may raise
-    // on a row that Spark skips (#7246).
-    op.sortOrder.drop(1).find(s => !cannotFail(s.child)).foreach { s =>
+    laterSortKeyThatCanFail(op.sortOrder).foreach { s =>
       return Incompatible(
         Some(s"sort key ${s.child.sql} can fail on rows that Spark's ordering skips"))
     }
diff --git a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
index ca8317ea174..5060421820f 100644
--- a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
+++ b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
@@ -51,6 +51,55 @@ query
 SELECT a, b FROM topk ORDER BY a, substr(CAST(b AS STRING), 1, 3), upper(CAST(b AS STRING)),
   length(lower(CAST(b AS STRING))) LIMIT 2
 
+-- A later key that can fail falls back even when an allowed expression wraps it
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, b FROM topk ORDER BY a, coalesce(b + 1, 0) LIMIT 2
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, b FROM topk ORDER BY a, IF(a > 3, b + 1, 0) LIMIT 2
+
+-- Rounding an integer overflows under ANSI mode
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, b FROM topk ORDER BY a, round(b, -1) LIMIT 2
+
+query
+SELECT a, b FROM topk ORDER BY a, try_cast(CAST(b AS STRING) AS INT) LIMIT 2
+
+statement
+CREATE TABLE topk_types(a int, b int, l bigint, d double, s string, dt date, ts timestamp)
+USING parquet
+
+statement
+INSERT INTO topk_types SELECT /*+ REPARTITION(1) */ * FROM VALUES
+  (1, 2147483647, 9223372036854775807, double('NaN'), ' x ', date'2024-01-31',
+    timestamp'2024-01-31 23:59:59'),
+  (2, NULL, NULL, -0.0D, NULL, NULL, NULL),
+  (3, -2147483648, -9223372036854775808, 0.0D, 'abc', date'1970-01-01',
+    timestamp'1970-01-01 00:00:00'),
+  (4, 7, 7, 1.5D, 'Spark', date'2000-02-29', timestamp'2000-02-29 12:30:45'),
+  (5, 0, 0, -1.0E308D, '', date'9999-12-31', timestamp'9999-12-31 23:59:59')
+  AS t(a, b, l, d, s, dt, ts)
+
+-- Later keys built from functions that cannot fail stay native
+query
+SELECT a, s FROM topk_types ORDER BY a, trim(s), ltrim(s), rtrim(s), replace(s, 'a', 'b'),
+  reverse(s), octet_length(s), bit_length(s), concat(s, 'x'), concat_ws('-', s, s) LIMIT 3
+
+query
+SELECT a, dt FROM topk_types ORDER BY a, year(dt), quarter(dt), month(dt), dayofmonth(dt),
+  dayofweek(dt), weekday(dt), dayofyear(dt), hour(ts), minute(ts), second(ts) LIMIT 3
+
+query
+SELECT a, d FROM topk_types ORDER BY a, floor(d), ceil(d), sqrt(d), exp(d), ln(d), signum(d),
+  isnan(d), nanvl(d, 0D), round(d, 1), greatest(b, 0), least(l, 0), b IN (1, 2) LIMIT 3
+
+query
+SELECT a, s FROM topk_types ORDER BY a, hash(b, s), xxhash64(s), md5(s), sha1(s), sha2(s, 256),
+  crc32(s) LIMIT 3
+
+query
+SELECT a, s FROM topk_types ORDER BY a, initcap(s) LIMIT 3
+
 -- The first key is evaluated for every row in Spark too
 query
 SELECT a, b FROM topk ORDER BY a + 1, b LIMIT 2
@@ -78,3 +127,9 @@ SELECT a, b FROM topk_ties ORDER BY a, b + 1 LIMIT 2
 
 query
 SELECT a, b FROM topk_ties ORDER BY a, -b, abs(b) LIMIT 2
+
+query
+SELECT a, b FROM topk ORDER BY a, round(b, -1), CAST(CAST(b AS STRING) AS INT) LIMIT 2
+
+query
+SELECT a, b FROM topk_types ORDER BY a, b / 0, b % 0, b DIV 0, l + 1, l * 2 LIMIT 3

From f073526c7bcd57908b63aee95157301371edb636 Mon Sep 17 00:00:00 2001
From: Tien-Lung Fu <0050211@gmail.com>
Date: Tue, 3 Nov 2026 15:55:21 -0800
Subject: [PATCH 4/7] fix: treat timestamp casts as fallible TopK later keys

Native timestamp cast kernels go through chrono, which rejects years Spark
accepts, and raise even for TRY casts.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 3ec1c07f-32b3-4cb1-a6ad-09293da13ab9
---
 .../scala/org/apache/comet/expressions/CometCast.scala |  8 +++++---
 .../sql-tests/operators/topk_later_key_ansi.sql        | 10 ++++++++++
 2 files changed, 15 insertions(+), 3 deletions(-)

diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index 4975fda2fb4..cb4016ea719 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -576,17 +576,19 @@ object CometCast
    * Whether `cast` cannot raise an error for any value. A TRY cast returns NULL for a value it
    * cannot convert, and so does a LEGACY cast between the atomic types listed here. An ANSI cast
    * cannot fail only where the conversion is always exact, such as a widening one or a string to
-   * its bytes.
+   * its bytes. A cast to or from a timestamp is excluded in every mode: the native kernels go
+   * through chrono, which rejects years Spark accepts, and raise even for TRY.
    */
   private[comet] def cannotFail(cast: Cast): Boolean = {
     val from = cast.child.dataType
     val to = cast.dataType
-    evalMode(cast) match {
+    def isTimestamp(t: DataType) = t == TimestampType || t == TimestampNTZType
+    !isTimestamp(from) && !isTimestamp(to) && (evalMode(cast) match {
       case CometEvalMode.TRY => true
       case CometEvalMode.LEGACY if isAtomicCastType(from) && isAtomicCastType(to) => true
       case _ if from.isInstanceOf[StringType] && to == DataTypes.BinaryType => true
       case _ => keyCastCannotFail(from, to)
-    }
+    })
   }
 
   private def isAtomicCastType(dataType: DataType): Boolean = dataType match {
diff --git a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
index 5060421820f..468d16856d0 100644
--- a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
+++ b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
@@ -100,6 +100,16 @@ SELECT a, s FROM topk_types ORDER BY a, hash(b, s), xxhash64(s), md5(s), sha1(s)
 query
 SELECT a, s FROM topk_types ORDER BY a, initcap(s) LIMIT 3
 
+-- Native timestamp casts reject years that Spark accepts, even for TRY
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, ts FROM topk_types ORDER BY a, CAST(ts AS STRING) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, ts FROM topk_types ORDER BY a, try_cast(ts AS DATE) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, dt FROM topk_types ORDER BY a, try_cast(dt AS TIMESTAMP) LIMIT 3
+
 -- The first key is evaluated for every row in Spark too
 query
 SELECT a, b FROM topk ORDER BY a + 1, b LIMIT 2

From 12509a2f5235f55a2c33ed587ff5fcf478d5d73f Mon Sep 17 00:00:00 2001
From: Tien-Lung Fu <0050211@gmail.com>
Date: Thu, 5 Nov 2026 17:41:39 -0800
Subject: [PATCH 5/7] Address review: allow float arithmetic and struct fields,
 exclude date to string

Floating-point add, subtract, multiply, negation and abs cannot fail in any
mode, and reading a struct field cannot fail, so a later TopK key built from
them stays native. A cast from a date to a string raises natively for years
beyond chrono's range in every mode, so it is treated as able to fail. The
test comment about the first key now notes it holds for at least two rows
(#7527).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 3ec1c07f-32b3-4cb1-a6ad-09293da13ab9
---
 .../apache/comet/expressions/CometCast.scala  |  6 +++--
 .../apache/comet/serde/QueryPlanSerde.scala   | 13 ++++++----
 .../operators/topk_later_key_ansi.sql         | 25 ++++++++++++++++++-
 3 files changed, 36 insertions(+), 8 deletions(-)

diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index cb4016ea719..7cfdda3e9ec 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -577,13 +577,15 @@ object CometCast
    * cannot convert, and so does a LEGACY cast between the atomic types listed here. An ANSI cast
    * cannot fail only where the conversion is always exact, such as a widening one or a string to
    * its bytes. A cast to or from a timestamp is excluded in every mode: the native kernels go
-   * through chrono, which rejects years Spark accepts, and raise even for TRY.
+   * through chrono, which rejects years Spark accepts, and raise even for TRY. So is a cast from
+   * a date to a string, whose native formatting raises in every mode for such a year.
    */
   private[comet] def cannotFail(cast: Cast): Boolean = {
     val from = cast.child.dataType
     val to = cast.dataType
     def isTimestamp(t: DataType) = t == TimestampType || t == TimestampNTZType
-    !isTimestamp(from) && !isTimestamp(to) && (evalMode(cast) match {
+    val dateToString = from == DataTypes.DateType && to.isInstanceOf[StringType]
+    !isTimestamp(from) && !isTimestamp(to) && !dateToString && (evalMode(cast) match {
       case CometEvalMode.TRY => true
       case CometEvalMode.LEGACY if isAtomicCastType(from) && isAtomicCastType(to) => true
       case _ if from.isInstanceOf[StringType] && to == DataTypes.BinaryType => true
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index dc750cdd3e1..552f72bbfd6 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -1405,8 +1405,11 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
   def cannotFail(expr: Expression): Boolean = {
     def numeric: Boolean = expr.dataType.isInstanceOf[NumericType]
     def nonAnsiNumeric(evalMode: EvalMode.Value): Boolean = evalMode != EvalMode.ANSI && numeric
+    // Floating-point arithmetic never overflows into an error, even in ANSI mode
+    def floating: Boolean = expr.dataType == FloatType || expr.dataType == DoubleType
     val nodeCannotFail = expr match {
       case _: Attribute | _: BoundReference | _: Literal => true
+      case _: GetStructField => true
       case _: BinaryComparison | _: And | _: Or | _: Not | _: IsNull | _: IsNotNull => true
       case _: In | _: InSet | _: Greatest | _: Least => true
       case _: If | _: CaseWhen | _: Coalesce | _: IsNaN | _: NaNvl => true
@@ -1429,11 +1432,11 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       case r: BRound => roundCannotFail(r.dataType, r.ansiEnabled)
       case c: Cast => CometCast.cannotFail(c)
       case _: UnaryPositive => true
-      case u: UnaryMinus => !u.failOnError && numeric
-      case a: Abs => !a.failOnError && numeric
-      case a: Add => nonAnsiNumeric(a.evalMode)
-      case s: Subtract => nonAnsiNumeric(s.evalMode)
-      case m: Multiply => nonAnsiNumeric(m.evalMode)
+      case u: UnaryMinus => floating || (!u.failOnError && numeric)
+      case a: Abs => floating || (!a.failOnError && numeric)
+      case a: Add => floating || nonAnsiNumeric(a.evalMode)
+      case s: Subtract => floating || nonAnsiNumeric(s.evalMode)
+      case m: Multiply => floating || nonAnsiNumeric(m.evalMode)
       case d: Divide => nonAnsiNumeric(d.evalMode)
       case d: IntegralDivide => nonAnsiNumeric(d.evalMode)
       case r: Remainder => nonAnsiNumeric(r.evalMode)
diff --git a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
index 468d16856d0..8ffcc12a4d7 100644
--- a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
+++ b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
@@ -110,7 +110,27 @@ SELECT a, ts FROM topk_types ORDER BY a, try_cast(ts AS DATE) LIMIT 3
 query expect_fallback(can fail on rows that Spark's ordering skips)
 SELECT a, dt FROM topk_types ORDER BY a, try_cast(dt AS TIMESTAMP) LIMIT 3
 
--- The first key is evaluated for every row in Spark too
+-- Native date formatting rejects years that Spark accepts, in every mode
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a, dt FROM topk_types ORDER BY a, CAST(dt AS STRING) LIMIT 3
+
+-- Floating-point arithmetic cannot fail even under ANSI mode
+query
+SELECT a, d FROM topk_types ORDER BY a, d * 1.1D, d + 1, d - 1, -d, abs(d) LIMIT 3
+
+statement
+CREATE TABLE topk_struct(a int, s struct<x: int, y: string>) USING parquet
+
+statement
+INSERT INTO topk_struct SELECT /*+ REPARTITION(1) */ a, IF(x IS NULL, NULL, named_struct('x', x,
+  'y', y)) FROM VALUES (1, 2147483647, 'p'), (1, 5, 'q'), (2, NULL, NULL), (3, -1, '') AS t(a, x, y)
+
+-- Reading a struct field cannot fail
+query
+SELECT a, s FROM topk_struct ORDER BY a, s.x, s.y LIMIT 3
+
+-- The first key is evaluated for every row in Spark too when the input has at least two rows
+-- (https://github.com/apache/datafusion-comet/issues/7527)
 query
 SELECT a, b FROM topk ORDER BY a + 1, b LIMIT 2
 
@@ -143,3 +163,6 @@ SELECT a, b FROM topk ORDER BY a, round(b, -1), CAST(CAST(b AS STRING) AS INT) L
 
 query
 SELECT a, b FROM topk_types ORDER BY a, b / 0, b % 0, b DIV 0, l + 1, l * 2 LIMIT 3
+
+query
+SELECT a, s FROM topk_struct ORDER BY a, s.x, s.y LIMIT 3

From c6f1dcece1d4a7f223d5b5f8c6478674457c209b Mon Sep 17 00:00:00 2001
From: Tien-Lung Fu <0050211@gmail.com>
Date: Thu, 5 Nov 2026 21:09:45 -0800
Subject: [PATCH 6/7] Treat nested timestamp and date-to-string casts as able
 to fail in TopK

A cast over an array, struct or map applies the same native kernels to each
element, field, key and value, so a nested timestamp, or a nested date cast to
a string, raises for years Spark accepts, even under TRY. Check them at any
depth and add TopK regressions for each nested shape.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 3ec1c07f-32b3-4cb1-a6ad-09293da13ab9
---
 .../apache/comet/expressions/CometCast.scala  | 19 +++++--
 .../operators/topk_later_key_ansi.sql         | 51 +++++++++++++++++++
 2 files changed, 67 insertions(+), 3 deletions(-)

diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index 7cfdda3e9ec..dfe3ce2adbe 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -578,14 +578,18 @@ object CometCast
    * cannot fail only where the conversion is always exact, such as a widening one or a string to
    * its bytes. A cast to or from a timestamp is excluded in every mode: the native kernels go
    * through chrono, which rejects years Spark accepts, and raise even for TRY. So is a cast from
-   * a date to a string, whose native formatting raises in every mode for such a year.
+   * a date to a string, whose native formatting raises in every mode for such a year. Both are
+   * checked at any depth, since a nested cast applies the same kernels to each element, field,
+   * key and value (https://github.com/apache/datafusion-comet/issues/7471).
    */
   private[comet] def cannotFail(cast: Cast): Boolean = {
     val from = cast.child.dataType
     val to = cast.dataType
     def isTimestamp(t: DataType) = t == TimestampType || t == TimestampNTZType
-    val dateToString = from == DataTypes.DateType && to.isInstanceOf[StringType]
-    !isTimestamp(from) && !isTimestamp(to) && !dateToString && (evalMode(cast) match {
+    val hasTimestamp = existsNested(from)(isTimestamp) || existsNested(to)(isTimestamp)
+    val dateToString = existsNested(from)(_ == DataTypes.DateType) &&
+      existsNested(to)(_.isInstanceOf[StringType])
+    !hasTimestamp && !dateToString && (evalMode(cast) match {
       case CometEvalMode.TRY => true
       case CometEvalMode.LEGACY if isAtomicCastType(from) && isAtomicCastType(to) => true
       case _ if from.isInstanceOf[StringType] && to == DataTypes.BinaryType => true
@@ -593,6 +597,15 @@ object CometCast
     })
   }
 
+  private def existsNested(dataType: DataType)(p: DataType => Boolean): Boolean =
+    p(dataType) || (dataType match {
+      case ArrayType(elementType, _) => existsNested(elementType)(p)
+      case MapType(keyType, valueType, _) =>
+        existsNested(keyType)(p) || existsNested(valueType)(p)
+      case StructType(fields) => fields.exists(f => existsNested(f.dataType)(p))
+      case _ => false
+    })
+
   private def isAtomicCastType(dataType: DataType): Boolean = dataType match {
     case _: NumericType | _: StringType | DataTypes.BooleanType | DataTypes.BinaryType |
         DataTypes.DateType | DataTypes.TimestampType | _: TimestampNTZType =>
diff --git a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
index 8ffcc12a4d7..33e34807e52 100644
--- a/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
+++ b/spark/src/test/resources/sql-tests/operators/topk_later_key_ansi.sql
@@ -22,6 +22,7 @@
 
 -- Config: spark.sql.ansi.enabled=true
 -- Config: spark.sql.adaptive.enabled=false
+-- Config: spark.sql.parquet.outputTimestampType=TIMESTAMP_MICROS
 
 statement
 CREATE TABLE topk(a int, b int) USING parquet
@@ -114,6 +115,56 @@ SELECT a, dt FROM topk_types ORDER BY a, try_cast(dt AS TIMESTAMP) LIMIT 3
 query expect_fallback(can fail on rows that Spark's ordering skips)
 SELECT a, dt FROM topk_types ORDER BY a, CAST(dt AS STRING) LIMIT 3
 
+statement
+CREATE TABLE topk_nested(a int, tsa array<timestamp>, dta array<date>, sts struct<f: timestamp>,
+  std struct<f: date>, md map<int, date>, mt map<timestamp, int>, ast array<struct<f: timestamp>>)
+USING parquet
+
+statement
+INSERT INTO topk_nested SELECT /*+ REPARTITION(1) */ a, array(ts), array(dt), named_struct('f', ts),
+  named_struct('f', dt), map(1, dt), map(ts, 1), array(named_struct('f', ts)) FROM VALUES
+  (1, timestamp_micros(9000000000000000000), date_from_unix_date(100000000)),
+  (1, timestamp'2024-01-31 23:59:59', date'2024-01-31'),
+  (2, timestamp'1970-01-01 00:00:00', date'1970-01-01') AS t(a, ts, dt)
+
+-- The same native kernels cast each element, field, key and value of a nested type
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(tsa AS STRING) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(dta AS STRING) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(tsa AS array<string>) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(dta AS array<string>) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(sts AS STRING) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(std AS STRING) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(std AS struct<f: string>) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(sts AS struct<f: date>) LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(std AS struct<f: timestamp>) LIMIT 3
+
+-- A map cannot be a sort key, so test it under IS NULL, which cannot fail by itself
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(md AS map<int, string>) IS NULL LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(mt AS map<string, int>) IS NULL LIMIT 3
+
+query expect_fallback(can fail on rows that Spark's ordering skips)
+SELECT a FROM topk_nested ORDER BY a, try_cast(ast AS STRING) LIMIT 3
+
 -- Floating-point arithmetic cannot fail even under ANSI mode
 query
 SELECT a, d FROM topk_types ORDER BY a, d * 1.1D, d + 1, d - 1, -d, abs(d) LIMIT 3

From 37be546ae3b04a281169322e4d7045171dfda9cc Mon Sep 17 00:00:00 2001
From: tlfua <0050211@gmail.com>
Date: Mon, 9 Nov 2026 16:40:35 -0800
Subject: [PATCH 7/7] Reuse SupportLevel.containsType in CometCast.cannotFail

Replace the private existsNested walker with SupportLevel.containsType, and
drop the timestamp arms of isAtomicCastType, which cannotFail never reaches
because it rejects any cast with a timestamp at any depth first. No change
in behavior.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 3ec1c07f-32b3-4cb1-a6ad-09293da13ab9
---
 .../apache/comet/expressions/CometCast.scala  | 22 ++++++-------------
 1 file changed, 7 insertions(+), 15 deletions(-)

diff --git a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
index dfe3ce2adbe..47350bcb348 100644
--- a/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
+++ b/spark/src/main/scala/org/apache/comet/expressions/CometCast.scala
@@ -24,7 +24,7 @@ import scala.util.Try
 import org.apache.spark.SPARK_VERSION
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression, Literal}
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DecimalType, MapType, NullType, NumericType, StringType, StructType, TimestampNTZType, TimestampType}
+import org.apache.spark.sql.types.{ArrayType, DataType, DataTypes, DateType, DecimalType, MapType, NullType, NumericType, StringType, StructType, TimestampNTZType, TimestampType}
 
 import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, withFallbackReason}
 import org.apache.comet.DataTypeSupport.{containsDuplicateFieldNames, isComplexType}
@@ -585,10 +585,10 @@ object CometCast
   private[comet] def cannotFail(cast: Cast): Boolean = {
     val from = cast.child.dataType
     val to = cast.dataType
-    def isTimestamp(t: DataType) = t == TimestampType || t == TimestampNTZType
-    val hasTimestamp = existsNested(from)(isTimestamp) || existsNested(to)(isTimestamp)
-    val dateToString = existsNested(from)(_ == DataTypes.DateType) &&
-      existsNested(to)(_.isInstanceOf[StringType])
+    val hasTimestamp = Seq(from, to).exists(
+      SupportLevel.containsType(_, classOf[TimestampType], classOf[TimestampNTZType]))
+    val dateToString = SupportLevel.containsType(from, classOf[DateType]) &&
+      SupportLevel.containsType(to, classOf[StringType])
     !hasTimestamp && !dateToString && (evalMode(cast) match {
       case CometEvalMode.TRY => true
       case CometEvalMode.LEGACY if isAtomicCastType(from) && isAtomicCastType(to) => true
@@ -597,18 +597,10 @@ object CometCast
     })
   }
 
-  private def existsNested(dataType: DataType)(p: DataType => Boolean): Boolean =
-    p(dataType) || (dataType match {
-      case ArrayType(elementType, _) => existsNested(elementType)(p)
-      case MapType(keyType, valueType, _) =>
-        existsNested(keyType)(p) || existsNested(valueType)(p)
-      case StructType(fields) => fields.exists(f => existsNested(f.dataType)(p))
-      case _ => false
-    })
-
+  // Timestamps are left out because `cannotFail` rejects them before reaching this check.
   private def isAtomicCastType(dataType: DataType): Boolean = dataType match {
     case _: NumericType | _: StringType | DataTypes.BooleanType | DataTypes.BinaryType |
-        DataTypes.DateType | DataTypes.TimestampType | _: TimestampNTZType =>
+        DataTypes.DateType =>
       true
     case _ => false
   }