dongjoon-hyun commented on code in PR #58153:
URL: https://github.com/apache/spark/pull/58153#discussion_r4164448216
##########
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:
**`TIMESTAMP` literals are printed as session-local wall time without an
offset, so they can replay as a different value or type.** `Literal.sql` prints
`TIMESTAMP '<wall time in spark.sql.session.timeZone>'`. With the default confs
and `America/Los_Angeles`, `ORDERED BY f(id, TIMESTAMP '2020-11-01
01:30:00-08:00')` stores 09:30Z, but SHOW CREATE TABLE prints `TIMESTAMP
'2020-11-01 01:30:00'`. Replaying that in the same session resolves the
ambiguous fall-back time to the earlier offset (PDT), so the key silently
becomes 08:30Z. Replaying in another session time zone shifts it by the offset
difference. With `spark.sql.timestampType=TIMESTAMP_NTZ`, a key written as
`TIMESTAMP_LTZ '2020-01-01 10:00:00'` is printed as `TIMESTAMP '...'` and
re-parsed as `TIMESTAMP_NTZ`. The round-trip test passes because it replays in
the same session, outside a DST overlap, with the default timestamp type.
Could we render `TimestampType` with an explicit offset in `toSQL`
(WriteDistributionAndOrdering.scala:72), e.g. `TIMESTAMP_LTZ '<UTC wall
time>Z'`, like the FLOAT special case? That re-parses as `TimestampType` in
both timestamp modes.
##########
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:
**`true`, `false` and `NULL` do not re-parse as constants inside a
transform, so `BooleanType` and `NullType` do not belong in this list.** In
`transformArgument : qualifiedName | constant`, these non-reserved keywords
match both alternatives, and ANTLR resolves the ambiguity to the first one,
`qualifiedName`. (`primaryExpression` lists `constant` before `identifier`,
which is why `SELECT true` is a literal.) If a connector reports RANGE with the
key `f(id, literal(true))`, SHOW CREATE TABLE prints `ORDERED BY (f(id, true)
ASC NULLS FIRST)`, and replaying it reads `true` as a column: the statement
fails with `WRITE_ORDERING_WITH_UNKNOWN_COLUMN`, or binds to a column named
`true`. `TRUE` is also in `ansiNonReserved`, so this happens in every mode.
`FALSE` and `NULL` behave the same unless
`spark.sql.ansi.enforceReservedKeywords` is true, so a table created with
`ORDERED BY f(id, NULL)` in an enforcing session does not replay in a default
one.
Could we drop `BooleanType` here and change L150 to `case (null, _) =>
false`? Deleting L150 would let a typed NULL fall through to the type list. The
comment at L146 and the `constant` in the new docs could mention this too.
##########
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:
**For a `CLUSTER BY` table, this advice cannot be followed either.** `CREATE
TABLE t (id INT, c STRING) USING foo CLUSTER BY (c) DISTRIBUTED BY PARTITION`,
pinned at CreateTableWriteOrderSuite.scala:343, gets this message. Adding
`PARTITIONED BY (id)` as suggested then fails with
`SPECIFY_CLUSTER_BY_WITH_PARTITIONED_BY_IS_NOT_ALLOWED`, and adding `CLUSTERED
BY (id) INTO 4 BUCKETS` fails with
`SPECIFY_CLUSTER_BY_WITH_BUCKETING_IS_NOT_ALLOWED`, because those checks
(AstBuilder.scala:6290-6296) run before `writeSpecsFrom`. Only the SQL
reference explains that `CLUSTER BY` has to be replaced. This is the `CLUSTER
BY` counterpart of point 8.
Could `writeSpecsFrom` raise a dedicated condition when `ctx.clusterBySpec`
is present, e.g.
`SPECIFY_CLUSTER_BY_WITH_DISTRIBUTED_BY_PARTITION_IS_NOT_ALLOWED` next to the
existing `SPECIFY_CLUSTER_BY_WITH_*` conditions, or at least add a sentence
about `CLUSTER BY` here, as for the schemaless case? Either way the rejection
stays at parse time, so HASH still always comes with non-empty `partitions()`.
##########
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:
**This extractor-based check introduced two problems in this commit.**
(a) `ClusterByTransform.unapply` casts every argument with
`arguments.map(_.asInstanceOf[NamedReference])` (expressions.scala:187), and
`hasPartitioning` is a strict `val` evaluated for every table whose ordering
passes L167, including tables that declare nothing, since an empty ordering
passes `forall`. So SHOW CREATE TABLE now throws a raw `ClassCastException` for
a table whose first partition transform is a `cluster_by` with a literal
argument, e.g. a connector-reported `Expressions.apply("cluster_by",
Expressions.literal(4))`. Master and d2aa8674fdc printed `PARTITIONED BY
(cluster_by(4))` there.
(b) The parser still counts such a transform as partitioning. `PARTITIONED
BY (cluster_by(a)) DISTRIBUTED BY PARTITION` becomes
`ApplyTransform("cluster_by", [a])`, passes the check at AstBuilder.scala:6320
and sends HASH, and a `DelegatingTable`-based catalog (like
`GenericClusterByTableCatalog` in the new suite) reports both back. SHOW CREATE
TABLE then prints `PARTITIONED BY (cluster_by(a))` but treats it as clustering
here and omits `DISTRIBUTED BY PARTITION`, so a replay silently loses HASH. The
notation is not exotic: v2 SHOW CREATE TABLE already prints every `CLUSTER BY`
table as `PARTITIONED BY (cluster_by(c))`.
Could we share one name-based predicate without a cast between
`writeSpecsFrom` and this method, e.g.
`WriteDistributionAndOrdering.hasPartitioning(p) = p.exists(_.name !=
"cluster_by")`, and evaluate it only in the HASH cases? That keeps szehon-ho's
request that a connector's generic `cluster_by` counts as clustering.
##########
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:
**`TimeType` is missing from this list, so a `TIME` literal in a sort key
drops every write clause from SHOW CREATE TABLE.** The parser accepts `ORDERED
BY f(id, TIME '12:00:00')` without any flag (`spark.sql.timeType.enabled` gates
schemas and casts, not literals), and `Literal.sql` spells it back as `TIME
'12:00:00'`, which re-parses as `TIME(6)`. But the key falls to `case _ =>
false` at L157, the `forall` at L167 fails, and `ORDERED BY` (or `LOCALLY
ORDERED BY`, or `DISTRIBUTED BY PARTITION ...`) disappears from the output.
Replaying it creates a table without the declared layout, which is the silent
drop from point 7. This is a regression in this commit: d2aa8674fdc accepted
every `Literal` and printed the clause. The nanosecond timestamp types (preview
flag) and the legacy `CalendarIntervalType` are dropped the same way.
Could we add `case (_, TimeType(TimeType.DEFAULT_PRECISION)) => true`, and
add `TIME '12:00:00'` to the round-trip test at
CreateTableWriteOrderSuite.scala:627? Precision 7-9 does not round-trip, since
`TIME '12:00:00.1000000'` prints as `TIME '12:00:00.1'` and re-parses as
`TIME(6)`.
##########
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:
Minor: the "same type" claim also fails for some connector-reported
literals. A `DECIMAL(10,2)` value 1.50 prints as `1.50BD` and comes back as
`DECIMAL(3,2)`, and 0.0000001 at `DECIMAL(10,7)` prints as `1E-7BD` and comes
back as `DECIMAL(7,7)`. An uncollated `CHAR(n)` or `VARCHAR(n)` literal passes
`isDefaultStringCharOrVarcharType` at L153 and comes back as `STRING`. Under
`spark.sql.legacy.interval.enabled=true` the ANSI intervals come back as
`CalendarIntervalType`, and under `spark.sql.parser.escapedStringLiterals=true`
every backslash doubles on replay. Also, the `Float.MaxValue` exclusion at L151
exists only because `toSQL` renders `<v>F`
(WriteDistributionAndOrdering.scala:72), and both are private methods in
different modules, so neither can be unit-tested.
Could we narrow `_: DecimalType` to `case (d: Decimal, t: DecimalType) =>
Literal(d.toJavaBigDecimal).dataType == t` (Catalyst `Literal`) and the string
case to `s eq StringType`? Moving this allow-list next to `toSQL` (which fits
SPARK-59946) would also let a table-driven test over connector-reported keys
assert "emitted iff the replayed statement declares the same keys". Comparing
through Catalyst `Literal` handles `Array[Byte]` for BINARY.
##########
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:
**`toSQL` still trusts connector-reported keys in two ways, so DESCRIBE
TABLE EXTENDED can fail entirely.**
(a) `CatalystLiteral.create` throws for some values that point 2's fix does
not cover. The public factory `Expressions.literal(new
java.math.BigDecimal("1E+3"))` creates `LiteralValue(1E+3, DecimalType(4,0))`,
since `Decimal.set` normalizes the negative scale, but `create` builds
`DecimalType(1, -3)` from the raw value and throws `NEGATIVE_SCALE_DISALLOWED`.
A value whose class does not match its type (`Integer` with `LongType`,
`ByteBuffer` with `BinaryType`) throws `IllegalArgumentException` from
`validateLiteralValue`.
`DescribeTableExec.addTableWriteDistributionAndOrdering` has no guard, and
`isSpellableLiteral` checks only the type tag, so SHOW CREATE TABLE throws
instead of omitting the pair.
(b) References are rendered with `describe()` (L75 and L78), whose default
is `toString()`, and only Spark's `FieldReference` quotes there. A connector's
own `NamedReference`, like the `MyNamedReference` classes in Spark's test
connectors, prints `...$MyNamedReference@1a2b` or an unquoted `order-id`, and
the DDL does not parse. PARTITIONED BY has the same pre-existing issue.
Could we wrap the conversion in `NonFatal`, falling back to `l.describe` for
DESCRIBE and treating the key as unspellable for SHOW CREATE TABLE, and render
references as `r.fieldNames.map(quoteIfNeeded).mkString(".")`? The latter
prints exactly what `FieldReference` prints and follows the same principle as
point 7.
##########
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:
Nit: these six tests (about 230 lines) copy assertions from "Test v2
CreateTable with default catalog" (L620) and "Test v2 CTAS with known catalog
in identifier" (L683) that are unrelated to this feature: catalog name, schema,
properties and `ignoreIfExists`. `parseAndResolve` (L291) runs neither
`PreprocessTableCreation` nor `checkAnalysis` by default, so these tests only
re-pin parser output that the parse tests at
CreateTableWriteOrderSuite.scala:80-190 already pin, and they can break for
reasons unrelated to the write clauses.
Could we drop them, or fold them into one table-driven test over `(sql,
mode, ordering)` that asserts only the write spec?
##########
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:
Minor: this caveat is only in the exec's scaladoc, while connector authors
read the public `TableCatalog.createTableLike` Javadoc. That Javadoc still says
`tableInfo` contains "columns and partitioning copied from the source, any
constraints copied from the source, ..." (TableCatalog.java:368-371) and calls
it the "complete description of the new table: columns, partitioning,
constraints, ..." (TableCatalog.java:379-381). A connector written against that
contract expects `tableInfo.writeOrdering()` to carry the source's layout like
the constraints, but it is always empty, so `CREATE TABLE ... LIKE` silently
creates the target without it unless the connector reads `sourceTable`. The
`TableInfo` class Javadoc (TableInfo.java:28) also lists only "columns,
properties, partitioning and constraints".
Could we add the same sentence to those Javadocs? That would complete
peter-toth's finding 9.
##########
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:
Minor, pre-existing but now reachable from ORDERED BY: `constant` in
`transformArgument` also matches a parameter marker, but
`visitTransformArgument` (L5913-5915) assumes a `Literal` and calls
`lit.value`. Without args, `spark.sql("CREATE TABLE cat.t (id INT) USING foo
ORDERED BY bucket(?, id)")` skips parameter substitution,
`visitPosParameterLiteral` returns a `PosParameter`, and parsing fails with a
raw `ClassCastException` (`PosParameter cannot be cast to Literal`) instead of
`UNBOUND_SQL_PARAMETER`. `PARTITIONED BY (bucket(?, id))` already fails the
same way, so I'm fine with a follow-up next to the `bucket(4294967300, id)` and
`IDENTIFIER('id')` items from my previous comment.
Could `visitTransformArgument` raise a parse error when the visited constant
is not a `Literal`?
##########
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:
Nit: the Javadoc says this must not be null and that `writeOrdering()` is
never null, but neither this setter nor `build()` (L156) enforces it. A
connector that passes null and returns `new DelegatingTable(info, name)` gets
an NPE in DESCRIBE TABLE EXTENDED (DescribeTableExec.scala:232) or SHOW CREATE
TABLE (ShowCreateTableExec.scala:167) instead of at `build()`. `withPartitions`
and `withConstraints` are unchecked too, but only this field documents "never
null".
Could `build()` add `Objects.requireNonNull(writeOrdering, ...)` next to the
`columns` check?
##########
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:
Minor: `isSpellable` checks only the argument kinds, but `visitTransform`
dispatches the exact names `bucket`, `years`, `months`, `days` and `hours`
(AstBuilder.scala:5869-5899) and rejects any other argument shape for them. The
parser's `PostProcessor` strips backticks before that, so quoting cannot escape
the dispatch. A connector key such as `Expressions.apply("bucket",
column("id"), literal(16))` (Trino-style argument order) prints `ORDERED BY
(bucket(id, 16) ASC NULLS FIRST)`, and replaying it fails for the whole
statement with `_LEGACY_ERROR_TEMP_0031`; `apply("days", ts, tz)` fails with
`TRANSFORM_WRONG_NUM_ARGS`. This is the unrunnable-DDL case from peter-toth's
finding 2.
Could we require the parser's shape for these five names here (a Short, Int
or Long literal followed by references for `bucket`, a single reference for the
others), and add such keys to `ORDERING_OVERRIDE`?
##########
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:
Minor: this assertion checks only `cols`, not the condition. About 13 error
checks in this suite are hand-rolled (`getCondition`,
`getMessageParameters.get`), while L516, L530, L558 and L570 already use
`checkError`. Some check only the condition (L214, L286, L307, L347, L550), and
none pins the SQLSTATE or the query context. No test pins the SQLSTATEs of the
three new conditions (`WRITE_ORDERING_WITH_UNKNOWN_COLUMN` with 42703,
`SPECIFY_WRITE_ORDERING_IS_NOT_ALLOWED`,
`SPECIFY_DISTRIBUTED_BY_PARTITION_WITHOUT_PARTITIONING_IS_NOT_ALLOWED`), and
`SparkThrowableSuite` does not pin per-condition values, so a change to the
JSON alone would pass. We settled 42703 in my first comment because error
classes are hard to change after a release.
Could we convert these to `checkError(..., sqlState = ..., parameters = ...,
context = ExpectedContext(...))`? The parse errors and the `failAnalysis` error
carry a query context, and `checkError` verifies the SQLSTATE only when
`sqlState` is passed.
##########
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:
Minor: Spark's own `DelegatingCatalogExtension` cannot keep this promise. It
forwards the delegate's `capabilities()`
(DelegatingCatalogExtension.java:57-58) but does not override
`createTable(Identifier, TableInfo)`, so it uses the default at
TableCatalog.java:359-361, which drops the request. Today the only in-tree
delegate, `V2SessionCatalog`, does not report the capability, so the statement
is rejected. Once a delegate reports it, e.g. if `V2SessionCatalog` gains
support later, subclasses that inherit `super.capabilities()`, such as Delta's
`AbstractDeltaCatalog` and Hudi's `HoodieCatalog`, pass the check and silently
lose the layout. That is the case the PR description says the capability
prevents, and `SUPPORT_TABLE_CONSTRAINT` has had the same gap since 4.1.0.
Forwarding `createTable(Identifier, TableInfo)` to the delegate is not a fix,
because it would bypass those subclasses' column-based overrides.
Could `DelegatingCatalogExtension.capabilities()` remove this new capability
from the forwarded set, so that only a subclass overriding all four `TableInfo`
overloads adds it back, or could this Javadoc at least mention it?
##########
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:
Nit: this comment gives a reason that no longer holds.
`PreprocessTableCreation` deliberately keeps an unresolvable ordering reference
for this check (rules.scala:354-355), so this is the only check rather than an
additional one, and analyzers without that sql/core rule need it as well. The
block also copies the partitioning check above with an explicit `(ref:
NamedReference) =>` ascription (the import at L33 exists only for it), a
trailing `.toSeq`, and a quote-then-reparse through `column.quoted` and
`toSQLId(String)`. With five or more references, `cols` follows the hash set
order rather than the declaration order.
Could we write it as
`create.writeOrdering.flatMap(_.expression().references().map(_.fieldNames().toImmutableArraySeq)).distinct.filter(create.tableSchema.findNestedField(_).isEmpty)`
with `toSQLId(parts: Seq[String])`, drop the import, and say why the check is
here?
--
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]
