From 84201b6dbe97e510191ae7a74f5e62eafd7441a5 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 8 Aug 2026 17:34:32 +0200 Subject: [PATCH 1/3] fix(arrow/csv): append null for later parse failures --- arrow/csv/reader.go | 76 ++++++++++++++++++++++------------------ arrow/csv/reader_test.go | 30 ++++++++++++++++ 2 files changed, 71 insertions(+), 35 deletions(-) diff --git a/arrow/csv/reader.go b/arrow/csv/reader.go index 0d4ee0f3..56875c57 100644 --- a/arrow/csv/reader.go +++ b/arrow/csv/reader.go @@ -385,6 +385,12 @@ func (r *Reader) isNull(val string) bool { return false } +func (r *Reader) setParseError(err error) { + if r.err == nil { + r.err = err + } +} + func (r *Reader) read(recs []string) { for i, str := range recs { r.fieldConverter[i](str) @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } @@ -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 } diff --git a/arrow/csv/reader_test.go b/arrow/csv/reader_test.go index 80c78482..211da126 100644 --- a/arrow/csv/reader_test.go +++ b/arrow/csv/reader_test.go @@ -901,6 +901,36 @@ func benchRead(b *testing.B, raw []byte, rows, cols, chunks int) { } } +func TestCSVReaderAppendsNullAfterPreviousParseError(t *testing.T) { + schema := arrow.NewSchema([]arrow.Field{ + {Name: "int8", Type: arrow.PrimitiveTypes.Int8}, + {Name: "int16", Type: arrow.PrimitiveTypes.Int16}, + {Name: "int32", Type: arrow.PrimitiveTypes.Int32}, + {Name: "int64", Type: arrow.PrimitiveTypes.Int64}, + {Name: "uint8", Type: arrow.PrimitiveTypes.Uint8}, + {Name: "uint16", Type: arrow.PrimitiveTypes.Uint16}, + {Name: "uint32", Type: arrow.PrimitiveTypes.Uint32}, + {Name: "uint64", Type: arrow.PrimitiveTypes.Uint64}, + {Name: "float16", Type: arrow.FixedWidthTypes.Float16}, + {Name: "float32", Type: arrow.PrimitiveTypes.Float32}, + {Name: "float64", Type: arrow.PrimitiveTypes.Float64}, + {Name: "timestamp", Type: arrow.FixedWidthTypes.Timestamp_ms}, + {Name: "date32", Type: arrow.PrimitiveTypes.Date32}, + {Name: "date64", Type: arrow.PrimitiveTypes.Date64}, + {Name: "decimal128", Type: &arrow.Decimal128Type{Precision: 10, Scale: 2}}, + {Name: "decimal256", Type: &arrow.Decimal256Type{Precision: 10, Scale: 2}}, + }, nil) + + r := csv.NewReader(strings.NewReader(strings.Repeat("bad,", schema.NumFields()-1)+"bad\n"), schema) + defer r.Release() + + require.True(t, r.Next()) + require.Error(t, r.Err()) + for i, col := range r.RecordBatch().Columns() { + require.Truef(t, col.IsNull(0), "column %d (%s) should be null", i, col.DataType()) + } +} + func TestInferringSchema(t *testing.T) { var b bytes.Buffer wr := stdcsv.NewWriter(&b) From 2eb2f10f972eed78a51f3a3508fab38362c27c64 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Wed, 12 Aug 2026 00:03:23 +0200 Subject: [PATCH 2/3] test(arrow/csv): handle uint64 example column --- arrow/csv/example_csv_test.go | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/arrow/csv/example_csv_test.go b/arrow/csv/example_csv_test.go index 268e6325..6e642892 100644 --- a/arrow/csv/example_csv_test.go +++ b/arrow/csv/example_csv_test.go @@ -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}, @@ -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() @@ -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++ { @@ -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 @@ -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 } From 5971924c4a4f9bc0385c86d836437a10f3ba0cf0 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Wed, 12 Aug 2026 22:42:35 +0200 Subject: [PATCH 3/3] fix(arrow/csv): keep builders aligned after parse errors --- arrow/csv/reader.go | 51 ++++++++++++++++++++++++++-------------- arrow/csv/reader_test.go | 34 +++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 18 deletions(-) diff --git a/arrow/csv/reader.go b/arrow/csv/reader.go index 56875c57..57f43821 100644 --- a/arrow/csv/reader.go +++ b/arrow/csv/reader.go @@ -817,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) } @@ -845,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) } } @@ -879,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 } @@ -895,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 } @@ -911,7 +924,7 @@ 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 } @@ -919,7 +932,8 @@ func (r *Reader) parseFixedSizeBinaryType(field array.Builder, str string, byteW 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() } } @@ -929,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 } } diff --git a/arrow/csv/reader_test.go b/arrow/csv/reader_test.go index 211da126..efa734ca 100644 --- a/arrow/csv/reader_test.go +++ b/arrow/csv/reader_test.go @@ -904,6 +904,7 @@ func benchRead(b *testing.B, raw []byte, rows, cols, chunks int) { func TestCSVReaderAppendsNullAfterPreviousParseError(t *testing.T) { schema := arrow.NewSchema([]arrow.Field{ {Name: "int8", Type: arrow.PrimitiveTypes.Int8}, + {Name: "bool", Type: arrow.FixedWidthTypes.Boolean}, {Name: "int16", Type: arrow.PrimitiveTypes.Int16}, {Name: "int32", Type: arrow.PrimitiveTypes.Int32}, {Name: "int64", Type: arrow.PrimitiveTypes.Int64}, @@ -931,6 +932,39 @@ func TestCSVReaderAppendsNullAfterPreviousParseError(t *testing.T) { } } +func TestCSVReaderAppendsNullAfterCompositeParseError(t *testing.T) { + tests := []struct { + name string + typ arrow.DataType + value string + }{ + {name: "list format", typ: arrow.ListOf(arrow.PrimitiveTypes.Int8), value: "bad"}, + {name: "fixed size list format", typ: arrow.FixedSizeListOf(2, arrow.PrimitiveTypes.Int8), value: "bad"}, + {name: "fixed size list length", typ: arrow.FixedSizeListOf(2, arrow.PrimitiveTypes.Int8), value: "{1}"}, + {name: "binary", typ: arrow.BinaryTypes.Binary, value: "%%%"}, + {name: "large binary", typ: arrow.BinaryTypes.LargeBinary, value: "%%%"}, + {name: "fixed size binary format", typ: &arrow.FixedSizeBinaryType{ByteWidth: 3}, value: "%%%"}, + {name: "fixed size binary", typ: &arrow.FixedSizeBinaryType{ByteWidth: 3}, value: "AQ=="}, + {name: "extension", typ: extensions.NewUUIDType(), value: "bad"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + schema := arrow.NewSchema([]arrow.Field{ + {Name: "int8", Type: arrow.PrimitiveTypes.Int8}, + {Name: "value", Type: tc.typ}, + }, nil) + r := csv.NewReader(strings.NewReader("bad;"+tc.value+"\n"), schema, csv.WithComma(';')) + defer r.Release() + + require.True(t, r.Next()) + require.ErrorContains(t, r.Err(), "strconv.ParseInt") + require.True(t, r.RecordBatch().Column(0).IsNull(0)) + require.True(t, r.RecordBatch().Column(1).IsNull(0)) + }) + } +} + func TestInferringSchema(t *testing.T) { var b bytes.Buffer wr := stdcsv.NewWriter(&b)