liting liu created FLINK-40142:
----------------------------------
Summary: [Table SQL / API] Support partition transform expressions
in PARTITIONED BY clause
Key: FLINK-40142
URL: https://issues.apache.org/jira/browse/FLINK-40142
Project: Flink
Issue Type: New Feature
Components: Table SQL / API
Affects Versions: 2.1.3, 1.18.2
Reporter: liting liu
h2. Problem
Flink SQL currently supports only column identifiers in the PARTITIONED BY
clause.
For example:
{code:sql}
CREATE TABLE t (
id BIGINT,
event_time TIMESTAMP(3),
payload STRING
)
PARTITIONED BY (event_time);
{code}
However, Flink SQL cannot express partition transforms such as:
{code:sql}
CREATE TABLE t (
id BIGINT,
event_time TIMESTAMP(3),
payload STRING
)
PARTITIONED BY (
days(event_time),
bucket(16, id)
);
{code}
This prevents table-format connectors, such as Apache Iceberg, from creating
tables with hidden partitioning through Flink SQL.
The limitation is in the Flink SQL and catalog layers, before the CREATE TABLE
operation is passed to a connector.
h2. Current Flink Limitations
* The Flink SQL parser accepts only simple identifiers in the PARTITIONED BY
clause.
* CatalogTable#getPartitionKeys() represents partitioning as List<String>.
* The catalog metadata model cannot represent:
** The partition transform name.
** Transform arguments.
** The source column referenced by a transform.
* Planner validation assumes that every partition key is an identity partition
column.
* SHOW CREATE TABLE and CREATE TABLE LIKE cannot preserve partition transform
expressions.
Consequently, a connector cannot obtain a complete partition specification from
Flink's CatalogTable metadata.
h2. Expected Behavior
Flink should be able to parse, validate, represent, and preserve partition
transform expressions in table DDL.
For example:
{code:sql}
PARTITIONED BY (
days(event_time),
bucket(16, id),
truncate(8, payload)
)
{code}
The resolved catalog metadata should preserve the complete partition
specification so that a connector can translate it into its native partition
definition.
Flink should not implement Iceberg-specific transform behavior. The connector
remains responsible for deciding which transforms it supports and translating
them into its own metadata model.
h2. Proposed Flink-Side Scope
* Extend the SQL grammar and AST for partition transform expressions.
* Introduce a structured representation of partition specifications in the
catalog metadata model.
* Support identity partitions using the same model or provide
backward-compatible conversion.
* Add planner validation for:
** Referenced source columns.
** Transform argument types.
** Invalid or unresolved partition expressions.
* Preserve partition transforms in SHOW CREATE TABLE.
* Preserve partition transforms in CREATE TABLE LIKE.
* Expose the structured partition specification to catalog and connector
implementations.
* Add parser, catalog, planner, and DDL tests.
* Update the Flink SQL documentation.
h2. Out of Scope
* Implementing Iceberg partition transforms inside Flink core.
* Modifying the Iceberg connector to create the final Iceberg PartitionSpec.
* Changing runtime read or write behavior for existing partitioned tables.
* Requiring all connectors to support partition transforms.
h2. Backward Compatibility
Existing identity partition definitions such as:
{code:sql}
PARTITIONED BY (region, dt)
{code}
must remain fully compatible.
Since CatalogTable and the catalog interfaces are part of Flink's public API,
changing the partition metadata representation may require a FLIP.
h2. Acceptance Criteria
* Flink SQL can parse partition transform expressions in PARTITIONED BY.
* The catalog metadata model preserves transform names, arguments, and source
columns.
* Catalog and connector implementations can access the structured partition
specification.
* Existing identity partitioning remains backward compatible.
* SHOW CREATE TABLE outputs an equivalent partition specification.
* CREATE TABLE LIKE preserves the partition specification.
* Invalid partition transforms produce clear validation errors.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
