anuragmantri commented on code in PR #58153:
URL: https://github.com/apache/spark/pull/58153#discussion_r3834580580
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowCreateTableExec.scala:
##########
@@ -125,6 +126,45 @@ case class ShowCreateTableExec(
}
}
+ /**
+ * Emits the write distribution and ordering the table declares as the
default for writes into it,
+ * so that a table created with those clauses can be recreated from this
statement.
+ *
+ * The pair a connector may report is wider than the syntax can spell:
`hash` on a table with no
+ * partitioning (the parser rejects `DISTRIBUTED BY PARTITION` there), a
`range` distribution with
+ * no ordering, an ordering with no distribution, and a mode this Spark
version does not know.
+ * Those are left out rather than guessed at, since emitting a clause that
means something else --
+ * or one that does not parse at all -- would be worse than emitting none.
DESCRIBE TABLE EXTENDED
+ * reports both values verbatim regardless.
+ */
+ private def showTableWriteDistributionAndOrdering(
+ table: Table,
+ builder: StringBuilder): Unit = {
+ val orderBy = if (table.writeOrdering().nonEmpty) {
+ Some(table.writeOrdering()
+ .map(WriteDistributionAndOrdering.describeSortOrder)
+ .mkString("ORDERED BY (", ", ", ")"))
Review Comment:
Done. I added `isSpellable()` and wrapped the rest with it.
##########
sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala:
##########
@@ -214,7 +214,7 @@ trait ThriftServerWithSparkContextSuite extends
SharedThriftServer {
val sessionHandle = client.openSession(user, "")
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,DIST
INCT,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,M
SCK,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,TAB
LESAMPLE,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,DIST
INCT,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,MOD
IFIES,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,SYS
TEM_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")
Review Comment:
Done.
##########
docs/sql-ref-syntax-ddl-create-table-datasource.md:
##########
@@ -98,6 +106,60 @@ as any order. For example, you can write COMMENT
table_comment after TBLPROPERTI
A list of key-value pairs that is used to tag the table definition.
+* **DISTRIBUTED BY PARTITION**
+
+ Requests that every write to the table be clustered by the table's
partitioning, so each
+ partition is written by a single task rather than by every task that holds
rows for it.
+
+ Requires the table to actually be partitioned, with `PARTITIONED BY` or
with
+ `CLUSTERED BY ... INTO ... BUCKETS`. Note that `CLUSTER BY` -- a different
clause from
+ `CLUSTERED BY ... INTO ... BUCKETS` -- does **not** qualify: it lists
clustering columns for the
+ data source to interpret rather than defining a partitioning, and it
cannot be combined with
+ `PARTITIONED BY` or `CLUSTERED BY ... INTO ... BUCKETS`, so a table using
it has no partitioning
+ to distribute by.
+
+* **ORDERED BY**
+
+ Requests a sort order for every write to the table, recorded on the table
so that later writes
+ honor it too. `UNORDERED` asks for no ordering at all, which is different
from omitting the
+ clause -- omitting it leaves the choice to the data source. The
parentheses are optional:
+ `ORDERED BY (a, b)` and `ORDERED BY a, b` are the same. The sort keys must
resolve against the
+ table's columns, so a `CREATE TABLE` with neither a column list nor `AS
SELECT` cannot use this
+ 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
+ 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.
+
+ When `DISTRIBUTED BY PARTITION` is given it decides the distribution
instead, and the order then
+ holds within each write task. `LOCALLY` therefore adds nothing beside it,
and `UNORDERED` beside
+ it contributes only "no sort keys":
+
+ ```sql
+ -- range-partition each write by id, so the order holds across the whole
table
+ CREATE TABLE t (id INT, c STRING) USING iceberg PARTITIONED BY (c) ORDERED
BY (id);
+
+ -- cluster each write by partition instead, and sort by id within each task
+ CREATE TABLE t (id INT, c STRING) USING iceberg PARTITIONED BY (c)
+ DISTRIBUTED BY PARTITION ORDERED BY (id);
+
+ -- cluster each write by partition, with no sort order; UNORDERED here is
the same as
+ -- omitting it, since DISTRIBUTED BY PARTITION already fixed the
distribution
+ CREATE TABLE t (id INT, c STRING) USING iceberg PARTITIONED BY (c)
+ DISTRIBUTED BY PARTITION UNORDERED;
+ ```
+
+ Both clauses are passed to the data source, which has to support them: a
data source that does
Review Comment:
Done.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -9137,6 +9151,11 @@
"message" : [
"Write for the binary file data source."
]
+ },
+ "WRITE_ORDERING_WITH_NESTED_COLUMN_IS_UNSUPPORTED" : {
Review Comment:
Changed it to `WRITE_ORDERING_WITH_UNKNOWN_COLUMN`.
##########
docs/sql-ref-syntax-ddl-create-table-datasource.md:
##########
@@ -98,6 +106,60 @@ as any order. For example, you can write COMMENT
table_comment after TBLPROPERTI
A list of key-value pairs that is used to tag the table definition.
+* **DISTRIBUTED BY PARTITION**
+
+ Requests that every write to the table be clustered by the table's
partitioning, so each
+ partition is written by a single task rather than by every task that holds
rows for it.
+
+ Requires the table to actually be partitioned, with `PARTITIONED BY` or
with
+ `CLUSTERED BY ... INTO ... BUCKETS`. Note that `CLUSTER BY` -- a different
clause from
+ `CLUSTERED BY ... INTO ... BUCKETS` -- does **not** qualify: it lists
clustering columns for the
+ data source to interpret rather than defining a partitioning, and it
cannot be combined with
+ `PARTITIONED BY` or `CLUSTERED BY ... INTO ... BUCKETS`, so a table using
it has no partitioning
+ to distribute by.
+
+* **ORDERED BY**
+
+ Requests a sort order for every write to the table, recorded on the table
so that later writes
+ honor it too. `UNORDERED` asks for no ordering at all, which is different
from omitting the
+ clause -- omitting it leaves the choice to the data source. The
parentheses are optional:
+ `ORDERED BY (a, b)` and `ORDERED BY a, b` are the same. The sort keys must
resolve against the
+ table's columns, so a `CREATE TABLE` with neither a column list nor `AS
SELECT` cannot use this
+ 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
+ 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.
+
+ When `DISTRIBUTED BY PARTITION` is given it decides the distribution
instead, and the order then
+ holds within each write task. `LOCALLY` therefore adds nothing beside it,
and `UNORDERED` beside
+ it contributes only "no sort keys":
+
+ ```sql
+ -- range-partition each write by id, so the order holds across the whole
table
+ CREATE TABLE t (id INT, c STRING) USING iceberg PARTITIONED BY (c) ORDERED
BY (id);
+
+ -- cluster each write by partition instead, and sort by id within each task
+ CREATE TABLE t (id INT, c STRING) USING iceberg PARTITIONED BY (c)
+ DISTRIBUTED BY PARTITION ORDERED BY (id);
+
+ -- cluster each write by partition, with no sort order; UNORDERED here is
the same as
+ -- omitting it, since DISTRIBUTED BY PARTITION already fixed the
distribution
+ CREATE TABLE t (id INT, c STRING) USING iceberg PARTITIONED BY (c)
+ DISTRIBUTED BY PARTITION UNORDERED;
+ ```
+
+ Both clauses are passed to the data source, which has to support them: a
data source 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
data sources do not
+ support them.
+
+ 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
Review Comment:
Done.
##########
docs/sql-ref-syntax-ddl-create-table-datasource.md:
##########
@@ -98,6 +106,60 @@ as any order. For example, you can write COMMENT
table_comment after TBLPROPERTI
A list of key-value pairs that is used to tag the table definition.
+* **DISTRIBUTED BY PARTITION**
+
+ Requests that every write to the table be clustered by the table's
partitioning, so each
+ partition is written by a single task rather than by every task that holds
rows for it.
+
+ Requires the table to actually be partitioned, with `PARTITIONED BY` or
with
+ `CLUSTERED BY ... INTO ... BUCKETS`. Note that `CLUSTER BY` -- a different
clause from
+ `CLUSTERED BY ... INTO ... BUCKETS` -- does **not** qualify: it lists
clustering columns for the
+ data source to interpret rather than defining a partitioning, and it
cannot be combined with
+ `PARTITIONED BY` or `CLUSTERED BY ... INTO ... BUCKETS`, so a table using
it has no partitioning
+ to distribute by.
+
+* **ORDERED BY**
+
+ Requests a sort order for every write to the table, recorded on the table
so that later writes
+ honor it too. `UNORDERED` asks for no ordering at all, which is different
from omitting the
+ clause -- omitting it leaves the choice to the data source. The
parentheses are optional:
+ `ORDERED BY (a, b)` and `ORDERED BY a, b` are the same. The sort keys must
resolve against the
+ table's columns, so a `CREATE TABLE` with neither a column list nor `AS
SELECT` cannot use this
+ 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
+ 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.
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
