This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 2991f50d9066 fix: Fix schema evolution decimal precision overflow
(#20130)
2991f50d9066 is described below
commit 2991f50d9066051bf4230cd650b8caa2914f00ef
Author: Ashok Raminedi <[email protected]>
AuthorDate: Fri Oct 2 10:20:41 2026 -0400
fix: Fix schema evolution decimal precision overflow (#20130)
* Fix schema evolution decimal precision overflow
* fix: handle decimal precision overflow in vectorized reads
* Fix schema evolution overflow assertion
---
.../client/utils/SparkInternalSchemaConverter.java | 32 +++-
.../util/TestSparkInternalSchemaConverter.scala | 42 ++++-
.../TestVectorizedReadWithSchemaEvolution.scala | 205 ++++++++++++++++-----
3 files changed, 228 insertions(+), 51 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkInternalSchemaConverter.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkInternalSchemaConverter.java
index f948b7ea5829..08feb7f38539 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkInternalSchemaConverter.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkInternalSchemaConverter.java
@@ -28,6 +28,7 @@ import org.apache.hudi.common.util.collection.Pair;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
+import org.apache.spark.sql.internal.SQLConf;
import org.apache.spark.sql.types.ArrayType;
import org.apache.spark.sql.types.ArrayType$;
import org.apache.spark.sql.types.BinaryType;
@@ -325,6 +326,22 @@ public class SparkInternalSchemaConverter {
}
}
+ private static void putDecimalOrNull(
+ WritableColumnVector newV, int rowId, Decimal decimal, DecimalType
decimalType) {
+ if (decimal.changePrecision(decimalType.precision(), decimalType.scale()))
{
+ newV.putDecimal(rowId, decimal, decimalType.precision());
+ } else if (SQLConf.get().ansiEnabled()) {
+ throw new ArithmeticException(
+ String.format(
+ "%s cannot be represented as Decimal(%d, %d)",
+ decimal,
+ decimalType.precision(),
+ decimalType.scale()));
+ } else {
+ newV.putNull(rowId);
+ }
+ }
+
/**
* Convert Int/long type to other Type.
* Now only support int/long -> long/float/double/string/Decimal
@@ -349,9 +366,9 @@ public class SparkInternalSchemaConverter {
} else if (newType instanceof StringType) {
newV.putByteArray(i, getUTF8Bytes((isInt ? oldV.getInt(i) :
oldV.getLong(i)) + ""));
} else if (newType instanceof DecimalType) {
+ DecimalType decimalType = (DecimalType) newType;
Decimal oldDecimal = Decimal.apply(isInt ? oldV.getInt(i) :
oldV.getLong(i));
- oldDecimal.changePrecision(((DecimalType) newType).precision(),
((DecimalType) newType).scale());
- newV.putDecimal(i, oldDecimal, ((DecimalType) newType).precision());
+ putDecimalOrNull(newV, i, oldDecimal, decimalType);
}
}
return true;
@@ -377,9 +394,9 @@ public class SparkInternalSchemaConverter {
} else if (newType instanceof StringType) {
newV.putByteArray(i, getUTF8Bytes(oldV.getFloat(i) + ""));
} else if (newType instanceof DecimalType) {
+ DecimalType decimalType = (DecimalType) newType;
Decimal oldDecimal = Decimal.apply(oldV.getFloat(i));
- oldDecimal.changePrecision(((DecimalType) newType).precision(),
((DecimalType) newType).scale());
- newV.putDecimal(i, oldDecimal, ((DecimalType) newType).precision());
+ putDecimalOrNull(newV, i, oldDecimal, decimalType);
}
}
return true;
@@ -401,9 +418,9 @@ public class SparkInternalSchemaConverter {
}
// double -> decimal/string
if (newType instanceof DecimalType) {
+ DecimalType decimalType = (DecimalType) newType;
Decimal oldDecimal = Decimal.apply(oldV.getDouble(i));
- oldDecimal.changePrecision(((DecimalType) newType).precision(),
((DecimalType) newType).scale());
- newV.putDecimal(i, oldDecimal, ((DecimalType) newType).precision());
+ putDecimalOrNull(newV, i, oldDecimal, decimalType);
} else if (newType instanceof StringType) {
newV.putByteArray(i, getUTF8Bytes(oldV.getDouble(i) + ""));
}
@@ -481,8 +498,7 @@ public class SparkInternalSchemaConverter {
DecimalType decimalType = (DecimalType) newType;
java.math.BigDecimal bigDecimal = new
java.math.BigDecimal(oldV.getUTF8String(i).toString().trim());
Decimal sparkDecimal = Decimal.apply(bigDecimal);
- sparkDecimal.changePrecision(decimalType.precision(),
decimalType.scale());
- newV.putDecimal(i, sparkDecimal, decimalType.precision());
+ putDecimalOrNull(newV, i, sparkDecimal, decimalType);
}
}
return true;
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/util/TestSparkInternalSchemaConverter.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/util/TestSparkInternalSchemaConverter.scala
index e2afc593d71f..0b55eb33f0cd 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/util/TestSparkInternalSchemaConverter.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/util/TestSparkInternalSchemaConverter.scala
@@ -20,12 +20,14 @@
package org.apache.hudi.util
import org.apache.hudi.SparkAdapterSupport
-import
org.apache.hudi.client.utils.SparkInternalSchemaConverter.collectColNamesFromSparkStruct
+import
org.apache.hudi.client.utils.SparkInternalSchemaConverter.{collectColNamesFromSparkStruct,
convertColumnVectorType}
import org.apache.hudi.common.schema.HoodieSchema
import
org.apache.hudi.common.schema.internal.convert.TestInternalSchemaConverter._
import org.apache.hudi.testutils.HoodieSparkClientTestHarness
import org.apache.spark.sql.avro.HoodieSparkSchemaConverters
+import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import org.junit.jupiter.api.Test
@@ -36,6 +38,44 @@ class TestSparkInternalSchemaConverter extends
HoodieSparkClientTestHarness with
HoodieSparkSchemaConverters.toSqlType(schema)._1
}
+ @Test
+ def testDecimalPrecisionOverflow(): Unit = {
+ val conf = SQLConf.get
+ val previousAnsi = conf.getConfString("spark.sql.ansi.enabled", null)
+ conf.setConfString("spark.sql.ansi.enabled", "false")
+
+ try {
+ val decimalType = DecimalType(4, 2)
+
+ Seq[(DataType, OnHeapColumnVector => Unit)](
+ IntegerType -> (_.putInt(0, 12345)),
+ LongType -> (_.putLong(0, 12345L)),
+ FloatType -> (_.putFloat(0, 12345.0f)),
+ DoubleType -> (_.putDouble(0, 12345.0)),
+ StringType -> (_.putByteArray(0, "12345".getBytes("UTF-8")))
+ ).foreach { case (sourceType, putValue) =>
+ val oldV = new OnHeapColumnVector(1, sourceType)
+ val newV = new OnHeapColumnVector(1, decimalType)
+
+ try {
+ putValue(oldV)
+
+ assertTrue(convertColumnVectorType(oldV, newV, 1))
+ assertTrue(newV.isNullAt(0), s"Expected NULL for $sourceType ->
$decimalType overflow")
+ } finally {
+ oldV.close()
+ newV.close()
+ }
+ }
+ } finally {
+ if (previousAnsi == null) {
+ conf.unsetConf("spark.sql.ansi.enabled")
+ } else {
+ conf.setConfString("spark.sql.ansi.enabled", previousAnsi)
+ }
+ }
+ }
+
@Test
def testCollectColumnNames(): Unit = {
val simpleSchema = StructType(Seq(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestVectorizedReadWithSchemaEvolution.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestVectorizedReadWithSchemaEvolution.scala
index 5738621bea18..263627943c74 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestVectorizedReadWithSchemaEvolution.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestVectorizedReadWithSchemaEvolution.scala
@@ -17,53 +17,174 @@
package org.apache.spark.sql.hudi.common
-import org.apache.hudi.HoodieSparkUtils
+import org.apache.spark.SparkException
class TestVectorizedReadWithSchemaEvolution extends HoodieSparkSqlTestBase {
Seq("cow", "mor").foreach { tableType =>
test(s"Test vectorized read for $tableType table") {
- if (HoodieSparkUtils.isSpark3) {
- withSQLConf(
- "hoodie.schema.on.read.enable" -> "true",
- "spark.sql.parquet.enableVectorizedReader" -> "true",
- "spark.sql.codegen.maxFields" -> "1",
- "hoodie.parquet.small.file.limit" -> "0"
- ) {
- withTempDir { tmp =>
- val tableName = generateTableName
- val tablePath = s"${tmp.getCanonicalPath}/$tableName"
- // create table
- spark.sql(
- s"""
- |create table $tableName (
- | id int,
- | name string,
- | price double,
- | ts long
- |) using hudi
- | partitioned by (ts)
- | location '$tablePath'
- | tblproperties (
- | type = '$tableType',
- | primaryKey = 'id',
- | orderingFields = 'ts'
- | )
- """.stripMargin)
- // insert data to table
- spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
-
- // alter table change data type of column 'id'
- spark.sql(s"alter table $tableName alter column price type string")
-
- // insert new records
- spark.sql(s"insert into $tableName values(2, 'a2', '20', 2000)")
-
- checkAnswer(s"select id, name, price, ts from $tableName")(
- Seq(1, "a1", "10.0", 1000),
- Seq(2, "a2", "20", 2000)
- )
- }
+ withSQLConf(
+ "hoodie.schema.on.read.enable" -> "true",
+ "spark.sql.parquet.enableVectorizedReader" -> "true",
+ "spark.sql.codegen.maxFields" -> "1",
+ "hoodie.parquet.small.file.limit" -> "0"
+ ) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ // create table
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | partitioned by (ts)
+ | location '$tablePath'
+ | tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ """.stripMargin)
+ // insert data to table
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+
+ // alter table change data type of column 'id'
+ spark.sql(s"alter table $tableName alter column price type string")
+
+ // insert new records
+ spark.sql(s"insert into $tableName values(2, 'a2', '20', 2000)")
+
+ checkAnswer(s"select id, name, price, ts from $tableName")(
+ Seq(1, "a1", "10.0", 1000),
+ Seq(2, "a2", "20", 2000)
+ )
+ }
+ }
+ }
+ }
+
+ test(s"Test INT to DECIMAL schema evolution with precision overflow for COW
table") {
+ withSQLConf(
+ "hoodie.schema.on.read.enable" -> "true",
+ "spark.sql.ansi.enabled" -> "false",
+ "spark.sql.parquet.enableVectorizedReader" -> "true",
+ "hoodie.parquet.small.file.limit" -> "0"
+ ) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | price int,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | type = 'cow',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ |""".stripMargin)
+
+ spark.sql(s"insert into $tableName values (1, 12345, 1000), (2, 12,
1000)")
+
+ spark.sql(s"alter table $tableName alter column price type decimal(4,
2)")
+
+ checkAnswer(s"select id, price from $tableName")(
+ Seq(1, null),
+ Seq(2, new java.math.BigDecimal("12.00"))
+ )
+ }
+ }
+ }
+
+ test(s"Test LONG to DECIMAL schema evolution with precision overflow for COW
table") {
+ withSQLConf(
+ "hoodie.schema.on.read.enable" -> "true",
+ "spark.sql.ansi.enabled" -> "false",
+ "spark.sql.parquet.enableVectorizedReader" -> "true",
+ "hoodie.parquet.small.file.limit" -> "0"
+ ) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | price bigint,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | type = 'cow',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ |""".stripMargin)
+
+ spark.sql(s"insert into $tableName values (1, 9223372036854775807,
1000)")
+
+ spark.sql(s"alter table $tableName alter column price type decimal(20,
2)")
+
+ checkAnswer(s"select id, price from $tableName")(
+ Seq(1, null)
+ )
+ }
+ }
+ }
+
+ test(s"Test INT to DECIMAL schema evolution with precision overflow in ANSI
mode for COW table") {
+ withSQLConf(
+ "hoodie.schema.on.read.enable" -> "true",
+ "spark.sql.ansi.enabled" -> "true",
+ "spark.sql.parquet.enableVectorizedReader" -> "true",
+ "hoodie.parquet.small.file.limit" -> "0"
+ ) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | price int,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | type = 'cow',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ |""".stripMargin)
+
+ spark.sql(s"insert into $tableName values (1, 12345, 1000)")
+
+ spark.sql(s"alter table $tableName alter column price type decimal(4,
2)")
+
+ val exception = intercept[SparkException] {
+ spark.sql(s"select id, price from $tableName").collect()
}
+
+ val messages = Iterator
+ .iterate[Throwable](exception)(_.getCause)
+ .takeWhile(_ != null)
+ .flatMap(e => Option(e.getMessage))
+ .mkString("\n")
+
+ assert(
+ messages.contains("12345 cannot be represented as Decimal(4, 2)"),
+ s"Unexpected Spark error chain:\n$messages"
+ )
}
}
}
