[
https://issues.apache.org/jira/browse/FLINK-41335?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Mika Naylor reassigned FLINK-41335:
-----------------------------------
Assignee: Mika Naylor
> Release Testing: Verify FLIP-604: Complete VARIANT Primitive Coverage with
> UUID and Timestamps
> ----------------------------------------------------------------------------------------------
>
> Key: FLINK-41335
> URL: https://issues.apache.org/jira/browse/FLINK-41335
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / API
> Reporter: Ramin Gharib
> Assignee: Mika Naylor
> Priority: Blocker
> Labels: release-testing
> Fix For: 2.4.0
>
>
> [FLIP-604|https://cwiki.apache.org/confluence/spaces/FLINK/pages/446070993]
> adds a native {{UUID}} type to Flink SQL and the Table API. It also lets a
> {{VARIANT}} hold {{TIME}}, nanosecond {{TIMESTAMP}}, nanosecond
> {{TIMESTAMP_LTZ}} and {{UUID}} values. These are type ids 17 to 20 of the
> Parquet Variant spec.
> h3. What is new
> * {{UUID}} as a column type, {{DataTypes.UUID()}} in the Table API and
> {{Types.UUID}} in the DataStream API.
> * The literal {{UUID 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'}} and the
> functions {{UUID_V4()}} and {{UUID_V7()}}. The old {{UUID()}} still returns
> {{CHAR(36)}}.
> * Explicit {{CAST}} between {{UUID}} and {{STRING}}, {{BINARY(16)}},
> {{BYTES}} and {{VARIANT}}. There is no implicit cast.
> * Comparison, {{ORDER BY}}, {{GROUP BY}}, {{DISTINCT}}, joins and set
> operations. The order is unsigned big-endian bytes, as in PostgreSQL and
> Iceberg.
> * {{CAST}} to and from {{VARIANT}} for {{TIME}}, {{TIMESTAMP(p)}},
> {{TIMESTAMP_LTZ(p)}} and {{UUID}}. A precision above 6 is stored in
> nanoseconds.
> Docs:
> [VARIANT|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/sql/reference/data-types/#variant],
> [built-in
> functions|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/sql/functions/built-in-functions/],
> UUID type section from FLINK-40494.
> h3. Setup
> Build the {{release-2.4}} branch, or use the 2.4.0 RC once it is out.
> {code:bash}
> git checkout release-2.4
> ./mvnw clean install -DskipTests -Dfast -Pskip-webui-build -T1C
> cd build-target
> ./bin/start-cluster.sh
> ./bin/sql-client.sh
> {code}
> No connector or format reads or writes a {{UUID}} column in 2.4. This
> includes datagen. Derive the UUIDs inside the query instead, and write them
> to the {{print}} or {{blackhole}} connector.
> {code:sql}
> SET 'table.local-time-zone' = 'UTC';
> SET 'sql-client.execution.result-mode' = 'tableau';
> CREATE TABLE src (
> k INT,
> pt AS PROCTIME()
> ) WITH (
> 'connector' = 'datagen',
> 'rows-per-second' = '5',
> 'fields.k.min' = '1',
> 'fields.k.max' = '5'
> );
> -- id repeats 5 keys: 00000000-0000-0000-0000-000000000001 ..
> 00000000-0000-0000-0000-000000000005
> -- event_id is a fresh time-ordered UUID per row
> CREATE VIEW events AS
> SELECT
> CAST(LPAD(CAST(k AS STRING), 32, '0') AS UUID) AS id,
> UUID_V7() AS event_id,
> k,
> pt
> FROM src;
> {code}
> h3. 1. Literals, generators and casts
> {code:sql}
> SELECT UUID 'A0EEBC99-9C0B-4EF8-BB6D-6BB9BD380A11';
> -- a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11, always canonical lower case
> SELECT UUID 'abcd';
> -- validation error: Invalid UUID string: abcd
> SELECT TYPEOF(UUID_V4()), TYPEOF(UUID_V7()), TYPEOF(UUID());
> -- UUID NOT NULL | UUID NOT NULL | CHAR(36) NOT NULL
> -- The version digit is the 15th character
> SELECT SUBSTR(CAST(UUID_V4() AS STRING), 15, 1), SUBSTR(CAST(UUID_V7() AS
> STRING), 15, 1);
> -- 4 | 7
> -- String parsing is lenient, like PostgreSQL. All three return
> a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11
> SELECT
> CAST('{a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11}' AS UUID),
> CAST('a0eebc999c0b4ef8bb6d6bb9bd380a11' AS UUID),
> CAST('a0ee-bc99-9c0b-4ef8-bb6d-6bb9-bd38-0a11' AS UUID);
> SELECT CAST(UUID 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11' AS BYTES);
> -- the 16 bytes a0 ee bc 99 9c 0b 4e f8 bb 6d 6b b9 bd 38 0a 11, big-endian
> SELECT CAST(x'a0eebc999c0b4ef8bb6d6bb9bd380a11' AS UUID);
> -- a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11
> -- Malformed input read from a table, so the runtime path runs and not
> constant folding
> SELECT TRY_CAST(CAST(k AS STRING) AS UUID), TRY_CAST(CAST(CAST(k AS STRING)
> AS BYTES) AS UUID) FROM events;
> -- NULL | NULL. Without TRY_ the job fails: a string needs 32 hex digits,
> bytes need exactly 16.
> SELECT CAST(UUID 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11' AS INT);
> -- validation error: Cast function cannot convert value of type UUID to type
> INTEGER
> SELECT id = '00000000-0000-0000-0000-000000000001' FROM events;
> -- validation error: Cannot apply '=' to arguments of type '<UUID> =
> <CHAR(36)>'
> {code}
> h3. 2. Comparison, ordering, grouping and joins
> Order is unsigned. A signed byte comparison would sort {{80000000-...}} first.
> {code:sql}
> SET 'execution.runtime-mode' = 'batch';
> SELECT u FROM (VALUES
> (UUID 'ffffffff-ffff-ffff-ffff-ffffffffffff'),
> (UUID '00000000-0000-0000-0000-000000000000'),
> (UUID '80000000-0000-0000-0000-000000000000'),
> (UUID '550e8400-e29b-41d4-a716-446655440000')
> ) AS T(u)
> ORDER BY u;
> -- 00000000-... , 550e8400-... , 80000000-... , ffffffff-...
> {code}
> In streaming mode, check that each query keys on {{id}} correctly. Repeat the
> ones that make sense in batch mode with a bounded source, for example
> {{'number-of-rows' = '100'}}.
> {code:sql}
> SET 'execution.runtime-mode' = 'streaming';
> SELECT a.id, COUNT(*) FROM events a JOIN (SELECT id FROM events WHERE k = 1)
> b ON a.id = b.id GROUP BY a.id;
> SELECT id FROM events WHERE id IN (SELECT id FROM events WHERE k <= 2);
> SELECT window_start, id, COUNT(*)
> FROM TABLE(TUMBLE(TABLE events, DESCRIPTOR(pt), INTERVAL '10' SECOND))
> GROUP BY window_start, window_end, id;
> SELECT id FROM events UNION SELECT id FROM events;
> -- 5 rows
> SELECT ARRAY_CONTAINS(ARRAY[id], UUID '00000000-0000-0000-0000-000000000001'),
> MAP[id, k][UUID '00000000-0000-0000-0000-000000000001']
> FROM events;
> -- TRUE and k only for key ...001
> {code}
> h3. 3. State, savepoints and compiled plans
> This is the most important part. The job below keeps {{UUID}} keys, {{UUID}}
> values and a {{VARIANT}} holding the new kinds in keyed state.
> {code:sql}
> SET 'execution.runtime-mode' = 'streaming';
> SET 'execution.checkpointing.interval' = '10 s';
> SET 'execution.checkpointing.savepoint-dir' =
> 'file:///tmp/flip604-savepoints';
> SET 'state.backend.type' = 'rocksdb'; -- repeat the section with 'hashmap'
> SET 'parallelism.default' = '2';
> CREATE TABLE agg_sink (id UUID, cnt BIGINT, last_event UUID, last_doc STRING)
> WITH ('connector' = 'print');
> INSERT INTO agg_sink
> SELECT
> id,
> COUNT(*),
> LAST_VALUE(event_id),
> JSON_STRING(LAST_VALUE(CAST(CAST(ROW(event_id, TIME '01:02:03.456',
> CAST(CURRENT_TIMESTAMP AS TIMESTAMP(9)))
> AS ROW<event UUID, t TIME(3), ts
> TIMESTAMP(9)>) AS VARIANT)))
> FROM events
> GROUP BY id;
> {code}
> # Let it run for a minute. The TaskManager {{.out}} file shows changelog rows
> per key.
> # Run {{SHOW JOBS;}} and {{STOP JOB '<job id>' WITH SAVEPOINT;}}.
> # Run {{SET 'execution.state-recovery.path' = '<savepoint path>';}} and {{SET
> 'parallelism.default' = '3';}}, then submit the same {{INSERT}} again.
> # Expected: the counts continue from the values before the stop. No
> serializer or restore errors appear in the logs.
> # Compile the same statement into a plan, run it, stop it with a savepoint
> and restore it with {{EXECUTE PLAN}}:
> {code:sql}
> COMPILE PLAN 'file:///tmp/flip604-plan.json' FOR INSERT INTO agg_sink SELECT
> ... ; -- the INSERT above
> EXECUTE PLAN 'file:///tmp/flip604-plan.json';
> {code}
> # Expected: the plan file contains {{"UUID"}} as a type, and the job restores
> like the plain {{INSERT}}.
> h3. 4. VARIANT with the new kinds
> {code:sql}
> SELECT
> JSON_STRING(CAST(UUID 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11' AS VARIANT)),
> JSON_STRING(CAST(TIME '12:34:56.789' AS VARIANT)),
> JSON_STRING(CAST(CAST('2026-11-04 10:00:00.123456789' AS TIMESTAMP(9)) AS
> VARIANT)),
> JSON_STRING(CAST(TO_TIMESTAMP_LTZ('2026-11-04 10:00:00.123456789',
> 'yyyy-MM-dd HH:mm:ss.SSSSSSSSS') AS VARIANT));
> -- "a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11" | "12:34:56.789" |
> "2026-11-04T10:00:00.123456789" | "2026-11-04T10:00:00.123456789Z"
> CREATE VIEW docs AS
> SELECT CAST(CAST(ROW(id, TIME '01:02:03.456', CAST('2026-11-04
> 10:00:00.123456789' AS TIMESTAMP(9)))
> AS ROW<id UUID, t TIME(3), ts TIMESTAMP(9)>) AS VARIANT) AS doc
> FROM events;
> SELECT
> JSON_STRING(doc), --
> {"id":"00000000-...-00000000000k","t":"01:02:03.456","ts":"2026-11-04T10:00:00.123456789"}
> CAST(doc['id'] AS UUID), -- 00000000-...-00000000000k
> CAST(doc['t'] AS TIME(3)), -- 01:02:03.456
> CAST(doc['ts'] AS TIMESTAMP(9)), -- 2026-11-04 10:00:00.123456789
> CAST(doc['ts'] AS TIMESTAMP(3)), -- 2026-11-04 10:00:00.123, extra
> digits truncated
> CAST(doc AS ROW<id UUID, ts TIMESTAMP(9)>)
> FROM docs;
> -- Nanoseconds only cover 1677-09-21 to 2262-04-11
> SELECT JSON_STRING(CAST(CAST('2300-01-01 00:00:00' AS TIMESTAMP(6)) AS
> VARIANT)); -- "2300-01-01T00:00:00"
> SELECT TRY_CAST(CAST('2300-01-01 00:00:00' AS TIMESTAMP(9)) AS VARIANT);
> -- NULL, CAST fails
> -- A VARIANT cast never reinterprets the stored kind
> SELECT TRY_CAST(PARSE_JSON('"a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"') AS
> UUID); -- NULL, the kind is STRING
> SELECT CAST(CAST(PARSE_JSON('"a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11"') AS
> STRING) AS UUID); -- works
> SELECT TRY_CAST(CAST(TIME '12:34:56.789' AS VARIANT) AS TIMESTAMP(3));
> -- NULL
> {code}
> Also write {{docs}} through the {{json}} and {{raw}} formats with the
> filesystem connector. Files are committed on checkpoints. Expected: each line
> is JSON, with the UUID, time and timestamp values as strings.
> h3. 5. Table API and DataStream (optional, Java)
> {code:java}
> DataStream<UUID> ds = env.fromData(Types.UUID, UUID.randomUUID()); //
> pass Types.UUID explicitly
> Table t = tEnv.fromDataStream(ds).as("u"); //
> schema: u UUID
> Table out = t.select($("u"), $("u").isEqual(lit(someUuid)),
> $("u").cast(DataTypes.STRING()), uuidV7());
> tEnv.toDataStream(out); //
> fields come back as java.util.UUID
> {code}
> In a UDF, build and read variants with
> {{Variant.newBuilder().of(LocalTime)}}, {{of(java.util.UUID)}},
> {{getTime()}}, {{getUuid()}} and {{getType()}}, which returns {{TIME}},
> {{TIMESTAMP_NS}}, {{TIMESTAMP_LTZ_NS}} or {{UUID}}.
> h3. Known issues. Do not file these again
> ||Behavior||Ticket||
> |{{COMPILE PLAN}} fails when the query contains a {{UUID}} literal: {{Literal
> type 'UUID' is not supported for serializing value ...}}. {{CAST('...' AS
> UUID)}} works as a workaround.|FLINK-41332|
> |{{id IN (UUID 'a', UUID 'b')}} and {{id = UUID 'a' OR id = UUID 'b'}} return
> no rows. {{NOT IN}} returns all rows. Same root cause as for
> {{BINARY}}.|FLINK-41327|
> |{{MIN}} and {{MAX}} on {{UUID}} fail in both modes. {{COUNT(DISTINCT id)}}
> fails in streaming mode. Aggregate over {{CAST(id AS STRING)}} instead, which
> sorts the same.|FLINK-41333|
> |Two {{UUID}} literals compared at planning time use signed order, so {{UUID
> '80000000-...' > UUID '00000000-...'}} returns FALSE. Column comparisons are
> correct.|CALCITE-7716, needs a Calcite upgrade|
> |The JDBC driver and the SQL Gateway REST API with the default
> {{rowFormat=JSON}} cannot return a {{UUID}} column. The SQL Client is not
> affected.|FLINK-41334|
> |No connector or format reads or writes a {{UUID}} column. Cast it to
> {{STRING}} or {{BYTES}} before writing.|FLINK-40544 for raw, others are
> future work in the FLIP|
> |PyFlink has no {{UUID}} type.|none|
> |The DataStream API extracts {{java.util.UUID}} as a generic type unless
> {{Types.UUID}} is given.|FLINK-40546, planned for 3.0|
> h3. Reporting
> Comment with the commit or RC, the runtime mode, the state backend and the
> parallelism, and give pass or fail per section. File new bugs as linked
> issues with Fix Version 2.4.0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
