From 7423d0f518c2caf4a85fb8e3b33975eb4b0d51e2 Mon Sep 17 00:00:00 2001 From: Matt Topol Date: Fri, 19 Jun 2026 17:35:46 -0400 Subject: [PATCH 1/2] fix(arrow/array): reject nulls in non-nullable fields when building records RecordBuilder.NewRecordBatch (and the deprecated NewRecord) now reject a record whose top-level field is declared non-nullable but whose column contains nulls, matching the Rust and C++ implementations. The new RecordBuilder.NewRecordBatchChecked returns the violation as an error instead of panicking. The check is scoped to top-level record columns only: a non-nullable child of a nullable struct is unaffected, so existing nested data keeps working. The CSV, JSON, and Avro readers route through NewRecordBatchChecked and surface a read error rather than building an invalid record. Closes #372 --- arrow/array/json_reader.go | 24 ++++-- arrow/array/record.go | 52 ++++++++++++- arrow/array/record_test.go | 101 +++++++++++++++++++++++++ arrow/array/table_test.go | 6 +- arrow/avro/loader.go | 22 +++++- arrow/compute/vector_sort_test.go | 14 ++-- arrow/csv/reader.go | 23 ++++-- arrow/csv/reader_test.go | 50 ++++++------ arrow/csv/writer_test.go | 54 ++++++------- arrow/example_test.go | 6 +- arrow/extensions/uuid_test.go | 2 +- arrow/internal/arrdata/arrdata.go | 4 +- arrow/internal/arrjson/arrjson_test.go | 4 +- arrow/ipc/cmd/arrow-ls/main_test.go | 4 +- arrow/ipc/ipc_test.go | 2 +- arrow/ipc/reader_test.go | 2 +- parquet/pqarrow/encode_arrow_test.go | 28 +++---- 17 files changed, 294 insertions(+), 104 deletions(-) diff --git a/arrow/array/json_reader.go b/arrow/array/json_reader.go index 6aea688dc..e3f3736bd 100644 --- a/arrow/array/json_reader.go +++ b/arrow/array/json_reader.go @@ -179,11 +179,26 @@ func (r *JSONReader) readNext() bool { return true } +func (r *JSONReader) build() bool { + rec, err := r.bldr.NewRecordBatchChecked() + if err != nil { + if r.err == nil { + r.err = err + } + r.done = true + return false + } + r.cur = rec + return true +} + func (r *JSONReader) nextall() bool { for r.readNext() { } - r.cur = r.bldr.NewRecordBatch() + if !r.build() { + return false + } return r.cur.NumRows() > 0 } @@ -192,8 +207,7 @@ func (r *JSONReader) next1() bool { return false } - r.cur = r.bldr.NewRecordBatch() - return true + return r.build() } func (r *JSONReader) nextn() bool { @@ -206,9 +220,9 @@ func (r *JSONReader) nextn() bool { } if n > 0 { - r.cur = r.bldr.NewRecordBatch() + return r.build() } - return n > 0 + return false } var _ RecordReader = (*JSONReader)(nil) diff --git a/arrow/array/record.go b/arrow/array/record.go index b7f84180a..9a2aee0b3 100644 --- a/arrow/array/record.go +++ b/arrow/array/record.go @@ -231,6 +231,28 @@ func (rec *simpleRecord) validate() error { return nil } +// nullCountInRows returns the number of top-level nulls among the first nrows +// rows of arr. It does not recurse into child arrays, so a nullable struct whose +// non-nullable children carry nulls in null parent slots is not flagged; only a +// field's own top-level nullability is enforced. Run-end-encoded logical nulls, +// held in the values child rather than a top-level bitmap, are not counted. +func nullCountInRows(arr arrow.Array, nrows int64) int { + if arr == nil || nrows <= 0 { + return 0 + } + n := int(nrows) + if n >= arr.Len() { + return arr.NullN() + } + count := 0 + for i := 0; i < n; i++ { + if arr.IsNull(i) { + count++ + } + } + return count +} + // Retain increases the reference count by 1. // Retain may be called simultaneously from multiple goroutines. func (rec *simpleRecord) Retain() { @@ -385,11 +407,26 @@ func (b *RecordBuilder) Resize(n int) { // // The returned RecordBatch must be Release()'d after use. // -// NewRecordBatch panics if the fields' builder do not have the same length. +// NewRecordBatch panics if the fields' builders do not have the same length or if +// a non-nullable field contains nulls. Use NewRecordBatchChecked for an error. func (b *RecordBuilder) NewRecordBatch() arrow.RecordBatch { + rec, err := b.newRecordBatch() + if err != nil { + panic(err) + } + return rec +} + +// NewRecordBatchChecked is like NewRecordBatch but returns an error instead of +// panicking when the record is invalid. +func (b *RecordBuilder) NewRecordBatchChecked() (arrow.RecordBatch, error) { + return b.newRecordBatch() +} + +func (b *RecordBuilder) newRecordBatch() (arrow.RecordBatch, error) { lower, upper := b.columnLenRange() if lower != upper { - panic(fmt.Errorf("arrow/array: some fields have excessive number of rows (want at most %d, have %d)", lower, upper)) + return nil, fmt.Errorf("arrow/array: some fields have excessive number of rows (want at most %d, have %d)", lower, upper) } cols := make([]arrow.Array, len(b.fields)) @@ -407,7 +444,16 @@ func (b *RecordBuilder) NewRecordBatch() arrow.RecordBatch { cols[i] = f.NewArray() } - return NewRecordBatch(b.schema, cols, int64(lower)) + for i := range cols { + f := b.schema.Field(i) + if !f.Nullable { + if n := nullCountInRows(cols[i], int64(lower)); n > 0 { + return nil, fmt.Errorf("arrow/array: field %q is declared non-nullable but contains %d null value(s)", f.Name, n) + } + } + } + + return NewRecordBatch(b.schema, cols, int64(lower)), nil } // Deprecated: Use [NewRecordBatch] instead. diff --git a/arrow/array/record_test.go b/arrow/array/record_test.go index a3924382a..c4a8e39c8 100644 --- a/arrow/array/record_test.go +++ b/arrow/array/record_test.go @@ -26,6 +26,7 @@ import ( "github.com/apache/arrow-go/v18/arrow/array" "github.com/apache/arrow-go/v18/arrow/memory" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestRecord(t *testing.T) { @@ -476,6 +477,106 @@ func TestRecordBuilderRespectsFixedSizeArrayNullability(t *testing.T) { } } +func TestRecordBuilderNonNullableNulls(t *testing.T) { + mem := memory.NewCheckedAllocator(memory.NewGoAllocator()) + defer mem.AssertSize(t, 0) + + t.Run("non-nullable field with nulls returns error", func(t *testing.T) { + schema := arrow.NewSchema([]arrow.Field{ + {Name: "a", Type: arrow.PrimitiveTypes.Int32, Nullable: false}, + }, nil) + b := array.NewRecordBuilder(mem, schema) + defer b.Release() + + ib := b.Field(0).(*array.Int32Builder) + ib.Append(1) + ib.AppendNull() + + rec, err := b.NewRecordBatchChecked() + assert.Nil(t, rec) + require.Error(t, err) + assert.ErrorContains(t, err, `field "a" is declared non-nullable`) + }) + + t.Run("non-nullable field with nulls panics", func(t *testing.T) { + schema := arrow.NewSchema([]arrow.Field{ + {Name: "a", Type: arrow.PrimitiveTypes.Int32, Nullable: false}, + }, nil) + b := array.NewRecordBuilder(mem, schema) + defer b.Release() + + b.Field(0).(*array.Int32Builder).Append(1) + b.Field(0).(*array.Int32Builder).AppendNull() + + assert.Panics(t, func() { + b.NewRecordBatch().Release() + }) + }) + + t.Run("nullable field with nulls is allowed", func(t *testing.T) { + schema := arrow.NewSchema([]arrow.Field{ + {Name: "a", Type: arrow.PrimitiveTypes.Int32, Nullable: true}, + }, nil) + b := array.NewRecordBuilder(mem, schema) + defer b.Release() + + b.Field(0).(*array.Int32Builder).Append(1) + b.Field(0).(*array.Int32Builder).AppendNull() + + rec, err := b.NewRecordBatchChecked() + require.NoError(t, err) + defer rec.Release() + assert.EqualValues(t, 2, rec.NumRows()) + }) + + t.Run("non-nullable field without nulls is allowed", func(t *testing.T) { + schema := arrow.NewSchema([]arrow.Field{ + {Name: "a", Type: arrow.PrimitiveTypes.Int32, Nullable: false}, + }, nil) + b := array.NewRecordBuilder(mem, schema) + defer b.Release() + + b.Field(0).(*array.Int32Builder).AppendValues([]int32{1, 2, 3}, nil) + + rec, err := b.NewRecordBatchChecked() + require.NoError(t, err) + defer rec.Release() + assert.EqualValues(t, 3, rec.NumRows()) + }) + + t.Run("nullable struct with non-nullable child is allowed", func(t *testing.T) { + structType := arrow.StructOf(arrow.Field{Name: "x", Type: arrow.PrimitiveTypes.Int32, Nullable: false}) + schema := arrow.NewSchema([]arrow.Field{ + {Name: "s", Type: structType, Nullable: true}, + }, nil) + b := array.NewRecordBuilder(mem, schema) + defer b.Release() + + sb := b.Field(0).(*array.StructBuilder) + sb.Append(true) + sb.FieldBuilder(0).(*array.Int32Builder).Append(10) + sb.AppendNull() + + rec, err := b.NewRecordBatchChecked() + require.NoError(t, err) + defer rec.Release() + assert.EqualValues(t, 2, rec.NumRows()) + }) + + t.Run("empty builder is allowed", func(t *testing.T) { + schema := arrow.NewSchema([]arrow.Field{ + {Name: "a", Type: arrow.PrimitiveTypes.Int32, Nullable: false}, + }, nil) + b := array.NewRecordBuilder(mem, schema) + defer b.Release() + + rec, err := b.NewRecordBatchChecked() + require.NoError(t, err) + defer rec.Release() + assert.EqualValues(t, 0, rec.NumRows()) + }) +} + func TestRecordBuilder(t *testing.T) { mem := memory.NewCheckedAllocator(memory.NewGoAllocator()) defer mem.AssertSize(t, 0) diff --git a/arrow/array/table_test.go b/arrow/array/table_test.go index e37a231f8..1539e7c66 100644 --- a/arrow/array/table_test.go +++ b/arrow/array/table_test.go @@ -619,7 +619,7 @@ func TestTableFromRecords(t *testing.T) { schema := arrow.NewSchema( []arrow.Field{ - {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32}, + {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32, Nullable: true}, {Name: "f2-f64", Type: arrow.PrimitiveTypes.Float64}, }, nil, @@ -793,7 +793,7 @@ func TestTableToString(t *testing.T) { schema := arrow.NewSchema( []arrow.Field{ - {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32}, + {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32, Nullable: true}, {Name: "f2-f64", Type: arrow.PrimitiveTypes.Float64}, }, nil, @@ -822,7 +822,7 @@ func TestTableToString(t *testing.T) { expected_str := `schema: fields: 2 - - f1-i32: type=int32 + - f1-i32: type=int32, nullable - f2-f64: type=float64 f1-i32: [[1 2 3 4 5 6 7 8 (null) 10], [111 112 113 114 115 116 117 118 119 120]] f2-f64: [[11 12 13 14 15 16 17 18 19 20], [211 212 213 214 215 216 217 218 219 220]] diff --git a/arrow/avro/loader.go b/arrow/avro/loader.go index a7199e661..4a7ba7ecd 100644 --- a/arrow/avro/loader.go +++ b/arrow/avro/loader.go @@ -46,6 +46,16 @@ func (r *OCFReader) decodeOCFToChan() { } } +func (r *OCFReader) emitRecord() bool { + rec, err := r.bld.NewRecordBatchChecked() + if err != nil { + r.err = err + return false + } + r.recChan <- rec + return true +} + func (r *OCFReader) recordFactory() { defer close(r.recChan) r.primed = true @@ -59,7 +69,9 @@ func (r *OCFReader) recordFactory() { return } } - r.recChan <- r.bld.NewRecordBatch() + if !r.emitRecord() { + return + } r.bldDone <- struct{}{} case r.chunk >= 1: for data := range r.avroChan { @@ -73,12 +85,16 @@ func (r *OCFReader) recordFactory() { } recChunk++ if recChunk >= r.chunk { - r.recChan <- r.bld.NewRecordBatch() + if !r.emitRecord() { + return + } recChunk = 0 } } if recChunk != 0 { - r.recChan <- r.bld.NewRecordBatch() + if !r.emitRecord() { + return + } } r.bldDone <- struct{}{} } diff --git a/arrow/compute/vector_sort_test.go b/arrow/compute/vector_sort_test.go index 39bf5e95f..db28d6e2a 100644 --- a/arrow/compute/vector_sort_test.go +++ b/arrow/compute/vector_sort_test.go @@ -1373,8 +1373,8 @@ func TestVectorSortIndicesCppRecordBatchParity(t *testing.T) { t.Run("Null", func(t *testing.T) { schema := arrow.NewSchema([]arrow.Field{ - {Name: "a", Type: arrow.PrimitiveTypes.Uint8}, - {Name: "b", Type: arrow.PrimitiveTypes.Uint32}, + {Name: "a", Type: arrow.PrimitiveTypes.Uint8, Nullable: true}, + {Name: "b", Type: arrow.PrimitiveTypes.Uint32, Nullable: true}, }, nil) jsonRows := `[ {"a": null, "b": 5}, @@ -1460,8 +1460,8 @@ func TestVectorSortIndicesCppRecordBatchParity(t *testing.T) { t.Run("Boolean", func(t *testing.T) { schema := arrow.NewSchema([]arrow.Field{ - {Name: "a", Type: arrow.FixedWidthTypes.Boolean}, - {Name: "b", Type: arrow.FixedWidthTypes.Boolean}, + {Name: "a", Type: arrow.FixedWidthTypes.Boolean, Nullable: true}, + {Name: "b", Type: arrow.FixedWidthTypes.Boolean, Nullable: true}, }, nil) jsonRows := `[ {"a": true, "b": null}, @@ -1536,7 +1536,7 @@ func TestVectorSortIndicesCppRecordBatchParity(t *testing.T) { d256 := &arrow.Decimal256Type{Precision: 4, Scale: 2} schema := arrow.NewSchema([]arrow.Field{ {Name: "a", Type: d128}, - {Name: "b", Type: d256}, + {Name: "b", Type: d256, Nullable: true}, }, nil) jsonRows := `[ {"a": "12.3", "b": "12.34"}, @@ -1610,8 +1610,8 @@ func TestVectorSortIndicesCppTableParity(t *testing.T) { ctx := context.Background() schemaAB := arrow.NewSchema([]arrow.Field{ - {Name: "a", Type: arrow.PrimitiveTypes.Uint8}, - {Name: "b", Type: arrow.PrimitiveTypes.Uint32}, + {Name: "a", Type: arrow.PrimitiveTypes.Uint8, Nullable: true}, + {Name: "b", Type: arrow.PrimitiveTypes.Uint32, Nullable: true}, }, nil) t.Run("EmptyTable", func(t *testing.T) { diff --git a/arrow/csv/reader.go b/arrow/csv/reader.go index 9f3c77242..fef3a124e 100644 --- a/arrow/csv/reader.go +++ b/arrow/csv/reader.go @@ -267,6 +267,19 @@ func (r *Reader) Next() bool { return r.next() } +func (r *Reader) build() bool { + rec, err := r.bld.NewRecordBatchChecked() + if err != nil { + if r.err == nil { + r.err = err + } + r.done = true + return false + } + r.cur = rec + return true +} + // next1 reads one row from the CSV file and creates a single Record // from that row. func (r *Reader) next1() bool { @@ -282,9 +295,8 @@ func (r *Reader) next1() bool { r.validate(recs) r.read(recs) - r.cur = r.bld.NewRecordBatch() - return true + return r.build() } // nextall reads the whole CSV file into memory and creates one single @@ -305,9 +317,8 @@ func (r *Reader) nextall() bool { r.validate(rec) r.read(rec) } - r.cur = r.bld.NewRecordBatch() - return true + return r.build() } // nextn reads n rows from the CSV file, where n is the chunk size, and creates @@ -338,7 +349,9 @@ func (r *Reader) nextn() bool { r.done = true } - r.cur = r.bld.NewRecordBatch() + if !r.build() { + return false + } return n > 0 } diff --git a/arrow/csv/reader_test.go b/arrow/csv/reader_test.go index 60af76faa..7d9b9e27f 100644 --- a/arrow/csv/reader_test.go +++ b/arrow/csv/reader_test.go @@ -236,7 +236,7 @@ func TestCSVReaderParseError(t *testing.T) { schema := arrow.NewSchema( []arrow.Field{ {Name: "i64", Type: arrow.PrimitiveTypes.Int64}, - {Name: "f64", Type: arrow.PrimitiveTypes.Float64}, + {Name: "f64", Type: arrow.PrimitiveTypes.Float64, Nullable: true}, {Name: "str", Type: arrow.BinaryTypes.String}, }, nil, @@ -335,30 +335,30 @@ func testCSVReader(t *testing.T, filepath string, withHeader bool, stringsCanBeN schema := arrow.NewSchema( []arrow.Field{ - {Name: "bool", Type: arrow.FixedWidthTypes.Boolean}, - {Name: "i8", Type: arrow.PrimitiveTypes.Int8}, - {Name: "i16", Type: arrow.PrimitiveTypes.Int16}, - {Name: "i32", Type: arrow.PrimitiveTypes.Int32}, - {Name: "i64", Type: arrow.PrimitiveTypes.Int64}, - {Name: "u8", Type: arrow.PrimitiveTypes.Uint8}, - {Name: "u16", Type: arrow.PrimitiveTypes.Uint16}, - {Name: "u32", Type: arrow.PrimitiveTypes.Uint32}, - {Name: "u64", Type: arrow.PrimitiveTypes.Uint64}, - {Name: "f16", Type: arrow.FixedWidthTypes.Float16}, - {Name: "f32", Type: arrow.PrimitiveTypes.Float32}, - {Name: "f64", Type: arrow.PrimitiveTypes.Float64}, - {Name: "str", Type: arrow.BinaryTypes.String}, - {Name: "large_str", Type: arrow.BinaryTypes.LargeString}, - {Name: "ts", Type: arrow.FixedWidthTypes.Timestamp_ms}, - {Name: "list(i64)", Type: arrow.ListOf(arrow.PrimitiveTypes.Int64)}, - {Name: "large_list(i64)", Type: arrow.LargeListOf(arrow.PrimitiveTypes.Int64)}, - {Name: "fixed_size_list(i64)", Type: arrow.FixedSizeListOf(3, arrow.PrimitiveTypes.Int64)}, - {Name: "binary", Type: arrow.BinaryTypes.Binary}, - {Name: "large_binary", Type: arrow.BinaryTypes.LargeBinary}, - {Name: "fixed_size_binary", Type: &arrow.FixedSizeBinaryType{ByteWidth: 3}}, - {Name: "uuid", Type: extensions.NewUUIDType()}, - {Name: "date32", Type: arrow.PrimitiveTypes.Date32}, - {Name: "date64", Type: arrow.PrimitiveTypes.Date64}, + {Name: "bool", Type: arrow.FixedWidthTypes.Boolean, Nullable: true}, + {Name: "i8", Type: arrow.PrimitiveTypes.Int8, Nullable: true}, + {Name: "i16", Type: arrow.PrimitiveTypes.Int16, Nullable: true}, + {Name: "i32", Type: arrow.PrimitiveTypes.Int32, Nullable: true}, + {Name: "i64", Type: arrow.PrimitiveTypes.Int64, Nullable: true}, + {Name: "u8", Type: arrow.PrimitiveTypes.Uint8, Nullable: true}, + {Name: "u16", Type: arrow.PrimitiveTypes.Uint16, Nullable: true}, + {Name: "u32", Type: arrow.PrimitiveTypes.Uint32, Nullable: true}, + {Name: "u64", Type: arrow.PrimitiveTypes.Uint64, Nullable: true}, + {Name: "f16", Type: arrow.FixedWidthTypes.Float16, Nullable: true}, + {Name: "f32", Type: arrow.PrimitiveTypes.Float32, Nullable: true}, + {Name: "f64", Type: arrow.PrimitiveTypes.Float64, Nullable: true}, + {Name: "str", Type: arrow.BinaryTypes.String, Nullable: true}, + {Name: "large_str", Type: arrow.BinaryTypes.LargeString, Nullable: true}, + {Name: "ts", Type: arrow.FixedWidthTypes.Timestamp_ms, Nullable: true}, + {Name: "list(i64)", Type: arrow.ListOf(arrow.PrimitiveTypes.Int64), Nullable: true}, + {Name: "large_list(i64)", Type: arrow.LargeListOf(arrow.PrimitiveTypes.Int64), Nullable: true}, + {Name: "fixed_size_list(i64)", Type: arrow.FixedSizeListOf(3, arrow.PrimitiveTypes.Int64), Nullable: true}, + {Name: "binary", Type: arrow.BinaryTypes.Binary, Nullable: true}, + {Name: "large_binary", Type: arrow.BinaryTypes.LargeBinary, Nullable: true}, + {Name: "fixed_size_binary", Type: &arrow.FixedSizeBinaryType{ByteWidth: 3}, Nullable: true}, + {Name: "uuid", Type: extensions.NewUUIDType(), Nullable: true}, + {Name: "date32", Type: arrow.PrimitiveTypes.Date32, Nullable: true}, + {Name: "date64", Type: arrow.PrimitiveTypes.Date64, Nullable: true}, }, nil, ) diff --git a/arrow/csv/writer_test.go b/arrow/csv/writer_test.go index e227b3b76..fab66a1e9 100644 --- a/arrow/csv/writer_test.go +++ b/arrow/csv/writer_test.go @@ -213,33 +213,33 @@ func testCSVWriter(t *testing.T, data [][]string, writeHeader bool, fmtr func(bo defer pool.AssertSize(t, 0) schema := arrow.NewSchema( []arrow.Field{ - {Name: "bool", Type: arrow.FixedWidthTypes.Boolean}, - {Name: "i8", Type: arrow.PrimitiveTypes.Int8}, - {Name: "i16", Type: arrow.PrimitiveTypes.Int16}, - {Name: "i32", Type: arrow.PrimitiveTypes.Int32}, - {Name: "i64", Type: arrow.PrimitiveTypes.Int64}, - {Name: "u8", Type: arrow.PrimitiveTypes.Uint8}, - {Name: "u16", Type: arrow.PrimitiveTypes.Uint16}, - {Name: "u32", Type: arrow.PrimitiveTypes.Uint32}, - {Name: "u64", Type: arrow.PrimitiveTypes.Uint64}, - {Name: "f16", Type: arrow.FixedWidthTypes.Float16}, - {Name: "f32", Type: arrow.PrimitiveTypes.Float32}, - {Name: "f64", Type: arrow.PrimitiveTypes.Float64}, - {Name: "str", Type: arrow.BinaryTypes.String}, - {Name: "large_str", Type: arrow.BinaryTypes.LargeString}, - {Name: "ts_s", Type: arrow.FixedWidthTypes.Timestamp_s}, - {Name: "d32", Type: arrow.FixedWidthTypes.Date32}, - {Name: "d64", Type: arrow.FixedWidthTypes.Date64}, - {Name: "dec128", Type: &arrow.Decimal128Type{Precision: 5, Scale: 2}}, - {Name: "dec256", Type: &arrow.Decimal256Type{Precision: 5, Scale: 2}}, - {Name: "list(i64)", Type: arrow.ListOf(arrow.PrimitiveTypes.Int64)}, - {Name: "large_list(i64)", Type: arrow.LargeListOf(arrow.PrimitiveTypes.Int64)}, - {Name: "fixed_size_list(i64)", Type: arrow.FixedSizeListOf(3, arrow.PrimitiveTypes.Int64)}, - {Name: "binary", Type: arrow.BinaryTypes.Binary}, - {Name: "large_binary", Type: arrow.BinaryTypes.LargeBinary}, - {Name: "fixed_size_binary", Type: &arrow.FixedSizeBinaryType{ByteWidth: 3}}, - {Name: "uuid", Type: extensions.NewUUIDType()}, - {Name: "null", Type: arrow.Null}, + {Name: "bool", Type: arrow.FixedWidthTypes.Boolean, Nullable: true}, + {Name: "i8", Type: arrow.PrimitiveTypes.Int8, Nullable: true}, + {Name: "i16", Type: arrow.PrimitiveTypes.Int16, Nullable: true}, + {Name: "i32", Type: arrow.PrimitiveTypes.Int32, Nullable: true}, + {Name: "i64", Type: arrow.PrimitiveTypes.Int64, Nullable: true}, + {Name: "u8", Type: arrow.PrimitiveTypes.Uint8, Nullable: true}, + {Name: "u16", Type: arrow.PrimitiveTypes.Uint16, Nullable: true}, + {Name: "u32", Type: arrow.PrimitiveTypes.Uint32, Nullable: true}, + {Name: "u64", Type: arrow.PrimitiveTypes.Uint64, Nullable: true}, + {Name: "f16", Type: arrow.FixedWidthTypes.Float16, Nullable: true}, + {Name: "f32", Type: arrow.PrimitiveTypes.Float32, Nullable: true}, + {Name: "f64", Type: arrow.PrimitiveTypes.Float64, Nullable: true}, + {Name: "str", Type: arrow.BinaryTypes.String, Nullable: true}, + {Name: "large_str", Type: arrow.BinaryTypes.LargeString, Nullable: true}, + {Name: "ts_s", Type: arrow.FixedWidthTypes.Timestamp_s, Nullable: true}, + {Name: "d32", Type: arrow.FixedWidthTypes.Date32, Nullable: true}, + {Name: "d64", Type: arrow.FixedWidthTypes.Date64, Nullable: true}, + {Name: "dec128", Type: &arrow.Decimal128Type{Precision: 5, Scale: 2}, Nullable: true}, + {Name: "dec256", Type: &arrow.Decimal256Type{Precision: 5, Scale: 2}, Nullable: true}, + {Name: "list(i64)", Type: arrow.ListOf(arrow.PrimitiveTypes.Int64), Nullable: true}, + {Name: "large_list(i64)", Type: arrow.LargeListOf(arrow.PrimitiveTypes.Int64), Nullable: true}, + {Name: "fixed_size_list(i64)", Type: arrow.FixedSizeListOf(3, arrow.PrimitiveTypes.Int64), Nullable: true}, + {Name: "binary", Type: arrow.BinaryTypes.Binary, Nullable: true}, + {Name: "large_binary", Type: arrow.BinaryTypes.LargeBinary, Nullable: true}, + {Name: "fixed_size_binary", Type: &arrow.FixedSizeBinaryType{ByteWidth: 3}, Nullable: true}, + {Name: "uuid", Type: extensions.NewUUIDType(), Nullable: true}, + {Name: "null", Type: arrow.Null, Nullable: true}, }, nil, ) diff --git a/arrow/example_test.go b/arrow/example_test.go index 2b0dc5d4f..5c25ead66 100644 --- a/arrow/example_test.go +++ b/arrow/example_test.go @@ -463,7 +463,7 @@ func Example_record() { schema := arrow.NewSchema( []arrow.Field{ - {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32}, + {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32, Nullable: true}, {Name: "f2-f64", Type: arrow.PrimitiveTypes.Float64}, }, nil, @@ -493,7 +493,7 @@ func Example_recordReader() { schema := arrow.NewSchema( []arrow.Field{ - {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32}, + {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32, Nullable: true}, {Name: "f2-f64", Type: arrow.PrimitiveTypes.Float64}, }, nil, @@ -542,7 +542,7 @@ func Example_table() { schema := arrow.NewSchema( []arrow.Field{ - {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32}, + {Name: "f1-i32", Type: arrow.PrimitiveTypes.Int32, Nullable: true}, {Name: "f2-f64", Type: arrow.PrimitiveTypes.Float64}, }, nil, diff --git a/arrow/extensions/uuid_test.go b/arrow/extensions/uuid_test.go index a76b77a91..36bb812b9 100644 --- a/arrow/extensions/uuid_test.go +++ b/arrow/extensions/uuid_test.go @@ -62,7 +62,7 @@ func TestUUIDExtensionBuilder(t *testing.T) { func TestUUIDExtensionRecordBuilder(t *testing.T) { schema := arrow.NewSchema([]arrow.Field{ - {Name: "uuid", Type: extensions.NewUUIDType()}, + {Name: "uuid", Type: extensions.NewUUIDType(), Nullable: true}, }, nil) builder := array.NewRecordBuilder(memory.DefaultAllocator, schema) builder.Field(0).(*extensions.UUIDBuilder).Append(testUUID) diff --git a/arrow/internal/arrdata/arrdata.go b/arrow/internal/arrdata/arrdata.go index 095571a8f..250413e83 100644 --- a/arrow/internal/arrdata/arrdata.go +++ b/arrow/internal/arrdata/arrdata.go @@ -433,8 +433,8 @@ func makeFixedSizeListsRecords() []arrow.RecordBatch { func makeStringsRecords() []arrow.RecordBatch { mem := memory.NewGoAllocator() schema := arrow.NewSchema([]arrow.Field{ - {Name: "strings", Type: arrow.BinaryTypes.String}, - {Name: "bytes", Type: arrow.BinaryTypes.Binary}, + {Name: "strings", Type: arrow.BinaryTypes.String, Nullable: true}, + {Name: "bytes", Type: arrow.BinaryTypes.Binary, Nullable: true}, }, nil) mask := []bool{true, false, false, true, true} diff --git a/arrow/internal/arrjson/arrjson_test.go b/arrow/internal/arrjson/arrjson_test.go index 7e2f386fc..08ef92fef 100644 --- a/arrow/internal/arrjson/arrjson_test.go +++ b/arrow/internal/arrjson/arrjson_test.go @@ -1957,7 +1957,7 @@ func makeStringsWantJSONs() string { "type": { "name": "utf8" }, - "nullable": false, + "nullable": true, "children": [] }, { @@ -1965,7 +1965,7 @@ func makeStringsWantJSONs() string { "type": { "name": "binary" }, - "nullable": false, + "nullable": true, "children": [] } ] diff --git a/arrow/ipc/cmd/arrow-ls/main_test.go b/arrow/ipc/cmd/arrow-ls/main_test.go index f90e4a800..f9d064e3c 100644 --- a/arrow/ipc/cmd/arrow-ls/main_test.go +++ b/arrow/ipc/cmd/arrow-ls/main_test.go @@ -75,8 +75,8 @@ records: 4 name: "strings", want: `schema: fields: 2 - - strings: type=utf8 - - bytes: type=binary + - strings: type=utf8, nullable + - bytes: type=binary, nullable records: 3 `, }, diff --git a/arrow/ipc/ipc_test.go b/arrow/ipc/ipc_test.go index 93bd987a6..395a3c6d4 100644 --- a/arrow/ipc/ipc_test.go +++ b/arrow/ipc/ipc_test.go @@ -282,7 +282,7 @@ func TestWriteColumnWithOffset(t *testing.T) { func TestIPCTable(t *testing.T) { pool := memory.NewGoAllocator() - schema := arrow.NewSchema([]arrow.Field{{Name: "f1", Type: arrow.PrimitiveTypes.Int32}}, nil) + schema := arrow.NewSchema([]arrow.Field{{Name: "f1", Type: arrow.PrimitiveTypes.Int32, Nullable: true}}, nil) b := array.NewRecordBuilder(pool, schema) defer b.Release() b.Field(0).(*array.Int32Builder).AppendValues([]int32{1, 2, 3, 4}, []bool{true, true, false, true}) diff --git a/arrow/ipc/reader_test.go b/arrow/ipc/reader_test.go index 29db5985a..8883f3ba8 100644 --- a/arrow/ipc/reader_test.go +++ b/arrow/ipc/reader_test.go @@ -100,7 +100,7 @@ func TestReaderCheckedAllocator(t *testing.T) { func TestMappedReader(t *testing.T) { pool := memory.NewCheckedAllocator(memory.NewGoAllocator()) defer pool.AssertSize(t, 0) - schema := arrow.NewSchema([]arrow.Field{{Name: "f1", Type: arrow.PrimitiveTypes.Int32}}, nil) + schema := arrow.NewSchema([]arrow.Field{{Name: "f1", Type: arrow.PrimitiveTypes.Int32, Nullable: true}}, nil) b := array.NewRecordBuilder(pool, schema) defer b.Release() b.Field(0).(*array.Int32Builder).AppendValues([]int32{1, 2, 3, 4}, []bool{true, true, false, true}) diff --git a/parquet/pqarrow/encode_arrow_test.go b/parquet/pqarrow/encode_arrow_test.go index 95a5b3e5d..c2b801d17 100644 --- a/parquet/pqarrow/encode_arrow_test.go +++ b/parquet/pqarrow/encode_arrow_test.go @@ -482,13 +482,13 @@ func TestWriteKeyValueMetadata(t *testing.T) { func TestWriteEmptyLists(t *testing.T) { sc := arrow.NewSchema([]arrow.Field{ - {Name: "f1", Type: arrow.ListOf(arrow.FixedWidthTypes.Date32)}, - {Name: "f2", Type: arrow.ListOf(arrow.FixedWidthTypes.Date64)}, - {Name: "f3", Type: arrow.ListOf(arrow.FixedWidthTypes.Timestamp_us)}, - {Name: "f4", Type: arrow.ListOf(arrow.FixedWidthTypes.Timestamp_ms)}, - {Name: "f5", Type: arrow.ListOf(arrow.FixedWidthTypes.Time32ms)}, - {Name: "f6", Type: arrow.ListOf(arrow.FixedWidthTypes.Time64ns)}, - {Name: "f7", Type: arrow.ListOf(arrow.FixedWidthTypes.Time64us)}, + {Name: "f1", Type: arrow.ListOf(arrow.FixedWidthTypes.Date32), Nullable: true}, + {Name: "f2", Type: arrow.ListOf(arrow.FixedWidthTypes.Date64), Nullable: true}, + {Name: "f3", Type: arrow.ListOf(arrow.FixedWidthTypes.Timestamp_us), Nullable: true}, + {Name: "f4", Type: arrow.ListOf(arrow.FixedWidthTypes.Timestamp_ms), Nullable: true}, + {Name: "f5", Type: arrow.ListOf(arrow.FixedWidthTypes.Time32ms), Nullable: true}, + {Name: "f6", Type: arrow.ListOf(arrow.FixedWidthTypes.Time64ns), Nullable: true}, + {Name: "f7", Type: arrow.ListOf(arrow.FixedWidthTypes.Time64us), Nullable: true}, }, nil) bldr := array.NewRecordBuilder(memory.DefaultAllocator, sc) defer bldr.Release() @@ -515,19 +515,19 @@ func TestWriteEmptyLists(t *testing.T) { func TestWriteAllNullsWithDeltaEncoding(t *testing.T) { sc := arrow.NewSchema([]arrow.Field{ {Name: "f1", Type: arrow.PrimitiveTypes.Int64, Nullable: true}, - {Name: "f2", Type: arrow.ListOf(arrow.FixedWidthTypes.Date32)}, + {Name: "f2", Type: arrow.ListOf(arrow.FixedWidthTypes.Date32), Nullable: true}, {Name: "f3", Type: arrow.BinaryTypes.String, Nullable: true}, - {Name: "f4", Type: arrow.ListOf(arrow.BinaryTypes.String)}, + {Name: "f4", Type: arrow.ListOf(arrow.BinaryTypes.String), Nullable: true}, {Name: "f5", Type: arrow.BinaryTypes.LargeString, Nullable: true}, - {Name: "f6", Type: arrow.ListOf(arrow.BinaryTypes.LargeString)}, + {Name: "f6", Type: arrow.ListOf(arrow.BinaryTypes.LargeString), Nullable: true}, {Name: "f7", Type: arrow.PrimitiveTypes.Float64, Nullable: true}, - {Name: "f8", Type: arrow.ListOf(arrow.FixedWidthTypes.Date64)}, + {Name: "f8", Type: arrow.ListOf(arrow.FixedWidthTypes.Date64), Nullable: true}, {Name: "f9", Type: arrow.BinaryTypes.String, Nullable: true}, - {Name: "f10", Type: arrow.ListOf(arrow.BinaryTypes.LargeString)}, + {Name: "f10", Type: arrow.ListOf(arrow.BinaryTypes.LargeString), Nullable: true}, {Name: "f11", Type: arrow.FixedWidthTypes.Boolean, Nullable: true}, - {Name: "f12", Type: arrow.ListOf(arrow.FixedWidthTypes.Boolean)}, + {Name: "f12", Type: arrow.ListOf(arrow.FixedWidthTypes.Boolean), Nullable: true}, {Name: "f13", Type: arrow.PrimitiveTypes.Int32, Nullable: true}, - {Name: "f14", Type: arrow.ListOf(arrow.PrimitiveTypes.Float32)}, + {Name: "f14", Type: arrow.ListOf(arrow.PrimitiveTypes.Float32), Nullable: true}, }, nil) bldr := array.NewRecordBuilder(memory.DefaultAllocator, sc) defer bldr.Release() From 3b128df0136df086746b06329ea7edcdaf448b33 Mon Sep 17 00:00:00 2001 From: Matt Topol Date: Tue, 23 Jun 2026 12:50:47 -0400 Subject: [PATCH 2/2] fix(arrow/array): use cached NullN for the non-nullable record check Address review feedback: drop the manual per-row null scan (and the nullCountInRows helper) in favor of the cached Array.NullN(). We only need to know whether any nulls are present, not the exact count, and NullN is already tracked, so this avoids iterating the column. --- arrow/array/record.go | 33 ++++++--------------------------- 1 file changed, 6 insertions(+), 27 deletions(-) diff --git a/arrow/array/record.go b/arrow/array/record.go index 9a2aee0b3..cb81d5af2 100644 --- a/arrow/array/record.go +++ b/arrow/array/record.go @@ -231,28 +231,6 @@ func (rec *simpleRecord) validate() error { return nil } -// nullCountInRows returns the number of top-level nulls among the first nrows -// rows of arr. It does not recurse into child arrays, so a nullable struct whose -// non-nullable children carry nulls in null parent slots is not flagged; only a -// field's own top-level nullability is enforced. Run-end-encoded logical nulls, -// held in the values child rather than a top-level bitmap, are not counted. -func nullCountInRows(arr arrow.Array, nrows int64) int { - if arr == nil || nrows <= 0 { - return 0 - } - n := int(nrows) - if n >= arr.Len() { - return arr.NullN() - } - count := 0 - for i := 0; i < n; i++ { - if arr.IsNull(i) { - count++ - } - } - return count -} - // Retain increases the reference count by 1. // Retain may be called simultaneously from multiple goroutines. func (rec *simpleRecord) Retain() { @@ -444,12 +422,13 @@ func (b *RecordBuilder) newRecordBatch() (arrow.RecordBatch, error) { cols[i] = f.NewArray() } + // Only each field's own top-level nullability is enforced; we do not recurse + // into children, so a non-nullable child of a nullable struct keeps its + // legitimate nulls in null parent slots. Run-end-encoded logical nulls live + // in the values child rather than a top-level bitmap, so they are not flagged. for i := range cols { - f := b.schema.Field(i) - if !f.Nullable { - if n := nullCountInRows(cols[i], int64(lower)); n > 0 { - return nil, fmt.Errorf("arrow/array: field %q is declared non-nullable but contains %d null value(s)", f.Name, n) - } + if f := b.schema.Field(i); !f.Nullable && cols[i].NullN() > 0 { + return nil, fmt.Errorf("arrow/array: field %q is declared non-nullable but contains nulls", f.Name) } }