Repository navigation
[SPARK-34586][SQL] Support declaring a write distribution and ordering in CREATE/REPLACE TABLE - #58153
Conversation
|
@peter-toth @szehon-ho could you please review this? I would especially like feedback on the the design design decisions mentioned in the PR description. Thanks. |
peter-toth
left a comment
peter-toth
left a comment
There was a problem hiding this comment.
Thanks for picking this up, @anuragmantri!
The shape reads right to me. The request rides on TableInfo, Table reports it back, and the capability check rejects the statement while planning, so a catalog that ignores TableInfo cannot hand back a table quietly missing the layout. On the design decisions you asked about: I'm a co-author, so my read of those isn't independent - all four still look right to me, and the value is in the findings below. Two things block. CI is red on a keyword-list test this misses, and SHOW CREATE TABLE can still emit DDL that does not parse - the ordering half of the hazard the hash-without-partitioning guard already covers. Findings 2, 3 and 5 are measured on this branch, not read off the diff.
Blocking
- 1. Connect JDBC keyword list not updated:
SparkConnectDatabaseMetaDataSuite."getSQLKeywords"asserts its own hardcoded keyword list and is the only failing test on this head. The Thrift-server sibling was updated; this one needs the same four keywords. [inline:sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala:217] - 2.
SHOW CREATE TABLEcan emit DDL that does not parse: the guard covers the distribution mode but never checks that the sort expressions are spellable, andSortOrder.expression()is typedExpression. A table reportingsort(a + 1, ASC, NULLS FIRST)yieldsORDERED BY (id + 1 ASC NULLS FIRST), which fails withPARSE_SYNTAX_ERRORon replay. [inline:sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:146] - 3. New error condition is named for cases it cannot report: nested struct columns are supported (the suite asserts
ORDERED BY p.xsucceeds), and the "or is in a map or array" clause is unreachable - those paths raiseINVALID_FIELD_NAME. A released condition name can't be renamed. [inline:common/utils/src/main/resources/error/error-conditions.json:9155]
Non-blocking
- 4. Docs credit the "data source" with a catalog capability: the gate is
TableCatalog.capabilities(), so a reader following this paragraph will inspect theUSINGprovider and find nothing there. [inline:docs/sql-ref-syntax-ddl-create-table-datasource.md:153] - 5. Docs promise
SHOW CREATE TABLEreproduces the clauses, with no omission caveat: the caveat is in the PR body and in the scaladoc, but not on the page users read, and the omitted case is the likely one. [inline:docs/sql-ref-syntax-ddl-create-table-datasource.md:160]
Minor
- 6. "written file" vs "write task" three lines apart: a task can write several files, so the two are different claims; the second is the accurate one. [inline:
docs/sql-ref-syntax-ddl-create-table-datasource.md:133]
| val infoValue = client.getInfo(sessionHandle, GetInfoType.CLI_ODBC_KEYWORDS) | ||
| // scalastyle:off line.size.limit | ||
| assert(infoValue.getStringValue == "ADD,AFTER,AGGREGATE,ALIGN,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPLY,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,ASOF,AT,ATOMIC,AUTHORIZATION,AUTO,BEGIN,BERNOULLI,BETWEEN,BIGINT,BIN,BINARY,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,EMPTY,END,ENFORCED,ERROR,ESCAPE,ESCAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HISTORY,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,JSON_EXISTS,JSON_TABLE,JSON_VALUE,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCALTIME,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATCH_CONDITION,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OF,OFFSET,ON,ONLY,OPEN,OPTION,OPTIONS,OR,ORDER,ORDINALITY,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNING,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCD,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRACK,TRAILING,TRANSACTION,TRANSACTIONS,TRANSFORM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNIFORM,UNION,UNIQUE,UNKNOWN,UNLOCK,UNNEST,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WIDTH,WINDOW,WITH,WITHIN,WITHOUT,X,YEAR,YEARS,ZONE") | ||
| assert(infoValue.getStringValue == "ADD,AFTER,AGGREGATE,ALIGN,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPLY,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,ASOF,AT,ATOMIC,AUTHORIZATION,AUTO,BEGIN,BERNOULLI,BETWEEN,BIGINT,BIN,BINARY,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DISTRIBUTED,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,EMPTY,END,ENFORCED,ERROR,ESCAPE,ESCAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HISTORY,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,JSON_EXISTS,JSON_TABLE,JSON_VALUE,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCALLY,LOCALTIME,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATCH_CONDITION,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OF,OFFSET,ON,ONLY,OPEN,OPTION,OPTIONS,OR,ORDER,ORDERED,ORDINALITY,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNING,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCD,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRACK,TRAILING,TRANSACTION,TRANSACTIONS,TRANSFORM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNIFORM,UNION,UNIQUE,UNKNOWN,UNLOCK,UNNEST,UNORDERED,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WIDTH,WINDOW,WITH,WITHIN,WITHOUT,X,YEAR,YEARS,ZONE") |
There was a problem hiding this comment.
Finding 1. This list got the four new keywords, but its Spark Connect JDBC sibling did not, and CI is red on it.
SparkConnectDatabaseMetaDataSuite."SparkConnectDatabaseMetaData getSQLKeywords" asserts its own hardcoded list at sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala:213. It is the only failing test on c6adea5c405 - one annotation on the "Report test results" check run, and the failing "Build modules: ... connect ..." job is the same test.
getSQLKeywords drops SQL:2003 reserved words, and all four new keywords are non-reserved, so all four need adding:
...,DISTRIBUTE,DISTRIBUTED,DIV,......,LOAD,LOCALLY,LOCATION,......,OPTIONS,ORDERED,ORDINALITY,......,UNLOCK,UNORDERED,UNPIVOT,...
| val orderBy = if (table.writeOrdering().nonEmpty) { | ||
| Some(table.writeOrdering() | ||
| .map(WriteDistributionAndOrdering.describeSortOrder) | ||
| .mkString("ORDERED BY (", ", ", ")")) |
There was a problem hiding this comment.
Finding 2. The hasPartitioning guard below covers the distribution side of this method's own doc - "emitting a clause that means something else -- or one that does not parse at all -- would be worse than emitting none" - but nothing checks that the sort expressions are spellable. SortOrder.expression() is typed Expression and Table.writeOrdering()'s new contract does not narrow it, so a connector may report an expression the writeOrderField : transform ... rule cannot represent.
Measured on this branch, with a table reporting mode range and sort(a + 1, ASCENDING, NULLS_FIRST):
CREATE TABLE p.t (
id INT)
USING foo
ORDERED BY (id + 1 ASC NULLS FIRST)
Replaying that gives [PARSE_SYNTAX_ERROR] Syntax error at or near '+' at line 4, pos 15. The whole statement is unrunnable, not one clause lost, which is the worse failure the doc calls out.
Same fix shape as the hash case: drop the pair when it cannot be spelled. Two things to get right. CreateTableWriteOrderSuite builds a bare FieldReference, not a Transform, and that one does round-trip, so the predicate has to admit it. And nulling out orderBy alone is not enough - (none, None) would then emit UNORDERED, declaring no ordering on a table that has one - so the whole method has to bail:
private def isSpellable(e: V2Expression): Boolean = e match {
case _: NamedReference => true
// Mirrors `transformArgument : qualifiedName | constant`.
case t: Transform =>
t.arguments().forall(a => a.isInstanceOf[NamedReference] || a.isInstanceOf[Literal[_]])
case _ => false
}then wrap the existing body in if (table.writeOrdering().forall(o => isSpellable(o.expression()))) { ... }. DESCRIBE TABLE EXTENDED still reports both values verbatim, so nothing is hidden.
There was a problem hiding this comment.
Done. I added isSpellable() and wrapped the rest with it.
| "Write for the binary file data source." | ||
| ] | ||
| }, | ||
| "WRITE_ORDERING_WITH_NESTED_COLUMN_IS_UNSUPPORTED" : { |
There was a problem hiding this comment.
Finding 3. Both the name and the message describe cases this condition cannot report.
A nested struct column is supported, and the suite asserts it - CREATE TABLE testcat.t (p STRUCT<x: INT>) USING foo ORDERED BY p.x is accepted. So WITH_NESTED_COLUMN_IS_UNSUPPORTED says the opposite of the tested behaviour.
"or is in a map or array" is unreachable. StructType.findNestedField is called with the default includeCollections = false, which throws INVALID_FIELD_NAME on a path through a non-struct rather than returning None. Measured on this branch:
ORDERED BY (m.key) on MAP<STRING, INT>
-> [INVALID_FIELD_NAME] Field name `m`.`key` is invalid: `m` is not a struct
ORDERED BY (a.element.x) on ARRAY<STRUCT<x: INT>>
-> [INVALID_FIELD_NAME] Field name `a`.`element`.`x` is invalid: `a` is not a struct
ORDERED BY (truncate(4, m.key)) on MAP<STRING, INT>
-> [INVALID_FIELD_NAME] Field name `m`.`key` is invalid: `m` is not a struct
The third one is raised from the CheckAnalysis check itself: truncate(...) is an ApplyTransform, so it is not rewritable and never reaches PreprocessTableCreation.
What this condition actually reports is a reference that is not a column of the table. I know it is a faithful copy of PARTITION_WITH_NESTED_COLUMN_IS_UNSUPPORTED, where both faults are equally present, and consistency with the sibling is a real argument - it just does not carry to a name being introduced now. The sibling can be left alone; this one cannot be renamed after a release. Suggest UNSUPPORTED_FEATURE.WRITE_ORDERING_WITH_UNKNOWN_COLUMN with "Invalid write ordering: <cols> is not a column of the table.", or a non-UNSUPPORTED_FEATURE parent, since a missing column is not a feature gap.
There was a problem hiding this comment.
Changed it to WRITE_ORDERING_WITH_UNKNOWN_COLUMN.
| DISTRIBUTED BY PARTITION UNORDERED; | ||
| ``` | ||
|
|
||
| Both clauses are passed to the data source, which has to support them: a data source that does |
There was a problem hiding this comment.
Finding 4. The advertiser is the catalog, not the data source. What gates this is TableCatalog.capabilities() returning TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING, checked in WriteDistributionAndOrdering.validateCatalogForWriteDistributionAndOrdering. A user who hits UNSUPPORTED_FEATURE.TABLE_OPERATION and follows this paragraph will inspect the USING provider, where there is nothing to inspect.
It also makes the last sentence hard to act on. "The built-in data sources do not support them" is true, but the reason is that V2SessionCatalog does not advertise the capability and the v1 conversion in ResolveSessionCatalog rejects the request outright - nothing to do with parquet or ORC.
Suggest saying catalog throughout: "Both clauses are passed to the catalog, which has to support them: a catalog that does not advertise support ... The built-in catalogs do not."
|
|
||
| What the data source records is a *default* for later writes, not a statement about the data | ||
| already in the table: an individual write may override it, and rewriting existing data to match | ||
| a newly requested layout is a separate operation. `SHOW CREATE TABLE` reproduces the clauses and |
There was a problem hiding this comment.
Finding 5. SHOW CREATE TABLE reproduces the clauses only for pairs the syntax can spell, and emits nothing at all for the rest. That caveat is in the PR description and in ShowCreateTableExec's scaladoc, but not here, and this page is what users read.
It is not a corner case. A connector that records a sort order without touching its distribution mode reports (null, non-empty), which has no clause form. Measured on this branch, with a table reporting mode null and ordering id DESC NULLS LAST:
CREATE TABLE n.t (
id INT)
USING foo
while DESCRIBE TABLE EXTENDED on the same table shows Ordering = id DESC NULLS LAST. So that DDL runs and creates a table without the ordering.
One sentence covers it: SHOW CREATE TABLE reproduces the clauses when the recorded pair has a clause form, and DESCRIBE TABLE EXTENDED reports both values in every case.
| The distribution decides how far the order reaches, and this clause picks one when | ||
| `DISTRIBUTED BY PARTITION` is absent: a bare `ORDERED BY` range-partitions each write, so the | ||
| order holds across the whole table, while `LOCALLY ORDERED BY` asks for it to hold within each | ||
| written file only, without a shuffle. `UNORDERED` on its own asks for no distribution either. |
There was a problem hiding this comment.
Finding 6. "within each written file only" here, "within each write task" three lines down at :136. A task can roll over several files, so these are different claims, and the second is the accurate one - it also matches TableInfo.DISTRIBUTION_MODE_NONE's javadoc ("any ordering holds within a write task only"). DISTRIBUTION_MODE_RANGE's javadoc has the same drift the other way ("across files, not only within one").
c6adea5 to
92cc539
Compare
peter-toth
left a comment
peter-toth
left a comment
There was a problem hiding this comment.
Re-checked through 92cc5395424 - findings 1, 3, 5 and 6 resolved (Connect JDBC keyword list, the condition rename, the omission caveat, "write task"). Finding 2's fix narrows the hole rather than closing it, and finding 4 has one occurrence left; both measured on this head.
On CREATE TABLE ... LIKE: leave it out. CreateTableLikeExec hands sourceTable to TableCatalog.createTableLike alongside the TableInfo, so a connector already has what it needs to carry the layout across, and that exec's own doc names Iceberg sort order as the example. Adding the clauses to LIKE syntax is separate work. Finding 9 is only about saying so in the doc.
Blocking
- 2.
SHOW CREATE TABLEcan still emit DDL that does not parse (round 1):isSpellablevalidates aTransform's arguments but never the transform itself.Expressions.apply("+", column("id"), literal(1))is public API and yieldsORDERED BY (+(id, 1) ASC NULLS FIRST), which fails replay withPARSE_SYNTAX_ERROR. [inline:sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:137] - 7. The new guard has no test (new): replacing the
ifatShowCreateTableExec.scala:154withif (true)leaves all 30CreateTableWriteOrderSuitetests green, so nothing pins finding 2's fix - which is why its remaining hole went unnoticed. [inline:sql/core/src/test/scala/org/apache/spark/sql/connector/CreateTableWriteOrderSuite.scala:757]
Non-blocking
- 4. One "data source" left where the gate is the catalog (round 1):
:125still says omitting the clause "leaves the choice to the data source". The other three are fixed;:117is aboutCLUSTER BYand is right as it stands. Following up on the existing thread rather than opening a new one. - 8. Docs claim a range ordering "holds across the whole table" (late catch): it holds across one write's tasks, and a later
INSERT INTOoverlaps it. This now contradictsDISTRIBUTION_MODE_RANGE's javadoc, which this round narrowed to "across write tasks". [inline:docs/sql-ref-syntax-ddl-create-table-datasource.md:131] - 9.
CreateTableLikeExec's doc enumerates what theTableInfocarries and now omits two fields (new): not copying them is right, but a connector author comparing againstconstraintswill expect otherwise. [inline:sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableInfo.java:159]
| private def isSpellable(e: V2Expression): Boolean = e match { | ||
| case _: NamedReference => true | ||
| case t: Transform => | ||
| t.arguments().forall(a => a.isInstanceOf[NamedReference] || a.isInstanceOf[Literal[_]]) |
There was a problem hiding this comment.
Finding 2. This closes the case on the parent thread (GeneralScalarExpression is neither a NamedReference nor a Transform), but it validates a Transform's arguments and never the transform itself. Two ways through:
- The name.
Expressions.apply(String name, Expression... args)is public and takes any name, andApplyTransform.describe()renders it unquoted.applyTransform : transformName=identifier LEFT_PAREN ...needs anidentifierthere. - No arguments.
forallon an emptyarguments()istrue, and the same rule requires at least onetransformArgument.
Measured on 92cc5395424, with a fixture reporting mode range and Expressions.sort(Expressions.apply("+", Expressions.column("id"), Expressions.literal(1)), ASCENDING, NULLS_FIRST):
CREATE TABLE reportcat.t (
id INT)
USING foo
ORDERED BY (+(id, 1) ASC NULLS FIRST)
Replaying that gives [PARSE_SYNTAX_ERROR] Syntax error at or near '+'. SQLSTATE: 42601 (line 4, pos 12) - the same whole-statement failure as before, through a narrower door.
| t.arguments().forall(a => a.isInstanceOf[NamedReference] || a.isInstanceOf[Literal[_]]) | |
| t.name().matches("[a-zA-Z_][a-zA-Z0-9_]*") && t.arguments().nonEmpty && | |
| t.arguments().forall(a => a.isInstanceOf[NamedReference] || a.isInstanceOf[Literal[_]]) |
The regex approximates identifier rather than matching it: a name that is a reserved word (select) passes and still fails to parse under ANSI. That hole is far smaller than the current one and I would not chase it, but it is worth a word in the comment so the next reader knows this is an approximation rather than a claim.
There was a problem hiding this comment.
Took your suggestion and also added a comment in the code about the approximation.
| .getOrElse(tableInfo.writeDistributionMode()) | ||
| } | ||
|
|
||
| override def writeOrdering(): Array[SortOrder] = tableInfo.writeOrdering() |
There was a problem hiding this comment.
Finding 7. Nothing pins the guard finding 2 asked for. Measured in a review worktree on this head: replace the if at ShowCreateTableExec.scala:154 with if (true) and sql/testOnly *CreateTableWriteOrderSuite still reports Tests: succeeded 30, failed 0.
This line is why. Every ordering the suite can build arrives through tableInfo.writeOrdering(), so it comes from the parser, and every parser-produced sort key is a NamedReference or a Transform over references and literals - isSpellable is true in all 30 tests. MODE_OVERRIDE gave the distribution side a way to fabricate a value no statement could ask for; the ordering side has no equivalent, which is also why finding 2's remaining hole went unnoticed.
Same shape as MODE_OVERRIDE:
override def writeOrdering(): Array[SortOrder] = {
if (tableInfo.properties().containsKey(ReportingInMemoryTable.ORDERING_OVERRIDE)) {
// A sort key `writeOrderField : transform ...` cannot represent.
Array(Expressions.sort(
Expressions.apply("+", Expressions.column("id"), Expressions.literal(1)),
SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST))
} else {
tableInfo.writeOrdering()
}
}Two assertions are worth having, one per trap named on the finding 2 thread:
- mode
range: noORDERED BYin the output, and the emitted DDL still runs; - mode
none: noUNORDEREDeither. That is the case where bailing out of theorderByhalf alone would declare "no ordering" on a table that has one, and it is the half a test written only againstrangewould miss.
| clause. | ||
|
|
||
| The distribution decides how far the order reaches, and this clause picks one when | ||
| `DISTRIBUTED BY PARTITION` is absent: a bare `ORDERED BY` range-partitions each write, so the |
There was a problem hiding this comment.
Finding 8. A range distribution orders one write's output across that write's tasks. It says nothing about the table over time: a later INSERT INTO range-partitions its own rows, so its files overlap the ranges already there and the table is not sorted end to end.
This round narrowed TableInfo.DISTRIBUTION_MODE_RANGE's javadoc from "across files, not only within one" to "across write tasks, not only within one", so the two now disagree and the javadoc is the accurate one. The same claim is in the SQL comment at :140.
The paragraph is about what the clause requests for every write ("recorded on the table so that later writes honor it too"), which is exactly the scope where the stronger claim fails. "so the order holds across the tasks of a write, not only within one" matches the javadoc and is what the mechanism delivers.
I raised the file-vs-task drift on this bullet last round and missed this one, which is the bigger of the two: it promises a table-level layout guarantee.
There was a problem hiding this comment.
I changed the docs, let me know if that is what you meant.
| * | ||
| * @since 4.4.0 | ||
| */ | ||
| public Builder withWriteOrdering(SortOrder[] writeOrdering) { |
There was a problem hiding this comment.
Finding 9. TableInfo now carries two more fields that CREATE TABLE ... LIKE could copy from its source, and CreateTableLikeExec does not. That is the right call - it passes sourceTable to TableCatalog.createTableLike, so the connector has everything, and its own doc already names Iceberg sort order as the example - but the exec's class doc enumerates what the TableInfo carries:
columns and partitioning copied from the source, constraints copied from
the source, user-specified TBLPROPERTIES / LOCATION / USING provider ...
so the omission now reads as an oversight (sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateTableLikeExec.scala:43-48). A connector author comparing against constraints, which is copied, will expect the same treatment. One sentence there - the declared write distribution and ordering are deliberately not copied, read them from sourceTable - saves the next reader deriving it from this PR.
There was a problem hiding this comment.
Added it in the docs
92cc539 to
19365b5
Compare
anuragmantri
left a comment
anuragmantri
left a comment
There was a problem hiding this comment.
Sorry for my late response. I have updated the PR with your comments. Ready for another round. @peter-toth.
| private def isSpellable(e: V2Expression): Boolean = e match { | ||
| case _: NamedReference => true | ||
| case t: Transform => | ||
| t.arguments().forall(a => a.isInstanceOf[NamedReference] || a.isInstanceOf[Literal[_]]) |
There was a problem hiding this comment.
Took your suggestion and also added a comment in the code about the approximation.
| * | ||
| * @since 4.4.0 | ||
| */ | ||
| public Builder withWriteOrdering(SortOrder[] writeOrdering) { |
There was a problem hiding this comment.
Added it in the docs
| .getOrElse(tableInfo.writeDistributionMode()) | ||
| } | ||
|
|
||
| override def writeOrdering(): Array[SortOrder] = tableInfo.writeOrdering() |
| clause. | ||
|
|
||
| The distribution decides how far the order reaches, and this clause picks one when | ||
| `DISTRIBUTED BY PARTITION` is absent: a bare `ORDERED BY` range-partitions each write, so the |
There was a problem hiding this comment.
I changed the docs, let me know if that is what you meant.
peter-toth
left a comment
peter-toth
left a comment
There was a problem hiding this comment.
Re-checked through 19365b5946a — findings 2, 4, 7, 8 and 9 resolved, nothing new.
Finding 7 re-measured rather than read: forcing the gate at ShowCreateTableExec.scala:157 to true now fails SHOW CREATE TABLE omits a pair the syntax cannot spell, and stays runnable on ORDERED BY (+(id, 1) ASC NULLS FIRST), where the same ablation left 30/30 green last round. That also pins finding 2's fix.
Thanks for working through all of these, @anuragmantri — nothing left open from my side.
|
@aokolnychyi, @szehon-ho, can you please take a look when you have some time? |
|
Any comments or suggestions @aokolnychyi, @szehon-ho? I would like to merge this PR this week if there isn't any. |
szehon-ho
left a comment
szehon-ho
left a comment
There was a problem hiding this comment.
How should the new write-layout clauses interact with existing table CLUSTER BY support? For example, CLUSTER BY (a) ORDERED BY (b) and CLUSTER BY (a) UNORDERED are accepted and send both declarations to the connector. Is the connector responsible for reconciling them or rejecting incompatible combinations? Could we document that contract and add coverage for these combinations?
Anton (@aokolnychyi) may have more thoughts on this than me, given his work on the DSv2 write distribution and ordering APIs.
| * would read back as a transform *named* `identity` rather than as a plain column reference. | ||
| */ | ||
| def describeSortOrder(sortOrder: SortOrder): String = { | ||
| s"${sortOrder.expression().describe()} ${sortOrder.direction()} ${sortOrder.nullOrdering()}" |
There was a problem hiding this comment.
[P2] describe() does not preserve literal types. For example, the parser stores DATE '1970-01-01' as LiteralValue(0, DateType), whose description is 0. Consequently, an ordering such as f(id, DATE '1970-01-01') passes isSpellable but becomes f(id, 0) in SHOW CREATE TABLE. Replaying that DDL changes the argument to IntegerType, potentially changing or invalidating the ordering. Could we render type-preserving SQL literals and add a round-trip test comparing the reconstructed ordering expressions?
There was a problem hiding this comment.
Good catch, thanks. describeSortOrder now renders literal arguments through Catalyst's Literal.sql, so their types survive, e.g. DATE '1970-01-01' instead of 0. I added a test that replays the SHOW CREATE TABLE output, and checks that the reconstructed ordering equals the original, literal data types included.
|
Thank you for working on this, @anuragmantri. I have a few additional comments.
|
|
Gentle ping, @anuragmantri . |
|
Gentle ping, @anuragmantri . Please resolve the conflicts and check my review comments. |
19365b5 to
d2aa867
Compare
anuragmantri
left a comment
anuragmantri
left a comment
There was a problem hiding this comment.
Thanks for the reviews @szehon-ho and @dongjoon-hyun.
I rebased onto the latest master to resolve the conflicts and addressed your points.
@dongjoon-hyun for your comments:
WRITE_ORDERING_WITH_UNKNOWN_COLUMNis now a top-level condition withSQLSTATE 42703instead of a subclass ofUNSUPPORTED_FEATURE.- Yes, I have verified this finding. For this PR, I added a test for case-insensitive session does not normalize an ApplyTransform's references to
CreateTableWriteOrderSuite. It showsORDERED BY IDis normalized against a column id whileORDERED BY truncate(4, ID)is rejected, and thatPARTITIONED BYbehaves the same way. - Added a note to the PR description that extensions pattern-matching on the create/replace plans and exec nodes need to add the two new fields.
- Trimmed the comments to describe only the current behavior, including the ones you pointed out in
rules.scalaandWriteDistributionAndOrdering.
How should the new write-layout clauses interact with existing table CLUSTER BY support? For example, CLUSTER BY (a) ORDERED BY (b) and CLUSTER BY (a) UNORDERED are accepted and send both declarations to the connector. Is the connector responsible for reconciling them or rejecting incompatible combinations? Could we document that contract and add coverage for these combinations?
Spark treats them as independent declarations. CLUSTER BY records clustering columns for the catalog to interpret, and the write clauses record a declared distribution and ordering. Spark passes both to the catalog without reconciling them, the same way it passes the write layout through without enforcing it. So yes, the connector is responsible for interpreting the combination, or rejecting it if it can't support it. I documented this contract in sql-ref-syntax-ddl-create-table-datasource.md and added a test showing that CLUSTER BY (a) ORDERED BY (b) and CLUSTER BY (a) UNORDERED both reach the catalog with both declarations intact.
I'm happy to adjust if you think Spark should validate any of these combinations.
| * would read back as a transform *named* `identity` rather than as a plain column reference. | ||
| */ | ||
| def describeSortOrder(sortOrder: SortOrder): String = { | ||
| s"${sortOrder.expression().describe()} ${sortOrder.direction()} ${sortOrder.nullOrdering()}" |
There was a problem hiding this comment.
Good catch, thanks. describeSortOrder now renders literal arguments through Catalyst's Literal.sql, so their types survive, e.g. DATE '1970-01-01' instead of 0. I added a test that replays the SHOW CREATE TABLE output, and checks that the reconstructed ordering equals the original, literal data types included.
|
Thank you for addressing the previous comments, @anuragmantri. I confirmed that my four points are resolved in
Minor:
The current CI failures look unrelated: |
szehon-ho
left a comment
szehon-ho
left a comment
There was a problem hiding this comment.
Thanks for clarifying the CLUSTER BY interaction. I'm fine keeping CLUSTER BY with ORDERED BY, LOCALLY ORDERED BY, or UNORDERED legal and leaving compatibility to the catalog. The existing rejection of CLUSTER BY with DISTRIBUTED BY PARTITION makes sense. The inline comments cover the rejection-path tests, a connector-transform case in the partitioning guard, and a wording correction.
| None | ||
| } | ||
| // Bucketing counts as partitioning here; CLUSTER BY does not. | ||
| val hasPartitioning = table.partitioning.exists(!_.isInstanceOf[ClusterByTransform]) |
There was a problem hiding this comment.
[P2] Could we use the ClusterByTransform(_) extractor here instead of an implementation-class check? A connector can return Expressions.apply("cluster_by", Expressions.column("a")) (or its own Transform implementation). The existing extractor recognizes that as clustering, but isInstanceOf[ClusterByTransform] is false, so this guard treats it as actual partitioning. If the table reports declared mode hash, SHOW CREATE TABLE then emits DISTRIBUTED BY PARTITION for a clustering-only table, although this method intends to omit that pair.
A focused regression test could stub Table.partitioning() with this generic transform and report mode hash, with both empty and non-empty write ordering. Assert that neither DISTRIBUTED BY PARTITION nor a standalone ORDERED BY is emitted for that unrepresentable pair. Keep an actual partition or bucket transform as a positive control. TransformExtractorSuite already covers recognition of a connector-defined cluster_by transform.
There was a problem hiding this comment.
Good catch. The guard now uses the ClusterByTransform(_) extractor, so a connector's generic cluster_by transform counts as clustering, not partitioning. I added "SHOW CREATE TABLE treats a connector's cluster_by transform as clustering". It uses a test catalog whose tables report CLUSTER BY as Expressions.apply("cluster_by", ...). It returns a DelegatingTable, because InMemoryTable rejects a generic cluster_by at creation. With mode hash and both an empty ordering and ORDERED BY (b), the test asserts that neither DISTRIBUTED BY PARTITION nor ORDERED BY is emitted, for both the native and the generic transform. PARTITIONED BY (a) and CLUSTERED BY (a) INTO 4 BUCKETS are the positive controls and still emit DISTRIBUTED BY PARTITION ORDERED BY (b ASC NULLS FIRST). I checked that the generic case fails with the old isInstanceOf check.
| } | ||
| } | ||
|
|
||
| test("CLUSTER BY and the write clauses both reach the catalog") { |
There was a problem hiding this comment.
Could we complement this successful pass-through test with a catalog that advertises SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING but deliberately rejects one combination, for example CLUSTER BY (a) UNORDERED? This would exercise the connector-rejection contract separately from the existing tests where the catalog lacks the entire capability.
For CREATE, assert that the catalog's error reaches the caller and no table is published. For staged REPLACE/RTAS, start with a populated table and assert that its rows, schema, and clustering metadata survive the rejection. The unchanged-table assertion should be scoped to staging: non-staging ReplaceTableExec drops the original before calling createTable, so late catalog rejection there retains the existing non-atomic replacement limitation.
It would also be useful to include CLUSTER BY (a) LOCALLY ORDERED BY (b) in this positive test, asserting mode none with the ordering retained.
There was a problem hiding this comment.
Added "a catalog with the capability can still reject a combination it does not support". It uses a staging catalog that advertises the capability but rejects CLUSTER BY with UNORDERED in createTable and in all three stage* methods. For CREATE TABLE (the createTable path) and CTAS (the stageCreate path), the catalog's error reaches the caller and no table is published. The test then populates a CLUSTER BY (a) table and runs REPLACE TABLE, CREATE OR REPLACE TABLE, RTAS and CREATE OR REPLACE ... AS SELECT, each with CLUSTER BY (a) UNORDERED. After each rejection, the rows, the schema and the clustering are unchanged. As you suggested, it covers only the staging path, because the non-staging ReplaceTableExec drops the original first. I also added CLUSTER BY (a) LOCALLY ORDERED BY (b) to this test, which asserts mode NONE with the ordering b ASC NULLS FIRST kept and the clustering intact.
| Both clauses are passed to the catalog, which has to support them: a catalog that does not | ||
| advertise support for a write distribution and ordering rejects the statement rather than | ||
| creating a table that silently lacks the requested layout. The built-in catalogs do not | ||
| support them. Both may also be combined with `CLUSTER BY`. Spark passes the clustering columns |
There was a problem hiding this comment.
Both may also be combined with CLUSTER BY appears to include DISTRIBUTED BY PARTITION, but that combination is rejected by the parser and by the test at CreateTableWriteOrderSuite:311. Could we name the allowed forms explicitly here: ORDERED BY, LOCALLY ORDERED BY, and UNORDERED may be combined with CLUSTER BY, subject to the catalog accepting the combination?
There was a problem hiding this comment.
Thanks, that sentence was too broad. It now reads: "ORDERED BY, LOCALLY ORDERED BY, and UNORDERED may also be combined with CLUSTER BY, subject to the catalog accepting the combination ... DISTRIBUTED BY PARTITION cannot be combined with CLUSTER BY, as described above."
|
Thank you for the update, @anuragmantri. I took one more pass over
Minor:
Pre-existing in the shared
|
|
Gentle ping, @anuragmantri . |
anuragmantri
left a comment
anuragmantri
left a comment
There was a problem hiding this comment.
@szehon-ho thanks for confirming the CLUSTER BY behavior. It stays combinable with ORDERED BY, LOCALLY ORDERED BY and UNORDERED, and it is still rejected with DISTRIBUTED BY PARTITION. Your three inline points are addressed in 5569461.
Thank you for the two detailed passes, @dongjoon-hyun. Keeping your numbering:
-
Done. A finite
FLOATnow renders as<v>F,isSpellablerejects non-finiteFLOAT/DOUBLE, and1.5Fis in the round-trip test. Two related cases came up:Float.MaxValuerenders as3.4028235E38F, which is outside the parser'sFLOATrange on replay, so it is treated as unspellable, andDESCRIBEnow prints a non-finiteFLOATasCAST('NaN' AS FLOAT)instead ofNaNF. -
Good catch. Switched to
CatalystLiteral.create. A connector-reportedjava.lang.Stringliteral is now tested in bothDESCRIBE TABLE EXTENDEDandSHOW CREATE TABLE. -
Done. The Javadoc of
SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERINGnow says that Spark passes the clustering columns and the requested distribution and ordering without reconciling them, and the catalog interprets or rejects the combination. -
Yes, it is intended. A
DelegatingTableexposes what the catalog stored, so reporting the layout keepsDESCRIBEandSHOW CREATE TABLEfaithful, and another engine reading the same catalog can apply it. Spark's own reads and writes go through the v1 path, which does not apply it, and the capability Javadoc now says so. -
Agreed. It is now an enum,
WriteDistributionMode(HASH,RANGE,NONE),@since 4.4.0, and theStringconstants are gone.SHOW CREATE TABLEno longer needs the unknown-mode case.DESCRIBEstill printshash/range/none. -
Added "CTAS and RTAS write their first load with the declared distribution and ordering". A test catalog builds
InMemoryTablewith the distribution and ordering from theTableInfo, and the test checks the inner write: a range shuffle plus sort forORDERED BY, a sort with no shuffle forLOCALLY ORDERED BY, and a hash shuffle plus sort forCREATE OR REPLACE ... DISTRIBUTED BY PARTITION ... AS SELECT. -
Good catch.
toSQLnow quotes the name withquoteIfNeeded, and the name check inisSpellableis gone. A new test round-tripsORDERED BY `z-order`(id), `+`(id, 1), and the unspellable fixture is now a transform with a transform argument,f(g(id)). -
I kept the parse-time check and fixed the message and the docs. Both now say that a statement that defines no schema (no column list, no typed partition columns, and no
AS SELECT) cannot declare partitioning, so it cannot useDISTRIBUTED BY PARTITION. I rephrased theORDERED BYdocs sentence the same way, sincePARTITIONED BY (p INT)defines a schema without a column list. I preferred this to moving the check because it keeps a simple guarantee for catalogs, thatHASHalways comes with non-emptypartitions(). Relaxing that later stays compatible, but tightening it after 4.4.0 would not. -
Added
DAYS(TS)andBUCKET(4, ID)to that test, next to the lowercase forms, which do normalize. Filed SPARK-59944 for makingApplyTransformaRewritableTransform. That alone would not stopBUCKET(4, id)from reaching the catalog as a generic transform. That needs case-insensitive dispatch invisitTransform, which also changesPARTITIONED BY, so the JIRA covers both. -
Done. The capability Javadoc and
TableInfo#writeDistributionMode()now listTableCatalog#createTable(Identifier, TableInfo)and the threeStagingTableCatalogstage*(Identifier, TableInfo)overloads, and say that a catalog reporting the capability must override each one it can be reached through. The capability Javadoc also notes that a staging catalog needs all four, because aCREATE TABLEwithoutAS SELECTis not staged. -
The comment on
toSQLnow says thatPARTITIONED BYand thePart Nrows ofDESCRIBErender throughdescribe. Sharing one renderer is SPARK-59946. -
I went with unwrapping.
ORDERED BY idnow reaches the catalog as a bareNamedReference, like the other v2SortOrderproducers, andPreprocessTableCreationnormalizes a bare reference the same way it normalizes a transform's references. TheTableInfo#writeOrdering()Javadoc now documents the shape (a column is aNamedReference, anything else aTransform), that the array is never null, and that Spark only checks that the referenced columns exist, so orderability and argument types are left to the catalog. -
Thanks for the SPARK-43529 precedent. I'd like to do this as a follow-up, SPARK-59943, and land it before 4.4.0. One thing to handle there: once the pair travels on
TableSpec, a custom strategy like Paimon's compiles again but skipsvalidateCatalogForWriteDistributionAndOrdering, which runs inDataSourceV2Strategy. So the follow-up should move that check into the execs or intoCheckAnalysis. -
Done. The
SHOW CREATE TABLEassertions now useddl.split("\n").contains(expected), the replay test compareswriteDistributionMode()as well as the ordering, and there is aPARTITIONED BY (c) ORDERED BY (id)case. The keyword test now setsENFORCE_RESERVED_KEYWORDSin its ANSI iteration.
| None | ||
| } | ||
| // Bucketing counts as partitioning here; CLUSTER BY does not. | ||
| val hasPartitioning = table.partitioning.exists(!_.isInstanceOf[ClusterByTransform]) |
There was a problem hiding this comment.
Good catch. The guard now uses the ClusterByTransform(_) extractor, so a connector's generic cluster_by transform counts as clustering, not partitioning. I added "SHOW CREATE TABLE treats a connector's cluster_by transform as clustering". It uses a test catalog whose tables report CLUSTER BY as Expressions.apply("cluster_by", ...). It returns a DelegatingTable, because InMemoryTable rejects a generic cluster_by at creation. With mode hash and both an empty ordering and ORDERED BY (b), the test asserts that neither DISTRIBUTED BY PARTITION nor ORDERED BY is emitted, for both the native and the generic transform. PARTITIONED BY (a) and CLUSTERED BY (a) INTO 4 BUCKETS are the positive controls and still emit DISTRIBUTED BY PARTITION ORDERED BY (b ASC NULLS FIRST). I checked that the generic case fails with the old isInstanceOf check.
| } | ||
| } | ||
|
|
||
| test("CLUSTER BY and the write clauses both reach the catalog") { |
There was a problem hiding this comment.
Added "a catalog with the capability can still reject a combination it does not support". It uses a staging catalog that advertises the capability but rejects CLUSTER BY with UNORDERED in createTable and in all three stage* methods. For CREATE TABLE (the createTable path) and CTAS (the stageCreate path), the catalog's error reaches the caller and no table is published. The test then populates a CLUSTER BY (a) table and runs REPLACE TABLE, CREATE OR REPLACE TABLE, RTAS and CREATE OR REPLACE ... AS SELECT, each with CLUSTER BY (a) UNORDERED. After each rejection, the rows, the schema and the clustering are unchanged. As you suggested, it covers only the staging path, because the non-staging ReplaceTableExec drops the original first. I also added CLUSTER BY (a) LOCALLY ORDERED BY (b) to this test, which asserts mode NONE with the ordering b ASC NULLS FIRST kept and the clustering intact.
| Both clauses are passed to the catalog, which has to support them: a catalog that does not | ||
| advertise support for a write distribution and ordering rejects the statement rather than | ||
| creating a table that silently lacks the requested layout. The built-in catalogs do not | ||
| support them. Both may also be combined with `CLUSTER BY`. Spark passes the clustering columns |
There was a problem hiding this comment.
Thanks, that sentence was too broad. It now reads: "ORDERED BY, LOCALLY ORDERED BY, and UNORDERED may also be combined with CLUSTER BY, subject to the catalog accepting the combination ... DISTRIBUTED BY PARTITION cannot be combined with CLUSTER BY, as described above."
|
Thank you for updating, @anuragmantri . |
dongjoon-hyun
left a comment
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for addressing points 1-14, @anuragmantri. I reviewed the latest commit, 5569461, and left 15 inline comments, continuing the numbering from points 1-14. Three of them (19, 21 and 29) cover cases that the fixes for points 8, 2 and 12 do not reach yet. Here is a summary, ordered by priority.
Correctness
ShowCreateTableExecL156:TimeTypeis missing from the allow-list, so aTIMEliteral in a sort key drops every write clause from SHOW CREATE TABLE. This is a regression in this commit and the silent drop from point 7.ShowCreateTableExecL154:true,falseandNULLre-parse as column references inside a transform, soBooleanTypeandNullTypedo not round-trip.ShowCreateTableExecL155:TIMESTAMPliterals are printed as session-local wall time without an offset, so a replay can shift the value (DST overlap, another time zone) or turn it intoTIMESTAMP_NTZ.ShowCreateTableExecL177: the extractor-basedhasPartitioningthrowsClassCastExceptionfor a table with acluster_bytransform that has a literal argument (a regression vs. master), and disagrees with the parser onPARTITIONED BY (cluster_by(a)) DISTRIBUTED BY PARTITION, so a replay loses HASH.error-conditions.jsonL7345: for aCLUSTER BYtable, the advice to addPARTITIONED BYorCLUSTERED BY ... INTO ... BUCKETScannot be followed.ShowCreateTableExecL146: the "same type" claim fails for connector-declaredDECIMAL(p,s)andCHAR/VARCHARliterals and under two legacy confs.WriteDistributionAndOrderingL73: a literal conversion that throws makes DESCRIBE TABLE EXTENDED fail entirely, and a connector's ownNamedReferenceis printed viadescribe().ShowCreateTableExecL137: connector keys namedbucket,years,months,daysorhourswith other argument shapes print DDL that the parser rejects.AstBuilderL6375: a parameter marker in a transform argument fails with a rawClassCastException. This is pre-existing in PARTITIONED BY, so a follow-up is fine.
Tests
CreateTableWriteOrderSuiteL244: the error checks are hand-rolled, one of them checks no condition, and nothing pins the SQLSTATEs of the three new conditions.PlanResolutionSuiteL3696: the six new tests copy unrelated assertions and only re-pin parser output covered elsewhere.
Docs and comments
CreateTableLikeExecL50: the "not copied" caveat is only in this scaladoc; the publicTableCatalog.createTableLikeandTableInfoJavadocs still describetableInfoas complete.- PR description: it still describes the mode as a
StringwithDISTRIBUTION_MODE_*constants (also in design decision 1), says the suite has 30 tests (now 38), and does not mention the two new abstract members ofV2CreateTablePlan(writeOrdering,withWriteOrdering) from the minor items under points 1-6.
API and design
TableCatalogCapabilityL108:DelegatingCatalogExtensionforwardscapabilities()but notcreateTable(Identifier, TableInfo), which leaves a latent hole in this Javadoc's promise.TableInfoL139: the "never null" ordering is not enforced atbuild().
Cleanup
CheckAnalysisL995: the comment gives a stale reason, and the block copies the partitioning check with workarounds.
The most important are 15-18. 15 is a regression in this commit, and 16-18 make SHOW CREATE TABLE emit DDL that fails, or silently changes the declared layout, on replay.
| case (_, s: StringType) => DataTypeUtils.isDefaultStringCharOrVarcharType(s) | ||
| case (_, BooleanType | ByteType | ShortType | IntegerType | LongType | _: DecimalType | | ||
| BinaryType | DateType | TimestampType | TimestampNTZType | _: DayTimeIntervalType | | ||
| _: YearMonthIntervalType) => true |
There was a problem hiding this 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: d2aa867 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).
There was a problem hiding this 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.
| 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 | |
There was a problem hiding this 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.
There was a problem hiding this 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.
| 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 | |
There was a problem hiding this 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.
There was a problem hiding this 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.
| } | ||
| // Bucketing counts as partitioning here; CLUSTER BY does not. | ||
| val hasPartitioning = table.partitioning.exists { | ||
| case ClusterByTransform(_) => false |
There was a problem hiding this 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 d2aa867 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.
There was a problem hiding this 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.
| "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." |
There was a problem hiding this 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().
There was a problem hiding this 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.
| } | ||
| } | ||
|
|
||
| test("SPARK-34586: v2 table creation (global writeOrdering)") { |
There was a problem hiding this 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?
There was a problem hiding this 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.
| * [[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 |
There was a problem hiding this 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.
There was a problem hiding this 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.
| * <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 |
There was a problem hiding this 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?
There was a problem hiding this 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.
| * | ||
| * @since 4.4.0 | ||
| */ | ||
| public Builder withWriteOrdering(SortOrder[] writeOrdering) { |
There was a problem hiding this 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?
There was a problem hiding this 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.
| "cols" -> badReferences.map(r => toSQLId(r)).mkString(", "))) | ||
| } | ||
|
|
||
| // PreprocessTableCreation only normalizes column and RewritableTransform references, |
There was a problem hiding this 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?
There was a problem hiding this 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.
|
Gentle ping, @anuragmantri ~ |
anuragmantri
left a comment
anuragmantri
left a comment
There was a problem hiding this comment.
Thank you for another careful pass, @dongjoon-hyun. Points 15-30 are addressed with replies on each thread.
One design change is worth calling out, because several of your points (15, 16, 17, 20, 21, 22) were the same problem: an allow-list could not cover every literal type, name and conf. SHOW CREATE TABLE now renders the keys, parses the clauses back with the session's parser, and emits them only when the parsed mode and keys equal the declared ones. It retries once with every name quoted, for reserved keywords. It also omits the clauses when the catalog does not report the capability or a key references a column the table does not have, since a replay would fail in both cases. WriteDistributionAndOrderingUtilsSuite checks "emitted iff it parses back to the same key" over connector-reported keys of every literal type and shape, including under the confs that change parsing.
| case (_, s: StringType) => DataTypeUtils.isDefaultStringCharOrVarcharType(s) | ||
| case (_, BooleanType | ByteType | ShortType | IntegerType | LongType | _: DecimalType | | ||
| BinaryType | DateType | TimestampType | TimestampNTZType | _: DayTimeIntervalType | | ||
| _: YearMonthIntervalType) => true |
There was a problem hiding this 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.
| 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 | |
There was a problem hiding this 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.
| 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 | |
There was a problem hiding this 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.
| } | ||
| // Bucketing counts as partitioning here; CLUSTER BY does not. | ||
| val hasPartitioning = table.partitioning.exists { | ||
| case ClusterByTransform(_) => false |
There was a problem hiding this 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.
| "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." |
There was a problem hiding this 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.
| * [[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 |
There was a problem hiding this 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.
| * <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 |
There was a problem hiding this 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.
| "cols" -> badReferences.map(r => toSQLId(r)).mkString(", "))) | ||
| } | ||
|
|
||
| // PreprocessTableCreation only normalizes column and RewritableTransform references, |
There was a problem hiding this 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.
| * | ||
| * @since 4.4.0 | ||
| */ | ||
| public Builder withWriteOrdering(SortOrder[] writeOrdering) { |
There was a problem hiding this 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.
| } | ||
|
|
||
| // A plain column is passed as a bare reference rather than as `identity(col)`. | ||
| val key = visitTransform(ctx.transform) match { |
There was a problem hiding this comment.
Thanks, filed SPARK-59977 for this since it seems like an existing bug.
dongjoon-hyun
left a comment
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for addressing points 15-30, @anuragmantri. I reviewed the latest commit, 60a4ea7, and left 8 inline comments, continuing the numbering from points 1-30. The parse-back check works well: I could not find a table created in SQL whose SHOW CREATE TABLE output fails or declares a different layout when replayed in the same session. 33 and 34 are the remaining cases of points 15 and 17, and 39 is point 27, which is still open. Here is a summary, ordered by priority.
Correctness
WriteDistributionAndOrderingL122:referencesExistdoes not see a column inside a nestedIdentityTransform, so SHOW CREATE TABLE can emit anORDERED BYthat fails on replay. This is new in this commit.WriteDistributionAndOrderingL173: DESCRIBE now prints anExtractor UDF key without its field or function name.WriteDistributionAndOrderingL188: aTIME(7-9)value with trailing zeros still drops the whole clause (point 15).WriteDistributionAndOrderingL185: the nanosecondTIMESTAMP_LTZtype is still printed in session-local time without an offset (point 17).
Tests
WriteDistributionAndOrderingUtilsSuiteL156: the HASH rows cannot fail, because this suite's replay has noPARTITIONED BY, and no test in the suite emits a HASH or NONE pair.PreprocessTableCreationL334:SPECIFY_WRITE_ORDERING_IS_NOT_ALLOWEDis the only new condition without a query context, and the convertedcheckErrornow pins that.
Docs and comments
TableInfoL100: the new sentence that the parser checks the arguments ofbucket,years,months,daysandhourspromises more than the parser does.sql-ref-syntax-ddl-create-table-datasource.mdL41: the syntax block still shows the parentheses ofORDERED BYas required.- PR description (point 27): it still describes a
Stringmode withDISTRIBUTION_MODE_*constants, including design decision 1, says the suite has 30 tests, and does not mention the two new abstract members ofV2CreateTablePlan(writeOrdering,withWriteOrdering). It should now also describe the parse-back check and theDelegatingCatalogExtensionchange.dev/merge_spark_pr.pyputs this text into the commit message.
Other notes
- The
GeneralScalarExpression.toString()recursion you found is pre-existing:V2ExpressionSQLBuilder.visitUnexpectedExprbuilds its error message withString.valueOf(expr), which callstoString()again. It can be fixed separately from this PR. - In the reply to point 16,
TRUEis not emitted underspark.sql.ansi.enforceReservedKeywordseither. It is also inansiNonReserved, so it always parses back as a column; onlyFALSEandNULLare reproduced in that mode. The new docs sentence is right as written. - The CI failures look unrelated.
SQLAppStatusListenerWithInMemoryStoreSuite"driver side SQL metrics" is a racy test that SPARK-59784 fixed on master after this branch's base, andKafkaRealTimeModeWindowSuite"tumbling window count" timed out.
The most important are 31 and 39: 31 makes SHOW CREATE TABLE emit DDL that fails on replay, and 39 becomes the commit message.
| * matches the ordering of a CREATE/REPLACE TABLE statement. | ||
| */ | ||
| def referencesExist(schema: StructType, writeOrdering: Seq[SortOrder]): Boolean = { | ||
| writeOrdering.flatMap(_.expression().references()).forall { ref => |
There was a problem hiding this comment.
referencesExist checks the declared key's references(), but ApplyTransform.references collects only top-level references, so a column inside a nested IdentityTransform is never checked. For a table t(id INT), a connector key Expressions.apply("f", Expressions.identity("missing")) passes this check vacuously. toSQL and keyOf both flatten the nested identity, so the parse-back comparison succeeds and SHOW CREATE TABLE prints ORDERED BY (f(missing) ASC NULLS FIRST). Replaying it fails with WRITE_ORDERING_WITH_UNKNOWN_COLUMN (CheckAnalysis.scala:996-1003). identity("ID") for a column id fails the same way, since an ApplyTransform is not normalized (SPARK-59944). This is new in this commit: the previous isSpellable rejected an argument that was neither a reference nor a literal.
Could we check the replayed ordering instead, which is what CheckAnalysis will see, e.g. referencesExist(schema, ordering) inside the find or in ShowCreateTableExec.replay?
There was a problem hiding this comment.
Good catch. The column check now runs on the replayed ordering, which is what CheckAnalysis sees, so f(identity(missing)) is omitted. A nested identity(...) is also no longer compared as a plain column: only a key's top level counts as identity(col), so f(identity(id)) is omitted too. Both are in the unit suite and in ORDERING_OVERRIDE with omit-and-replay rows.
| case g: GeneralScalarExpression => | ||
| g.children().map(toSQL(_, quote)).mkString(s"${g.name}(", ", ", ")") | ||
| case other => | ||
| other.children().map(toSQL(_, quote)).mkString(s"${other.getClass.getSimpleName}(", ", ", ")") |
There was a problem hiding this comment.
Minor: this fallback now applies to every connector expression, not only to the GeneralScalarExpression names that V2ExpressionSQLBuilder does not know. An Extract key prints as Extract(ts) instead of EXTRACT(YEAR FROM ts), because Extract.children() is just the source (Extract.java:60), and a UserDefinedScalarFunc prints as UserDefinedScalarFunc(id) instead of my_udf(id). So DESCRIBE TABLE EXTENDED loses the field and the function name that 5569461 printed with describe. The GeneralScalarExpression case at L170-171 has a similar effect on known names, e.g. +(id, 1) instead of id + 1.
Could describe stay the default here, with the children-based rendering only for a GeneralScalarExpression name that V2ExpressionSQLBuilder does not know? Once the pre-existing recursion is fixed separately, that special case could go away too.
There was a problem hiding this comment.
Thanks. DESCRIBE renders connector expressions with ToStringSQLBuilder again, so EXTRACT(YEAR FROM ts), my_udf(id) and id + 1 print as describe printed them. The builder is a subclass that renders literals and references with their types and quoting, and renders GetArrayItem and VariantGet through their children rather than through toString. It renders an expression it does not know as name(children) instead of throwing, so a known parent keeps its name, e.g. my_udf(DATE_TRUNC(id)). Tests cover these cases, plus a sort order and a null child nested in a key.
| case (micros: Long, TimestampType) => | ||
| val utc = TimestampFormatter.getFractionFormatter(ZoneOffset.UTC).format(micros) | ||
| s"TIMESTAMP_LTZ '${utc}Z'" | ||
| case _ => l.sql |
There was a problem hiding this comment.
Minor: point 15 is not complete for a TIME(7-9) value with trailing zeros. TIME '12:00:00.100000000' is a TIME(9) literal, but Literal.sql prints TIME '12:00:00.1' (literals.scala:691, whose formatter drops trailing zeros), which parses back as TIME(6). The comparison fails and the whole clause is dropped, although TIME '12:00:00.100000000' would replay correctly. A value without trailing zeros, such as TIME '12:00:00.123456789', does round-trip, so the reply's "a TIME with 7 to 9 fractional digits is omitted" holds only for these values.
Could literalToSQL pad the fraction to the type's precision for TimeType, as Literal.sql already does with padToNanosPrecision for the nanosecond timestamp types (literals.scala:700-702)?
There was a problem hiding this comment.
Done. literalToSQL pads a TIME with more than microsecond precision to its precision, so TIME '12:00:00.100000000' round-trips as TIME(9). The unit suite has TIME(7) and TIME(9) rows with and without trailing zeros.
|
|
||
| private def literalToSQL(l: CatalystLiteral): String = (l.value, l.dataType) match { | ||
| case (f: Float, FloatType) if java.lang.Float.isFinite(f) => s"${f}F" | ||
| case (micros: Long, TimestampType) => |
There was a problem hiding this comment.
Minor (preview): point 17 is not complete for the nanosecond timestamps. With spark.sql.timestampNanosTypes.enabled, a TIMESTAMP literal with 7-9 fractional digits is a TimestampLTZNanosType, which this case does not match, so it falls to l.sql and prints TIMESTAMP_LTZ '<session-local wall time>' without an offset (literals.scala:702). In America/Los_Angeles, ORDERED BY f(id, TIMESTAMP '2020-11-01 01:30:00.123456789-08:00') then fails the comparison in the DST overlap and drops the clause, and a value outside the overlap replays shifted in another session time zone.
Could TimestampLTZNanosType also be printed in UTC with a Z suffix, padded to its precision?
There was a problem hiding this comment.
Done. A TimestampLTZNanosType literal now renders in UTC with a Z suffix, padded to its precision. A test checks TIMESTAMP_LTZ '2020-11-01 01:30:00.123456789-08:00' in America/Los_Angeles with spark.sql.timestampNanosTypes.enabled, and a nanosecond TIMESTAMP_NTZ row was added too.
| test("the pairs with no clause form are not emitted") { | ||
| val keys = Seq(key(id)) | ||
| Seq( | ||
| (WriteDistributionMode.HASH, keys, Seq.empty[Transform]), |
There was a problem hiding this comment.
Minor: these two HASH rows cannot fail. This suite's replay (L51) parses CREATE TABLE t (id INT) USING foo $clauses, which has no PARTITIONED BY, so the parser rejects every DISTRIBUTED BY PARTITION ... rendering and writeClausesSQL returns None whatever it emits. Production replays with PARTITIONED BY (p) (ShowCreateTableExec.scala:162). Removing the if partitioned guards at WriteDistributionAndOrdering.scala:102-104 would still pass here, and since emitted (L55-57) always uses RANGE, no test in this suite emits a HASH or NONE pair.
Could this test use a replay that accepts any clauses, e.g. _ => Some((mode, ordering)), and could the suite add a positive HASH row such as (HASH, Seq(key(id)), Seq(Expressions.identity("p"))) expecting DISTRIBUTED BY PARTITION ORDERED BY (id ASC NULLS FIRST)?
There was a problem hiding this comment.
Thanks. The suite now replays the same statement as SHOW CREATE TABLE (replayStatement, shared with ShowCreateTableExec). The no-clause-form test uses a replay that accepts any clauses, so only the guards decide, and it covers a null mode. A new test emits each mode's clause form, including DISTRIBUTED BY PARTITION ORDERED BY (id ASC NULLS FIRST) and LOCALLY ORDERED BY .... Another omits an unspellable key under HASH, RANGE and NONE.
| throw QueryCompilationErrors.specifyPartitionNotAllowedWhenTableSchemaNotDefinedError() | ||
| } | ||
| if (create.writeOrdering.nonEmpty) { | ||
| throw QueryCompilationErrors |
There was a problem hiding this comment.
Nit: SPECIFY_WRITE_ORDERING_IS_NOT_ALLOWED is the only one of the four new conditions without a query context. specifyWriteOrderingNotAllowedWhenTableSchemaNotDefinedError() creates the AnalysisException without an origin, while WRITE_ORDERING_WITH_UNKNOWN_COLUMN and the two parse errors point at the statement, and the converted checkError at CreateTableWriteOrderSuite.scala:329 now pins an empty context. The legacy _LEGACY_ERROR_TEMP_1165 at L331 has the same gap, so this is optional, but changing it later means changing the test again.
Could this error take create.origin?
There was a problem hiding this comment.
Done. The error now takes create.origin, and the test pins the statement as its context.
| * <p> | ||
| * A plain column is a {@link org.apache.spark.sql.connector.expressions.NamedReference}; any | ||
| * other key is a {@link Transform}, such as {@code bucket(16, id)}. Spark checks that each | ||
| * referenced column exists in the table schema, and the parser checks the arguments of |
There was a problem hiding this comment.
This new sentence promises more validation than the parser does. visitTransform dispatches these names case-sensitively (AstBuilder.scala:5869), so ORDERED BY BUCKET(id, 16) or ORDERED BY DAYS(1) reaches the catalog as a generic transform with no argument check, and DAYS(1) references no column, so the column check does not apply either. For lowercase bucket, the parser checks only the argument kinds: bucket(16) without a column, a zero or negative count, and bucket(4294967300, id), which .toInt truncates to 4 (AstBuilder.scala:5877), are all accepted. WriteDistributionAndOrderingUtilsSuite even lists "bucket without columns" as spellable. A catalog author reading this sentence may skip their own validation.
Could we drop the parser clause, or say that only the lowercase names have their argument kinds checked and that bucket counts and column lists are not validated?
There was a problem hiding this comment.
Thanks. It now says that the parser checks the argument kinds of the lowercase bucket, years, months, days and hours, and that Spark does not check orderability, a positive bucket count, a bucket's column list, or the arguments of any other transform.
| [ COMMENT table_comment ] | ||
| [ TBLPROPERTIES ( key1=val1, key2=val2, ... ) ] | ||
| [ DISTRIBUTED BY PARTITION ] | ||
| [ [ LOCALLY ] ORDERED BY ( write_order_field [ , ... ] ) | UNORDERED ] |
There was a problem hiding this comment.
Nit: the parentheses are optional, as L130-131 says and the AstBuilder scaladoc now shows with [(] ... [)], but this syntax block still shows them as required. Could it be [ [ LOCALLY ] ORDERED BY { ( write_order_field [ , ... ] ) | write_order_field [ , ... ] } | UNORDERED ]?
There was a problem hiding this comment.
Done: [ LOCALLY ] ORDERED BY { ( write_order_field [ , ... ] ) | write_order_field [ , ... ] }.
anuragmantri
left a comment
anuragmantri
left a comment
There was a problem hiding this comment.
Thank you, @dongjoon-hyun. Points 31-39 are addressed in 88882a2, with replies on each thread.
On your other notes:
Point 16: you are right, thanks for the correction. TRUE is in ansiNonReserved, so it always parses back as a column and is never reproduced. Only FALSE and NULL are reproduced under spark.sql.ansi.enforceReservedKeywords. The docs sentence was already right.
The GeneralScalarExpression.toString() recursion: I filed SPARK-60075 for the fix in V2ExpressionSQLBuilder.visitUnexpectedExpr. This PR only avoids it in DESCRIBE.
While checking the class behind 31 and 32, a few more cases came up and are fixed in the same commit.
SHOW CREATE TABLEnow also omits the clauses when the statement would create a v1 table, because the session catalog with a provider that is not a v2 source rejects them on replay.DESCRIBEno longer recurses throughGetArrayItem,VariantGetor a nested sort order. A nestedidentity(...)is no longer compared as a plain column. AndORDERED BYid.x on an INT column reportsWRITE_ORDERING_WITH_UNKNOWN_COLUMNwith the statement as its context instead of a context-freeINVALID_FIELD_NAME.
| * matches the ordering of a CREATE/REPLACE TABLE statement. | ||
| */ | ||
| def referencesExist(schema: StructType, writeOrdering: Seq[SortOrder]): Boolean = { | ||
| writeOrdering.flatMap(_.expression().references()).forall { ref => |
There was a problem hiding this comment.
Good catch. The column check now runs on the replayed ordering, which is what CheckAnalysis sees, so f(identity(missing)) is omitted. A nested identity(...) is also no longer compared as a plain column: only a key's top level counts as identity(col), so f(identity(id)) is omitted too. Both are in the unit suite and in ORDERING_OVERRIDE with omit-and-replay rows.
| case g: GeneralScalarExpression => | ||
| g.children().map(toSQL(_, quote)).mkString(s"${g.name}(", ", ", ")") | ||
| case other => | ||
| other.children().map(toSQL(_, quote)).mkString(s"${other.getClass.getSimpleName}(", ", ", ")") |
There was a problem hiding this comment.
Thanks. DESCRIBE renders connector expressions with ToStringSQLBuilder again, so EXTRACT(YEAR FROM ts), my_udf(id) and id + 1 print as describe printed them. The builder is a subclass that renders literals and references with their types and quoting, and renders GetArrayItem and VariantGet through their children rather than through toString. It renders an expression it does not know as name(children) instead of throwing, so a known parent keeps its name, e.g. my_udf(DATE_TRUNC(id)). Tests cover these cases, plus a sort order and a null child nested in a key.
| case (micros: Long, TimestampType) => | ||
| val utc = TimestampFormatter.getFractionFormatter(ZoneOffset.UTC).format(micros) | ||
| s"TIMESTAMP_LTZ '${utc}Z'" | ||
| case _ => l.sql |
There was a problem hiding this comment.
Done. literalToSQL pads a TIME with more than microsecond precision to its precision, so TIME '12:00:00.100000000' round-trips as TIME(9). The unit suite has TIME(7) and TIME(9) rows with and without trailing zeros.
|
|
||
| private def literalToSQL(l: CatalystLiteral): String = (l.value, l.dataType) match { | ||
| case (f: Float, FloatType) if java.lang.Float.isFinite(f) => s"${f}F" | ||
| case (micros: Long, TimestampType) => |
There was a problem hiding this comment.
Done. A TimestampLTZNanosType literal now renders in UTC with a Z suffix, padded to its precision. A test checks TIMESTAMP_LTZ '2020-11-01 01:30:00.123456789-08:00' in America/Los_Angeles with spark.sql.timestampNanosTypes.enabled, and a nanosecond TIMESTAMP_NTZ row was added too.
| test("the pairs with no clause form are not emitted") { | ||
| val keys = Seq(key(id)) | ||
| Seq( | ||
| (WriteDistributionMode.HASH, keys, Seq.empty[Transform]), |
There was a problem hiding this comment.
Thanks. The suite now replays the same statement as SHOW CREATE TABLE (replayStatement, shared with ShowCreateTableExec). The no-clause-form test uses a replay that accepts any clauses, so only the guards decide, and it covers a null mode. A new test emits each mode's clause form, including DISTRIBUTED BY PARTITION ORDERED BY (id ASC NULLS FIRST) and LOCALLY ORDERED BY .... Another omits an unspellable key under HASH, RANGE and NONE.
| throw QueryCompilationErrors.specifyPartitionNotAllowedWhenTableSchemaNotDefinedError() | ||
| } | ||
| if (create.writeOrdering.nonEmpty) { | ||
| throw QueryCompilationErrors |
There was a problem hiding this comment.
Done. The error now takes create.origin, and the test pins the statement as its context.
| * <p> | ||
| * A plain column is a {@link org.apache.spark.sql.connector.expressions.NamedReference}; any | ||
| * other key is a {@link Transform}, such as {@code bucket(16, id)}. Spark checks that each | ||
| * referenced column exists in the table schema, and the parser checks the arguments of |
There was a problem hiding this comment.
Thanks. It now says that the parser checks the argument kinds of the lowercase bucket, years, months, days and hours, and that Spark does not check orderability, a positive bucket count, a bucket's column list, or the arguments of any other transform.
| [ COMMENT table_comment ] | ||
| [ TBLPROPERTIES ( key1=val1, key2=val2, ... ) ] | ||
| [ DISTRIBUTED BY PARTITION ] | ||
| [ [ LOCALLY ] ORDERED BY ( write_order_field [ , ... ] ) | UNORDERED ] |
There was a problem hiding this comment.
Done: [ LOCALLY ] ORDERED BY { ( write_order_field [ , ... ] ) | write_order_field [ , ... ] }.
dongjoon-hyun
left a comment
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for addressing points 31-38, @anuragmantri. I reviewed the latest commit, 88882a2, and left 3 inline comments, continuing the numbering from points 1-39. None of them is a correctness issue. Point 39 is still open, so I repeated it here as 43. Here is a summary, ordered by file.
Cleanup
WriteDistributionAndOrderingL194: the reason this comment gives forDescribingSQLBuilderno longer holds on master. SPARK-59983 (#59241) already fixed theGeneralScalarExpressionrecursion inToStringSQLBuilder.visitUnexpectedExpr, so SPARK-60075 looks like a duplicate.ShowCreateTableExecL161:createsV1TablecopiesResolveSessionCatalog.supportsV1CommandandisV2Providerinline, so SHOW CREATE TABLE can drift from what CREATE TABLE actually does.DescribeTableExecL234:table.writeDistributionMode()andtable.writeOrdering()are each called up to three times. Reading them once would avoid repeated connector work and inconsistent snapshots.
Docs
- PR description (point 39, not in 88882a2): the description is a separate edit from the commits, and it still says the following.
- It describes
writeDistributionModeas aStringwithDISTRIBUTION_MODE_HASH/_RANGE/_NONEvalues, and design decision 1 still says "AStringmode", while the API is theWriteDistributionModeenum. - It says
CreateTableWriteOrderSuitehas 30 tests, but it now has 47. - It does not mention the two new abstract members of
V2CreateTablePlan(writeOrdering,withWriteOrdering), the parse-back check in SHOW CREATE TABLE, or thatDelegatingCatalogExtensionno longer forwards the new capability.
- It describes
The most important is 43, because dev/merge_spark_pr.py puts this text into the commit message.
| e.children().map(toSQL(_, quote)).mkString(s"$name(", ", ", ")") | ||
| } | ||
|
|
||
| // `ToStringSQLBuilder` renders a few expressions through their children's `describe`, and for an |
There was a problem hiding this comment.
[40] The recursion this comment describes is already fixed on master.
SPARK-59983 (#59241, 5926a10) changed ToStringSQLBuilder.visitUnexpectedExpr to render a GeneralScalarExpression with an unknown name as a function call. It went into master after this branch's base, so "recurses without end for a GeneralScalarExpression name it does not know" no longer holds after a rebase. SPARK-60075, which you filed for this in your last reply, looks like a duplicate of SPARK-59983 and can probably be closed.
DescribingSQLBuilder itself is still useful, because it routes literals and references through toSQL and renders other unknown expressions from their children. Could we rebase and reword this comment to give only those reasons?
There was a problem hiding this comment.
Done. After rebasing on master, the comment gives only the reasons that still hold. DescribingSQLBuilder renders literals and references with toSQL, so they keep their type and quoting, and renders the children of GetArrayItem and VariantGet itself rather than through their toString. It renders an expression ToStringSQLBuilder does not know, and a PartitionPredicate, from its children. The last part is still needed on master: a connector PartitionPredicate that does not override describe still recurses. I rescoped SPARK-60075 to cover this.
|
|
||
| // In the session catalog, a CREATE TABLE whose provider is not a v2 source creates a v1 table, | ||
| // which cannot record the clauses, so the replay would be rejected. | ||
| private def createsV1Table(resolvedTable: ResolvedTable): Boolean = { |
There was a problem hiding this comment.
[41] createsV1Table copies ResolveSessionCatalog's v1-fallback rule inline.
v1Capable is ResolveSessionCatalog.supportsV1Command (L1030), and the provider check is isV2Provider (L950). Both are private there, so this is a second copy that has to change whenever the rule changes, such as the provider/serde resolution in getStorageFormatAndProvider (L765) or the v1 source list. If the copies drift, SHOW CREATE TABLE either emits clauses that the replayed CREATE TABLE rejects with UNSUPPORTED_FEATURE.TABLE_OPERATION, or drops clauses that would have worked, and the parse-back check cannot catch this because it only parses. Could we move these two predicates into a shared helper, e.g. next to DataSourceV2Utils.getTableProvider, and call it from both places?
There was a problem hiding this comment.
Agreed. The copy had in fact already drifted. For a table with no provider it used spark.sql.sources.default, while CREATE TABLE creates a Hive serde table when spark.sql.legacy.createHiveTableByDefault is set. supportsV1Command, isV2Provider, createsHiveTableByDefault and createTableProvider now live in DataSourceV2Utils, and both ResolveSessionCatalog and ShowCreateTableExec call them; ResolveSessionCatalog's behavior is unchanged. A new test covers a provider-less table in a session catalog extension with spark.sql.sources.default set to a v2 source: the clauses are emitted with the legacy conf off and omitted with it on.
| * Reports the table's declared write distribution and ordering, whether or not SHOW CREATE TABLE | ||
| * can reproduce them. | ||
| */ | ||
| private def addTableWriteDistributionAndOrdering(rows: ArrayBuffer[InternalRow]): Unit = { |
There was a problem hiding this comment.
[42] Each accessor is called up to three times.
table.writeDistributionMode() and table.writeOrdering() are each called once for isRequested, once for the null/empty check and once for the row. A connector may build these on every call, for example by converting its own sort order into a new SortOrder[] from refreshed metadata. That repeats the work, and the rows can come from different snapshots than the isRequested check. Reading both into local vals once at the top would avoid both. ShowCreateTableExec.showTableWriteDistributionAndOrdering already does this for writeOrdering.
There was a problem hiding this comment.
Done. addTableWriteDistributionAndOrdering now reads both into vals once. I also checked the rest of the diff for repeated connector accessor calls; ShowCreateTableExec already reads each one once.
…g in CREATE/REPLACE TABLE
88882a2 to
8338ca9
Compare
anuragmantri
left a comment
anuragmantri
left a comment
There was a problem hiding this comment.
Thank you, @dongjoon-hyun. Points 40-42 are addressed in 8338ca9, with replies on each thread. I rebased the branch on master to pick up SPARK-59983.
On SPARK-60075: it is not a duplicate. After SPARK-59983, a connector PartitionPredicate that overrides neither toString nor describe still recurses, because visitPartitionPredicate returns describe(), which defaults to toString(), which builds through the same builder again. I re-scoped SPARK-60075 to that case.
| e.children().map(toSQL(_, quote)).mkString(s"$name(", ", ", ")") | ||
| } | ||
|
|
||
| // `ToStringSQLBuilder` renders a few expressions through their children's `describe`, and for an |
There was a problem hiding this comment.
Done. After rebasing on master, the comment gives only the reasons that still hold. DescribingSQLBuilder renders literals and references with toSQL, so they keep their type and quoting, and renders the children of GetArrayItem and VariantGet itself rather than through their toString. It renders an expression ToStringSQLBuilder does not know, and a PartitionPredicate, from its children. The last part is still needed on master: a connector PartitionPredicate that does not override describe still recurses. I rescoped SPARK-60075 to cover this.
|
|
||
| // In the session catalog, a CREATE TABLE whose provider is not a v2 source creates a v1 table, | ||
| // which cannot record the clauses, so the replay would be rejected. | ||
| private def createsV1Table(resolvedTable: ResolvedTable): Boolean = { |
There was a problem hiding this comment.
Agreed. The copy had in fact already drifted. For a table with no provider it used spark.sql.sources.default, while CREATE TABLE creates a Hive serde table when spark.sql.legacy.createHiveTableByDefault is set. supportsV1Command, isV2Provider, createsHiveTableByDefault and createTableProvider now live in DataSourceV2Utils, and both ResolveSessionCatalog and ShowCreateTableExec call them; ResolveSessionCatalog's behavior is unchanged. A new test covers a provider-less table in a session catalog extension with spark.sql.sources.default set to a v2 source: the clauses are emitted with the legacy conf off and omitted with it on.
| * Reports the table's declared write distribution and ordering, whether or not SHOW CREATE TABLE | ||
| * can reproduce them. | ||
| */ | ||
| private def addTableWriteDistributionAndOrdering(rows: ArrayBuffer[InternalRow]): Unit = { |
There was a problem hiding this comment.
Done. addTableWriteDistributionAndOrdering now reads both into vals once. I also checked the rest of the diff for repeated connector accessor calls; ShowCreateTableExec already reads each one once.
This branch has not been deployed
4 participants
This PR is based on the previous works of @aokolnychyi and @RussellSpitzer. Both are co-authors on the commit.
What changes were proposed in this pull request?
Lets
CREATE/REPLACE TABLE, and theirAS SELECTforms, carry the write distribution and sort order the table should be written with:The request travels to the catalog on
TableInfo, via two new builder methods (withWriteDistributionMode,withWriteOrdering) and two accessors. A plain column inORDERED BYreaches the catalog as aNamedReference, any other key as aTransform, and the ordering is never null.TablegainswriteDistributionMode()/writeOrdering()so a catalog can report back what a table declares,DelegatingTableforwards both, andSHOW CREATE TABLE/DESCRIBE TABLE EXTENDEDread them.A catalog has to advertise the new
TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING, otherwise the statement fails withUNSUPPORTED_FEATURE.TABLE_OPERATIONbefore anything is created: during planning for a v2 catalog, and during analysis when the session catalog would create a v1 table. The request reaches the catalog only through the fourTableInfooverloads (createTableand the threeStagingTableCatalog.stage*), so a catalog that reports the capability must override each one it can be reached through.DelegatingCatalogExtensiondoes not forward this capability from its delegate, because it does not forward those overloads.Layers:
AstBuilder(which also rejectsDISTRIBUTED BY PARTITIONon an unpartitioned table),CreateTable/ReplaceTable(AsSelect) plans,PreprocessTableCreation(normalizes the ordering) andCheckAnalysis(rejects unknown columns)TableInfo, andDataSourceV2Strategy, which checks the capabilityA sort key has to resolve against the table's columns, so
ORDERED BYneeds a statement that defines a schema (a column list, typed partition columns, orAS SELECT); on a v2 catalog a schemaless statement is rejected withSPECIFY_WRITE_ORDERING_IS_NOT_ALLOWED, the positionPARTITIONED BYalready takes there.DISTRIBUTED BY PARTITIONneeds the statement to declare a partitioning (SPECIFY_DISTRIBUTED_BY_PARTITION_WITHOUT_PARTITIONING_IS_NOT_ALLOWED) and cannot be combined withCLUSTER BY(SPECIFY_CLUSTER_BY_WITH_DISTRIBUTED_BY_PARTITION_IS_NOT_ALLOWED); acluster_by(...)transform inPARTITIONED BYcounts as clustering, not partitioning.writeDistributionModeis a newWriteDistributionModeenum (HASH/RANGE/NONE).nullmeans the statement said nothing, which is distinct fromNONE.nullleaves the choice to the catalog's own default;NONEis an explicit request not to distribute. See the design decisions below.Why are the changes needed?
Spark can already enforce a write layout: a connector reports one from
RequiresDistributionAndOrderingon itsWrite, andDistributionAndOrderingUtilsinserts the shuffle and sort. What is missing is a way for the user to author it, so it is persisted as table metadata and honored by the first write and by every later one, including writes from other engines.Without it, a user who wants a sorted table has to run three statements:
CREATE, then a connector-specificALTER TABLE, thenINSERT. That reaches the same physical layout but is not atomic: the table is visible unsorted in between, a concurrent writer can land unsorted data, and forREPLACE ... AS SELECTthe table is temporarily empty. Tools that generate CTAS have no hook between create and load, so for them the first load can never be sorted.Iceberg's community has asked for this in the engine's DDL twice apache/iceberg#3547 and apache/iceberg#14612
Does this PR introduce any user-facing change?
Yes, additive optional clauses on
CREATE/REPLACE TABLE, and four new keywords (DISTRIBUTED,LOCALLY,ORDERED,UNORDERED), all non-reserved in every mode, so existing identifiers with those names keep working. A statement that does not use the clauses builds the sameTableInfoas before.CREATE TEMPORARY TABLE ... USING,CREATE MATERIALIZED VIEWandCREATE STREAMING TABLEreject the clauses withINVALID_STATEMENT_OR_CLAUSE.New error conditions:
WRITE_ORDERING_WITH_UNKNOWN_COLUMN(42703),SPECIFY_WRITE_ORDERING_IS_NOT_ALLOWED(42601),SPECIFY_DISTRIBUTED_BY_PARTITION_WITHOUT_PARTITIONING_IS_NOT_ALLOWED(42908) andSPECIFY_CLUSTER_BY_WITH_DISTRIBUTED_BY_PARTITION_IS_NOT_ALLOWED(42908). The JDBCgetSQLKeywordsand ThriftCLI_ODBC_KEYWORDSlists include the four new keywords.Source compatibility for extensions and connectors:
CreateTable,ReplaceTable,CreateTableAsSelectandReplaceTableAsSelectplans need to add the two new fields (writeDistributionMode,writeOrdering) to their patterns. The fields are defaulted, so constructing the plans by name still compiles, but extensions compiled against an earlier release must be recompiled. Moving the pair ontoTableSpecto restore the signatures is SPARK-59943.CreateTableAsSelectExecandReplaceTableAsSelectExecthey come before the defaultedtransaction.V2CreateTablePlanneed to add its two new abstract members (writeOrdering,withWriteOrdering).DelegatingCatalogExtension.capabilities()now returns the delegate's capabilities withoutSUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING. A subclass that overrides theTableInfooverloads adds it back by overridingcapabilities().TableInforejects a null write ordering at construction.Design decisions
Four choices here are worth spelling out, with what they cost.
1. A mode enum rather than a typed
Distribution. Spark already hasDistributions.unspecified()/clustered(...)/ordered(...), used byRequiresDistributionAndOrdering. But a mode is a policy for all future writes while aDistributiondescribes one write, so a typedDistributionwould have to freeze the partition expressions at create time. The table therefore declares aWriteDistributionMode(HASH/RANGE/NONE), a closed set likeSortDirectionandNullOrdering, which maps directly to connector settings such as Iceberg'swrite.distribution-mode.TableInfois a builder, so a typedwithDistribution(Distribution)can be added later without breaking callers.2.
Tableexposes what the table declares, and both display paths read it.Table.writeDistributionMode()/writeOrdering()default tonull/ empty, mirroringTable.constraints(). The contract, documented on the methods, is that these are a declared default for future writes and nothing more: an individual write may override it, a connector may narrow it (a hash distribution is meaningless on an unpartitioned table),RequiresDistributionAndOrderingon theWritestays authoritative for what a given write actually requires, and none of it claims anything about how the data already in the table is laid out. A scan reports that itself, viaoutputPartitioning/outputOrdering.SHOW CREATE TABLEreproduces the clauses from them andDESCRIBE TABLE EXTENDEDreports them, which is what makes a table created with these clauses recreatable. Iceberg, for example, downgrades a requestedhashtononeon an unpartitioned table at write time, so a consumer of these accessors must not read them as what the next write will do. There is also noALTER TABLEversion yet. A connector that has one of its own (Iceberg'sALTER TABLE ... WRITE) is unaffected, but changing the declared default through Spark is follow-up work.A connector can report more. For example,
hashon a table with no partitioning (the parser rejectsDISTRIBUTED BY PARTITIONthere), arangedistribution with no ordering, or a sort key the syntax cannot spell.SHOW CREATE TABLEtherefore emits the clauses only when the catalog accepts them, the statement would not create a v1 table (the session catalog with a provider that is not a v2 source, decided by the same rule CREATE TABLE uses), and parsing the rendered clauses with the session's parser gives back the same mode and keys, with every column they reference in the table. It retries once with every name quoted, for reserved keywords. Otherwise it omits the pair rather than emitting a clause that would mean something else or would not parse. Literals keep their type: aFLOATrenders as<v>F, aTIMESTAMPor nanosecondTIMESTAMP_LTZin UTC with an explicit offset, and aTIMEwith its precision.DESCRIBE TABLE EXTENDEDprints both values in every case, renders connector expressions as theirdescribedoes, and does not fail on values Catalyst cannot represent or on expressions Spark's SQL builder does not know.CREATE TABLE ... LIKEdeliberately does not copy the declared layout, and is not gated on the capability, because it is not aV2CreateTablePlan, so gating it would mean extending the check to a fifth plan. It does hand the sourceTableto the connector, which can carry the layout across itself, as theTableCatalog.createTableLikeJavadoc says. Happy to foldLIKEin if reviewers would rather have it here.3.
ORDERED BYimplies a distribution, and the mode is what says how far the order reaches.ORDERED BY (...)rangeLOCALLY ORDERED BY (...)noneUNORDEREDnoneDISTRIBUTED BY PARTITIONhashDISTRIBUTED BY PARTITION [LOCALLY] ORDERED BY (...)hashDISTRIBUTED BY PARTITION UNORDEREDhashThere is no separate global-vs-local flag on the recorded pair, because the distribution already is one:
rangemeans the order holds across the table,hashandnonemean it holds within a write task. A bareORDERED BYhas to implyrange: sorting within each task does not make the table sorted, so on its own it would record an order the writes cannot achieve. That is also whyLOCALLYis the escape hatch for the within-task case, the same word Iceberg'sALTER TABLE ... WRITE LOCALLY ORDERED BYuses.It also means the implication only applies when
DISTRIBUTED BY PARTITIONis absent. Beside it the distribution is already fixed and already local, soLOCALLYadds nothing andUNORDEREDcontributes only "no sort keys". Both are accepted and both produce the same pair as leaving them out. Spark's usual answer for a clause with no effect in a combination is to reject it (seeSPECIFY_CLUSTER_BY_WITH_PARTITIONED_BY_IS_NOT_ALLOWED), and that would be defensible here too; they are accepted because both spellings are already valid in Iceberg'sALTER TABLE ... WRITE, and a word that means the same thing onCREATE TABLEas it does onALTER TABLEseems worth more than the extra strictness.The cost of the coupling: there is no way to say "range-distribute but record no ordering", nor "record this ordering and leave the distribution unset". Decoupling needs more syntax; There is no spelling for a range distribution on its own (
DISTRIBUTED BY PARTITIONis the hash one), so one would have to be invented, and the common cases would then take two clauses instead of one.4.
CLUSTER BYis independent, and it is notCLUSTERED BY ... INTO ... BUCKETS.DISTRIBUTED BY PARTITIONrequires the table to be partitioned, byPARTITIONED BYor byCLUSTERED BY ... INTO n BUCKETS(bucketing is a partition transform).CLUSTER BY (...)is a different clause: it records clustering columns for the data source to interpret, it is not a partition transform, and the grammar already forbids combining it withPARTITIONED BYorCLUSTERED BY ... INTO ... BUCKETS. So a table usingCLUSTER BYhas no partitioning at all, andDISTRIBUTED BY PARTITIONon it is unsatisfiable by construction rather than by a policy choice here. The cost:CLUSTER BYusers cannot ask for a per-partition write distribution without moving toPARTITIONED BY. IfCLUSTER BYshould count as a partitioning for this, that is a change toCLUSTER BYand belongs in its own PR.CLUSTER BYcan still be combined withORDERED BY,LOCALLY ORDERED BYorUNORDERED: Spark passes the clustering columns and the declared distribution and ordering to the catalog without reconciling them, so the catalog decides how they interact and may reject a combination it does not support.DISTRIBUTED BY PARTITIONon aCLUSTER BYtable gets its own error,SPECIFY_CLUSTER_BY_WITH_DISTRIBUTED_BY_PARTITION_IS_NOT_ALLOWED, which says to replaceCLUSTER BYwithPARTITIONED BYor bucketing. The parser andSHOW CREATE TABLEshare one rule for what counts as partitioning: any transform not namedcluster_by.One pre-existing hole to be aware of:
TableCatalog.createTable(ident, TableInfo)'s default implementation forwards to the deprecated 4-argcreateTable(ident, columns, partitions, properties), so everyTableInfo-only field is dropped for a catalog that implements only that overload.constraintsalready is, today. The capability check is what keeps that from becoming a silent wrong result here: without the capability the statement fails, so a catalog that never looks atTableInfocannot quietly create a table lacking the requested layout. Fixing the default itself is out of scope.How was this patch tested?
CreateTableWriteOrderSuite(48 tests): parsing, analysis, what reaches the catalog (createTableand thestage*methods), the capability checks (includingDelegatingCatalogExtension), a catalog rejecting a combination, the first write of a CTAS/RTAS planning the declared shuffle and sort,SHOW CREATE TABLE/DESCRIBE TABLE EXTENDED, and every new error condition withcheckErrorand its SQLSTATE.WriteDistributionAndOrderingUtilsSuite(14 tests): table-driven, over connector-reported sort keys of every literal type and shape, assertingSHOW CREATE TABLEemits a key exactly when parsing it back gives the same key with every column it references in the schema, including under the parser confs that change literals and keywords; each mode's clause form; and DESCRIBE's rendering of connector expressions.PlanResolutionSuiteover the resolved CREATE/CTAS/REPLACE/RTAS plans, and one inCreatePipelineDatasetAsSelectParserSuiteBase.keywords*.sql.out,SparkConnectDatabaseMetaDataSuiteandThriftServerWithSparkContextSuite.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Co-authored-by: Peter Toth peter.toth@gmail.com
Co-authored-by: Anton Okolnychyi aokolnychyi@apache.org
Co-authored-by: Russell Spitzer russell.spitzer@gmail.com