Skip to content

fix: preserve external CSV integer values and whitespace records #30262

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
mergify merged 9 commits into from
Oct 19, 2026
Merged

fix: preserve external CSV integer values and whitespace records #30262

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
1e64ad9
fix: preserve external CSV integer assignment and whitespace records
iamlinjunhong Oct 15, 2026
b3e6fa9
fix: scale external CSV decimals before integer range checks
iamlinjunhong Oct 15, 2026
2ca34a1
fix(external): preserve JSONLINE whitespace record skipping
iamlinjunhong Oct 15, 2026
feb9fb9
fix(load): share JSONLINE whitespace framing with pre-scan
XuPeng-SH Oct 18, 2026
3e43b13
Merge branch 'main' into fix/csv-external-integer-whitespace
XuPeng-SH Oct 18, 2026
4f10c03
Merge branch 'main' into fix/csv-external-integer-whitespace
mergify[bot] Oct 19, 2026
f435d6b
Merge branch 'main' into fix/csv-external-integer-whitespace
mergify[bot] Oct 19, 2026
d4ceca3
test(csvparser): preserve record lines under whitespace policies
iamlinjunhong Oct 19, 2026
9b7c74c
Merge branch 'main' into fix/csv-external-integer-whitespace
mergify[bot] Oct 19, 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
85 changes: 85 additions & 0 deletions pkg/sql/colexec/external/error_mode_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,17 @@ func TestCSVErrorModeTolerates(t *testing.T) {
})
require.Equal(t, "/tmp/rows.dat", string(bat.Vecs[colPath].GetBytesAt(1)),
"__mo_filepath is set on a failed row too")
t.Run("rounded overflow rolls back partial row", func(t *testing.T) {
param, proc, bat := errorModeParam(t, tree.CSV, "", numTestCols)
// Materialize the string column first so a numeric failure must roll it back.
param.Attrs[0], param.Attrs[1] = param.Attrs[1], param.Attrs[0]
require.NoError(t, readAllText(t, param, proc, bat, "42949672960e-10,alpha\n2147483647.5,beta\n25000000000e-10,gamma\n"))
requireRows(t, bat, []wantRow{
{a: 4, s: "alpha", line: 1},
{line: 2, message: "is not int32 type", text: "2147483647.5,beta"},
{a: 3, s: "gamma", line: 3},
})
})
}

// TestCSVWithoutErrorColumnsStillFails: with the error columns pruned the scan
Expand Down Expand Up @@ -307,6 +318,80 @@ func TestJSONLineWithoutErrorColumnsStillFails(t *testing.T) {
require.Contains(t, err.Error(), "not a well-formed json object")
}

func TestJSONLineWhitespaceRecords(t *testing.T) {
for _, shape := range []struct {
name, jsonData, first, second string
}{
{"object", tree.OBJECT, `{"a":1,"s":"alpha"}`, `{"a":2,"s":"beta"}`},
{"array", tree.ARRAY, `[1,"alpha"]`, `[2,"beta"]`},
} {
for _, keep := range []int{colMsg, numTestCols} {
name := "ordinary"
if keep == numTestCols {
name = "tolerant"
}
t.Run(shape.name+"/"+name, func(t *testing.T) {
param, proc, bat := errorModeParam(t, tree.JSONLINE, shape.jsonData, keep)
content := " \r\n" + shape.first + "\r\n \r\n\r\n" + shape.second + "\r\n \n"
require.NoError(t, readAllText(t, param, proc, bat, content))
require.Equal(t, 2, bat.RowCount())
for i, line := range []int64{2, 5} {
require.False(t, bat.Vecs[colA].IsNull(uint64(i)))
require.Equal(t, int32(i+1), vector.GetFixedAtWithTypeCheck[int32](bat.Vecs[colA], i))
require.Equal(t, []string{"alpha", "beta"}[i], string(bat.Vecs[colS].GetBytesAt(i)))
require.Equal(t, line, vector.GetFixedAtWithTypeCheck[int64](bat.Vecs[colLine], i))
if keep == numTestCols {
require.True(t, bat.Vecs[colMsg].IsNull(uint64(i)))
require.True(t, bat.Vecs[colText].IsNull(uint64(i)))
}
}
})
}
}

// A quoted empty scalar and a tab-delimited malformed record must still
// reach JSON validation, even though the first decoded field is empty.
for _, content := range []string{"\"\"\n", "\" \"\n", "\tbad\n"} {
t.Run("invalid/"+content, func(t *testing.T) {
param, proc, bat := errorModeParam(t, tree.JSONLINE, tree.OBJECT, colMsg)
require.Error(t, readAllText(t, param, proc, bat, content))
param, proc, bat = errorModeParam(t, tree.JSONLINE, tree.OBJECT, numTestCols)
require.NoError(t, readAllText(t, param, proc, bat, content))
require.Equal(t, 1, bat.RowCount())
require.True(t, bat.Vecs[colA].IsNull(0))
require.False(t, bat.Vecs[colMsg].IsNull(0))
})
}

for _, tc := range []struct {
name, content string
prevStr string
ignore int
maxBytes uint64
want []wantRow
}{
{"ignored records", " \n{\"a\":1,\"s\":\"alpha\"}\n \n{\"a\":2,\"s\":\"beta\"}\n", "", 1, 1 << 20, []wantRow{{a: 2, s: "beta", line: 4}}},
{"continuation", " \n123}\n", `{"a":1,"s":`, 0, 1 << 20, []wantRow{{a: 1, s: "123", line: 2}}},
{"batch bytes", strings.Repeat(" ", 80) + "\n{\"a\":1,\"s\":\"alpha\"}\n{\"a\":2,\"s\":\"beta\"}\n", "", 0, 40, []wantRow{{a: 1, s: "alpha", line: 2}, {a: 2, s: "beta", line: 3}}},
} {
t.Run(tc.name, func(t *testing.T) {
param, proc, bat := errorModeParam(t, tree.JSONLINE, tree.OBJECT, numTestCols)
param.maxBatchSize = tc.maxBytes
if tc.name == "continuation" {
param.ErrorMode.Tolerate = false
}
r := &CsvReader{param: param, prevStr: tc.prevStr, ignoreLine: tc.ignore, reader: io.NopCloser(strings.NewReader(tc.content))}
t.Cleanup(func() { require.NoError(t, r.Close()) })
parser, err := newCSVParserFromReader(param.Extern, r.reader)
require.NoError(t, err)
r.plh = &ParseLineHandler{csvReader: parser}
_, err = r.makeBatchRows(proc, bat)
require.NoError(t, err)
requireRows(t, bat, tc.want)
})
}
}

