diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml index d25223c5f20ae8..ea8cc881368cbe 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 85e28b198c3631..ab3cfa539c1a9d 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 42f3d587499808..9a6ffc600f299f 100644 --- a/flink-python/pyflink/table/expressions.py +++ b/flink-python/pyflink/table/expressions.py @@ -32,8 +32,9 @@ '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 f894cc2afd5ecb..8fdc1af2f262db 100644 --- a/flink-python/pyflink/table/tests/test_expression.py +++ b/flink-python/pyflink/table/tests/test_expression.py @@ -30,7 +30,7 @@ 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 @@ def test_expressions(self): 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 a2a9544fb3558c..c91441869f5aa1 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 static ApiExpression concatWs(Object separator, Object string, Object... /** * 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 RFC 9562 */ public static ApiExpression uuid() { return apiCall(BuiltInFunctionDefinitions.UUID); } + /** + * Returns a random RFC 9562 version 4 (pseudo randomly generated) {@code UUID} value. + * + *

Compared to {@link #uuid()}, this function returns a value of the {@code UUID} data type. + * + * @see RFC 9562 + */ + 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. + * + *

Compared to {@link #uuid()}, this function returns a value of the {@code UUID} data type. + * + * @see RFC 9562 + */ + 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 01b3f48f6fd5ab..ce1e09f98a0fe3 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 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 b1b59ed663aa3e..0c41031688fa3c 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 fdfa00390d6f4c..1b16c3f52bd4d6 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 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL) .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 00000000000000..f3f7bd0ac0d501 --- /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 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 00000000000000..117d10df2edcd4 --- /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 00000000000000..5b78a469e1f7f7 --- /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 00000000000000..34beaaa67cff5e --- /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; + } +}