You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/06/03 18:17:44 UTC
(iceberg-go) branch main updated: fix(table): normalize timestamp units for partitioned writes (#1112)
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/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new bd81e3fc fix(table): normalize timestamp units for partitioned writes (#1112)
bd81e3fc is described below
commit bd81e3fcca4747c4fd97fe3556d287844a4c6955
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Wed Jun 3 20:17:37 2026 +0200
fix(table): normalize timestamp units for partitioned writes (#1112)
## Summary
- convert Arrow timestamp partition values according to their Arrow unit
before applying Iceberg partition transforms
- use the table source type to choose microsecond vs nanosecond Iceberg
timestamp literals
- add partition fanout regression coverage for timestamp seconds,
milliseconds, microseconds, and nanoseconds
## Why
Partitioned writes compute partition keys before `ToRequestedSchema`
normalizes Arrow timestamp arrays to the table schema. The old partition
path cast raw Arrow timestamp values directly to Iceberg microsecond
timestamps, so `timestamp[s]` and `timestamp[ms]` inputs could be routed
to the wrong day/hour partition even though the data values were later
written with normalized units.
Fixes #1111.
## Testing
- `go test ./table -run TestFanoutWriter -count=1`
- `git diff --check`
---
internal/utils.go | 13 +++
table/arrow_utils.go | 62 +++++++++--
table/arrow_utils_test.go | 58 ++++++++++
table/partitioned_fanout_writer.go | 116 ++++++++++++++++----
table/partitioned_fanout_writer_test.go | 183 ++++++++++++++++++++++++++++++++
table/rolling_data_writer.go | 6 +-
table/write_records.go | 11 +-
table/write_records_test.go | 74 +++++++++++++
table/writer.go | 6 +-
transforms.go | 23 +---
types.go | 5 +-
11 files changed, 504 insertions(+), 53 deletions(-)
diff --git a/internal/utils.go b/internal/utils.go
index 06399d57..65b393fe 100644
--- a/internal/utils.go
+++ b/internal/utils.go
@@ -25,8 +25,21 @@ import (
"iter"
"slices"
"sync/atomic"
+
+ "golang.org/x/exp/constraints"
)
+// FloorDiv performs floored integer division, rounding toward negative infinity.
+// This matches Java's Math.floorDiv behavior for negative dividends.
+func FloorDiv[T constraints.Integer](a, b T) T {
+ d := a / b
+ if (a^b) < 0 && d*b != a {
+ d--
+ }
+
+ return d
+}
+
// Helper function to find the difference between two slices (a - b).
func Difference(a, b []string) []string {
m := make(map[string]bool)
diff --git a/table/arrow_utils.go b/table/arrow_utils.go
index 378f13d2..bcd70587 100644
--- a/table/arrow_utils.go
+++ b/table/arrow_utils.go
@@ -914,14 +914,16 @@ func (a *arrowProjectionVisitor) castIfNeeded(field iceberg.NestedField, vals ar
}
if !field.Type.Equals(typ) {
- promoted := retOrPanic(iceberg.PromoteType(fileField.Type, field.Type))
- targetType := retOrPanic(TypeToArrowType(promoted, a.includeFieldIDs, a.useLargeTypes))
- if !a.useLargeTypes {
- targetType = retOrPanic(ensureSmallArrowTypes(targetType))
- }
+ if !canDowncastTimestampPrecision(fileField.Type, field.Type, a.downcastNsTimestamp) {
+ promoted := retOrPanic(iceberg.PromoteType(fileField.Type, field.Type))
+ targetType := retOrPanic(TypeToArrowType(promoted, a.includeFieldIDs, a.useLargeTypes))
+ if !a.useLargeTypes {
+ targetType = retOrPanic(ensureSmallArrowTypes(targetType))
+ }
- return retOrPanic(compute.CastArray(a.ctx, vals,
- compute.SafeCastOptions(targetType)))
+ return retOrPanic(compute.CastArray(a.ctx, vals,
+ compute.SafeCastOptions(targetType)))
+ }
}
targetType := retOrPanic(TypeToArrowType(field.Type, a.includeFieldIDs, a.useLargeTypes))
@@ -933,7 +935,8 @@ func (a *arrowProjectionVisitor) castIfNeeded(field iceberg.NestedField, vals ar
if tgtok && valok && tt.TimeZone == "" && vt.TimeZone == "" && tt.Unit == arrow.Microsecond {
if vt.Unit == arrow.Nanosecond && a.downcastNsTimestamp {
- return retOrPanic(compute.CastArray(a.ctx, vals, compute.UnsafeCastOptions(tt)))
+ return floorTimestampNanosecondsToMicroseconds(
+ compute.GetAllocator(a.ctx), vals.(*array.Timestamp), tt)
} else if vt.Unit == arrow.Second || vt.Unit == arrow.Millisecond {
return retOrPanic(compute.CastArray(a.ctx, vals, compute.SafeCastOptions(tt)))
}
@@ -948,7 +951,8 @@ func (a *arrowProjectionVisitor) castIfNeeded(field iceberg.NestedField, vals ar
if tgtok && valok && tt.TimeZone == "UTC" &&
slices.Contains(utcAliases, vt.TimeZone) && tt.Unit == arrow.Microsecond {
if vt.Unit == arrow.Nanosecond && a.downcastNsTimestamp {
- return retOrPanic(compute.CastArray(a.ctx, vals, compute.UnsafeCastOptions(tt)))
+ return floorTimestampNanosecondsToMicroseconds(
+ compute.GetAllocator(a.ctx), vals.(*array.Timestamp), tt)
} else if vt.Unit != arrow.Nanosecond {
return retOrPanic(compute.CastArray(a.ctx, vals, compute.SafeCastOptions(tt)))
}
@@ -966,6 +970,46 @@ func (a *arrowProjectionVisitor) castIfNeeded(field iceberg.NestedField, vals ar
return vals
}
+func canDowncastTimestampPrecision(fileType, readType iceberg.Type, enabled bool) bool {
+ if !enabled {
+ return false
+ }
+
+ switch fileType.(type) {
+ case iceberg.TimestampNsType:
+ _, ok := readType.(iceberg.TimestampType)
+
+ return ok
+ case iceberg.TimestampTzNsType:
+ _, ok := readType.(iceberg.TimestampTzType)
+
+ return ok
+ default:
+ return false
+ }
+}
+
+func floorTimestampNanosecondsToMicroseconds(
+ mem memory.Allocator,
+ vals *array.Timestamp,
+ targetType *arrow.TimestampType,
+) arrow.Array {
+ bldr := array.NewTimestampBuilder(mem, targetType)
+ defer bldr.Release()
+
+ for i := range vals.Len() {
+ if vals.IsNull(i) {
+ bldr.AppendNull()
+
+ continue
+ }
+
+ bldr.Append(arrow.Timestamp(internal.FloorDiv(int64(vals.Value(i)), int64(1_000))))
+ }
+
+ return bldr.NewArray()
+}
+
func (a *arrowProjectionVisitor) constructField(field iceberg.NestedField, arrowType arrow.DataType) arrow.Field {
metadata := map[string]string{}
if field.Doc != "" {
diff --git a/table/arrow_utils_test.go b/table/arrow_utils_test.go
index 92b78c92..1ecc5f1f 100644
--- a/table/arrow_utils_test.go
+++ b/table/arrow_utils_test.go
@@ -636,6 +636,64 @@ func TestToRequestedSchemaTimestamps(t *testing.T) {
}
}
+func TestToRequestedSchemaTimestampDowncastFloorsNegativeNanoseconds(t *testing.T) {
+ tests := []struct {
+ name string
+ sourceArrow *arrow.TimestampType
+ sourceType iceberg.Type
+ requested iceberg.Type
+ expectedType *arrow.TimestampType
+ }{
+ {
+ name: "timestamp",
+ sourceArrow: &arrow.TimestampType{Unit: arrow.Nanosecond},
+ sourceType: iceberg.PrimitiveTypes.TimestampNs,
+ requested: iceberg.PrimitiveTypes.Timestamp,
+ expectedType: &arrow.TimestampType{Unit: arrow.Microsecond},
+ },
+ {
+ name: "timestamptz",
+ sourceArrow: &arrow.TimestampType{Unit: arrow.Nanosecond, TimeZone: "UTC"},
+ sourceType: iceberg.PrimitiveTypes.TimestampTzNs,
+ requested: iceberg.PrimitiveTypes.TimestampTz,
+ expectedType: &arrow.TimestampType{Unit: arrow.Microsecond, TimeZone: "UTC"},
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ctx := context.Background()
+ mem := memory.NewCheckedAllocator(memory.NewGoAllocator())
+ defer mem.AssertSize(t, 0)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{{Name: "ts", Type: tt.sourceArrow, Nullable: true}}, nil)
+ bldr := array.NewRecordBuilder(mem, arrowSchema)
+ defer bldr.Release()
+ bldr.Field(0).(*array.TimestampBuilder).Append(arrow.Timestamp(-1_500))
+
+ rec := bldr.NewRecordBatch()
+ defer rec.Release()
+
+ requested := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "ts", Type: tt.requested, Required: false},
+ )
+ fileSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "ts", Type: tt.sourceType, Required: false},
+ )
+
+ converted, err := table.ToRequestedSchema(ctx, requested, fileSchema, rec, table.SchemaOptions{
+ DowncastTimestamp: true,
+ })
+ require.NoError(t, err)
+ defer converted.Release()
+
+ col := converted.Column(0).(*array.Timestamp)
+ require.True(t, arrow.TypeEqual(tt.expectedType, col.DataType()))
+ assert.Equal(t, arrow.Timestamp(-2), col.Value(0))
+ })
+ }
+}
+
func TestToRequestedSchema(t *testing.T) {
mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
defer mem.AssertSize(t, 0)
diff --git a/table/partitioned_fanout_writer.go b/table/partitioned_fanout_writer.go
index edc1dd0b..1616158f 100644
--- a/table/partitioned_fanout_writer.go
+++ b/table/partitioned_fanout_writer.go
@@ -22,6 +22,7 @@ import (
"errors"
"fmt"
"iter"
+ "math"
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
@@ -29,6 +30,7 @@ import (
"github.com/apache/arrow-go/v18/arrow/decimal128"
"github.com/apache/arrow-go/v18/arrow/extensions"
"github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/internal"
"golang.org/x/sync/errgroup"
)
@@ -51,6 +53,12 @@ type partitionInfo struct {
partitionRec partitionRecord // The actual partition values for generating the path
}
+type partitionFieldInfo struct {
+ sourceField iceberg.PartitionField
+ fieldID int
+ sourceType iceberg.Type
+}
+
// NewPartitionedFanoutWriter creates a new PartitionedFanoutWriter with the specified
// partition specification, schema, record iterator, and writerFactory.
func newPartitionedFanoutWriter(partitionSpec iceberg.PartitionSpec, concurrentWriter *concurrentDataFileWriter, schema *iceberg.Schema, itr iter.Seq2[arrow.RecordBatch, error], writerFactory *writerFactory) *partitionedFanoutWriter {
@@ -214,28 +222,37 @@ func getRecordPartitions(spec iceberg.PartitionSpec, schema *iceberg.Schema, rec
partitionRec := make(partitionRecord, len(partitionFields))
partitionColumns := make([]arrow.Array, len(partitionFields))
- partitionFieldsInfo := make([]struct {
- sourceField *iceberg.PartitionField
- fieldID int
- }, len(partitionFields))
+ partitionFieldsInfo := make([]partitionFieldInfo, len(partitionFields))
for i := range partitionFields {
sourceField := spec.Field(i)
- colName, _ := schema.FindColumnName(sourceField.SourceID())
- colIdx := record.Schema().FieldIndices(colName)[0]
- partitionColumns[i] = record.Column(colIdx)
- partitionFieldsInfo[i] = struct {
- sourceField *iceberg.PartitionField
- fieldID int
- }{&sourceField, sourceField.FieldID}
+ colName, ok := schema.FindColumnName(sourceField.SourceID())
+ if !ok {
+ return nil, fmt.Errorf("failed to find source field ID %d in schema", sourceField.SourceID())
+ }
+ colIndices := record.Schema().FieldIndices(colName)
+ if len(colIndices) == 0 {
+ return nil, fmt.Errorf("failed to find source column %q in record schema", colName)
+ }
+ sourceType, ok := schema.FindTypeByID(sourceField.SourceID())
+ if !ok {
+ return nil, fmt.Errorf("failed to find type for source field ID %d in schema", sourceField.SourceID())
+ }
+ partitionColumns[i] = record.Column(colIndices[0])
+ partitionFieldsInfo[i] = partitionFieldInfo{
+ sourceField: sourceField,
+ fieldID: sourceField.FieldID,
+ sourceType: sourceType,
+ }
}
for row := range record.NumRows() {
for i := range partitionFields {
col := partitionColumns[i]
if !col.IsNull(int(row)) {
- sourceField := partitionFieldsInfo[i].sourceField
- val, err := getArrowValueAsIcebergLiteral(col, int(row))
+ fieldInfo := partitionFieldsInfo[i]
+ sourceField := fieldInfo.sourceField
+ val, err := getArrowValueAsIcebergLiteral(col, int(row), fieldInfo.sourceType)
if err != nil {
return nil, fmt.Errorf("failed to get arrow values as iceberg literal: %w", err)
}
@@ -278,11 +295,7 @@ func newPartitionMapNode() *partitionMapNode {
// getOrCreate navigates the tree and returns the partitionInfo for the given partition key,
// creating nodes along the way if they don't exist
-func (n *partitionMapNode) getOrCreate(partitionRec partitionRecord, fieldInfo []struct {
- sourceField *iceberg.PartitionField
- fieldID int
-},
-) *partitionInfo {
+func (n *partitionMapNode) getOrCreate(partitionRec partitionRecord, fieldInfo []partitionFieldInfo) *partitionInfo {
// Navigate through all but the last partition field
node := n
for _, part := range partitionRec[:len(partitionRec)-1] {
@@ -369,7 +382,7 @@ func partitionBatchByKey(ctx context.Context) partitionBatchFn {
}
}
-func getArrowValueAsIcebergLiteral(column arrow.Array, row int) (iceberg.Literal, error) {
+func getArrowValueAsIcebergLiteral(column arrow.Array, row int, sourceType iceberg.Type) (iceberg.Literal, error) {
if column.IsNull(row) {
return nil, nil
}
@@ -383,7 +396,7 @@ func getArrowValueAsIcebergLiteral(column arrow.Array, row int) (iceberg.Literal
return iceberg.NewLiteral(iceberg.Time(arr.Value(row))), nil
case *array.Timestamp:
- return iceberg.NewLiteral(iceberg.Timestamp(arr.Value(row))), nil
+ return timestampLiteralFromArrow(arr, row, sourceType)
case *array.Decimal32:
val := arr.Value(row)
dec := iceberg.Decimal{
@@ -464,3 +477,66 @@ func getArrowValueAsIcebergLiteral(column arrow.Array, row int) (iceberg.Literal
return nil, fmt.Errorf("unsupported value type: %T", val)
}
}
+
+func timestampLiteralFromArrow(arr *array.Timestamp, row int, sourceType iceberg.Type) (iceberg.Literal, error) {
+ timestampType := arr.DataType().(*arrow.TimestampType)
+ value := int64(arr.Value(row))
+
+ switch sourceType.(type) {
+ case iceberg.TimestampType, iceberg.TimestampTzType:
+ micros, err := arrowTimestampToMicros(value, timestampType.Unit)
+ if err != nil {
+ return nil, err
+ }
+
+ return iceberg.NewLiteral(iceberg.Timestamp(micros)), nil
+ case iceberg.TimestampNsType, iceberg.TimestampTzNsType:
+ nanos, err := arrowTimestampToNanos(value, timestampType.Unit)
+ if err != nil {
+ return nil, err
+ }
+
+ return iceberg.NewLiteral(iceberg.TimestampNano(nanos)), nil
+ default:
+ return nil, fmt.Errorf("cannot convert arrow timestamp to iceberg literal for source type %v", sourceType)
+ }
+}
+
+func arrowTimestampToMicros(value int64, unit arrow.TimeUnit) (int64, error) {
+ switch unit {
+ case arrow.Second:
+ return scaleTimestamp(value, 1_000_000)
+ case arrow.Millisecond:
+ return scaleTimestamp(value, 1_000)
+ case arrow.Microsecond:
+ return value, nil
+ case arrow.Nanosecond:
+ return internal.FloorDiv(value, int64(1_000)), nil
+ default:
+ return 0, fmt.Errorf("unsupported arrow timestamp unit: %s", unit)
+ }
+}
+
+func arrowTimestampToNanos(value int64, unit arrow.TimeUnit) (int64, error) {
+ switch unit {
+ case arrow.Second:
+ return scaleTimestamp(value, 1_000_000_000)
+ case arrow.Millisecond:
+ return scaleTimestamp(value, 1_000_000)
+ case arrow.Microsecond:
+ return scaleTimestamp(value, 1_000)
+ case arrow.Nanosecond:
+ return value, nil
+ default:
+ return 0, fmt.Errorf("unsupported arrow timestamp unit: %s", unit)
+ }
+}
+
+func scaleTimestamp(value, factor int64) (int64, error) {
+ if (value > 0 && value > math.MaxInt64/factor) ||
+ (value < 0 && value < math.MinInt64/factor) {
+ return 0, fmt.Errorf("arrow timestamp value %d overflows int64 when scaled by %d", value, factor)
+ }
+
+ return value * factor, nil
+}
diff --git a/table/partitioned_fanout_writer_test.go b/table/partitioned_fanout_writer_test.go
index 35c63afa..c48c45b1 100644
--- a/table/partitioned_fanout_writer_test.go
+++ b/table/partitioned_fanout_writer_test.go
@@ -20,6 +20,7 @@ package table
import (
"context"
"fmt"
+ "math"
"path/filepath"
"reflect"
"strconv"
@@ -300,6 +301,188 @@ func (s *FanoutWriterTestSuite) TestHourTransform() {
s.testTransformPartition(iceberg.HourTransform{}, "created_ts", "hour", testRecord, 3)
}
+func (s *FanoutWriterTestSuite) TestTimestampPartitionUsesArrowUnit() {
+ tests := []struct {
+ name string
+ arrowType *arrow.TimestampType
+ value arrow.Timestamp
+ expectedSource iceberg.Type
+ }{
+ {
+ name: "second",
+ arrowType: &arrow.TimestampType{Unit: arrow.Second},
+ value: arrow.Timestamp(1_700_000_000),
+ expectedSource: iceberg.PrimitiveTypes.Timestamp,
+ },
+ {
+ name: "millisecond",
+ arrowType: &arrow.TimestampType{Unit: arrow.Millisecond},
+ value: arrow.Timestamp(1_700_000_000_000),
+ expectedSource: iceberg.PrimitiveTypes.Timestamp,
+ },
+ {
+ name: "microsecond",
+ arrowType: &arrow.TimestampType{Unit: arrow.Microsecond},
+ value: arrow.Timestamp(1_700_000_000_000_000),
+ expectedSource: iceberg.PrimitiveTypes.Timestamp,
+ },
+ {
+ name: "millisecond UTC",
+ arrowType: &arrow.TimestampType{Unit: arrow.Millisecond, TimeZone: "UTC"},
+ value: arrow.Timestamp(1_700_000_000_000),
+ expectedSource: iceberg.PrimitiveTypes.TimestampTz,
+ },
+ }
+
+ for _, tt := range tests {
+ s.Run(tt.name, func() {
+ arrSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+ {Name: "created_ts", Type: tt.arrowType, Nullable: true},
+ }, nil)
+
+ testRecord := s.createCustomTestRecord(arrSchema, [][]any{
+ {int32(1), tt.value},
+ })
+ defer testRecord.Release()
+
+ icebergSchema, err := ArrowSchemaToIcebergWithFreshIDs(testRecord.Schema(), false)
+ s.Require().NoError(err)
+
+ sourceField, ok := icebergSchema.FindFieldByName("created_ts")
+ s.Require().True(ok)
+ s.True(tt.expectedSource.Equals(sourceField.Type))
+
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{sourceField.ID},
+ FieldID: 1000,
+ Transform: iceberg.DayTransform{},
+ Name: "created_day",
+ })
+
+ partitions, err := getRecordPartitions(spec, icebergSchema, testRecord)
+ s.Require().NoError(err)
+ s.Require().Len(partitions, 1)
+
+ partitionPath := spec.PartitionToPath(partitions[0].partitionRec, icebergSchema)
+ s.Equal("created_day=2023-11-14", partitionPath)
+ })
+ }
+}
+
+func (s *FanoutWriterTestSuite) TestTimestampNsPartitionUsesArrowUnit() {
+ tests := []struct {
+ name string
+ arrowType *arrow.TimestampType
+ expectedSource iceberg.Type
+ }{
+ {
+ name: "nanosecond",
+ arrowType: &arrow.TimestampType{Unit: arrow.Nanosecond},
+ expectedSource: iceberg.PrimitiveTypes.TimestampNs,
+ },
+ {
+ name: "nanosecond UTC",
+ arrowType: &arrow.TimestampType{Unit: arrow.Nanosecond, TimeZone: "UTC"},
+ expectedSource: iceberg.PrimitiveTypes.TimestampTzNs,
+ },
+ }
+
+ for _, tt := range tests {
+ s.Run(tt.name, func() {
+ arrSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+ {Name: "created_ts", Type: tt.arrowType, Nullable: true},
+ }, nil)
+
+ testRecord := s.createCustomTestRecord(arrSchema, [][]any{
+ {int32(1), arrow.Timestamp(1_700_000_000_000_000_000)},
+ })
+ defer testRecord.Release()
+
+ icebergSchema, err := ArrowSchemaToIcebergWithFreshIDs(testRecord.Schema(), false)
+ s.Require().NoError(err)
+
+ sourceField, ok := icebergSchema.FindFieldByName("created_ts")
+ s.Require().True(ok)
+ s.True(tt.expectedSource.Equals(sourceField.Type))
+
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{sourceField.ID},
+ FieldID: 1000,
+ Transform: iceberg.DayTransform{},
+ Name: "created_day",
+ })
+
+ partitions, err := getRecordPartitions(spec, icebergSchema, testRecord)
+ s.Require().NoError(err)
+ s.Require().Len(partitions, 1)
+
+ partitionPath := spec.PartitionToPath(partitions[0].partitionRec, icebergSchema)
+ s.Equal("created_day=2023-11-14", partitionPath)
+ })
+ }
+}
+
+func (s *FanoutWriterTestSuite) TestTimestampPartitionFloorsNegativeNanoseconds() {
+ arrSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+ {Name: "created_ts", Type: &arrow.TimestampType{Unit: arrow.Nanosecond}, Nullable: true},
+ }, nil)
+
+ testRecord := s.createCustomTestRecord(arrSchema, [][]any{
+ {int32(1), arrow.Timestamp(-1_500)},
+ })
+ defer testRecord.Release()
+
+ icebergSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32},
+ iceberg.NestedField{ID: 2, Name: "created_ts", Type: iceberg.PrimitiveTypes.Timestamp},
+ )
+
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{2},
+ FieldID: 1000,
+ Transform: iceberg.IdentityTransform{},
+ Name: "created_ts",
+ })
+
+ partitions, err := getRecordPartitions(spec, icebergSchema, testRecord)
+ s.Require().NoError(err)
+ s.Require().Len(partitions, 1)
+ s.Equal(iceberg.Timestamp(-2), partitions[0].partitionRec.Get(0))
+
+ partitionPath := spec.PartitionToPath(partitions[0].partitionRec, icebergSchema)
+ s.Equal("created_ts=1969-12-31T23%3A59%3A59.999998", partitionPath)
+}
+
+func (s *FanoutWriterTestSuite) TestTimestampPartitionRejectsOverflowWhenScalingArrowUnit() {
+ arrSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+ {Name: "created_ts", Type: &arrow.TimestampType{Unit: arrow.Second}, Nullable: true},
+ }, nil)
+
+ testRecord := s.createCustomTestRecord(arrSchema, [][]any{
+ {int32(1), arrow.Timestamp(math.MaxInt64)},
+ })
+ defer testRecord.Release()
+
+ icebergSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32},
+ iceberg.NestedField{ID: 2, Name: "created_ts", Type: iceberg.PrimitiveTypes.Timestamp},
+ )
+
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{2},
+ FieldID: 1000,
+ Transform: iceberg.DayTransform{},
+ Name: "created_day",
+ })
+
+ _, err := getRecordPartitions(spec, icebergSchema, testRecord)
+ s.Require().ErrorContains(err, "overflows int64")
+}
+
func (s *FanoutWriterTestSuite) TestVoidTransform() {
arrSchema := arrow.NewSchema([]arrow.Field{
{Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
diff --git a/table/rolling_data_writer.go b/table/rolling_data_writer.go
index 11520a2b..efc2f261 100644
--- a/table/rolling_data_writer.go
+++ b/table/rolling_data_writer.go
@@ -343,7 +343,11 @@ func (r *RollingDataWriter) stream(outputDataFilesCh chan<- iceberg.DataFile) {
}
converted, err := ToRequestedSchema(r.ctx, r.factory.fileSchema,
- r.factory.taskSchema, record, SchemaOptions{IncludeFieldIDs: true, UseWriteDefault: true})
+ r.factory.taskSchema, record, SchemaOptions{
+ DowncastTimestamp: true,
+ IncludeFieldIDs: true,
+ UseWriteDefault: true,
+ })
if err != nil {
record.Release()
r.sendError(err)
diff --git a/table/write_records.go b/table/write_records.go
index 9c215fbf..98fd3848 100644
--- a/table/write_records.go
+++ b/table/write_records.go
@@ -122,7 +122,8 @@ func WithPreserveRowLineage(schema *iceberg.Schema) WriteRecordOption {
// field ID (or by name via the table's name mapping if field IDs are absent).
// The Arrow schema may be a subset of the table schema (projection), but every
// field present must have a type that is promotable to the corresponding table
-// field type.
+// field type. When the table uses microsecond timestamps, Arrow nanosecond
+// timestamps are also accepted and are downcast during the write path.
//
// WriteRecords releases each RecordBatch it consumes. If the caller needs a
// batch to remain valid after it has been yielded, it must call Retain before
@@ -161,7 +162,13 @@ func WriteRecords(ctx context.Context, tbl *Table,
if cfg.fileSchema != nil {
checkSchema = cfg.fileSchema
}
- if err := checkArrowSchemaCompat(checkSchema, schema, false); err != nil {
+ err := checkArrowSchemaCompat(checkSchema, schema, false)
+ if err != nil {
+ if downcastErr := checkArrowSchemaCompat(checkSchema, schema, true); downcastErr == nil {
+ err = nil
+ }
+ }
+ if err != nil {
return internal.SingleErrorIter[iceberg.DataFile](
fmt.Errorf("arrow schema is not compatible with the table schema: %w", err))
}
diff --git a/table/write_records_test.go b/table/write_records_test.go
index ac3c61cf..c125967d 100644
--- a/table/write_records_test.go
+++ b/table/write_records_test.go
@@ -27,6 +27,8 @@ import (
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/compute"
"github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet/file"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table"
@@ -253,6 +255,78 @@ func (s *WriteRecordsTestSuite) TestEmptyInput() {
s.Equal(0, count)
}
+func (s *WriteRecordsTestSuite) TestTimestampNanosecondsShouldFloorNegativeValuesIfTableUsesMicroseconds() {
+ loc := filepath.ToSlash(s.T().TempDir())
+ iceSch := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ iceberg.NestedField{ID: 2, Name: "created_ts", Type: iceberg.PrimitiveTypes.Timestamp, Required: false},
+ )
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{2},
+ FieldID: 1000,
+ Transform: iceberg.IdentityTransform{},
+ Name: "created_ts",
+ })
+ meta, err := table.NewMetadata(iceSch, &spec, table.UnsortedSortOrder, loc, iceberg.Properties{})
+ s.Require().NoError(err)
+
+ tbl := table.New(
+ table.Identifier{"test", "write_records_timestamps"},
+ meta,
+ filepath.Join(loc, "metadata", "v1.metadata.json"),
+ func(ctx context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil },
+ nil,
+ )
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+ {Name: "created_ts", Type: &arrow.TimestampType{Unit: arrow.Nanosecond}, Nullable: true},
+ }, nil)
+
+ records := func(yield func(arrow.RecordBatch, error) bool) {
+ bldr := array.NewRecordBuilder(s.mem, arrowSchema)
+ defer bldr.Release()
+
+ bldr.Field(0).(*array.Int32Builder).Append(1)
+ bldr.Field(1).(*array.TimestampBuilder).Append(arrow.Timestamp(-1_500))
+
+ yield(bldr.NewRecordBatch(), nil)
+ }
+
+ var dataFiles []iceberg.DataFile
+ for df, err := range table.WriteRecords(s.ctx, tbl, arrowSchema, records) {
+ s.Require().NoError(err)
+ dataFiles = append(dataFiles, df)
+ }
+
+ s.Require().Len(dataFiles, 1)
+
+ partitionRec := table.GetPartitionRecord(dataFiles[0], spec.PartitionType(iceSch))
+ s.Equal("created_ts=1969-12-31T23%3A59%3A59.999998", spec.PartitionToPath(partitionRec, iceSch))
+
+ rdr, err := file.OpenParquetFile(dataFiles[0].FilePath(), false)
+ s.Require().NoError(err)
+ defer rdr.Close()
+
+ arrowRdr, err := pqarrow.NewFileReader(rdr, pqarrow.ArrowReadProperties{}, memory.DefaultAllocator)
+ s.Require().NoError(err)
+
+ written, err := arrowRdr.ReadTable(s.ctx)
+ s.Require().NoError(err)
+ defer written.Release()
+
+ s.Equal(int64(1), written.NumRows())
+ chunks := written.Column(1).Data().Chunks()
+ s.Require().Len(chunks, 1)
+
+ timestamps := chunks[0].(*array.Timestamp)
+ s.Equal(arrow.Timestamp(-2), timestamps.Value(0))
+
+ writtenType := timestamps.DataType().(*arrow.TimestampType)
+ s.Equal(arrow.Microsecond, writtenType.Unit)
+ s.Empty(writtenType.TimeZone)
+}
+
type readOnlyFS struct{}
func (readOnlyFS) Open(name string) (iceio.File, error) {
diff --git a/table/writer.go b/table/writer.go
index 5e616f90..816e1eac 100644
--- a/table/writer.go
+++ b/table/writer.go
@@ -125,7 +125,11 @@ func (w *defaultDataFileWriter) writeFile(ctx context.Context, partitionValues m
batches := make([]arrow.RecordBatch, len(task.Batches))
for i, b := range task.Batches {
rec, err := ToRequestedSchema(ctx, w.fileSchema,
- task.Schema, b, SchemaOptions{IncludeFieldIDs: true, UseWriteDefault: true})
+ task.Schema, b, SchemaOptions{
+ DowncastTimestamp: true,
+ IncludeFieldIDs: true,
+ UseWriteDefault: true,
+ })
if err != nil {
return nil, err
}
diff --git a/transforms.go b/transforms.go
index 1f9fa73d..c40cf790 100644
--- a/transforms.go
+++ b/transforms.go
@@ -30,9 +30,9 @@ import (
"unsafe"
"github.com/apache/arrow-go/v18/arrow/decimal128"
+ "github.com/apache/iceberg-go/internal"
"github.com/google/uuid"
"github.com/twmb/murmur3"
- "golang.org/x/exp/constraints"
)
// ParseTransform takes the string representation of a transform as
@@ -572,19 +572,6 @@ type TimeTransform interface {
Transformer(Type) (func(any) Optional[int32], error)
}
-// floorDiv performs floored integer division, rounding toward negative infinity.
-// Unlike Go's truncated division (which rounds toward zero), this correctly
-// handles negative dividends — e.g., floorDiv(-1, 3600000000) = -1, not 0.
-// This matches the behavior of Java's Math.floorDiv.
-func floorDiv[T constraints.Integer](a, b T) T {
- d := a / b
- if (a^b) < 0 && d*b != a {
- d--
- }
-
- return d
-}
-
func canTransformTime(t TimeTransform, sourceType Type) bool {
switch sourceType.(type) {
case DateType, TimestampType, TimestampTzType, TimestampNsType, TimestampTzNsType:
@@ -919,7 +906,7 @@ func (HourTransform) Transformer(src Type) (func(any) Optional[int32], error) {
return Optional[int32]{
Valid: true,
- Val: int32(floorDiv(int64(v.(Timestamp)), factor)),
+ Val: int32(internal.FloorDiv(int64(v.(Timestamp)), factor)),
}
}, nil
case TimestampNsType, TimestampTzNsType:
@@ -932,7 +919,7 @@ func (HourTransform) Transformer(src Type) (func(any) Optional[int32], error) {
return Optional[int32]{
Valid: true,
- Val: int32(floorDiv(int64(v.(TimestampNano)), factor)),
+ Val: int32(internal.FloorDiv(int64(v.(TimestampNano)), factor)),
}
}, nil
}
@@ -949,10 +936,10 @@ func (HourTransform) Apply(value Optional[Literal]) (out Optional[Literal]) {
switch v := value.Val.(type) {
case TimestampLiteral:
const factor = int64(time.Hour / time.Microsecond)
- out.Valid, out.Val = true, Int32Literal(int32(floorDiv(int64(v), factor)))
+ out.Valid, out.Val = true, Int32Literal(int32(internal.FloorDiv(int64(v), factor)))
case TimestampNsLiteral:
const factor = int64(time.Hour)
- out.Valid, out.Val = true, Int32Literal(int32(floorDiv(int64(v), factor)))
+ out.Valid, out.Val = true, Int32Literal(int32(internal.FloorDiv(int64(v), factor)))
}
return out
diff --git a/types.go b/types.go
index c87f2c1f..3be85017 100644
--- a/types.go
+++ b/types.go
@@ -27,6 +27,7 @@ import (
"time"
"github.com/apache/arrow-go/v18/arrow/decimal"
+ "github.com/apache/iceberg-go/internal"
"github.com/geoarrow/geoarrow-go"
)
@@ -697,7 +698,7 @@ func (t Timestamp) ToTime() time.Time {
func (t Timestamp) ToDate() Date {
const microsecondsPerDay = int64((time.Hour * 24) / time.Microsecond)
- return Date(int32(floorDiv(int64(t), microsecondsPerDay)))
+ return Date(int32(internal.FloorDiv(int64(t), microsecondsPerDay)))
}
func (t Timestamp) ToNanos() TimestampNano {
@@ -717,7 +718,7 @@ func (t TimestampNano) ToMicros() Timestamp {
func (t TimestampNano) ToDate() Date {
const nanosecondsPerDay = int64(time.Hour * 24)
- return Date(int32(floorDiv(int64(t), nanosecondsPerDay)))
+ return Date(int32(internal.FloorDiv(int64(t), nanosecondsPerDay)))
}
// TimestampType represents a number of microseconds since the unix epoch