Skip to content

fix(arrow/array): reject nulls in non-nullable fields when building records #858

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zeroshade wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix(arrow/array): reject nulls in non-nullable fields when building records #858

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7423d0f
fix(arrow/array): reject nulls in non-nullable fields when building r…
zeroshade Jun 19, 2026
3b128df
fix(arrow/array): use cached NullN for the non-nullable record check
zeroshade Jun 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 19 additions & 5 deletions arrow/array/json_reader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand All @@ -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 {
Expand All @@ -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)
31 changes: 28 additions & 3 deletions arrow/array/record.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -385,11 +385,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.
Comment on lines +388 to +389

@lidavidm lidavidm Jun 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

While not strictly breaking, this might be a pretty painful change, if NewRecordBatch now "randomly" panics on things it would've accepted before; maybe we should deprecate this too as a stronger signal?

@zeroshade zeroshade Jun 23, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I dug into deprecating it, but it turns out to be disproportionate for this PR. staticcheck's SA1019 fires on cross-package use, so deprecating NewRecordBatch would mean migrating ~76 call sites across ~20 files to NewRecordBatchChecked (golangci-lint caps output at 3 per message, which hid the real count at first). That set also includes the new TestRecordBuilderNonNullableNulls, which deliberately calls NewRecordBatch() to assert it panics, so it would need a //nolint:staticcheck,SA1019.

Since NewRecordBatch already panics today on a whole-builder length mismatch, rejecting nulls in a non-nullable column reads as a consistent extension of that existing contract rather than brand-new panic behavior — and NewRecordBatchChecked is there for callers who'd rather have an error. So for now I've left NewRecordBatch as-is with the clarified doc. Happy to do the deprecation + full caller migration as a dedicated follow-up if you think the stronger signal is worth it; it just felt too sweeping to fold into this fix. WDYT?

@lidavidm lidavidm Jun 23, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Um, I'm not sure if I'm talking to you or a bot, but I'd think accidentally inserting a null (especially because nullable defaults to false due to false being the zero value) is far more common, and a panic is fairly unwelcome.

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))
Expand All @@ -407,7 +422,17 @@ func (b *RecordBuilder) NewRecordBatch() arrow.RecordBatch {
cols[i] = f.NewArray()
}

return NewRecordBatch(b.schema, cols, int64(lower))
// 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 {
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)
}
}

return NewRecordBatch(b.schema, cols, int64(lower)), nil
}

// Deprecated: Use [NewRecordBatch] instead.
Expand Down
101 changes: 101 additions & 0 deletions arrow/array/record_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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)
Expand Down
6 changes: 3 additions & 3 deletions arrow/array/table_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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]]
Expand Down
22 changes: 19 additions & 3 deletions arrow/avro/loader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 {
Expand All @@ -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{}{}
}
Expand Down
14 changes: 7 additions & 7 deletions arrow/compute/vector_sort_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand Down Expand Up @@ -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},
Expand Down Expand Up @@ -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"},
Expand Down Expand Up @@ -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) {
Expand Down
23 changes: 18 additions & 5 deletions arrow/csv/reader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
}

Expand Down
Loading
Loading