This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new c6ff968a424 [FLINK-40489][table] Add UUID_V4() and UUID_V7() functions
c6ff968a424 is described below
commit c6ff968a424c7aa41822d8db101de21b16cf5ab9
Author: Moritz Manner <[email protected]>
AuthorDate: Tue Sep 15 10:09:56 2026 +0200
[FLINK-40489][table] Add UUID_V4() and UUID_V7() functions
This closes #29154.
---
docs/data/sql_functions.yml | 8 ++-
docs/data/sql_functions_zh.yml | 14 ++++-
flink-python/pyflink/table/expressions.py | 29 ++++++++-
.../pyflink/table/tests/test_expression.py | 4 +-
.../org/apache/flink/table/api/Expressions.java | 27 ++++++++-
.../expressions/ExpressionSerializationTest.java | 2 +
.../table/api/ImplicitExpressionConversions.scala | 2 +-
.../functions/BuiltInFunctionDefinitions.java | 20 +++++++
.../planner/functions/UuidFunctionsITCase.java | 69 ++++++++++++++++++++++
.../runtime/functions/scalar/UuidV4Function.java | 44 ++++++++++++++
.../runtime/functions/scalar/UuidV7Function.java | 67 +++++++++++++++++++++
.../functions/scalar/UuidV7FunctionTest.java | 62 +++++++++++++++++++
12 files changed, 340 insertions(+), 8 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index d25223c5f20..ea8cc881368 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -227,7 +227,13 @@ arithmetic:
description: Returns a pseudorandom integer value in the range [0, INT2)
with an initial seed INT1. Two RAND_INTEGER functions will return identical
sequences of numbers if they have the same initial seed and bound.
- sql: UUID()
table: uuid()
- description: Returns an UUID (Universally Unique Identifier) string (e.g.,
"3d3c68f7-f608-473f-b60c-b0c44ad4cc4e") according to RFC 4122 type 4 (pseudo
randomly generated) UUID. The UUID is generated using a cryptographically
strong pseudo random number generator.
+ description: Returns an UUID (Universally Unique Identifier) string (e.g.,
"3d3c68f7-f608-473f-b60c-b0c44ad4cc4e") according to RFC 9562 version 4 (pseudo
randomly generated) UUID. The UUID is generated using a cryptographically
strong pseudo random number generator.
+ - sql: UUID_V4()
+ table: uuidV4()
+ description: Returns a random RFC 9562 version 4 UUID value. The UUID is
generated using a cryptographically strong pseudo random number generator.
Compared to UUID(), this function returns a value of the UUID data type.
+ - sql: UUID_V7()
+ table: uuidV7()
+ description: Returns a time-ordered RFC 9562 version 7 UUID value,
generated from the current timestamp and a random component. Compared to
UUID(), this function returns a value of the UUID data type.
- sql: BIN(INT)
table: INT.bin()
description: Returns a string representation of INTEGER in binary format.
Returns NULL if INTEGER is NULL. E.g., 4.bin() returns "100" and 12.bin()
returns "1100".
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index 85e28b198c3..ab3cfa539c1 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -280,8 +280,20 @@ arithmetic:
- sql: UUID()
table: uuid()
description: |
- 根据 RFC 4122 类型 4（伪随机生成）UUID，返回 UUID（通用唯一标识符）字符串。
+ 根据 RFC 9562 类型 4（伪随机生成）UUID，返回 UUID（通用唯一标识符）字符串。
例如“3d3c68f7-f608-473f-b60c-b0c44ad4cc4e”，UUID 是使用加密强的伪随机数生成器生成的。
+ - sql: UUID_V4()
+ table: uuidV4()
+ description: |
+ Returns a random RFC 9562 version 4 UUID value. The UUID is generated
using a
+ cryptographically strong pseudo random number generator. Compared to
UUID(), this
+ function returns a value of the UUID data type.
+ - sql: UUID_V7()
+ table: uuidV7()
+ description: |
+ Returns a time-ordered RFC 9562 version 7 UUID value, generated from the
current
+ timestamp and a random component. Compared to UUID(), this function
returns a value
+ of the UUID data type.
- sql: BIN(INT)
table: INT.bin()
description: |
diff --git a/flink-python/pyflink/table/expressions.py
b/flink-python/pyflink/table/expressions.py
index 42f3d587499..9a6ffc600f2 100644
--- a/flink-python/pyflink/table/expressions.py
+++ b/flink-python/pyflink/table/expressions.py
@@ -32,8 +32,9 @@ __all__ = ['if_then_else', 'lit', 'col', 'range_', 'and_',
'or_', 'not_', 'UNBOU
'current_watermark', 'local_time', 'local_timestamp',
'temporal_overlaps', 'date_format', 'timestamp_diff', 'array',
'row', 'map_',
'row_interval', 'pi', 'e', 'rand', 'rand_integer', 'atan2',
'negative', 'concat',
- 'concat_ws', 'uuid', 'null_of', 'log', 'with_columns',
'without_columns', 'json',
- 'json_string', 'json_object', 'json_object_agg', 'json_array',
'json_array_agg',
+ 'concat_ws', 'uuid', 'uuid_v4', 'uuid_v7', 'null_of', 'log',
'with_columns',
+ 'without_columns', 'json', 'json_string', 'json_object',
'json_object_agg',
+ 'json_array', 'json_array_agg',
'call', 'call_sql', 'source_watermark', 'to_timestamp_ltz',
'from_unixtime', 'to_date',
'to_timestamp', 'convert_tz', 'unix_timestamp', 'descriptor']
@@ -765,13 +766,35 @@ def concat_ws(separator: Union[str, Expression[str]],
def uuid() -> Expression[str]:
"""
Returns an UUID (Universally Unique Identifier) string (e.g.,
- "3d3c68f7-f608-473f-b60c-b0c44ad4cc4e") according to RFC 4122 type 4
(pseudo randomly
+ "3d3c68f7-f608-473f-b60c-b0c44ad4cc4e") according to RFC 9562 version 4
(pseudo randomly
generated) UUID. The UUID is generated using a cryptographically strong
pseudo random number
generator.
"""
return _leaf_op("uuid")
+@PublicEvolving()
+def uuid_v4() -> Expression:
+ """
+ Returns a random RFC 9562 version 4 (pseudo randomly generated) UUID
value. The UUID is
+ generated using a cryptographically strong pseudo random number generator.
+
+ Compared to uuid(), this function returns a value of the UUID data type.
+ """
+ return _leaf_op("uuidV4")
+
+
+@PublicEvolving()
+def uuid_v7() -> Expression:
+ """
+ Returns a time-ordered RFC 9562 version 7 UUID value, generated from the
current timestamp
+ and a random component.
+
+ Compared to uuid(), this function returns a value of the UUID data type.
+ """
+ return _leaf_op("uuidV7")
+
+
@PublicEvolving()
def null_of(data_type: DataType) -> Expression:
"""
diff --git a/flink-python/pyflink/table/tests/test_expression.py
b/flink-python/pyflink/table/tests/test_expression.py
index f894cc2afd5..8fdc1af2f26 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -30,7 +30,7 @@ from pyflink.table.expressions import (col, lit, range_,
and_, or_, current_date
rand, rand_integer, atan2, negative,
concat, concat_ws, uuid,
null_of, log, if_then_else,
with_columns, call,
to_timestamp_ltz, from_unixtime,
to_date, to_timestamp,
- convert_tz, unix_timestamp)
+ convert_tz, unix_timestamp, uuid_v4,
uuid_v7)
from pyflink.testing.test_case_utils import PyFlinkTestCase
@@ -375,6 +375,8 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
self.assertEqual('concat(a, b, c)', str(concat(expr1, expr2, expr3)))
self.assertEqual("concat_ws(', ', b, c)", str(concat_ws(', ', expr2,
expr3)))
self.assertEqual('uuid()', str(uuid()))
+ self.assertEqual('UUID_V4()', str(uuid_v4()))
+ self.assertEqual('UUID_V7()', str(uuid_v7()))
self.assertEqual('null', str(null_of(DataTypes.BIGINT())))
self.assertEqual('log(a)', str(log(expr1)))
self.assertEqual('ifThenElse(a, b, c)', str(if_then_else(expr1, expr2,
expr3)))
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
index a2a9544fb35..c91441869f5 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
@@ -827,14 +827,39 @@ public final class Expressions {
/**
* Returns an UUID (Universally Unique Identifier) string (e.g.,
- * "3d3c68f7-f608-473f-b60c-b0c44ad4cc4e") according to RFC 4122 type 4
(pseudo randomly
+ * "3d3c68f7-f608-473f-b60c-b0c44ad4cc4e") according to RFC 9562 version 4
(pseudo randomly
* generated) UUID. The UUID is generated using a cryptographically strong
pseudo random number
* generator.
+ *
+ * @see <a href="https://www.rfc-editor.org/rfc/rfc9562.html";>RFC 9562</a>
*/
public static ApiExpression uuid() {
return apiCall(BuiltInFunctionDefinitions.UUID);
}
+ /**
+ * Returns a random RFC 9562 version 4 (pseudo randomly generated) {@code
UUID} value.
+ *
+ * <p>Compared to {@link #uuid()}, this function returns a value of the
{@code UUID} data type.
+ *
+ * @see <a href="https://www.rfc-editor.org/rfc/rfc9562.html";>RFC 9562</a>
+ */
+ public static ApiExpression uuidV4() {
+ return apiCall(BuiltInFunctionDefinitions.UUID_V4);
+ }
+
+ /**
+ * Returns a time-ordered RFC 9562 version 7 {@code UUID} value, generated
from the current
+ * timestamp and a random component.
+ *
+ * <p>Compared to {@link #uuid()}, this function returns a value of the
{@code UUID} data type.
+ *
+ * @see <a href="https://www.rfc-editor.org/rfc/rfc9562.html";>RFC 9562</a>
+ */
+ public static ApiExpression uuidV7() {
+ return apiCall(BuiltInFunctionDefinitions.UUID_V7);
+ }
+
/**
* Returns a null literal value of a given data type.
*
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/ExpressionSerializationTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/ExpressionSerializationTest.java
index 01b3f48f6fd..ce1e09f98a0 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/ExpressionSerializationTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/ExpressionSerializationTest.java
@@ -58,6 +58,8 @@ public class ExpressionSerializationTest {
public static Stream<TestSpec> testData() {
return Stream.of(
TestSpec.forExpr(Expressions.uuid()).expectStr("UUID()"),
+ TestSpec.forExpr(Expressions.uuidV4()).expectStr("UUID_V4()"),
+ TestSpec.forExpr(Expressions.uuidV7()).expectStr("UUID_V7()"),
TestSpec.forExpr($("f0").abs())
.withField("f0", DataTypes.BIGINT())
.expectStr("ABS(`f0`)"),
diff --git
a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
index b1b59ed663a..0c41031688f 100644
---
a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
+++
b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
@@ -730,7 +730,7 @@ trait ImplicitExpressionConversions {
/**
* Returns an UUID (Universally Unique Identifier) string (e.g.,
- * "3d3c68f7-f608-473f-b60c-b0c44ad4cc4e") according to RFC 4122 type 4
(pseudo randomly
+ * "3d3c68f7-f608-473f-b60c-b0c44ad4cc4e") according to RFC 9562 version 4
(pseudo randomly
* generated) UUID. The UUID is generated using a cryptographically strong
pseudo random number
* generator.
*/
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index fdfa00390d6..1b16c3f52bd 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -1713,6 +1713,26 @@ public final class BuiltInFunctionDefinitions {
.outputTypeStrategy(explicit(DataTypes.CHAR(36).notNull()))
.build();
+ public static final BuiltInFunctionDefinition UUID_V4 =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("UUID_V4")
+ .kind(SCALAR)
+ .notDeterministic()
+ .inputTypeStrategy(NO_ARGS)
+ .outputTypeStrategy(explicit(DataTypes.UUID().notNull()))
+
.runtimeClass("org.apache.flink.table.runtime.functions.scalar.UuidV4Function")
+ .build();
+
+ public static final BuiltInFunctionDefinition UUID_V7 =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("UUID_V7")
+ .kind(SCALAR)
+ .notDeterministic()
+ .inputTypeStrategy(NO_ARGS)
+ .outputTypeStrategy(explicit(DataTypes.UUID().notNull()))
+
.runtimeClass("org.apache.flink.table.runtime.functions.scalar.UuidV7Function")
+ .build();
+
public static final BuiltInFunctionDefinition LTRIM =
BuiltInFunctionDefinition.newBuilder()
.name("ltrim")
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/UuidFunctionsITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/UuidFunctionsITCase.java
new file mode 100644
index 00000000000..f3f7bd0ac0d
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/UuidFunctionsITCase.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+
+import java.util.stream.Stream;
+
+import static org.apache.flink.table.api.Expressions.uuidV4;
+import static org.apache.flink.table.api.Expressions.uuidV7;
+
+/**
+ * Test for {@link BuiltInFunctionDefinitions#UUID_V4} and {@link
+ * BuiltInFunctionDefinitions#UUID_V7} and their return type.
+ */
+public class UuidFunctionsITCase extends BuiltInFunctionTestBase {
+
+ @Override
+ Stream<TestSetSpec> getTestSetSpecs() {
+ return Stream.of(
+ // UUID_V4()
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.UUID_V4)
+ .testSqlResult("UUID_V4()", DataTypes.UUID().notNull())
+ .testTableApiResult(uuidV4(),
DataTypes.UUID().notNull()),
+ // UUID_V4() produces a version 4 UUID, i.e. the canonical
string form has '4' as
+ // the first character of the third group.
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.UUID_V4)
+ .testSqlResult(
+ "CHAR_LENGTH(SPLIT_INDEX(CAST(UUID_V4() AS
STRING), '-', 2))",
+ 4,
+ DataTypes.INT())
+ .testSqlResult(
+ "SUBSTR(SPLIT_INDEX(CAST(UUID_V4() AS STRING),
'-', 2), 1, 1)",
+ "4",
+ DataTypes.STRING()),
+ // UUID_V7()
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.UUID_V7)
+ .testSqlResult("UUID_V7()", DataTypes.UUID().notNull())
+ .testTableApiResult(uuidV7(),
DataTypes.UUID().notNull()),
+ // UUID_V7() produces a version 7 UUID, i.e. the canonical
string form has '7' as
+ // the first character of the third group.
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.UUID_V7)
+ .testSqlResult(
+ "CHAR_LENGTH(SPLIT_INDEX(CAST(UUID_V7() AS
STRING), '-', 2))",
+ 4,
+ DataTypes.INT())
+ .testSqlResult(
+ "SUBSTR(SPLIT_INDEX(CAST(UUID_V7() AS STRING),
'-', 2), 1, 1)",
+ "7",
+ DataTypes.STRING()));
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/UuidV4Function.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/UuidV4Function.java
new file mode 100644
index 00000000000..117d10df2ed
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/UuidV4Function.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.logical.UuidType;
+
+import java.nio.ByteBuffer;
+import java.util.UUID;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#UUID_V4}. */
+@Internal
+public class UuidV4Function extends BuiltInScalarFunction {
+
+ public UuidV4Function(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.UUID_V4, context);
+ }
+
+ public byte[] eval() {
+ final UUID uuid = UUID.randomUUID();
+ final ByteBuffer buffer = ByteBuffer.allocate(UuidType.BYTE_LENGTH);
+ buffer.putLong(uuid.getMostSignificantBits());
+ buffer.putLong(uuid.getLeastSignificantBits());
+ return buffer.array();
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/UuidV7Function.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/UuidV7Function.java
new file mode 100644
index 00000000000..5b78a469e1f
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/UuidV7Function.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.types.logical.UuidType;
+
+import java.security.SecureRandom;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#UUID_V7}. */
+@Internal
+public class UuidV7Function extends BuiltInScalarFunction {
+
+ private static final SecureRandom SECURE_RANDOM = new SecureRandom();
+
+ public UuidV7Function(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.UUID_V7, context);
+ }
+
+ public byte[] eval() {
+ return generate();
+ }
+
+ @VisibleForTesting
+ static byte[] generate() {
+ final long timestamp = System.currentTimeMillis();
+ final byte[] bytes = new byte[UuidType.BYTE_LENGTH];
+ SECURE_RANDOM.nextBytes(bytes);
+
+ // embed the timestamp into the first 6 bytes
+ bytes[0] = (byte) (timestamp >>> 40);
+ bytes[1] = (byte) (timestamp >>> 32);
+ bytes[2] = (byte) (timestamp >>> 24);
+ bytes[3] = (byte) (timestamp >>> 16);
+ bytes[4] = (byte) (timestamp >>> 8);
+ bytes[5] = (byte) timestamp;
+
+ // set the version to 7
+ bytes[6] &= 0x0F;
+ bytes[6] |= 0x70;
+
+ // set the variant to IETF
+ bytes[8] &= 0x3F;
+ bytes[8] |= (byte) 0x80;
+
+ return bytes;
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/scalar/UuidV7FunctionTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/scalar/UuidV7FunctionTest.java
new file mode 100644
index 00000000000..34beaaa67cf
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/scalar/UuidV7FunctionTest.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.table.types.logical.UuidType;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link UuidV7Function#generate()}. */
+class UuidV7FunctionTest {
+
+ @Test
+ void testGenerateLength() {
+ assertThat(UuidV7Function.generate()).hasSize(UuidType.BYTE_LENGTH);
+ }
+
+ @Test
+ void testGenerateVersionAndVariant() {
+ byte[] bytes = UuidV7Function.generate();
+ assertThat(bytes[6] & 0xF0).as("version nibble").isEqualTo(0x70);
+ assertThat(bytes[8] & 0xC0).as("variant bits").isEqualTo(0x80);
+ }
+
+ @Test
+ void testGenerateTimestamp() {
+ long now = System.currentTimeMillis();
+ byte[] bytes = UuidV7Function.generate();
+ long timestamp = extractTimestamp(bytes);
+
+ // The tolerance absorbs scheduling gaps and small clock corrections.
+ long tolerance = Duration.ofSeconds(10).toMillis();
+ assertThat(timestamp).isBetween(now - tolerance, now + tolerance);
+ }
+
+ private static long extractTimestamp(byte[] bytes) {
+ long timestamp = 0;
+ for (int i = 0; i < 6; i++) {
+ timestamp = (timestamp << 8) | (bytes[i] & 0xFFL);
+ }
+ return timestamp;
+ }
+}
