dongjoon-hyun commented on code in PR #58153:
URL: https://github.com/apache/spark/pull/58153#discussion_r4383848516
##########
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:
**[93] The four `TableInfo` defaults this sentence names still drop a
request without an error, so each class has to add its own guard.**
71 and 86 were each fixed with a throwing `createTable(Identifier,
TableInfo)` override in one class (V2SessionCatalog.scala:279-286 and
DelegatingCatalogExtension.java:144-151). The defaults that turn a `TableInfo`
into the `Column[]` call are unchanged: TableCatalog.java:382-385 and
StagingTableCatalog.java:109-112, 175-178 and 239-245 forward only the columns,
partitions and properties. So a catalog that reports the capability but misses
one override still loses the request silently. For example, take a
`StagingTableCatalog` that overrides `createTable`, `stageCreate` and
`stageReplace` with their `TableInfo` forms, but implements
`stageCreateOrReplace` only through the `Column[]` overload, e.g. while it
migrates one overload at a time:
- `CREATE OR REPLACE TABLE cat.t (id INT) USING foo ORDERED BY id` passes
the check at DataSourceV2Strategy.scala:331.
- `AtomicReplaceTableExec` calls `catalog.stageCreateOrReplace(identifier,
tableInfo)` (ReplaceTableExec.scala:98), which falls through to the default at
StagingTableCatalog.java:239-245.
- The table is created without the ordering, and nothing reports it. `CREATE
OR REPLACE TABLE ... AS SELECT` does the same through
WriteToDataSourceV2Exec.scala:273.
The same happens to a catalog that forwards `capabilities()` from an inner
catalog but implements only the `Column[]` `createTable`, which is the shape of
a transaction's catalog that L118-120 ask to report the capability alike.
Could the four defaults reject a request, e.g. throw
`WriteDistributionAndOrdering.unsupportedClausesException(this, ident)` when
`writeDistributionMode()` is not null or `writeOrdering()` is not empty, and
otherwise call the `Column[]` overload as now? This changes nothing for a
catalog without the capability or for a statement without the clauses, because
Spark sends a request only after the capability check. A guarded `stage*`
default throws before anything is staged, so a staging catalog keeps its table.
Java defaults here already throw Spark errors (TableCatalog.java:348 and 474),
and only method bodies change. The guards in V2SessionCatalog and
`DelegatingCatalogExtension` would then repeat the defaults.
`DelegatingCatalogExtension.capabilities()` should still strip the capability,
so that a non-staging REPLACE is rejected before the drop, and L108-110 and
L128-139 could say that the defaults reject the request. A catalog that honors
the request in its own override and then
calls a default would have to call the `Column[]` overload instead, but the
capability is new in 4.4.0, so no released catalog does that. This is narrower
than the "Fixing the default itself is out of scope" note in the PR
description, which is about forwarding every `TableInfo`-only field.
##########
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:
**[94] Nothing can interrupt this `join`, so a regression of point 85 hangs
the build instead of failing this test.**
L334-335 say that SparkFunSuite's per-test timeout can interrupt the `join`.
SparkFunSuite.test wraps each test in `failAfter(Span(timeout,
Minutes))(testBody)` (SparkFunSuite.scala:42-44), but no implicit `Signaler` is
in scope there, so ScalaTest uses its default, `DoNotSignal`, which never
interrupts the test thread; `failAfter` only reports the timeout after the body
returns. A `Signaler` declared in this suite would not help either, because the
implicit is resolved inside SparkFunSuite.test. Your reply on 85 says the same
about the rendering loop ("`SparkTestSuite` mixes in `TimeLimits` with no
signaler ... so the timeout cannot interrupt it"), and a thread waiting in
`join` is no different.
So suppose a later change makes `attemptNode` catch `StackOverflowError`
again. The test at L317 fails at once, as L350-351 intend, but this test then
renders the self-containing key on the 256 KiB thread, where each level renders
its subtree twice, and `t.join()` never returns. The catalyst test JVM waits
until the job's time limit instead of failing, and the suites that would run
after this one in the same JVM never start.
Could the `join` be bounded, e.g. `t.join(TimeUnit.MINUTES.toMillis(1))`
followed by `assert(!t.isAlive, "rendering did not finish")`, or wrapped in
`failAfter` with a local `implicit val signaler: Signaler = ThreadSignaler`, as
KeyLockSuite.scala:32 does? Either way, L334-335 need to change.
##########
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:
**[95] This test cannot tell whether the request reached
`Transaction#catalog()` or the plugin itself.**
`calls("txncat")` (L73-75) reads what the plugin recorded, and
`RecordingTransactionalTableCatalog` records only in its own `createTable`
(L1227-1230). The transaction's catalog, `TxnTableCatalog`, records nothing and
forwards to the plugin (txns.scala:242-245), so the recorded call is the same
whether `CreateTableAsSelectExec` calls the transaction's catalog or the
plugin. L548 only shows that a transaction was begun and closed:
`observedTransactions` grows in `clearActiveTransaction`, which `Txn.close`
calls (txns.scala:90-95) whichever catalog created the table. If the exec
called the plugin directly, e.g. after resolving the catalog again outside the
transaction, this test and the one at L554 would still pass.
Could this test also check the transaction's own catalog, e.g.
`assert(catalog.lastTransaction.catalog.txnTables.contains(Identifier.of(Array.empty[String],
"t")))`? `TxnTableCatalog.createTable` pins the new table through `loadTable`
(txns.scala:195-201), and its `writeTarget` (txns.scala:139) is set only when
the write goes through it. The `REPLACE TABLE ... AS SELECT` check through the
transaction's catalog (DataSourceV2Strategy.scala:505) has no test either; an
RTAS statement here would cover it.
##########
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:
**[100] `.asScala.toMap` is not needed here.**
`SparkUnsupportedOperationException` already has a `(String,
java.util.Map[String, String])` constructor (SparkException.scala:306-309), so
this can be `new SparkUnsupportedOperationException(error.getCondition,
error.getMessageParameters)`. The `scala.jdk.CollectionConverters._` import
that 99becb1 added at L22 is used only here.
Nit: could the conversion and the import go?
##########
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:
**[98] 99becb1 states the same fact in four Javadocs.**
That `createTable(Identifier, TableInfo)` rejects a requested layout because
the `Column[]` overload cannot carry it is now said here (L38-41), in
`capabilities()` (L66-68), on the method itself (L135-137) and in
TableCatalogCapability.java:128-130. Three of them are in this file, and this
commit already had to rewrite the two older copies when the behavior changed.
The paragraph at L38-41 also came from my docs-only alternative in 86, which
the code change made unnecessary.
Nit: could the full statement stay on the method (L134-143), and could the
class Javadoc, `capabilities()` and TableCatalogCapability.java:128-131 link to
it with `{@link #createTable(Identifier, TableInfo)}`? With 93, most of this
would reduce to "the defaults reject the request".
##########
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:
**[102] `describeSortOrder(null)` returns "null" only after five caught
`NullPointerException`s.**
For a null element, `sortOrderToSQL` throws at L255, `sortOrder.describe`
throws, and so do the three `attempt`s at L144-146, before
`Option(sortOrder).fold("null")(...)` at L147 returns "null". The result is
right, but neither the code nor the Scaladoc (L131-136) shows that null is
handled, while the same file checks for null explicitly at L265 and L294.
Nit: could this start with `if (sortOrder == null) "null" else ...`, and
could L147 end with `sortOrder.getClass.getName`? Both give the same result for
every input.
##########
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:
**[104] This helper sits between two tests, away from the suite's other
members.**
Every other member of this suite is in the block at L40-149, before the
first test at L151, including single-use helpers such as `ConnectorFunction`
(L45) and `UnrenderableTransform` (L52). The repository's AGENTS.md asks to
"put the new member with the code it belongs with" instead of "dropping it
wherever it is first used". The by-name parameter `f` also shadows the suite's
`f(arg)` at L66.
Nit: could it move to that block, or become local to its test as `callsFor`
is, and could `f` be renamed, e.g. to `body`?
##########
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:
**[97] The ordering is rendered before anything checks that the pair has a
clause form, and rendered again after a failure that quoting cannot change.**
Both quote functions run in turn until `find` accepts a rendering, and the
match at L173-182 that decides the clause form runs only after the whole
ordering has been rendered (L168-170). So:
- For a pair with no clause form, such as an ordering with a null mode,
which point 75 called the likely case for a connector that records only a sort
order, or `HASH` on a table without partitioning, every SHOW CREATE TABLE
renders the ordering twice and drops both results at L181.
- When a rendering throws, from a connector expression that fails, a
`LinkageError` or a key deeper than the stack, the second rendering fails the
same way, because neither `quoteIfNeeded` nor `quoteIdentifier` throws for a
name. For the self-containing key in
WriteDistributionAndOrderingUtilsSuite.scala:352, `emitted` overflows the stack
twice.
Nit: could the clause form be decided first, from the mode,
`writeOrdering.nonEmpty` and `partitioned`, and could a rendering that throws
end the search? The retry is still needed when the first rendering does not
parse, or parses as another pair, e.g. a column named `true` under
`spark.sql.ansi.enforceReservedKeywords`. The cost is small, a few microseconds
per key and only in a catalog that reports the capability.
##########
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:
**[96] `checkRejected` no longer pins the exception class of either
rejection path.**
Before 99becb1, the test that uses this helper (L627-640) called
`checkUnsupported`, which intercepts an `AnalysisException` (L102).
`intercept[Exception](sql(stmt)).asInstanceOf[SparkThrowable]` accepts any
class. The `ForwardingSessionCatalog` rows go through V2SessionCatalog, which
throws an `AnalysisException`, and the `SuperCallingSessionCatalog` rows go
through `DelegatingCatalogExtension`, which throws a
`SparkUnsupportedOperationException`, as its Javadoc promises
(DelegatingCatalogExtension.java:39 and 136-137). A change to either class
still passes. An exception that is not a `SparkThrowable`, e.g. an
`IllegalStateException`, fails here as a `ClassCastException` without the
original exception as its cause.
Nit: could this use `intercept[SparkThrowable]`, as
DataFrameToSchemaSuite.scala:58 does, and check the expected class per catalog,
e.g. `classOf[AnalysisException]` for `ForwardingSessionCatalog` and
`classOf[SparkUnsupportedOperationException]` for `SuperCallingSessionCatalog`?
##########
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:
**[99] Correction to my point 86: this line repeats the body of the
`TableCatalog` default, for the reason I gave in 92.**
In 86 I suggested calling `createTable(ident, info.columns(),
info.partitions(), info.properties())` "as the default does", and this line
does exactly that. But in 92 I asked V2SessionCatalog to call
`super.createTable(ident, tableInfo)`, so that a later change to the default,
e.g. one that passes on `constraints()`, reaches it too. Before 99becb1, a
subclass's `super.createTable(ident, info)` bound to that default
(TableCatalog.java:382-385). Now it binds to this copy, so every
`DelegatingCatalogExtension` subclass that does not override this method would
miss such a change. L147 also writes out
`WriteDistributionAndOrdering.isRequested` again, which
V2SessionCatalog.scala:280-281 calls.
Nit: could this be `return CatalogExtension.super.createTable(ident,
tableInfo);`? `CatalogExtension` is this class's only direct superinterface and
does not redeclare the method, so this compiles (`TableCatalog.super` would
not), and it still dispatches to this class's `Column[]` overload. An
`isRequested(TableInfo)` overload in `WriteDistributionAndOrdering` would let
both overrides share the check, since the current one takes a Scala `Seq`. With
93, this override would reduce to the `super` call.
##########
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:
**[103] Correction to my point 44: since the fix for 88, this `isRequested`
guard no longer changes anything.**
I asked for this check in 44 so that the provider lookup would not run for a
table that declares nothing. Since 99becb1 applies `createsV1Table` as a filter
on the result of `writeClausesSQL` (L166), a pair with a null mode and an empty
ordering never reaches it: `writeClausesSQL` renders nothing, the match at
WriteDistributionAndOrdering.scala:173-182 returns `None` (a null mode does not
match `WriteDistributionMode.HASH`, which is compared with `==`), `replay` is
never called, and `None.filter` does not call `createsV1Table`. The output is
the same, and the `"plain" -> false` row of the counting test at
CreateTableWriteOrderSuite.scala:812 still holds.
Nit: could the inner `if` go, to save a nesting level? DescribeTableExec
still needs its own check for its header row.
##########
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:
**[101] `attempt` repeats both cases of `attemptNode`.**
The two methods differ only in `StackOverflowError`, but L62-63 list the
`NonFatal` and `LinkageError` cases again, and two comments (L46-50 and L58-59)
explain the difference.
Nit: could this be `private def attempt[T](body: => T): Try[T] = try
attemptNode(body) catch { case e: StackOverflowError => Failure(e) }`, with one
comment? It behaves the same for every `Throwable`: `NonFatal` and
`LinkageError` become the inner `Failure`, a `StackOverflowError` passes
through `attemptNode` and becomes the outer one, and anything else propagates
as now. The body is still evaluated once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
