anuragmantri opened a new pull request, #58153:
URL: https://github.com/apache/spark/pull/58153
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 their `AS SELECT` forms, carry the write
distribution and sort order the table should be written with:
[DISTRIBUTED BY PARTITION]
[[LOCALLY] ORDERED BY (transform(col) [ASC|DESC] [NULLS {FIRST|LAST}],
...) | UNORDERED]
The request travels to the catalog on `TableInfo`, via two new builder
methods (`withWriteDistributionMode`, `withWriteOrdering`) and two accessors.
`Table` gains `writeDistributionMode()` / `writeOrdering()` so a catalog can
report back what a table declares, `DelegatingTable` forwards both, and `SHOW
CREATE TABLE` / `DESCRIBE TABLE EXTENDED` read them.
A catalog has to advertise the new
`TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_WRITE_DISTRIBUTION_AND_ORDERING`,
otherwise the statement fails with `UNSUPPORTED_FEATURE.TABLE_OPERATION` while
it is planned.
Layers:
- Grammar (4 new non-reserved keywords),
- `AstBuilder` (which also rejects `DISTRIBUTED BY PARTITION` on an
unpartitioned table),
- Four `CreateTable`/`ReplaceTable`(`AsSelect`) plans
A sort key has to resolve against the table's columns, so `ORDERED BY` needs
either a column list or `AS SELECT`; on a schemaless `CREATE TABLE` it is
rejected with `SPECIFY_WRITE_ORDERING_IS_NOT_ALLOWED`, the position
`PARTITIONED BY` already takes there.
`writeDistributionMode` is a `String` with three
values(`DISTRIBUTION_MODE_HASH`/`_RANGE`/`_NONE`). `null` means the statement
said nothing, which is distinct from `none`. `null` leaves the choice to the
catalog's own default; `none` is 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
`RequiresDistributionAndOrdering` on its `Write`, and
`DistributionAndOrderingUtils` inserts 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-specific `ALTER TABLE`, then `INSERT`. 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 for `REPLACE ... AS
SELECT` the 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
https://github.com/apache/iceberg/issues/3547 and
https://github.com/apache/iceberg/issues/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 same `TableInfo` as before.
### Design decisions
Four choices here are worth spelling out, with what they cost.
**1. A `String` mode rather than a typed `Distribution`.** Spark already has
`Distributions.unspecified()` / `clustered(...)` / `ordered(...)`, used by
`RequiresDistributionAndOrdering`, and a typed value would remove the string
vocabulary. But a mode is a *policy* for all future writes while a
`Distribution` describes one write. For example, Iceberg's
`write.distribution-mode` takes exactly `hash`/`range`/`none` and Spark only
ever produces one of the three. `TableInfo` is a builder, so a typed
`withDistribution(Distribution)` can be added later without breaking callers.
**2. `Table` exposes what the table declares, and both display paths read
it.** `Table.writeDistributionMode()` / `writeOrdering()` default to `null` /
empty, mirroring `Table.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),
`RequiresDistributionAndOrdering` on the `Write` stays 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, via
`outputPartitioning`/`outputOrdering`. `SHOW CREATE TABLE` reproduces the
clauses from them and `DESCRIBE TABLE EXTENDED` reports them, which is what
makes a table created with these clauses recreatable. Iceberg, for example,
downgrades a requested `hash` to `none` on an unpartitioned table at write
time, so a consumer of these acc
essors must not read them as what the next write will do. There is also no
`ALTER TABLE` version yet. A connector that has one of its own (Iceberg's
`ALTER TABLE ... WRITE`) is unaffected, but changing the declared default
through Spark is follow-up work.
A connector can potentially can report more. For example, `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, or a
mode this Spark version does not know. `SHOW CREATE TABLE` omits those rather
than emitting a clause that would mean something else or would not parse;
`DESCRIBE TABLE EXTENDED` prints both values verbatim, so nothing is hidden.
There is a test for it.
`CREATE TABLE ... LIKE` deliberately does not copy the declared layout, and
is not gated on the capability, because it is not a `V2CreateTablePlan`, so
gating it would mean extending the check to a fifth plan. It does hand the
source `Table` to the connector, which can carry the layout across itself, the
way `CreateTableLikeExec`'s own doc already anticipates for Iceberg sort order.
Happy to fold `LIKE` in if reviewers would rather have it here.
**3. `ORDERED BY` implies a distribution, and the mode is what says how far
the order reaches.**
| clause | distribution | ordering |
|---|---|---|
| *(none)* | unset, catalog's own default | none |
| `ORDERED BY (...)` | `range` | as written |
| `LOCALLY ORDERED BY (...)` | `none` | as written |
| `UNORDERED` | `none` | none |
| `DISTRIBUTED BY PARTITION` | `hash` | none |
| `DISTRIBUTED BY PARTITION [LOCALLY] ORDERED BY (...)` | `hash` | as
written |
| `DISTRIBUTED BY PARTITION UNORDERED` | `hash` | none |
There is no separate global-vs-local flag on the recorded pair, because the
distribution already is one: `range` means the order holds across the table,
`hash` and `none` mean it holds within a write task. A bare `ORDERED BY` has to
imply `range`: 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 why
`LOCALLY` is the escape hatch for the within-task case, the same word Iceberg's
`ALTER TABLE ... WRITE LOCALLY ORDERED BY` uses.
It also means the implication only applies when `DISTRIBUTED BY PARTITION`
is absent. Beside it the distribution is already fixed and already local, so
`LOCALLY` adds nothing and `UNORDERED` contributes 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 (see
`SPECIFY_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's `ALTER TABLE ... WRITE`, and a word that means the same thing on
`CREATE TABLE` as it does on `ALTER TABLE` seems 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 PARTITION` is the hash one), so one
would have to be invented, and the common cases would then take two clauses
instead of one.
**4. `CLUSTER BY` is independent, and it is not `CLUSTERED BY ... INTO ...
BUCKETS`.** `DISTRIBUTED BY PARTITION` requires the table to be partitioned, by
`PARTITIONED BY` or by `CLUSTERED 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 with `PARTITIONED BY`
or `CLUSTERED BY ... INTO ... BUCKETS`. So a table using `CLUSTER BY` has no
partitioning at all, and `DISTRIBUTED BY PARTITION` on it is unsatisfiable by
construction rather than by a policy choice here. The cost: `CLUSTER BY` users
cannot ask for a per-partition write distribution without moving to
`PARTITIONED BY`. If `CLUSTER BY` should count as a partitioning for this, that
is a change to `CLUSTER BY` and belongs in its own PR.
One pre-existing hole to be aware of: `TableCatalog.createTable(ident,
TableInfo)`'s default implementation forwards to the deprecated 4-arg
`createTable(ident, columns, partitions, properties)`, so *every*
`TableInfo`-only field is dropped for a catalog that implements only that
overload. `constraints` already 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 at `TableInfo` cannot quietly
create a table lacking the requested layout. Fixing the default itself is out
of scope.
### How was this patch tested?
- New `CreateTableWriteOrderSuite` (30 tests)
- Six more cases in `PlanResolutionSuite` covering the resolved plans, and
one in `CreatePipelineDatasetAsSelectParserSuiteBase`
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Co-authored-by: Peter Toth <[email protected]>
Co-authored-by: Anton Okolnychyi <[email protected]>
Co-authored-by: Russell Spitzer <[email protected]>
--
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]
