AHeise commented on code in PR #29370:
URL: https://github.com/apache/flink/pull/29370#discussion_r4181763905
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java:
##########
@@ -65,6 +83,74 @@ void testCastToVariantRejectsValuesOverTheSizeLimit() {
.hasMessageStartingWith("Cannot cast a string value of
16777212 bytes to VARIANT.");
}
+ @ParameterizedTest
+ @ValueSource(strings = {"", "hello", "Grüße, 世界 🚀"})
+ void testCastStringToVariantStoresItsUtf8Bytes(final String str) {
+
assertThat(fromString(binaryString(str.getBytes(UTF_8)))).isEqualTo(BUILDER.of(str));
+ }
+
+ @ParameterizedTest
+ @MethodSource("invalidUtf8")
+ void testCastStringToVariantReplacesInvalidUtf8(final byte[] invalid) {
+ final Variant variant = fromString(binaryString(invalid));
+
+ assertThat(variant.getString()).contains(REPLACEMENT_CHARACTER);
+ assertThat(variant).isEqualTo(BUILDER.of(new String(invalid, UTF_8)));
Review Comment:
nit: `BUILDER.of(new String(invalid, UTF_8))` restates the new fallback.
`BUILDER.of(stringData.toString())` is what the old code did, so asserting
against it pins the "stored exactly as before" claim. The same applies at line
118.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java:
##########
@@ -593,14 +594,43 @@ public static Variant fromDouble(double value) {
}
public static Variant fromDecimal(DecimalData value) {
- return BUILDER.of(value.toBigDecimal());
+ if (!value.isCompact()) {
+ return BUILDER.of(value.toBigDecimal());
+ }
+ final BinaryVariantInternalBuilder builder = new
BinaryVariantInternalBuilder(false);
+ builder.appendDecimal(value.toUnscaledLong(), value.scale());
+ return builder.build();
}
+ /**
+ * Stores the UTF-8 bytes of the string as they are. A {@link StringData}
may hold invalid
+ * UTF-8, which the variant spec does not allow, so such a value is
decoded first and every
+ * malformed sequence is stored as the U+FFFD replacement character.
+ */
public static Variant fromString(StringData value) {
Review Comment:
A `StringData.fromString` value (literals, most UDF and built-in function
results) has no binary section yet, so `getSegments()` materializes it through
`StringUtf8Utils.encodeUTF8` and then validates bytes that are valid by
construction. The result is identical (lone surrogates become `?` either way),
but none of the tests use this form: every `Layout` goes through `fromAddress`.
Could you add a `JAVA_OBJECT(StringData::fromString)`-style case, ideally with
a lone surrogate? Skipping the check when the string started as a Java object
would also be possible, but `BinaryStringData` doesn't expose that, so I'd only
add the test.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java:
##########
@@ -593,14 +594,43 @@ public static Variant fromDouble(double value) {
}
public static Variant fromDecimal(DecimalData value) {
- return BUILDER.of(value.toBigDecimal());
+ if (!value.isCompact()) {
+ return BUILDER.of(value.toBigDecimal());
+ }
+ final BinaryVariantInternalBuilder builder = new
BinaryVariantInternalBuilder(false);
+ builder.appendDecimal(value.toUnscaledLong(), value.scale());
+ return builder.build();
}
+ /**
+ * Stores the UTF-8 bytes of the string as they are. A {@link StringData}
may hold invalid
+ * UTF-8, which the variant spec does not allow, so such a value is
decoded first and every
+ * malformed sequence is stored as the U+FFFD replacement character.
+ */
public static Variant fromString(StringData value) {
+ final BinaryStringData string = (BinaryStringData) value;
+ final MemorySegment[] segments = string.getSegments();
+ final int length = string.getSizeInBytes();
+ final byte[] utf8;
+ final int offset;
+ // Read a string that lies in one heap segment in place, which saves
copying it out.
+ if (segments.length == 1 && !segments[0].isOffHeap()) {
Review Comment:
nit: a string that lies entirely in the first segment of a multi-segment row
still takes the copy. `!segments[0].isOffHeap() && offset + length <=
segments[0].size()` covers both cases, like `BinaryStringData#inFirstSegment`.
Fine to leave if multi-segment rows are rare on this path.
##########
flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java:
##########
@@ -201,4 +208,58 @@ void testAppendFloat() {
assertThatCode(() ->
floatList.forEach(builder::appendFloat)).doesNotThrowAnyException();
}
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "",
+ "Grüße, 世界 🚀",
+ "A string longer than 63 bytes is stored with a 4-byte length
header."
+ })
+ void testAppendStringStoresUtf8BytesAsTheyAre(final String str) {
+ final byte[] utf8 = str.getBytes(UTF_8);
+ final BinaryVariantInternalBuilder builder = new
BinaryVariantInternalBuilder(false);
+ builder.appendString(utf8);
+ final BinaryVariant variant = builder.build();
+
+ final byte[] value = variant.getValue();
+ final int headerSize = utf8.length > MAX_SHORT_STR_SIZE ? 1 + U32_SIZE
: 1;
Review Comment:
This would break the test: it drops the 1-byte type header, and a 64-byte
string would count as short. The current `1 + (length > 63 ? 4 : 0)` is correct.
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java:
##########
@@ -65,6 +79,49 @@ void testCastToVariantRejectsValuesOverTheSizeLimit() {
.hasMessageStartingWith("Cannot cast a string value of
16777212 bytes to VARIANT.");
}
+ @ParameterizedTest
+ @ValueSource(strings = {"", "hello", "Grüße, 世界 🚀"})
+ void testCastStringToVariantStoresItsUtf8Bytes(final String str) {
+
assertThat(fromString(binaryString(str.getBytes(UTF_8)))).isEqualTo(BUILDER.of(str));
+ }
+
+ @ParameterizedTest
+ @MethodSource("invalidUtf8")
+ void testCastStringToVariantReplacesInvalidUtf8(final byte[] invalid) {
+ final Variant variant = fromString(binaryString(invalid));
+
+ assertThat(variant.getString()).contains(REPLACEMENT_CHARACTER);
+ assertThat(variant).isEqualTo(BUILDER.of(new String(invalid, UTF_8)));
+ }
+
+ private static Stream<byte[]> invalidUtf8() {
+ return Stream.of(
+ new byte[] {'a', (byte) 0xFF, 'b'},
+ new byte[] {'a', (byte) 0xC3},
+ new byte[] {(byte) 0xC0, (byte) 0xAF},
+ new byte[] {(byte) 0xED, (byte) 0xA0, (byte) 0x80});
+ }
+
+ @ParameterizedTest(name = "{0} as DECIMAL({1}, {2})")
+ @CsvSource({
+ "0, 1, 0",
+ "1.50, 10, 2",
+ "999999999, 9, 0",
+ "-0.999999999, 9, 9",
+ "1000000000, 10, 0",
+ "0.0000000001, 10, 10",
+ "-999999999999999999, 18, 0",
Review Comment:
Not reachable through `fromDecimal`, since `DecimalType` requires a scale
between 0 and the precision. `BinaryVariantInternalBuilderTest` covers scale -1
at the builder level.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
