This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9b8faccb884 [FLINK-41056][table] Keep non-compact decimals with
unscaled value fitting in a long as long to avoid BigDecimal on read, add,
subtract, cast and deserialize
9b8faccb884 is described below
commit 9b8faccb8842a6be4f84e3f71da7c82b0c499e00
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Sat Oct 24 21:36:25 2026 +0200
[FLINK-41056][table] Keep non-compact decimals with unscaled value fitting
in a long as long to avoid BigDecimal on read, add, subtract, cast and
deserialize
---
.../org/apache/flink/table/data/DecimalData.java | 43 +++++--
.../flink/table/data/DecimalDataInternalUtils.java | 42 +++++++
.../table/data/binary/BinarySegmentUtils.java | 22 ++++
.../apache/flink/table/data/DecimalDataUtils.java | 80 ++++++++----
.../flink/table/data/BinaryArrayDataTest.java | 32 +++++
.../apache/flink/table/data/BinaryRowDataTest.java | 36 ++++++
.../apache/flink/table/data/DecimalDataTest.java | 138 ++++++++++++++++++++-
.../apache/flink/table/data/NestedRowDataTest.java | 36 ++++++
.../flink/table/data/util/DataFormatTestUtil.java | 36 ++++++
.../runtime/typeutils/DecimalDataSerializer.java | 10 ++
.../runtime/typeutils/DecimalSerializerTest.java | 112 ++++++++++++++---
11 files changed, 531 insertions(+), 56 deletions(-)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/DecimalData.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/DecimalData.java
index 5f4e4941163..7ba1c1b6ab1 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/DecimalData.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/DecimalData.java
@@ -61,27 +61,33 @@ public final class DecimalData implements
Comparable<DecimalData> {
// The semantics of the fields are as follows:
// - `precision` and `scale` represent the precision and scale of SQL
decimal type
- // - If `decimalVal` is set, it represents the whole decimal value
- // - Otherwise, the decimal value is longVal/(10^scale).
+ // - if `longBacked`, the decimal value is longVal/(10^scale);
+ // `decimalVal` may be set and cached
+ // - otherwise (only if precision > MAX_COMPACT_PRECISION),
+ // `decimalVal` represents the value and `longVal` is undefined
//
// Note that the (precision, scale) must be correct.
- // if precision > MAX_COMPACT_PRECISION,
- // `decimalVal` represents the value. `longVal` is undefined
- // otherwise, (longVal, scale) represents the value
- // `decimalVal` may be set and cached
final int precision;
final int scale;
final long longVal;
+ final boolean longBacked;
BigDecimal decimalVal;
// this constructor does not perform any sanity check.
+ // non-compact: long-backed iff `decimalVal` is null
DecimalData(int precision, int scale, long longVal, BigDecimal decimalVal)
{
+ this(precision, scale, longVal, decimalVal, isCompact(precision) ||
decimalVal == null);
+ }
+
+ private DecimalData(
+ int precision, int scale, long longVal, BigDecimal decimalVal,
boolean longBacked) {
this.precision = precision;
this.scale = scale;
this.longVal = longVal;
this.decimalVal = decimalVal;
+ this.longBacked = longBacked;
}
//
------------------------------------------------------------------------------------------
@@ -117,7 +123,7 @@ public final class DecimalData implements
Comparable<DecimalData> {
* @throws ArithmeticException if this {@link DecimalData} does not
exactly fit in a long.
*/
public long toUnscaledLong() {
- if (isCompact()) {
+ if (longBacked) {
return longVal;
} else {
return toBigDecimal().unscaledValue().longValueExact();
@@ -130,27 +136,42 @@ public final class DecimalData implements
Comparable<DecimalData> {
* @return the unscaled byte array of this {@link DecimalData}.
*/
public byte[] toUnscaledBytes() {
+ if (longBacked) {
+ // same bytes as BigInteger#toByteArray: minimal big-endian two's
complement
+ final int length = (64 - Long.numberOfLeadingZeros(longVal ^
(longVal >> 63))) / 8 + 1;
+ final byte[] bytes = new byte[length];
+ for (int i = 0; i < length; i++) {
+ bytes[length - 1 - i] = (byte) (longVal >> (8 * i));
+ }
+ return bytes;
+ }
return toBigDecimal().unscaledValue().toByteArray();
}
- /** Returns whether the decimal value is small enough to be stored in a
long. */
+ /** Returns whether the precision is at most 18. */
public boolean isCompact() {
return precision <= MAX_COMPACT_PRECISION;
}
/** Returns a copy of this {@link DecimalData} object. */
public DecimalData copy() {
- return new DecimalData(precision, scale, longVal, decimalVal);
+ return new DecimalData(precision, scale, longVal, decimalVal,
longBacked);
}
@Override
public int hashCode() {
+ if (longBacked && longVal != Long.MIN_VALUE) {
+ // BigDecimal#hashCode without allocating; must match, hashes may
feed partitioning
+ final long abs = Math.abs(longVal);
+ final int h = (int) ((int) (abs >>> 32) * 31 + (abs &
0xFFFFFFFFL));
+ return 31 * (longVal < 0 ? -h : h) + scale;
+ }
return toBigDecimal().hashCode();
}
@Override
public int compareTo(@Nonnull DecimalData that) {
- if (this.isCompact() && that.isCompact() && this.scale == that.scale) {
+ if (this.longBacked && that.longBacked && this.scale == that.scale) {
return Long.compare(this.longVal, that.longVal);
}
return this.toBigDecimal().compareTo(that.toBigDecimal());
@@ -197,7 +218,7 @@ public final class DecimalData implements
Comparable<DecimalData> {
/**
* Rescales a compact unscaled value to the given {@code precision}/{@code
scale} with HALF_UP
* in long arithmetic (no {@link BigDecimal}). Returns {@code null} on
precision overflow, like
- * {@link #fromBigDecimal}. Callers must ensure the source and target are
compact.
+ * {@link #fromBigDecimal}. Requires source scale and target precision at
most 18.
*/
static @Nullable DecimalData fromCompactRescale(
long unscaled, int fromScale, int precision, int scale) {
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/DecimalDataInternalUtils.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/DecimalDataInternalUtils.java
new file mode 100644
index 00000000000..67ee8ecf442
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/DecimalDataInternalUtils.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.data;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.types.logical.DecimalType;
+
+import java.math.BigDecimal;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Internal factories for {@link DecimalData}. */
+@Internal
+public final class DecimalDataInternalUtils {
+
+ private DecimalDataInternalUtils() {}
+
+ /**
+ * Like {@link DecimalData#fromUnscaledLong}, but for a non-compact
precision, keeping the value
+ * as a long instead of a {@link BigDecimal}. Any long fits a non-compact
precision.
+ */
+ public static DecimalData fromUnscaledLong(long unscaledLong, int
precision, int scale) {
+ checkArgument(!DecimalData.isCompact(precision) && precision <=
DecimalType.MAX_PRECISION);
+ return new DecimalData(precision, scale, unscaledLong, null);
+ }
+}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java
index 653fc559df9..819f5ceab6e 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java
@@ -24,6 +24,7 @@ import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.DecimalDataInternalUtils;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RawValueData;
import org.apache.flink.table.data.RowData;
@@ -1005,11 +1006,32 @@ public final class BinarySegmentUtils {
int scale) {
final int size = ((int) offsetAndSize);
int subOffset = (int) (offsetAndSize >> 32);
+ // up to 8 bytes fit a long; size 0 keeps the old path
+ if (size > 0 && size <= 8) {
+ final int pos = baseOffset + subOffset;
+ final long unscaled =
+ inFirstSegment(segments, pos, 8)
+ // value is in the high bytes, the rest is shifted
out
+ ? segments[0].getLongBigEndian(pos) >> (64 - 8 *
size)
+ : readUnscaledLongMultiSegments(segments, pos,
size);
+ // at most 19 digits, so it fits any non-compact precision
+ return DecimalDataInternalUtils.fromUnscaledLong(unscaled,
precision, scale);
+ }
byte[] bytes = new byte[size];
copyToBytes(segments, baseOffset + subOffset, bytes, 0, size);
return DecimalData.fromUnscaledBytes(bytes, precision, scale);
}
+ private static long readUnscaledLongMultiSegments(
+ MemorySegment[] segments, int offset, int size) {
+ // signed first byte sign-extends
+ long unscaled = getByte(segments, offset);
+ for (int i = 1; i < size; i++) {
+ unscaled = (unscaled << 8) | (getByte(segments, offset + i) &
0xFF);
+ }
+ return unscaled;
+ }
+
/**
* Gets an instance of {@link TimestampData} from underlying {@link
MemorySegment}.
*
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/DecimalDataUtils.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/DecimalDataUtils.java
index 55608af3b81..9c9d6d09dbd 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/DecimalDataUtils.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/DecimalDataUtils.java
@@ -20,6 +20,8 @@ package org.apache.flink.table.data;
import org.apache.flink.table.types.logical.DecimalType;
+import javax.annotation.Nullable;
+
import java.math.BigDecimal;
import java.math.MathContext;
import java.math.RoundingMode;
@@ -46,7 +48,7 @@ public final class DecimalDataUtils {
if (decimal.isCompact()) {
return ((double) decimal.longVal) / POW10[decimal.scale];
} else {
- return decimal.decimalVal.doubleValue();
+ return decimal.toBigDecimal().doubleValue();
}
}
@@ -57,35 +59,35 @@ public final class DecimalDataUtils {
* @return the signum function of this decimal.
*/
public static int signum(DecimalData decimal) {
- if (decimal.isCompact()) {
- return Long.signum(decimal.toUnscaledLong());
+ if (decimal.longBacked) {
+ return Long.signum(decimal.longVal);
} else {
return decimal.toBigDecimal().signum();
}
}
public static DecimalData negate(DecimalData decimal) {
- if (decimal.isCompact()) {
+ if (decimal.longBacked && decimal.longVal != Long.MIN_VALUE) {
return new DecimalData(decimal.precision, decimal.scale,
-decimal.longVal, null);
} else {
return new DecimalData(
- decimal.precision, decimal.scale, -1,
decimal.decimalVal.negate());
+ decimal.precision, decimal.scale, -1,
decimal.toBigDecimal().negate());
}
}
public static DecimalData abs(DecimalData decimal) {
- if (decimal.isCompact()) {
+ if (decimal.longBacked && decimal.longVal != Long.MIN_VALUE) {
if (decimal.longVal >= 0) {
return decimal;
} else {
return new DecimalData(decimal.precision, decimal.scale,
-decimal.longVal, null);
}
} else {
- if (decimal.decimalVal.signum() >= 0) {
+ final BigDecimal bd = decimal.toBigDecimal();
+ if (bd.signum() >= 0) {
return decimal;
} else {
- return new DecimalData(
- decimal.precision, decimal.scale, -1,
decimal.decimalVal.negate());
+ return new DecimalData(decimal.precision, decimal.scale, -1,
bd.negate());
}
}
}
@@ -104,11 +106,7 @@ public final class DecimalDataUtils {
}
public static DecimalData add(DecimalData v1, DecimalData v2, int
precision, int scale) {
- if (v1.isCompact()
- && v2.isCompact()
- && v1.scale == v2.scale
- && DecimalData.isCompact(precision)) {
- assert scale == v1.scale; // no need to rescale
+ if (canAddOrSubtractAsLong(v1, v2, precision, scale)) {
try {
long ls = Math.addExact(v1.longVal, v2.longVal); // checks
overflow
return new DecimalData(precision, scale, ls, null);
@@ -121,11 +119,7 @@ public final class DecimalDataUtils {
}
public static DecimalData subtract(DecimalData v1, DecimalData v2, int
precision, int scale) {
- if (v1.isCompact()
- && v2.isCompact()
- && v1.scale == v2.scale
- && DecimalData.isCompact(precision)) {
- assert scale == v1.scale; // no need to rescale
+ if (canAddOrSubtractAsLong(v1, v2, precision, scale)) {
try {
long ls = Math.subtractExact(v1.longVal, v2.longVal); //
checks overflow
return new DecimalData(precision, scale, ls, null);
@@ -137,6 +131,15 @@ public final class DecimalDataUtils {
return fromBigDecimal(bd, precision, scale);
}
+ private static boolean canAddOrSubtractAsLong(
+ DecimalData v1, DecimalData v2, int precision, int scale) {
+ return (DecimalData.isCompact(precision)
+ ? v1.isCompact() && v2.isCompact()
+ : v1.longBacked && v2.longBacked)
+ && v1.scale == scale
+ && v2.scale == scale;
+ }
+
public static DecimalData multiply(DecimalData v1, DecimalData v2, int
precision, int scale) {
BigDecimal bd = v1.toBigDecimal().multiply(v2.toBigDecimal());
return fromBigDecimal(bd, precision, scale);
@@ -180,12 +183,37 @@ public final class DecimalDataUtils {
}
public static DecimalData castToDecimal(DecimalData dec, int precision,
int scale) {
- if (dec.isCompact() && precision <= DecimalData.MAX_COMPACT_PRECISION)
{
- return DecimalData.fromCompactRescale(dec.longVal, dec.scale,
precision, scale);
+ if (dec.longBacked) {
+ if (DecimalData.isCompact(precision)) {
+ if (dec.scale < POW10.length) {
+ return DecimalData.fromCompactRescale(dec.longVal,
dec.scale, precision, scale);
+ }
+ } else {
+ final DecimalData scaled =
+ rescaleUpAsLong(dec.longVal, dec.scale, precision,
scale);
+ if (scaled != null) {
+ return scaled;
+ }
+ }
}
return fromBigDecimal(dec.toBigDecimal(), precision, scale);
}
+ // scale must not decrease; null if the result overflows a long
+ private static @Nullable DecimalData rescaleUpAsLong(
+ long unscaled, int fromScale, int precision, int scale) {
+ final int scaleDiff = scale - fromScale;
+ if (scaleDiff < 0 || scaleDiff >= POW10.length) {
+ return null;
+ }
+ try {
+ return new DecimalData(
+ precision, scale, Math.multiplyExact(unscaled,
POW10[scaleDiff]), null);
+ } catch (ArithmeticException e) {
+ return null;
+ }
+ }
+
public static DecimalData castFrom(DecimalData dec, int precision, int
scale) {
return castToDecimal(dec, precision, scale);
}
@@ -199,9 +227,13 @@ public final class DecimalDataUtils {
}
public static DecimalData castFrom(long val, int p, int s) {
- if (p <= DecimalData.MAX_COMPACT_PRECISION) {
+ if (DecimalData.isCompact(p)) {
return DecimalData.fromCompactRescale(val, 0, p, s);
}
+ final DecimalData scaled = rescaleUpAsLong(val, 0, p, s);
+ if (scaled != null) {
+ return scaled;
+ }
return fromBigDecimal(BigDecimal.valueOf(val), p, s);
}
@@ -225,8 +257,8 @@ public final class DecimalDataUtils {
}
public static int compare(DecimalData b1, long n2) {
- if (!b1.isCompact()) {
- return b1.decimalVal.compareTo(BigDecimal.valueOf(n2));
+ if (!b1.longBacked || b1.scale >= POW10.length) {
+ return b1.toBigDecimal().compareTo(BigDecimal.valueOf(n2));
}
if (b1.scale == 0) {
return Long.compare(b1.longVal, n2);
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryArrayDataTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryArrayDataTest.java
index cb132cea5b5..838aca87bc8 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryArrayDataTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryArrayDataTest.java
@@ -45,6 +45,8 @@ import java.sql.Timestamp;
import java.time.LocalDateTime;
import static org.apache.flink.table.data.StringData.fromString;
+import static
org.apache.flink.table.data.util.DataFormatTestUtil.nonCompactBoundaryDecimals;
+import static org.apache.flink.table.data.util.DataFormatTestUtil.toSegments;
import static org.apache.flink.table.utils.RawValueDataAsserter.equivalent;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.HamcrestCondition.matching;
@@ -523,6 +525,36 @@ class BinaryArrayDataTest {
array.setDecimal(0, decimal2, precision);
assertThat(array.getDecimal(0, precision,
scale).toString()).isEqualTo("6.55000");
}
+
+ // 3.not compact, values around the 8-byte boundary, non-zero offset
and across segments
+ {
+ int precision = 38;
+ int scale = 2;
+ DecimalData[] decimals = nonCompactBoundaryDecimals(precision,
scale);
+ BinaryArrayData nonCompactArray = new BinaryArrayData();
+ BinaryArrayWriter nonCompactWriter =
+ new BinaryArrayWriter(nonCompactArray, decimals.length, 8);
+ for (int i = 0; i < decimals.length; i++) {
+ nonCompactWriter.writeDecimal(i, decimals[i], precision);
+ }
+ nonCompactWriter.complete();
+
+ int size = nonCompactArray.getSizeInBytes();
+ byte[] bytes =
+ BinarySegmentUtils.copyToBytes(
+ nonCompactArray.getSegments(),
nonCompactArray.getOffset(), size);
+ int baseOffset = 3;
+ for (int segmentSize = 1; segmentSize <= baseOffset + size;
segmentSize++) {
+ BinaryArrayData splitArray = new BinaryArrayData();
+ splitArray.pointTo(toSegments(bytes, baseOffset, segmentSize),
baseOffset, size);
+ for (int i = 0; i < decimals.length; i++) {
+ DecimalData decimal = splitArray.getDecimal(i, precision,
scale);
+
assertThat(decimal.toBigDecimal()).isEqualTo(decimals[i].toBigDecimal());
+ assertThat(decimal.longBacked)
+ .isEqualTo(decimals[i].toUnscaledBytes().length <=
8);
+ }
+ }
+ }
}
@Test
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryRowDataTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryRowDataTest.java
index 18b756f9ea9..b9b74029c7a 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryRowDataTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryRowDataTest.java
@@ -37,6 +37,7 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.binary.BinaryArrayData;
import org.apache.flink.table.data.binary.BinaryMapData;
import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.binary.BinarySegmentUtils;
import org.apache.flink.table.data.binary.BinaryStringData;
import org.apache.flink.table.data.util.DataFormatTestUtil;
import org.apache.flink.table.data.writer.BinaryArrayWriter;
@@ -81,6 +82,8 @@ import java.util.Set;
import static org.apache.flink.table.data.StringData.fromBytes;
import static org.apache.flink.table.data.StringData.fromString;
import static org.apache.flink.table.data.util.DataFormatTestUtil.MyObj;
+import static
org.apache.flink.table.data.util.DataFormatTestUtil.nonCompactBoundaryDecimals;
+import static org.apache.flink.table.data.util.DataFormatTestUtil.toSegments;
import static org.apache.flink.table.data.util.MapDataUtil.convertToJavaMap;
import static org.apache.flink.table.utils.RawValueDataAsserter.equivalent;
import static org.assertj.core.api.Assertions.assertThat;
@@ -453,6 +456,39 @@ class BinaryRowDataTest {
row.setDecimal(0, decimal2, precision);
assertThat(row.getDecimal(0, precision,
scale).toString()).isEqualTo("6.55000");
}
+
+ // 3.not compact, values around the 8-byte boundary
+ {
+ int precision = 38;
+ int scale = 2;
+ for (DecimalData decimal : nonCompactBoundaryDecimals(precision,
scale)) {
+ BinaryRowData row = new BinaryRowData(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeDecimal(0, decimal, precision);
+ writer.complete();
+
+ assertThat(row.getDecimal(0, precision, scale).toBigDecimal())
+ .isEqualTo(decimal.toBigDecimal());
+ assertThat(row.getDecimal(0, precision, scale).longBacked)
+ .isEqualTo(decimal.toUnscaledBytes().length <= 8);
+ row.setDecimal(0, row.getDecimal(0, precision, scale),
precision);
+ assertThat(row.getDecimal(0, precision, scale).toBigDecimal())
+ .isEqualTo(decimal.toBigDecimal());
+
+ // split the row so that the decimal bytes cross a segment
boundary
+ int fixedSize = row.getFixedLengthPartSize();
+ byte[] bytes =
+ BinarySegmentUtils.copyToBytes(row.getSegments(), 0,
row.getSizeInBytes());
+ for (int split = 1; split < 8; split++) {
+ BinaryRowData splitRow = new BinaryRowData(1);
+ splitRow.pointTo(
+ toSegments(bytes, 0, fixedSize + split), 0,
row.getSizeInBytes());
+
+ assertThat(splitRow.getDecimal(0, precision,
scale).toBigDecimal())
+ .isEqualTo(decimal.toBigDecimal());
+ }
+ }
+ }
}
@Test
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/DecimalDataTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/DecimalDataTest.java
index f64c0e5108e..4f6e444856f 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/DecimalDataTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/DecimalDataTest.java
@@ -90,7 +90,7 @@ class DecimalDataTest {
DecimalData.fromBigDecimal(BigDecimal.valueOf(val),
targetPrecision, targetScale);
final DecimalData actual = castFrom(val, targetPrecision, targetScale);
- assertCastMatches(actual, expected);
+ assertDecimalMatches(actual, expected);
}
@ParameterizedTest(name = "unscaled={0} DECIMAL({1},{2}) ->
DECIMAL({3},{4})")
@@ -102,10 +102,142 @@ class DecimalDataTest {
DecimalData.fromBigDecimal(src.toBigDecimal(),
targetPrecision, targetScale);
final DecimalData actual = castToDecimal(src, targetPrecision,
targetScale);
- assertCastMatches(actual, expected);
+ assertDecimalMatches(actual, expected);
}
- private static void assertCastMatches(DecimalData actual, DecimalData
expected) {
+ private static Stream<Arguments> unscaledLongCases() {
+ return Stream.of(
+ arguments(0L),
+ arguments(1L),
+ arguments(-1L),
+ arguments(127L),
+ arguments(128L),
+ arguments(-128L),
+ arguments(-129L),
+ arguments(1234L),
+ arguments(32767L),
+ arguments(32768L),
+ arguments(-32768L),
+ arguments(-32769L),
+ arguments((long) Integer.MAX_VALUE),
+ arguments((long) Integer.MAX_VALUE + 1),
+ arguments((long) Integer.MIN_VALUE),
+ arguments((long) Integer.MIN_VALUE - 1),
+ arguments((1L << 55) - 1),
+ arguments(1L << 55),
+ arguments(-(1L << 55)),
+ arguments(-(1L << 55) - 1),
+ arguments(Long.MAX_VALUE),
+ arguments(Long.MIN_VALUE));
+ }
+
+ @ParameterizedTest(name = "unscaled={0}")
+ @MethodSource("unscaledLongCases")
+ void testNonCompactLongBackedMatchesBigDecimal(long unscaled) {
+ final DecimalData longBacked =
DecimalDataInternalUtils.fromUnscaledLong(unscaled, 38, 2);
+ final DecimalData bigBacked =
+ DecimalData.fromBigDecimal(BigDecimal.valueOf(unscaled, 2),
38, 2);
+
+ assertThat(longBacked.longBacked).isTrue();
+
assertThat(longBacked.toUnscaledBytes()).isEqualTo(bigBacked.toUnscaledBytes());
+ assertThat(longBacked.toUnscaledLong()).isEqualTo(unscaled);
+ assertThat(longBacked.compareTo(bigBacked)).isZero();
+
assertThat(longBacked).isEqualTo(bigBacked).hasSameHashCodeAs(bigBacked);
+
+ // caching the BigDecimal must keep the long fast paths, also for
copies
+ longBacked.toBigDecimal();
+ assertThat(longBacked.longBacked).isTrue();
+ assertThat(longBacked.copy().longBacked).isTrue();
+
+ assertThat(doubleValue(longBacked))
+ .isEqualTo(BigDecimal.valueOf(unscaled, 2).doubleValue());
+ assertThat(signum(longBacked)).isEqualTo(Long.signum(unscaled));
+ assertThat(compare(longBacked, unscaled / 100))
+
.isEqualTo(bigBacked.toBigDecimal().compareTo(BigDecimal.valueOf(unscaled /
100)));
+
+ final DecimalData otherScale =
DecimalDataInternalUtils.fromUnscaledLong(unscaled, 38, 3);
+ assertThat(Integer.signum(longBacked.compareTo(otherScale)))
+ .isEqualTo(
+ Integer.signum(
+
bigBacked.toBigDecimal().compareTo(otherScale.toBigDecimal())));
+
+ assertDecimalMatches(negate(longBacked), negate(bigBacked));
+ assertThat(negate(longBacked).longBacked).isEqualTo(unscaled !=
Long.MIN_VALUE);
+ assertDecimalMatches(abs(longBacked), abs(bigBacked));
+
+ final DecimalData one = DecimalDataInternalUtils.fromUnscaledLong(1,
38, 2);
+ assertDecimalMatches(add(longBacked, one, 38, 2), add(bigBacked, one,
38, 2));
+ assertThat(add(longBacked, one, 38, 2).longBacked).isEqualTo(unscaled
!= Long.MAX_VALUE);
+ assertDecimalMatches(subtract(longBacked, one, 38, 2),
subtract(bigBacked, one, 38, 2));
+ assertThat(subtract(longBacked, one, 38, 2).longBacked)
+ .isEqualTo(unscaled != Long.MIN_VALUE);
+ assertDecimalMatches(
+ add(longBacked, longBacked, 38, 2),
+ DecimalData.fromBigDecimal(
+
bigBacked.toBigDecimal().add(bigBacked.toBigDecimal()), 38, 2));
+ assertDecimalMatches(
+ subtract(longBacked, negate(longBacked), 38, 2),
+ DecimalData.fromBigDecimal(
+
bigBacked.toBigDecimal().subtract(bigBacked.toBigDecimal().negate()),
+ 38,
+ 2));
+ }
+
+ @ParameterizedTest(name = "unscaled={0}")
+ @MethodSource("unscaledLongCases")
+ void testCompactHashCodeMatchesBigDecimal(long unscaled) {
+ final long compact = unscaled % 1_000_000_000_000_000L;
+ final BigDecimal bd = BigDecimal.valueOf(compact, 2);
+
+ assertThat(DecimalData.fromUnscaledLong(compact, 18, 2).hashCode())
+ .isEqualTo(bd.hashCode());
+ assertThat(DecimalData.fromBigDecimal(bd, 18,
2).hashCode()).isEqualTo(bd.hashCode());
+ }
+
+ private static Stream<Arguments> unscaledLongWithTargetCases() {
+ final int[][] targets = {
+ {38, 2}, {38, 4}, {38, 20}, {38, 1}, {19, 0}, {20, 2}, {18, 2},
{18, 0}, {10, 1}
+ };
+ return unscaledLongCases()
+ .flatMap(
+ value ->
+ Stream.of(targets).map(t ->
arguments(value.get()[0], t[0], t[1])));
+ }
+
+ @ParameterizedTest(name = "unscaled={0} -> DECIMAL({1},{2})")
+ @MethodSource("unscaledLongWithTargetCases")
+ void testCastLongBackedMatchesBigDecimal(long unscaled, int precision, int
scale) {
+ final DecimalData longBacked =
DecimalDataInternalUtils.fromUnscaledLong(unscaled, 38, 2);
+ final DecimalData bigBacked =
+ DecimalData.fromBigDecimal(BigDecimal.valueOf(unscaled, 2),
38, 2);
+
+ final DecimalData actual = castToDecimal(longBacked, precision, scale);
+ assertDecimalMatches(actual, castToDecimal(bigBacked, precision,
scale));
+ if (actual != null && scale >= 2 && !DecimalData.isCompact(precision))
{
+ assertThat(actual.longBacked)
+
.isEqualTo(actual.toBigDecimal().unscaledValue().bitLength() < 64);
+ }
+ }
+
+ @ParameterizedTest(name = "unscaled={0} -> DECIMAL({1},{2})")
+ @MethodSource("unscaledLongWithTargetCases")
+ void testCastCompactMatchesBigDecimal(long unscaled, int precision, int
scale) {
+ final DecimalData compact = DecimalData.fromUnscaledLong(unscaled %
10_000_000_000L, 10, 2);
+
+ assertDecimalMatches(
+ castToDecimal(compact, precision, scale),
+ DecimalData.fromBigDecimal(compact.toBigDecimal(), precision,
scale));
+ }
+
+ @ParameterizedTest(name = "{0} -> DECIMAL({1},{2})")
+ @MethodSource("unscaledLongWithTargetCases")
+ void testCastFromLongMatchesBigDecimal(long val, int precision, int scale)
{
+ assertDecimalMatches(
+ castFrom(val, precision, scale),
+ DecimalData.fromBigDecimal(BigDecimal.valueOf(val), precision,
scale));
+ }
+
+ private static void assertDecimalMatches(DecimalData actual, DecimalData
expected) {
if (expected == null) {
assertThat(actual).isNull();
} else {
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/NestedRowDataTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/NestedRowDataTest.java
index 34a98ccd843..1b1e7c93d86 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/NestedRowDataTest.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/NestedRowDataTest.java
@@ -26,17 +26,23 @@ import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.binary.BinarySegmentUtils;
import org.apache.flink.table.data.binary.NestedRowData;
import org.apache.flink.table.data.writer.BinaryRowWriter;
import org.apache.flink.table.runtime.typeutils.RawValueDataSerializer;
import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
import org.apache.flink.table.runtime.typeutils.StringDataSerializer;
+import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.LogicalType;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
+
import static org.apache.flink.table.data.util.DataFormatTestUtil.MyObj;
+import static
org.apache.flink.table.data.util.DataFormatTestUtil.nonCompactBoundaryDecimals;
import static org.apache.flink.table.data.util.DataFormatTestUtil.splitBytes;
+import static org.apache.flink.table.data.util.DataFormatTestUtil.toSegments;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link NestedRowData}s. */
@@ -129,6 +135,36 @@ class NestedRowDataTest {
assertThat(nestedRow.isNullAt(3)).isTrue();
}
+ @Test
+ void testNonCompactDecimalsWithMultipleSegments() {
+ int precision = 38;
+ int scale = 2;
+ DecimalData[] decimals = nonCompactBoundaryDecimals(precision, scale);
+ LogicalType[] types = new LogicalType[decimals.length];
+ Arrays.fill(types, new DecimalType(precision, scale));
+
+ BinaryRowData row = new BinaryRowData(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeRow(0, GenericRowData.of((Object[]) decimals), new
RowDataSerializer(types));
+ writer.complete();
+
+ int size = row.getSizeInBytes();
+ byte[] bytes = BinarySegmentUtils.copyToBytes(row.getSegments(),
row.getOffset(), size);
+ int baseOffset = 3;
+ // BinaryRowData requires its fixed-length part in the first segment
+ for (int segmentSize = baseOffset + row.getFixedLengthPartSize();
+ segmentSize <= baseOffset + size;
+ segmentSize++) {
+ row.pointTo(toSegments(bytes, baseOffset, segmentSize),
baseOffset, size);
+ RowData nestedRow = row.getRow(0, decimals.length);
+ for (int i = 0; i < decimals.length; i++) {
+ DecimalData decimal = nestedRow.getDecimal(i, precision,
scale);
+
assertThat(decimal.toBigDecimal()).isEqualTo(decimals[i].toBigDecimal());
+
assertThat(decimal.longBacked).isEqualTo(decimals[i].toUnscaledBytes().length
<= 8);
+ }
+ }
+ }
+
private BinaryRowData getBinaryRowData() {
BinaryRowData row = new BinaryRowData(1);
BinaryRowWriter writer = new BinaryRowWriter(row);
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/util/DataFormatTestUtil.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/util/DataFormatTestUtil.java
index 4402056fea6..599209dc8f2 100644
---
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/util/DataFormatTestUtil.java
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/util/DataFormatTestUtil.java
@@ -19,6 +19,7 @@ package org.apache.flink.table.data.util;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.binary.BinaryRowData;
@@ -27,6 +28,10 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.commons.lang3.RandomStringUtils;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.stream.Stream;
+
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.assertj.core.api.Assertions.assertThat;
@@ -121,6 +126,37 @@ public class DataFormatTestUtil {
return ret;
}
+ /** Non-compact decimals with unscaled values around the 8-byte boundary.
*/
+ public static DecimalData[] nonCompactBoundaryDecimals(int precision, int
scale) {
+ return Stream.of(
+ "0",
+ "-1",
+ "128",
+ "-129",
+ "9223372036854775807",
+ "-9223372036854775808",
+ "9223372036854775808",
+ "-9223372036854775809",
+ "12345678901234567890123456789012")
+ .map(v -> new BigDecimal(new BigInteger(v), scale))
+ .map(v -> DecimalData.fromBigDecimal(v, precision, scale))
+ .toArray(DecimalData[]::new);
+ }
+
+ /** Copies bytes to equally sized segments, starting at baseOffset of the
first one. */
+ public static MemorySegment[] toSegments(byte[] bytes, int baseOffset, int
segmentSize) {
+ MemorySegment[] ret =
+ new MemorySegment[(baseOffset + bytes.length + segmentSize -
1) / segmentSize];
+ for (int i = 0; i < ret.length; i++) {
+ ret[i] = MemorySegmentFactory.wrap(new byte[segmentSize]);
+ }
+ for (int i = 0; i < bytes.length; i++) {
+ int pos = baseOffset + i;
+ ret[pos / segmentSize].put(pos % segmentSize, bytes[i]);
+ }
+ return ret;
+ }
+
/** A simple class for testing generic type getting / setting on data
formats. */
public static class MyObj {
public int i;
diff --git
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/DecimalDataSerializer.java
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/DecimalDataSerializer.java
index 05f2757008b..f67e331bad5 100644
---
a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/DecimalDataSerializer.java
+++
b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/DecimalDataSerializer.java
@@ -25,6 +25,7 @@ import
org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.DecimalDataInternalUtils;
import java.io.IOException;
@@ -86,6 +87,15 @@ public final class DecimalDataSerializer extends
TypeSerializer<DecimalData> {
return DecimalData.fromUnscaledLong(longVal, precision, scale);
} else {
int length = source.readInt();
+ if (length > 0 && length <= 8) {
+ // signed first byte sign-extends
+ long unscaled = source.readByte();
+ for (int i = 1; i < length; i++) {
+ unscaled = (unscaled << 8) | source.readUnsignedByte();
+ }
+ // at most 19 digits, so it fits any non-compact precision
+ return DecimalDataInternalUtils.fromUnscaledLong(unscaled,
precision, scale);
+ }
byte[] bytes = new byte[length];
source.readFully(bytes);
return DecimalData.fromUnscaledBytes(bytes, precision, scale);
diff --git
a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/DecimalSerializerTest.java
b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/DecimalSerializerTest.java
index 821e2bd99c7..c69085c67c0 100644
---
a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/DecimalSerializerTest.java
+++
b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/DecimalSerializerTest.java
@@ -19,33 +19,109 @@
package org.apache.flink.table.runtime.typeutils;
import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.DecimalDataInternalUtils;
+import org.apache.flink.util.StringUtils;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
/** A test for the {@link DecimalDataSerializer}. */
-class DecimalSerializerTest extends SerializerTestBase<DecimalData> {
+class DecimalSerializerTest {
+
+ /** Bytes as written by the BigDecimal-only serializer, which must stay
readable. */
+ @ParameterizedTest(name = "{0} -> {1}")
+ @CsvSource({
+ "00, 0",
+ "7f, 127",
+ "0080, 128",
+ "80, -128",
+ "ff7f, -129",
+ "7fffffffffffffff, 9223372036854775807",
+ "8000000000000000, -9223372036854775808",
+ "008000000000000000, 9223372036854775808",
+ "ff7fffffffffffffff, -9223372036854775809"
+ })
+ void testDeserializeLegacyNonCompactBytes(String hex, String unscaled)
throws IOException {
+ DecimalDataSerializer serializer = new DecimalDataSerializer(38, 2);
+ byte[] bytes = StringUtils.hexStringToByte(hex);
+ BigDecimal expected = new BigDecimal(new BigInteger(unscaled), 2);
+
+ DataOutputSerializer out = new DataOutputSerializer(16);
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ byte[] serialized = out.getCopyOfBuffer();
- @Override
- protected DecimalDataSerializer createSerializer() {
- return new DecimalDataSerializer(5, 2);
+ DecimalData decimal = serializer.deserialize(new
DataInputDeserializer(serialized));
+ assertThat(decimal.toBigDecimal()).isEqualTo(expected);
+ assertThat(decimal.hashCode()).isEqualTo(expected.hashCode());
+
+ out.clear();
+ serializer.serialize(decimal, out);
+ assertThat(out.getCopyOfBuffer()).isEqualTo(serialized);
}
- @Override
- protected int getLength() {
- return -1;
+ @Nested
+ final class CompactDecimalSerializerTest extends DecimalSerializerTestBase
{
+ @Override
+ protected DecimalDataSerializer createSerializer() {
+ return new DecimalDataSerializer(5, 2);
+ }
+
+ @Override
+ protected DecimalData[] getTestData() {
+ return new DecimalData[] {
+ DecimalData.fromUnscaledLong(1, 5, 2),
+ DecimalData.fromUnscaledLong(2, 5, 2),
+ DecimalData.fromUnscaledLong(3, 5, 2),
+ DecimalData.fromUnscaledLong(4, 5, 2)
+ };
+ }
}
- @Override
- protected Class<DecimalData> getTypeClass() {
- return DecimalData.class;
+ @Nested
+ final class NonCompactDecimalSerializerTest extends
DecimalSerializerTestBase {
+ @Override
+ protected DecimalDataSerializer createSerializer() {
+ return new DecimalDataSerializer(38, 2);
+ }
+
+ @Override
+ protected DecimalData[] getTestData() {
+ return new DecimalData[] {
+ DecimalDataInternalUtils.fromUnscaledLong(0, 38, 2),
+ DecimalDataInternalUtils.fromUnscaledLong(-1, 38, 2),
+ DecimalDataInternalUtils.fromUnscaledLong(128, 38, 2),
+ DecimalDataInternalUtils.fromUnscaledLong(-129, 38, 2),
+ DecimalDataInternalUtils.fromUnscaledLong(Long.MAX_VALUE, 38,
2),
+ DecimalDataInternalUtils.fromUnscaledLong(Long.MIN_VALUE, 38,
2),
+ DecimalData.fromBigDecimal(new BigDecimal("12.34"), 38, 2),
+ DecimalData.fromBigDecimal(new
BigDecimal("92233720368547758.08"), 38, 2),
+ DecimalData.fromBigDecimal(
+ new BigDecimal("-123456789012345678901234567890.12"),
38, 2)
+ };
+ }
}
- @Override
- protected DecimalData[] getTestData() {
- return new DecimalData[] {
- DecimalData.fromUnscaledLong(1, 5, 2),
- DecimalData.fromUnscaledLong(2, 5, 2),
- DecimalData.fromUnscaledLong(3, 5, 2),
- DecimalData.fromUnscaledLong(4, 5, 2)
- };
+ abstract static class DecimalSerializerTestBase extends
SerializerTestBase<DecimalData> {
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected Class<DecimalData> getTypeClass() {
+ return DecimalData.class;
+ }
}
}
