feat: ddl statements + catalog

docs/configuration/index.md

@@ -1896,6 +1896,7 @@ The Druid SQL server is configured through the following properties on the Broke
 |`druid.sql.planner.metadataSegmentCacheEnable`|Whether to keep a cache of published segments on Broker that can be used to serve queries against `sys.segments`. If true, broker polls coordinator in background to get segments from metadata store and maintains a local cache. If false, coordinator's REST API will be invoked when broker needs published segments info.|true|
 |`druid.sql.planner.metadataSegmentPollPeriod`|How often to poll coordinator for published segments list if `druid.sql.planner.metadataSegmentCacheEnable` is set to true. Poll period is in milliseconds. |60000|
 |`druid.sql.planner.authorizeSystemTablesDirectly`|If true, Druid authorizes queries against any of the system schema tables (`sys` in SQL) as `SYSTEM_TABLE` resources which require `READ` access, in addition to permissions based content filtering.|false|
+|`druid.sql.planner.enableCatalogDdl`|If true, `CREATE TABLE` and `ALTER TABLE` statements may be used to define [catalog](../development/extensions-core/catalog.md) tables. These statements require `WRITE` permission on the datasource, the same permission the catalog API requires, so enabling this lets anyone who can ingest into a datasource also change its catalog definition. Requires the `druid-catalog` extension. Cannot be overridden per query.|false|
 |`druid.sql.planner.useNativeQueryExplain`|If true, `EXPLAIN PLAN FOR` will return the explain plan as a JSON representation of equivalent native query(s), else it will return the original version of explain plan generated by Calcite. It can be overridden per query with `useNativeQueryExplain` context key.|true|
 |`druid.sql.planner.maxNumericInFilters`|Max limit for the amount of numeric values that can be compared for a string type dimension when the entire SQL WHERE clause of a query translates to an [OR](../querying/filters.md#or) of [Bound filter](../querying/filters.md#bound-filter). By default, Druid does not restrict the amount of numeric Bound Filters on String columns, although this situation may block other queries from running. Set this property to a smaller value to prevent Druid from running queries that have prohibitively long segment processing times. The optimal limit requires some trial and error; we recommend starting with 100.  Users who submit a query that exceeds the limit of `maxNumericInFilters` should instead rewrite their queries to use strings in the `WHERE` clause instead of numbers. For example, `WHERE someString IN (‘123’, ‘456’)`. If this value is disabled, `maxNumericInFilters` set through query context is ignored.|`-1` (disabled)|
 |`druid.sql.approxCountDistinct.function`|Implementation to use for the [`APPROX_COUNT_DISTINCT` function](../querying/sql-aggregations.md). Without extensions loaded, the only valid value is `APPROX_COUNT_DISTINCT_BUILTIN` (a HyperLogLog, or HLL, based implementation). If the [DataSketches extension](../development/extensions-core/datasketches-extension.md) is loaded, this can also be `APPROX_COUNT_DISTINCT_DS_HLL` (alternative HLL implementation) or `APPROX_COUNT_DISTINCT_DS_THETA`.<br /><br />Theta sketches use significantly more memory than HLL sketches, so you should prefer one of the two HLL implementations.|`APPROX_COUNT_DISTINCT_BUILTIN`|

docs/development/extensions-core/catalog.md

@@ -43,6 +43,165 @@ allowing queries to be more concise, and simpler to write. This also allows the
 written into a defined column of the table is consistent with that columns definition, minimizing errors where unexpected
 data is written into a particular column of the table.
 
+### SQL DDL
+
+Tables can be defined with SQL instead of by posting a table specification. `CREATE TABLE` and `ALTER TABLE` are
+submitted to the Broker like any other SQL statement, and write the same catalog metadata the REST API does. They
+return no rows.
+
+These statements change catalog metadata only. They never create, modify, or delete segments: defining a table does
+not ingest anything, and altering a column does not rewrite existing data. Column changes take effect for subsequent
+ingestion.
+
+These statements are disabled by default. Set `druid.sql.planner.enableCatalogDdl` to `true` on the Broker to enable
+them. They require `WRITE` permission on the datasource, the same permission the catalog API requires, so enabling
+them lets anyone who can ingest into a datasource also change its catalog definition; leave them disabled if you
+manage catalog entries with your own tooling. The setting cannot be overridden per query.
+
+The `druid-catalog` extension must be loaded on both the Broker and the Coordinator; without it, these statements
+report that the extension is not available.
+
+```sql
+CREATE [OR REPLACE] TABLE [IF NOT EXISTS] <table>
+  [ ( { <column> <type> | PROJECTION <name> AS ( <select> ) } [, ...] ) ]
+  [ PARTITIONED BY <granularity> ]
+  [ CLUSTERED BY <column> [, ...] ]
+  [ SEALED ]
+```
+
+`OR REPLACE` replaces the specification of an existing table; `IF NOT EXISTS` leaves an existing table unchanged.
+The two cannot be combined. `PARTITIONED BY` sets [`segmentGranularity`](#table-properties) and `CLUSTERED BY` sets
+`clusterKeys`, both of which a later `INSERT` or `REPLACE` inherits unless it states its own. `SEALED` sets
+[`sealed`](#table-properties), which requires every ingested column to be declared.
+
+Note that the table-level `CLUSTERED BY` is a sort order applied to each ingestion, which is a different thing from
+the `CLUSTERED BY` inside a [`__base` projection](#the-base-table), which defines how segments physically group rows.
+
+Column types are written as SQL types, such as `VARCHAR`, `BIGINT`, `DOUBLE`, or `VARCHAR ARRAY`. The `__time` column
+is written as `TIMESTAMP`. Types that have no SQL spelling, such as complex types, use `TYPE('...')` with the Druid
+native type string:
+
+```sql
+CREATE TABLE "druid"."visits" (
+  __time TIMESTAMP,
+  user_id VARCHAR,
+  pages_visited BIGINT,
+  sketch TYPE('COMPLEX<thetaSketch>')
+)
+PARTITIONED BY DAY
+CLUSTERED BY user_id
+```
+
+`ALTER TABLE` supports one change per statement, so that each statement is a single atomic catalog operation:
+
+```sql
+ALTER TABLE <table> ADD COLUMN <column> <type>
+ALTER TABLE <table> DROP COLUMN <column>
+ALTER TABLE <table> ALTER COLUMN <column> SET DATA TYPE <type>
+ALTER TABLE <table> ADD [IF NOT EXISTS] PROJECTION <name> AS ( <select> )
+ALTER TABLE <table> DROP PROJECTION [IF EXISTS] <name>
+ALTER TABLE <table> SET PROPERTIES ( <property> = <value> [, ...] )
+```
+
+#### Projections
+
+A table may declare [projections](../../querying/projections.md), which are pre-aggregated views stored inside each
+segment. A projection is written as a `SELECT` over the table's own columns, with no `FROM` clause:
+
+```sql
+CREATE TABLE "druid"."visits" (
+  __time TIMESTAMP,
+  user_id VARCHAR,
+  user_agent VARCHAR,
+  pages_visited BIGINT,
+  PROJECTION daily_by_agent AS (
+    SELECT TIME_FLOOR(__time, 'P1D'), user_agent, SUM(pages_visited) AS total_pages
+    WHERE user_agent IS NOT NULL
+    GROUP BY 1, 2
+  )
+)
+PARTITIONED BY DAY
+```
+
+The body is planned exactly as the equivalent query would be, so a projection matches the queries it was written to
+serve. Every aggregate needs an alias, which becomes the name of the stored column. Time granularity is expressed
+with `TIME_FLOOR`, as it would be in a query.
+
+A projection body accepts a select list, an optional `WHERE` and an optional `GROUP BY`. It cannot use `ORDER BY`,
+`LIMIT` or `HAVING`: a projection's ordering follows its grouping columns and is not something you choose. It also
+cannot use joins, subqueries, or expressions computed over aggregates. Store the aggregates instead: `SUM(x)` and
+`COUNT(x)` rather than `AVG(x)`.
+
+Projections may also be added to and removed from an existing table:
+
+```sql
+ALTER TABLE "druid"."visits" ADD [IF NOT EXISTS] PROJECTION by_agent AS (
+  SELECT user_agent, SUM(pages_visited) AS total_pages GROUP BY user_agent
+)
+ALTER TABLE "druid"."visits" DROP PROJECTION [IF EXISTS] by_agent
+```
+
+Both take effect for subsequent ingestion. Segments already built keep whatever projections they were built with, so
+dropping a projection does not rewrite data.
+
+#### The base table
+
+The reserved projection name `__base` describes the table's own physical layout rather than an additional
+pre-aggregation. Defining it makes the table a 'clustered' table: rows of segments are stored grouped by the clustering
+columns.
+
+Its body lists the columns in the order segments store them, so it must name every declared column, in declared
+order. An item written as `<expr> AS <name>` makes that column computed at ingest time, from the columns it reads:
+
+```sql
+CREATE TABLE "druid"."events" (
+  tenant VARCHAR,
+  bucket BIGINT,
+  __time TIMESTAMP,
+  user_id BIGINT,
+  payload TYPE('COMPLEX<json>'),
+  PROJECTION __base AS (
+    SELECT tenant, ABS(user_id) % 128 AS bucket, __time, user_id, payload
+    CLUSTERED BY tenant, bucket
+  )
+)
+PARTITIONED BY DAY
+SEALED
+```
+
+The clustering columns must be the leading columns of the table, because the declared order is the physical order.
+`SEALED` is required: the declared columns define the physical segment schema, so a column that is not declared
+cannot be stored.
+
+A computed column is written by the expression, not by the ingestion query, so an `INSERT` must supply the
+expression's inputs and leave the computed column out. Above, that means supplying `user_id` and letting `bucket`
+be derived.
+
+Unlike an aggregate projection, a `__base` body cannot filter or group: the base table stores every ingested row.
+It is the only projection that chooses a clustering.
+
+`ALTER TABLE ... ADD PROJECTION __base AS ( ... )` gives an existing table a layout, and
+`ALTER TABLE ... DROP PROJECTION __base` removes it. Both affect future segments only.
+
+Every other name beginning with `__` remains reserved.
+
+#### Setting table properties
+
+`SET PROPERTIES` merges the given [table properties](#table-properties) into the table. A value of `NULL` removes a
+property. Values must be literals:
+
+```sql
+ALTER TABLE "druid"."visits" SET PROPERTIES (targetSegmentRows = 3000000, sealed = TRUE)
+ALTER TABLE "druid"."visits" SET PROPERTIES (sealed = NULL)
+```
+
+Both statements require the `WRITE` permission on the datasource, the same permission the REST API checks. Table
+names may be unqualified or qualified with the `druid` schema; other schemas are rejected. Names are case-sensitive.
+
+There is no `DROP TABLE`. Deleting a table's catalog entry without deleting its data would be a surprising meaning
+for the statement, so removing a specification is left to the [delete API](#delete-a-table) until the semantics are
+settled.
+
 ### API Objects
 
 #### TableSpec
@@ -90,10 +249,11 @@ The endpoint supports a set of optional query parameters to enforce optimistic l
 is meant to update a table rather than create a new one. In the default case, with no query parameters set, this request
 will return an error if a table of the same name already exists in the schema specified.
 
-| Parameter   | Type    | Description                                                                                                                   |
-|-------------|---------|-------------------------------------------------------------------------------------------------------------------------------|
-| `version`   | Long    | the expected version of an existing table. The version must match. If not (or if the table does not exist), returns an error. |
-| `overwrite` | boolean | if true, then overwrites any existing table. Otherwise, the operation fails if the table already exists.                      |
+| Parameter     | Type    | Description                                                                                                                   |
+|---------------|---------|-------------------------------------------------------------------------------------------------------------------------------|
+| `version`     | Long    | the expected version of an existing table. The version must match. If not (or if the table does not exist), returns an error. |
+| `overwrite`   | boolean | if true, then overwrites any existing table. Otherwise, the operation fails if the table already exists.                      |
+| `ifNotExists` | boolean | if true, then leaves an existing table unchanged and reports a version of 0. Otherwise, the operation fails if the table already exists. |
 
 ##### Responses