// kafkaErrorModeParam declares `(a int, s varchar)` plus __mo_message_id, the
// three error-mode columns and __mo_message_value.
func kafkaErrorModeParam(t *testing.T, ks *plan.KafkaScan) (*ExternalParam, *process.Process, *batch.Batch) {
Expand Down
72 changes: 56 additions & 16 deletions pkg/sql/colexec/external/external.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1889,6 +1889,45 @@ func loadTimeTextInput(param *ExternalParam, id types.T) bool {
param.Extern.Format != tree.PARQUET && param.Extern.Format != tree.ARROW
}

// parseCSVInt preserves the integer fast path and LOAD conversion policy. Only
// unquoted external CSV decimal text uses exact assignment rounding.
// An initial integer overflow can precede an exponent that scales into range;
// validate the complete decimal value before enforcing the destination width.
func parseCSVInt(field csvparser.Field, bits int, round bool) (int64, error) {
value, err := strconv.ParseInt(field.Val, 10, bits)
if err == nil || !round || field.HasStringQuote {
return value, err
}
integer, err := function.ParseDecimalInteger(field.Val)
if err != nil {
return 0, err
}
return strconv.ParseInt(integer, 10, bits)
}

func parseCSVUint(field csvparser.Field, bits int, round bool) (uint64, error) {
value, err := strconv.ParseUint(field.Val, 10, bits)
if err == nil || !round || field.HasStringQuote {
return value, err
}
integer, err := function.ParseDecimalInteger(field.Val)
if err != nil {
return 0, err
}
// Keep negative zero accepted, while negative fractions that round to zero
// still fail unsigned conversion as before.
if integer == "-0" {
mantissa := field.Val
if exponent := strings.IndexAny(mantissa, "eE"); exponent >= 0 {
mantissa = mantissa[:exponent]
}
if strings.Trim(mantissa, "-0.") == "" {
integer = "0"
}
}
return strconv.ParseUint(integer, 10, bits)
}

func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *ExternalParam, mp *mpool.MPool, attr plan.ExternAttr, proc *process.Process) error {
colIdx := attr.ColIndex
colName := attr.ColName
Expand Down Expand Up @@ -1974,6 +2013,7 @@ func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *Ext
return nil
}

