This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new b3dacd2a fix(parquet/pqarrow): reject decimal overflow when writing
integers (#1161)
b3dacd2a is described below
commit b3dacd2a2ce0bf709a29a63a233f1626b5eee817
Author: Minh Vu <[email protected]>
AuthorDate: Fri Aug 21 21:15:40 2026 +0200
fix(parquet/pqarrow): reject decimal overflow when writing integers (#1161)
### Rationale for this change
When decimals are stored as Parquet integers, values outside the
declared or physical precision can be narrowed to int32 or int64 without
a runtime error. That can silently write a different value.
### What changes are included in this PR?
Validate non-null Decimal128 and Decimal256 values against both their
declared precision and the target integer precision before conversion,
and return arrow.ErrInvalid when a value does not fit.
### Are these changes tested?
Yes. The test covers Decimal128 and Decimal256 with both int32 and int64
Parquet storage. The full parquet/pqarrow package suite passes.
### Are there any user-facing changes?
Invalid decimal arrays now return an error instead of producing
corrupted integer values.
---
parquet/pqarrow/encode_arrow.go | 95 +++++++++++++----
parquet/pqarrow/encode_arrow_test.go | 198 +++++++++++++++++++++++++++++++++++
2 files changed, 275 insertions(+), 18 deletions(-)
diff --git a/parquet/pqarrow/encode_arrow.go b/parquet/pqarrow/encode_arrow.go
index b71452f9..8b49fb95 100644
--- a/parquet/pqarrow/encode_arrow.go
+++ b/parquet/pqarrow/encode_arrow.go
@@ -121,11 +121,58 @@ func targetDecimalPrecision(cw file.ColumnChunkWriter)
(int32, bool) {
}
func decimal128FitsTargetPrecision(val decimal128.Num, precision int32) bool {
- return precision > 0 && precision <= decimal128.MaxPrecision &&
val.FitsInPrecision(precision)
+ if precision <= 0 || precision > decimal128.MaxPrecision {
+ return false
+ }
+ // Abs overflows for the minimum two's-complement value.
+ if val.Sign() < 0 && val.Negate() == val {
+ return false
+ }
+ return val.FitsInPrecision(precision)
}
func decimal256FitsTargetPrecision(val decimal256.Num, precision int32) bool {
- return precision > 0 && precision <= decimal256.MaxPrecision &&
val.FitsInPrecision(precision)
+ if precision <= 0 || precision > decimal256.MaxPrecision {
+ return false
+ }
+ // Abs overflows for the minimum two's-complement value.
+ if val.Sign() < 0 && val.Negate() == val {
+ return false
+ }
+ return val.FitsInPrecision(precision)
+}
+
+// validatePresentDecimalValues visits only values represented by a definition
+// level. Values below the repeated ancestor definition level are list
+// placeholders and do not have a corresponding entry in arr.
+func validatePresentDecimalValues(arr arrow.Array, defLevels []int16,
levelInfo file.LevelInfo, validate func(int) error) error {
+ if len(defLevels) == 0 {
+ for idx := 0; idx < arr.Len(); idx++ {
+ if arr.IsValid(idx) {
+ if err := validate(idx); err != nil {
+ return err
+ }
+ }
+ }
+ return nil
+ }
+
+ valueIdx := 0
+ for _, defLevel := range defLevels {
+ if defLevel < levelInfo.RepeatedAncestorDefLevel {
+ continue
+ }
+ if valueIdx >= arr.Len() {
+ break
+ }
+ if defLevel == levelInfo.DefLevel && arr.IsValid(valueIdx) {
+ if err := validate(valueIdx); err != nil {
+ return err
+ }
+ }
+ valueIdx++
+ }
+ return nil
}
// arrowColumnWriter is a convenience object for easily writing arrow data to
a specific
@@ -414,31 +461,37 @@ func writeDenseArrow(ctx *arrowWriteContext, cw
file.ColumnChunkWriter, leafArr
case arrow.DECIMAL128:
arr := leafArr.(*array.Decimal128)
precision, hasPrecision :=
targetDecimalPrecision(cw)
- for idx, val := range arr.Values() {
- if arr.IsNull(idx) {
- continue
- }
+ if err := validatePresentDecimalValues(arr,
defLevels, cw.LevelInfo(), func(idx int) error {
+ val := arr.Value(idx)
if !decimal128FitsInt32(val) {
return fmt.Errorf("%w:
Decimal128 value at index %d does not fit in Parquet INT32", arrow.ErrInvalid,
idx)
}
if hasPrecision &&
!decimal128FitsTargetPrecision(val, precision) {
return fmt.Errorf("%w:
Decimal128 value at index %d does not fit Parquet DECIMAL precision %d",
arrow.ErrInvalid, idx, precision)
}
+ return nil
+ }); err != nil {
+ return err
+ }
+ for idx, val := range arr.Values() {
data[idx] = int32(val.LowBits())
}
case arrow.DECIMAL256:
arr := leafArr.(*array.Decimal256)
precision, hasPrecision :=
targetDecimalPrecision(cw)
- for idx, val := range arr.Values() {
- if arr.IsNull(idx) {
- continue
- }
+ if err := validatePresentDecimalValues(arr,
defLevels, cw.LevelInfo(), func(idx int) error {
+ val := arr.Value(idx)
if !decimal256FitsInt32(val) {
return fmt.Errorf("%w:
Decimal256 value at index %d does not fit in Parquet INT32", arrow.ErrInvalid,
idx)
}
if hasPrecision &&
!decimal256FitsTargetPrecision(val, precision) {
return fmt.Errorf("%w:
Decimal256 value at index %d does not fit Parquet DECIMAL precision %d",
arrow.ErrInvalid, idx, precision)
}
+ return nil
+ }); err != nil {
+ return err
+ }
+ for idx, val := range arr.Values() {
data[idx] = int32(val.LowBits())
}
default:
@@ -515,16 +568,19 @@ func writeDenseArrow(ctx *arrowWriteContext, cw
file.ColumnChunkWriter, leafArr
data =
arrow.Int64Traits.CastFromBytes(ctx.dataBuffer.Bytes())
arr := leafArr.(*array.Decimal128)
precision, hasPrecision := targetDecimalPrecision(cw)
- for idx, val := range arr.Values() {
- if arr.IsNull(idx) {
- continue
- }
+ if err := validatePresentDecimalValues(arr, defLevels,
cw.LevelInfo(), func(idx int) error {
+ val := arr.Value(idx)
if !decimal128FitsInt64(val) {
return fmt.Errorf("%w: Decimal128 value
at index %d does not fit in Parquet INT64", arrow.ErrInvalid, idx)
}
if hasPrecision &&
!decimal128FitsTargetPrecision(val, precision) {
return fmt.Errorf("%w: Decimal128 value
at index %d does not fit Parquet DECIMAL precision %d", arrow.ErrInvalid, idx,
precision)
}
+ return nil
+ }); err != nil {
+ return err
+ }
+ for idx, val := range arr.Values() {
data[idx] = int64(val.LowBits())
}
case arrow.DECIMAL256:
@@ -532,16 +588,19 @@ func writeDenseArrow(ctx *arrowWriteContext, cw
file.ColumnChunkWriter, leafArr
data =
arrow.Int64Traits.CastFromBytes(ctx.dataBuffer.Bytes())
arr := leafArr.(*array.Decimal256)
precision, hasPrecision := targetDecimalPrecision(cw)
- for idx, val := range arr.Values() {
- if arr.IsNull(idx) {
- continue
- }
+ if err := validatePresentDecimalValues(arr, defLevels,
cw.LevelInfo(), func(idx int) error {
+ val := arr.Value(idx)
if !decimal256FitsInt64(val) {
return fmt.Errorf("%w: Decimal256 value
at index %d does not fit in Parquet INT64", arrow.ErrInvalid, idx)
}
if hasPrecision &&
!decimal256FitsTargetPrecision(val, precision) {
return fmt.Errorf("%w: Decimal256 value
at index %d does not fit Parquet DECIMAL precision %d", arrow.ErrInvalid, idx,
precision)
}
+ return nil
+ }); err != nil {
+ return err
+ }
+ for idx, val := range arr.Values() {
data[idx] = int64(val.LowBits())
}
default:
diff --git a/parquet/pqarrow/encode_arrow_test.go
b/parquet/pqarrow/encode_arrow_test.go
index 7a4fdd01..7f49aeff 100644
--- a/parquet/pqarrow/encode_arrow_test.go
+++ b/parquet/pqarrow/encode_arrow_test.go
@@ -309,6 +309,204 @@ func TestWriteArrowCols(t *testing.T) {
}
}
+func TestWriteDecimalRejectsValuesOutsidePrecision(t *testing.T) {
+ tests := []struct {
+ name string
+ dtype arrow.DecimalType
+ value string
+ valuePrec int32
+ }{
+ {"decimal128_int32", &arrow.Decimal128Type{Precision: 9},
"3000000000", 10},
+ {"decimal128_int64", &arrow.Decimal128Type{Precision: 18},
"10000000000000000000", 20},
+ {"decimal256_int32", &arrow.Decimal256Type{Precision: 9},
"3000000000", 10},
+ {"decimal256_int64", &arrow.Decimal256Type{Precision: 18},
"10000000000000000000", 20},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ var values arrow.Array
+ switch dtype := tt.dtype.(type) {
+ case *arrow.Decimal128Type:
+ value, err := decimal128.FromString(tt.value,
tt.valuePrec, 0)
+ require.NoError(t, err)
+ builder := array.NewDecimal128Builder(mem,
dtype)
+ builder.Append(value)
+ values = builder.NewDecimal128Array()
+ builder.Release()
+ case *arrow.Decimal256Type:
+ value, err := decimal256.FromString(tt.value,
tt.valuePrec, 0)
+ require.NoError(t, err)
+ builder := array.NewDecimal256Builder(mem,
dtype)
+ builder.Append(value)
+ values = builder.NewDecimal256Array()
+ builder.Release()
+ }
+ defer values.Release()
+
+ sc := arrow.NewSchema([]arrow.Field{{Name: "decimal",
Type: tt.dtype}}, nil)
+ rec := array.NewRecordBatch(sc, []arrow.Array{values},
int64(values.Len()))
+ defer rec.Release()
+
+ var sink bytes.Buffer
+ writer, err := pqarrow.NewFileWriter(sc, &sink,
+
parquet.NewWriterProperties(parquet.WithStoreDecimalAsInteger(true)),
+
pqarrow.NewArrowWriterProperties(pqarrow.WithAllocator(mem)))
+ require.NoError(t, err)
+ err = writer.Write(rec)
+ require.ErrorIs(t, err, arrow.ErrInvalid)
+ assert.ErrorContains(t, err, "does not fit")
+ require.NoError(t, writer.Close())
+ })
+ }
+}
+
+func TestWriteDecimalRejectsTwoComplementMinimum(t *testing.T) {
+ tests := []struct {
+ name string
+ dtype arrow.DecimalType
+ value any
+ }{
+ {"decimal128_int32", &arrow.Decimal128Type{Precision: 9},
decimal128.New(-1<<63, 0)},
+ {"decimal128_int64", &arrow.Decimal128Type{Precision: 18},
decimal128.New(-1<<63, 0)},
+ {"decimal256_int32", &arrow.Decimal256Type{Precision: 9},
decimal256.New(1<<63, 0, 0, 0)},
+ {"decimal256_int64", &arrow.Decimal256Type{Precision: 18},
decimal256.New(1<<63, 0, 0, 0)},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ var values arrow.Array
+ switch dtype := tt.dtype.(type) {
+ case *arrow.Decimal128Type:
+ builder := array.NewDecimal128Builder(mem,
dtype)
+ builder.Append(tt.value.(decimal128.Num))
+ values = builder.NewDecimal128Array()
+ builder.Release()
+ case *arrow.Decimal256Type:
+ builder := array.NewDecimal256Builder(mem,
dtype)
+ builder.Append(tt.value.(decimal256.Num))
+ values = builder.NewDecimal256Array()
+ builder.Release()
+ }
+ defer values.Release()
+
+ sc := arrow.NewSchema([]arrow.Field{{Name: "decimal",
Type: tt.dtype}}, nil)
+ rec := array.NewRecordBatch(sc, []arrow.Array{values},
int64(values.Len()))
+ defer rec.Release()
+
+ var sink bytes.Buffer
+ writer, err := pqarrow.NewFileWriter(sc, &sink,
+
parquet.NewWriterProperties(parquet.WithStoreDecimalAsInteger(true)),
+
pqarrow.NewArrowWriterProperties(pqarrow.WithAllocator(mem)))
+ require.NoError(t, err)
+ err = writer.Write(rec)
+ require.ErrorIs(t, err, arrow.ErrInvalid)
+ require.NoError(t, writer.Close())
+ })
+ }
+}
+
+func TestWriteDecimalIgnoresValuesUnderNullParent(t *testing.T) {
+ tests := []struct {
+ name string
+ dtype arrow.DecimalType
+ value string
+ valuePrec int32
+ }{
+ {"decimal128_int32", &arrow.Decimal128Type{Precision: 9},
"3000000000", 10},
+ {"decimal128_int64", &arrow.Decimal128Type{Precision: 18},
"10000000000000000000", 20},
+ {"decimal256_int32", &arrow.Decimal256Type{Precision: 9},
"3000000000", 10},
+ {"decimal256_int64", &arrow.Decimal256Type{Precision: 18},
"10000000000000000000", 20},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ var values arrow.Array
+ switch dtype := tt.dtype.(type) {
+ case *arrow.Decimal128Type:
+ value, err := decimal128.FromString(tt.value,
tt.valuePrec, 0)
+ require.NoError(t, err)
+ builder := array.NewDecimal128Builder(mem,
dtype)
+ builder.Append(value)
+ builder.Append(decimal128.FromI64(1))
+ values = builder.NewDecimal128Array()
+ builder.Release()
+ case *arrow.Decimal256Type:
+ value, err := decimal256.FromString(tt.value,
tt.valuePrec, 0)
+ require.NoError(t, err)
+ builder := array.NewDecimal256Builder(mem,
dtype)
+ builder.Append(value)
+ builder.Append(decimal256.FromI64(1))
+ values = builder.NewDecimal256Array()
+ builder.Release()
+ }
+ defer values.Release()
+
+ childField := arrow.Field{Name: "decimal", Type:
tt.dtype}
+ structField := arrow.Field{Name: "parent", Type:
arrow.StructOf(childField), Nullable: true}
+ validity := memory.NewBufferBytes([]byte{0x02})
+ defer validity.Release()
+ structData := array.NewData(structField.Type, 2,
[]*memory.Buffer{validity}, []arrow.ArrayData{values.Data()}, 1, 0)
+ defer structData.Release()
+ structArray := array.NewStructData(structData)
+ defer structArray.Release()
+
+ sc := arrow.NewSchema([]arrow.Field{structField}, nil)
+ rec := array.NewRecordBatch(sc,
[]arrow.Array{structArray}, 2)
+ defer rec.Release()
+
+ var sink bytes.Buffer
+ writer, err := pqarrow.NewFileWriter(sc, &sink,
+
parquet.NewWriterProperties(parquet.WithStoreDecimalAsInteger(true)),
+
pqarrow.NewArrowWriterProperties(pqarrow.WithAllocator(mem)))
+ require.NoError(t, err)
+ require.NoError(t, writer.Write(rec))
+ require.NoError(t, writer.Close())
+ })
+ }
+}
+
+func TestWriteDecimalRejectsOverflowAfterEmptyList(t *testing.T) {
+ mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ dtype := &arrow.Decimal128Type{Precision: 4, Scale: 0}
+ listBuilder := array.NewListBuilder(mem, dtype)
+ valueBuilder := listBuilder.ValueBuilder().(*array.Decimal128Builder)
+
+ listBuilder.Append(true)
+ listBuilder.Append(true)
+ valueBuilder.Append(decimal128.FromI64(99999))
+ listBuilder.Append(true)
+ valueBuilder.Append(decimal128.FromI64(1))
+ lists := listBuilder.NewListArray()
+ listBuilder.Release()
+ defer lists.Release()
+
+ sc := arrow.NewSchema([]arrow.Field{{Name: "decimal", Type:
lists.DataType()}}, nil)
+ rec := array.NewRecordBatch(sc, []arrow.Array{lists},
int64(lists.Len()))
+ defer rec.Release()
+
+ var sink bytes.Buffer
+ writer, err := pqarrow.NewFileWriter(sc, &sink,
+
parquet.NewWriterProperties(parquet.WithStoreDecimalAsInteger(true)),
+ pqarrow.NewArrowWriterProperties(pqarrow.WithAllocator(mem)))
+ require.NoError(t, err)
+
+ err = writer.Write(rec)
+ require.ErrorIs(t, err, arrow.ErrInvalid)
+ assert.ErrorContains(t, err, "does not fit")
+ require.NoError(t, writer.Close())
+}
+
func TestWriteArrowInt96(t *testing.T) {
mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
defer mem.AssertSize(t, 0)
