feat: ddl statements + catalog

docs/configuration/index.md

@@ -1919,6 +1919,7 @@ The Druid SQL server is configured through the following properties on the Broke
 |`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.authorizeTableVisibility`|Whether [READ DATASOURCE](../multi-stage-query/security.md) permissions are required for table visibility in the validator. When this is set, users that query unauthorized tables see a "not found" error rather than "forbidden". Additionally, when this is set, INSERT and REPLACE require both READ and WRITE access to the target table. (If this property is not set, they require only WRITE.) Regardless of the value of this property, READ access is required for tables to show up in the INFORMATION_SCHEMA.|true|
+|`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 both `READ` and `WRITE` permission on the datasource, the same permissions the catalog API requires for its write operations, 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,212 @@ 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.
 
+### Effects of a table definition
+
+A table definition takes effect when it is read, not when it is written: changing one, whether through SQL DDL or the
+REST API, rewrites nothing. What it affects:
+
+- **SELECT queries** report each declared column with its declared type, in declared order. Segments that store a
+  column as a different physical type are converted to the declared type at query time. Columns present in segments
+  but not declared in the catalog remain queryable, and follow the declared columns with their physical types.
+- **INSERT and REPLACE** (the catalog applies only to SQL-based ingestion) validate against the definition rather
+  than merely defaulting from it. A query column targeting a declared column must be convertible to the declared type
+  without changing it: inserting a `BIGINT` value into a column declared `VARCHAR` stores the value as a string, while
+  inserting a `VARCHAR` value into a column declared `BIGINT` is an error. Columns the query produces that the table
+  does not declare are rejected when the table is [`sealed`](#table-properties) and ingested normally otherwise.
+  Table properties such as `segmentGranularity` and `clusterKeys` act as defaults that an individual statement may
+  override, such as with its own `PARTITIONED BY`.
+- **Streaming and native batch ingestion** do not consult the catalog; their specs are unaffected by any table
+  definition.
+
+### 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. Changes take effect for subsequent queries
+and ingestion, as described in [Effects of a table definition](#effects-of-a-table-definition).
+
+These statements are disabled by default. Set `druid.sql.planner.enableCatalogDdl` to `true` on the Broker to enable
+them. They require both `READ` and `WRITE` permission on the datasource, the same permissions the catalog API
+requires for its write operations, 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>
+  [ [ SEALED ] ( <table element> [, ...] ) ]
+  [ PARTITIONED BY <granularity> ]
+  [ CLUSTERED BY <column> [, ...] ]
+
+<table element> ::=
+    <column> <type>
+  | PROJECTION <name> AS ( <select> )
+```
+
+`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. It is written just before the
+column list because that is what it describes: the list is the table's whole schema, so the statement does not accept
+`SEALED` without one. (The `sealed` property itself can still be set on any table through `SET PROPERTIES` or the
+REST API.)
+
+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
+```
+
+A complex type must be provided by a loaded extension: the statement fails if the named type is not registered, so a
+typo or a missing extension surfaces when the table is defined rather than at the first ingestion or query that
+touches the column. The extension must be loaded on the Coordinator, which validates catalog writes.
+
+`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 PROJECTION [IF NOT EXISTS] <name> AS ( <select> )
+ALTER TABLE <table> DROP PROJECTION [IF EXISTS] <name>
+ALTER TABLE <table> SET PROPERTIES ( <property> = <value> [, ...] )
+```
+
+`ADD COLUMN` fails if the column already exists, and `ALTER COLUMN` fails if it does not, so a misspelled column name
+is reported rather than quietly creating or replacing a column. Each statement is also checked against the rest of the
+table definition, not only the part it changes: adding a column, changing a type, or setting a property is rejected if
+the resulting table would be invalid, such as a segment granularity coarser than a projection the table declares.
+
+`DROP COLUMN` removes the column's declaration, not its data. A table's SQL schema is the declared columns followed by
+any other columns present in its segments, so a dropped column that still has data remains queryable: it loses its
+declared type and position and appears after the declared columns, like any column the catalog does not know about.
+Whether it can still be ingested then follows the usual rule: rejected if the table is sealed, accepted as an
+undeclared column otherwise. To remove a column from query results without rewriting data, use the catalog's
+`hiddenColumns` property instead, which currently has no SQL spelling and is set through the REST API.
+
+#### 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.
+
+Because the body is planned like a query, it is planned under the statement's own query context, including any `SET`
+clauses. Only context parameters that affect planning can change the stored definition; parameters that only affect
+query execution have no effect, since the body is planned and stored rather than run. The context itself is not part
+of the definition: what the catalog stores is the projection the body planned to, so nothing from the statement's
+context is carried over to queries that later use it. Note also that a projection is matched to a query by its shape,
+so a definition planned under a context that changes that shape only matches queries run under the same context.
+
+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, such as `SUM(x) / COUNT(x)`: store the two
+aggregates as separate columns and divide at query time. A projection can store the same
+[aggregation functions that are supported for rollup at ingestion time](../../multi-stage-query/concepts.md#rollup).
+
+Projections may also be added to and removed from an existing table:
+
+```sql
+ALTER TABLE "druid"."visits" ADD PROJECTION [IF NOT EXISTS] 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" SEALED (
+  tenant VARCHAR,
+  bucket BIGINT,
+  __time TIMESTAMP,
+  user_id BIGINT,
+  payload TYPE('COMPLEX<json>'),
+  PROJECTION __base AS (
+    SELECT tenant, MOD(ABS(user_id), 128) AS bucket, __time, user_id, payload
+    CLUSTERED BY tenant, bucket
+  )
+)
+PARTITIONED BY DAY
+```
+
+The clustering columns must be the leading columns of the table, because the declared order is the physical order.
+`SEALED` is optional: a column the query produces but the table does not declare is stored after the declared
+layout, in the order it arrives. Declare `SEALED` to reject such columns instead.
+
+Computed columns, like `bucket` in the example above, are computed based on inputs provided by the `INSERT` or
+`REPLACE`. In terms of the example, the `INSERT` or `REPLACE` command should provide `user_id`, not `bucket`.
+
+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 `READ` and `WRITE` permission on the datasource, the same permissions 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 +296,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