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 d80b252e fix(parquet/pqarrow): reject out-of-range INT96 timestamps
(#1128)
d80b252e is described below
commit d80b252e47ae0aac7ac669b610e83802a8968bd4
Author: Minh Vu <[email protected]>
AuthorDate: Tue Aug 18 18:56:03 2026 +0200
fix(parquet/pqarrow): reject out-of-range INT96 timestamps (#1128)
### Rationale for this change
transferInt96 converts Parquet INT96 values through time.Time.UnixNano
without checking the int64 nanosecond range. Very wide dates can
therefore wrap when read as Arrow timestamp[ns].
### What changes are included in this PR?
Add checked INT96 to nanosecond conversion, propagate conversion errors
through the Parquet reader, and normalize Int96.ToTime without
overflowing before the range check.
### Are these changes tested?
- `go test ./parquet/pqarrow -run
'Test(ArrowTimestampToImpalaTimestamp|ReadInt96RejectsOutOfRangeTimestamp|WriteArrowInt96)$'`
### Are there any user-facing changes?
In-range INT96 behavior is unchanged. Out-of-range INT96 timestamps now
return an error instead of being wrapped.
---
parquet/pqarrow/column_readers.go | 33 +++++---
parquet/pqarrow/encode_arrow_int96_test.go | 130 +++++++++++++++++++++++++++++
parquet/types.go | 35 +++++++-
parquet/types_test.go | 125 +++++++++++++++++++++++++++
4 files changed, 309 insertions(+), 14 deletions(-)
diff --git a/parquet/pqarrow/column_readers.go
b/parquet/pqarrow/column_readers.go
index 5757e863..1ab16bd7 100644
--- a/parquet/pqarrow/column_readers.go
+++ b/parquet/pqarrow/column_readers.go
@@ -662,7 +662,10 @@ func transferColumnData(rdr file.RecordReader, valueType
arrow.DataType, descr *
dt = valueType.(arrow.ExtensionType).StorageType()
}
- var data arrow.ArrayData
+ var (
+ data arrow.ArrayData
+ err error
+ )
switch dt.ID() {
case arrow.DICTIONARY:
return transferDictionary(rdr, valueType, mem)
@@ -702,7 +705,10 @@ func transferColumnData(rdr file.RecordReader, valueType
arrow.DataType, descr *
data = transferZeroCopy(rdr, valueType)
case arrow.Nanosecond:
if descr.PhysicalType() == parquet.Types.Int96 {
- data = transferInt96(rdr, valueType)
+ data, err = transferInt96(rdr, valueType)
+ if err != nil {
+ return nil, err
+ }
} else {
data = transferZeroCopy(rdr, valueType)
}
@@ -887,28 +893,31 @@ func transferDate64(rdr file.RecordReader, dt
arrow.DataType) arrow.ArrayData {
}
// coerce int96 to nanosecond timestamp
-func transferInt96(rdr file.RecordReader, dt arrow.DataType) arrow.ArrayData {
+func transferInt96(rdr file.RecordReader, dt arrow.DataType) (arrow.ArrayData,
error) {
length := rdr.ValuesWritten()
values := parquet.Int96Traits.CastFromBytes(rdr.Values())
+ bitmap := rdr.ReleaseValidBits()
+ if bitmap != nil {
+ defer bitmap.Release()
+ }
+
data := make([]byte, arrow.Int64SizeBytes*length)
out := arrow.Int64Traits.CastFromBytes(data)
for idx, val := range values[:length] {
- if binary.LittleEndian.Uint32(val[8:]) == 0 {
- out[idx] = 0
- } else {
- out[idx] = val.ToTime().UnixNano()
+ if bitmap == nil || bitutil.BitIsSet(bitmap.Bytes(), idx) {
+ timestamp, err := val.ToTimestamp()
+ if err != nil {
+ return nil, fmt.Errorf("parquet INT96 timestamp
at index %d: %w", idx, err)
+ }
+ out[idx] = int64(timestamp)
}
}
- bitmap := rdr.ReleaseValidBits()
- if bitmap != nil {
- defer bitmap.Release()
- }
return array.NewData(dt, length, []*memory.Buffer{
bitmap, memory.NewBufferBytes(data),
- }, nil, int(rdr.NullCount()), 0)
+ }, nil, int(rdr.NullCount()), 0), nil
}
// convert physical integer storage of a decimal logical type to a decimal128
typed array
diff --git a/parquet/pqarrow/encode_arrow_int96_test.go
b/parquet/pqarrow/encode_arrow_int96_test.go
index 4e5cbf2d..7ca74d04 100644
--- a/parquet/pqarrow/encode_arrow_int96_test.go
+++ b/parquet/pqarrow/encode_arrow_int96_test.go
@@ -17,11 +17,19 @@
package pqarrow
import (
+ "bytes"
+ "context"
+ "encoding/binary"
"testing"
+ "time"
"github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/file"
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
)
func TestArrowTimestampToImpalaTimestamp(t *testing.T) {
@@ -49,3 +57,125 @@ func TestArrowTimestampToImpalaTimestamp(t *testing.T) {
})
}
}
+
+func TestReadInt96RejectsOutOfRangeTimestamp(t *testing.T) {
+ nanosPerDay := uint64(24 * time.Hour)
+ var epoch parquet.Int96
+ arrowTimestampToImpalaTimestamp(arrow.Nanosecond, 0, &epoch)
+ nanosecondsAtEndOfDay := epoch
+ binary.LittleEndian.PutUint64(nanosecondsAtEndOfDay[:8], nanosPerDay)
+ tests := []struct {
+ name string
+ corrupt parquet.Int96
+ }{
+ {
+ name: "zero value",
+ corrupt: parquet.NewInt96([3]uint32{0, 0, 0}),
+ },
+ {
+ name: "julian day out of range",
+ corrupt: parquet.NewInt96([3]uint32{0, 0, ^uint32(0)}),
+ },
+ {
+ name: "nanoseconds at end of day",
+ corrupt: nanosecondsAtEndOfDay,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ err := readCorruptInt96(t, tt.corrupt)
+ require.ErrorIs(t, err, arrow.ErrInvalid)
+ })
+ }
+}
+
+func readCorruptInt96(t *testing.T, corrupt parquet.Int96) error {
+ t.Helper()
+ mem := memory.NewGoAllocator()
+ timestampType := &arrow.TimestampType{Unit: arrow.Nanosecond}
+ sc := arrow.NewSchema([]arrow.Field{{Name: "ts", Type: timestampType}},
nil)
+ builder := array.NewTimestampBuilder(mem, timestampType)
+ builder.Append(0)
+ record := array.NewRecordBatch(sc, []arrow.Array{builder.NewArray()}, 1)
+ builder.Release()
+ defer record.Release()
+
+ var buf bytes.Buffer
+ writer, err := NewFileWriter(
+ sc,
+ &buf,
+ parquet.NewWriterProperties(
+ parquet.WithDictionaryDefault(false),
+ parquet.WithEncodingFor("ts", parquet.Encodings.Plain),
+ ),
+ NewArrowWriterProperties(WithDeprecatedInt96Timestamps(true)),
+ )
+ require.NoError(t, err)
+ require.NoError(t, writer.Write(record))
+ require.NoError(t, writer.Close())
+
+ var valid parquet.Int96
+ arrowTimestampToImpalaTimestamp(arrow.Nanosecond, 0, &valid)
+ encoded := buf.Bytes()
+ idx := bytes.Index(encoded, valid[:])
+ require.GreaterOrEqual(t, idx, 0)
+ copy(encoded[idx:idx+parquet.Int96SizeBytes], corrupt[:])
+
+ fileReader, err := file.NewParquetReader(bytes.NewReader(encoded))
+ require.NoError(t, err)
+ defer fileReader.Close()
+
+ arrowReader, err := NewFileReader(fileReader, ArrowReadProperties{},
mem)
+ require.NoError(t, err)
+ columnReader, err := arrowReader.GetColumn(context.Background(), 0)
+ require.NoError(t, err)
+ defer columnReader.Release()
+
+ _, err = columnReader.NextBatch(1)
+ return err
+}
+
+func TestReadInt96SkipsNullPhysicalValues(t *testing.T) {
+ mem := memory.NewGoAllocator()
+ timestampType := &arrow.TimestampType{Unit: arrow.Nanosecond}
+ sc := arrow.NewSchema([]arrow.Field{{Name: "ts", Type: timestampType,
Nullable: true}}, nil)
+ builder := array.NewTimestampBuilder(mem, timestampType)
+ builder.AppendNull()
+ builder.Append(0)
+ record := array.NewRecordBatch(sc, []arrow.Array{builder.NewArray()}, 2)
+ builder.Release()
+ defer record.Release()
+
+ var buf bytes.Buffer
+ writer, err := NewFileWriter(
+ sc,
+ &buf,
+ parquet.NewWriterProperties(
+ parquet.WithDictionaryDefault(false),
+ parquet.WithEncodingFor("ts", parquet.Encodings.Plain),
+ ),
+ NewArrowWriterProperties(WithDeprecatedInt96Timestamps(true)),
+ )
+ require.NoError(t, err)
+ require.NoError(t, writer.Write(record))
+ require.NoError(t, writer.Close())
+
+ fileReader, err := file.NewParquetReader(bytes.NewReader(buf.Bytes()))
+ require.NoError(t, err)
+ defer fileReader.Close()
+
+ arrowReader, err := NewFileReader(fileReader, ArrowReadProperties{},
mem)
+ require.NoError(t, err)
+ columnReader, err := arrowReader.GetColumn(context.Background(), 0)
+ require.NoError(t, err)
+ defer columnReader.Release()
+
+ chunked, err := columnReader.NextBatch(2)
+ require.NoError(t, err)
+ defer chunked.Release()
+
+ values := chunked.Chunk(0).(*array.Timestamp)
+ assert.True(t, values.IsNull(0))
+ assert.Equal(t, arrow.Timestamp(0), values.Value(1))
+}
diff --git a/parquet/types.go b/parquet/types.go
index f8c904c8..c1bd4372 100644
--- a/parquet/types.go
+++ b/parquet/types.go
@@ -18,7 +18,9 @@ package parquet
import (
"encoding/binary"
+ "fmt"
"io"
+ "math"
"reflect"
"strings"
"time"
@@ -85,11 +87,40 @@ func (i96 Int96) ToTime() time.Time {
nanos := binary.LittleEndian.Uint64(i96[:8])
jdays := binary.LittleEndian.Uint32(i96[8:])
- nanos = (uint64(jdays)-uint64(julianUnixEpoch))*uint64(nanosPerDay) +
nanos
- t := time.Unix(0, int64(nanos))
+ days := int64(jdays) - julianUnixEpoch
+ seconds := days*86400 + int64(nanos/1_000_000_000)
+ t := time.Unix(seconds, int64(nanos%1_000_000_000))
return t.UTC()
}
+// ToTimestamp converts an Int96 value to an Arrow nanosecond timestamp.
+func (i96 Int96) ToTimestamp() (arrow.Timestamp, error) {
+ nanosOfDay := binary.LittleEndian.Uint64(i96[:8])
+ if nanosOfDay >= uint64(nanosPerDay) {
+ return 0, fmt.Errorf("%w: invalid INT96 nanoseconds of day:
%d", arrow.ErrInvalid, nanosOfDay)
+ }
+
+ days := int64(binary.LittleEndian.Uint32(i96[8:])) - julianUnixEpoch
+ maxDays := math.MaxInt64 / nanosPerDay
+ if days > maxDays || (days == maxDays && int64(nanosOfDay) >
math.MaxInt64%nanosPerDay) {
+ return 0, fmt.Errorf("%w: INT96 timestamp is outside the Arrow
nanosecond range", arrow.ErrInvalid)
+ }
+
+ minDays := math.MinInt64 / nanosPerDay
+ minRemainder := math.MinInt64 % nanosPerDay
+ minNanos := nanosPerDay + minRemainder
+ // Go division truncates toward zero, so the minimum timestamp can fall
on
+ // the day before minDays with a positive nanoseconds-of-day remainder.
+ if days < minDays {
+ if days != minDays-1 || int64(nanosOfDay) < minNanos {
+ return 0, fmt.Errorf("%w: INT96 timestamp is outside
the Arrow nanosecond range", arrow.ErrInvalid)
+ }
+ return arrow.Timestamp(math.MinInt64 + int64(nanosOfDay) -
minNanos), nil
+ }
+
+ return arrow.Timestamp(days*nanosPerDay + int64(nanosOfDay)), nil
+}
+
type int96Traits struct{}
func (int96Traits) BytesRequired(n int) int { return Int96SizeBytes * n }
diff --git a/parquet/types_test.go b/parquet/types_test.go
new file mode 100644
index 00000000..6946b887
--- /dev/null
+++ b/parquet/types_test.go
@@ -0,0 +1,125 @@
+// 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 parquet
+
+import (
+ "math"
+ "testing"
+ "time"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/stretchr/testify/require"
+)
+
+func TestInt96ToTimestamp(t *testing.T) {
+ maxDays := math.MaxInt64 / nanosPerDay
+ maxNanos := math.MaxInt64 % nanosPerDay
+ minDays := math.MinInt64 / nanosPerDay
+ minNanos := nanosPerDay + math.MinInt64%nanosPerDay
+
+ tests := []struct {
+ name string
+ value Int96
+ want arrow.Timestamp
+ wantErr error
+ }{
+ {
+ name: "exact minimum",
+ value: newInt96Timestamp(minDays-1, minNanos),
+ want: arrow.Timestamp(math.MinInt64),
+ },
+ {
+ name: "below minimum",
+ value: newInt96Timestamp(minDays-1, minNanos-1),
+ wantErr: arrow.ErrInvalid,
+ },
+ {
+ name: "exact maximum",
+ value: newInt96Timestamp(maxDays, maxNanos),
+ want: arrow.Timestamp(math.MaxInt64),
+ },
+ {
+ name: "above maximum",
+ value: newInt96Timestamp(maxDays, maxNanos+1),
+ wantErr: arrow.ErrInvalid,
+ },
+ {
+ name: "last nanosecond of day",
+ value: newInt96Timestamp(0, nanosPerDay-1),
+ want: arrow.Timestamp(nanosPerDay - 1),
+ },
+ {
+ name: "nanoseconds at end of day",
+ value: newInt96Timestamp(0, nanosPerDay),
+ wantErr: arrow.ErrInvalid,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, err := tt.value.ToTimestamp()
+ if tt.wantErr != nil {
+ require.ErrorIs(t, err, tt.wantErr)
+ return
+ }
+
+ require.NoError(t, err)
+ require.Equal(t, tt.want, got)
+ })
+ }
+}
+
+func TestInt96ToTime(t *testing.T) {
+ wideDate := time.Date(3000, time.January, 1, 0, 0, 0, 0, time.UTC)
+ tests := []struct {
+ name string
+ value Int96
+ want time.Time
+ }{
+ {
+ name: "unix epoch",
+ value: newInt96Timestamp(0, 0),
+ want: time.Unix(0, 0).UTC(),
+ },
+ {
+ name: "before unix epoch",
+ value: newInt96Timestamp(-1, nanosPerDay-1),
+ want: time.Unix(0, -1).UTC(),
+ },
+ {
+ name: "wide date",
+ value: newInt96Timestamp(wideDate.Unix()/86400, 0),
+ want: wideDate,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ require.Equal(t, tt.want, tt.value.ToTime())
+ require.Equal(t, tt.want.String(), tt.value.String())
+ })
+ }
+}
+
+func newInt96Timestamp(days, nanosOfDay int64) Int96 {
+ nanos := uint64(nanosOfDay)
+ return NewInt96([3]uint32{
+ uint32(nanos),
+ uint32(nanos >> 32),
+ uint32(days + julianUnixEpoch),
+ })
+}
