Ramin Gharib created FLINK-41335:
------------------------------------
Summary: 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
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)
