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/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new 1e0b757e fix(arrow/csv): append null for later parse failures (#1149)
1e0b757e is described below
commit 1e0b757e146da4fb08c29fb6f62c0104d38af4fa
Author: Minh Vu <[email protected]>
AuthorDate: Fri Aug 14 18:10:47 2026 +0200
fix(arrow/csv): append null for later parse failures (#1149)
## What
CSV scalar parsers already append a null when their own parse fails, but
several only did so while the reader had no earlier error. A malformed
value in a later column could therefore append the parser zero value
after another column failed. This keeps the first error and appends null
for every failed scalar conversion.
## Test
- go test ./arrow/csv -run
TestCSVReaderAppendsNullAfterPreviousParseError -count=1
---
arrow/csv/example_csv_test.go | 17 +++++-
arrow/csv/reader.go | 127 ++++++++++++++++++++++++------------------
arrow/csv/reader_test.go | 64 +++++++++++++++++++++
3 files changed, 152 insertions(+), 56 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
}
diff --git a/arrow/csv/reader.go b/arrow/csv/reader.go
index 9387f23d..f2a1f0b2 100644
--- a/arrow/csv/reader.go
+++ b/arrow/csv/reader.go
@@ -404,6 +404,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)
@@ -551,7 +557,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
}
@@ -566,8 +572,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
}
@@ -582,8 +588,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
}
@@ -598,8 +604,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
}
@@ -614,8 +620,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
}
@@ -630,8 +636,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
}
@@ -646,8 +652,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
}
@@ -662,8 +668,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
}
@@ -678,8 +684,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
}
@@ -694,8 +700,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
}
@@ -709,8 +715,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
}
@@ -724,8 +730,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
}
@@ -740,8 +746,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
}
@@ -756,8 +762,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
}
@@ -771,8 +777,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
}
@@ -786,8 +792,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
}
@@ -801,8 +807,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
}
@@ -816,8 +822,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
}
@@ -830,23 +836,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)
}
@@ -858,29 +867,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)
}
}
@@ -892,7 +911,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
}
@@ -908,7 +927,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
}
@@ -924,7 +943,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
}
@@ -932,7 +951,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()
}
}
@@ -942,7 +962,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 c7cd61f4..8623f70b 100644
--- a/arrow/csv/reader_test.go
+++ b/arrow/csv/reader_test.go
@@ -901,6 +901,70 @@ 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},
+ {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 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)
