Skip to content

fix(arrow/csv): append null for later parse failures #1149

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

Merged
zeroshade merged 3 commits into from
Aug 14, 2026
+152 −56
Merged

fix(arrow/csv): append null for later parse failures #1149

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
84201b6
fix(arrow/csv): append null for later parse failures
fallintoplace Aug 8, 2026
2eb2f10
test(arrow/csv): handle uint64 example column
fallintoplace Aug 11, 2026
5971924
fix(arrow/csv): keep builders aligned after parse errors
fallintoplace Aug 12, 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
17 changes: 14 additions & 3 deletions arrow/csv/example_csv_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ func Example_reader() {
{Name: "c7", Type: arrow.PrimitiveTypes.Int64, Nullable: true},
{Name: "c8", Type: arrow.PrimitiveTypes.Int64, Nullable: true},
{Name: "c9", Type: arrow.PrimitiveTypes.Int64, Nullable: true},
{Name: "c10", Type: arrow.PrimitiveTypes.Int64, Nullable: true},
{Name: "c10", Type: arrow.PrimitiveTypes.Uint64, Nullable: true},
{Name: "c11", Type: arrow.PrimitiveTypes.Float64, Nullable: true},
{Name: "c12", Type: arrow.PrimitiveTypes.Float64, Nullable: true},
{Name: "c13", Type: arrow.BinaryTypes.String, Nullable: true},
Expand All @@ -64,6 +64,10 @@ func Example_reader() {
fmt.Println("No records found")
return
}
if err := reader.Err(); err != nil {
fmt.Printf("Error reading CSV: %v\n", err)
return
}

// Get the record but don't release it - the reader will handle that
record := reader.RecordBatch()
Expand All @@ -73,7 +77,7 @@ func Example_reader() {
fmt.Println()

fmt.Println("Basic statistics for numeric columns:")
for i := 1; i < 10; i++ { // cols c2 through c10 are Int64
for i := 1; i < 9; i++ { // cols c2 through c9 are Int64
col := record.Column(i).(*array.Int64)
var sum int64
for j := 0; j < col.Len(); j++ {
Expand All @@ -83,6 +87,13 @@ func Example_reader() {
fmt.Printf("Column c%d: Average = %.2f\n", i+1, avg)
}

col := record.Column(9).(*array.Uint64)
var sum float64
for j := 0; j < col.Len(); j++ {
sum += float64(col.Value(j))
}
fmt.Printf("Column c10: Average = %.2f\n", sum/float64(col.Len()))

for i := 10; i < 12; i++ { // cols c11 and c12 are Float64
col := record.Column(i).(*array.Float64)
var sum float64
Expand All @@ -106,7 +117,7 @@ func Example_reader() {
// Column c7: Average = 130.60
// Column c8: Average = 30176.41
// Column c9: Average = 2220897700.60
// Column c10: Average = -86834033398685392.00
// Column c10: Average = 8652627809237404672.00
// Column c11: Average = 0.4793
// Column c12: Average = 0.5090
}
127 changes: 74 additions & 53 deletions arrow/csv/reader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -385,6 +385,12 @@ func (r *Reader) isNull(val string) bool {
return false
}

func (r *Reader) setParseError(err error) {

@zeroshade zeroshade Aug 12, 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.

Please route all conversion failures through this first-error helper, not only the numeric, temporal, and decimal subset. The binary converters below still assign r.err directly.

Reproduced with first-error,%%% under an int8,binary schema: the later base64 error replaces the original integer error. This contradicts the intended “keep the first error” behavior.

if r.err == nil {
r.err = err
}
}

func (r *Reader) read(recs []string) {
for i, str := range recs {
r.fieldConverter[i](str)
Expand Down Expand Up @@ -532,7 +538,7 @@ func (r *Reader) parseBool(field array.Builder, str string) {

v, err := strconv.ParseBool(str)
if err != nil {
r.err = fmt.Errorf("%w: unrecognized boolean: %s", err, str)
r.setParseError(fmt.Errorf("%w: unrecognized boolean: %s", err, str))
field.AppendNull()
return
}
Expand All @@ -547,8 +553,8 @@ func (r *Reader) parseInt8(field array.Builder, str string) {
}

v, err := strconv.ParseInt(str, 10, 8)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -563,8 +569,8 @@ func (r *Reader) parseInt16(field array.Builder, str string) {
}

v, err := strconv.ParseInt(str, 10, 16)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -579,8 +585,8 @@ func (r *Reader) parseInt32(field array.Builder, str string) {
}

v, err := strconv.ParseInt(str, 10, 32)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -595,8 +601,8 @@ func (r *Reader) parseInt64(field array.Builder, str string) {
}

v, err := strconv.ParseInt(str, 10, 64)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -611,8 +617,8 @@ func (r *Reader) parseUint8(field array.Builder, str string) {
}

v, err := strconv.ParseUint(str, 10, 8)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -627,8 +633,8 @@ func (r *Reader) parseUint16(field array.Builder, str string) {
}

v, err := strconv.ParseUint(str, 10, 16)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -643,8 +649,8 @@ func (r *Reader) parseUint32(field array.Builder, str string) {
}

v, err := strconv.ParseUint(str, 10, 32)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -659,8 +665,8 @@ func (r *Reader) parseUint64(field array.Builder, str string) {
}

v, err := strconv.ParseUint(str, 10, 64)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -675,8 +681,8 @@ func (r *Reader) parseFloat16(field array.Builder, str string) {
}

v, err := strconv.ParseFloat(str, 32)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -690,8 +696,8 @@ func (r *Reader) parseFloat32(field array.Builder, str string) {
}

v, err := strconv.ParseFloat(str, 32)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -705,8 +711,8 @@ func (r *Reader) parseFloat64(field array.Builder, str string) {
}

v, err := strconv.ParseFloat(str, 64)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -721,8 +727,8 @@ func (r *Reader) parseTimestamp(field array.Builder, str string, unit arrow.Time
}

v, err := arrow.TimestampFromString(str, unit)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -737,8 +743,8 @@ func (r *Reader) parseDate32(field array.Builder, str string) {
}

tm, err := time.Parse("2006-01-02", str)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -752,8 +758,8 @@ func (r *Reader) parseDate64(field array.Builder, str string) {
}

tm, err := time.Parse("2006-01-02", str)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -767,8 +773,8 @@ func (r *Reader) parseTime32(field array.Builder, str string, unit arrow.TimeUni
}

val, err := arrow.Time32FromString(str, unit)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -782,8 +788,8 @@ func (r *Reader) parseDecimal128(field array.Builder, str string, prec, scale in
}

val, err := decimal128.FromString(str, prec, scale)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -797,8 +803,8 @@ func (r *Reader) parseDecimal256(field array.Builder, str string, prec, scale in
}

val, err := decimal256.FromString(str, prec, scale)
if err != nil && r.err == nil {
r.err = err
if err != nil {
r.setParseError(err)
field.AppendNull()
return
}
Expand All @@ -811,23 +817,26 @@ func (r *Reader) parseListLike(field array.ListLikeBuilder, str string) {
return
}
if !strings.HasPrefix(str, "{") || !strings.HasSuffix(str, "}") {
r.err = errors.New("invalid list format. should start with '{' and end with '}'")
r.setParseError(errors.New("invalid list format. should start with '{' and end with '}'"))
field.AppendNull()
return
}
str = strings.Trim(str, "{}")
field.Append(true)
if len(str) == 0 {
// we don't want to create the csv reader if we already know the
// string is empty
field.Append(true)
return
}
valueBldr := field.ValueBuilder()
reader := csv.NewReader(strings.NewReader(str))
items, err := reader.Read()
if err != nil {
r.err = err
r.setParseError(err)
field.AppendNull()
return
}
field.Append(true)
valueBldr := field.ValueBuilder()
for _, str := range items {
r.initFieldConverter(valueBldr)(str)
}
Expand All @@ -839,29 +848,39 @@ func (r *Reader) parseFixedSizeList(field *array.FixedSizeListBuilder, str strin
return
}
if !strings.HasPrefix(str, "{") || !strings.HasSuffix(str, "}") {
r.err = errors.New("invalid list format. should start with '{' and end with '}'")
r.setParseError(errors.New("invalid list format. should start with '{' and end with '}'"))
field.AppendNull()
return
}
str = strings.Trim(str, "{}")
field.Append(true)
if len(str) == 0 {
// we don't want to create the csv reader if we already know the
// string is empty
if n != 0 {
r.setParseError(fmt.Errorf("%w: fixed size list items should match the fixed size list length, expected %d, got 0", arrow.ErrInvalid, n))
field.AppendNull()
return
}
field.Append(true)
return
}
valueBldr := field.ValueBuilder()
reader := csv.NewReader(strings.NewReader(str))
items, err := reader.Read()
if err != nil {
r.err = err
r.setParseError(err)
field.AppendNull()
return
}
if len(items) == n {
for _, str := range items {
r.initFieldConverter(valueBldr)(str)
}
} else {
r.err = fmt.Errorf("%w: fixed size list items should match the fixed size list length, expected %d, got %d", arrow.ErrInvalid, n, len(items))
if len(items) != n {
r.setParseError(fmt.Errorf("%w: fixed size list items should match the fixed size list length, expected %d, got %d", arrow.ErrInvalid, n, len(items)))
field.AppendNull()
return
}

field.Append(true)
for _, str := range items {
r.initFieldConverter(valueBldr)(str)
}
}

Expand All @@ -873,7 +892,7 @@ func (r *Reader) parseBinaryType(field array.Builder, str string) {
}
decodedVal, err := base64.StdEncoding.DecodeString(str)
if err != nil {
r.err = fmt.Errorf("cannot decode base64 string %s", str)
r.setParseError(fmt.Errorf("cannot decode base64 string %s", str))
field.AppendNull()
return
}
Expand All @@ -889,7 +908,7 @@ func (r *Reader) parseLargeBinaryType(field array.Builder, str string) {
}
decodedVal, err := base64.StdEncoding.DecodeString(str)
if err != nil {
r.err = fmt.Errorf("cannot decode base64 string %s", str)
r.setParseError(fmt.Errorf("cannot decode base64 string %s", str))
field.AppendNull()
return
}
Expand All @@ -905,15 +924,16 @@ func (r *Reader) parseFixedSizeBinaryType(field array.Builder, str string, byteW
}
decodedVal, err := base64.StdEncoding.DecodeString(str)
if err != nil {
r.err = fmt.Errorf("cannot decode base64 string %s", str)
r.setParseError(fmt.Errorf("cannot decode base64 string %s", str))
field.AppendNull()
return
}

if len(decodedVal) == byteWidth {
field.(*array.FixedSizeBinaryBuilder).Append(decodedVal)
} else {
r.err = fmt.Errorf("%w: the length of fixed size binary value should match the fixed size binary byte width, expected %d, got %d", arrow.ErrInvalid, byteWidth, len(decodedVal))
r.setParseError(fmt.Errorf("%w: the length of fixed size binary value should match the fixed size binary byte width, expected %d, got %d", arrow.ErrInvalid, byteWidth, len(decodedVal)))
field.AppendNull()
}
}

Expand All @@ -923,7 +943,8 @@ func (r *Reader) parseExtension(field array.Builder, str string) {
return
}
if err := field.AppendValueFromString(str); err != nil {
r.err = err
r.setParseError(err)
field.AppendNull()
return
}
}
Expand Down
Loading
Loading