From 619c4962d22a9c71f1f6573a3e3c397ffbe1475c Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Fri, 11 Sep 2026 15:45:31 +0800 Subject: [PATCH 1/7] [kafka] Add strict JSON scalar conversion Local JSON schema-aware Produce split; no remote publication. Co-Authored-By: Codex AI-Model: gpt-6-astra AI-Contributed/Feature: 363/363 AI-Contributed/UT: 140/140 --- fluss-kafka/pom.xml | 5 + .../format/json/JsonToFlussConverter.java | 32 ++ .../format/json/JsonToFlussConverters.java | 326 ++++++++++++++++++ .../json/JsonToFlussConvertersTest.java | 140 ++++++++ 4 files changed, 503 insertions(+) create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverter.java create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java diff --git a/fluss-kafka/pom.xml b/fluss-kafka/pom.xml index e38487db564..e2ed458a1a3 100644 --- a/fluss-kafka/pom.xml +++ b/fluss-kafka/pom.xml @@ -63,6 +63,11 @@ ${project.version} + + org.apache.fluss + fluss-shaded-jackson + + ${project.groupId} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverter.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverter.java new file mode 100644 index 00000000000..e1bbf8ae52b --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverter.java @@ -0,0 +1,32 @@ +/* + * 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.fluss.kafka.format.json; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; + +import javax.annotation.Nullable; + +/** Converts one JSON node into the internal representation of one Fluss field. */ +@Internal +interface JsonToFlussConverter { + + /** Converts the node, which is null when the JSON property is missing. */ + @Nullable + Object convert(@Nullable JsonNode node, String path); +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java new file mode 100644 index 00000000000..b3dc7e478b6 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java @@ -0,0 +1,326 @@ +/* + * 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.fluss.kafka.format.json; + +import org.apache.fluss.kafka.schema.KafkaTopicSchemaException; +import org.apache.fluss.kafka.transcode.KafkaRecordEncodingException; +import org.apache.fluss.row.BinaryString; +import org.apache.fluss.row.Decimal; +import org.apache.fluss.row.TimestampLtz; +import org.apache.fluss.row.TimestampNtz; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.fluss.types.BinaryType; +import org.apache.fluss.types.CharType; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DecimalType; +import org.apache.fluss.types.LocalZonedTimestampType; +import org.apache.fluss.types.TimeType; +import org.apache.fluss.types.TimestampType; + +import javax.annotation.Nullable; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.time.DateTimeException; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.OffsetDateTime; +import java.time.format.DateTimeParseException; +import java.util.Base64; + +/** Strict JSON-to-Fluss scalar converter construction. */ +final class JsonToFlussConverters { + + private JsonToFlussConverters() {} + + static JsonToFlussConverter create(DataType dataType) { + final JsonToFlussConverter notNullConverter = createNotNull(dataType); + return (node, path) -> { + if (node == null || node.isNull()) { + if (!dataType.isNullable()) { + throw invalid(path, dataType, "field is missing or null"); + } + return null; + } + try { + return notNullConverter.convert(node, path); + } catch (KafkaRecordEncodingException e) { + throw e; + } catch (RuntimeException e) { + throw invalid(path, dataType, "value cannot be converted", e); + } + }; + } + + private static JsonToFlussConverter createNotNull(DataType dataType) { + switch (dataType.getTypeRoot()) { + case BOOLEAN: + return (node, path) -> { + require(node.isBoolean(), path, dataType, "expected a JSON boolean"); + return node.booleanValue(); + }; + case TINYINT: + return (node, path) -> { + long value = integralValue(node, path, dataType); + require( + value >= Byte.MIN_VALUE && value <= Byte.MAX_VALUE, + path, + dataType, + "integer overflow"); + return (byte) value; + }; + case SMALLINT: + return (node, path) -> { + long value = integralValue(node, path, dataType); + require( + value >= Short.MIN_VALUE && value <= Short.MAX_VALUE, + path, + dataType, + "integer overflow"); + return (short) value; + }; + case INTEGER: + return (node, path) -> { + long value = integralValue(node, path, dataType); + require( + value >= Integer.MIN_VALUE && value <= Integer.MAX_VALUE, + path, + dataType, + "integer overflow"); + return (int) value; + }; + case BIGINT: + return (node, path) -> integralValue(node, path, dataType); + case FLOAT: + return (node, path) -> { + require(node.isNumber(), path, dataType, "expected a JSON number"); + double doubleValue = node.doubleValue(); + float value = (float) doubleValue; + require( + Double.isFinite(doubleValue) && Float.isFinite(value), + path, + dataType, + "non-finite or overflowing number"); + return value; + }; + case DOUBLE: + return (node, path) -> { + require(node.isNumber(), path, dataType, "expected a JSON number"); + double value = node.doubleValue(); + require( + Double.isFinite(value), + path, + dataType, + "non-finite or overflowing number"); + return value; + }; + case DECIMAL: + return decimalConverter((DecimalType) dataType); + case CHAR: + return charConverter((CharType) dataType); + case STRING: + return (node, path) -> { + require(node.isTextual(), path, dataType, "expected a JSON string"); + return BinaryString.fromString(node.textValue()); + }; + case BINARY: + return binaryConverter((BinaryType) dataType); + case BYTES: + return (node, path) -> decodeBase64(node, path, dataType); + case DATE: + return (node, path) -> { + String value = textualValue(node, path, dataType); + try { + return Math.toIntExact(LocalDate.parse(value).toEpochDay()); + } catch (DateTimeException | ArithmeticException e) { + throw invalid(path, dataType, "invalid ISO-8601 date", e); + } + }; + case TIME_WITHOUT_TIME_ZONE: + return timeConverter((TimeType) dataType); + case TIMESTAMP_WITHOUT_TIME_ZONE: + return timestampConverter((TimestampType) dataType); + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + return timestampLtzConverter((LocalZonedTimestampType) dataType); + default: + throw new KafkaTopicSchemaException( + "Kafka JSON format does not support Fluss type " + + dataType.asSummaryString() + + "."); + } + } + + private static JsonToFlussConverter decimalConverter(DecimalType dataType) { + return (node, path) -> { + require(node.isNumber(), path, dataType, "expected a JSON number"); + BigDecimal value = node.decimalValue(); + final BigDecimal scaled; + try { + scaled = value.setScale(dataType.getScale(), RoundingMode.UNNECESSARY); + } catch (ArithmeticException e) { + throw invalid(path, dataType, "decimal scale exceeds target", e); + } + Decimal decimal = + Decimal.fromBigDecimal(scaled, dataType.getPrecision(), dataType.getScale()); + if (decimal == null) { + throw invalid(path, dataType, "decimal precision exceeds target"); + } + return decimal; + }; + } + + private static JsonToFlussConverter charConverter(CharType dataType) { + return (node, path) -> { + String value = textualValue(node, path, dataType); + int codePoints = value.codePointCount(0, value.length()); + require( + codePoints <= dataType.getLength(), + path, + dataType, + "character length exceeds target"); + return BinaryString.fromString(value); + }; + } + + private static JsonToFlussConverter binaryConverter(BinaryType dataType) { + return (node, path) -> { + byte[] value = decodeBase64(node, path, dataType); + require( + value.length == dataType.getLength(), + path, + dataType, + "decoded binary length must equal " + dataType.getLength()); + return value; + }; + } + + private static JsonToFlussConverter timeConverter(TimeType dataType) { + return (node, path) -> { + String value = textualValue(node, path, dataType); + final LocalTime time; + try { + time = LocalTime.parse(value); + } catch (DateTimeParseException e) { + throw invalid(path, dataType, "invalid ISO-8601 local time", e); + } + validatePrecision(time.getNano(), dataType.getPrecision(), path, dataType); + require( + time.getNano() % 1_000_000 == 0, + path, + dataType, + "Fluss TIME storage currently requires millisecond precision"); + return Math.toIntExact(time.toNanoOfDay() / 1_000_000); + }; + } + + private static JsonToFlussConverter timestampConverter(TimestampType dataType) { + return (node, path) -> { + String value = textualValue(node, path, dataType); + final LocalDateTime timestamp; + try { + timestamp = LocalDateTime.parse(value); + } catch (DateTimeParseException e) { + throw invalid(path, dataType, "invalid ISO-8601 local timestamp", e); + } + validatePrecision(timestamp.getNano(), dataType.getPrecision(), path, dataType); + return TimestampNtz.fromLocalDateTime(timestamp); + }; + } + + private static JsonToFlussConverter timestampLtzConverter(LocalZonedTimestampType dataType) { + return (node, path) -> { + String value = textualValue(node, path, dataType); + final OffsetDateTime timestamp; + try { + timestamp = OffsetDateTime.parse(value); + } catch (DateTimeParseException e) { + throw invalid( + path, + dataType, + "invalid ISO-8601 timestamp; an offset or Z is required", + e); + } + validatePrecision(timestamp.getNano(), dataType.getPrecision(), path, dataType); + return TimestampLtz.fromInstant(timestamp.toInstant()); + }; + } + + private static long integralValue(JsonNode node, String path, DataType dataType) { + require(node.isIntegralNumber(), path, dataType, "expected a JSON integer"); + require(node.canConvertToLong(), path, dataType, "integer overflow"); + return node.longValue(); + } + + private static String textualValue(JsonNode node, String path, DataType dataType) { + require(node.isTextual(), path, dataType, "expected a JSON string"); + return node.textValue(); + } + + private static byte[] decodeBase64(JsonNode node, String path, DataType dataType) { + String value = textualValue(node, path, dataType); + try { + return Base64.getDecoder().decode(value); + } catch (IllegalArgumentException e) { + throw invalid(path, dataType, "invalid Base64 string", e); + } + } + + private static void validatePrecision( + int nanos, int precision, String path, DataType dataType) { + if (precision >= 9) { + return; + } + int unit = 1; + for (int i = precision; i < 9; i++) { + unit *= 10; + } + require(nanos % unit == 0, path, dataType, "fractional seconds exceed target precision"); + } + + private static void require(boolean condition, String path, DataType dataType, String reason) { + if (!condition) { + throw invalid(path, dataType, reason); + } + } + + private static KafkaRecordEncodingException invalid( + String path, DataType dataType, String reason) { + return new KafkaRecordEncodingException( + "Invalid Kafka record value at " + + path + + " for " + + dataType.asSummaryString() + + ": " + + reason + + "."); + } + + private static KafkaRecordEncodingException invalid( + String path, DataType dataType, String reason, @Nullable Throwable cause) { + return new KafkaRecordEncodingException( + "Invalid Kafka record value at " + + path + + " for " + + dataType.asSummaryString() + + ": " + + reason + + ".", + cause); + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java new file mode 100644 index 00000000000..785edd9e5e4 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java @@ -0,0 +1,140 @@ +/* + * 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.fluss.kafka.format.json; + +import org.apache.fluss.kafka.schema.KafkaTopicSchemaException; +import org.apache.fluss.kafka.transcode.KafkaRecordEncodingException; +import org.apache.fluss.row.BinaryString; +import org.apache.fluss.row.Decimal; +import org.apache.fluss.row.TimestampLtz; +import org.apache.fluss.row.TimestampNtz; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.math.BigDecimal; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Verifies strict scalar conversion without transport or Arrow dependencies. */ +class JsonToFlussConvertersTest { + private static final ObjectMapper MAPPER = + new ObjectMapper().enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS); + + @Test + void testNumericRepresentationsAndExactDecimals() throws Exception { + assertThat(convert(DataTypes.BOOLEAN(), "true")).isEqualTo(true); + assertThat(convert(DataTypes.TINYINT(), "-128")).isEqualTo((byte) -128); + assertThat(convert(DataTypes.SMALLINT(), "32767")).isEqualTo((short) 32767); + assertThat(convert(DataTypes.INT(), "2147483647")).isEqualTo(Integer.MAX_VALUE); + assertThat(convert(DataTypes.BIGINT(), "9223372036854775807")).isEqualTo(Long.MAX_VALUE); + assertThat(convert(DataTypes.FLOAT(), "1.25")).isEqualTo(1.25F); + assertThat(convert(DataTypes.DOUBLE(), "2.5")).isEqualTo(2.5D); + Decimal decimal = + (Decimal) convert(DataTypes.DECIMAL(30, 10), "12345678901234567890.1234567890"); + assertThat(decimal.toBigDecimal()) + .isEqualByComparingTo(new BigDecimal("12345678901234567890.1234567890")); + } + + @Test + void testStringsBinaryAndTemporalRepresentations() throws Exception { + assertThat(convert(DataTypes.CHAR(2), "\"😀a\"")).isEqualTo(BinaryString.fromString("😀a")); + assertThat(convert(DataTypes.STRING(), "\"文本\"")).isEqualTo(BinaryString.fromString("文本")); + assertThat((byte[]) convert(DataTypes.BINARY(2), "\"AQI=\"")).containsExactly(1, 2); + assertThat((byte[]) convert(DataTypes.BYTES(), "\"\"")).isEmpty(); + assertThat(convert(DataTypes.DATE(), "\"1970-01-02\"")) + .isEqualTo((int) LocalDate.of(1970, 1, 2).toEpochDay()); + assertThat(convert(DataTypes.TIME(3), "\"00:00:01.123\"")).isEqualTo(1123); + assertThat(convert(DataTypes.TIMESTAMP(3), "\"2026-09-11T12:00:00.123\"")) + .isEqualTo( + TimestampNtz.fromLocalDateTime( + LocalDateTime.of(2026, 9, 11, 12, 0, 0, 123000000))); + assertThat(convert(DataTypes.TIMESTAMP_LTZ(3), "\"2026-09-11T12:00:00.123+08:00\"")) + .isEqualTo(TimestampLtz.fromInstant(Instant.parse("2026-09-11T04:00:00.123Z"))); + } + + @Test + void testMissingAndNullRespectNullability() throws Exception { + assertThat(JsonToFlussConverters.create(DataTypes.INT()).convert(null, "$.field")).isNull(); + assertThat(convert(DataTypes.INT(), "null")).isNull(); + assertThatThrownBy( + () -> + JsonToFlussConverters.create(DataTypes.INT().copy(false)) + .convert(null, "$.field")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.field") + .hasMessageContaining("missing or null"); + assertThatThrownBy(() -> convert(DataTypes.INT().copy(false), "null")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("missing or null"); + } + + @ParameterizedTest + @MethodSource("invalidScalars") + void testRejectsCoercionOverflowAndPrecisionLoss(DataType type, String json) { + assertThatThrownBy(() -> convert(type, json)) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.field"); + } + + private static Stream invalidScalars() { + return Stream.of( + Arguments.of(DataTypes.BOOLEAN(), "1"), + Arguments.of(DataTypes.INT(), "\"1\""), + Arguments.of(DataTypes.INT(), "1.0"), + Arguments.of(DataTypes.INT(), "2147483648"), + Arguments.of(DataTypes.TINYINT(), "128"), + Arguments.of(DataTypes.SMALLINT(), "-32769"), + Arguments.of(DataTypes.BIGINT(), "9223372036854775808"), + Arguments.of(DataTypes.FLOAT(), "1e39"), + Arguments.of(DataTypes.DOUBLE(), "1e309"), + Arguments.of(DataTypes.DECIMAL(5, 2), "1.234"), + Arguments.of(DataTypes.DECIMAL(5, 2), "1000"), + Arguments.of(DataTypes.CHAR(1), "\"ab\""), + Arguments.of(DataTypes.STRING(), "false"), + Arguments.of(DataTypes.BINARY(2), "\"AQ==\""), + Arguments.of(DataTypes.BYTES(), "\"!invalid!\""), + Arguments.of(DataTypes.DATE(), "\"2026-02-30\""), + Arguments.of(DataTypes.TIME(0), "\"00:00:00.001\""), + Arguments.of(DataTypes.TIME(3), "\"00:00:00.0001\""), + Arguments.of(DataTypes.TIMESTAMP(3), "\"2026-09-11T00:00:00.0001\""), + Arguments.of(DataTypes.TIMESTAMP_LTZ(3), "\"2026-09-11T00:00:00\"")); + } + + @Test + void testComplexTypesRemainUnsupportedAtThisStage() { + assertThatThrownBy(() -> JsonToFlussConverters.create(DataTypes.ARRAY(DataTypes.INT()))) + .isInstanceOf(KafkaTopicSchemaException.class) + .hasMessageContaining("does not support"); + } + + private static Object convert(DataType type, String json) throws Exception { + return JsonToFlussConverters.create(type).convert(MAPPER.readTree(json), "$.field"); + } +} From 9762dbb282ae8ae3ab239a53594f9dfcdb8f46bb Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Sun, 20 Sep 2026 11:54:43 +0800 Subject: [PATCH 2/7] [kafka] Fix JSON scalar and timestamp boundary conversions Bound decimal rescaling, reject unpaired Unicode surrogates, and convert FLOAT directly to avoid double rounding. Check timestamp storage overflow and preserve negative and large-date timestamps when reading them back. Add regression coverage for the conversion and Arrow round-trip boundaries. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 147/147 AI-Contributed/UT: 341/341 --- .../org/apache/fluss/row/TimestampNtz.java | 8 +- .../ArrowTimestampLtzColumnVector.java | 6 +- .../ArrowTimestampNtzColumnVector.java | 6 +- .../writers/ArrowTimestampLtzWriter.java | 9 +- .../writers/ArrowTimestampNtzWriter.java | 6 +- .../org/apache/fluss/utils/DateTimeUtils.java | 29 ++++ .../apache/fluss/row/TimestampNtzTest.java | 25 +++ .../row/arrow/ArrowTimestampWriterTest.java | 153 ++++++++++++++++ .../format/json/JsonToFlussConverters.java | 83 +++++++-- .../json/JsonToFlussConvertersTest.java | 163 ++++++++++++++++++ 10 files changed, 463 insertions(+), 25 deletions(-) create mode 100644 fluss-common/src/test/java/org/apache/fluss/row/arrow/ArrowTimestampWriterTest.java diff --git a/fluss-common/src/main/java/org/apache/fluss/row/TimestampNtz.java b/fluss-common/src/main/java/org/apache/fluss/row/TimestampNtz.java index c58141da950..c7dd14b9387 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/TimestampNtz.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/TimestampNtz.java @@ -137,12 +137,8 @@ public long toEpochMicros() { /** Converts this {@link TimestampNtz} object to a {@link LocalDateTime}. */ public LocalDateTime toLocalDateTime() { - int date = (int) (millisecond / MILLIS_PER_DAY); - long time = millisecond % MILLIS_PER_DAY; - if (time < 0) { - --date; - time += MILLIS_PER_DAY; - } + long date = Math.floorDiv(millisecond, MILLIS_PER_DAY); + long time = Math.floorMod(millisecond, MILLIS_PER_DAY); long nanoOfDay = time * 1_000_000L + nanoOfMillisecond; LocalDate localDate = LocalDate.ofEpochDay(date); LocalTime localTime = LocalTime.ofNanoOfDay(nanoOfDay); diff --git a/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampLtzColumnVector.java b/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampLtzColumnVector.java index 9b870cdf1ff..9af6601bc53 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampLtzColumnVector.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampLtzColumnVector.java @@ -57,10 +57,12 @@ public TimestampLtz getTimestampLtz(int i, int precision) { return TimestampLtz.fromEpochMillis(((TimeStampMilliVector) valueVector).get(i)); } else if (valueVector instanceof TimeStampMicroVector) { long micros = ((TimeStampMicroVector) valueVector).get(i); - return TimestampLtz.fromEpochMillis(micros / 1000, (int) (micros % 1000) * 1000); + return TimestampLtz.fromEpochMillis( + Math.floorDiv(micros, 1_000L), (int) Math.floorMod(micros, 1_000L) * 1_000); } else { long nanos = ((TimeStampNanoVector) valueVector).get(i); - return TimestampLtz.fromEpochMillis(nanos / 1_000_000, (int) (nanos % 1_000_000)); + return TimestampLtz.fromEpochMillis( + Math.floorDiv(nanos, 1_000_000L), (int) Math.floorMod(nanos, 1_000_000L)); } } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampNtzColumnVector.java b/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampNtzColumnVector.java index 37bc9f7b9df..19ac10479c1 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampNtzColumnVector.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampNtzColumnVector.java @@ -56,10 +56,12 @@ public TimestampNtz getTimestampNtz(int i, int precision) { return TimestampNtz.fromMillis(((TimeStampMilliVector) valueVector).get(i)); } else if (valueVector instanceof TimeStampMicroVector) { long micros = ((TimeStampMicroVector) valueVector).get(i); - return TimestampNtz.fromMillis(micros / 1000, (int) (micros % 1000) * 1000); + return TimestampNtz.fromMillis( + Math.floorDiv(micros, 1_000L), (int) Math.floorMod(micros, 1_000L) * 1_000); } else { long nanos = ((TimeStampNanoVector) valueVector).get(i); - return TimestampNtz.fromMillis(nanos / 1_000_000, (int) (nanos % 1_000_000)); + return TimestampNtz.fromMillis( + Math.floorDiv(nanos, 1_000_000L), (int) Math.floorMod(nanos, 1_000_000L)); } } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampLtzWriter.java b/fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampLtzWriter.java index fe10b374e34..42a4bbc582f 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampLtzWriter.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampLtzWriter.java @@ -27,6 +27,8 @@ import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.TimeStampVector; import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.types.pojo.ArrowType; +import static org.apache.fluss.utils.DateTimeUtils.timestampToMicros; +import static org.apache.fluss.utils.DateTimeUtils.timestampToNanos; import static org.apache.fluss.utils.Preconditions.checkState; /** {@link ArrowFieldWriter} for TimestampLtz. */ @@ -64,8 +66,8 @@ public void doWrite(int rowIndex, DataGetters row, int ordinal, boolean handleSa } } else if (vector instanceof TimeStampMicroVector) { long microSec = - timestamp.getEpochMillisecond() * 1000 - + timestamp.getNanoOfMillisecond() / 1000; + timestampToMicros( + timestamp.getEpochMillisecond(), timestamp.getNanoOfMillisecond()); if (handleSafe) { vector.setSafe(rowIndex, microSec); } else { @@ -73,7 +75,8 @@ public void doWrite(int rowIndex, DataGetters row, int ordinal, boolean handleSa } } else { long nanoSec = - timestamp.getEpochMillisecond() * 1_000_000 + timestamp.getNanoOfMillisecond(); + timestampToNanos( + timestamp.getEpochMillisecond(), timestamp.getNanoOfMillisecond()); if (handleSafe) { vector.setSafe(rowIndex, nanoSec); } else { diff --git a/fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampNtzWriter.java b/fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampNtzWriter.java index 89f965ad362..5d79bff0d1b 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampNtzWriter.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampNtzWriter.java @@ -27,6 +27,8 @@ import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.TimeStampVector; import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.types.pojo.ArrowType; +import static org.apache.fluss.utils.DateTimeUtils.timestampToMicros; +import static org.apache.fluss.utils.DateTimeUtils.timestampToNanos; import static org.apache.fluss.utils.Preconditions.checkState; /** {@link ArrowFieldWriter} for TimestampNtz. */ @@ -64,7 +66,7 @@ public void doWrite(int rowIndex, DataGetters row, int ordinal, boolean handleSa } } else if (vector instanceof TimeStampMicroVector) { long microSec = - timestamp.getMillisecond() * 1000 + timestamp.getNanoOfMillisecond() / 1000; + timestampToMicros(timestamp.getMillisecond(), timestamp.getNanoOfMillisecond()); if (handleSafe) { vector.setSafe(rowIndex, microSec); } else { @@ -72,7 +74,7 @@ public void doWrite(int rowIndex, DataGetters row, int ordinal, boolean handleSa } } else { long nanoSec = - timestamp.getMillisecond() * 1_000_000 + timestamp.getNanoOfMillisecond(); + timestampToNanos(timestamp.getMillisecond(), timestamp.getNanoOfMillisecond()); if (handleSafe) { vector.setSafe(rowIndex, nanoSec); } else { diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/DateTimeUtils.java b/fluss-common/src/main/java/org/apache/fluss/utils/DateTimeUtils.java index 3019df4e33c..1140c8c1165 100644 --- a/fluss-common/src/main/java/org/apache/fluss/utils/DateTimeUtils.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/DateTimeUtils.java @@ -71,6 +71,35 @@ public class DateTimeUtils { .optionalEnd() .toFormatter(); + /** + * Converts a timestamp's milliseconds and nanoseconds within the millisecond to microseconds. + * Sub-microsecond digits are discarded. + * + * @throws ArithmeticException if the result does not fit in a long + */ + public static long timestampToMicros(long millisecond, int nanoOfMillisecond) { + return scaleTimestamp(millisecond, MICROS_PER_MILLIS, nanoOfMillisecond / NANOS_PER_MICROS); + } + + /** + * Converts a timestamp's milliseconds and nanoseconds within the millisecond to nanoseconds. + * + * @throws ArithmeticException if the result does not fit in a long + */ + public static long timestampToNanos(long millisecond, int nanoOfMillisecond) { + return scaleTimestamp(millisecond, 1_000_000L, nanoOfMillisecond); + } + + private static long scaleTimestamp(long millisecond, long unitsPerMillis, long fraction) { + if (millisecond < 0) { + // The non-negative fraction can bring the final value back into range even when + // multiplying the floor-rounded millisecond would underflow at Long.MIN_VALUE. + return Math.addExact( + Math.multiplyExact(millisecond + 1, unitsPerMillis), fraction - unitsPerMillis); + } + return Math.addExact(Math.multiplyExact(millisecond, unitsPerMillis), fraction); + } + public static LocalDate toLocalDate(int date) { return julianToLocalDate(date + EPOCH_JULIAN); } diff --git a/fluss-common/src/test/java/org/apache/fluss/row/TimestampNtzTest.java b/fluss-common/src/test/java/org/apache/fluss/row/TimestampNtzTest.java index 1a939842450..dcd4df406b9 100644 --- a/fluss-common/src/test/java/org/apache/fluss/row/TimestampNtzTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/row/TimestampNtzTest.java @@ -19,7 +19,9 @@ import org.junit.jupiter.api.Test; +import java.time.Instant; import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.TimeZone; import static org.assertj.core.api.Assertions.assertThat; @@ -46,6 +48,29 @@ public void testNormal() { assertThat(TimestampNtz.fromLocalDateTime(ldt3).toLocalDateTime()).isEqualTo(ldt3); } + @Test + void testToLocalDateTimeAcrossMillisecondRange() { + long[] milliseconds = { + Long.MIN_VALUE, + (long) Integer.MIN_VALUE * 86_400_000 - 1, + -86_400_001, + -1, + 0, + 86_400_000, + ((long) Integer.MAX_VALUE + 1) * 86_400_000, + Long.MAX_VALUE + }; + for (long millis : milliseconds) { + for (int nanos : new int[] {0, 999_999}) { + LocalDateTime expected = + LocalDateTime.ofInstant( + Instant.ofEpochMilli(millis).plusNanos(nanos), ZoneOffset.UTC); + assertThat(TimestampNtz.fromMillis(millis, nanos).toLocalDateTime()) + .isEqualTo(expected); + } + } + } + @Test public void testDaylightSavingTime() { TimeZone tz = TimeZone.getDefault(); diff --git a/fluss-common/src/test/java/org/apache/fluss/row/arrow/ArrowTimestampWriterTest.java b/fluss-common/src/test/java/org/apache/fluss/row/arrow/ArrowTimestampWriterTest.java new file mode 100644 index 00000000000..5c5becd1a8e --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/row/arrow/ArrowTimestampWriterTest.java @@ -0,0 +1,153 @@ +/* + * 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.fluss.row.arrow; + +import org.apache.fluss.row.GenericRow; +import org.apache.fluss.row.TimestampLtz; +import org.apache.fluss.row.TimestampNtz; +import org.apache.fluss.row.arrow.writers.ArrowFieldWriter; +import org.apache.fluss.row.columnar.ColumnVector; +import org.apache.fluss.row.columnar.ColumnarRow; +import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.RootAllocator; +import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.TimeStampVector; +import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.types.RowType; +import org.apache.fluss.utils.ArrowUtils; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Verifies timestamp range checks and round trips through Arrow vectors. */ +class ArrowTimestampWriterTest { + + @ParameterizedTest + @MethodSource("timestampWriters") + void testRoundTripsTimestampBoundaries(int precision, boolean withZone, boolean handleSafe) { + DataType type = + withZone ? DataTypes.TIMESTAMP_LTZ(precision) : DataTypes.TIMESTAMP(precision); + RowType rowType = DataTypes.ROW(type); + long unitsPerMilli = precision == 6 ? 1_000 : 1_000_000; + int nanosPerUnit = precision == 6 ? 1_000 : 1; + try (RootAllocator allocator = new RootAllocator(1024 * 1024); + VectorSchemaRoot root = + VectorSchemaRoot.create(ArrowUtils.toArrowSchema(rowType), allocator)) { + root.allocateNew(); + TimeStampVector vector = (TimeStampVector) root.getVector(0); + ArrowFieldWriter writer = ArrowUtils.createArrowFieldWriter(vector, type); + long[] values = { + Long.MIN_VALUE, + -1_000_001, + -unitsPerMilli, + -unitsPerMilli + 1, + -1, + 0, + 1, + 1_000_001, + Long.MAX_VALUE + }; + for (int i = 0; i < values.length; i++) { + long millis = Math.floorDiv(values[i], unitsPerMilli); + int nanos = (int) Math.floorMod(values[i], unitsPerMilli) * nanosPerUnit; + writer.write(i, timestampRow(millis, nanos, withZone), 0, handleSafe); + assertThat(vector.get(i)).isEqualTo(values[i]); + } + root.setRowCount(values.length); + ArrowReader reader = + new ArrowReader( + new ColumnVector[] {ArrowUtils.createArrowColumnVector(vector, type)}, + root.getRowCount()); + for (int i = 0; i < values.length; i++) { + long millis = Math.floorDiv(values[i], unitsPerMilli); + int nanos = (int) Math.floorMod(values[i], unitsPerMilli) * nanosPerUnit; + ColumnarRow row = reader.read(i); + if (withZone) { + assertThat(row.getTimestampLtz(0, precision)) + .isEqualTo(TimestampLtz.fromEpochMillis(millis, nanos)); + } else { + assertThat(row.getTimestampNtz(0, precision)) + .isEqualTo(TimestampNtz.fromMillis(millis, nanos)); + } + } + } + } + + @ParameterizedTest + @MethodSource("timestampWriters") + void testRejectsOverflowBeforeChangingVector( + int precision, boolean withZone, boolean handleSafe) { + DataType type = + withZone ? DataTypes.TIMESTAMP_LTZ(precision) : DataTypes.TIMESTAMP(precision); + RowType rowType = DataTypes.ROW(type); + long unitsPerMilli = precision == 6 ? 1_000 : 1_000_000; + int nanosPerUnit = precision == 6 ? 1_000 : 1; + try (RootAllocator allocator = new RootAllocator(1024 * 1024); + VectorSchemaRoot root = + VectorSchemaRoot.create(ArrowUtils.toArrowSchema(rowType), allocator)) { + root.allocateNew(); + TimeStampVector vector = (TimeStampVector) root.getVector(0); + ArrowFieldWriter writer = ArrowUtils.createArrowFieldWriter(vector, type); + vector.set(0, 42); + for (long limit : new long[] {Long.MIN_VALUE, Long.MAX_VALUE}) { + long millis = Math.floorDiv(limit, unitsPerMilli); + int nanos = (int) Math.floorMod(limit, unitsPerMilli) * nanosPerUnit; + int outsideNanos = nanos + (limit < 0 ? -nanosPerUnit : nanosPerUnit); + assertThatThrownBy( + () -> + writer.write( + 0, + timestampRow(millis, outsideNanos, withZone), + 0, + handleSafe)) + .isInstanceOf(ArithmeticException.class); + long outsideMillis = millis + (limit < 0 ? -1 : 1); + assertThatThrownBy( + () -> + writer.write( + 0, + timestampRow(outsideMillis, nanos, withZone), + 0, + handleSafe)) + .isInstanceOf(ArithmeticException.class); + assertThat(vector.get(0)).isEqualTo(42); + } + } + } + + private static GenericRow timestampRow(long millis, int nanos, boolean withZone) { + return GenericRow.of( + withZone + ? TimestampLtz.fromEpochMillis(millis, nanos) + : TimestampNtz.fromMillis(millis, nanos)); + } + + private static Stream timestampWriters() { + return Stream.of( + Arguments.of(6, false, false), Arguments.of(6, false, true), + Arguments.of(6, true, false), Arguments.of(6, true, true), + Arguments.of(9, false, false), Arguments.of(9, false, true), + Arguments.of(9, true, false), Arguments.of(9, true, true)); + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java index b3dc7e478b6..79fb79c98d1 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java @@ -37,13 +37,18 @@ import java.math.BigDecimal; import java.math.RoundingMode; import java.time.DateTimeException; +import java.time.Instant; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; import java.time.OffsetDateTime; +import java.time.ZoneOffset; import java.time.format.DateTimeParseException; import java.util.Base64; +import static org.apache.fluss.utils.DateTimeUtils.timestampToMicros; +import static org.apache.fluss.utils.DateTimeUtils.timestampToNanos; + /** Strict JSON-to-Fluss scalar converter construction. */ final class JsonToFlussConverters { @@ -110,10 +115,9 @@ private static JsonToFlussConverter createNotNull(DataType dataType) { case FLOAT: return (node, path) -> { require(node.isNumber(), path, dataType, "expected a JSON number"); - double doubleValue = node.doubleValue(); - float value = (float) doubleValue; + float value = node.floatValue(); require( - Double.isFinite(doubleValue) && Float.isFinite(value), + Float.isFinite(value), path, dataType, "non-finite or overflowing number"); @@ -135,10 +139,7 @@ private static JsonToFlussConverter createNotNull(DataType dataType) { case CHAR: return charConverter((CharType) dataType); case STRING: - return (node, path) -> { - require(node.isTextual(), path, dataType, "expected a JSON string"); - return BinaryString.fromString(node.textValue()); - }; + return (node, path) -> BinaryString.fromString(textualValue(node, path, dataType)); case BINARY: return binaryConverter((BinaryType) dataType); case BYTES: @@ -170,6 +171,22 @@ private static JsonToFlussConverter decimalConverter(DecimalType dataType) { return (node, path) -> { require(node.isNumber(), path, dataType, "expected a JSON number"); BigDecimal value = node.decimalValue(); + if (value.signum() == 0) { + return Decimal.zero(dataType.getPrecision(), dataType.getScale()); + } + // Check the resulting digits before setScale can expand a large JSON exponent. + // Use long arithmetic because BigDecimal scales may span the entire int range. + long scaledPrecision = (long) value.precision() - value.scale() + dataType.getScale(); + require( + scaledPrecision <= dataType.getPrecision(), + path, + dataType, + "decimal precision exceeds target"); + require( + value.stripTrailingZeros().scale() <= dataType.getScale(), + path, + dataType, + "decimal scale exceeds target"); final BigDecimal scaled; try { scaled = value.setScale(dataType.getScale(), RoundingMode.UNNECESSARY); @@ -239,7 +256,13 @@ private static JsonToFlussConverter timestampConverter(TimestampType dataType) { throw invalid(path, dataType, "invalid ISO-8601 local timestamp", e); } validatePrecision(timestamp.getNano(), dataType.getPrecision(), path, dataType); - return TimestampNtz.fromLocalDateTime(timestamp); + long millis = + timestampMillis( + timestamp.toInstant(ZoneOffset.UTC), + dataType.getPrecision(), + path, + dataType); + return TimestampNtz.fromMillis(millis, timestamp.getNano() % 1_000_000); }; } @@ -257,10 +280,30 @@ private static JsonToFlussConverter timestampLtzConverter(LocalZonedTimestampTyp e); } validatePrecision(timestamp.getNano(), dataType.getPrecision(), path, dataType); - return TimestampLtz.fromInstant(timestamp.toInstant()); + long millis = + timestampMillis(timestamp.toInstant(), dataType.getPrecision(), path, dataType); + return TimestampLtz.fromEpochMillis(millis, timestamp.getNano() % 1_000_000); }; } + private static long timestampMillis( + Instant timestamp, int precision, String path, DataType dataType) { + try { + long millis = timestamp.toEpochMilli(); + int nanos = timestamp.getNano() % 1_000_000; + // Arrow stores precision 4-6 in microseconds and precision 7-9 in nanoseconds. + // Check here as well so Kafka reports the offending JSON field before encoding. + if (precision > 6) { + timestampToNanos(millis, nanos); + } else if (precision > 3) { + timestampToMicros(millis, nanos); + } + return millis; + } catch (ArithmeticException e) { + throw invalid(path, dataType, "timestamp exceeds storage range", e); + } + } + private static long integralValue(JsonNode node, String path, DataType dataType) { require(node.isIntegralNumber(), path, dataType, "expected a JSON integer"); require(node.canConvertToLong(), path, dataType, "integer overflow"); @@ -269,7 +312,27 @@ private static long integralValue(JsonNode node, String path, DataType dataType) private static String textualValue(JsonNode node, String path, DataType dataType) { require(node.isTextual(), path, dataType, "expected a JSON string"); - return node.textValue(); + String value = node.textValue(); + // Jackson can retain unpaired JSON escapes in Java strings. Reject them before + // BinaryString's UTF-8 encoding silently replaces them with a question mark. + for (int i = 0; i < value.length(); i++) { + char character = value.charAt(i); + if (Character.isHighSurrogate(character)) { + require( + i + 1 < value.length() && Character.isLowSurrogate(value.charAt(i + 1)), + path, + dataType, + "unpaired UTF-16 surrogate"); + i++; + } else { + require( + !Character.isLowSurrogate(character), + path, + dataType, + "unpaired UTF-16 surrogate"); + } + } + return value; } private static byte[] decodeBase64(JsonNode node, String path, DataType dataType) { diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java index 785edd9e5e4..05a18072c89 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java @@ -25,6 +25,7 @@ import org.apache.fluss.row.TimestampNtz; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.DecimalNode; import org.apache.fluss.types.DataType; import org.apache.fluss.types.DataTypes; @@ -32,11 +33,14 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; import java.math.BigDecimal; +import java.math.BigInteger; import java.time.Instant; import java.time.LocalDate; import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; @@ -62,6 +66,29 @@ void testNumericRepresentationsAndExactDecimals() throws Exception { .isEqualByComparingTo(new BigDecimal("12345678901234567890.1234567890")); } + @ParameterizedTest + @ValueSource(strings = {"", "-"}) + void testFloatRoundsDirectlyWithoutDoubleRounding(String sign) throws Exception { + float direction = sign.isEmpty() ? 1.0F : -1.0F; + assertThat(convert(DataTypes.FLOAT(), sign + "1.0000000596046448")) + .isEqualTo(direction * 1.0000001F); + assertThat(convert(DataTypes.FLOAT(), sign + "1.0000000596046447")).isEqualTo(direction); + // The integer and decimal nodes must both round to the largest finite float. + String belowOverflow = "340282356779733661637539395458142568447"; + assertThat(convert(DataTypes.FLOAT(), sign + belowOverflow)) + .isEqualTo(direction * Float.MAX_VALUE); + assertThat(convert(DataTypes.FLOAT(), sign + belowOverflow + ".0")) + .isEqualTo(direction * Float.MAX_VALUE); + assertThatThrownBy( + () -> + convert( + DataTypes.FLOAT(), + sign + "340282356779733661637539395458142568448")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.field") + .hasMessageContaining("overflowing number"); + } + @Test void testStringsBinaryAndTemporalRepresentations() throws Exception { assertThat(convert(DataTypes.CHAR(2), "\"😀a\"")).isEqualTo(BinaryString.fromString("😀a")); @@ -95,6 +122,142 @@ void testMissingAndNullRespectNullability() throws Exception { .hasMessageContaining("missing or null"); } + @Test + void testDecimalBoundsBeforeRescaling() throws Exception { + for (int scale : new int[] {100_000_000, Integer.MAX_VALUE, Integer.MIN_VALUE}) { + assertThatThrownBy( + () -> + JsonToFlussConverters.create(DataTypes.DECIMAL(5, 2)) + .convert( + DecimalNode.valueOf( + new BigDecimal(BigInteger.ONE, scale)), + "$.amount")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.amount") + .hasMessageContaining(scale > 0 ? "scale" : "precision"); + Decimal zero = + (Decimal) + JsonToFlussConverters.create(DataTypes.DECIMAL(5, 2)) + .convert( + DecimalNode.valueOf( + new BigDecimal(BigInteger.ZERO, scale)), + "$.amount"); + assertThat(zero.toBigDecimal()).isEqualTo(new BigDecimal("0.00")); + } + for (String json : new String[] {"12.3400", "1.234e1", "-12.3400", "1e2", "1e-2"}) { + Decimal decimal = (Decimal) convert(DataTypes.DECIMAL(5, 2), json); + assertThat(decimal.toBigDecimal()).isEqualByComparingTo(new BigDecimal(json)); + } + } + + @ParameterizedTest + @ValueSource(strings = {"\\uD800", "\\uDC00", "a\\uD800b", "\\uD800\\uD800", "\\uDC00\\uD800"}) + void testRejectsUnpairedSurrogates(String escaped) { + for (DataType type : new DataType[] {DataTypes.STRING(), DataTypes.CHAR(8)}) { + assertThatThrownBy(() -> convert(type, "\"" + escaped + "\"")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.field") + .hasMessageContaining("surrogate"); + } + } + + @Test + void testAcceptsPairedSurrogatesAndLiteralEscape() throws Exception { + String escapedEmoji = "\"\\uD83D\\uDE00\""; + assertThat(convert(DataTypes.STRING(), escapedEmoji)) + .isEqualTo(BinaryString.fromString("😀")); + assertThat(convert(DataTypes.CHAR(1), escapedEmoji)) + .isEqualTo(BinaryString.fromString("😀")); + assertThat(convert(DataTypes.STRING(), "\"\\\\uD800\"")) + .isEqualTo(BinaryString.fromString("\\uD800")); + } + + @ParameterizedTest + @ValueSource(ints = {3, 6, 9}) + void testTimestampStorageBoundaries(int precision) throws Exception { + long unitsPerSecond = precision == 3 ? 1_000 : precision == 6 ? 1_000_000 : 1_000_000_000; + long nanosPerUnit = 1_000_000_000 / unitsPerSecond; + for (long units : new long[] {Long.MIN_VALUE, -1, 0, 1, Long.MAX_VALUE}) { + Instant instant = + Instant.ofEpochSecond( + Math.floorDiv(units, unitsPerSecond), + Math.floorMod(units, unitsPerSecond) * nanosPerUnit); + for (boolean withZone : new boolean[] {false, true}) { + Object actual = + convert( + timestampType(precision, withZone), + timestampJson(instant, withZone)); + Object expected = + withZone + ? TimestampLtz.fromEpochMillis( + instant.toEpochMilli(), instant.getNano() % 1_000_000) + : TimestampNtz.fromMillis( + instant.toEpochMilli(), instant.getNano() % 1_000_000); + assertThat(actual).isEqualTo(expected); + if (withZone) { + assertThat(((TimestampLtz) actual).toInstant()).isEqualTo(instant); + } else { + assertThat(((TimestampNtz) actual).toLocalDateTime()) + .isEqualTo(LocalDateTime.ofInstant(instant, ZoneOffset.UTC)); + } + } + } + for (long limit : new long[] {Long.MIN_VALUE, Long.MAX_VALUE}) { + Instant outside = + Instant.ofEpochSecond( + Math.floorDiv(limit, unitsPerSecond), + Math.floorMod(limit, unitsPerSecond) * nanosPerUnit) + .plusNanos(limit < 0 ? -nanosPerUnit : nanosPerUnit); + for (boolean withZone : new boolean[] {false, true}) { + assertThatThrownBy( + () -> + convert( + timestampType(precision, withZone), + timestampJson(outside, withZone))) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.field") + .hasMessageContaining("timestamp exceeds storage range"); + } + } + } + + @Test + void testRejectsTimestampOverflowAndKeepsRepresentableDates() throws Exception { + for (boolean withZone : new boolean[] {false, true}) { + String suffix = withZone ? "Z" : ""; + for (String date : new String[] {"3000-01-01T00:00:00", "1600-01-01T00:00:00"}) { + assertThatThrownBy( + () -> + convert( + timestampType(9, withZone), + "\"" + date + suffix + "\"")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("timestamp exceeds storage range"); + assertThat(convert(timestampType(6, withZone), "\"" + date + suffix + "\"")) + .isNotNull(); + } + assertThatThrownBy( + () -> + convert( + timestampType(3, withZone), + "\"+999999999-01-01T00:00:00" + suffix + "\"")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("timestamp exceeds storage range"); + } + } + + private static DataType timestampType(int precision, boolean withZone) { + return withZone ? DataTypes.TIMESTAMP_LTZ(precision) : DataTypes.TIMESTAMP(precision); + } + + private static String timestampJson(Instant instant, boolean withZone) { + return "\"" + + (withZone + ? instant.toString() + : LocalDateTime.ofInstant(instant, ZoneOffset.UTC).toString()) + + "\""; + } + @ParameterizedTest @MethodSource("invalidScalars") void testRejectsCoercionOverflowAndPrecisionLoss(DataType type, String json) { From 5b24673d4e2f326d718109978e7b208bd52b2f71 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Mon, 12 Oct 2026 21:14:56 +0800 Subject: [PATCH 3/7] [kafka] Reuse public JSON schema validation for scalar converters Validate schemas with KafkaTableSchema before converter construction and translate failures to KafkaTopicSchemaException. Cover shared map-key and depth rules while retaining explicit rejection of complex converters. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 12/12 AI-Contributed/UT: 67/67 --- .../format/json/JsonToFlussConverters.java | 12 ++++ .../json/JsonToFlussConvertersTest.java | 67 ++++++++++++++++++- 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java index 79fb79c98d1..eb1433f4e78 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java @@ -17,8 +17,10 @@ package org.apache.fluss.kafka.format.json; +import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.kafka.schema.KafkaTopicSchemaException; import org.apache.fluss.kafka.transcode.KafkaRecordEncodingException; +import org.apache.fluss.metadata.KafkaTableSchema; import org.apache.fluss.row.BinaryString; import org.apache.fluss.row.Decimal; import org.apache.fluss.row.TimestampLtz; @@ -54,7 +56,17 @@ final class JsonToFlussConverters { private JsonToFlussConverters() {} + /** + * Creates a converter for one top-level JSON business field, excluding the JSON root container. + * The shared schema contract is validated before checking decoder support; complex types still + * require their own converter implementations. + */ static JsonToFlussConverter create(DataType dataType) { + try { + KafkaTableSchema.validateJsonType(dataType); + } catch (InvalidTableException e) { + throw new KafkaTopicSchemaException(e.getMessage()); + } final JsonToFlussConverter notNullConverter = createNotNull(dataType); return (node, path) -> { if (node == null || node.isNull()) { diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java index 05a18072c89..f8d35148564 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java @@ -17,8 +17,10 @@ package org.apache.fluss.kafka.format.json; +import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.kafka.schema.KafkaTopicSchemaException; import org.apache.fluss.kafka.transcode.KafkaRecordEncodingException; +import org.apache.fluss.metadata.KafkaTableSchema; import org.apache.fluss.row.BinaryString; import org.apache.fluss.row.Decimal; import org.apache.fluss.row.TimestampLtz; @@ -292,9 +294,72 @@ private static Stream invalidScalars() { @Test void testComplexTypesRemainUnsupportedAtThisStage() { - assertThatThrownBy(() -> JsonToFlussConverters.create(DataTypes.ARRAY(DataTypes.INT()))) + for (DataType type : + new DataType[] { + DataTypes.ARRAY(DataTypes.INT()), + DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()), + DataTypes.ROW(DataTypes.INT()) + }) { + KafkaTableSchema.validateJsonType(type); + assertThatThrownBy(() -> JsonToFlussConverters.create(type)) + .isInstanceOf(KafkaTopicSchemaException.class) + .hasMessageContaining("does not support"); + } + } + + @ParameterizedTest + @MethodSource("invalidMapKeyTypes") + void testInvalidMapKeysUseThePublicSchemaError(DataType type) { + assertThatThrownBy(() -> KafkaTableSchema.validateJsonType(type)) + .isInstanceOf(InvalidTableException.class) + .hasMessageContaining("only supports STRING map keys"); + assertThatThrownBy(() -> JsonToFlussConverters.create(type)) + .isInstanceOf(KafkaTopicSchemaException.class) + .hasMessageContaining("only supports STRING map keys"); + } + + private static Stream invalidMapKeyTypes() { + return Stream.of( + DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()), + DataTypes.MAP(DataTypes.CHAR(8), DataTypes.STRING()), + DataTypes.ROW(DataTypes.ARRAY(DataTypes.MAP(DataTypes.BYTES(), DataTypes.INT()))), + DataTypes.MAP( + DataTypes.STRING(), DataTypes.MAP(DataTypes.BOOLEAN(), DataTypes.INT()))); + } + + @Test + void testSchemaDepthUsesTheSharedBusinessFieldLimit() { + DataType type = DataTypes.INT(); + for (int depth = 0; depth < KafkaTableSchema.MAX_JSON_SCHEMA_NESTING_DEPTH; depth++) { + switch (depth % 3) { + case 0: + type = DataTypes.ROW(type); + break; + case 1: + type = DataTypes.ARRAY(type); + break; + default: + type = DataTypes.MAP(DataTypes.STRING(), type); + } + } + // The top-level business field starts at zero; each child consumes one level. + DataType atLimit = type; + KafkaTableSchema.validateJsonType(atLimit); + assertThatThrownBy(() -> JsonToFlussConverters.create(atLimit)) .isInstanceOf(KafkaTopicSchemaException.class) .hasMessageContaining("does not support"); + + DataType tooDeep = DataTypes.ROW(atLimit); + String message = + "Kafka JSON schema exceeds the maximum nesting depth of " + + KafkaTableSchema.MAX_JSON_SCHEMA_NESTING_DEPTH + + "."; + assertThatThrownBy(() -> KafkaTableSchema.validateJsonType(tooDeep)) + .isInstanceOf(InvalidTableException.class) + .hasMessage(message); + assertThatThrownBy(() -> JsonToFlussConverters.create(tooDeep)) + .isInstanceOf(KafkaTopicSchemaException.class) + .hasMessage(message); } private static Object convert(DataType type, String json) throws Exception { From a7eac151f4de8a9accfa02aab497374a200eb354 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Thu, 15 Oct 2026 23:29:09 +0800 Subject: [PATCH 4/7] [kafka] Cover JSON scalar edge cases on the updated FIP base Verify signed floating-point zero and underflow, rejection of non-finite numeric nodes and malformed Base64, and timestamp offsets before epoch. Preserve the existing scalar conversion and Arrow timestamp behavior. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 0/0 AI-Contributed/UT: 74/74 --- .../json/JsonToFlussConvertersTest.java | 74 +++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java index f8d35148564..c7273ca7401 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java @@ -26,8 +26,11 @@ import org.apache.fluss.row.TimestampLtz; import org.apache.fluss.row.TimestampNtz; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.DecimalNode; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.DoubleNode; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.FloatNode; import org.apache.fluss.types.DataType; import org.apache.fluss.types.DataTypes; @@ -91,6 +94,77 @@ void testFloatRoundsDirectlyWithoutDoubleRounding(String sign) throws Exception .hasMessageContaining("overflowing number"); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testFloatingPointPreservesZeroSignIncludingUnderflow(boolean negative) { + double zero = negative ? -0.0D : 0.0D; + for (JsonNode node : + new JsonNode[] { + DoubleNode.valueOf(zero), + FloatNode.valueOf((float) zero), + DecimalNode.valueOf(new BigDecimal(negative ? "-1e-1000" : "1e-1000")) + }) { + float floatValue = + (float) JsonToFlussConverters.create(DataTypes.FLOAT()).convert(node, "$.f"); + double doubleValue = + (double) JsonToFlussConverters.create(DataTypes.DOUBLE()).convert(node, "$.d"); + assertThat(Float.floatToRawIntBits(floatValue)) + .isEqualTo(Float.floatToRawIntBits((float) zero)); + assertThat(Double.doubleToRawLongBits(doubleValue)) + .isEqualTo(Double.doubleToRawLongBits(zero)); + } + } + + @ParameterizedTest + @MethodSource("nonFiniteNumbers") + void testRejectsNonFiniteNumericNodes(JsonNode node) { + for (DataType type : + new DataType[] {DataTypes.FLOAT(), DataTypes.DOUBLE(), DataTypes.DECIMAL(38, 18)}) { + assertThatThrownBy(() -> JsonToFlussConverters.create(type).convert(node, "$.number")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.number") + .hasMessageContaining(type.asSummaryString()); + } + } + + private static Stream nonFiniteNumbers() { + return Stream.of( + DoubleNode.valueOf(Double.NaN), + DoubleNode.valueOf(Double.NEGATIVE_INFINITY), + DoubleNode.valueOf(Double.POSITIVE_INFINITY), + FloatNode.valueOf(Float.NaN), + FloatNode.valueOf(Float.NEGATIVE_INFINITY), + FloatNode.valueOf(Float.POSITIVE_INFINITY)); + } + + @ParameterizedTest + @ValueSource(strings = {"A", "AQ===", "AQ==x", "A Q==", "AQ==\n", "_w=="}) + void testBase64RejectsInvalidPaddingAlphabetAndWhitespace(String value) { + for (DataType type : new DataType[] {DataTypes.BYTES(), DataTypes.BINARY(1)}) { + assertThatThrownBy( + () -> + JsonToFlussConverters.create(type) + .convert( + MAPPER.getNodeFactory().textNode(value), "$.b")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.b") + .hasMessageContaining("invalid Base64"); + } + } + + @ParameterizedTest + @ValueSource( + strings = { + "1969-12-31T23:59:59.999999999Z", + "1970-01-01T05:44:59.999999999+05:45", + "1969-12-31T20:29:59.999999999-03:30" + }) + void testTimestampLtzNormalizesOffsetsBeforeEpoch(String value) throws Exception { + TimestampLtz timestamp = + (TimestampLtz) convert(DataTypes.TIMESTAMP_LTZ(9), "\"" + value + "\""); + assertThat(timestamp.toInstant()).isEqualTo(Instant.EPOCH.minusNanos(1)); + } + @Test void testStringsBinaryAndTemporalRepresentations() throws Exception { assertThat(convert(DataTypes.CHAR(2), "\"😀a\"")).isEqualTo(BinaryString.fromString("😀a")); From 0881c7c0eddad8c032a20b29af4dbcff830cf6f5 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Sun, 25 Oct 2026 12:33:22 +0800 Subject: [PATCH 5/7] [kafka] Support recursive JSON conversion and normalize scalar values Add ARRAY, MAP, and ROW conversion with nested null handling and bounded error paths. Round decimals with HALF_UP, truncate temporal fractions, accept SQL timestamp separators and local-zone LTZ input, and normalize fixed-width binary values for Arrow. Reuse BinaryString string semantics. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 322/322 AI-Contributed/UT: 631/631 --- .../fluss/kafka/format/json/JsonPath.java | 97 ++++++ .../format/json/JsonToFlussConverters.java | 225 ++++++++------ .../format/json/JsonToFlussArrowTest.java | 158 ++++++++++ .../JsonToFlussComplexConvertersTest.java | 278 ++++++++++++++++++ .../json/JsonToFlussConvertersTest.java | 195 ++++++++++-- 5 files changed, 833 insertions(+), 120 deletions(-) create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonPath.java create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussArrowTest.java create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussComplexConvertersTest.java diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonPath.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonPath.java new file mode 100644 index 00000000000..c6a8eb5b958 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonPath.java @@ -0,0 +1,97 @@ +/* + * 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.fluss.kafka.format.json; + +/** Builds bounded, escaped JSON paths for Kafka record errors. */ +final class JsonPath { + + static final String ROOT = "$"; + + private static final int MAX_SEGMENT_LENGTH = 128; + private static final int MAX_PATH_LENGTH = 512; + private static final String TRUNCATED = "..."; + + private JsonPath() {} + + static String field(String path, String fieldName) { + return append(path, "[\"" + escapeAndTruncate(fieldName) + "\"]"); + } + + static String index(String path, int index) { + return append(path, "[" + index + "]"); + } + + private static String append(String path, String segment) { + if (path.length() + segment.length() <= MAX_PATH_LENGTH) { + return path + segment; + } + int prefixLength = MAX_PATH_LENGTH - TRUNCATED.length() - segment.length(); + if (prefixLength <= ROOT.length()) { + return (ROOT + TRUNCATED + segment).substring(0, MAX_PATH_LENGTH); + } + return path.substring(0, Math.min(path.length(), prefixLength)) + TRUNCATED + segment; + } + + private static String escapeAndTruncate(String value) { + StringBuilder escaped = new StringBuilder(MAX_SEGMENT_LENGTH); + int contentLimit = MAX_SEGMENT_LENGTH - TRUNCATED.length(); + int index = 0; + while (index < value.length()) { + String encodedCharacter; + char character = value.charAt(index); + switch (character) { + case '\\': + encodedCharacter = "\\\\"; + break; + case '"': + encodedCharacter = "\\\""; + break; + case '\b': + encodedCharacter = "\\b"; + break; + case '\f': + encodedCharacter = "\\f"; + break; + case '\n': + encodedCharacter = "\\n"; + break; + case '\r': + encodedCharacter = "\\r"; + break; + case '\t': + encodedCharacter = "\\t"; + break; + default: + if (character < 0x20) { + encodedCharacter = String.format("\\u%04x", (int) character); + } else { + encodedCharacter = String.valueOf(character); + } + } + if (escaped.length() + encodedCharacter.length() > contentLimit) { + break; + } + escaped.append(encodedCharacter); + index++; + } + if (index < value.length()) { + escaped.append(TRUNCATED); + } + return escaped.toString(); + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java index eb1433f4e78..a543495e7d8 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java @@ -23,43 +23,61 @@ import org.apache.fluss.metadata.KafkaTableSchema; import org.apache.fluss.row.BinaryString; import org.apache.fluss.row.Decimal; +import org.apache.fluss.row.GenericArray; +import org.apache.fluss.row.GenericMap; +import org.apache.fluss.row.GenericRow; import org.apache.fluss.row.TimestampLtz; import org.apache.fluss.row.TimestampNtz; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.fluss.types.ArrayType; import org.apache.fluss.types.BinaryType; -import org.apache.fluss.types.CharType; +import org.apache.fluss.types.DataField; import org.apache.fluss.types.DataType; import org.apache.fluss.types.DecimalType; import org.apache.fluss.types.LocalZonedTimestampType; +import org.apache.fluss.types.MapType; +import org.apache.fluss.types.RowType; import org.apache.fluss.types.TimeType; import org.apache.fluss.types.TimestampType; import javax.annotation.Nullable; import java.math.BigDecimal; -import java.math.RoundingMode; import java.time.DateTimeException; import java.time.Instant; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; import java.time.OffsetDateTime; +import java.time.ZoneId; import java.time.ZoneOffset; import java.time.format.DateTimeParseException; +import java.util.Arrays; import java.util.Base64; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import static org.apache.fluss.utils.DateTimeUtils.timestampToMicros; import static org.apache.fluss.utils.DateTimeUtils.timestampToNanos; -/** Strict JSON-to-Fluss scalar converter construction. */ +/** + * JSON-to-Fluss converter construction. + * + *

Recursive ARRAY, MAP, and ROW conversion follows Apache Flink's {@code + * JsonToRowDataConverters}, adapted to Fluss internal values, schema nullability, and bounded + * containers. + */ final class JsonToFlussConverters { + static final int MAX_CONTAINER_ELEMENTS = 10_000; + private JsonToFlussConverters() {} /** * Creates a converter for one top-level JSON business field, excluding the JSON root container. - * The shared schema contract is validated before checking decoder support; complex types still - * require their own converter implementations. + * The shared schema contract validates the entire type before recursive converter construction. */ static JsonToFlussConverter create(DataType dataType) { try { @@ -67,9 +85,13 @@ static JsonToFlussConverter create(DataType dataType) { } catch (InvalidTableException e) { throw new KafkaTopicSchemaException(e.getMessage()); } + return createValidated(dataType); + } + + private static JsonToFlussConverter createValidated(DataType dataType) { final JsonToFlussConverter notNullConverter = createNotNull(dataType); return (node, path) -> { - if (node == null || node.isNull()) { + if (node == null || node.isNull() || node.isMissingNode()) { if (!dataType.isNullable()) { throw invalid(path, dataType, "field is missing or null"); } @@ -149,9 +171,10 @@ private static JsonToFlussConverter createNotNull(DataType dataType) { case DECIMAL: return decimalConverter((DecimalType) dataType); case CHAR: - return charConverter((CharType) dataType); case STRING: - return (node, path) -> BinaryString.fromString(textualValue(node, path, dataType)); + return (node, path) -> + BinaryString.fromString( + node.isContainerNode() ? node.toString() : node.asText()); case BINARY: return binaryConverter((BinaryType) dataType); case BYTES: @@ -171,6 +194,12 @@ private static JsonToFlussConverter createNotNull(DataType dataType) { return timestampConverter((TimestampType) dataType); case TIMESTAMP_WITH_LOCAL_TIME_ZONE: return timestampLtzConverter((LocalZonedTimestampType) dataType); + case ARRAY: + return arrayConverter((ArrayType) dataType); + case MAP: + return mapConverter((MapType) dataType); + case ROW: + return rowConverter((RowType) dataType); default: throw new KafkaTopicSchemaException( "Kafka JSON format does not support Fluss type " @@ -194,19 +223,14 @@ private static JsonToFlussConverter decimalConverter(DecimalType dataType) { path, dataType, "decimal precision exceeds target"); - require( - value.stripTrailingZeros().scale() <= dataType.getScale(), - path, - dataType, - "decimal scale exceeds target"); - final BigDecimal scaled; - try { - scaled = value.setScale(dataType.getScale(), RoundingMode.UNNECESSARY); - } catch (ArithmeticException e) { - throw invalid(path, dataType, "decimal scale exceeds target", e); + // Values below a tenth of the target unit round to zero. Avoid constructing a huge + // power of ten when the input has an extreme negative exponent. + if (scaledPrecision < 0) { + return Decimal.zero(dataType.getPrecision(), dataType.getScale()); } + // Decimal uses HALF_UP and checks precision again after a possible rounding carry. Decimal decimal = - Decimal.fromBigDecimal(scaled, dataType.getPrecision(), dataType.getScale()); + Decimal.fromBigDecimal(value, dataType.getPrecision(), dataType.getScale()); if (decimal == null) { throw invalid(path, dataType, "decimal precision exceeds target"); } @@ -214,32 +238,18 @@ private static JsonToFlussConverter decimalConverter(DecimalType dataType) { }; } - private static JsonToFlussConverter charConverter(CharType dataType) { - return (node, path) -> { - String value = textualValue(node, path, dataType); - int codePoints = value.codePointCount(0, value.length()); - require( - codePoints <= dataType.getLength(), - path, - dataType, - "character length exceeds target"); - return BinaryString.fromString(value); - }; - } - private static JsonToFlussConverter binaryConverter(BinaryType dataType) { return (node, path) -> { byte[] value = decodeBase64(node, path, dataType); - require( - value.length == dataType.getLength(), - path, - dataType, - "decoded binary length must equal " + dataType.getLength()); - return value; + // Arrow uses FixedSizeBinaryVector: zero-pad short values and truncate long ones. + return value.length == dataType.getLength() + ? value + : Arrays.copyOf(value, dataType.getLength()); }; } private static JsonToFlussConverter timeConverter(TimeType dataType) { + final int nanosPerUnit = nanosPerUnit(Math.min(dataType.getPrecision(), 3)); return (node, path) -> { String value = textualValue(node, path, dataType); final LocalTime time; @@ -248,56 +258,119 @@ private static JsonToFlussConverter timeConverter(TimeType dataType) { } catch (DateTimeParseException e) { throw invalid(path, dataType, "invalid ISO-8601 local time", e); } - validatePrecision(time.getNano(), dataType.getPrecision(), path, dataType); - require( - time.getNano() % 1_000_000 == 0, - path, - dataType, - "Fluss TIME storage currently requires millisecond precision"); - return Math.toIntExact(time.toNanoOfDay() / 1_000_000); + return time.toSecondOfDay() * 1_000 + + (time.getNano() / nanosPerUnit * nanosPerUnit) / 1_000_000; }; } private static JsonToFlussConverter timestampConverter(TimestampType dataType) { + final int nanosPerUnit = nanosPerUnit(dataType.getPrecision()); return (node, path) -> { - String value = textualValue(node, path, dataType); + String value = textualValue(node, path, dataType).replace(' ', 'T'); final LocalDateTime timestamp; try { timestamp = LocalDateTime.parse(value); } catch (DateTimeParseException e) { throw invalid(path, dataType, "invalid ISO-8601 local timestamp", e); } - validatePrecision(timestamp.getNano(), dataType.getPrecision(), path, dataType); + LocalDateTime normalized = + timestamp.withNano(timestamp.getNano() / nanosPerUnit * nanosPerUnit); long millis = timestampMillis( - timestamp.toInstant(ZoneOffset.UTC), + normalized.toInstant(ZoneOffset.UTC), dataType.getPrecision(), path, dataType); - return TimestampNtz.fromMillis(millis, timestamp.getNano() % 1_000_000); + return TimestampNtz.fromMillis(millis, normalized.getNano() % 1_000_000); }; } private static JsonToFlussConverter timestampLtzConverter(LocalZonedTimestampType dataType) { + final int nanosPerUnit = nanosPerUnit(dataType.getPrecision()); + final ZoneId localZone = ZoneId.systemDefault(); return (node, path) -> { - String value = textualValue(node, path, dataType); - final OffsetDateTime timestamp; + String value = textualValue(node, path, dataType).replace(' ', 'T'); + Instant timestamp; try { - timestamp = OffsetDateTime.parse(value); + timestamp = OffsetDateTime.parse(value).toInstant(); } catch (DateTimeParseException e) { - throw invalid( - path, - dataType, - "invalid ISO-8601 timestamp; an offset or Z is required", - e); + try { + timestamp = LocalDateTime.parse(value).atZone(localZone).toInstant(); + } catch (DateTimeParseException localError) { + throw invalid(path, dataType, "invalid timestamp", localError); + } } - validatePrecision(timestamp.getNano(), dataType.getPrecision(), path, dataType); - long millis = - timestampMillis(timestamp.toInstant(), dataType.getPrecision(), path, dataType); - return TimestampLtz.fromEpochMillis(millis, timestamp.getNano() % 1_000_000); + Instant normalized = + Instant.ofEpochSecond( + timestamp.getEpochSecond(), + timestamp.getNano() / nanosPerUnit * nanosPerUnit); + long millis = timestampMillis(normalized, dataType.getPrecision(), path, dataType); + return TimestampLtz.fromEpochMillis(millis, normalized.getNano() % 1_000_000); + }; + } + + private static JsonToFlussConverter arrayConverter(ArrayType dataType) { + JsonToFlussConverter elementConverter = createValidated(dataType.getElementType()); + return (node, path) -> { + require(node.isArray(), path, dataType, "expected a JSON array"); + requireContainerSize(node.size(), path, dataType); + Object[] elements = new Object[node.size()]; + for (int i = 0; i < elements.length; i++) { + elements[i] = elementConverter.convert(node.get(i), JsonPath.index(path, i)); + } + return new GenericArray(elements); + }; + } + + private static JsonToFlussConverter mapConverter(MapType dataType) { + JsonToFlussConverter valueConverter = createValidated(dataType.getValueType()); + return (node, path) -> { + require(node.isObject(), path, dataType, "expected a JSON object"); + requireContainerSize(node.size(), path, dataType); + Map values = new LinkedHashMap<>(); + Iterator> fields = node.fields(); + while (fields.hasNext()) { + Map.Entry field = fields.next(); + values.put( + BinaryString.fromString(field.getKey()), + valueConverter.convert( + field.getValue(), JsonPath.field(path, field.getKey()))); + } + return new GenericMap(values); + }; + } + + private static JsonToFlussConverter rowConverter(RowType dataType) { + List fields = dataType.getFields(); + String[] fieldNames = new String[fields.size()]; + JsonToFlussConverter[] fieldConverters = new JsonToFlussConverter[fields.size()]; + for (int i = 0; i < fields.size(); i++) { + DataField field = fields.get(i); + fieldNames[i] = field.getName(); + fieldConverters[i] = createValidated(field.getType()); + } + return (node, path) -> { + require(node.isObject(), path, dataType, "expected a JSON object"); + requireContainerSize(node.size(), path, dataType); + GenericRow row = new GenericRow(fieldNames.length); + for (int i = 0; i < fieldNames.length; i++) { + row.setField( + i, + fieldConverters[i].convert( + node.get(fieldNames[i]), JsonPath.field(path, fieldNames[i]))); + } + return row; }; } + private static void requireContainerSize(int size, String path, DataType dataType) { + require( + size <= MAX_CONTAINER_ELEMENTS, + path, + dataType, + "container size exceeds " + MAX_CONTAINER_ELEMENTS); + } + private static long timestampMillis( Instant timestamp, int precision, String path, DataType dataType) { try { @@ -324,27 +397,7 @@ private static long integralValue(JsonNode node, String path, DataType dataType) private static String textualValue(JsonNode node, String path, DataType dataType) { require(node.isTextual(), path, dataType, "expected a JSON string"); - String value = node.textValue(); - // Jackson can retain unpaired JSON escapes in Java strings. Reject them before - // BinaryString's UTF-8 encoding silently replaces them with a question mark. - for (int i = 0; i < value.length(); i++) { - char character = value.charAt(i); - if (Character.isHighSurrogate(character)) { - require( - i + 1 < value.length() && Character.isLowSurrogate(value.charAt(i + 1)), - path, - dataType, - "unpaired UTF-16 surrogate"); - i++; - } else { - require( - !Character.isLowSurrogate(character), - path, - dataType, - "unpaired UTF-16 surrogate"); - } - } - return value; + return node.textValue(); } private static byte[] decodeBase64(JsonNode node, String path, DataType dataType) { @@ -356,16 +409,12 @@ private static byte[] decodeBase64(JsonNode node, String path, DataType dataType } } - private static void validatePrecision( - int nanos, int precision, String path, DataType dataType) { - if (precision >= 9) { - return; - } + private static int nanosPerUnit(int precision) { int unit = 1; for (int i = precision; i < 9; i++) { unit *= 10; } - require(nanos % unit == 0, path, dataType, "fractional seconds exceed target precision"); + return unit; } private static void require(boolean condition, String path, DataType dataType, String reason) { diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussArrowTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussArrowTest.java new file mode 100644 index 00000000000..85abf920539 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussArrowTest.java @@ -0,0 +1,158 @@ +/* + * 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.fluss.kafka.format.json; + +import org.apache.fluss.row.GenericRow; +import org.apache.fluss.row.InternalArray; +import org.apache.fluss.row.InternalMap; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.TimestampNtz; +import org.apache.fluss.row.arrow.ArrowReader; +import org.apache.fluss.row.arrow.writers.ArrowFieldWriter; +import org.apache.fluss.row.columnar.ColumnVector; +import org.apache.fluss.row.columnar.ColumnarRow; +import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.RootAllocator; +import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.types.RowType; +import org.apache.fluss.utils.ArrowUtils; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.math.BigDecimal; +import java.time.LocalDateTime; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Verifies converted JSON values through the Fluss Arrow writers and readers. */ +class JsonToFlussArrowTest { + private static final ObjectMapper MAPPER = + new ObjectMapper().enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS); + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testBinaryPadsAndTruncatesWithoutCorruptingAdjacentValues(boolean handleSafe) + throws Exception { + RowType schema = DataTypes.ROW(DataTypes.BINARY(2)); + JsonToFlussConverter converter = JsonToFlussConverters.create(DataTypes.BINARY(2)); + String[] json = {"\"AQID\"", "\"AQ==\"", "\"\"", "null", "\"AwQ=\""}; + byte[][] expected = { + new byte[] {1, 2}, new byte[] {1, 0}, new byte[] {0, 0}, null, new byte[] {3, 4} + }; + try (RootAllocator allocator = new RootAllocator(1024 * 1024); + VectorSchemaRoot root = + VectorSchemaRoot.create(ArrowUtils.toArrowSchema(schema), allocator)) { + root.allocateNew(); + ArrowFieldWriter writer = + ArrowUtils.createArrowFieldWriter(root.getVector(0), schema.getTypeAt(0)); + for (int i = 0; i < json.length; i++) { + Object value = converter.convert(MAPPER.readTree(json[i]), "$[\"binary\"]"); + writer.write(i, GenericRow.of(value), 0, handleSafe); + } + root.setRowCount(json.length); + ArrowReader reader = reader(root, schema); + for (int i = 0; i < expected.length; i++) { + ColumnarRow row = reader.read(i); + if (expected[i] == null) { + assertThat(row.isNullAt(0)).isTrue(); + } else { + assertThat(row.getBinary(0, 2)).containsExactly(expected[i]); + } + } + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testNestedContainersNullsAndNormalizedScalarsRoundTrip(boolean handleSafe) + throws Exception { + RowType valueType = + DataTypes.ROW( + DataTypes.FIELD("payload", DataTypes.BINARY(2)), + DataTypes.FIELD("amounts", DataTypes.ARRAY(DataTypes.DECIMAL(5, 2))), + DataTypes.FIELD("time", DataTypes.TIME(3)), + DataTypes.FIELD("ts", DataTypes.TIMESTAMP(6))); + RowType schema = + DataTypes.ROW( + DataTypes.FIELD("name", DataTypes.CHAR(1)), + DataTypes.FIELD( + "groups", + DataTypes.ARRAY(DataTypes.MAP(DataTypes.STRING(), valueType)))); + String[] json = { + "{\"name\":\"full value\",\"groups\":[{\"a\":{\"payload\":\"AQ==\"," + + "\"amounts\":[12.345,null],\"time\":\"12:34:56.123456\"," + + "\"ts\":\"1969-12-31 23:59:59.999999999\"},\"empty\":null},null,{}]}", + "{\"name\":\"second\",\"groups\":null}", + "{\"name\":\"third\",\"groups\":[]}" + }; + JsonToFlussConverter converter = JsonToFlussConverters.create(schema); + try (RootAllocator allocator = new RootAllocator(8 * 1024 * 1024); + VectorSchemaRoot root = + VectorSchemaRoot.create(ArrowUtils.toArrowSchema(schema), allocator)) { + root.allocateNew(); + ArrowFieldWriter[] writers = new ArrowFieldWriter[schema.getFieldCount()]; + for (int field = 0; field < writers.length; field++) { + writers[field] = + ArrowUtils.createArrowFieldWriter( + root.getVector(field), schema.getTypeAt(field)); + } + for (int i = 0; i < json.length; i++) { + GenericRow row = (GenericRow) converter.convert(MAPPER.readTree(json[i]), "$"); + for (int field = 0; field < writers.length; field++) { + writers[field].write(i, row, field, handleSafe); + } + } + root.setRowCount(json.length); + ArrowReader reader = reader(root, schema); + ColumnarRow first = reader.read(0); + assertThat(first.getChar(0, 1).toString()).isEqualTo("full value"); + InternalArray groups = first.getArray(1); + assertThat(groups.size()).isEqualTo(3); + assertThat(groups.isNullAt(1)).isTrue(); + assertThat(groups.getMap(2).size()).isZero(); + InternalMap entries = groups.getMap(0); + assertThat(entries.size()).isEqualTo(2); + assertThat(entries.keyArray().getString(0).toString()).isEqualTo("a"); + assertThat(entries.keyArray().getString(1).toString()).isEqualTo("empty"); + assertThat(entries.valueArray().isNullAt(1)).isTrue(); + InternalRow value = entries.valueArray().getRow(0, 4); + assertThat(value.getBinary(0, 2)).containsExactly(1, 0); + assertThat(value.getArray(1).getDecimal(0, 5, 2).toBigDecimal()) + .isEqualTo(new BigDecimal("12.35")); + assertThat(value.getArray(1).isNullAt(1)).isTrue(); + assertThat(value.getInt(2)).isEqualTo(45_296_123); + assertThat(value.getTimestampNtz(3, 6)) + .isEqualTo( + TimestampNtz.fromLocalDateTime( + LocalDateTime.parse("1969-12-31T23:59:59.999999"))); + assertThat(reader.read(1).isNullAt(1)).isTrue(); + assertThat(reader.read(2).getArray(1).size()).isZero(); + } + } + + private static ArrowReader reader(VectorSchemaRoot root, RowType schema) { + ColumnVector[] columns = new ColumnVector[schema.getFieldCount()]; + for (int i = 0; i < columns.length; i++) { + columns[i] = ArrowUtils.createArrowColumnVector(root.getVector(i), schema.getTypeAt(i)); + } + return new ArrowReader(columns, root.getRowCount()); + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussComplexConvertersTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussComplexConvertersTest.java new file mode 100644 index 00000000000..2450b047411 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussComplexConvertersTest.java @@ -0,0 +1,278 @@ +/* + * 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.fluss.kafka.format.json; + +import org.apache.fluss.kafka.transcode.KafkaRecordEncodingException; +import org.apache.fluss.metadata.KafkaTableSchema; +import org.apache.fluss.row.BinaryString; +import org.apache.fluss.row.GenericArray; +import org.apache.fluss.row.GenericMap; +import org.apache.fluss.row.GenericRow; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.math.BigDecimal; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests recursive containers, nullability, and bounded conversion errors. */ +class JsonToFlussComplexConvertersTest { + private static final ObjectMapper MAPPER = + new ObjectMapper().enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS); + + @Test + void testNestedRowsArraysAndMaps() throws Exception { + RowType type = + DataTypes.ROW( + DataTypes.FIELD( + "customer", + DataTypes.ROW( + DataTypes.FIELD("name", DataTypes.STRING()), + DataTypes.FIELD( + "address", + DataTypes.ROW( + DataTypes.FIELD( + "city", DataTypes.STRING()))))), + DataTypes.FIELD( + "items", + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD("sku", DataTypes.STRING()), + DataTypes.FIELD("quantity", DataTypes.INT())))), + DataTypes.FIELD( + "attributes", + DataTypes.MAP( + DataTypes.STRING(), DataTypes.ARRAY(DataTypes.INT())))); + GenericRow row = + (GenericRow) + convert( + type, + "{\"customer\":{\"name\":\"Alice\",\"address\":{\"city\":\"Hangzhou\"}}," + + "\"items\":[{\"sku\":\"A-100\",\"quantity\":2},null]," + + "\"attributes\":{\"scores\":[7,null,9],\"empty\":[],\"missing\":null}}"); + assertThat(row.getRow(0, 2).getString(0).toString()).isEqualTo("Alice"); + assertThat(row.getRow(0, 2).getRow(1, 1).getString(0).toString()).isEqualTo("Hangzhou"); + assertThat(row.getArray(1).getRow(0, 2).getString(0).toString()).isEqualTo("A-100"); + assertThat(row.getArray(1).getRow(0, 2).getInt(1)).isEqualTo(2); + assertThat(row.getArray(1).isNullAt(1)).isTrue(); + GenericMap attributes = (GenericMap) row.getMap(2); + assertThat( + ((GenericArray) attributes.get(BinaryString.fromString("scores"))) + .toObjectArray()) + .containsExactly(7, null, 9); + assertThat(((GenericArray) attributes.get(BinaryString.fromString("empty"))).size()) + .isZero(); + assertThat(attributes.get(BinaryString.fromString("missing"))).isNull(); + } + + @Test + void testArrayOfMapsOfRowsUsesScalarNormalization() throws Exception { + DataType type = + DataTypes.ARRAY( + DataTypes.MAP( + DataTypes.STRING(), + DataTypes.ROW( + DataTypes.FIELD("amount", DataTypes.DECIMAL(5, 2)), + DataTypes.FIELD( + "labels", + DataTypes.MAP( + DataTypes.STRING(), DataTypes.STRING()))))); + GenericArray array = + (GenericArray) + convert( + type, + "[{\"entry\":{\"amount\":12.345,\"labels\":{\"boolean\":true,\"array\":[1,2]}}}]"); + GenericRow row = + (GenericRow) ((GenericMap) array.getMap(0)).get(BinaryString.fromString("entry")); + assertThat(row.getDecimal(0, 5, 2).toBigDecimal()).isEqualTo(new BigDecimal("12.35")); + GenericMap labels = (GenericMap) row.getMap(1); + assertThat(labels.get(BinaryString.fromString("boolean"))) + .isEqualTo(BinaryString.fromString("true")); + assertThat(labels.get(BinaryString.fromString("array"))) + .isEqualTo(BinaryString.fromString("[1,2]")); + } + + @Test + void testEmptyContainersNullsAndMissingFields() throws Exception { + RowType type = + DataTypes.ROW( + DataTypes.FIELD("array", DataTypes.ARRAY(DataTypes.INT())), + DataTypes.FIELD("map", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())), + DataTypes.FIELD( + "row", DataTypes.ROW(DataTypes.FIELD("value", DataTypes.INT())))); + GenericRow empty = (GenericRow) convert(type, "{\"array\":[],\"map\":{},\"row\":{}}"); + assertThat(empty.getArray(0).size()).isZero(); + assertThat(empty.getMap(1).size()).isZero(); + assertThat(empty.getRow(2, 1).isNullAt(0)).isTrue(); + for (String json : new String[] {"{}", "{\"array\":null,\"map\":null,\"row\":null}"}) { + GenericRow missing = (GenericRow) convert(type, json); + for (int i = 0; i < 3; i++) { + assertThat(missing.isNullAt(i)).isTrue(); + } + } + GenericArray arrays = + (GenericArray) + convert( + DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT())), + "[[],null,[null]]"); + assertThat(arrays.getArray(0).size()).isZero(); + assertThat(arrays.isNullAt(1)).isTrue(); + assertThat(arrays.getArray(2).isNullAt(0)).isTrue(); + assertThat(convert(type, "null")).isNull(); + } + + @Test + void testRowsIgnoreUnknownFieldsAndMapKeysRemainDynamic() throws Exception { + RowType type = + DataTypes.ROW( + DataTypes.FIELD( + "nested", DataTypes.ROW(DataTypes.FIELD("id", DataTypes.INT())))); + GenericRow row = + (GenericRow) + convert(type, "{\"extra\":true,\"nested\":{\"extra\":[1,2],\"id\":3}}"); + assertThat(row.getRow(0, 1).getInt(0)).isEqualTo(3); + GenericMap map = + (GenericMap) + convert( + DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()), + "{\"\":1,\"😀\":2,\"dynamic-key\":null}"); + assertThat(map.size()).isEqualTo(3); + assertThat(map.get(BinaryString.fromString(""))).isEqualTo(1); + assertThat(map.get(BinaryString.fromString("😀"))).isEqualTo(2); + assertThat(map.get(BinaryString.fromString("dynamic-key"))).isNull(); + } + + @ParameterizedTest + @MethodSource("invalidContainers") + void testRejectsInvalidShapesAndValuesAtNestedPaths(DataType type, String json, String path) + throws Exception { + assertThatThrownBy(() -> convert(type, json)) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining(path); + } + + private static Stream invalidContainers() { + return Stream.of( + Arguments.of(DataTypes.ARRAY(DataTypes.INT()), "{}", "$"), + Arguments.of(DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()), "[]", "$"), + Arguments.of(DataTypes.ROW(DataTypes.INT()), "[]", "$"), + Arguments.of(DataTypes.ARRAY(DataTypes.INT()), "[1,\"bad\"]", "$[1]"), + Arguments.of(DataTypes.ARRAY(DataTypes.INT().copy(false)), "[null]", "$[0]"), + Arguments.of( + DataTypes.MAP(DataTypes.STRING(), DataTypes.INT().copy(false)), + "{\"key\":null}", + "$[\"key\"]"), + Arguments.of( + DataTypes.ROW(DataTypes.FIELD("value", DataTypes.INT().copy(false))), + "{}", + "$[\"value\"]"), + Arguments.of( + DataTypes.ARRAY( + DataTypes.MAP( + DataTypes.STRING(), + DataTypes.ROW( + DataTypes.FIELD("quantity", DataTypes.INT())))), + "[{\"bad\\\"key\":{\"quantity\":\"bad\"}}]", + "$[0][\"bad\\\"key\"][\"quantity\"]")); + } + + @Test + void testContainersAtTheLimitAndBeyond() { + ArrayNode array = MAPPER.createArrayNode(); + ObjectNode object = MAPPER.createObjectNode(); + for (int i = 0; i < JsonToFlussConverters.MAX_CONTAINER_ELEMENTS; i++) { + array.add(i); + object.put("k" + i, i); + } + JsonToFlussConverter arrayConverter = + JsonToFlussConverters.create(DataTypes.ARRAY(DataTypes.INT())); + JsonToFlussConverter mapConverter = + JsonToFlussConverters.create(DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())); + JsonToFlussConverter rowConverter = + JsonToFlussConverters.create(DataTypes.ROW(DataTypes.FIELD("k0", DataTypes.INT()))); + assertThat(((GenericArray) arrayConverter.convert(array, "$")).size()) + .isEqualTo(JsonToFlussConverters.MAX_CONTAINER_ELEMENTS); + assertThat(((GenericMap) mapConverter.convert(object, "$")).size()) + .isEqualTo(JsonToFlussConverters.MAX_CONTAINER_ELEMENTS); + assertThat(((GenericRow) rowConverter.convert(object, "$")).getInt(0)).isZero(); + array.add(0); + object.put("excess", 0); + assertThatThrownBy(() -> arrayConverter.convert(array, "$")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("container size exceeds"); + for (JsonToFlussConverter converter : + new JsonToFlussConverter[] {mapConverter, rowConverter}) { + assertThatThrownBy(() -> converter.convert(object, "$")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("container size exceeds"); + } + } + + @Test + void testConvertsAtSchemaNestingLimit() { + DataType type = DataTypes.INT(); + JsonNode value = MAPPER.getNodeFactory().numberNode(42); + for (int i = 0; i < KafkaTableSchema.MAX_JSON_SCHEMA_NESTING_DEPTH; i++) { + type = DataTypes.ARRAY(type); + value = MAPPER.createArrayNode().add(value); + } + Object result = JsonToFlussConverters.create(type).convert(value, "$"); + for (int i = 0; i < KafkaTableSchema.MAX_JSON_SCHEMA_NESTING_DEPTH; i++) { + result = ((GenericArray) result).toObjectArray()[0]; + } + assertThat(result).isEqualTo(42); + } + + @Test + void testErrorPathsEscapeNamesAndStayBounded() { + StringBuilder name = new StringBuilder("quoted\"\\\n\t\r\b\f"); + for (int i = 0; i < 1000; i++) { + name.append('x'); + } + ObjectNode object = MAPPER.createObjectNode().put(name.toString(), "invalid"); + JsonToFlussConverter converter = + JsonToFlussConverters.create(DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())); + assertThatThrownBy(() -> converter.convert(object, "$")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$[\"quoted\\\"\\\\\\n\\t\\r\\b\\f") + .hasMessageContaining("...") + .satisfies(error -> assertThat(error.getMessage().length()).isLessThan(300)); + String path = "$"; + for (int i = 0; i < 64; i++) { + path = JsonPath.field(path, name.toString()); + } + assertThat(path).hasSize(512).startsWith("$").endsWith("...\"]"); + } + + private static Object convert(DataType type, String json) throws Exception { + return JsonToFlussConverters.create(type).convert(MAPPER.readTree(json), "$"); + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java index c7273ca7401..84227993ab2 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java @@ -35,6 +35,7 @@ import org.apache.fluss.types.DataTypes; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.ResourceLock; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -46,12 +47,13 @@ import java.time.LocalDate; import java.time.LocalDateTime; import java.time.ZoneOffset; +import java.util.TimeZone; import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -/** Verifies strict scalar conversion without transport or Arrow dependencies. */ +/** Verifies scalar conversion, normalization, and storage bounds. */ class JsonToFlussConvertersTest { private static final ObjectMapper MAPPER = new ObjectMapper().enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS); @@ -186,6 +188,10 @@ void testStringsBinaryAndTemporalRepresentations() throws Exception { void testMissingAndNullRespectNullability() throws Exception { assertThat(JsonToFlussConverters.create(DataTypes.INT()).convert(null, "$.field")).isNull(); assertThat(convert(DataTypes.INT(), "null")).isNull(); + assertThat( + JsonToFlussConverters.create(DataTypes.INT()) + .convert(MAPPER.missingNode(), "$.field")) + .isNull(); assertThatThrownBy( () -> JsonToFlussConverters.create(DataTypes.INT().copy(false)) @@ -201,16 +207,17 @@ void testMissingAndNullRespectNullability() throws Exception { @Test void testDecimalBoundsBeforeRescaling() throws Exception { for (int scale : new int[] {100_000_000, Integer.MAX_VALUE, Integer.MIN_VALUE}) { - assertThatThrownBy( - () -> - JsonToFlussConverters.create(DataTypes.DECIMAL(5, 2)) - .convert( - DecimalNode.valueOf( - new BigDecimal(BigInteger.ONE, scale)), - "$.amount")) - .isInstanceOf(KafkaRecordEncodingException.class) - .hasMessageContaining("$.amount") - .hasMessageContaining(scale > 0 ? "scale" : "precision"); + JsonNode value = DecimalNode.valueOf(new BigDecimal(BigInteger.ONE, scale)); + JsonToFlussConverter converter = JsonToFlussConverters.create(DataTypes.DECIMAL(5, 2)); + if (scale > 0) { + assertThat(((Decimal) converter.convert(value, "$.amount")).toBigDecimal()) + .isEqualTo(new BigDecimal("0.00")); + } else { + assertThatThrownBy(() -> converter.convert(value, "$.amount")) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.amount") + .hasMessageContaining("precision"); + } Decimal zero = (Decimal) JsonToFlussConverters.create(DataTypes.DECIMAL(5, 2)) @@ -228,12 +235,143 @@ void testDecimalBoundsBeforeRescaling() throws Exception { @ParameterizedTest @ValueSource(strings = {"\\uD800", "\\uDC00", "a\\uD800b", "\\uD800\\uD800", "\\uDC00\\uD800"}) - void testRejectsUnpairedSurrogates(String escaped) { + void testUnpairedSurrogatesFollowBinaryStringEncoding(String escaped) throws Exception { for (DataType type : new DataType[] {DataTypes.STRING(), DataTypes.CHAR(8)}) { - assertThatThrownBy(() -> convert(type, "\"" + escaped + "\"")) - .isInstanceOf(KafkaRecordEncodingException.class) - .hasMessageContaining("$.field") - .hasMessageContaining("surrogate"); + String json = "\"" + escaped + "\""; + assertThat(convert(type, json)) + .isEqualTo(BinaryString.fromString(MAPPER.readTree(json).textValue())); + } + } + + @ParameterizedTest + @MethodSource("roundedDecimals") + void testDecimalRoundsHalfUp(String json, String expected) throws Exception { + assertThat(((Decimal) convert(DataTypes.DECIMAL(5, 2), json)).toBigDecimal()) + .isEqualTo(new BigDecimal(expected)); + } + + private static Stream roundedDecimals() { + return Stream.of( + Arguments.of("12.345", "12.35"), + Arguments.of("-12.345", "-12.35"), + Arguments.of("999.994", "999.99"), + Arguments.of("0.005", "0.01"), + Arguments.of("-0.005", "-0.01"), + Arguments.of("0.0049", "0.00"), + Arguments.of("-1e-1000", "0.00")); + } + + @ParameterizedTest + @ValueSource(strings = {"false", "123", "1.25", "[1,null,\"x\"]", "{\"a\":true}"}) + void testStringsAcceptOtherJsonTypes(String json) throws Exception { + for (DataType type : new DataType[] {DataTypes.STRING(), DataTypes.CHAR(1)}) { + assertThat(convert(type, json)).isEqualTo(BinaryString.fromString(json)); + } + } + + @Test + void testCharKeepsFullStringAndBinaryNormalizesWidth() throws Exception { + assertThat(convert(DataTypes.CHAR(1), "\"ab😀\"")) + .isEqualTo(BinaryString.fromString("ab😀")); + assertThat((byte[]) convert(DataTypes.BINARY(2), "\"\"")).containsExactly(0, 0); + assertThat((byte[]) convert(DataTypes.BINARY(2), "\"AQ==\"")).containsExactly(1, 0); + assertThat((byte[]) convert(DataTypes.BINARY(2), "\"AQID\"")).containsExactly(1, 2); + assertThat((byte[]) convert(DataTypes.BYTES(), "\"AQID\"")).containsExactly(1, 2, 3); + } + + @ParameterizedTest + @ValueSource(strings = {"T", " "}) + void testTimestampSeparatorsWithAndWithoutFractions(String separator) throws Exception { + for (String fraction : new String[] {"", ".123456"}) { + LocalDateTime expected = LocalDateTime.parse("2026-10-25T12:34:56" + fraction); + String local = "2026-10-25" + separator + "12:34:56" + fraction; + assertThat(convert(DataTypes.TIMESTAMP(6), "\"" + local + "\"")) + .isEqualTo(TimestampNtz.fromLocalDateTime(expected)); + for (String offset : new String[] {"Z", "+08:00"}) { + assertThat(convert(DataTypes.TIMESTAMP_LTZ(6), "\"" + local + offset + "\"")) + .isEqualTo( + TimestampLtz.fromInstant( + expected.toInstant(ZoneOffset.of(offset)))); + } + } + } + + @Test + @ResourceLock("java.util.TimeZone.default") + void testTimestampLtzFallsBackToLocalZoneAndPreservesExplicitOffsets() throws Exception { + TimeZone previous = TimeZone.getDefault(); + try { + TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai")); + for (String separator : new String[] {"T", " "}) { + for (String fraction : new String[] {"", ".123456"}) { + String local = "2026-10-25" + separator + "12:34:56" + fraction; + TimestampLtz expected = + TimestampLtz.fromInstant( + Instant.parse("2026-10-25T04:34:56" + fraction + "Z")); + assertThat(convert(DataTypes.TIMESTAMP_LTZ(6), "\"" + local + "\"")) + .isEqualTo(expected); + assertThat(convert(DataTypes.TIMESTAMP_LTZ(6), "\"" + local + "+08:00\"")) + .isEqualTo(expected); + assertThat(convert(DataTypes.TIMESTAMP_LTZ(6), "\"" + local + "Z\"")) + .isEqualTo( + TimestampLtz.fromInstant( + Instant.parse("2026-10-25T12:34:56" + fraction + "Z"))); + } + } + } finally { + TimeZone.setDefault(previous); + } + } + + @ParameterizedTest + @MethodSource("temporalPrecisions") + void testTruncatesTemporalFractions(int precision, int nanos, int timeMillis) throws Exception { + assertThat(convert(DataTypes.TIME(precision), "\"12:34:56.123456789\"")) + .isEqualTo(45_296_000 + timeMillis); + for (boolean withZone : new boolean[] {false, true}) { + for (String date : new String[] {"1969-12-31", "2026-10-25"}) { + Instant expected = Instant.parse(date + "T12:34:56Z").plusNanos(nanos); + Object actual = + convert( + timestampType(precision, withZone), + "\"" + date + "T12:34:56.123456789" + (withZone ? "Z\"" : "\"")); + assertThat(actual) + .isEqualTo( + withZone + ? TimestampLtz.fromInstant(expected) + : TimestampNtz.fromLocalDateTime( + LocalDateTime.ofInstant(expected, ZoneOffset.UTC))); + } + } + } + + private static Stream temporalPrecisions() { + return Stream.of( + Arguments.of(0, 0, 0), + Arguments.of(1, 100_000_000, 100), + Arguments.of(2, 120_000_000, 120), + Arguments.of(3, 123_000_000, 123), + Arguments.of(4, 123_400_000, 123), + Arguments.of(5, 123_450_000, 123), + Arguments.of(6, 123_456_000, 123), + Arguments.of(7, 123_456_700, 123), + Arguments.of(8, 123_456_780, 123), + Arguments.of(9, 123_456_789, 123)); + } + + @ParameterizedTest + @ValueSource(ints = {3, 6}) + void testTruncatesBeforeCheckingTimestampStorageRange(int precision) throws Exception { + long unitsPerSecond = precision == 3 ? 1_000 : 1_000_000; + long nanosPerUnit = 1_000_000_000 / unitsPerSecond; + Instant maximum = + Instant.ofEpochSecond( + Math.floorDiv(Long.MAX_VALUE, unitsPerSecond), + Math.floorMod(Long.MAX_VALUE, unitsPerSecond) * nanosPerUnit); + for (boolean withZone : new boolean[] {false, true}) { + DataType type = timestampType(precision, withZone); + assertThat(convert(type, timestampJson(maximum.plusNanos(nanosPerUnit - 1), withZone))) + .isEqualTo(convert(type, timestampJson(maximum, withZone))); } } @@ -336,7 +474,7 @@ private static String timestampJson(Instant instant, boolean withZone) { @ParameterizedTest @MethodSource("invalidScalars") - void testRejectsCoercionOverflowAndPrecisionLoss(DataType type, String json) { + void testRejectsInvalidValuesAndOverflow(DataType type, String json) { assertThatThrownBy(() -> convert(type, json)) .isInstanceOf(KafkaRecordEncodingException.class) .hasMessageContaining("$.field"); @@ -353,21 +491,18 @@ private static Stream invalidScalars() { Arguments.of(DataTypes.BIGINT(), "9223372036854775808"), Arguments.of(DataTypes.FLOAT(), "1e39"), Arguments.of(DataTypes.DOUBLE(), "1e309"), - Arguments.of(DataTypes.DECIMAL(5, 2), "1.234"), + Arguments.of(DataTypes.DECIMAL(5, 2), "999.995"), + Arguments.of(DataTypes.DECIMAL(5, 2), "-999.995"), Arguments.of(DataTypes.DECIMAL(5, 2), "1000"), - Arguments.of(DataTypes.CHAR(1), "\"ab\""), - Arguments.of(DataTypes.STRING(), "false"), - Arguments.of(DataTypes.BINARY(2), "\"AQ==\""), Arguments.of(DataTypes.BYTES(), "\"!invalid!\""), Arguments.of(DataTypes.DATE(), "\"2026-02-30\""), - Arguments.of(DataTypes.TIME(0), "\"00:00:00.001\""), - Arguments.of(DataTypes.TIME(3), "\"00:00:00.0001\""), - Arguments.of(DataTypes.TIMESTAMP(3), "\"2026-09-11T00:00:00.0001\""), - Arguments.of(DataTypes.TIMESTAMP_LTZ(3), "\"2026-09-11T00:00:00\"")); + Arguments.of(DataTypes.TIME(3), "\"24:00:00\""), + Arguments.of(DataTypes.TIMESTAMP(3), "\"2026-02-30 12:34:56\""), + Arguments.of(DataTypes.TIMESTAMP_LTZ(3), "\"2026-10-25 12:34:56+25:00\"")); } @Test - void testComplexTypesRemainUnsupportedAtThisStage() { + void testComplexTypesFollowTheSharedSchemaContract() { for (DataType type : new DataType[] { DataTypes.ARRAY(DataTypes.INT()), @@ -375,9 +510,7 @@ void testComplexTypesRemainUnsupportedAtThisStage() { DataTypes.ROW(DataTypes.INT()) }) { KafkaTableSchema.validateJsonType(type); - assertThatThrownBy(() -> JsonToFlussConverters.create(type)) - .isInstanceOf(KafkaTopicSchemaException.class) - .hasMessageContaining("does not support"); + assertThat(JsonToFlussConverters.create(type).convert(null, "$.field")).isNull(); } } @@ -419,9 +552,7 @@ void testSchemaDepthUsesTheSharedBusinessFieldLimit() { // The top-level business field starts at zero; each child consumes one level. DataType atLimit = type; KafkaTableSchema.validateJsonType(atLimit); - assertThatThrownBy(() -> JsonToFlussConverters.create(atLimit)) - .isInstanceOf(KafkaTopicSchemaException.class) - .hasMessageContaining("does not support"); + assertThat(JsonToFlussConverters.create(atLimit)).isNotNull(); DataType tooDeep = DataTypes.ROW(atLimit); String message = From efe1e3895bdd664c97c1cf74967901e3956c4e27 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Sun, 25 Oct 2026 13:52:29 +0800 Subject: [PATCH 6/7] [kafka] Align JSON scalar coercions with Flink Accept Flink-compatible boolean and numeric text, decimal node conversion, and Jackson Base64 decoding. Retain Fluss precision, nullability, and storage-bound checks and cover scalar coercions inside nested containers. Validation: 459 related tests passed, including 149 JSON conversion tests; 100 cases compared against Flink 1.20; Java 8, Spotless, Checkstyle and RAT passed. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 90/90 AI-Contributed/UT: 183/183 --- .../format/json/JsonToFlussConverters.java | 90 ++++----- .../json/JsonToFlussConvertersTest.java | 7 +- .../JsonToFlussFlinkCompatibilityTest.java | 176 ++++++++++++++++++ 3 files changed, 218 insertions(+), 55 deletions(-) create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussFlinkCompatibilityTest.java diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java index a543495e7d8..4d238d8a6d5 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java @@ -42,6 +42,7 @@ import javax.annotation.Nullable; +import java.io.IOException; import java.math.BigDecimal; import java.time.DateTimeException; import java.time.Instant; @@ -53,7 +54,6 @@ import java.time.ZoneOffset; import java.time.format.DateTimeParseException; import java.util.Arrays; -import java.util.Base64; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; @@ -65,9 +65,11 @@ /** * JSON-to-Fluss converter construction. * - *

Recursive ARRAY, MAP, and ROW conversion follows Apache Flink's {@code + *

Scalar coercion and recursive conversion follow Apache Flink 1.20's {@code * JsonToRowDataConverters}, adapted to Fluss internal values, schema nullability, and bounded - * containers. + * containers. Fluss additionally normalizes temporal precision and fixed binary widths, supports + * timestamp offsets and local timestamps, and reports numeric overflow instead of storing + * non-finite numbers or a null decimal. */ final class JsonToFlussConverters { @@ -110,46 +112,34 @@ private static JsonToFlussConverter createValidated(DataType dataType) { private static JsonToFlussConverter createNotNull(DataType dataType) { switch (dataType.getTypeRoot()) { case BOOLEAN: - return (node, path) -> { - require(node.isBoolean(), path, dataType, "expected a JSON boolean"); - return node.booleanValue(); - }; + // Like Flink, parseBoolean treats every non-true string as false. + return (node, path) -> + node.isBoolean() + ? node.booleanValue() + : Boolean.parseBoolean(node.asText().trim()); case TINYINT: - return (node, path) -> { - long value = integralValue(node, path, dataType); - require( - value >= Byte.MIN_VALUE && value <= Byte.MAX_VALUE, - path, - dataType, - "integer overflow"); - return (byte) value; - }; + return (node, path) -> Byte.parseByte(node.asText().trim()); case SMALLINT: - return (node, path) -> { - long value = integralValue(node, path, dataType); - require( - value >= Short.MIN_VALUE && value <= Short.MAX_VALUE, - path, - dataType, - "integer overflow"); - return (short) value; - }; + return (node, path) -> Short.parseShort(node.asText().trim()); case INTEGER: - return (node, path) -> { - long value = integralValue(node, path, dataType); - require( - value >= Integer.MIN_VALUE && value <= Integer.MAX_VALUE, - path, - dataType, - "integer overflow"); - return (int) value; - }; + // Flink truncates fractional numeric nodes, but parses text as an exact integer. + return (node, path) -> + node.canConvertToInt() + ? node.asInt() + : Integer.parseInt(node.asText().trim()); case BIGINT: - return (node, path) -> integralValue(node, path, dataType); + return (node, path) -> + node.canConvertToLong() + ? node.asLong() + : Long.parseLong(node.asText().trim()); case FLOAT: return (node, path) -> { - require(node.isNumber(), path, dataType, "expected a JSON number"); - float value = node.floatValue(); + // Parse decimal text directly to float to avoid an intermediate double + // rounding. + float value = + node.isDouble() + ? (float) node.asDouble() + : Float.parseFloat(node.asText().trim()); require( Float.isFinite(value), path, @@ -159,8 +149,10 @@ private static JsonToFlussConverter createNotNull(DataType dataType) { }; case DOUBLE: return (node, path) -> { - require(node.isNumber(), path, dataType, "expected a JSON number"); - double value = node.doubleValue(); + double value = + node.isDouble() + ? node.asDouble() + : Double.parseDouble(node.asText().trim()); require( Double.isFinite(value), path, @@ -210,8 +202,8 @@ private static JsonToFlussConverter createNotNull(DataType dataType) { private static JsonToFlussConverter decimalConverter(DecimalType dataType) { return (node, path) -> { - require(node.isNumber(), path, dataType, "expected a JSON number"); - BigDecimal value = node.decimalValue(); + BigDecimal value = + node.isBigDecimal() ? node.decimalValue() : new BigDecimal(node.asText()); if (value.signum() == 0) { return Decimal.zero(dataType.getPrecision(), dataType.getScale()); } @@ -389,22 +381,20 @@ private static long timestampMillis( } } - private static long integralValue(JsonNode node, String path, DataType dataType) { - require(node.isIntegralNumber(), path, dataType, "expected a JSON integer"); - require(node.canConvertToLong(), path, dataType, "integer overflow"); - return node.longValue(); - } - private static String textualValue(JsonNode node, String path, DataType dataType) { require(node.isTextual(), path, dataType, "expected a JSON string"); return node.textValue(); } private static byte[] decodeBase64(JsonNode node, String path, DataType dataType) { - String value = textualValue(node, path, dataType); + require( + node.isTextual() || node.isBinary(), + path, + dataType, + "expected a Base64 string or binary node"); try { - return Base64.getDecoder().decode(value); - } catch (IllegalArgumentException e) { + return node.binaryValue(); + } catch (IOException e) { throw invalid(path, dataType, "invalid Base64 string", e); } } diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java index 84227993ab2..8547a98d18e 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java @@ -140,8 +140,8 @@ private static Stream nonFiniteNumbers() { } @ParameterizedTest - @ValueSource(strings = {"A", "AQ===", "AQ==x", "A Q==", "AQ==\n", "_w=="}) - void testBase64RejectsInvalidPaddingAlphabetAndWhitespace(String value) { + @ValueSource(strings = {"A", "AQ", "AQI", "AQ===", "AQ==x", "A Q==", "_w=="}) + void testBase64RejectsInvalidPaddingAlphabetAndEmbeddedWhitespace(String value) { for (DataType type : new DataType[] {DataTypes.BYTES(), DataTypes.BINARY(1)}) { assertThatThrownBy( () -> @@ -482,9 +482,6 @@ void testRejectsInvalidValuesAndOverflow(DataType type, String json) { private static Stream invalidScalars() { return Stream.of( - Arguments.of(DataTypes.BOOLEAN(), "1"), - Arguments.of(DataTypes.INT(), "\"1\""), - Arguments.of(DataTypes.INT(), "1.0"), Arguments.of(DataTypes.INT(), "2147483648"), Arguments.of(DataTypes.TINYINT(), "128"), Arguments.of(DataTypes.SMALLINT(), "-32769"), diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussFlinkCompatibilityTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussFlinkCompatibilityTest.java new file mode 100644 index 00000000000..79fb235d948 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussFlinkCompatibilityTest.java @@ -0,0 +1,176 @@ +/* + * 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.fluss.kafka.format.json; + +import org.apache.fluss.kafka.transcode.KafkaRecordEncodingException; +import org.apache.fluss.row.Decimal; +import org.apache.fluss.row.GenericRow; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.BinaryNode; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.FloatNode; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.math.BigDecimal; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Covers scalar coercion rules from Flink 1.20 JsonToRowDataConverters. */ +class JsonToFlussFlinkCompatibilityTest { + private static final ObjectMapper MAPPER = + new ObjectMapper().enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS); + + @ParameterizedTest + @MethodSource("coercibleScalars") + void testFlinkScalarCoercions(DataType type, String json, Object expected) throws Exception { + assertThat(convert(type, json)).isEqualTo(expected); + } + + private static Stream coercibleScalars() { + return Stream.of( + Arguments.of(DataTypes.BOOLEAN(), "\" TrUe \"", true), + Arguments.of(DataTypes.BOOLEAN(), "\"false\"", false), + // Boolean.parseBoolean returns false for every string except case-insensitive true. + Arguments.of(DataTypes.BOOLEAN(), "\"not-a-boolean\"", false), + Arguments.of(DataTypes.BOOLEAN(), "1", false), + Arguments.of(DataTypes.BOOLEAN(), "{}", false), + Arguments.of(DataTypes.TINYINT(), "\" -128 \"", Byte.MIN_VALUE), + Arguments.of(DataTypes.TINYINT(), "\"+127\"", Byte.MAX_VALUE), + Arguments.of(DataTypes.SMALLINT(), "\" -32768 \"", Short.MIN_VALUE), + Arguments.of(DataTypes.SMALLINT(), "\"32767\"", Short.MAX_VALUE), + Arguments.of(DataTypes.INT(), "\" -2147483648 \"", Integer.MIN_VALUE), + Arguments.of(DataTypes.INT(), "\"+2147483647\"", Integer.MAX_VALUE), + Arguments.of(DataTypes.INT(), "112.013", 112), + Arguments.of(DataTypes.INT(), "-112.013", -112), + Arguments.of(DataTypes.INT(), "1.0", 1), + Arguments.of(DataTypes.BIGINT(), "\" -9223372036854775808 \"", Long.MIN_VALUE), + Arguments.of(DataTypes.BIGINT(), "\"9223372036854775807\"", Long.MAX_VALUE), + Arguments.of(DataTypes.BIGINT(), "112.013", 112L), + Arguments.of(DataTypes.BIGINT(), "-112.013", -112L), + Arguments.of(DataTypes.FLOAT(), "\" 1.25e2 \"", 125F), + Arguments.of(DataTypes.FLOAT(), "\"-1.0000000596046448\"", -1.0000001F), + Arguments.of(DataTypes.FLOAT(), "\"-0.0\"", -0.0F), + Arguments.of(DataTypes.DOUBLE(), "\" -2.5e2 \"", -250D), + Arguments.of(DataTypes.DOUBLE(), "\"-0.0\"", -0.0D)); + } + + @Test + void testDecimalTextAndFloatingPointNodes() throws Exception { + assertThat(((Decimal) convert(DataTypes.DECIMAL(5, 2), "\"12.345\"")).toBigDecimal()) + .isEqualByComparingTo("12.35"); + assertThat(((Decimal) convert(DataTypes.DECIMAL(5, 2), "\"1.2345e1\"")).toBigDecimal()) + .isEqualByComparingTo("12.35"); + assertThat(((Decimal) convert(DataTypes.DECIMAL(5, 2), "\"1e-2147483647\"")).toBigDecimal()) + .isEqualByComparingTo(BigDecimal.ZERO); + JsonNode floatNode = FloatNode.valueOf(0.1F); + Decimal decimal = + (Decimal) + JsonToFlussConverters.create(DataTypes.DECIMAL(20, 10)) + .convert(floatNode, "$.field"); + assertThat(decimal.toBigDecimal()).isEqualByComparingTo("0.1"); + assertThat(JsonToFlussConverters.create(DataTypes.DOUBLE()).convert(floatNode, "$.field")) + .isEqualTo(0.1D); + } + + @Test + void testJacksonBinaryNodesAndBase64Whitespace() throws Exception { + byte[] bytes = {1, 2, 3, 4}; + JsonNode binary = BinaryNode.valueOf(bytes); + assertThat((byte[]) JsonToFlussConverters.create(DataTypes.BYTES()).convert(binary, "$.b")) + .containsExactly(bytes); + assertThat( + (byte[]) + JsonToFlussConverters.create(DataTypes.BINARY(2)) + .convert(binary, "$.b")) + .containsExactly((byte) 1, (byte) 2); + assertThat((byte[]) convert(DataTypes.BYTES(), "\" AQID\\nBA== \"")).containsExactly(bytes); + assertThat((byte[]) convert(DataTypes.BINARY(6), "\" AQID\\nBA== \"")) + .containsExactly((byte) 1, (byte) 2, (byte) 3, (byte) 4, (byte) 0, (byte) 0); + } + + @Test + void testCoercionsInsideRowsArraysAndMaps() throws Exception { + RowType type = + DataTypes.ROW( + DataTypes.FIELD("active", DataTypes.BOOLEAN()), + DataTypes.FIELD("counts", DataTypes.ARRAY(DataTypes.INT())), + DataTypes.FIELD( + "prices", + DataTypes.MAP(DataTypes.STRING(), DataTypes.DECIMAL(5, 2)))); + GenericRow row = + (GenericRow) + convert( + type, + "{\"active\":\"TRUE\",\"counts\":[\"1\",-2.9,null],\"prices\":{\"a\":\"12.345\"}}"); + assertThat(row.getBoolean(0)).isTrue(); + assertThat(row.getArray(1).getInt(0)).isEqualTo(1); + assertThat(row.getArray(1).getInt(1)).isEqualTo(-2); + assertThat(row.getArray(1).isNullAt(2)).isTrue(); + assertThat(row.getMap(2).valueArray().getDecimal(0, 5, 2).toBigDecimal()) + .isEqualByComparingTo("12.35"); + } + + @ParameterizedTest + @MethodSource("invalidScalarCoercions") + void testRejectsInvalidTextAndOutOfRangeValues(DataType type, String json) { + assertThatThrownBy(() -> convert(type, json)) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.field"); + } + + private static Stream invalidScalarCoercions() { + return Stream.of( + Arguments.of(DataTypes.TINYINT(), "\"128\""), + Arguments.of(DataTypes.TINYINT(), "1.5"), + Arguments.of(DataTypes.SMALLINT(), "\"-32769\""), + Arguments.of(DataTypes.SMALLINT(), "-1.5"), + Arguments.of(DataTypes.INT(), "\"2147483648\""), + Arguments.of(DataTypes.INT(), "\"1.0\""), + Arguments.of(DataTypes.INT(), "2147483647.1"), + Arguments.of(DataTypes.INT(), "true"), + Arguments.of(DataTypes.BIGINT(), "\"9223372036854775808\""), + Arguments.of(DataTypes.BIGINT(), "\"1.0\""), + Arguments.of(DataTypes.BIGINT(), "9223372036854775807.1"), + Arguments.of(DataTypes.FLOAT(), "\"not-a-number\""), + Arguments.of(DataTypes.DOUBLE(), "[]"), + // Retain Fluss overflow errors rather than storing non-finite values or null. + Arguments.of(DataTypes.FLOAT(), "\"NaN\""), + Arguments.of(DataTypes.FLOAT(), "\"1e39\""), + Arguments.of(DataTypes.DOUBLE(), "\"Infinity\""), + Arguments.of(DataTypes.DOUBLE(), "\"1e309\""), + Arguments.of(DataTypes.DECIMAL(5, 2), "\"999.995\""), + Arguments.of(DataTypes.DECIMAL(5, 2), "\"1e2147483647\""), + Arguments.of(DataTypes.DECIMAL(5, 2), "\" 12.34 \""), + Arguments.of(DataTypes.DECIMAL(5, 2), "true"), + Arguments.of(DataTypes.BYTES(), "1")); + } + + private static Object convert(DataType type, String json) throws Exception { + return JsonToFlussConverters.create(type).convert(MAPPER.readTree(json), "$.field"); + } +} From a8c4621b777bc55373574aa5b005385df679a67d Mon Sep 17 00:00:00 2001 From: Jark Wu Date: Sun, 25 Oct 2026 15:41:22 +0800 Subject: [PATCH 7/7] [json] Refactor JsonToFlussConverters for improved readability and maintainability Co-Authored-By: Codex AI-Contributed/Feature: 151/151 AI-Contributed/UT: 33/33 --- .../format/json/JsonToFlussConverters.java | 151 ++++++++++-------- .../JsonToFlussComplexConvertersTest.java | 33 ---- 2 files changed, 80 insertions(+), 104 deletions(-) diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java index 4d238d8a6d5..020167e69c7 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java @@ -66,15 +66,13 @@ * JSON-to-Fluss converter construction. * *

Scalar coercion and recursive conversion follow Apache Flink 1.20's {@code - * JsonToRowDataConverters}, adapted to Fluss internal values, schema nullability, and bounded - * containers. Fluss additionally normalizes temporal precision and fixed binary widths, supports - * timestamp offsets and local timestamps, and reports numeric overflow instead of storing - * non-finite numbers or a null decimal. + * JsonToRowDataConverters}, adapted to Fluss internal values and schema nullability. Fluss + * additionally normalizes temporal precision and fixed binary widths, supports timestamp offsets + * and local timestamps, and reports numeric overflow instead of storing non-finite numbers or a + * null decimal. */ final class JsonToFlussConverters { - static final int MAX_CONTAINER_ELEMENTS = 10_000; - private JsonToFlussConverters() {} /** @@ -112,74 +110,30 @@ private static JsonToFlussConverter createValidated(DataType dataType) { private static JsonToFlussConverter createNotNull(DataType dataType) { switch (dataType.getTypeRoot()) { case BOOLEAN: - // Like Flink, parseBoolean treats every non-true string as false. - return (node, path) -> - node.isBoolean() - ? node.booleanValue() - : Boolean.parseBoolean(node.asText().trim()); + return booleanConverter(); case TINYINT: - return (node, path) -> Byte.parseByte(node.asText().trim()); + return tinyIntConverter(); case SMALLINT: - return (node, path) -> Short.parseShort(node.asText().trim()); + return smallIntConverter(); case INTEGER: - // Flink truncates fractional numeric nodes, but parses text as an exact integer. - return (node, path) -> - node.canConvertToInt() - ? node.asInt() - : Integer.parseInt(node.asText().trim()); + return intConverter(); case BIGINT: - return (node, path) -> - node.canConvertToLong() - ? node.asLong() - : Long.parseLong(node.asText().trim()); + return bigIntConverter(); case FLOAT: - return (node, path) -> { - // Parse decimal text directly to float to avoid an intermediate double - // rounding. - float value = - node.isDouble() - ? (float) node.asDouble() - : Float.parseFloat(node.asText().trim()); - require( - Float.isFinite(value), - path, - dataType, - "non-finite or overflowing number"); - return value; - }; + return floatConverter(dataType); case DOUBLE: - return (node, path) -> { - double value = - node.isDouble() - ? node.asDouble() - : Double.parseDouble(node.asText().trim()); - require( - Double.isFinite(value), - path, - dataType, - "non-finite or overflowing number"); - return value; - }; + return doubleConverter(dataType); case DECIMAL: return decimalConverter((DecimalType) dataType); case CHAR: case STRING: - return (node, path) -> - BinaryString.fromString( - node.isContainerNode() ? node.toString() : node.asText()); + return stringConverter(); case BINARY: return binaryConverter((BinaryType) dataType); case BYTES: - return (node, path) -> decodeBase64(node, path, dataType); + return bytesConverter(dataType); case DATE: - return (node, path) -> { - String value = textualValue(node, path, dataType); - try { - return Math.toIntExact(LocalDate.parse(value).toEpochDay()); - } catch (DateTimeException | ArithmeticException e) { - throw invalid(path, dataType, "invalid ISO-8601 date", e); - } - }; + return dateConverter(dataType); case TIME_WITHOUT_TIME_ZONE: return timeConverter((TimeType) dataType); case TIMESTAMP_WITHOUT_TIME_ZONE: @@ -200,6 +154,52 @@ private static JsonToFlussConverter createNotNull(DataType dataType) { } } + private static JsonToFlussConverter booleanConverter() { + // parseBoolean treats every non-true string as false. + return (node, path) -> + node.isBoolean() ? node.booleanValue() : Boolean.parseBoolean(node.asText().trim()); + } + + private static JsonToFlussConverter tinyIntConverter() { + return (node, path) -> Byte.parseByte(node.asText().trim()); + } + + private static JsonToFlussConverter smallIntConverter() { + return (node, path) -> Short.parseShort(node.asText().trim()); + } + + private static JsonToFlussConverter intConverter() { + // Flink truncates fractional numeric nodes, but parses text as an exact integer. + return (node, path) -> + node.canConvertToInt() ? node.asInt() : Integer.parseInt(node.asText().trim()); + } + + private static JsonToFlussConverter bigIntConverter() { + return (node, path) -> + node.canConvertToLong() ? node.asLong() : Long.parseLong(node.asText().trim()); + } + + private static JsonToFlussConverter floatConverter(DataType dataType) { + return (node, path) -> { + // Parse decimal text directly to float to avoid an intermediate double rounding. + float value = + node.isDouble() + ? (float) node.asDouble() + : Float.parseFloat(node.asText().trim()); + require(Float.isFinite(value), path, dataType, "non-finite or overflowing number"); + return value; + }; + } + + private static JsonToFlussConverter doubleConverter(DataType dataType) { + return (node, path) -> { + double value = + node.isDouble() ? node.asDouble() : Double.parseDouble(node.asText().trim()); + require(Double.isFinite(value), path, dataType, "non-finite or overflowing number"); + return value; + }; + } + private static JsonToFlussConverter decimalConverter(DecimalType dataType) { return (node, path) -> { BigDecimal value = @@ -230,6 +230,11 @@ private static JsonToFlussConverter decimalConverter(DecimalType dataType) { }; } + private static JsonToFlussConverter stringConverter() { + return (node, path) -> + BinaryString.fromString(node.isContainerNode() ? node.toString() : node.asText()); + } + private static JsonToFlussConverter binaryConverter(BinaryType dataType) { return (node, path) -> { byte[] value = decodeBase64(node, path, dataType); @@ -240,6 +245,21 @@ private static JsonToFlussConverter binaryConverter(BinaryType dataType) { }; } + private static JsonToFlussConverter bytesConverter(DataType dataType) { + return (node, path) -> decodeBase64(node, path, dataType); + } + + private static JsonToFlussConverter dateConverter(DataType dataType) { + return (node, path) -> { + String value = textualValue(node, path, dataType); + try { + return Math.toIntExact(LocalDate.parse(value).toEpochDay()); + } catch (DateTimeException | ArithmeticException e) { + throw invalid(path, dataType, "invalid ISO-8601 date", e); + } + }; + } + private static JsonToFlussConverter timeConverter(TimeType dataType) { final int nanosPerUnit = nanosPerUnit(Math.min(dataType.getPrecision(), 3)); return (node, path) -> { @@ -305,7 +325,6 @@ private static JsonToFlussConverter arrayConverter(ArrayType dataType) { JsonToFlussConverter elementConverter = createValidated(dataType.getElementType()); return (node, path) -> { require(node.isArray(), path, dataType, "expected a JSON array"); - requireContainerSize(node.size(), path, dataType); Object[] elements = new Object[node.size()]; for (int i = 0; i < elements.length; i++) { elements[i] = elementConverter.convert(node.get(i), JsonPath.index(path, i)); @@ -318,7 +337,6 @@ private static JsonToFlussConverter mapConverter(MapType dataType) { JsonToFlussConverter valueConverter = createValidated(dataType.getValueType()); return (node, path) -> { require(node.isObject(), path, dataType, "expected a JSON object"); - requireContainerSize(node.size(), path, dataType); Map values = new LinkedHashMap<>(); Iterator> fields = node.fields(); while (fields.hasNext()) { @@ -343,7 +361,6 @@ private static JsonToFlussConverter rowConverter(RowType dataType) { } return (node, path) -> { require(node.isObject(), path, dataType, "expected a JSON object"); - requireContainerSize(node.size(), path, dataType); GenericRow row = new GenericRow(fieldNames.length); for (int i = 0; i < fieldNames.length; i++) { row.setField( @@ -355,14 +372,6 @@ private static JsonToFlussConverter rowConverter(RowType dataType) { }; } - private static void requireContainerSize(int size, String path, DataType dataType) { - require( - size <= MAX_CONTAINER_ELEMENTS, - path, - dataType, - "container size exceeds " + MAX_CONTAINER_ELEMENTS); - } - private static long timestampMillis( Instant timestamp, int precision, String path, DataType dataType) { try { diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussComplexConvertersTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussComplexConvertersTest.java index 2450b047411..890bbee396f 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussComplexConvertersTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussComplexConvertersTest.java @@ -26,7 +26,6 @@ import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; -import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode; import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.fluss.types.DataType; import org.apache.fluss.types.DataTypes; @@ -204,38 +203,6 @@ private static Stream invalidContainers() { "$[0][\"bad\\\"key\"][\"quantity\"]")); } - @Test - void testContainersAtTheLimitAndBeyond() { - ArrayNode array = MAPPER.createArrayNode(); - ObjectNode object = MAPPER.createObjectNode(); - for (int i = 0; i < JsonToFlussConverters.MAX_CONTAINER_ELEMENTS; i++) { - array.add(i); - object.put("k" + i, i); - } - JsonToFlussConverter arrayConverter = - JsonToFlussConverters.create(DataTypes.ARRAY(DataTypes.INT())); - JsonToFlussConverter mapConverter = - JsonToFlussConverters.create(DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())); - JsonToFlussConverter rowConverter = - JsonToFlussConverters.create(DataTypes.ROW(DataTypes.FIELD("k0", DataTypes.INT()))); - assertThat(((GenericArray) arrayConverter.convert(array, "$")).size()) - .isEqualTo(JsonToFlussConverters.MAX_CONTAINER_ELEMENTS); - assertThat(((GenericMap) mapConverter.convert(object, "$")).size()) - .isEqualTo(JsonToFlussConverters.MAX_CONTAINER_ELEMENTS); - assertThat(((GenericRow) rowConverter.convert(object, "$")).getInt(0)).isZero(); - array.add(0); - object.put("excess", 0); - assertThatThrownBy(() -> arrayConverter.convert(array, "$")) - .isInstanceOf(KafkaRecordEncodingException.class) - .hasMessageContaining("container size exceeds"); - for (JsonToFlussConverter converter : - new JsonToFlussConverter[] {mapConverter, rowConverter}) { - assertThatThrownBy(() -> converter.convert(object, "$")) - .isInstanceOf(KafkaRecordEncodingException.class) - .hasMessageContaining("container size exceeds"); - } - } - @Test void testConvertsAtSchemaNestingLimit() { DataType type = DataTypes.INT();