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/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/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/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..020167e69c7 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/json/JsonToFlussConverters.java @@ -0,0 +1,449 @@ +/* + * 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.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.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.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.io.IOException; +import java.math.BigDecimal; +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.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; + +/** + * JSON-to-Fluss converter construction. + * + *

Scalar coercion and recursive conversion follow Apache Flink 1.20's {@code + * 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 { + + private JsonToFlussConverters() {} + + /** + * Creates a converter for one top-level JSON business field, excluding the JSON root container. + * The shared schema contract validates the entire type before recursive converter construction. + */ + static JsonToFlussConverter create(DataType dataType) { + try { + KafkaTableSchema.validateJsonType(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() || node.isMissingNode()) { + 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 booleanConverter(); + case TINYINT: + return tinyIntConverter(); + case SMALLINT: + return smallIntConverter(); + case INTEGER: + return intConverter(); + case BIGINT: + return bigIntConverter(); + case FLOAT: + return floatConverter(dataType); + case DOUBLE: + return doubleConverter(dataType); + case DECIMAL: + return decimalConverter((DecimalType) dataType); + case CHAR: + case STRING: + return stringConverter(); + case BINARY: + return binaryConverter((BinaryType) dataType); + case BYTES: + return bytesConverter(dataType); + case DATE: + return dateConverter(dataType); + 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); + 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 " + + dataType.asSummaryString() + + "."); + } + } + + 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 = + node.isBigDecimal() ? node.decimalValue() : new BigDecimal(node.asText()); + 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"); + // 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(value, dataType.getPrecision(), dataType.getScale()); + if (decimal == null) { + throw invalid(path, dataType, "decimal precision exceeds target"); + } + return decimal; + }; + } + + 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); + // 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 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) -> { + 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); + } + 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).replace(' ', 'T'); + final LocalDateTime timestamp; + try { + timestamp = LocalDateTime.parse(value); + } catch (DateTimeParseException e) { + throw invalid(path, dataType, "invalid ISO-8601 local timestamp", e); + } + LocalDateTime normalized = + timestamp.withNano(timestamp.getNano() / nanosPerUnit * nanosPerUnit); + long millis = + timestampMillis( + normalized.toInstant(ZoneOffset.UTC), + dataType.getPrecision(), + path, + dataType); + 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).replace(' ', 'T'); + Instant timestamp; + try { + timestamp = OffsetDateTime.parse(value).toInstant(); + } catch (DateTimeParseException e) { + try { + timestamp = LocalDateTime.parse(value).atZone(localZone).toInstant(); + } catch (DateTimeParseException localError) { + throw invalid(path, dataType, "invalid timestamp", localError); + } + } + 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"); + 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"); + 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"); + 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 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 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) { + require( + node.isTextual() || node.isBinary(), + path, + dataType, + "expected a Base64 string or binary node"); + try { + return node.binaryValue(); + } catch (IOException e) { + throw invalid(path, dataType, "invalid Base64 string", e); + } + } + + private static int nanosPerUnit(int precision) { + int unit = 1; + for (int i = precision; i < 9; i++) { + unit *= 10; + } + return unit; + } + + 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/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..890bbee396f --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussComplexConvertersTest.java @@ -0,0 +1,245 @@ +/* + * 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.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 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 new file mode 100644 index 00000000000..8547a98d18e --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/format/json/JsonToFlussConvertersTest.java @@ -0,0 +1,570 @@ +/* + * 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.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; +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; + +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; +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.TimeZone; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Verifies scalar conversion, normalization, and storage bounds. */ +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")); + } + + @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"); + } + + @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", "AQI", "AQ===", "AQ==x", "A Q==", "_w=="}) + void testBase64RejectsInvalidPaddingAlphabetAndEmbeddedWhitespace(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")); + 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(); + assertThat( + JsonToFlussConverters.create(DataTypes.INT()) + .convert(MAPPER.missingNode(), "$.field")) + .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"); + } + + @Test + void testDecimalBoundsBeforeRescaling() throws Exception { + for (int scale : new int[] {100_000_000, Integer.MAX_VALUE, Integer.MIN_VALUE}) { + 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)) + .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 testUnpairedSurrogatesFollowBinaryStringEncoding(String escaped) throws Exception { + for (DataType type : new DataType[] {DataTypes.STRING(), DataTypes.CHAR(8)}) { + 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))); + } + } + + @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 testRejectsInvalidValuesAndOverflow(DataType type, String json) { + assertThatThrownBy(() -> convert(type, json)) + .isInstanceOf(KafkaRecordEncodingException.class) + .hasMessageContaining("$.field"); + } + + private static Stream invalidScalars() { + return Stream.of( + 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), "999.995"), + Arguments.of(DataTypes.DECIMAL(5, 2), "-999.995"), + Arguments.of(DataTypes.DECIMAL(5, 2), "1000"), + Arguments.of(DataTypes.BYTES(), "\"!invalid!\""), + Arguments.of(DataTypes.DATE(), "\"2026-02-30\""), + 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 testComplexTypesFollowTheSharedSchemaContract() { + for (DataType type : + new DataType[] { + DataTypes.ARRAY(DataTypes.INT()), + DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()), + DataTypes.ROW(DataTypes.INT()) + }) { + KafkaTableSchema.validateJsonType(type); + assertThat(JsonToFlussConverters.create(type).convert(null, "$.field")).isNull(); + } + } + + @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); + assertThat(JsonToFlussConverters.create(atLimit)).isNotNull(); + + 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 { + return JsonToFlussConverters.create(type).convert(MAPPER.readTree(json), "$.field"); + } +} 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"); + } +}