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 a1a3ca7ea64 [FLINK-41017][table] Use `DecimalData#isCompact` while
casting
a1a3ca7ea64 is described below
commit a1a3ca7ea6415f0491d1db2960f6702ed4557b69
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Sun Oct 18 11:40:06 2026 +0200
[FLINK-41017][table] Use `DecimalData#isCompact` while casting
---
.../org/apache/flink/table/data/DecimalData.java | 28 +++++++++
.../apache/flink/table/data/DecimalDataUtils.java | 8 ++-
.../apache/flink/table/data/DecimalDataTest.java | 67 ++++++++++++++++++++++
3 files changed, 102 insertions(+), 1 deletion(-)
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 31081c0a3c6..5f4e4941163 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
@@ -194,6 +194,34 @@ public final class DecimalData implements
Comparable<DecimalData> {
return new DecimalData(precision, scale, longVal, bd);
}
+ /**
+ * 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.
+ */
+ static @Nullable DecimalData fromCompactRescale(
+ long unscaled, int fromScale, int precision, int scale) {
+ final int scaleDiff = scale - fromScale; // both scales in [0, 18] =>
scaleDiff in [-18, 18]
+ final long result;
+ if (scaleDiff >= 0) {
+ final long factor = POW10[scaleDiff];
+ result = unscaled * factor; // exact scale-up
+ if (result / factor != unscaled) { // long overflow => precision
overflow
+ return null;
+ }
+ } else {
+ final long divisor = POW10[-scaleDiff];
+ final long rem = unscaled % divisor; // HALF_UP: round half away
from zero
+ result =
+ unscaled / divisor + (Math.abs(rem) * 2 >= divisor ?
Long.signum(unscaled) : 0);
+ }
+ final long bound = POW10[precision]; // digit count must not exceed
precision
+ if (result <= -bound || result >= bound) {
+ return null;
+ }
+ return new DecimalData(precision, scale, result, null);
+ }
+
/**
* Creates an instance of {@link DecimalData} from an unscaled long value
and the given
* precision and scale.
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 d88848ab311..55608af3b81 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
@@ -180,11 +180,14 @@ 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);
+ }
return fromBigDecimal(dec.toBigDecimal(), precision, scale);
}
public static DecimalData castFrom(DecimalData dec, int precision, int
scale) {
- return fromBigDecimal(dec.toBigDecimal(), precision, scale);
+ return castToDecimal(dec, precision, scale);
}
public static DecimalData castFrom(String string, int precision, int
scale) {
@@ -196,6 +199,9 @@ public final class DecimalDataUtils {
}
public static DecimalData castFrom(long val, int p, int s) {
+ if (p <= DecimalData.MAX_COMPACT_PRECISION) {
+ return DecimalData.fromCompactRescale(val, 0, p, s);
+ }
return fromBigDecimal(BigDecimal.valueOf(val), p, s);
}
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 76fb7dbf61b..f64c0e5108e 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
@@ -19,8 +19,12 @@
package org.apache.flink.table.data;
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.apache.flink.table.data.DecimalDataUtils.abs;
import static org.apache.flink.table.data.DecimalDataUtils.add;
@@ -45,10 +49,73 @@ import static
org.apache.flink.table.data.DecimalDataUtils.signum;
import static org.apache.flink.table.data.DecimalDataUtils.sround;
import static org.apache.flink.table.data.DecimalDataUtils.subtract;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
/** Test for {@link DecimalData}. */
class DecimalDataTest {
+ private static Stream<Arguments> compactCastCases() {
+ return Stream.of(
+ // unscaled, srcPrecision, srcScale, targetPrecision,
targetScale
+ arguments(0L, 5, 2, 3, 1),
+ arguments(123L, 5, 2, 6, 2),
+ arguments(12345L, 5, 0, 10, 2),
+ arguments(24L, 2, 1, 2, 0),
+ arguments(25L, 2, 1, 2, 0),
+ arguments(-25L, 2, 1, 2, 0),
+ arguments(995L, 3, 2, 2, 1),
+ arguments(-995L, 3, 2, 2, 1),
+ arguments(999_999_999_999_999_999L, 18, 0, 18, 2),
+ arguments(999_999_999_999_999_999L, 18, 0, 18, 0),
+ arguments(-999_999_999_999_999_999L, 18, 0, 18, 0),
+ arguments(12345L, 5, 2, 3, 2));
+ }
+
+ private static Stream<Arguments> longCastCases() {
+ return Stream.of(
+ // val, targetPrecision, targetScale
+ arguments(0L, 3, 1),
+ arguments(123L, 5, 2),
+ arguments(-123L, 5, 2),
+ arguments(Long.MAX_VALUE, 18, 0),
+ arguments(Long.MIN_VALUE, 18, 0),
+ arguments(999_999_999_999_999_999L, 18, 0),
+ arguments(99L, 2, 1));
+ }
+
+ @ParameterizedTest(name = "{0} -> DECIMAL({1},{2})")
+ @MethodSource("longCastCases")
+ void testCastFromLongCompactMatchesBigDecimal(long val, int
targetPrecision, int targetScale) {
+ final DecimalData expected =
+ DecimalData.fromBigDecimal(BigDecimal.valueOf(val),
targetPrecision, targetScale);
+ final DecimalData actual = castFrom(val, targetPrecision, targetScale);
+
+ assertCastMatches(actual, expected);
+ }
+
+ @ParameterizedTest(name = "unscaled={0} DECIMAL({1},{2}) ->
DECIMAL({3},{4})")
+ @MethodSource("compactCastCases")
+ void testCastToDecimalCompactMatchesBigDecimal(
+ long unscaled, int srcPrecision, int srcScale, int
targetPrecision, int targetScale) {
+ final DecimalData src = DecimalData.fromUnscaledLong(unscaled,
srcPrecision, srcScale);
+ final DecimalData expected =
+ DecimalData.fromBigDecimal(src.toBigDecimal(),
targetPrecision, targetScale);
+ final DecimalData actual = castToDecimal(src, targetPrecision,
targetScale);
+
+ assertCastMatches(actual, expected);
+ }
+
+ private static void assertCastMatches(DecimalData actual, DecimalData
expected) {
+ if (expected == null) {
+ assertThat(actual).isNull();
+ } else {
+ assertThat(actual).isNotNull();
+ assertThat(actual.precision()).isEqualTo(expected.precision());
+ assertThat(actual.scale()).isEqualTo(expected.scale());
+
assertThat(actual.toBigDecimal()).isEqualTo(expected.toBigDecimal());
+ }
+ }
+
@SuppressWarnings("ConstantConditions")
@Test
void testNormal() {
