anuragmantri commented on code in PR #58153:
URL: https://github.com/apache/spark/pull/58153#discussion_r4186475570
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:
##########
@@ -125,6 +127,72 @@ case class ShowCreateTableExec(
}
}
+ /**
+ * True for a sort key expression the `transform` grammar rule can
represent: a column
+ * reference, or a transform with at least one argument, each a column
reference or a literal
+ * the `constant` rule can spell with the same type.
+ */
+ private def isSpellable(e: V2Expression): Boolean = e match {
+ case _: NamedReference => true
+ case t: Transform =>
+ t.arguments().nonEmpty && t.arguments().forall {
+ case _: NamedReference => true
+ case l: Literal[_] => isSpellableLiteral(l)
+ case _ => false
+ }
+ case _ => false
+ }
+
+ // Lists only the types whose rendering parses back as a `constant` of the
same type. A typed
+ // NULL, a collated string, NaN, and infinity have none, nor does a FLOAT of
maximal magnitude,
+ // whose shortest rendering exceeds the parser's range.
+ private def isSpellableLiteral(l: Literal[_]): Boolean = (l.value,
l.dataType) match {
+ case (null, dataType) => dataType == NullType
+ case (f: Float, FloatType) => java.lang.Float.isFinite(f) && math.abs(f) <
Float.MaxValue
+ case (d: Double, DoubleType) => java.lang.Double.isFinite(d)
+ case (_, s: StringType) =>
DataTypeUtils.isDefaultStringCharOrVarcharType(s)
+ case (_, BooleanType | ByteType | ShortType | IntegerType | LongType | _:
DecimalType |
Review Comment:
Thanks. A `TRUE`, `FALSE` or `NULL` argument now parses back as a column
reference when the keyword is not reserved, so the comparison fails and the key
is omitted. Under `spark.sql.ansi.enforceReservedKeywords` they parse back as
constants and are emitted. The docs now say so.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:
##########
@@ -125,6 +127,72 @@ case class ShowCreateTableExec(
}
}
+ /**
+ * True for a sort key expression the `transform` grammar rule can
represent: a column
+ * reference, or a transform with at least one argument, each a column
reference or a literal
+ * the `constant` rule can spell with the same type.
+ */
+ private def isSpellable(e: V2Expression): Boolean = e match {
+ case _: NamedReference => true
+ case t: Transform =>
+ t.arguments().nonEmpty && t.arguments().forall {
+ case _: NamedReference => true
+ case l: Literal[_] => isSpellableLiteral(l)
+ case _ => false
+ }
+ case _ => false
+ }
+
+ // Lists only the types whose rendering parses back as a `constant` of the
same type. A typed
+ // NULL, a collated string, NaN, and infinity have none, nor does a FLOAT of
maximal magnitude,
+ // whose shortest rendering exceeds the parser's range.
+ private def isSpellableLiteral(l: Literal[_]): Boolean = (l.value,
l.dataType) match {
+ case (null, dataType) => dataType == NullType
+ case (f: Float, FloatType) => java.lang.Float.isFinite(f) && math.abs(f) <
Float.MaxValue
+ case (d: Double, DoubleType) => java.lang.Double.isFinite(d)
+ case (_, s: StringType) =>
DataTypeUtils.isDefaultStringCharOrVarcharType(s)
+ case (_, BooleanType | ByteType | ShortType | IntegerType | LongType | _:
DecimalType |
+ BinaryType | DateType | TimestampType | TimestampNTZType | _:
DayTimeIntervalType |
+ _: YearMonthIntervalType) => true
+ case _ => false
+ }
+
+ /**
+ * Emits the table's declared write distribution and ordering as clauses.
Pairs with no clause
+ * form, such as `HASH` without partitioning or an unspellable sort key, are
omitted.
+ */
+ private def showTableWriteDistributionAndOrdering(
+ table: Table,
+ builder: StringBuilder): Unit = {
+ if (table.writeOrdering().forall(o => isSpellable(o.expression()))) {
+ val orderBy = if (table.writeOrdering().nonEmpty) {
+ Some(table.writeOrdering()
+ .map(WriteDistributionAndOrdering.describeSortOrder)
+ .mkString("ORDERED BY (", ", ", ")"))
+ } else {
+ None
+ }
+ // Bucketing counts as partitioning here; CLUSTER BY does not.
+ val hasPartitioning = table.partitioning.exists {
+ case ClusterByTransform(_) => false
Review Comment:
Good catch on both. `writeSpecsFrom` and `SHOW CREATE TABLE` now share
`WriteDistributionAndOrdering.hasPartitioning`, which matches by name (any
transform not named `cluster_by`), casts nothing, and is evaluated only in the
HASH cases. A connector's `cluster_by(4)` prints as `PARTITIONED BY
(cluster_by(4))` again. The parser now rejects `PARTITIONED BY (cluster_by(a))
DISTRIBUTED BY PARTITION`, so a replay can no longer lose HASH, and @szehon-ho
's case still holds.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:
##########
@@ -125,6 +127,72 @@ case class ShowCreateTableExec(
}
}
+ /**
+ * True for a sort key expression the `transform` grammar rule can
represent: a column
+ * reference, or a transform with at least one argument, each a column
reference or a literal
+ * the `constant` rule can spell with the same type.
+ */
+ private def isSpellable(e: V2Expression): Boolean = e match {
+ case _: NamedReference => true
+ case t: Transform =>
+ t.arguments().nonEmpty && t.arguments().forall {
+ case _: NamedReference => true
+ case l: Literal[_] => isSpellableLiteral(l)
+ case _ => false
+ }
+ case _ => false
+ }
+
+ // Lists only the types whose rendering parses back as a `constant` of the
same type. A typed
+ // NULL, a collated string, NaN, and infinity have none, nor does a FLOAT of
maximal magnitude,
+ // whose shortest rendering exceeds the parser's range.
+ private def isSpellableLiteral(l: Literal[_]): Boolean = (l.value,
l.dataType) match {
+ case (null, dataType) => dataType == NullType
+ case (f: Float, FloatType) => java.lang.Float.isFinite(f) && math.abs(f) <
Float.MaxValue
+ case (d: Double, DoubleType) => java.lang.Double.isFinite(d)
+ case (_, s: StringType) =>
DataTypeUtils.isDefaultStringCharOrVarcharType(s)
+ case (_, BooleanType | ByteType | ShortType | IntegerType | LongType | _:
DecimalType |
+ BinaryType | DateType | TimestampType | TimestampNTZType | _:
DayTimeIntervalType |
Review Comment:
Done. `toSQL` renders a TIMESTAMP as `TIMESTAMP_LTZ '<UTC wall time>Z'`. A
new test creates the table in America/Los_Angeles with `TIMESTAMP '2020-11-01
01:30:00-08:00'` and replays the DDL in Los Angeles, in Asia/Tokyo and with
`spark.sql.timestampType=TIMESTAMP_NTZ`. Each replay declares the same key.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -7339,6 +7339,13 @@
],
"sqlState" : "42908"
},
+ "SPECIFY_DISTRIBUTED_BY_PARTITION_WITHOUT_PARTITIONING_IS_NOT_ALLOWED" : {
+ "message" : [
+ "Cannot specify DISTRIBUTED BY PARTITION for a table that has no
partitioning.",
+ "Please add PARTITIONED BY or CLUSTERED BY ... INTO ... BUCKETS, or drop
the clause. A statement that defines no schema (no column list, no typed
partition columns, and no AS SELECT) cannot declare partitioning, so it cannot
use DISTRIBUTED BY PARTITION."
Review Comment:
Done. `writeSpecsFrom` raises a dedicated
`SPECIFY_CLUSTER_BY_WITH_DISTRIBUTED_BY_PARTITION_IS_NOT_ALLOWED` (42908) when
`CLUSTER BY` is present. The message says to replace `CLUSTER BY` with
`PARTITIONED BY` or bucketing, or to drop `DISTRIBUTED BY PARTITION`. The
missing-partitioning message also says now that a `cluster_by(...)` transform
in `PARTITIONED BY` is clustering. Both stay at parse time.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableInfo.java:
##########
@@ -82,6 +120,27 @@ public Builder withConstraints(Constraint[] constraints) {
return this;
}
+ /**
+ * Sets the requested write distribution. See {@link
TableInfo#writeDistributionMode()}.
+ *
+ * @since 4.4.0
+ */
+ public Builder withWriteDistributionMode(WriteDistributionMode
writeDistributionMode) {
+ this.writeDistributionMode = writeDistributionMode;
+ return this;
+ }
+
+ /**
+ * Sets the requested write ordering, which must not be null. See
+ * {@link TableInfo#writeOrdering()}.
+ *
+ * @since 4.4.0
+ */
+ public Builder withWriteOrdering(SortOrder[] writeOrdering) {
Review Comment:
Done. `TableInfo` now rejects a null write ordering in its constructor, so
both `build()` and a subclass are covered, and a new test checks it.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalogCapability.java:
##########
@@ -92,5 +92,33 @@ 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:
Good catch. `DelegatingCatalogExtension.capabilities()` now removes
`SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING` from the
delegate's set, so a subclass reports it only if it overrides the `TableInfo`
overloads and adds it back. `createTable(Identifier, TableInfo)` is still not
forwarded. Both Javadocs say so, and a new test checks that such an extension
rejects `ORDERED BY` before anything is created. I left
`SUPPORT_TABLE_CONSTRAINT` as it is, since that gap predates this PR.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:
##########
@@ -125,6 +127,72 @@ case class ShowCreateTableExec(
}
}
+ /**
+ * True for a sort key expression the `transform` grammar rule can
represent: a column
+ * reference, or a transform with at least one argument, each a column
reference or a literal
+ * the `constant` rule can spell with the same type.
+ */
+ private def isSpellable(e: V2Expression): Boolean = e match {
+ case _: NamedReference => true
+ case t: Transform =>
+ t.arguments().nonEmpty && t.arguments().forall {
+ case _: NamedReference => true
+ case l: Literal[_] => isSpellableLiteral(l)
+ case _ => false
+ }
+ case _ => false
+ }
+
+ // Lists only the types whose rendering parses back as a `constant` of the
same type. A typed
+ // NULL, a collated string, NaN, and infinity have none, nor does a FLOAT of
maximal magnitude,
+ // whose shortest rendering exceeds the parser's range.
+ private def isSpellableLiteral(l: Literal[_]): Boolean = (l.value,
l.dataType) match {
+ case (null, dataType) => dataType == NullType
+ case (f: Float, FloatType) => java.lang.Float.isFinite(f) && math.abs(f) <
Float.MaxValue
+ case (d: Double, DoubleType) => java.lang.Double.isFinite(d)
+ case (_, s: StringType) =>
DataTypeUtils.isDefaultStringCharOrVarcharType(s)
+ case (_, BooleanType | ByteType | ShortType | IntegerType | LongType | _:
DecimalType |
+ BinaryType | DateType | TimestampType | TimestampNTZType | _:
DayTimeIntervalType |
+ _: YearMonthIntervalType) => true
Review Comment:
Good catch, and thanks for spotting that it was a regression. With the
parse-back check there is no type list to miss. `TIME '12:00:00'` is in the
round-trip test, and a TIME with 7 to 9 fractional digits is omitted, because
it parses back as TIME(6). The nanosecond timestamp types and
`CalendarIntervalType` are handled by the same check.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala:
##########
@@ -3692,6 +3693,236 @@ class PlanResolutionSuite extends SharedSparkSession
with AnalysisTest {
}
}
+ test("SPARK-34586: v2 table creation (global writeOrdering)") {
Review Comment:
Folded the six into one table-driven test over CREATE, CTAS, REPLACE and
RTAS with the default and a known catalog. It asserts only the plan type, the
mode, the ordering and, for `DISTRIBUTED BY PARTITION`, the partitioning.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:
##########
@@ -125,6 +127,72 @@ case class ShowCreateTableExec(
}
}
+ /**
+ * True for a sort key expression the `transform` grammar rule can
represent: a column
+ * reference, or a transform with at least one argument, each a column
reference or a literal
+ * the `constant` rule can spell with the same type.
+ */
+ private def isSpellable(e: V2Expression): Boolean = e match {
+ case _: NamedReference => true
+ case t: Transform =>
Review Comment:
Covered by the parse-back check. `bucket(id, 16)` fails to parse and
`days(id, 'UTC')` fails with `TRANSFORM_WRONG_NUM_ARGS`, so both are omitted,
and both are in `ORDERING_OVERRIDE` with omit-and-replay rows. A connector
bucket with a BIGINT count is omitted too, because the parser narrows it to INT.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/WriteDistributionAndOrdering.scala:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.expressions.{Literal => CatalystLiteral}
+import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog,
TableCatalogCapability, WriteDistributionMode}
+import org.apache.spark.sql.connector.expressions.{Expression,
IdentityTransform, Literal, SortOrder, Transform}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.types.FloatType
+
+/**
+ * Utility methods for the create-time write distribution and ordering
requested by
+ * `CREATE`/`REPLACE TABLE ... DISTRIBUTED BY PARTITION / [LOCALLY] ORDERED
BY`.
+ */
+object WriteDistributionAndOrdering {
+
+ /**
+ * 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
+ }
+
+ /**
+ * 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) &&
+ !catalog.capabilities().contains(
+
TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING))
{
+ throw QueryCompilationErrors.unsupportedTableOperationError(
+ catalog, ident, s"$operation ... DISTRIBUTED BY/ORDERED BY/UNORDERED")
+ }
+ }
+
+ /** Renders a requested sort key as SQL, for SHOW CREATE TABLE and DESCRIBE.
*/
+ def describeSortOrder(sortOrder: SortOrder): String = {
+ s"${toSQL(sortOrder.expression())} ${sortOrder.direction()}
${sortOrder.nullOrdering()}"
+ }
+
+ // `describe` prints a literal's internal value, e.g. `0` for DATE
'1970-01-01', so literals are
+ // rendered through Catalyst to keep their type. Catalyst renders a FLOAT as
a CAST, which the
+ // parser does not accept here, so a finite FLOAT is rendered as `<v>F`
instead. PARTITIONED BY
+ // in SHOW CREATE TABLE and the `Part N` rows of DESCRIBE render transforms
with `describe`, so
+ // the same transform can print differently there.
+ private def toSQL(e: Expression): String = e match {
+ case l: Literal[_] => (l.value, l.dataType) match {
+ case (f: Float, FloatType) if java.lang.Float.isFinite(f) => s"${f}F"
+ case (value, dataType) => CatalystLiteral.create(value, dataType).sql
Review Comment:
Done.
(a) The Catalyst conversion is wrapped in `NonFatal`: DESCRIBE falls back to
`describe`, or to the raw value when `LiteralValue` itself rejects it, and SHOW
CREATE TABLE omits the key.
(b) References render as `fieldNames.map(quoteIfNeeded).mkString(".")`, so a
connector's own `NamedReference` prints like `FieldReference`. A related case
came up: a connector expression `V2ExpressionSQLBuilder` does not know, such as
`GeneralScalarExpression("DATE_TRUNC", ...)`, made `describe` recurse until
`StackOverflowError`. DESCRIBE now renders those from their children.
PARTITIONED BY is left to SPARK-59946.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:
##########
@@ -125,6 +127,72 @@ case class ShowCreateTableExec(
}
}
+ /**
+ * True for a sort key expression the `transform` grammar rule can
represent: a column
+ * reference, or a transform with at least one argument, each a column
reference or a literal
+ * the `constant` rule can spell with the same type.
+ */
+ private def isSpellable(e: V2Expression): Boolean = e match {
+ case _: NamedReference => true
+ case t: Transform =>
+ t.arguments().nonEmpty && t.arguments().forall {
+ case _: NamedReference => true
+ case l: Literal[_] => isSpellableLiteral(l)
+ case _ => false
+ }
+ case _ => false
+ }
+
+ // Lists only the types whose rendering parses back as a `constant` of the
same type. A typed
Review Comment:
The parse-back check makes the "same type" claim hold by construction.
`1.50` at DECIMAL(10,2), `1E-7` at DECIMAL(10,7), CHAR/VARCHAR literals, and
ANSI intervals under the legacy interval conf all fail the comparison and are
omitted. So do strings with a backslash under escaped string literals, dates
under the LEGACY time parser and multi-field day-time intervals under the
legacy day-time string conf, which came up while testing. The rendering and the
check now sit together in `WriteDistributionAndOrdering`, and
`WriteDistributionAndOrderingUtilsSuite` has the table-driven test you
suggested.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/CreateTableWriteOrderSuite.scala:
##########
@@ -0,0 +1,1044 @@
+/*
+ * 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 scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.analysis.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.connector.catalog.{Column, DelegatingTable,
Identifier, InMemoryTable, InMemoryTableCatalog, StagedTable,
StagingInMemoryTableCatalog, Table, TableCatalogCapability, TableInfo,
WriteDistributionMode}
+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.distributions.Distributions
+import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
Expression, FieldReference, LogicalExpressions, 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, StringType}
+
+/**
+ * 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 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 FIRST") 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_FIRST)
+ case other => fail(s"unexpected plan: $other")
+ }
+ }
+
+ test("bare ORDERED BY implies range, LOCALLY implies none, UNORDERED implies
none") {
+ Seq(
+ ("ORDERED BY id", RANGE, 1),
+ ("LOCALLY ORDERED BY id", NONE, 1),
+ ("UNORDERED", NONE, 0)
+ ).foreach { case (clause, expectedMode, expectedOrderingSize) =>
+ parse(s"CREATE TABLE t (id INT) USING foo $clause") match {
+ case c: CreateTable =>
+ assert(c.writeDistributionMode === expectedMode, s"for $clause")
+ assert(c.writeOrdering.length === expectedOrderingSize, s"for
$clause")
+ case other => fail(s"unexpected plan for $clause: $other")
+ }
+ }
+ }
+
+ test("an explicit DISTRIBUTED BY PARTITION decides the distribution on its
own") {
+ Seq(
+ ("ORDERED BY id", 1),
+ ("LOCALLY ORDERED BY id", 1),
+ ("UNORDERED", 0)
+ ).foreach { case (clause, expectedOrderingSize) =>
+ parse(s"CREATE TABLE t (id INT, c STRING) USING foo PARTITIONED BY (c) "
+
+ s"DISTRIBUTED BY PARTITION $clause") match {
+ case c: CreateTable =>
+ assert(c.writeDistributionMode === HASH, s"for $clause")
+ assert(c.writeOrdering.length === expectedOrderingSize, s"for
$clause")
+ case other => fail(s"unexpected plan for $clause: $other")
+ }
+ }
+ }
+
+ test("no clause leaves the mode unset (null), distinct from an explicit
none") {
+ parse("CREATE TABLE t (id INT) USING foo") match {
+ case c: CreateTable =>
+ assert(c.writeDistributionMode === null)
+ assert(c.writeOrdering.isEmpty)
+ case other => fail(s"unexpected plan: $other")
+ }
+ }
+
+ 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("the write clauses may appear in any position among the other
create-table clauses") {
+ Seq(
+ "ORDERED BY id PARTITIONED BY (c) DISTRIBUTED BY PARTITION",
+ "DISTRIBUTED BY PARTITION PARTITIONED BY (c) LOCALLY ORDERED BY id",
+ "PARTITIONED BY (c) ORDERED BY id COMMENT 'a table' DISTRIBUTED BY
PARTITION"
+ ).foreach { clauses =>
+ parse(s"CREATE TABLE t (id INT, c STRING) USING foo $clauses") match {
+ case c: CreateTable =>
+ assert(c.writeDistributionMode === HASH, s"for $clauses")
+ assert(c.writeOrdering.length === 1, s"for $clauses")
+ case other => fail(s"unexpected plan for $clauses: $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 e = intercept[ParseException] {
+ parse(s"CREATE TABLE t (id INT, c STRING) USING foo PARTITIONED BY (c)
$clauses")
+ }
+ assert(e.getCondition === "DUPLICATE_CLAUSES", s"for $clauses")
+ assert(e.getMessageParameters.get("clauseName") === clauseName, s"for
$clauses")
+ }
+ }
+
+ 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 =>
+ val e = intercept[ParseException](parse(stmt))
+ assert(e.getCondition ===
+
"SPECIFY_DISTRIBUTED_BY_PARTITION_WITHOUT_PARTITIONING_IS_NOT_ALLOWED", s"for
$stmt")
+ }
+ }
+
+ 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 e = intercept[AnalysisException] {
+ analyze("CREATE TABLE testcat.t (ID INT) USING foo ORDERED BY
truncate(4, id)")
+ }
+ assert(e.getCondition === "WRITE_ORDERING_WITH_UNKNOWN_COLUMN")
+ assert(e.getMessageParameters.get("cols") === "`id`")
+
+ // references are normalized independently, so only the unresolvable one
is reported
+ val multi = intercept[AnalysisException] {
+ analyze("CREATE TABLE testcat.t (ID INT) USING foo ORDERED BY
bucket(4, id, nope)")
+ }
+ assert(multi.getMessageParameters.get("cols") === "`nope`")
Review Comment:
Done. Every error check in the suite now uses `checkError` with `sqlState`,
`parameters`, and the query context where there is one. The SQLSTATEs of the
four new conditions are pinned: 42703, 42601, 42908 and 42908.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateTableLikeExec.scala:
##########
@@ -46,6 +46,10 @@ import org.apache.spark.sql.errors.QueryCompilationErrors
* ROW FORMAT / STORED AS converted to hive properties (if given), and
* [[TableCatalog.PROP_OWNER]] set to the current user. Source table
properties are intentionally
* excluded so that connectors can decide which custom properties to clone via
[[sourceTable]].
+ *
+ * The source's declared write distribution and ordering are likewise not
copied onto the
Review Comment:
Done. The `TableCatalog.createTableLike` Javadoc and its `@param tableInfo`
say the source's write distribution and ordering are not copied
(`writeDistributionMode()` is null, `writeOrdering()` is empty) and that a
connector can read them from `sourceTable`. The `TableInfo` class summary lists
them too.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala:
##########
@@ -991,6 +992,25 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
"cols" -> badReferences.map(r => toSQLId(r)).mkString(", ")))
}
+ // PreprocessTableCreation only normalizes column and
RewritableTransform references,
Review Comment:
Done as you suggested. The comment now says this is the only check for an
unknown ordering column, because `PreprocessTableCreation` keeps such a
reference for it and analyzers without that rule need it. The `NamedReference`
import is gone, and a new test pins declaration order with six unknown
references.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -6292,6 +6300,85 @@ class AstBuilder extends DataTypeAstBuilder
collation, serdeInfo, clusterBySpec)
}
+ /**
+ * Parse the optional write distribution and ordering clauses of a
CREATE/REPLACE TABLE statement:
+ * `DISTRIBUTED BY PARTITION` and `[LOCALLY] ORDERED BY ... | UNORDERED`.
+ *
+ * Returns the distribution mode and the ordering. The mode is null when the
statement did not ask
+ * for one, which leaves the choice to the catalog; `NONE` is an explicit
request.
+ * `DISTRIBUTED BY PARTITION` is rejected when `partitionTransforms` is
empty.
+ */
+ private def writeSpecsFrom(
+ ctx: CreateTableClausesContext,
+ partitionTransforms: Seq[Transform]): (WriteDistributionMode,
Seq[V2SortOrder]) = {
+ checkDuplicateClauses(ctx.writeDistributionSpec, "DISTRIBUTED BY
PARTITION", ctx)
+ checkDuplicateClauses(ctx.writeOrderingSpec, "ORDERED BY/UNORDERED", ctx)
+
+ val distributionSpec = ctx.writeDistributionSpec.asScala.headOption.orNull
+ val orderingSpec = ctx.writeOrderingSpec.asScala.headOption.orNull
+ // Bucketing counts as partitioning here; CLUSTER BY does not.
+ if (distributionSpec != null && partitionTransforms.isEmpty) {
+ throw
QueryParsingErrors.distributedByPartitionWithoutPartitioningError(ctx)
+ }
+ if (distributionSpec == null && orderingSpec == null) {
+ (null, Seq.empty)
+ } else {
+ (toDistributionMode(distributionSpec, orderingSpec),
toOrdering(orderingSpec))
+ }
+ }
+
+ /**
+ * Derive the write distribution mode from how the ordering was asked for.
An explicit
+ * `DISTRIBUTED BY PARTITION` wins, which leaves `LOCALLY` with no effect
beside it.
+ */
+ private def toDistributionMode(
+ distributionSpec: WriteDistributionSpecContext,
+ orderingSpec: WriteOrderingSpecContext): WriteDistributionMode = {
+ // Only called when at least one of the two clauses is present, so
orderingSpec is non-null
+ // here whenever distributionSpec is null.
+ if (distributionSpec != null) {
+ WriteDistributionMode.HASH
+ } else if (orderingSpec.UNORDERED != null || orderingSpec.LOCALLY != null)
{
+ WriteDistributionMode.NONE
+ } else {
+ WriteDistributionMode.RANGE
+ }
+ }
+
+ private def toOrdering(orderingSpec: WriteOrderingSpecContext):
Seq[V2SortOrder] = {
+ if (orderingSpec != null && orderingSpec.writeOrder != null) {
+ orderingSpec.writeOrder.fields.asScala.map(visitWriteOrderField).toSeq
+ } else {
+ Seq.empty
+ }
+ }
+
+ /**
+ * Create a v2 sort order for one field of a write ordering.
+ */
+ override def visitWriteOrderField(ctx: WriteOrderFieldContext): V2SortOrder
= withOrigin(ctx) {
+ val direction = if (ctx.DESC != null) {
+ SortDirection.DESCENDING
+ } else {
+ SortDirection.ASCENDING
+ }
+
+ val nullOrdering = if (ctx.FIRST != null) {
+ NullOrdering.NULLS_FIRST
+ } else if (ctx.LAST != null) {
+ NullOrdering.NULLS_LAST
+ } else {
+ direction.defaultNullOrdering
+ }
+
+ // A plain column is passed as a bare reference rather than as
`identity(col)`.
+ val key = visitTransform(ctx.transform) match {
Review Comment:
Thanks, filed SPARK-59977 for this since it seems like an existing bug.
--
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]
