Re: [PR] [SPARK-34586][SQL] Support declaring a write distribution and ordering in CREATE/REPLACE TABLE [spark]

[via GitHub]

anuragmantri commented on code in PR #58153:
URL: https://github.com/apache/spark/pull/58153#discussion_r4391146600


##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalogCapability.java:
##########
@@ -92,5 +92,65 @@ public enum TableCatalogCapability {
    * {@link TableCatalog#createTable}.
    * See {@link Column#identityColumnSpec()}.
    */
-  SUPPORTS_CREATE_TABLE_WITH_IDENTITY_COLUMNS
+  SUPPORTS_CREATE_TABLE_WITH_IDENTITY_COLUMNS,
+
+  /**
+   * Signals that the TableCatalog supports a write distribution and ordering 
requested upon table
+   * creation in SQL, with {@code DISTRIBUTED BY PARTITION} and {@code 
[LOCALLY] ORDERED BY ...} /
+   * {@code UNORDERED}. The request reaches the catalog only through the 
{@link TableInfo}
+   * overloads:
+   * <ul>
+   *   <li>{@link TableCatalog#createTable(Identifier, TableInfo)}</li>
+   *   <li>{@link StagingTableCatalog#stageCreate(Identifier, TableInfo)}</li>
+   *   <li>{@link StagingTableCatalog#stageReplace(Identifier, TableInfo)}</li>
+   *   <li>{@link StagingTableCatalog#stageCreateOrReplace(Identifier, 
TableInfo)}</li>
+   * </ul>
+   * Their default implementations drop the request, so a catalog that reports 
this capability must

Review Comment:
   Done in c8f45f6. The four defaults call `rejectRequested`. I removed the 
guards in `V2SessionCatalog` (the file matches master again) and in 
`DelegatingCatalogExtension`. `capabilities()` still strips the capability. 
Please look at the `@throws` text and the capability paragraph in 
`TableCatalogCapability`, since the Javadoc now says the defaults reject.



##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/WriteDistributionAndOrderingUtilsSuite.scala:
##########
@@ -0,0 +1,456 @@
+/*
+ * 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.catalyst.util
+
+import java.math.BigDecimal
+import java.nio.ByteBuffer
+import java.time.{Duration, Instant, LocalDate, LocalDateTime, LocalTime, 
Period}
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Literal => CatalystLiteral}
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
+import org.apache.spark.sql.catalyst.plans.SQLHelper
+import org.apache.spark.sql.connector.catalog.WriteDistributionMode
+import org.apache.spark.sql.connector.expressions.{Cast, Expression, 
Expressions, Extract, FieldReference, GeneralScalarExpression, GetArrayItem, 
IdentityTransform, NamedReference, NullOrdering, SortDirection, SortOrder, 
Transform, UserDefinedScalarFunc, VariantGet}
+import org.apache.spark.sql.connector.expressions.LogicalExpressions.{apply => 
transform, literal, sort}
+import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+class WriteDistributionAndOrderingUtilsSuite extends SparkFunSuite with 
SQLHelper {
+  import WriteDistributionAndOrdering._
+
+  private class ConnectorReference(names: String*) extends NamedReference {
+    override def fieldNames(): Array[String] = names.toArray
+  }
+
+  // A connector's own expression, which Spark renders with its `describe`.
+  private class ConnectorFunction(rendered: String, args: Expression*) extends 
Expression {
+    override def children(): Array[Expression] = args.toArray
+    override def describe(): String = rendered
+    override def toString: String = rendered
+  }
+
+  // A key whose arguments and `describe` both throw `error`.
+  private class UnrenderableTransform(
+      error: () => Throwable,
+      calls: AtomicInteger = new AtomicInteger(0)) extends Transform {
+    override def name(): String = "unrenderable"
+    override def references(): Array[NamedReference] = Array.empty
+    override def arguments(): Array[Expression] = {
+      calls.incrementAndGet()
+      throw error()
+    }
+    override def toString: String = s"unrenderable(${arguments().mkString(", 
")})"
+  }
+
+  private val id = FieldReference("id")
+
+  private def f(arg: Expression): Transform = transform("f", id, arg)
+
+  private def key(e: Expression): SortOrder =
+    sort(e, SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)
+
+  private val schema = StructType.fromDDL(
+    "id INT, `order` INT, `order-id` INT, s STRUCT<x: INT>, a ARRAY<STRUCT<x: 
INT>>")
+
+  private val replay = replayWith(CatalystSqlParser) _
+
+  private def emitted(e: Expression): Option[String] = {
+    writeClausesSQL(WriteDistributionMode.RANGE, Seq(key(e)), Seq.empty, 
schema, replay)
+  }
+
+  private val spellable: Seq[(String, Expression)] = Seq(
+    "reference" -> id,
+    "nested field" -> FieldReference(Seq("s", "x")),
+    "connector reference" -> transform("f", new 
ConnectorReference("order-id")),
+    "identity" -> IdentityTransform(id),
+    "identity by name" -> transform("identity", id),
+    "byte" -> f(literal(1.toByte)),
+    "short" -> f(literal(1.toShort)),
+    "int" -> f(literal(1)),
+    "long" -> f(literal(1L)),
+    "float" -> f(literal(1.5F)),
+    "double" -> f(literal(1.5D)),
+    "decimal" -> f(literal(new BigDecimal("1.5"))),
+    "string" -> f(literal("x")),
+    "string with escapes" -> f(literal("it's a \\")),
+    "binary" -> f(literal(Array[Byte](1, 2))),
+    "date" -> f(literal(LocalDate.of(1970, 1, 1))),
+    "timestamp" -> f(literal(Instant.parse("2020-11-01T09:30:00Z"))),
+    "timestamp_ntz" -> f(literal(LocalDateTime.of(2020, 1, 1, 10, 0))),
+    "time" -> f(literal(LocalTime.of(12, 0))),
+    "time(7) with trailing zeros" -> f(literal(LocalTime.of(12, 0, 0, 
100000000), TimeType(7))),
+    "time(9) with trailing zeros" -> f(literal(LocalTime.of(12, 0, 0, 
100000000), TimeType(9))),
+    "time(9) without trailing zeros" ->
+      f(literal(LocalTime.of(12, 0, 0, 123456789), TimeType(9))),
+    "time(7) at a whole second" -> f(literal(LocalTime.of(12, 0), 
TimeType(7))),
+    "day-time interval" -> f(literal(Duration.ofHours(1))),
+    "year-month interval" -> f(literal(Period.ofMonths(14))),
+    "bucket" -> transform("bucket", literal(16), id),
+    "bucket without columns" -> transform("bucket", literal(16)),
+    "Spark's bucket" -> Expressions.bucket(16, "id"),
+    "days" -> transform("days", id),
+    "unparsed name" -> transform("Days", id, literal("UTC")))
+
+  private val unspellable: Seq[(String, Expression)] = Seq(
+    "nested transform" -> transform("f", transform("g", id)),
+    "nested identity" -> transform("f", IdentityTransform(id)),
+    "true" -> f(literal(true)),
+    "false" -> f(literal(false)),
+    "null" -> f(literal(null)),
+    "typed null" -> f(literal(null, IntegerType)),
+    "decimal with trailing zeros" ->
+      f(literal(new BigDecimal("1.50"), DecimalType(10, 2))),
+    "decimal in exponent form" ->
+      f(literal(new BigDecimal("0.0000001"), DecimalType(10, 7))),
+    "char" -> f(literal("x", CharType(3))),
+    "varchar" -> f(literal("x", VarcharType(3))),
+    "collated string" -> f(literal("x", StringType("UTF8_LCASE"))),
+    "NaN" -> f(literal(Float.NaN)),
+    "infinity" -> f(literal(Double.PositiveInfinity)),
+    "maximal float" -> f(literal(Float.MaxValue)),
+    "time(6) holding a nanosecond" -> f(literal(LocalTime.ofNanoOfDay(1))),
+    "year interval not in whole years" ->
+      f(literal(13, YearMonthIntervalType(YearMonthIntervalType.YEAR))),
+    "hour interval not in whole hours" ->
+      f(literal(Duration.ofMinutes(90).toNanos / 1000, DayTimeIntervalType(
+        DayTimeIntervalType.HOUR))),
+    "negative-scale decimal" -> f(Expressions.literal(new BigDecimal("1E+3"))),
+    "int value with long type" -> f(literal(5, LongType)),
+    "byte buffer value" -> f(literal(ByteBuffer.wrap(Array[Byte](1, 2)), 
BinaryType)),
+    "bucket with columns first" -> transform("bucket", id, literal(16)),
+    "bucket with long count" -> transform("bucket", literal(16L), id),
+    "bucket with literal column" -> transform("bucket", literal(16), 
literal(1)),
+    "days with two arguments" -> transform("days", id, literal("UTC")),
+    "hours of a literal" -> transform("hours", literal(1)),
+    "missing column" -> FieldReference("missing"),
+    "missing column inside a nested identity" ->
+      transform("f", Expressions.identity("missing")),
+    "column in a different case" -> FieldReference("ID"),
+    "field of an array element" -> FieldReference(Seq("a", "x")),
+    "reference with no field names" -> new ConnectorReference())
+
+  test("a sort key is emitted iff its rendering replays as the same key") {
+    assert(spellable.filter { case (_, e) =>
+      !emitted(e).contains(s"ORDERED BY (${describeSortOrder(key(e))})")
+    }.map(_._1) === Seq.empty)
+    assert(unspellable.filter { case (_, e) => emitted(e).isDefined 
}.map(_._1) === Seq.empty)
+  }
+
+  test("a sort key is emitted iff it replays as the same key under parser 
confs") {
+    val timestamp = f(literal(Instant.parse("2020-11-01T09:30:00Z")))
+    val wrong = Seq(
+      (SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles", timestamp, 
true),
+      (SQLConf.TIMESTAMP_TYPE.key -> "TIMESTAMP_NTZ", timestamp, true),
+      (SQLConf.LEGACY_INTERVAL_ENABLED.key -> "true", 
f(literal(Duration.ofHours(1))), false),
+      (SQLConf.LEGACY_INTERVAL_ENABLED.key -> "true", 
f(literal(Period.ofMonths(14))), false),
+      (SQLConf.ESCAPED_STRING_LITERALS.key -> "true", f(literal("it's")), 
false),
+      (SQLConf.ESCAPED_STRING_LITERALS.key -> "true", f(literal("a\\b")), 
false),
+      (SQLConf.ESCAPED_STRING_LITERALS.key -> "true", f(literal("x")), true),
+      (SQLConf.LEGACY_TIME_PARSER_POLICY.key -> "LEGACY",
+        f(literal(LocalDate.of(-1, 1, 1))), false),
+      (SQLConf.LEGACY_FROM_DAYTIME_STRING.key -> "true",
+        f(literal(Duration.ofHours(26), DayTimeIntervalType(
+          DayTimeIntervalType.DAY, DayTimeIntervalType.HOUR))), false)
+    ).filter { case (conf, e, spelled) => 
withSQLConf(conf)(emitted(e).isDefined != spelled) }
+    assert(wrong.map(_._2.describe) === Seq.empty)
+  }
+
+  test("a name that is a reserved keyword is emitted quoted only where the 
parser needs it") {
+    val order = FieldReference("order")
+    assert(emitted(order) === Some("ORDERED BY (order ASC NULLS FIRST)"))
+    withSQLConf(
+        SQLConf.ANSI_ENABLED.key -> "true",
+        SQLConf.ENFORCE_RESERVED_KEYWORDS.key -> "true") {
+      assert(emitted(order) === Some("ORDERED BY (`order` ASC NULLS FIRST)"))
+      assert(emitted(transform("bucket", literal(4), order)) ===
+        Some("ORDERED BY (`bucket`(4, `order`) ASC NULLS FIRST)"))
+    }
+  }
+
+  test("each mode with a clause form is emitted as that clause") {
+    val keys = Seq(key(id))
+    val partitioning = Seq(Expressions.identity("p"))
+    Seq(
+      (WriteDistributionMode.HASH, keys,
+        "DISTRIBUTED BY PARTITION ORDERED BY (id ASC NULLS FIRST)"),
+      (WriteDistributionMode.HASH, Seq.empty, "DISTRIBUTED BY PARTITION"),
+      (WriteDistributionMode.RANGE, keys, "ORDERED BY (id ASC NULLS FIRST)"),
+      (WriteDistributionMode.NONE, keys, "LOCALLY ORDERED BY (id ASC NULLS 
FIRST)"),
+      (WriteDistributionMode.NONE, Seq.empty, "UNORDERED")
+    ).foreach { case (mode, ordering, clauses) =>
+      assert(writeClausesSQL(mode, ordering, partitioning, schema, replay) === 
Some(clauses),
+        s"for $mode $ordering")
+    }
+  }
+
+  test("a key that does not parse back omits the pair in every mode") {
+    val keys = Seq(key(transform("f", transform("g", id))))
+    Seq(WriteDistributionMode.HASH, WriteDistributionMode.RANGE, 
WriteDistributionMode.NONE)
+      .foreach { mode =>
+        assert(writeClausesSQL(mode, keys, Seq(Expressions.identity("p")), 
schema, replay).isEmpty,
+          s"for $mode")
+      }
+  }
+
+  test("the pairs with no clause form are not emitted") {
+    val keys = Seq(key(id))
+    Seq(
+      (WriteDistributionMode.HASH, keys, Seq.empty[Transform]),
+      (WriteDistributionMode.HASH, Seq.empty, 
Seq(Expressions.apply("cluster_by", id))),
+      (WriteDistributionMode.RANGE, Seq.empty, Seq.empty[Transform]),
+      (null, keys, Seq.empty[Transform])
+    ).foreach { case (mode, ordering, partitioning) =>
+      val anyClauses = (_: String) => Some((mode, ordering))
+      assert(writeClausesSQL(mode, ordering, partitioning, schema, 
anyClauses).isEmpty,
+        s"for $mode")
+    }
+  }
+
+  test("a nanosecond TIMESTAMP_LTZ renders in UTC with an explicit offset") {
+    withSQLConf(
+        SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+        SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+      val parsed = CatalystSqlParser.parseExpression(
+        "TIMESTAMP_LTZ '2020-11-01 
01:30:00.123456789-08:00'").asInstanceOf[CatalystLiteral]
+      val e = f(literal(parsed.value, parsed.dataType))
+      assert(emitted(e) ===
+        Some("ORDERED BY (f(id, TIMESTAMP_LTZ '2020-11-01 
09:30:00.123456789Z') ASC NULLS FIRST)"))
+
+      val wholeSecond = CatalystSqlParser.parseExpression(
+        "TIMESTAMP_LTZ '2020-11-01 
01:30:00.000000000-08:00'").asInstanceOf[CatalystLiteral]
+      assert(emitted(f(literal(wholeSecond.value, wholeSecond.dataType))) ===
+        Some("ORDERED BY (f(id, TIMESTAMP_LTZ '2020-11-01 
09:30:00.000000000Z') ASC NULLS FIRST)"))
+
+      val ntz = CatalystSqlParser.parseExpression("TIMESTAMP_NTZ '2020-01-01 
10:00:00.100000000'")
+        .asInstanceOf[CatalystLiteral]
+      assert(emitted(f(literal(ntz.value, ntz.dataType))).isDefined)
+    }
+  }
+
+  test("a timestamp renders in UTC with an explicit offset") {
+    val e = f(literal(Instant.parse("2020-11-01T09:30:00Z")))
+    Seq("UTC", "America/Los_Angeles", "Asia/Tokyo").foreach { tz =>
+      withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
+        assert(describeSortOrder(key(e)) ===
+          "f(id, TIMESTAMP_LTZ '2020-11-01 09:30:00Z') ASC NULLS FIRST", s"in 
$tz")
+      }
+    }
+  }
+
+  test("DESCRIBE renders a literal Catalyst cannot represent with describe") {
+    Seq(
+      Expressions.literal(new BigDecimal("1E+3")) -> "1E+3",
+      literal(5, LongType) -> "5"
+    ).foreach { case (lit, rendered) =>
+      assert(describeSortOrder(key(f(lit))) === s"f(id, $rendered) ASC NULLS 
FIRST")
+    }
+    val buffer = ByteBuffer.wrap(Array[Byte](1, 2))
+    assert(describeSortOrder(key(f(literal(buffer, BinaryType)))) ===
+      s"f(id, $buffer) ASC NULLS FIRST")
+  }
+
+  test("DESCRIBE renders a connector expression in its SQL form") {
+    Seq(
+      new Extract("YEAR", id) -> "EXTRACT(YEAR FROM id)",
+      new UserDefinedScalarFunc("my_udf", "my_udf", Array[Expression](id)) -> 
"my_udf(id)",
+      new GeneralScalarExpression("+", Array[Expression](id, literal(1))) -> 
"id + 1",
+      new GeneralScalarExpression("+", 
Array[Expression](IdentityTransform(id), literal(1))) ->
+        "identity(id) + 1",
+      new Cast(Expressions.bucket(16, "id"), StringType) -> "CAST(bucket(16, 
id) AS STRING)",
+      new Cast(id, ArrayType(IntegerType)) -> "CAST(id AS ARRAY<INT>)",
+      new UserDefinedScalarFunc("zorder_key", "zorder_key", Array[Expression](
+        transform("zorder", id, FieldReference("b")))) -> 
"zorder_key(zorder(id, b))",
+      new VariantGet(id, null, IntegerType, true, null) -> "variant_get(id, 
'null', int)"
+    ).foreach { case (e, rendered) =>
+      assert(describeSortOrder(key(transform("f", e))) === s"f($rendered) ASC 
NULLS FIRST")
+    }
+  }
+
+  test("DESCRIBE renders a connector's own expression under a Spark expression 
with its describe") {
+    val truncate = new ConnectorFunction("truncate(4, id)", id, literal(4))
+    Seq(
+      new GeneralScalarExpression("ABS", Array[Expression](truncate)) -> 
"ABS(truncate(4, id))",
+      new GeneralScalarExpression("+", Array[Expression](truncate, 
literal(1))) ->
+        "(truncate(4, id)) + 1",
+      new Cast(truncate, IntegerType) -> "CAST(truncate(4, id) AS INT)",
+      new Extract("YEAR", truncate) -> "EXTRACT(YEAR FROM truncate(4, id))",
+      new GetArrayItem(truncate, literal(0), true) -> "truncate(4, id)[0]",
+      truncate -> "truncate(4, id)"
+    ).foreach { case (e, rendered) =>
+      assert(describeSortOrder(key(transform("f", e))) === s"f($rendered) ASC 
NULLS FIRST")
+      assert(describeSortOrder(key(e)) === s"$rendered ASC NULLS FIRST")
+    }
+  }
+
+  test("DESCRIBE renders a sort order that cannot be rendered with the class 
of its key") {
+    Seq[() => Throwable](
+      () => new IllegalStateException("broken"),
+      () => new NoClassDefFoundError("org/example/MissingClass"),
+      () => new StackOverflowError()
+    ).foreach { error =>
+      val unrenderable = new UnrenderableTransform(error)
+      assert(describeSortOrder(key(unrenderable)) ===
+        s"${classOf[UnrenderableTransform].getName} ASC NULLS FIRST", 
error().getClass.getName)
+      assert(emitted(unrenderable).isEmpty, error().getClass.getName)
+    }
+  }
+
+  test("a key that fails with a StackOverflowError costs the same at any 
depth") {
+    def callsFor(depth: Int): Int = {
+      val calls = new AtomicInteger(0)
+      val leaf = new UnrenderableTransform(() => new StackOverflowError(), 
calls)
+      val nested = (1 to depth).foldLeft[Expression](leaf) { (child, _) =>
+        new GeneralScalarExpression("ABS", Array[Expression](child))
+      }
+      assert(describeSortOrder(key(nested)) ===
+        s"${classOf[GeneralScalarExpression].getName} ASC NULLS FIRST")
+      assert(emitted(nested).isEmpty)
+      calls.get()
+    }
+
+    assert(callsFor(10) === callsFor(1))
+  }
+
+  // Runs `f` on a thread with a small (256 KiB) stack so that a recursive 
rendering overflows
+  // deterministically, and returns whatever it threw (or null if nothing 
did). The test thread
+  // waits in `join`, which SparkFunSuite's per-test timeout can interrupt.
+  private def runOnSmallStack(f: => Unit): Throwable = {
+    @volatile var thrown: Throwable = null
+    val runnable = new Runnable {
+      override def run(): Unit = {
+        try f catch { case t: Throwable => thrown = t }
+      }
+    }
+    val t = new Thread(null, runnable, "write-ordering-deep-key", 256 * 1024)
+    t.setDaemon(true)
+    t.start()
+    t.join()

Review Comment:
   Done. The wait is one minute, followed by an `isAlive` assert. I also fixed 
the comment. A scratch suite with a one-minute timeout was still running after 
five, so the per-test timeout does not interrupt the `join`.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/WriteDistributionAndOrdering.scala:
##########
@@ -0,0 +1,355 @@
+/*
+ * 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.catalyst.util
+
+import java.time.ZoneOffset
+
+import scala.jdk.CollectionConverters._
+import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.{Literal => CatalystLiteral}
+import org.apache.spark.sql.catalyst.parser.ParserInterface
+import org.apache.spark.sql.catalyst.plans.logical.CreateTable
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, 
Table, TableCatalog, TableCatalogCapability, WriteDistributionMode}
+import org.apache.spark.sql.connector.expressions.{Expression, 
GeneralScalarExpression, GetArrayItem, IdentityTransform, Literal, 
NamedReference, SortOrder, Transform, VariantGet}
+import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.connector.{ExpressionWithToString, 
ToStringSQLBuilder}
+import org.apache.spark.sql.types.{DataType, FloatType, StructField, 
StructType, TimestampLTZNanosType, TimestampType}
+import org.apache.spark.unsafe.types.TimestampNanosVal
+import org.apache.spark.util.ArrayImplicits._
+
+/**
+ * Utility methods for the create-time write distribution and ordering 
requested by
+ * `CREATE`/`REPLACE TABLE ... DISTRIBUTED BY PARTITION / [LOCALLY] ORDERED 
BY`.
+ */
+object WriteDistributionAndOrdering {
+
+  // Runs a call into a connector's expression or literal at one node of a 
key. `NonFatal` does not
+  // cover a `LinkageError`, from a connector class whose dependency is 
missing, and it should not
+  // fail DESCRIBE or SHOW CREATE TABLE. A `StackOverflowError` is left to 
`attempt` at the entry
+  // points: caught at every node of a key deeper than the stack, the fallback 
would render each
+  // subtree again at every level and never finish.
+  private def attemptNode[T](body: => T): Try[T] = {
+    try Success(body) catch {
+      case NonFatal(e) => Failure(e)
+      case e: LinkageError => Failure(e)
+    }
+  }
+
+  // Runs a whole rendering at an entry point, where a `StackOverflowError` 
from a key nested
+  // deeper than the stack is caught once, after the stack has unwound.
+  private def attempt[T](body: => T): Try[T] = {
+    try Success(body) catch {
+      case NonFatal(e) => Failure(e)
+      case e @ (_: LinkageError | _: StackOverflowError) => Failure(e)
+    }
+  }
+
+  /** Names the write clauses in an error's operation, as in `CREATE TABLE ... 
<CLAUSES>`. */
+  val CLAUSES = "DISTRIBUTED BY/ORDERED BY/UNORDERED"
+
+  /**
+   * True when a CREATE/REPLACE TABLE statement asked for a write distribution 
or ordering.
+   * `UNORDERED` counts as a request.
+   */
+  def isRequested(
+      writeDistributionMode: WriteDistributionMode,
+      writeOrdering: Seq[SortOrder]): Boolean = {
+    writeDistributionMode != null || writeOrdering.nonEmpty
+  }
+
+  /**
+   * True when the transforms define a partitioning that `DISTRIBUTED BY 
PARTITION` can distribute
+   * by. Bucketing counts; a `cluster_by` transform, from CLUSTER BY or 
reported by a connector,
+   * does not. Matches by name only, since a connector's `cluster_by` may take 
any arguments.
+   */
+  def hasPartitioning(partitioning: Seq[Transform]): Boolean = {
+    partitioning.exists(_.name != "cluster_by")
+  }
+
+  /** The write ordering a table declares, which is empty also for a connector 
that returns null. */
+  def writeOrderingOf(table: Table): Seq[SortOrder] = {
+    
Option(table.writeOrdering()).map(_.toImmutableArraySeq).getOrElse(Seq.empty)
+  }
+
+  /**
+   * The `UNSUPPORTED_FEATURE.TABLE_OPERATION` error for a requested write 
distribution and
+   * ordering that `catalog` cannot carry, as an unchecked exception that a 
Java override of
+   * `createTable` can throw.
+   */
+  def unsupportedClausesException(
+      catalog: CatalogPlugin,
+      ident: Identifier): SparkUnsupportedOperationException = {
+    val error = QueryCompilationErrors.unsupportedTableOperationError(catalog, 
ident, CLAUSES)
+      .asInstanceOf[SparkThrowable]
+    new SparkUnsupportedOperationException(
+      error.getCondition, error.getMessageParameters.asScala.toMap)
+  }
+
+  /** True when `catalog` accepts a create-time write distribution and 
ordering. */
+  def supportsWriteDistributionAndOrdering(catalog: TableCatalog): Boolean = {
+    catalog.capabilities().contains(
+      
TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING)
+  }
+
+  /**
+   * Rejects a create-time write distribution or ordering that the catalog has 
not advertised
+   * support for, before anything is created or dropped.
+   */
+  def validateCatalogForWriteDistributionAndOrdering(
+      catalog: TableCatalog,
+      ident: Identifier,
+      operation: String,
+      writeDistributionMode: WriteDistributionMode,
+      writeOrdering: Seq[SortOrder]): Unit = {
+    if (isRequested(writeDistributionMode, writeOrdering) &&
+        !supportsWriteDistributionAndOrdering(catalog)) {
+      throw QueryCompilationErrors.unsupportedTableOperationError(
+        catalog, ident, s"$operation ... $CLAUSES")
+    }
+  }
+
+  /**
+   * Renders a requested sort key as SQL, for DESCRIBE. When a connector 
expression in it fails to
+   * render, the sort order is rendered with its own `describe`, and when that 
fails too, with
+   * whichever of the class name of its key, its direction and its null 
ordering can be read, or
+   * with its own class name when none can.
+   */
+  def describeSortOrder(sortOrder: SortOrder): String = {
+    attempt(sortOrderToSQL(sortOrder, 
quoteIfNeeded)).orElse(attempt(sortOrder.describe))
+      .getOrElse(describeUnrenderable(sortOrder))
+  }
+
+  private def describeUnrenderable(sortOrder: SortOrder): String = {
+    val parts = Seq(
+      attempt(sortOrder.expression().getClass.getName),
+      attempt(sortOrder.direction().toString),
+      attempt(sortOrder.nullOrdering().toString)).flatMap(_.toOption)
+    if (parts.nonEmpty) parts.mkString(" ") else 
Option(sortOrder).fold("null")(_.getClass.getName)
+  }
+
+  /**
+   * Renders the declared write distribution and ordering as `DISTRIBUTED BY 
PARTITION`,
+   * `[LOCALLY] ORDERED BY` or `UNORDERED` clauses for SHOW CREATE TABLE. 
Returns None when the
+   * pair has no clause form, such as `HASH` without partitioning, when a 
connector expression in
+   * the ordering fails to render, or when `replay`, such as [[replayWith]], 
does not give back the
+   * same pair with every referenced column in `schema`. Names are quoted only 
where needed, and on
+   * a second try all of them, which a name that is a reserved keyword under
+   * `spark.sql.ansi.enforceReservedKeywords` needs.
+   */
+  def writeClausesSQL(
+      writeDistributionMode: WriteDistributionMode,
+      writeOrdering: Seq[SortOrder],
+      partitioning: Seq[Transform],
+      schema: StructType,
+      replay: String => Option[(WriteDistributionMode, Seq[SortOrder])]): 
Option[String] = {
+    lazy val partitioned = hasPartitioning(partitioning)
+    Seq[String => String](quoteIfNeeded, quoteIdentifier).iterator.flatMap { 
quote =>

Review Comment:
   Done. The clause form is decided first, and a failed rendering ends the 
search. I also skip the quoted retry when it renders the same text. Please 
check the new `render` and `replaysAsSamePair` shape in `writeClausesSQL`.



##########
sql/core/src/test/scala/org/apache/spark/sql/connector/CreateTableWriteOrderSuite.scala:
##########
@@ -0,0 +1,1512 @@
+/*
+ * 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.connector
+
+import java.util
+import java.util.Locale
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.SparkThrowable
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, 
TableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.catalyst.plans.logical.{CreateTable, 
CreateTableAsSelect, LogicalPlan, ReplaceTable, ReplaceTableAsSelect, 
V2CreateTablePlan}
+import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, 
RangePartitioning}
+import org.apache.spark.sql.catalyst.util.WriteDistributionAndOrdering
+import org.apache.spark.sql.connector.catalog.{Column, 
DelegatingCatalogExtension, DelegatingTable, Identifier, 
InMemoryRowLevelOperationTableCatalog, InMemoryTable, InMemoryTableCatalog, 
StagedTable, StagingInMemoryTableCatalog, Table, TableCatalogCapability, 
TableInfo, Txn, TxnTableCatalog, WriteDistributionMode}
+import 
org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+import org.apache.spark.sql.connector.catalog.WriteDistributionMode.{HASH, 
NONE, RANGE}
+import org.apache.spark.sql.connector.catalog.transactions.{Transaction, 
TransactionInfo}
+import org.apache.spark.sql.connector.distributions.Distributions
+import org.apache.spark.sql.connector.expressions.{ClusterByTransform, 
Expression, FieldReference, LogicalExpressions, NamedReference, NullOrdering, 
SortDirection, SortOrder, Transform}
+import org.apache.spark.sql.connector.expressions.LogicalExpressions.literal
+import org.apache.spark.sql.execution.{SortExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Tests the create-time write distribution and ordering clauses: 
CREATE/REPLACE TABLE ...
+ * DISTRIBUTED BY PARTITION ... [LOCALLY] ORDERED BY ... | UNORDERED. Covers 
what the parser
+ * produces, what reaches the catalog, and how a catalog without the 
capability is rejected.
+ */
+class CreateTableWriteOrderSuite extends QueryTest with SharedSparkSession {
+
+  // The catalog manager caches catalogs for the session; reset it so tests 
don't share tables.
+  override def afterEach(): Unit = {
+    spark.sessionState.catalogManager.reset()
+    super.afterEach()
+  }
+
+  private def parse(sql: String): LogicalPlan = 
spark.sessionState.sqlParser.parsePlan(sql)
+
+  private def analyze(sql: String): LogicalPlan =
+    spark.sessionState.executePlan(parse(sql)).analyzed
+
+  private def orderingOf(plan: LogicalPlan): Seq[String] = plan match {
+    case create: V2CreateTablePlan => 
WriteSpecCall.render(create.writeOrdering)
+    case other => fail(s"unexpected plan: $other")
+  }
+
+  private def calls(catalogName: String): Seq[WriteSpecCall] =
+    spark.sessionState.catalogManager.catalog(catalogName)
+      .asInstanceOf[RecordsWriteSpecs].recordedCalls
+
+  private def writeClauseLines(ddl: String): Seq[String] = 
ddl.split("\n").toSeq.filter { line =>
+    Seq("DISTRIBUTED BY", "ORDERED BY", "LOCALLY ORDERED BY", 
"UNORDERED").exists(line.startsWith)
+  }
+
+  private def checkUnknownColumns(stmt: String, cols: String): Unit = {
+    checkError(
+      exception = intercept[AnalysisException](analyze(stmt)),
+      condition = "WRITE_ORDERING_WITH_UNKNOWN_COLUMN",
+      sqlState = "42703",
+      parameters = Map("cols" -> cols),
+      context = ExpectedContext(stmt, 0, stmt.length - 1))
+  }
+
+  private def checkRejected(stmt: String, tableName: String): Unit = {
+    checkError(
+      exception = intercept[Exception](sql(stmt)).asInstanceOf[SparkThrowable],
+      condition = "UNSUPPORTED_FEATURE.TABLE_OPERATION",
+      sqlState = "0A000",
+      parameters = Map(
+        "tableName" -> tableName,
+        "operation" -> WriteDistributionAndOrdering.CLAUSES))
+  }
+
+  private def checkUnsupported(stmt: String, tableName: String, operation: 
String): Unit = {
+    checkError(
+      exception = intercept[AnalysisException](sql(stmt)),
+      condition = "UNSUPPORTED_FEATURE.TABLE_OPERATION",
+      sqlState = "0A000",
+      parameters = Map(
+        "tableName" -> tableName,
+        "operation" -> s"$operation ... DISTRIBUTED BY/ORDERED BY/UNORDERED"))
+  }
+
+  private def showCreate(table: String): String = {
+    sql(s"SHOW CREATE TABLE $table").head().getString(0)
+  }
+
+  private def described(table: String): Map[String, String] = {
+    sql(s"DESCRIBE TABLE EXTENDED $table").collect()
+      .map(r => r.getString(0) -> r.getString(1)).toMap
+  }
+
+  private def loadTable(catalogName: String, name: String): Table =
+    spark.sessionState.catalogManager.catalog(catalogName).asTableCatalog
+      .loadTable(Identifier.of(Array.empty, name))
+
+  private def clusteringOf(table: Table): Seq[Seq[String]] =
+    table.partitioning().toSeq.collect {
+      case ClusterByTransform(columns) => columns.map(_.describe)
+    }
+
+  test("parse DISTRIBUTED BY PARTITION / ORDERED BY on CREATE TABLE") {
+    parse("CREATE TABLE t (id INT, c STRING) USING foo PARTITIONED BY (c) " +
+      "DISTRIBUTED BY PARTITION ORDERED BY id ASC NULLS LAST") match {
+      case c: CreateTable =>
+        assert(c.writeDistributionMode === HASH)
+        assert(c.writeOrdering.length === 1)
+        val order = c.writeOrdering.head
+        assert(order.expression() === FieldReference(Seq("id")))
+        assert(order.direction() === SortDirection.ASCENDING)
+        assert(order.nullOrdering() === NullOrdering.NULLS_LAST)
+      case other => fail(s"unexpected plan: $other")
+    }
+  }
+
+  test("each combination of the write clauses parses to its distribution and 
ordering") {
+    Seq(
+      ("", null, 0),
+      ("ORDERED BY id", RANGE, 1),
+      ("LOCALLY ORDERED BY id", NONE, 1),
+      ("UNORDERED", NONE, 0),
+      ("PARTITIONED BY (c) DISTRIBUTED BY PARTITION", HASH, 0),
+      ("PARTITIONED BY (c) DISTRIBUTED BY PARTITION ORDERED BY id", HASH, 1),
+      ("PARTITIONED BY (c) DISTRIBUTED BY PARTITION LOCALLY ORDERED BY id", 
HASH, 1),
+      ("PARTITIONED BY (c) DISTRIBUTED BY PARTITION UNORDERED", HASH, 0),
+      ("ORDERED BY id PARTITIONED BY (c) DISTRIBUTED BY PARTITION", HASH, 1),
+      ("DISTRIBUTED BY PARTITION PARTITIONED BY (c) LOCALLY ORDERED BY id", 
HASH, 1),
+      ("PARTITIONED BY (c) ORDERED BY id COMMENT 'a table' DISTRIBUTED BY 
PARTITION", HASH, 1)
+    ).foreach { case (clauses, mode, orderingSize) =>
+      parse(s"CREATE TABLE t (id INT, c STRING) USING foo $clauses") match {
+        case c: CreateTable =>
+          assert(c.writeDistributionMode === mode, s"for [$clauses]")
+          assert(c.writeOrdering.length === orderingSize, s"for [$clauses]")
+        case other => fail(s"unexpected plan for [$clauses]: $other")
+      }
+    }
+  }
+
+  test("a column written as identity(col) reaches the catalog as a plain 
column") {
+    parse("CREATE TABLE t (id INT) USING foo ORDERED BY identity(id)") match {
+      case c: CreateTable =>
+        assert(c.writeOrdering.map(_.expression()) === 
Seq(FieldReference("id")))
+      case other => fail(s"unexpected plan: $other")
+    }
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      assert(orderingOf(analyze(
+        "CREATE TABLE testcat.t (id INT) USING foo ORDERED BY identity(ID)")) 
===
+        Seq("id ASC NULLS FIRST"))
+    }
+  }
+
+  test("CTAS / REPLACE / RTAS carry the clauses too") {
+    parse("CREATE TABLE t USING foo ORDERED BY id AS SELECT 1 AS id") match {
+      case c: CreateTableAsSelect =>
+        assert(c.writeDistributionMode === RANGE && c.writeOrdering.length == 
1)
+      case other => fail(s"unexpected plan: $other")
+    }
+    parse("REPLACE TABLE t (id INT) USING foo ORDERED BY id DESC NULLS LAST") 
match {
+      case r: ReplaceTable =>
+        assert(r.writeDistributionMode === RANGE)
+        assert(r.writeOrdering.head.direction() === SortDirection.DESCENDING)
+        assert(r.writeOrdering.head.nullOrdering() === NullOrdering.NULLS_LAST)
+      case other => fail(s"unexpected plan: $other")
+    }
+    parse("REPLACE TABLE t USING foo UNORDERED AS SELECT 1 AS id") match {
+      case r: ReplaceTableAsSelect =>
+        assert(r.writeDistributionMode === NONE && r.writeOrdering.isEmpty)
+      case other => fail(s"unexpected plan: $other")
+    }
+  }
+
+  test("transforms are allowed in the ordering") {
+    val stmt = "CREATE TABLE t (id INT, ts TIMESTAMP) USING foo " +
+      "ORDERED BY days(ts), bucket(16, id)"
+    parse(stmt) match {
+      case c: CreateTable =>
+        assert(c.writeOrdering.length === 2)
+        assert(c.writeOrdering.map(_.expression().describe()) ===
+          Seq("days(ts)", "bucket(16, id)"))
+      case other => fail(s"unexpected plan: $other")
+    }
+  }
+
+  test("the parenthesised ORDERED BY form parses the same") {
+    Seq("ORDERED BY id DESC, c", "ORDERED BY (id DESC, c)").foreach { clause =>
+      parse(s"CREATE TABLE t (id INT, c STRING) USING foo $clause") match {
+        case c: CreateTable =>
+          assert(WriteSpecCall.render(c.writeOrdering) ===
+            Seq("id DESC NULLS LAST", "c ASC NULLS FIRST"), s"for $clause")
+        case other => fail(s"unexpected plan for $clause: $other")
+      }
+    }
+  }
+
+  test("a repeated write clause is rejected") {
+    Seq(
+      ("DISTRIBUTED BY PARTITION DISTRIBUTED BY PARTITION", "DISTRIBUTED BY 
PARTITION"),
+      ("ORDERED BY id UNORDERED", "ORDERED BY/UNORDERED"),
+      ("UNORDERED LOCALLY ORDERED BY id", "ORDERED BY/UNORDERED")
+    ).foreach { case (clauses, clauseName) =>
+      val stmt = s"CREATE TABLE t (id INT, c STRING) USING foo PARTITIONED BY 
(c) $clauses"
+      checkError(
+        exception = intercept[ParseException](parse(stmt)),
+        condition = "DUPLICATE_CLAUSES",
+        sqlState = "42614",
+        parameters = Map("clauseName" -> clauseName),
+        context = ExpectedContext(stmt, 0, stmt.length - 1))
+    }
+  }
+
+  test("DISTRIBUTED BY PARTITION requires a partitioned table") {
+    Seq(
+      "CREATE TABLE t (id INT) USING foo DISTRIBUTED BY PARTITION",
+      "CREATE TABLE t USING foo DISTRIBUTED BY PARTITION AS SELECT 1 AS id",
+      "REPLACE TABLE t (id INT) USING foo DISTRIBUTED BY PARTITION",
+      "REPLACE TABLE t USING foo DISTRIBUTED BY PARTITION AS SELECT 1 AS id"
+    ).foreach { stmt =>
+      checkError(
+        exception = intercept[ParseException](parse(stmt)),
+        condition = 
"SPECIFY_DISTRIBUTED_BY_PARTITION_WITHOUT_PARTITIONING_IS_NOT_ALLOWED",
+        sqlState = "42908",
+        parameters = Map.empty,
+        context = ExpectedContext(stmt, 0, stmt.length - 1))
+    }
+  }
+
+  test("ordering references are normalized to the schema's spelling") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      assert(orderingOf(analyze("CREATE TABLE testcat.t (ID INT, TS TIMESTAMP) 
USING foo " +
+        "ORDERED BY id, days(ts) DESC")) === Seq("ID ASC NULLS FIRST", 
"days(TS) DESC NULLS LAST"))
+
+      assert(orderingOf(analyze(
+        "CREATE TABLE testcat.t USING foo ORDERED BY id AS SELECT 1 AS ID")) 
===
+        Seq("ID ASC NULLS FIRST"))
+
+      assert(orderingOf(analyze(
+        "CREATE TABLE testcat.t (P STRUCT<X: INT>) USING foo ORDERED BY p.x")) 
===
+        Seq("P.X ASC NULLS FIRST"))
+
+      assert(orderingOf(analyze("CREATE TABLE testcat.t (ID INT) USING foo " +
+        "ORDERED BY truncate(4, ID)")) === Seq("truncate(4, ID) ASC NULLS 
FIRST"))
+      val truncate = "CREATE TABLE testcat.t (ID INT) USING foo ORDERED BY 
truncate(4, id)"
+      checkUnknownColumns(truncate, "`id`")
+
+      // references are normalized independently, so only the unresolvable one 
is reported
+      val multi = "CREATE TABLE testcat.t (ID INT) USING foo ORDERED BY 
bucket(4, id, nope)"
+      checkUnknownColumns(multi, "`nope`")
+    }
+  }
+
+  test("REPLACE TABLE and RTAS normalize the ordering too") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      assert(orderingOf(analyze("REPLACE TABLE testcat.t (ID INT) USING foo 
ORDERED BY id")) ===
+        Seq("ID ASC NULLS FIRST"))
+      assert(orderingOf(analyze(
+        "REPLACE TABLE testcat.t USING foo ORDERED BY id AS SELECT 1 AS ID")) 
===
+        Seq("ID ASC NULLS FIRST"))
+    }
+  }
+
+  test("a case-sensitive session resolves the ordering case-sensitively") {
+    withSQLConf(
+      "spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName,
+      SQLConf.CASE_SENSITIVE.key -> "true") {
+      assert(orderingOf(analyze("CREATE TABLE testcat.t (ID INT) USING foo 
ORDERED BY ID")) ===
+        Seq("ID ASC NULLS FIRST"))
+      val stmt = "CREATE TABLE testcat.t (ID INT) USING foo ORDERED BY id"
+      checkUnknownColumns(stmt, "`id`")
+    }
+  }
+
+  test("a case-insensitive session does not normalize an ApplyTransform's 
references") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      assert(orderingOf(analyze("CREATE TABLE testcat.t (id INT) USING foo 
ORDERED BY ID")) ===
+        Seq("id ASC NULLS FIRST"))
+
+      val ordered = "CREATE TABLE testcat.t (id INT) USING foo ORDERED BY 
truncate(4, ID)"
+      checkUnknownColumns(ordered, "`ID`")
+
+      val partitioned =
+        "CREATE TABLE testcat.t (id INT) USING foo PARTITIONED BY (truncate(4, 
ID))"
+      checkError(
+        exception = intercept[AnalysisException](analyze(partitioned)),
+        condition = 
"UNSUPPORTED_FEATURE.PARTITION_WITH_NESTED_COLUMN_IS_UNSUPPORTED",
+        sqlState = "0A000",
+        parameters = Map("cols" -> "`ID`"),
+        context = ExpectedContext(partitioned, 0, partitioned.length - 1))
+
+      assert(orderingOf(analyze("CREATE TABLE testcat.t (id INT, ts TIMESTAMP) 
USING foo " +
+        "ORDERED BY days(TS), bucket(4, ID)")) ===
+        Seq("days(ts) ASC NULLS FIRST", "bucket(4, id) ASC NULLS FIRST"))
+      Seq("DAYS(TS)" -> "`TS`", "BUCKET(4, ID)" -> "`ID`").foreach { case 
(key, cols) =>
+        val stmt = s"CREATE TABLE testcat.t (id INT, ts TIMESTAMP) USING foo 
ORDERED BY $key"
+        checkUnknownColumns(stmt, cols)
+      }
+    }
+  }
+
+  test("ORDERED BY needs a schema to resolve against") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      Seq("ORDERED BY id", "LOCALLY ORDERED BY id").foreach { clause =>
+        val stmt = s"CREATE TABLE testcat.t USING foo $clause"
+        checkError(
+          exception = intercept[AnalysisException](analyze(stmt)),
+          condition = "SPECIFY_WRITE_ORDERING_IS_NOT_ALLOWED",
+          sqlState = "42601",
+          parameters = Map.empty,
+          context = ExpectedContext(stmt, 0, stmt.length - 1))
+      }
+
+      // UNORDERED asks for no ordering at all, so it has nothing to resolve 
and stays allowed
+      analyze("CREATE TABLE testcat.t USING foo UNORDERED")
+    }
+  }
+
+  test("an ordering on an unknown column is rejected during analysis") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      Seq(
+        "CREATE TABLE testcat.t (id INT) USING foo ORDERED BY nope",
+        "CREATE TABLE testcat.t (id INT) USING foo ORDERED BY days(nope)",
+        "CREATE TABLE testcat.t (id INT) USING foo ORDERED BY truncate(4, 
nope)",
+        "CREATE TABLE testcat.t USING foo ORDERED BY nope AS SELECT 1 AS id"
+      ).foreach { stmt =>
+        checkUnknownColumns(stmt, "`nope`")
+      }
+
+      // a nested struct field resolves
+      analyze("CREATE TABLE testcat.t (p STRUCT<x: INT>) USING foo ORDERED BY 
p.x")
+
+      val throughInt = "CREATE TABLE testcat.t (id INT) USING foo ORDERED BY 
id.x"
+      checkUnknownColumns(throughInt, "`id`.`x`")
+    }
+  }
+
+  test("unknown ordering columns are reported once each, in declaration 
order") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      val stmt = "CREATE TABLE testcat.t (id INT) USING foo " +
+        "ORDERED BY e, d, bucket(4, c, a), id, b, days(f), d"
+      checkUnknownColumns(stmt, "`e`, `d`, `c`, `a`, `b`, `f`")
+    }
+  }
+
+  test("a repeated ordering column is accepted, unlike a repeated partition 
column") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      assert(orderingOf(analyze("CREATE TABLE testcat.t (id INT) USING foo " +
+        "ORDERED BY bucket(4, id), bucket(8, id)")) ===
+        Seq("bucket(4, id) ASC NULLS FIRST", "bucket(8, id) ASC NULLS FIRST"))
+      analyze("CREATE TABLE testcat.t (id INT) USING foo ORDERED BY id, id 
DESC")
+      analyze("CREATE TABLE testcat.t (ts TIMESTAMP) USING foo ORDERED BY 
days(ts), hours(ts)")
+    }
+  }
+
+  test("DISTRIBUTED BY PARTITION is not satisfied by CLUSTER BY") {
+    Seq(
+      "CREATE TABLE t (id INT, c STRING) USING foo CLUSTER BY (c) DISTRIBUTED 
BY PARTITION",
+      "REPLACE TABLE t USING foo CLUSTER BY (c) DISTRIBUTED BY PARTITION AS 
SELECT 'x' AS c"
+    ).foreach { stmt =>
+      checkError(
+        exception = intercept[ParseException](parse(stmt)),
+        condition = 
"SPECIFY_CLUSTER_BY_WITH_DISTRIBUTED_BY_PARTITION_IS_NOT_ALLOWED",
+        sqlState = "42908",
+        parameters = Map.empty,
+        context = ExpectedContext(stmt, 0, stmt.length - 1))
+    }
+
+    parse("CREATE TABLE t (id INT, c STRING) USING foo CLUSTERED BY (c) INTO 4 
BUCKETS " +
+      "DISTRIBUTED BY PARTITION") match {
+      case c: CreateTable => assert(c.writeDistributionMode === HASH)
+      case other => fail(s"unexpected plan: $other")
+    }
+  }
+
+  test("CLUSTER BY and the write clauses both reach the catalog") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      sql("CREATE TABLE testcat.ordered (a INT, b INT) USING foo CLUSTER BY 
(a) ORDERED BY (b)")
+      sql("CREATE TABLE testcat.local (a INT, b INT) USING foo CLUSTER BY (a) 
" +
+        "LOCALLY ORDERED BY (b)")
+      sql("CREATE TABLE testcat.unordered (a INT, b INT) USING foo CLUSTER BY 
(a) UNORDERED")
+
+      assert(calls("testcat") === Seq(
+        WriteSpecCall("createTable", "ordered", RANGE, Seq("b ASC NULLS 
FIRST")),
+        WriteSpecCall("createTable", "local", NONE, Seq("b ASC NULLS FIRST")),
+        WriteSpecCall("createTable", "unordered", NONE, Seq.empty)))
+      Seq("ordered", "local", "unordered").foreach { name =>
+        assert(clusteringOf(loadTable("testcat", name)) === Seq(Seq("a")), 
name)
+      }
+    }
+  }
+
+  test("a catalog with the capability can still reject a combination it does 
not support") {
+    val catalogClass = classOf[ClusterByUnorderedRejectingCatalog].getName
+    withSQLConf("spark.sql.catalog.rejectcat" -> catalogClass) {
+      Seq(
+        "CREATE TABLE rejectcat.t (a INT, b INT) USING foo CLUSTER BY (a) 
UNORDERED",
+        "CREATE TABLE rejectcat.t USING foo CLUSTER BY (a) UNORDERED AS SELECT 
1 AS a, 2 AS b"
+      ).foreach { stmt =>
+        val e = intercept[IllegalArgumentException](sql(stmt))
+        assert(e.getMessage === ClusterByUnorderedRejectingCatalog.MESSAGE, 
stmt)
+        assert(sql("SHOW TABLES IN rejectcat").count() === 0, stmt)
+      }
+
+      sql("CREATE TABLE rejectcat.t (a INT, b INT) USING foo CLUSTER BY (a)")
+      sql("INSERT INTO rejectcat.t VALUES (1, 2)")
+      Seq(
+        "REPLACE TABLE rejectcat.t (a INT, b INT, c INT) USING foo CLUSTER BY 
(a) UNORDERED",
+        "CREATE OR REPLACE TABLE rejectcat.t (a INT, b INT, c INT) USING foo 
CLUSTER BY (a) " +
+          "UNORDERED",
+        "REPLACE TABLE rejectcat.t USING foo CLUSTER BY (a) UNORDERED " +
+          "AS SELECT 3 AS a, 4 AS b, 5 AS c",
+        "CREATE OR REPLACE TABLE rejectcat.t USING foo CLUSTER BY (a) 
UNORDERED " +
+          "AS SELECT 3 AS a, 4 AS b, 5 AS c"
+      ).foreach { stmt =>
+        val e = intercept[IllegalArgumentException](sql(stmt))
+        assert(e.getMessage === ClusterByUnorderedRejectingCatalog.MESSAGE, 
stmt)
+        checkAnswer(sql("SELECT * FROM rejectcat.t"), Row(1, 2))
+        val table = loadTable("rejectcat", "t")
+        assert(table.columns().map(_.name).toSeq === Seq("a", "b"), stmt)
+        assert(clusteringOf(table) === Seq(Seq("a")), stmt)
+      }
+    }
+  }
+
+  test("CREATE TABLE / CTAS / REPLACE TABLE / RTAS hand the distribution and 
ordering " +
+    "to the catalog") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      sql("CREATE TABLE testcat.t (id INT, c STRING) USING foo PARTITIONED BY 
(c) " +
+        "DISTRIBUTED BY PARTITION ORDERED BY id DESC")
+      sql("CREATE TABLE testcat.ctas USING foo ORDERED BY id AS SELECT 1 AS 
id")
+      // a non-staging catalog replaces by dropping and re-creating, so this 
is createTable as well
+      sql("REPLACE TABLE testcat.t (id INT) USING foo LOCALLY ORDERED BY id")
+      sql("REPLACE TABLE testcat.t USING foo PARTITIONED BY (c) DISTRIBUTED BY 
PARTITION " +
+        "ORDERED BY id AS SELECT 1 AS id, 'a' AS c")
+
+      assert(calls("testcat") === Seq(
+        WriteSpecCall("createTable", "t", HASH, Seq("id DESC NULLS LAST")),
+        WriteSpecCall("createTable", "ctas", RANGE, Seq("id ASC NULLS FIRST")),
+        WriteSpecCall("createTable", "t", NONE, Seq("id ASC NULLS FIRST")),
+        WriteSpecCall("createTable", "t", HASH, Seq("id ASC NULLS FIRST"))))
+    }
+  }
+
+  test("CTAS and RTAS write their first load with the declared distribution 
and ordering") {
+    withSQLConf(
+      "spark.sql.catalog.layoutcat" -> 
classOf[LayoutEnforcingInMemoryTableCatalog].getName,
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+      def innerWrite(stmt: String): SparkPlan = {
+        val executions = 
QueryTest.withQueryExecutionsCaptured(spark)(sql(stmt))
+        executions.map(_.executedPlan).collectFirst { case w: V2TableWriteExec 
=> w.query }
+          .getOrElse(fail(s"no inner write for $stmt"))
+      }
+
+      val global = innerWrite(
+        "CREATE TABLE layoutcat.g USING foo ORDERED BY id DESC AS SELECT id 
FROM range(10)")
+      assert(global.collect { case s: ShuffleExchangeExec => 
s.outputPartitioning }
+        .exists(_.isInstanceOf[RangePartitioning]), global.treeString)
+      assert(global.collect { case s: SortExec => s }.nonEmpty, 
global.treeString)
+
+      val local = innerWrite(
+        "CREATE TABLE layoutcat.l USING foo LOCALLY ORDERED BY id DESC AS 
SELECT id FROM range(10)")
+      assert(local.collect { case s: ShuffleExchangeExec => s }.isEmpty, 
local.treeString)
+      assert(local.collect { case s: SortExec => s }.nonEmpty, 
local.treeString)
+
+      val clustered = innerWrite("CREATE OR REPLACE TABLE layoutcat.h USING 
foo " +
+        "PARTITIONED BY (c) DISTRIBUTED BY PARTITION ORDERED BY id " +
+        "AS SELECT id, CAST(id % 3 AS STRING) AS c FROM range(10)")
+      assert(clustered.collect { case s: ShuffleExchangeExec => 
s.outputPartitioning }
+        .exists(_.isInstanceOf[HashPartitioning]), clustered.treeString)
+      assert(clustered.collect { case s: SortExec => s }.nonEmpty, 
clustered.treeString)
+    }
+  }
+
+  test("a statement with no write clause carries neither a distribution nor an 
ordering") {
+    withSQLConf("spark.sql.catalog.testcat" -> 
classOf[RecordingInMemoryTableCatalog].getName) {
+      sql("CREATE TABLE testcat.plain (id INT) USING foo")
+      sql("CREATE TABLE testcat.plain_ctas USING foo AS SELECT 1 AS id")
+      sql("REPLACE TABLE testcat.plain (id INT) USING foo")
+
+      assert(calls("testcat") === Seq(
+        WriteSpecCall("createTable", "plain", null, Seq.empty),
+        WriteSpecCall("createTable", "plain_ctas", null, Seq.empty),
+        WriteSpecCall("createTable", "plain", null, Seq.empty)))
+    }
+  }
+
+  test("the staging catalog gets them on stageCreate / stageReplace / 
stageCreateOrReplace") {
+    val catalogClass = classOf[RecordingStagingInMemoryTableCatalog].getName
+    withSQLConf("spark.sql.catalog.stagingcat" -> catalogClass) {
+      sql("CREATE TABLE stagingcat.t USING foo ORDERED BY id AS SELECT 1 AS 
id")
+      sql("REPLACE TABLE stagingcat.t USING foo LOCALLY ORDERED BY id AS 
SELECT 2 AS id")
+      sql("CREATE OR REPLACE TABLE stagingcat.t USING foo UNORDERED AS SELECT 
3 AS id")
+      sql("REPLACE TABLE stagingcat.t (id INT) USING foo ORDERED BY id DESC 
NULLS FIRST")
+      sql("CREATE OR REPLACE TABLE stagingcat.t (id INT) USING foo LOCALLY 
ORDERED BY id")
+      sql("CREATE TABLE stagingcat.plain USING foo AS SELECT 1 AS id")
+
+      assert(calls("stagingcat") === Seq(
+        WriteSpecCall("stageCreate", "t", RANGE, Seq("id ASC NULLS FIRST")),
+        WriteSpecCall("stageReplace", "t", NONE, Seq("id ASC NULLS FIRST")),
+        WriteSpecCall("stageCreateOrReplace", "t", NONE, Seq.empty),
+        WriteSpecCall("stageReplace", "t", RANGE, Seq("id DESC NULLS FIRST")),
+        WriteSpecCall("stageCreateOrReplace", "t", NONE, Seq("id ASC NULLS 
FIRST")),
+        WriteSpecCall("stageCreate", "plain", null, Seq.empty)))
+    }
+  }
+
+  test("a staging catalog sees an unstaged CREATE TABLE through createTable") {
+    val catalogClass = classOf[RecordingStagingInMemoryTableCatalog].getName
+    withSQLConf("spark.sql.catalog.stagingcat" -> catalogClass) {
+      sql("CREATE TABLE stagingcat.t (id INT) USING foo ORDERED BY id")
+
+      assert(calls("stagingcat") ===
+        Seq(WriteSpecCall("createTable", "t", RANGE, Seq("id ASC NULLS 
FIRST"))))
+    }
+  }
+
+  test("a transactional catalog gets them through the catalog of its 
transaction") {
+    val catalogClass = classOf[RecordingTransactionalTableCatalog].getName
+    withSQLConf("spark.sql.catalog.txncat" -> catalogClass) {
+      sql("CREATE TABLE txncat.t USING foo ORDERED BY id AS SELECT 1 AS id")
+
+      val catalog = spark.sessionState.catalogManager.catalog("txncat")
+        .asInstanceOf[RecordingTransactionalTableCatalog]
+      assert(catalog.observedTransactions.size === 1)

Review Comment:
   Done. Both transaction tests check `txnTables`. I added an RTAS test through 
the transaction's catalog, one that accepts and one that rejects.



##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DelegatingCatalogExtension.java:
##########
@@ -111,6 +131,25 @@ public Table createTable(
     return asTableCatalog().createTable(ident, columns, partitions, 
properties);
   }
 
+  /**
+   * Creates the table with the {@code Column[]} overload, as the default 
does, but rejects a
+   * requested write distribution and ordering, which that overload cannot 
carry, with an unchecked
+   * {@code UNSUPPORTED_FEATURE.TABLE_OPERATION} error. A subclass that reports
+   * {@link 
TableCatalogCapability#SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING}
+   * therefore gets an error instead of a silently dropped request when it 
calls
+   * {@code super.createTable(ident, tableInfo)}. For a {@code REPLACE TABLE} 
of an existing table
+   * the error comes after Spark has dropped the table, because this class is 
not a
+   * {@link StagingTableCatalog}.
+   */
+  @Override
+  public Table createTable(Identifier ident, TableInfo tableInfo)
+      throws TableAlreadyExistsException, NoSuchNamespaceException {
+    if (tableInfo.writeDistributionMode() != null || 
tableInfo.writeOrdering().length > 0) {
+      throw WriteDistributionAndOrdering.unsupportedClausesException(this, 
ident);
+    }
+    return createTable(ident, tableInfo.columns(), tableInfo.partitions(), 
tableInfo.properties());

Review Comment:
   Done differently. After 93 the override had nothing left to do, so I deleted 
it. `super.createTable` now reaches the `TableCatalog` default.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/WriteDistributionAndOrdering.scala:
##########
@@ -0,0 +1,355 @@
+/*
+ * 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.catalyst.util
+
+import java.time.ZoneOffset
+
+import scala.jdk.CollectionConverters._
+import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.{Literal => CatalystLiteral}
+import org.apache.spark.sql.catalyst.parser.ParserInterface
+import org.apache.spark.sql.catalyst.plans.logical.CreateTable
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, 
Table, TableCatalog, TableCatalogCapability, WriteDistributionMode}
+import org.apache.spark.sql.connector.expressions.{Expression, 
GeneralScalarExpression, GetArrayItem, IdentityTransform, Literal, 
NamedReference, SortOrder, Transform, VariantGet}
+import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.connector.{ExpressionWithToString, 
ToStringSQLBuilder}
+import org.apache.spark.sql.types.{DataType, FloatType, StructField, 
StructType, TimestampLTZNanosType, TimestampType}
+import org.apache.spark.unsafe.types.TimestampNanosVal
+import org.apache.spark.util.ArrayImplicits._
+
+/**
+ * Utility methods for the create-time write distribution and ordering 
requested by
+ * `CREATE`/`REPLACE TABLE ... DISTRIBUTED BY PARTITION / [LOCALLY] ORDERED 
BY`.
+ */
+object WriteDistributionAndOrdering {
+
+  // Runs a call into a connector's expression or literal at one node of a 
key. `NonFatal` does not
+  // cover a `LinkageError`, from a connector class whose dependency is 
missing, and it should not
+  // fail DESCRIBE or SHOW CREATE TABLE. A `StackOverflowError` is left to 
`attempt` at the entry
+  // points: caught at every node of a key deeper than the stack, the fallback 
would render each
+  // subtree again at every level and never finish.
+  private def attemptNode[T](body: => T): Try[T] = {
+    try Success(body) catch {
+      case NonFatal(e) => Failure(e)
+      case e: LinkageError => Failure(e)
+    }
+  }
+
+  // Runs a whole rendering at an entry point, where a `StackOverflowError` 
from a key nested
+  // deeper than the stack is caught once, after the stack has unwound.
+  private def attempt[T](body: => T): Try[T] = {
+    try Success(body) catch {
+      case NonFatal(e) => Failure(e)
+      case e @ (_: LinkageError | _: StackOverflowError) => Failure(e)
+    }
+  }
+
+  /** Names the write clauses in an error's operation, as in `CREATE TABLE ... 
<CLAUSES>`. */
+  val CLAUSES = "DISTRIBUTED BY/ORDERED BY/UNORDERED"
+
+  /**
+   * True when a CREATE/REPLACE TABLE statement asked for a write distribution 
or ordering.
+   * `UNORDERED` counts as a request.
+   */
+  def isRequested(
+      writeDistributionMode: WriteDistributionMode,
+      writeOrdering: Seq[SortOrder]): Boolean = {
+    writeDistributionMode != null || writeOrdering.nonEmpty
+  }
+
+  /**
+   * True when the transforms define a partitioning that `DISTRIBUTED BY 
PARTITION` can distribute
+   * by. Bucketing counts; a `cluster_by` transform, from CLUSTER BY or 
reported by a connector,
+   * does not. Matches by name only, since a connector's `cluster_by` may take 
any arguments.
+   */
+  def hasPartitioning(partitioning: Seq[Transform]): Boolean = {
+    partitioning.exists(_.name != "cluster_by")
+  }
+
+  /** The write ordering a table declares, which is empty also for a connector 
that returns null. */
+  def writeOrderingOf(table: Table): Seq[SortOrder] = {
+    
Option(table.writeOrdering()).map(_.toImmutableArraySeq).getOrElse(Seq.empty)
+  }
+
+  /**
+   * The `UNSUPPORTED_FEATURE.TABLE_OPERATION` error for a requested write 
distribution and
+   * ordering that `catalog` cannot carry, as an unchecked exception that a 
Java override of
+   * `createTable` can throw.
+   */
+  def unsupportedClausesException(
+      catalog: CatalogPlugin,
+      ident: Identifier): SparkUnsupportedOperationException = {
+    val error = QueryCompilationErrors.unsupportedTableOperationError(catalog, 
ident, CLAUSES)
+      .asInstanceOf[SparkThrowable]
+    new SparkUnsupportedOperationException(
+      error.getCondition, error.getMessageParameters.asScala.toMap)
+  }
+
+  /** True when `catalog` accepts a create-time write distribution and 
ordering. */
+  def supportsWriteDistributionAndOrdering(catalog: TableCatalog): Boolean = {
+    catalog.capabilities().contains(
+      
TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING)
+  }
+
+  /**
+   * Rejects a create-time write distribution or ordering that the catalog has 
not advertised
+   * support for, before anything is created or dropped.
+   */
+  def validateCatalogForWriteDistributionAndOrdering(
+      catalog: TableCatalog,
+      ident: Identifier,
+      operation: String,
+      writeDistributionMode: WriteDistributionMode,
+      writeOrdering: Seq[SortOrder]): Unit = {
+    if (isRequested(writeDistributionMode, writeOrdering) &&
+        !supportsWriteDistributionAndOrdering(catalog)) {
+      throw QueryCompilationErrors.unsupportedTableOperationError(
+        catalog, ident, s"$operation ... $CLAUSES")
+    }
+  }
+
+  /**
+   * Renders a requested sort key as SQL, for DESCRIBE. When a connector 
expression in it fails to
+   * render, the sort order is rendered with its own `describe`, and when that 
fails too, with
+   * whichever of the class name of its key, its direction and its null 
ordering can be read, or
+   * with its own class name when none can.
+   */
+  def describeSortOrder(sortOrder: SortOrder): String = {

Review Comment:
   Done.



##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/DelegatingCatalogExtension.java:
##########
@@ -32,6 +34,11 @@
  * by calling the built-in session catalog directly. This is created for 
convenience, so that users
  * only need to override some methods where they want to apply custom logic. 
For example, they can
  * override {@code createTable}, do something else before calling {@code 
super.createTable}.
+ * <p>
+ * {@code super.createTable(ident, tableInfo)} cannot carry a requested write 
distribution and

Review Comment:
   Done. The full statement is in `TableCatalogCapability`. The other copies 
link to it or went with the override.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/WriteDistributionAndOrdering.scala:
##########
@@ -0,0 +1,355 @@
+/*
+ * 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.catalyst.util
+
+import java.time.ZoneOffset
+
+import scala.jdk.CollectionConverters._
+import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.{Literal => CatalystLiteral}
+import org.apache.spark.sql.catalyst.parser.ParserInterface
+import org.apache.spark.sql.catalyst.plans.logical.CreateTable
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, 
Table, TableCatalog, TableCatalogCapability, WriteDistributionMode}
+import org.apache.spark.sql.connector.expressions.{Expression, 
GeneralScalarExpression, GetArrayItem, IdentityTransform, Literal, 
NamedReference, SortOrder, Transform, VariantGet}
+import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.connector.{ExpressionWithToString, 
ToStringSQLBuilder}
+import org.apache.spark.sql.types.{DataType, FloatType, StructField, 
StructType, TimestampLTZNanosType, TimestampType}
+import org.apache.spark.unsafe.types.TimestampNanosVal
+import org.apache.spark.util.ArrayImplicits._
+
+/**
+ * Utility methods for the create-time write distribution and ordering 
requested by
+ * `CREATE`/`REPLACE TABLE ... DISTRIBUTED BY PARTITION / [LOCALLY] ORDERED 
BY`.
+ */
+object WriteDistributionAndOrdering {
+
+  // Runs a call into a connector's expression or literal at one node of a 
key. `NonFatal` does not
+  // cover a `LinkageError`, from a connector class whose dependency is 
missing, and it should not
+  // fail DESCRIBE or SHOW CREATE TABLE. A `StackOverflowError` is left to 
`attempt` at the entry
+  // points: caught at every node of a key deeper than the stack, the fallback 
would render each
+  // subtree again at every level and never finish.
+  private def attemptNode[T](body: => T): Try[T] = {
+    try Success(body) catch {
+      case NonFatal(e) => Failure(e)
+      case e: LinkageError => Failure(e)
+    }
+  }
+
+  // Runs a whole rendering at an entry point, where a `StackOverflowError` 
from a key nested
+  // deeper than the stack is caught once, after the stack has unwound.
+  private def attempt[T](body: => T): Try[T] = {
+    try Success(body) catch {
+      case NonFatal(e) => Failure(e)
+      case e @ (_: LinkageError | _: StackOverflowError) => Failure(e)
+    }
+  }
+
+  /** Names the write clauses in an error's operation, as in `CREATE TABLE ... 
<CLAUSES>`. */
+  val CLAUSES = "DISTRIBUTED BY/ORDERED BY/UNORDERED"
+
+  /**
+   * True when a CREATE/REPLACE TABLE statement asked for a write distribution 
or ordering.
+   * `UNORDERED` counts as a request.
+   */
+  def isRequested(
+      writeDistributionMode: WriteDistributionMode,
+      writeOrdering: Seq[SortOrder]): Boolean = {
+    writeDistributionMode != null || writeOrdering.nonEmpty
+  }
+
+  /**
+   * True when the transforms define a partitioning that `DISTRIBUTED BY 
PARTITION` can distribute
+   * by. Bucketing counts; a `cluster_by` transform, from CLUSTER BY or 
reported by a connector,
+   * does not. Matches by name only, since a connector's `cluster_by` may take 
any arguments.
+   */
+  def hasPartitioning(partitioning: Seq[Transform]): Boolean = {
+    partitioning.exists(_.name != "cluster_by")
+  }
+
+  /** The write ordering a table declares, which is empty also for a connector 
that returns null. */
+  def writeOrderingOf(table: Table): Seq[SortOrder] = {
+    
Option(table.writeOrdering()).map(_.toImmutableArraySeq).getOrElse(Seq.empty)
+  }
+
+  /**
+   * The `UNSUPPORTED_FEATURE.TABLE_OPERATION` error for a requested write 
distribution and
+   * ordering that `catalog` cannot carry, as an unchecked exception that a 
Java override of
+   * `createTable` can throw.
+   */
+  def unsupportedClausesException(
+      catalog: CatalogPlugin,
+      ident: Identifier): SparkUnsupportedOperationException = {
+    val error = QueryCompilationErrors.unsupportedTableOperationError(catalog, 
ident, CLAUSES)
+      .asInstanceOf[SparkThrowable]
+    new SparkUnsupportedOperationException(
+      error.getCondition, error.getMessageParameters.asScala.toMap)

Review Comment:
   Done.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/WriteDistributionAndOrdering.scala:
##########
@@ -0,0 +1,355 @@
+/*
+ * 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.catalyst.util
+
+import java.time.ZoneOffset
+
+import scala.jdk.CollectionConverters._
+import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.{Literal => CatalystLiteral}
+import org.apache.spark.sql.catalyst.parser.ParserInterface
+import org.apache.spark.sql.catalyst.plans.logical.CreateTable
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, 
Table, TableCatalog, TableCatalogCapability, WriteDistributionMode}
+import org.apache.spark.sql.connector.expressions.{Expression, 
GeneralScalarExpression, GetArrayItem, IdentityTransform, Literal, 
NamedReference, SortOrder, Transform, VariantGet}
+import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.connector.{ExpressionWithToString, 
ToStringSQLBuilder}
+import org.apache.spark.sql.types.{DataType, FloatType, StructField, 
StructType, TimestampLTZNanosType, TimestampType}
+import org.apache.spark.unsafe.types.TimestampNanosVal
+import org.apache.spark.util.ArrayImplicits._
+
+/**
+ * Utility methods for the create-time write distribution and ordering 
requested by
+ * `CREATE`/`REPLACE TABLE ... DISTRIBUTED BY PARTITION / [LOCALLY] ORDERED 
BY`.
+ */
+object WriteDistributionAndOrdering {
+
+  // Runs a call into a connector's expression or literal at one node of a 
key. `NonFatal` does not
+  // cover a `LinkageError`, from a connector class whose dependency is 
missing, and it should not
+  // fail DESCRIBE or SHOW CREATE TABLE. A `StackOverflowError` is left to 
`attempt` at the entry
+  // points: caught at every node of a key deeper than the stack, the fallback 
would render each
+  // subtree again at every level and never finish.
+  private def attemptNode[T](body: => T): Try[T] = {
+    try Success(body) catch {
+      case NonFatal(e) => Failure(e)
+      case e: LinkageError => Failure(e)
+    }
+  }
+
+  // Runs a whole rendering at an entry point, where a `StackOverflowError` 
from a key nested
+  // deeper than the stack is caught once, after the stack has unwound.
+  private def attempt[T](body: => T): Try[T] = {

Review Comment:
   Done.



##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/WriteDistributionAndOrderingUtilsSuite.scala:
##########
@@ -0,0 +1,456 @@
+/*
+ * 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.catalyst.util
+
+import java.math.BigDecimal
+import java.nio.ByteBuffer
+import java.time.{Duration, Instant, LocalDate, LocalDateTime, LocalTime, 
Period}
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Literal => CatalystLiteral}
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
+import org.apache.spark.sql.catalyst.plans.SQLHelper
+import org.apache.spark.sql.connector.catalog.WriteDistributionMode
+import org.apache.spark.sql.connector.expressions.{Cast, Expression, 
Expressions, Extract, FieldReference, GeneralScalarExpression, GetArrayItem, 
IdentityTransform, NamedReference, NullOrdering, SortDirection, SortOrder, 
Transform, UserDefinedScalarFunc, VariantGet}
+import org.apache.spark.sql.connector.expressions.LogicalExpressions.{apply => 
transform, literal, sort}
+import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+class WriteDistributionAndOrderingUtilsSuite extends SparkFunSuite with 
SQLHelper {
+  import WriteDistributionAndOrdering._
+
+  private class ConnectorReference(names: String*) extends NamedReference {
+    override def fieldNames(): Array[String] = names.toArray
+  }
+
+  // A connector's own expression, which Spark renders with its `describe`.
+  private class ConnectorFunction(rendered: String, args: Expression*) extends 
Expression {
+    override def children(): Array[Expression] = args.toArray
+    override def describe(): String = rendered
+    override def toString: String = rendered
+  }
+
+  // A key whose arguments and `describe` both throw `error`.
+  private class UnrenderableTransform(
+      error: () => Throwable,
+      calls: AtomicInteger = new AtomicInteger(0)) extends Transform {
+    override def name(): String = "unrenderable"
+    override def references(): Array[NamedReference] = Array.empty
+    override def arguments(): Array[Expression] = {
+      calls.incrementAndGet()
+      throw error()
+    }
+    override def toString: String = s"unrenderable(${arguments().mkString(", 
")})"
+  }
+
+  private val id = FieldReference("id")
+
+  private def f(arg: Expression): Transform = transform("f", id, arg)
+
+  private def key(e: Expression): SortOrder =
+    sort(e, SortDirection.ASCENDING, NullOrdering.NULLS_FIRST)
+
+  private val schema = StructType.fromDDL(
+    "id INT, `order` INT, `order-id` INT, s STRUCT<x: INT>, a ARRAY<STRUCT<x: 
INT>>")
+
+  private val replay = replayWith(CatalystSqlParser) _
+
+  private def emitted(e: Expression): Option[String] = {
+    writeClausesSQL(WriteDistributionMode.RANGE, Seq(key(e)), Seq.empty, 
schema, replay)
+  }
+
+  private val spellable: Seq[(String, Expression)] = Seq(
+    "reference" -> id,
+    "nested field" -> FieldReference(Seq("s", "x")),
+    "connector reference" -> transform("f", new 
ConnectorReference("order-id")),
+    "identity" -> IdentityTransform(id),
+    "identity by name" -> transform("identity", id),
+    "byte" -> f(literal(1.toByte)),
+    "short" -> f(literal(1.toShort)),
+    "int" -> f(literal(1)),
+    "long" -> f(literal(1L)),
+    "float" -> f(literal(1.5F)),
+    "double" -> f(literal(1.5D)),
+    "decimal" -> f(literal(new BigDecimal("1.5"))),
+    "string" -> f(literal("x")),
+    "string with escapes" -> f(literal("it's a \\")),
+    "binary" -> f(literal(Array[Byte](1, 2))),
+    "date" -> f(literal(LocalDate.of(1970, 1, 1))),
+    "timestamp" -> f(literal(Instant.parse("2020-11-01T09:30:00Z"))),
+    "timestamp_ntz" -> f(literal(LocalDateTime.of(2020, 1, 1, 10, 0))),
+    "time" -> f(literal(LocalTime.of(12, 0))),
+    "time(7) with trailing zeros" -> f(literal(LocalTime.of(12, 0, 0, 
100000000), TimeType(7))),
+    "time(9) with trailing zeros" -> f(literal(LocalTime.of(12, 0, 0, 
100000000), TimeType(9))),
+    "time(9) without trailing zeros" ->
+      f(literal(LocalTime.of(12, 0, 0, 123456789), TimeType(9))),
+    "time(7) at a whole second" -> f(literal(LocalTime.of(12, 0), 
TimeType(7))),
+    "day-time interval" -> f(literal(Duration.ofHours(1))),
+    "year-month interval" -> f(literal(Period.ofMonths(14))),
+    "bucket" -> transform("bucket", literal(16), id),
+    "bucket without columns" -> transform("bucket", literal(16)),
+    "Spark's bucket" -> Expressions.bucket(16, "id"),
+    "days" -> transform("days", id),
+    "unparsed name" -> transform("Days", id, literal("UTC")))
+
+  private val unspellable: Seq[(String, Expression)] = Seq(
+    "nested transform" -> transform("f", transform("g", id)),
+    "nested identity" -> transform("f", IdentityTransform(id)),
+    "true" -> f(literal(true)),
+    "false" -> f(literal(false)),
+    "null" -> f(literal(null)),
+    "typed null" -> f(literal(null, IntegerType)),
+    "decimal with trailing zeros" ->
+      f(literal(new BigDecimal("1.50"), DecimalType(10, 2))),
+    "decimal in exponent form" ->
+      f(literal(new BigDecimal("0.0000001"), DecimalType(10, 7))),
+    "char" -> f(literal("x", CharType(3))),
+    "varchar" -> f(literal("x", VarcharType(3))),
+    "collated string" -> f(literal("x", StringType("UTF8_LCASE"))),
+    "NaN" -> f(literal(Float.NaN)),
+    "infinity" -> f(literal(Double.PositiveInfinity)),
+    "maximal float" -> f(literal(Float.MaxValue)),
+    "time(6) holding a nanosecond" -> f(literal(LocalTime.ofNanoOfDay(1))),
+    "year interval not in whole years" ->
+      f(literal(13, YearMonthIntervalType(YearMonthIntervalType.YEAR))),
+    "hour interval not in whole hours" ->
+      f(literal(Duration.ofMinutes(90).toNanos / 1000, DayTimeIntervalType(
+        DayTimeIntervalType.HOUR))),
+    "negative-scale decimal" -> f(Expressions.literal(new BigDecimal("1E+3"))),
+    "int value with long type" -> f(literal(5, LongType)),
+    "byte buffer value" -> f(literal(ByteBuffer.wrap(Array[Byte](1, 2)), 
BinaryType)),
+    "bucket with columns first" -> transform("bucket", id, literal(16)),
+    "bucket with long count" -> transform("bucket", literal(16L), id),
+    "bucket with literal column" -> transform("bucket", literal(16), 
literal(1)),
+    "days with two arguments" -> transform("days", id, literal("UTC")),
+    "hours of a literal" -> transform("hours", literal(1)),
+    "missing column" -> FieldReference("missing"),
+    "missing column inside a nested identity" ->
+      transform("f", Expressions.identity("missing")),
+    "column in a different case" -> FieldReference("ID"),
+    "field of an array element" -> FieldReference(Seq("a", "x")),
+    "reference with no field names" -> new ConnectorReference())
+
+  test("a sort key is emitted iff its rendering replays as the same key") {
+    assert(spellable.filter { case (_, e) =>
+      !emitted(e).contains(s"ORDERED BY (${describeSortOrder(key(e))})")
+    }.map(_._1) === Seq.empty)
+    assert(unspellable.filter { case (_, e) => emitted(e).isDefined 
}.map(_._1) === Seq.empty)
+  }
+
+  test("a sort key is emitted iff it replays as the same key under parser 
confs") {
+    val timestamp = f(literal(Instant.parse("2020-11-01T09:30:00Z")))
+    val wrong = Seq(
+      (SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles", timestamp, 
true),
+      (SQLConf.TIMESTAMP_TYPE.key -> "TIMESTAMP_NTZ", timestamp, true),
+      (SQLConf.LEGACY_INTERVAL_ENABLED.key -> "true", 
f(literal(Duration.ofHours(1))), false),
+      (SQLConf.LEGACY_INTERVAL_ENABLED.key -> "true", 
f(literal(Period.ofMonths(14))), false),
+      (SQLConf.ESCAPED_STRING_LITERALS.key -> "true", f(literal("it's")), 
false),
+      (SQLConf.ESCAPED_STRING_LITERALS.key -> "true", f(literal("a\\b")), 
false),
+      (SQLConf.ESCAPED_STRING_LITERALS.key -> "true", f(literal("x")), true),
+      (SQLConf.LEGACY_TIME_PARSER_POLICY.key -> "LEGACY",
+        f(literal(LocalDate.of(-1, 1, 1))), false),
+      (SQLConf.LEGACY_FROM_DAYTIME_STRING.key -> "true",
+        f(literal(Duration.ofHours(26), DayTimeIntervalType(
+          DayTimeIntervalType.DAY, DayTimeIntervalType.HOUR))), false)
+    ).filter { case (conf, e, spelled) => 
withSQLConf(conf)(emitted(e).isDefined != spelled) }
+    assert(wrong.map(_._2.describe) === Seq.empty)
+  }
+
+  test("a name that is a reserved keyword is emitted quoted only where the 
parser needs it") {
+    val order = FieldReference("order")
+    assert(emitted(order) === Some("ORDERED BY (order ASC NULLS FIRST)"))
+    withSQLConf(
+        SQLConf.ANSI_ENABLED.key -> "true",
+        SQLConf.ENFORCE_RESERVED_KEYWORDS.key -> "true") {
+      assert(emitted(order) === Some("ORDERED BY (`order` ASC NULLS FIRST)"))
+      assert(emitted(transform("bucket", literal(4), order)) ===
+        Some("ORDERED BY (`bucket`(4, `order`) ASC NULLS FIRST)"))
+    }
+  }
+
+  test("each mode with a clause form is emitted as that clause") {
+    val keys = Seq(key(id))
+    val partitioning = Seq(Expressions.identity("p"))
+    Seq(
+      (WriteDistributionMode.HASH, keys,
+        "DISTRIBUTED BY PARTITION ORDERED BY (id ASC NULLS FIRST)"),
+      (WriteDistributionMode.HASH, Seq.empty, "DISTRIBUTED BY PARTITION"),
+      (WriteDistributionMode.RANGE, keys, "ORDERED BY (id ASC NULLS FIRST)"),
+      (WriteDistributionMode.NONE, keys, "LOCALLY ORDERED BY (id ASC NULLS 
FIRST)"),
+      (WriteDistributionMode.NONE, Seq.empty, "UNORDERED")
+    ).foreach { case (mode, ordering, clauses) =>
+      assert(writeClausesSQL(mode, ordering, partitioning, schema, replay) === 
Some(clauses),
+        s"for $mode $ordering")
+    }
+  }
+
+  test("a key that does not parse back omits the pair in every mode") {
+    val keys = Seq(key(transform("f", transform("g", id))))
+    Seq(WriteDistributionMode.HASH, WriteDistributionMode.RANGE, 
WriteDistributionMode.NONE)
+      .foreach { mode =>
+        assert(writeClausesSQL(mode, keys, Seq(Expressions.identity("p")), 
schema, replay).isEmpty,
+          s"for $mode")
+      }
+  }
+
+  test("the pairs with no clause form are not emitted") {
+    val keys = Seq(key(id))
+    Seq(
+      (WriteDistributionMode.HASH, keys, Seq.empty[Transform]),
+      (WriteDistributionMode.HASH, Seq.empty, 
Seq(Expressions.apply("cluster_by", id))),
+      (WriteDistributionMode.RANGE, Seq.empty, Seq.empty[Transform]),
+      (null, keys, Seq.empty[Transform])
+    ).foreach { case (mode, ordering, partitioning) =>
+      val anyClauses = (_: String) => Some((mode, ordering))
+      assert(writeClausesSQL(mode, ordering, partitioning, schema, 
anyClauses).isEmpty,
+        s"for $mode")
+    }
+  }
+
+  test("a nanosecond TIMESTAMP_LTZ renders in UTC with an explicit offset") {
+    withSQLConf(
+        SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+        SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+      val parsed = CatalystSqlParser.parseExpression(
+        "TIMESTAMP_LTZ '2020-11-01 
01:30:00.123456789-08:00'").asInstanceOf[CatalystLiteral]
+      val e = f(literal(parsed.value, parsed.dataType))
+      assert(emitted(e) ===
+        Some("ORDERED BY (f(id, TIMESTAMP_LTZ '2020-11-01 
09:30:00.123456789Z') ASC NULLS FIRST)"))
+
+      val wholeSecond = CatalystSqlParser.parseExpression(
+        "TIMESTAMP_LTZ '2020-11-01 
01:30:00.000000000-08:00'").asInstanceOf[CatalystLiteral]
+      assert(emitted(f(literal(wholeSecond.value, wholeSecond.dataType))) ===
+        Some("ORDERED BY (f(id, TIMESTAMP_LTZ '2020-11-01 
09:30:00.000000000Z') ASC NULLS FIRST)"))
+
+      val ntz = CatalystSqlParser.parseExpression("TIMESTAMP_NTZ '2020-01-01 
10:00:00.100000000'")
+        .asInstanceOf[CatalystLiteral]
+      assert(emitted(f(literal(ntz.value, ntz.dataType))).isDefined)
+    }
+  }
+
+  test("a timestamp renders in UTC with an explicit offset") {
+    val e = f(literal(Instant.parse("2020-11-01T09:30:00Z")))
+    Seq("UTC", "America/Los_Angeles", "Asia/Tokyo").foreach { tz =>
+      withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) {
+        assert(describeSortOrder(key(e)) ===
+          "f(id, TIMESTAMP_LTZ '2020-11-01 09:30:00Z') ASC NULLS FIRST", s"in 
$tz")
+      }
+    }
+  }
+
+  test("DESCRIBE renders a literal Catalyst cannot represent with describe") {
+    Seq(
+      Expressions.literal(new BigDecimal("1E+3")) -> "1E+3",
+      literal(5, LongType) -> "5"
+    ).foreach { case (lit, rendered) =>
+      assert(describeSortOrder(key(f(lit))) === s"f(id, $rendered) ASC NULLS 
FIRST")
+    }
+    val buffer = ByteBuffer.wrap(Array[Byte](1, 2))
+    assert(describeSortOrder(key(f(literal(buffer, BinaryType)))) ===
+      s"f(id, $buffer) ASC NULLS FIRST")
+  }
+
+  test("DESCRIBE renders a connector expression in its SQL form") {
+    Seq(
+      new Extract("YEAR", id) -> "EXTRACT(YEAR FROM id)",
+      new UserDefinedScalarFunc("my_udf", "my_udf", Array[Expression](id)) -> 
"my_udf(id)",
+      new GeneralScalarExpression("+", Array[Expression](id, literal(1))) -> 
"id + 1",
+      new GeneralScalarExpression("+", 
Array[Expression](IdentityTransform(id), literal(1))) ->
+        "identity(id) + 1",
+      new Cast(Expressions.bucket(16, "id"), StringType) -> "CAST(bucket(16, 
id) AS STRING)",
+      new Cast(id, ArrayType(IntegerType)) -> "CAST(id AS ARRAY<INT>)",
+      new UserDefinedScalarFunc("zorder_key", "zorder_key", Array[Expression](
+        transform("zorder", id, FieldReference("b")))) -> 
"zorder_key(zorder(id, b))",
+      new VariantGet(id, null, IntegerType, true, null) -> "variant_get(id, 
'null', int)"
+    ).foreach { case (e, rendered) =>
+      assert(describeSortOrder(key(transform("f", e))) === s"f($rendered) ASC 
NULLS FIRST")
+    }
+  }
+
+  test("DESCRIBE renders a connector's own expression under a Spark expression 
with its describe") {
+    val truncate = new ConnectorFunction("truncate(4, id)", id, literal(4))
+    Seq(
+      new GeneralScalarExpression("ABS", Array[Expression](truncate)) -> 
"ABS(truncate(4, id))",
+      new GeneralScalarExpression("+", Array[Expression](truncate, 
literal(1))) ->
+        "(truncate(4, id)) + 1",
+      new Cast(truncate, IntegerType) -> "CAST(truncate(4, id) AS INT)",
+      new Extract("YEAR", truncate) -> "EXTRACT(YEAR FROM truncate(4, id))",
+      new GetArrayItem(truncate, literal(0), true) -> "truncate(4, id)[0]",
+      truncate -> "truncate(4, id)"
+    ).foreach { case (e, rendered) =>
+      assert(describeSortOrder(key(transform("f", e))) === s"f($rendered) ASC 
NULLS FIRST")
+      assert(describeSortOrder(key(e)) === s"$rendered ASC NULLS FIRST")
+    }
+  }
+
+  test("DESCRIBE renders a sort order that cannot be rendered with the class 
of its key") {
+    Seq[() => Throwable](
+      () => new IllegalStateException("broken"),
+      () => new NoClassDefFoundError("org/example/MissingClass"),
+      () => new StackOverflowError()
+    ).foreach { error =>
+      val unrenderable = new UnrenderableTransform(error)
+      assert(describeSortOrder(key(unrenderable)) ===
+        s"${classOf[UnrenderableTransform].getName} ASC NULLS FIRST", 
error().getClass.getName)
+      assert(emitted(unrenderable).isEmpty, error().getClass.getName)
+    }
+  }
+
+  test("a key that fails with a StackOverflowError costs the same at any 
depth") {
+    def callsFor(depth: Int): Int = {
+      val calls = new AtomicInteger(0)
+      val leaf = new UnrenderableTransform(() => new StackOverflowError(), 
calls)
+      val nested = (1 to depth).foldLeft[Expression](leaf) { (child, _) =>
+        new GeneralScalarExpression("ABS", Array[Expression](child))
+      }
+      assert(describeSortOrder(key(nested)) ===
+        s"${classOf[GeneralScalarExpression].getName} ASC NULLS FIRST")
+      assert(emitted(nested).isEmpty)
+      calls.get()
+    }
+
+    assert(callsFor(10) === callsFor(1))
+  }
+
+  // Runs `f` on a thread with a small (256 KiB) stack so that a recursive 
rendering overflows
+  // deterministically, and returns whatever it threw (or null if nothing 
did). The test thread
+  // waits in `join`, which SparkFunSuite's per-test timeout can interrupt.
+  private def runOnSmallStack(f: => Unit): Throwable = {

Review Comment:
   Done. I renamed `f` to `body` as well.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:
##########
@@ -125,22 +140,70 @@ case class ShowCreateTableExec(
     }
   }
 
-  private def showTableLocation(table: Table, builder: StringBuilder): Unit = {
-    val isManagedOption = 
Option(table.properties.get(TableCatalog.PROP_IS_MANAGED_LOCATION))
+  /**
+   * Emits the table's declared write distribution and ordering as clauses 
when parsing them in
+   * this session declares the same pair. Otherwise the pair is omitted: it 
has no clause form, or
+   * the catalog does not accept the clauses, or the statement would create a 
v1 table, or a sort
+   * key references a column the table does not have or does not parse back to 
the same key.
+   */
+  private def showTableWriteDistributionAndOrdering(
+      resolvedTable: ResolvedTable,
+      schema: StructType,
+      partitioning: Array[Transform],
+      provider: Option[String],
+      builder: StringBuilder): Unit = {
+    val table = resolvedTable.table
+    if 
(WriteDistributionAndOrdering.supportsWriteDistributionAndOrdering(resolvedTable.catalog))
 {
+      val writeDistributionMode = table.writeDistributionMode()
+      val writeOrdering = WriteDistributionAndOrdering.writeOrderingOf(table)
+      if (WriteDistributionAndOrdering.isRequested(writeDistributionMode, 
writeOrdering)) {

Review Comment:
   Done.



##########
sql/core/src/test/scala/org/apache/spark/sql/connector/CreateTableWriteOrderSuite.scala:
##########
@@ -0,0 +1,1512 @@
+/*
+ * 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.connector
+
+import java.util
+import java.util.Locale
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.SparkThrowable
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, 
TableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.catalyst.plans.logical.{CreateTable, 
CreateTableAsSelect, LogicalPlan, ReplaceTable, ReplaceTableAsSelect, 
V2CreateTablePlan}
+import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, 
RangePartitioning}
+import org.apache.spark.sql.catalyst.util.WriteDistributionAndOrdering
+import org.apache.spark.sql.connector.catalog.{Column, 
DelegatingCatalogExtension, DelegatingTable, Identifier, 
InMemoryRowLevelOperationTableCatalog, InMemoryTable, InMemoryTableCatalog, 
StagedTable, StagingInMemoryTableCatalog, Table, TableCatalogCapability, 
TableInfo, Txn, TxnTableCatalog, WriteDistributionMode}
+import 
org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+import org.apache.spark.sql.connector.catalog.WriteDistributionMode.{HASH, 
NONE, RANGE}
+import org.apache.spark.sql.connector.catalog.transactions.{Transaction, 
TransactionInfo}
+import org.apache.spark.sql.connector.distributions.Distributions
+import org.apache.spark.sql.connector.expressions.{ClusterByTransform, 
Expression, FieldReference, LogicalExpressions, NamedReference, NullOrdering, 
SortDirection, SortOrder, Transform}
+import org.apache.spark.sql.connector.expressions.LogicalExpressions.literal
+import org.apache.spark.sql.execution.{SortExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Tests the create-time write distribution and ordering clauses: 
CREATE/REPLACE TABLE ...
+ * DISTRIBUTED BY PARTITION ... [LOCALLY] ORDERED BY ... | UNORDERED. Covers 
what the parser
+ * produces, what reaches the catalog, and how a catalog without the 
capability is rejected.
+ */
+class CreateTableWriteOrderSuite extends QueryTest with SharedSparkSession {
+
+  // The catalog manager caches catalogs for the session; reset it so tests 
don't share tables.
+  override def afterEach(): Unit = {
+    spark.sessionState.catalogManager.reset()
+    super.afterEach()
+  }
+
+  private def parse(sql: String): LogicalPlan = 
spark.sessionState.sqlParser.parsePlan(sql)
+
+  private def analyze(sql: String): LogicalPlan =
+    spark.sessionState.executePlan(parse(sql)).analyzed
+
+  private def orderingOf(plan: LogicalPlan): Seq[String] = plan match {
+    case create: V2CreateTablePlan => 
WriteSpecCall.render(create.writeOrdering)
+    case other => fail(s"unexpected plan: $other")
+  }
+
+  private def calls(catalogName: String): Seq[WriteSpecCall] =
+    spark.sessionState.catalogManager.catalog(catalogName)
+      .asInstanceOf[RecordsWriteSpecs].recordedCalls
+
+  private def writeClauseLines(ddl: String): Seq[String] = 
ddl.split("\n").toSeq.filter { line =>
+    Seq("DISTRIBUTED BY", "ORDERED BY", "LOCALLY ORDERED BY", 
"UNORDERED").exists(line.startsWith)
+  }
+
+  private def checkUnknownColumns(stmt: String, cols: String): Unit = {
+    checkError(
+      exception = intercept[AnalysisException](analyze(stmt)),
+      condition = "WRITE_ORDERING_WITH_UNKNOWN_COLUMN",
+      sqlState = "42703",
+      parameters = Map("cols" -> cols),
+      context = ExpectedContext(stmt, 0, stmt.length - 1))
+  }
+
+  private def checkRejected(stmt: String, tableName: String): Unit = {
+    checkError(
+      exception = intercept[Exception](sql(stmt)).asInstanceOf[SparkThrowable],

Review Comment:
   Done, with one change. Both catalogs reach the Java defaults now, so both 
throw `SparkUnsupportedOperationException`. `checkRejected` pins that class.



-- 
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: reviews-unsubscr...@spark.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org