roundExternalInteger := param.Extern.ExternType == int32(plan.ExternType_EXTERNAL_TB) && param.Extern.Format == tree.CSV
switch id {
case types.T_bool:
b, err := types.ParseBool(field.Val)
Expand Down Expand Up @@ -2001,14 +2041,14 @@ func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *Ext
return err
}
case types.T_int8:
d, err := strconv.ParseInt(field.Val, 10, 8)
d, err := parseCSVInt(field, 8, roundExternalInteger)
if err == nil {
if err := vector.AppendFixed(vec, int8(d), false, mp); err != nil {
return err
}
} else {
// if field.HasStringQuote is true, like load "1.9" into int type, return error, if load 1.9, can load successful
if errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
if roundExternalInteger || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
logutil.Errorf("parse field[%v] err:%v", field.Val, err)
return moerr.NewInternalErrorf(param.Ctx, "the input value '%v' is not int8 type for column %d", field.Val, colIdx)
}
Expand All @@ -2022,13 +2062,13 @@ func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *Ext
}
}
case types.T_int16:
d, err := strconv.ParseInt(field.Val, 10, 16)
d, err := parseCSVInt(field, 16, roundExternalInteger)
if err == nil {
if err := vector.AppendFixed(vec, int16(d), false, mp); err != nil {
return err
}
} else {
if errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
if roundExternalInteger || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
logutil.Errorf("parse field[%v] err:%v", field.Val, err)
return moerr.NewInternalErrorf(param.Ctx, "the input value '%v' is not int16 type for column %d", field.Val, colIdx)
}
Expand All @@ -2050,13 +2090,13 @@ func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *Ext
min, max = -(1 << 23), (1<<23)-1
valueType = "mediumint"
}
d, err := strconv.ParseInt(field.Val, 10, bits)
d, err := parseCSVInt(field, bits, roundExternalInteger)
if err == nil {
if err := vector.AppendFixed(vec, int32(d), false, mp); err != nil {
return err
}
} else {
if errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
if roundExternalInteger || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
logutil.Errorf("parse field[%v] err:%v", field.Val, err)
return moerr.NewInternalErrorf(param.Ctx, "the input value '%v' is not %s type for column %d", field.Val, valueType, colIdx)
}
Expand All @@ -2070,13 +2110,13 @@ func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *Ext
}
}
case types.T_int64:
d, err := strconv.ParseInt(field.Val, 10, 64)
d, err := parseCSVInt(field, 64, roundExternalInteger)
if err == nil {
if err := vector.AppendFixed(vec, d, false, mp); err != nil {
return err
}
} else {
if errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
if roundExternalInteger || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
logutil.Errorf("parse field[%v] err:%v", field.Val, err)
return moerr.NewInternalErrorf(param.Ctx, "the input value '%v' is not int64 type for column %d", field.Val, colIdx)
}
Expand All @@ -2090,13 +2130,13 @@ func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *Ext
}
}
case types.T_uint8:
d, err := strconv.ParseUint(field.Val, 10, 8)
d, err := parseCSVUint(field, 8, roundExternalInteger)
if err == nil {
if err := vector.AppendFixed(vec, uint8(d), false, mp); err != nil {
return err
}
} else {
if errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
if roundExternalInteger || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
logutil.Errorf("parse field[%v] err:%v", field.Val, err)
return moerr.NewInternalErrorf(param.Ctx, "the input value '%v' is not uint8 type for column %d", field.Val, colIdx)
}
Expand All @@ -2110,13 +2150,13 @@ func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *Ext
}
}
case types.T_uint16:
d, err := strconv.ParseUint(field.Val, 10, 16)
d, err := parseCSVUint(field, 16, roundExternalInteger)
if err == nil {
if err := vector.AppendFixed(vec, uint16(d), false, mp); err != nil {
return err
}
} else {
if errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
if roundExternalInteger || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
logutil.Errorf("parse field[%v] err:%v", field.Val, err)
return moerr.NewInternalErrorf(param.Ctx, "the input value '%v' is not uint16 type for column %d", field.Val, colIdx)
}
Expand All @@ -2138,13 +2178,13 @@ func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *Ext
max = (1 << 24) - 1
valueType = "mediumint unsigned"
}
d, err := strconv.ParseUint(field.Val, 10, bits)
d, err := parseCSVUint(field, bits, roundExternalInteger)
if err == nil {
if err := vector.AppendFixed(vec, uint32(d), false, mp); err != nil {
return err
}
} else {
if errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
if roundExternalInteger || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
logutil.Errorf("parse field[%v] err:%v", field.Val, err)
return moerr.NewInternalErrorf(param.Ctx, "the input value '%v' is not %s type for column %d", field.Val, valueType, colIdx)
}
Expand All @@ -2169,13 +2209,13 @@ func getColData(bat *batch.Batch, line []csvparser.Field, rowIdx int, param *Ext
return err
}
} else {
d, err := strconv.ParseUint(field.Val, 10, 64)
d, err := parseCSVUint(field, 64, roundExternalInteger)
if err == nil {
if err := vector.AppendFixed(vec, d, false, mp); err != nil {
return err
}
} else {
if errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
if roundExternalInteger || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
logutil.Errorf("parse field[%v] err:%v", field.Val, err)
return moerr.NewInternalErrorf(param.Ctx, "the input value '%v' is not uint64 type for column %d", field.Val, colIdx)
}
Expand Down
Loading