From 1e64ad9c7accbf7ed7023682ebd438147779309a Mon Sep 17 00:00:00 2001
From: iamlinjunhong <1030420200@qq.com>
Date: Thu, 15 Oct 2026 14:40:25 +0800
Subject: [PATCH 1/5] fix: preserve external CSV integer assignment and
 whitespace records

---
 pkg/sql/colexec/external/error_mode_test.go   |  11 ++
 pkg/sql/colexec/external/external.go          |  70 +++++++--
 pkg/sql/colexec/external/external_test.go     | 148 ++++++++++++++++++
 pkg/sql/plan/function/func_cast.go            |  12 ++
 .../func_cast_integer_assignment_test.go      |  30 ++++
 pkg/sql/util/csvparser/csv_parser.go          |  19 ++-
 pkg/sql/util/csvparser/csv_parser_test.go     |  61 +++++++-
 .../load_data/external_csv_values.result      |  56 +++++++
 .../cases/load_data/external_csv_values.sql   |  41 +++++
 .../csv_integer_assignment.csv                |   1 +
 .../csv_integer_overflow.csv                  |   1 +
 .../csv_whitespace_records.csv                |   8 +
 12 files changed, 428 insertions(+), 30 deletions(-)
 create mode 100644 test/distributed/cases/load_data/external_csv_values.result
 create mode 100644 test/distributed/cases/load_data/external_csv_values.sql
 create mode 100644 test/distributed/resources/external_table_file/csv_integer_assignment.csv
 create mode 100644 test/distributed/resources/external_table_file/csv_integer_overflow.csv
 create mode 100644 test/distributed/resources/external_table_file/csv_whitespace_records.csv

diff --git a/pkg/sql/colexec/external/error_mode_test.go b/pkg/sql/colexec/external/error_mode_test.go
index fb9533b1e5959..c6ecc86ff79bc 100644
--- a/pkg/sql/colexec/external/error_mode_test.go
+++ b/pkg/sql/colexec/external/error_mode_test.go
@@ -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, "1,alpha\n2147483647.5,beta\n2.5,gamma\n"))
+		requireRows(t, bat, []wantRow{
+			{a: 1, 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
diff --git a/pkg/sql/colexec/external/external.go b/pkg/sql/colexec/external/external.go
index 62fcd1d5b8a05..458dbfdfe4b32 100644
--- a/pkg/sql/colexec/external/external.go
+++ b/pkg/sql/colexec/external/external.go
@@ -1889,6 +1889,43 @@ 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.
+func parseCSVInt(field csvparser.Field, bits int, round bool) (int64, error) {
+	value, err := strconv.ParseInt(field.Val, 10, bits)
+	if err == nil || !round || errors.Is(err, strconv.ErrRange) || 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 || errors.Is(err, strconv.ErrRange) || 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
@@ -1974,6 +2011,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)
@@ -2001,14 +2039,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)
 			}
@@ -2022,13 +2060,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)
 			}
@@ -2050,13 +2088,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)
 			}
@@ -2070,13 +2108,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)
 			}
@@ -2090,13 +2128,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)
 			}
@@ -2110,13 +2148,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)
 			}
@@ -2138,13 +2176,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)
 			}
@@ -2169,13 +2207,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)
 				}
diff --git a/pkg/sql/colexec/external/external_test.go b/pkg/sql/colexec/external/external_test.go
index 7e1a0678de74c..d47185ef0efba 100644
--- a/pkg/sql/colexec/external/external_test.go
+++ b/pkg/sql/colexec/external/external_test.go
@@ -2763,3 +2763,151 @@ func TestExternalOwnerPreservesCloseError(t *testing.T) {
 	require.Nil(t, op.reader)
 	require.False(t, op.fileOpened)
 }
+
+func TestGetColDataExternalCSVIntegerRounding(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	defer proc.Free()
+	for _, typ := range []types.Type{
+		types.T_int8.ToType(), types.T_int16.ToType(), types.T_int32.ToType(), types.T_int64.ToType(),
+		types.T_uint8.ToType(), types.T_uint16.ToType(), types.T_uint32.ToType(), types.T_uint64.ToType(),
+		types.New(types.T_int32, 24, 0), types.New(types.T_uint32, 24, 0),
+	} {
+		name := typ.String()
+		if typ.IsMediumInt() {
+			name += "_mediumint"
+		}
+		t.Run(name, func(t *testing.T) {
+			type testCase struct {
+				value, want     string
+				quoted, wantErr bool
+			}
+			cases := []testCase{
+				{value: "1.9", want: "2"}, {value: "2.4", want: "2"}, {value: "2.5", want: "3"},
+				{value: "1e2", want: "100"}, {value: "2.5e0", want: "3"}, {value: ".5", want: "1"},
+				{value: "1.", want: "1"}, {value: "+1.9", want: "2"}, {value: " 1.9 ", want: "2"},
+				{value: "7", want: "7", quoted: true}, {value: "1.9", quoted: true, wantErr: true},
+				{value: "1e2", quoted: true, wantErr: true}, {value: "NaN", wantErr: true},
+				{value: "+Inf", wantErr: true}, {value: "-Inf", wantErr: true}, {value: "1e+", wantErr: true},
+				{value: "1 2", wantErr: true}, {value: "1+2", wantErr: true}, {value: "1.9tail", wantErr: true},
+				{value: "0x1p2", wantErr: true}, {value: "1_0.5", wantErr: true},
+				{value: "1e999999999999999999", wantErr: true}, {value: "1e-999999999999999999", want: "0"},
+				{value: "0.10000000000000000000000000000000000000000e1", want: "1"},
+			}
+			if typ.Oid.IsUnsignedInt() {
+				cases = append(cases, testCase{value: "-0.1", wantErr: true},
+					testCase{value: "-1.9", wantErr: true},
+					testCase{value: "-0.0", want: "0"},
+					testCase{value: "-0e2", want: "0"})
+			} else {
+				cases = append(cases, testCase{value: "-1.9", want: "-2"},
+					testCase{value: "-2.5", want: "-3"},
+					testCase{value: "-0.49", want: "0"})
+			}
+			var min, max string
+			switch typ.Oid {
+			case types.T_int8:
+				min, max = "-128", "127"
+			case types.T_int16:
+				min, max = "-32768", "32767"
+			case types.T_int32:
+				min, max = "-2147483648", "2147483647"
+			case types.T_int64:
+				min, max = "-9223372036854775808", "9223372036854775807"
+			case types.T_uint8:
+				max = "255"
+			case types.T_uint16:
+				max = "65535"
+			case types.T_uint32:
+				max = "4294967295"
+			case types.T_uint64:
+				max = "18446744073709551615"
+			}
+			if typ.IsMediumInt() {
+				if typ.Oid.IsUnsignedInt() {
+					max = "16777215"
+				} else {
+					min, max = "-8388608", "8388607"
+				}
+			}
+			cases = append(cases, testCase{value: max, want: max},
+				testCase{value: max + ".0", want: max},
+				testCase{value: max + ".4", want: max},
+				testCase{value: max + ".5", wantErr: true})
+			if min != "" {
+				cases = append(cases, testCase{value: min + ".4", want: min},
+					testCase{value: min + ".5", wantErr: true})
+			}
+			if typ.Oid == types.T_int64 || typ.Oid == types.T_uint64 {
+				cases = append(cases,
+					testCase{value: "9007199254740993.0", want: "9007199254740993"})
+			}
+			for _, tc := range cases {
+				t.Run(fmt.Sprintf("%s_quoted_%t", tc.value, tc.quoted), func(t *testing.T) {
+					bat := batch.NewWithSize(1)
+					bat.Vecs[0] = vector.NewVec(typ)
+					defer bat.Clean(proc.Mp())
+					param := &ExternalParam{ExParamConst: ExParamConst{Ctx: context.Background(),
+						Cols: []*plan.ColDef{{Name: "i", Typ: plan.Type{Id: int32(typ.Oid), Width: typ.Width}}},
+						Extern: &tree.ExternParam{ExParamConst: tree.ExParamConst{Format: tree.CSV},
+							ExParam: tree.ExParam{ExternType: int32(plan.ExternType_EXTERNAL_TB)}}}, ExParam: ExParam{Fileparam: &ExFileparam{}}}
+					err := getColData(bat, []csvparser.Field{{Val: tc.value, HasStringQuote: tc.quoted}}, 0, param, proc.Mp(),
+						plan.ExternAttr{ColName: "i", ColIndex: 0, ColFieldIndex: 0}, proc)
+					if tc.wantErr {
+						require.Error(t, err)
+						require.Zero(t, bat.Vecs[0].Length())
+						return
+					}
+					require.NoError(t, err)
+					require.Equal(t, 1, bat.Vecs[0].Length())
+					var got string
+					switch typ.Oid {
+					case types.T_int8:
+						got = fmt.Sprint(vector.GetFixedAtWithTypeCheck[int8](bat.Vecs[0], 0))
+					case types.T_int16:
+						got = fmt.Sprint(vector.GetFixedAtWithTypeCheck[int16](bat.Vecs[0], 0))
+					case types.T_int32:
+						got = fmt.Sprint(vector.GetFixedAtWithTypeCheck[int32](bat.Vecs[0], 0))
+					case types.T_int64:
+						got = fmt.Sprint(vector.GetFixedAtWithTypeCheck[int64](bat.Vecs[0], 0))
+					case types.T_uint8:
+						got = fmt.Sprint(vector.GetFixedAtWithTypeCheck[uint8](bat.Vecs[0], 0))
+					case types.T_uint16:
+						got = fmt.Sprint(vector.GetFixedAtWithTypeCheck[uint16](bat.Vecs[0], 0))
+					case types.T_uint32:
+						got = fmt.Sprint(vector.GetFixedAtWithTypeCheck[uint32](bat.Vecs[0], 0))
+					case types.T_uint64:
+						got = fmt.Sprint(vector.GetFixedAtWithTypeCheck[uint64](bat.Vecs[0], 0))
+					}
+					require.Equal(t, tc.want, got)
+				})
+			}
+		})
+	}
+}
+
+func TestGetColDataIntegerRoundingScope(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	defer proc.Free()
+	for _, tc := range []struct {
+		name, format      string
+		externType        int32
+		enumValues, value string
+		want              uint64
+	}{
+		{"load csv", tree.CSV, int32(plan.ExternType_LOAD), "", "1.9", 1},
+		{"external jsonline", tree.JSONLINE, int32(plan.ExternType_EXTERNAL_TB), "", "1.9", 1},
+		{"external set", tree.CSV, int32(plan.ExternType_EXTERNAL_TB), "1.9,two", "1.9", 1},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			bat := batch.NewWithSize(1)
+			bat.Vecs[0] = vector.NewVec(types.T_uint64.ToType())
+			defer bat.Clean(proc.Mp())
+			param := &ExternalParam{ExParamConst: ExParamConst{Ctx: context.Background(), StrictSqlMode: true,
+				Cols:   []*plan.ColDef{{Name: "i", Typ: plan.Type{Id: int32(types.T_uint64), Enumvalues: tc.enumValues}}},
+				Extern: &tree.ExternParam{ExParamConst: tree.ExParamConst{Format: tc.format}, ExParam: tree.ExParam{ExternType: tc.externType}}}, ExParam: ExParam{Fileparam: &ExFileparam{}}}
+			err := getColData(bat, []csvparser.Field{{Val: tc.value}}, 0, param, proc.Mp(), plan.ExternAttr{ColName: "i"}, proc)
+			require.NoError(t, err)
+			require.Equal(t, []uint64{tc.want}, vector.MustFixedColWithTypeCheck[uint64](bat.Vecs[0]))
+		})
+	}
+}
diff --git a/pkg/sql/plan/function/func_cast.go b/pkg/sql/plan/function/func_cast.go
index b69233b1f7b1d..ff39e0f0efd93 100644
--- a/pkg/sql/plan/function/func_cast.go
+++ b/pkg/sql/plan/function/func_cast.go
@@ -7297,6 +7297,18 @@ func appendTemporalAssignmentConversionWarning(proc *process.Process, targetType
 	)
 }
 
+// ParseDecimalInteger rounds decimal text to an integer using assignment
+// semantics. Unlike assignmentIntegerPrefix, it requires the whole value to be
+// numeric. The shared parser keeps precision beyond float64 and bounds work by
+// input length, even for large exponents.
+func ParseDecimalInteger(value string) (string, error) {
+	integer, hasPrefix, truncated, err := assignmentIntegerPrefix(value)
+	if !hasPrefix || truncated {
+		return "", strconv.ErrSyntax
+	}
+	return integer, err
+}
+
 // assignmentIntegerPrefix returns the exact decimal prefix rounded to an
 // integer using assignment semantics and whether a non-whitespace suffix is
 // discarded. It is intentionally separate from leadingDecimalIntegerPrefix,
diff --git a/pkg/sql/plan/function/func_cast_integer_assignment_test.go b/pkg/sql/plan/function/func_cast_integer_assignment_test.go
index b9ccfa2738444..4ae02969939d3 100644
--- a/pkg/sql/plan/function/func_cast_integer_assignment_test.go
+++ b/pkg/sql/plan/function/func_cast_integer_assignment_test.go
@@ -306,3 +306,33 @@ func TestMediumIntAssignmentBounds(t *testing.T) {
 		}
 	})
 }
+
+func TestParseDecimalInteger(t *testing.T) {
+	for _, tc := range []struct {
+		input, want string
+		wantErr     bool
+	}{
+		{input: "1.9", want: "2"}, {input: "-1.9", want: "-2"}, {input: "2.5", want: "3"},
+		{input: "2.5e0", want: "3"}, {input: "-0.1", want: "-0"}, {input: ".5", want: "1"},
+		{input: "1e2", want: "100"}, {input: "9007199254740993.0", want: "9007199254740993"},
+		{input: "18446744073709551615.4", want: "18446744073709551615"},
+		{input: "18446744073709551615.5", wantErr: true},
+		{input: "10000000000000000000000000000000000000000e-40", want: "1"},
+		{input: "1e-99999999999999999", want: "0"}, {input: "0e99999999999999999", want: "0"},
+		{input: "1e99999999999999999", wantErr: true},
+		{input: "", wantErr: true}, {input: "abc", wantErr: true}, {input: "1x", wantErr: true},
+		{input: "1e", wantErr: true}, {input: "1e+", wantErr: true}, {input: "1 2", wantErr: true},
+		{input: "1+2", wantErr: true}, {input: "NaN", wantErr: true}, {input: "Inf", wantErr: true},
+		{input: "0x1p2", wantErr: true}, {input: "1_0.5", wantErr: true},
+	} {
+		t.Run(tc.input, func(t *testing.T) {
+			got, err := ParseDecimalInteger(tc.input)
+			if tc.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.want, got)
+		})
+	}
+}
diff --git a/pkg/sql/util/csvparser/csv_parser.go b/pkg/sql/util/csvparser/csv_parser.go
index 68da4beb51c0d..5c195e7aa927a 100644
--- a/pkg/sql/util/csvparser/csv_parser.go
+++ b/pkg/sql/util/csvparser/csv_parser.go
@@ -608,7 +608,6 @@ func (parser *CSVParser) readRecord() error {
 	parser.fieldIsQuoted = parser.fieldIsQuoted[:0]
 
 	isEmptyLine := true
-	whitespaceLine := true
 	foundStartingByThisLine := false
 	prevToken := csvTokenNewLine
 	fieldIsQuoted := false
@@ -695,14 +694,12 @@ outside:
 
 		switch firstToken {
 		case csvTokenComma:
-			whitespaceLine = false
 			parser.fieldIndexes = append(parser.fieldIndexes, len(parser.recordBuffer))
 			parser.fieldIsQuoted = append(parser.fieldIsQuoted, fieldIsQuoted)
 			fieldIsQuoted = false
 		case csvTokenDelimiter:
 			if prevToken != csvTokenComma && prevToken != csvTokenNewLine {
 				if parser.unescapedQuote {
-					whitespaceLine = false
 					parser.recordBuffer = append(parser.recordBuffer, parser.quote...)
 					continue
 				}
@@ -712,11 +709,19 @@ outside:
 				return err
 			}
 			fieldIsQuoted = true
-			whitespaceLine = false
 		case csvTokenNewLine:
 			foundStartingByThisLine = false
 			// new line = end of field (ignore empty lines)
 			prevToken = firstToken
+			// An explicit LF terminator also accepts CRLF. A lone unquoted CR
+			// before that terminator is an empty record, not field data. Keep
+			// quoted CR and a CR at EOF as data.
+			if err == nil && bytes.Equal(parser.newLine, newlineByte) &&
+				len(parser.fieldIndexes) == 0 && !fieldIsQuoted &&
+				len(parser.recordBuffer) == 1 && parser.recordBuffer[0] == '\r' {
+				parser.recordBuffer = parser.recordBuffer[:0]
+				isEmptyLine = true
+			}
 			if !parser.allowEmptyLine {
 				if isEmptyLine {
 					// The record has not started yet: it begins at the first
@@ -724,12 +729,6 @@ outside:
 					parser.recordStartLine = parser.lineNo
 					continue
 				}
-				// skip lines only contain whitespaces
-				if err == nil && whitespaceLine && len(bytes.TrimSpace(parser.recordBuffer)) == 0 {
-					parser.recordBuffer = parser.recordBuffer[:0]
-					parser.recordStartLine = parser.lineNo
-					continue
-				}
 			}
 			if bytes.Equal(parser.newLine, []byte{'\n'}) {
 				if n := len(parser.recordBuffer); n > 1 && parser.recordBuffer[n-1] == '\r' {
diff --git a/pkg/sql/util/csvparser/csv_parser_test.go b/pkg/sql/util/csvparser/csv_parser_test.go
index afd626a90e170..834ae43370588 100644
--- a/pkg/sql/util/csvparser/csv_parser_test.go
+++ b/pkg/sql/util/csvparser/csv_parser_test.go
@@ -676,14 +676,16 @@ func TestRecordLine(t *testing.T) {
 
 	t.Run("skipped blank lines still count", func(t *testing.T) {
 		// line 1: a,1   lines 2-3 blank   line 4: b,2   line 5 whitespace
-		// line 6: c,3 -- a record starts where its content starts, not where
-		// the scan for it started.
+		// line 6: c,3. Whitespace is field data; only lines 2-3 are skipped.
 		parser, err := NewCSVParser(&cfg, NewStringReader("a,1\n\n\nb,2\n   \nc,3\n"), int64(ReadBlockSize), false)
 		require.NoError(t, err)
-		for _, want := range []int64{1, 4, 6} {
-			_, err = parser.Read(nil)
+		for _, want := range []int64{1, 4, 5, 6} {
+			row, err := parser.Read(nil)
 			require.NoError(t, err)
 			require.Equal(t, want, parser.RecordLine())
+			if want == 5 {
+				require.Equal(t, []Field{newStringField("   ", false, false)}, row)
+			}
 		}
 	})
 
@@ -884,3 +886,54 @@ func TestTrailingCRBelongsToSealedField(t *testing.T) {
 		require.ErrorIs(t, err, io.EOF)
 	})
 }
+
+func TestWhitespaceOnlyRecord(t *testing.T) {
+	for _, tc := range []struct {
+		name, input, terminator, comment, startingBy string
+		allowEmpty                                   bool
+		rows                                         [][]Field
+		lines                                        []int64
+	}{
+		{name: "lf", input: "\n \n\t\n \t \nx\n", rows: [][]Field{
+			{newStringField(" ", false, false)}, {newStringField("\t", false, false)},
+			{newStringField(" \t ", false, false)}, {newStringField("x", false, false)}}, lines: []int64{2, 3, 4, 5}},
+		{name: "explicit crlf", terminator: "\r\n", input: "\r\n \r\n\t\r\nx\r\n", rows: [][]Field{
+			{newStringField(" ", false, false)}, {newStringField("\t", false, false)}, {newStringField("x", false, false)}}, lines: []int64{2, 3, 4}},
+		{name: "eof", input: " \nx\n\t", rows: [][]Field{
+			{newStringField(" ", false, false)}, {newStringField("x", false, false)}, {newStringField("\t", false, false)}}, lines: []int64{1, 2, 3}},
+		{name: "unicode whitespace", input: "\u00a0\n\u2003\n", rows: [][]Field{
+			{newStringField("\u00a0", false, false)}, {newStringField("\u2003", false, false)}}, lines: []int64{1, 2}},
+		{name: "quoted controls", terminator: "\n", input: "\"\"\n\" \"\n\"\r\"\n", rows: [][]Field{
+			{newStringField("", false, true)}, {newStringField(" ", false, true)}, {newStringField("\r", false, true)}}, lines: []int64{1, 2, 3}},
+		{name: "allow empty", allowEmpty: true, input: "\n \n", rows: [][]Field{
+			{newStringField("", false, false)}, {newStringField(" ", false, false)}}, lines: []int64{1, 2}},
+		{name: "allow empty crlf", allowEmpty: true, terminator: "\r\n", input: "\r\n \r\n", rows: [][]Field{
+			{newStringField("", false, false)}, {newStringField(" ", false, false)}}, lines: []int64{1, 2}},
+		{name: "comments", comment: "#", input: "# skip\n \n# skip\nx\n", rows: [][]Field{
+			{newStringField(" ", false, false)}, {newStringField("x", false, false)}}},
+		{name: "starting by", startingBy: "p:", terminator: "\n", input: "skip\np: \np:x\n", rows: [][]Field{
+			{newStringField(" ", false, false)}, {newStringField("x", false, false)}}},
+		{name: "multiple columns", input: " ,\n,\n", rows: [][]Field{
+			{newStringField(" ", false, false), newStringField("", false, false)},
+			{newStringField("", false, false), newStringField("", false, false)}}, lines: []int64{1, 2}},
+		{name: "cr at eof", terminator: "\n", input: "\r", rows: [][]Field{{newStringField("\r", false, false)}}, lines: []int64{1}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg := CSVConfig{FieldsTerminatedBy: ",", FieldsEnclosedBy: `"`, LinesTerminatedBy: tc.terminator,
+				Comment: tc.comment, LinesStartingBy: tc.startingBy, AllowEmptyLine: tc.allowEmpty}
+			parser, err := NewCSVParser(&cfg, strings.NewReader(tc.input), 1, false)
+			require.NoError(t, err)
+			var row []Field
+			for i, want := range tc.rows {
+				row, err = parser.Read(row)
+				require.NoError(t, err)
+				require.Equal(t, want, row)
+				if tc.lines != nil {
+					require.Equal(t, tc.lines[i], parser.RecordLine())
+				}
+			}
+			_, err = parser.Read(row)
+			require.ErrorIs(t, err, io.EOF)
+		})
+	}
+}
diff --git a/test/distributed/cases/load_data/external_csv_values.result b/test/distributed/cases/load_data/external_csv_values.result
new file mode 100644
index 0000000000000..70c97218a514e
--- /dev/null
+++ b/test/distributed/cases/load_data/external_csv_values.result
@@ -0,0 +1,56 @@
+drop database if exists csv_value_assignment;
+create database csv_value_assignment;
+use csv_value_assignment;
+create external table csv_numbers(a tinyint, b smallint, c int, d bigint, ua tinyint unsigned, ub smallint unsigned, uc int unsigned, ud bigint unsigned, m mediumint, um mediumint unsigned, exact_value bigint)
+infile{'filepath'='$resources/external_table_file/csv_integer_assignment.csv', 'format'='csv'}
+fields terminated by ',' lines terminated by '\n';
+select * from csv_numbers;
+➤ a[-6,8,0]  ¦  b[5,16,0]  ¦  c[4,32,0]  ¦  d[-5,64,0]  ¦  ua[-6,8,0]  ¦  ub[5,16,0]  ¦  uc[4,32,0]  ¦  ud[-5,64,0]  ¦  m[4,9,0]  ¦  um[4,8,0]  ¦  exact_value[-5,64,0]  𝄀
+2  ¦  2  ¦  100  ¦  -2  ¦  3  ¦  4  ¦  5  ¦  18446744073709551615  ¦  8388607  ¦  16777215  ¦  9007199254740993
+create table native_numbers(a tinyint, b smallint, c int, d bigint, ua tinyint unsigned, ub smallint unsigned, uc int unsigned, ud bigint unsigned, m mediumint, um mediumint unsigned, exact_value bigint);
+insert into native_numbers values (1.9,2.4,1e2,-1.9,2.5,3.9,4.5,18446744073709551615.0,8388606.9,16777214.9,9007199254740993.0);
+select count(*) as mismatches from csv_numbers e cross join native_numbers n
+where e.a <> n.a or e.b <> n.b or e.c <> n.c or e.d <> n.d
+or e.ua <> n.ua or e.ub <> n.ub or e.uc <> n.uc or e.ud <> n.ud
+or e.m <> n.m or e.um <> n.um or e.exact_value <> n.exact_value;
+➤ mismatches[-5,64,0]  𝄀
+0
+create external table csv_spaces(v varchar(8))
+infile{'filepath'='$resources/external_table_file/csv_whitespace_records.csv', 'format'='csv'}
+fields terminated by ',' enclosed by '"' escaped by '\\' lines terminated by '\n';
+select count(*) as records, count(v) as non_null from csv_spaces;
+➤ records[-5,64,0]  ¦  non_null[-5,64,0]  𝄀
+7  ¦  6
+select __mo_file_line as line, concat('[',hex(v),']') as value_hex from csv_spaces order by line;
+➤ line[-5,64,0]  ¦  value_hex[12,66,0]  𝄀
+2  ¦  [20]  𝄀
+3  ¦  [09]  𝄀
+4  ¦  [20]  𝄀
+5  ¦  []  𝄀
+6  ¦  [78]  𝄀
+7  ¦  null  𝄀
+8  ¦  [20]
+create table loaded_spaces(v varchar(8));
+load data infile '$resources/external_table_file/csv_whitespace_records.csv'
+into table loaded_spaces fields terminated by ',' enclosed by '"' escaped by '\\' lines terminated by '\n';
+select count(*) as records, count(v) as non_null from loaded_spaces;
+➤ records[-5,64,0]  ¦  non_null[-5,64,0]  𝄀
+7  ¦  6
+select coalesce(concat('[',hex(v),']'),'NULL') as value_hex, count(*) as n
+from loaded_spaces group by v order by value_hex;
+➤ value_hex[12,66,0]  ¦  n[-5,64,0]  𝄀
+NULL  ¦  1  𝄀
+[09]  ¦  1  𝄀
+[20]  ¦  3  𝄀
+[78]  ¦  1  𝄀
+[]  ¦  1
+create external table csv_overflow(v tinyint)
+infile{'filepath'='$resources/external_table_file/csv_integer_overflow.csv', 'format'='csv'}
+fields terminated by ',' lines terminated by '\n';
+select v from csv_overflow;
+internal error: the input value '127.5' is not int8 type for column 0
+select v is null as rejected, __mo_file_line as line,
+__mo_error_message like '%not int8 type%' as conversion_error from csv_overflow;
+➤ rejected[-7,1,0]  ¦  line[-5,64,0]  ¦  conversion_error[-7,1,0]  𝄀
+1  ¦  1  ¦  1
+drop database csv_value_assignment;
diff --git a/test/distributed/cases/load_data/external_csv_values.sql b/test/distributed/cases/load_data/external_csv_values.sql
new file mode 100644
index 0000000000000..9f64265c67ffc
--- /dev/null
+++ b/test/distributed/cases/load_data/external_csv_values.sql
@@ -0,0 +1,41 @@
+-- External CSV preserves decimal assignment values and whitespace field records.
+-- Numeric text uses exact assignment rounding; approximate SQL FLOAT ties have
+-- separate semantics. The whitespace fixture includes a genuine blank line,
+-- quoted empty/space, NULL, and a final space without a line terminator.
+drop database if exists csv_value_assignment;
+create database csv_value_assignment;
+use csv_value_assignment;
+
+create external table csv_numbers(a tinyint, b smallint, c int, d bigint, ua tinyint unsigned, ub smallint unsigned, uc int unsigned, ud bigint unsigned, m mediumint, um mediumint unsigned, exact_value bigint)
+infile{'filepath'='$resources/external_table_file/csv_integer_assignment.csv', 'format'='csv'}
+fields terminated by ',' lines terminated by '\n';
+select * from csv_numbers;
+create table native_numbers(a tinyint, b smallint, c int, d bigint, ua tinyint unsigned, ub smallint unsigned, uc int unsigned, ud bigint unsigned, m mediumint, um mediumint unsigned, exact_value bigint);
+insert into native_numbers values (1.9,2.4,1e2,-1.9,2.5,3.9,4.5,18446744073709551615.0,8388606.9,16777214.9,9007199254740993.0);
+select count(*) as mismatches from csv_numbers e cross join native_numbers n
+where e.a <> n.a or e.b <> n.b or e.c <> n.c or e.d <> n.d
+or e.ua <> n.ua or e.ub <> n.ub or e.uc <> n.uc or e.ud <> n.ud
+or e.m <> n.m or e.um <> n.um or e.exact_value <> n.exact_value;
+
+create external table csv_spaces(v varchar(8))
+infile{'filepath'='$resources/external_table_file/csv_whitespace_records.csv', 'format'='csv'}
+fields terminated by ',' enclosed by '"' escaped by '\\' lines terminated by '\n';
+select count(*) as records, count(v) as non_null from csv_spaces;
+select __mo_file_line as line, concat('[',hex(v),']') as value_hex from csv_spaces order by line;
+
+-- The shared parser preserves the same VARCHAR records for LOAD DATA.
+create table loaded_spaces(v varchar(8));
+load data infile '$resources/external_table_file/csv_whitespace_records.csv'
+into table loaded_spaces fields terminated by ',' enclosed by '"' escaped by '\\' lines terminated by '\n';
+select count(*) as records, count(v) as non_null from loaded_spaces;
+select coalesce(concat('[',hex(v),']'),'NULL') as value_hex, count(*) as n
+from loaded_spaces group by v order by value_hex;
+
+-- Check range after rounding, before narrowing or publishing a value.
+create external table csv_overflow(v tinyint)
+infile{'filepath'='$resources/external_table_file/csv_integer_overflow.csv', 'format'='csv'}
+fields terminated by ',' lines terminated by '\n';
+select v from csv_overflow;
+select v is null as rejected, __mo_file_line as line,
+__mo_error_message like '%not int8 type%' as conversion_error from csv_overflow;
+drop database csv_value_assignment;
diff --git a/test/distributed/resources/external_table_file/csv_integer_assignment.csv b/test/distributed/resources/external_table_file/csv_integer_assignment.csv
new file mode 100644
index 0000000000000..d7d8db4a20742
--- /dev/null
+++ b/test/distributed/resources/external_table_file/csv_integer_assignment.csv
@@ -0,0 +1 @@
+1.9,2.4,1e2,-1.9,2.5,3.9,4.5,18446744073709551615.0,8388606.9,16777214.9,9007199254740993.0
diff --git a/test/distributed/resources/external_table_file/csv_integer_overflow.csv b/test/distributed/resources/external_table_file/csv_integer_overflow.csv
new file mode 100644
index 0000000000000..4a02f0eebff73
--- /dev/null
+++ b/test/distributed/resources/external_table_file/csv_integer_overflow.csv
@@ -0,0 +1 @@
+127.5
diff --git a/test/distributed/resources/external_table_file/csv_whitespace_records.csv b/test/distributed/resources/external_table_file/csv_whitespace_records.csv
new file mode 100644
index 0000000000000..f094566d7579b
--- /dev/null
+++ b/test/distributed/resources/external_table_file/csv_whitespace_records.csv
@@ -0,0 +1,8 @@
+
+ 
+	
+" "
+""
+x
+\N
+ 
\ No newline at end of file

From b3e6fa906d6d16974beb6dbc16c672b1aa065675 Mon Sep 17 00:00:00 2001
From: iamlinjunhong <1030420200@qq.com>
Date: Thu, 15 Oct 2026 17:45:17 +0800
Subject: [PATCH 2/5] fix: scale external CSV decimals before integer range
 checks

---
 pkg/sql/colexec/external/error_mode_test.go   |  4 ++--
 pkg/sql/colexec/external/external.go          |  6 +++--
 pkg/sql/colexec/external/external_test.go     | 22 ++++++++++++++++---
 .../cases/load_data/external_csv_values.sql   |  1 +
 .../csv_integer_assignment.csv                |  2 +-
 5 files changed, 27 insertions(+), 8 deletions(-)

diff --git a/pkg/sql/colexec/external/error_mode_test.go b/pkg/sql/colexec/external/error_mode_test.go
index c6ecc86ff79bc..2f4cbe8067dba 100644
--- a/pkg/sql/colexec/external/error_mode_test.go
+++ b/pkg/sql/colexec/external/error_mode_test.go
@@ -207,9 +207,9 @@ func TestCSVErrorModeTolerates(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, "1,alpha\n2147483647.5,beta\n2.5,gamma\n"))
+		require.NoError(t, readAllText(t, param, proc, bat, "42949672960e-10,alpha\n2147483647.5,beta\n25000000000e-10,gamma\n"))
 		requireRows(t, bat, []wantRow{
-			{a: 1, s: "alpha", line: 1},
+			{a: 4, s: "alpha", line: 1},
 			{line: 2, message: "is not int32 type", text: "2147483647.5,beta"},
 			{a: 3, s: "gamma", line: 3},
 		})
diff --git a/pkg/sql/colexec/external/external.go b/pkg/sql/colexec/external/external.go
index 458dbfdfe4b32..a739f0b91fcd6 100644
--- a/pkg/sql/colexec/external/external.go
+++ b/pkg/sql/colexec/external/external.go
@@ -1891,9 +1891,11 @@ func loadTimeTextInput(param *ExternalParam, id types.T) bool {
 
 // 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 || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
+	if err == nil || !round || field.HasStringQuote {
 		return value, err
 	}
 	integer, err := function.ParseDecimalInteger(field.Val)
@@ -1905,7 +1907,7 @@ func parseCSVInt(field csvparser.Field, bits int, round bool) (int64, error) {
 
 func parseCSVUint(field csvparser.Field, bits int, round bool) (uint64, error) {
 	value, err := strconv.ParseUint(field.Val, 10, bits)
-	if err == nil || !round || errors.Is(err, strconv.ErrRange) || field.HasStringQuote {
+	if err == nil || !round || field.HasStringQuote {
 		return value, err
 	}
 	integer, err := function.ParseDecimalInteger(field.Val)
diff --git a/pkg/sql/colexec/external/external_test.go b/pkg/sql/colexec/external/external_test.go
index d47185ef0efba..e25e8853c2ec9 100644
--- a/pkg/sql/colexec/external/external_test.go
+++ b/pkg/sql/colexec/external/external_test.go
@@ -2784,9 +2784,12 @@ func TestGetColDataExternalCSVIntegerRounding(t *testing.T) {
 			cases := []testCase{
 				{value: "1.9", want: "2"}, {value: "2.4", want: "2"}, {value: "2.5", want: "3"},
 				{value: "1e2", want: "100"}, {value: "2.5e0", want: "3"}, {value: ".5", want: "1"},
+				{value: "1000e-3", want: "1"}, {value: "42949672960e-10", want: "4"},
+				{value: "1000000000000000000000e-21", want: "1"},
 				{value: "1.", want: "1"}, {value: "+1.9", want: "2"}, {value: " 1.9 ", want: "2"},
 				{value: "7", want: "7", quoted: true}, {value: "1.9", quoted: true, wantErr: true},
 				{value: "1e2", quoted: true, wantErr: true}, {value: "NaN", wantErr: true},
+				{value: "1000000000000000000000e-21", quoted: true, wantErr: true},
 				{value: "+Inf", wantErr: true}, {value: "-Inf", wantErr: true}, {value: "1e+", wantErr: true},
 				{value: "1 2", wantErr: true}, {value: "1+2", wantErr: true}, {value: "1.9tail", wantErr: true},
 				{value: "0x1p2", wantErr: true}, {value: "1_0.5", wantErr: true},
@@ -2795,11 +2798,13 @@ func TestGetColDataExternalCSVIntegerRounding(t *testing.T) {
 			}
 			if typ.Oid.IsUnsignedInt() {
 				cases = append(cases, testCase{value: "-0.1", wantErr: true},
+					testCase{value: "-1000000000000000000000e-21", wantErr: true},
 					testCase{value: "-1.9", wantErr: true},
 					testCase{value: "-0.0", want: "0"},
 					testCase{value: "-0e2", want: "0"})
 			} else {
 				cases = append(cases, testCase{value: "-1.9", want: "-2"},
+					testCase{value: "-1000000000000000000000e-21", want: "-1"},
 					testCase{value: "-2.5", want: "-3"},
 					testCase{value: "-0.49", want: "0"})
 			}
@@ -2830,11 +2835,14 @@ func TestGetColDataExternalCSVIntegerRounding(t *testing.T) {
 				}
 			}
 			cases = append(cases, testCase{value: max, want: max},
+				testCase{value: max + "0", wantErr: true},
+				testCase{value: max + "0e-1", want: max},
 				testCase{value: max + ".0", want: max},
 				testCase{value: max + ".4", want: max},
 				testCase{value: max + ".5", wantErr: true})
 			if min != "" {
 				cases = append(cases, testCase{value: min + ".4", want: min},
+					testCase{value: min + "0e-1", want: min},
 					testCase{value: min + ".5", wantErr: true})
 			}
 			if typ.Oid == types.T_int64 || typ.Oid == types.T_uint64 {
@@ -2893,10 +2901,13 @@ func TestGetColDataIntegerRoundingScope(t *testing.T) {
 		externType        int32
 		enumValues, value string
 		want              uint64
+		wantErr           bool
 	}{
-		{"load csv", tree.CSV, int32(plan.ExternType_LOAD), "", "1.9", 1},
-		{"external jsonline", tree.JSONLINE, int32(plan.ExternType_EXTERNAL_TB), "", "1.9", 1},
-		{"external set", tree.CSV, int32(plan.ExternType_EXTERNAL_TB), "1.9,two", "1.9", 1},
+		{"load csv", tree.CSV, int32(plan.ExternType_LOAD), "", "1.9", 1, false},
+		{"external jsonline", tree.JSONLINE, int32(plan.ExternType_EXTERNAL_TB), "", "1.9", 1, false},
+		{"external set", tree.CSV, int32(plan.ExternType_EXTERNAL_TB), "1.9,two", "1.9", 1, false},
+		{"load csv overflow mantissa", tree.CSV, int32(plan.ExternType_LOAD), "", "1000000000000000000000e-21", 0, true},
+		{"external jsonline overflow mantissa", tree.JSONLINE, int32(plan.ExternType_EXTERNAL_TB), "", "1000000000000000000000e-21", 0, true},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			bat := batch.NewWithSize(1)
@@ -2906,6 +2917,11 @@ func TestGetColDataIntegerRoundingScope(t *testing.T) {
 				Cols:   []*plan.ColDef{{Name: "i", Typ: plan.Type{Id: int32(types.T_uint64), Enumvalues: tc.enumValues}}},
 				Extern: &tree.ExternParam{ExParamConst: tree.ExParamConst{Format: tc.format}, ExParam: tree.ExParam{ExternType: tc.externType}}}, ExParam: ExParam{Fileparam: &ExFileparam{}}}
 			err := getColData(bat, []csvparser.Field{{Val: tc.value}}, 0, param, proc.Mp(), plan.ExternAttr{ColName: "i"}, proc)
+			if tc.wantErr {
+				require.Error(t, err)
+				require.Zero(t, bat.Vecs[0].Length())
+				return
+			}
 			require.NoError(t, err)
 			require.Equal(t, []uint64{tc.want}, vector.MustFixedColWithTypeCheck[uint64](bat.Vecs[0]))
 		})
diff --git a/test/distributed/cases/load_data/external_csv_values.sql b/test/distributed/cases/load_data/external_csv_values.sql
index 9f64265c67ffc..29bc19ef816a6 100644
--- a/test/distributed/cases/load_data/external_csv_values.sql
+++ b/test/distributed/cases/load_data/external_csv_values.sql
@@ -1,4 +1,5 @@
 -- External CSV preserves decimal assignment values and whitespace field records.
+-- Oversized mantissas must be scaled and rounded before destination range checks.
 -- Numeric text uses exact assignment rounding; approximate SQL FLOAT ties have
 -- separate semantics. The whitespace fixture includes a genuine blank line,
 -- quoted empty/space, NULL, and a final space without a line terminator.
diff --git a/test/distributed/resources/external_table_file/csv_integer_assignment.csv b/test/distributed/resources/external_table_file/csv_integer_assignment.csv
index d7d8db4a20742..fbf1273ebbb01 100644
--- a/test/distributed/resources/external_table_file/csv_integer_assignment.csv
+++ b/test/distributed/resources/external_table_file/csv_integer_assignment.csv
@@ -1 +1 @@
-1.9,2.4,1e2,-1.9,2.5,3.9,4.5,18446744073709551615.0,8388606.9,16777214.9,9007199254740993.0
+1900e-3,240000e-5,1000000000000000000000e-19,-1900000000000000000000e-21,2500e-3,390000e-5,45000000000e-10,184467440737095516150e-1,8388606.9,16777214.9,9007199254740993.0

From 2ca34a1999c081ecd7df38624c04fc169282f795 Mon Sep 17 00:00:00 2001
From: iamlinjunhong <1030420200@qq.com>
Date: Thu, 15 Oct 2026 23:32:20 +0800
Subject: [PATCH 3/5] fix(external): preserve JSONLINE whitespace record
 skipping

---
 pkg/sql/colexec/external/error_mode_test.go   | 74 +++++++++++++++++++
 pkg/sql/colexec/external/reader_csv.go        |  9 +++
 .../load_data/external_csv_values.result      | 17 +++++
 .../cases/load_data/external_csv_values.sql   |  9 +++
 .../jsonline_whitespace_records.jsonl         |  3 +
 5 files changed, 112 insertions(+)
 create mode 100644 test/distributed/resources/external_table_file/jsonline_whitespace_records.jsonl

diff --git a/pkg/sql/colexec/external/error_mode_test.go b/pkg/sql/colexec/external/error_mode_test.go
index 2f4cbe8067dba..0530a22a9ec53 100644
--- a/pkg/sql/colexec/external/error_mode_test.go
+++ b/pkg/sql/colexec/external/error_mode_test.go
@@ -301,6 +301,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) {
diff --git a/pkg/sql/colexec/external/reader_csv.go b/pkg/sql/colexec/external/reader_csv.go
index a29b69bc4dd45..ac6f310cee8fa 100644
--- a/pkg/sql/colexec/external/reader_csv.go
+++ b/pkg/sql/colexec/external/reader_csv.go
@@ -165,6 +165,15 @@ func (r *CsvReader) makeBatchRows(proc *process.Process, bat *batch.Batch) (file
 			break
 		}
 
+		// CSV whitespace records are data, but JSONLINE keeps its existing
+		// blank-line policy. Skip before ignore/batch accounting, as the shared
+		// parser used to do; quoted scalars and split records still need validation.
+		if param.Extern.Format == tree.JSONLINE && len(row) == 1 &&
+			!row[0].HasStringQuote && strings.TrimSpace(row[0].Val) == "" {
+			i--
+			continue
+		}
+
 		if r.ignoreLine > 0 {
 			if !param.Extern.Parallel || param.FileOffset[0] == 0 {
 				r.ignoreLine--
diff --git a/test/distributed/cases/load_data/external_csv_values.result b/test/distributed/cases/load_data/external_csv_values.result
index 70c97218a514e..07772880893ac 100644
--- a/test/distributed/cases/load_data/external_csv_values.result
+++ b/test/distributed/cases/load_data/external_csv_values.result
@@ -53,4 +53,21 @@ select v is null as rejected, __mo_file_line as line,
 __mo_error_message like '%not int8 type%' as conversion_error from csv_overflow;
 ➤ rejected[-7,1,0]  ¦  line[-5,64,0]  ¦  conversion_error[-7,1,0]  𝄀
 1  ¦  1  ¦  1
+create external table json_lines(a int, s varchar(8))
+infile{'filepath'='$resources/external_table_file/jsonline_whitespace_records.jsonl', 'format'='jsonline', 'jsondata'='object'};
+select a, s, __mo_file_line as line from json_lines order by a;
+➤ a[4,32,0]  ¦  s[12,8,0]  ¦  line[-5,64,0]  𝄀
+1  ¦  alpha  ¦  1  𝄀
+2  ¦  beta  ¦  3
+select a, s, __mo_file_line as line, __mo_error_message is null as valid_record
+from json_lines order by a;
+➤ a[4,32,0]  ¦  s[12,8,0]  ¦  line[-5,64,0]  ¦  valid_record[-7,1,0]  𝄀
+1  ¦  alpha  ¦  1  ¦  1  𝄀
+2  ¦  beta  ¦  3  ¦  1
+create table loaded_json_lines(a int, s varchar(8));
+load data infile{'filepath'='$resources/external_table_file/jsonline_whitespace_records.jsonl', 'format'='jsonline', 'jsondata'='object'} into table loaded_json_lines;
+select a, s from loaded_json_lines order by a;
+➤ a[4,32,0]  ¦  s[12,8,0]  𝄀
+1  ¦  alpha  𝄀
+2  ¦  beta
 drop database csv_value_assignment;
diff --git a/test/distributed/cases/load_data/external_csv_values.sql b/test/distributed/cases/load_data/external_csv_values.sql
index 29bc19ef816a6..eacb659f678b2 100644
--- a/test/distributed/cases/load_data/external_csv_values.sql
+++ b/test/distributed/cases/load_data/external_csv_values.sql
@@ -39,4 +39,13 @@ fields terminated by ',' lines terminated by '\n';
 select v from csv_overflow;
 select v is null as rejected, __mo_file_line as line,
 __mo_error_message like '%not int8 type%' as conversion_error from csv_overflow;
+-- JSONLINE retains its format-specific blank-record policy.
+create external table json_lines(a int, s varchar(8))
+infile{'filepath'='$resources/external_table_file/jsonline_whitespace_records.jsonl', 'format'='jsonline', 'jsondata'='object'};
+select a, s, __mo_file_line as line from json_lines order by a;
+select a, s, __mo_file_line as line, __mo_error_message is null as valid_record
+from json_lines order by a;
+create table loaded_json_lines(a int, s varchar(8));
+load data infile{'filepath'='$resources/external_table_file/jsonline_whitespace_records.jsonl', 'format'='jsonline', 'jsondata'='object'} into table loaded_json_lines;
+select a, s from loaded_json_lines order by a;
 drop database csv_value_assignment;
diff --git a/test/distributed/resources/external_table_file/jsonline_whitespace_records.jsonl b/test/distributed/resources/external_table_file/jsonline_whitespace_records.jsonl
new file mode 100644
index 0000000000000..2f06fe1d57d78
--- /dev/null
+++ b/test/distributed/resources/external_table_file/jsonline_whitespace_records.jsonl
@@ -0,0 +1,3 @@
+{"a":1,"s":"alpha"}
+   
+{"a":2,"s":"beta"}

From feb9fb90ca6ef82935e99f0ae59a9d83ed253f0a Mon Sep 17 00:00:00 2001
From: XuPeng-SH <xupeng3112@163.com>
Date: Sun, 18 Oct 2026 16:39:43 +0800
Subject: [PATCH 4/5] fix(load): share JSONLINE whitespace framing with
 pre-scan

---
 pkg/sql/colexec/external/reader_csv.go        |  9 ---
 pkg/sql/colexec/external/types.go             |  5 ++
 pkg/sql/plan/build_load.go                    |  8 +-
 pkg/sql/plan/build_load_prescan_test.go       | 27 +++++++
 pkg/sql/util/csvparser/csv_parser.go          | 31 ++++++++
 pkg/sql/util/csvparser/scan_record_test.go    | 36 ++++++++-
 .../sqlintegration/csv_multiline_load_test.go | 78 +++++++++++++++++++
 7 files changed, 182 insertions(+), 12 deletions(-)

diff --git a/pkg/sql/colexec/external/reader_csv.go b/pkg/sql/colexec/external/reader_csv.go
index ac6f310cee8fa..a29b69bc4dd45 100644
--- a/pkg/sql/colexec/external/reader_csv.go
+++ b/pkg/sql/colexec/external/reader_csv.go
@@ -165,15 +165,6 @@ func (r *CsvReader) makeBatchRows(proc *process.Process, bat *batch.Batch) (file
 			break
 		}
 
-		// CSV whitespace records are data, but JSONLINE keeps its existing
-		// blank-line policy. Skip before ignore/batch accounting, as the shared
-		// parser used to do; quoted scalars and split records still need validation.
-		if param.Extern.Format == tree.JSONLINE && len(row) == 1 &&
-			!row[0].HasStringQuote && strings.TrimSpace(row[0].Val) == "" {
-			i--
-			continue
-		}
-
 		if r.ignoreLine > 0 {
 			if !param.Extern.Parallel || param.FileOffset[0] == 0 {
 				r.ignoreLine--
diff --git a/pkg/sql/colexec/external/types.go b/pkg/sql/colexec/external/types.go
index 5778b022590da..8dfa7205b1df0 100644
--- a/pkg/sql/colexec/external/types.go
+++ b/pkg/sql/colexec/external/types.go
@@ -531,6 +531,11 @@ func newCSVParserFromReader(extern *tree.ExternParam, r io.Reader) (*csvparser.C
 		Comment: plan.GetCSVComment(extern),
 	}
 
+	if extern.Format == tree.JSONLINE {
+		config.SkipWhitespaceRecords = true
+		config.Context = extern.Ctx
+	}
+
 	return csvparser.NewCSVParser(&config, bufio.NewReader(r), csvparser.ReadBlockSize, false)
 }
 
diff --git a/pkg/sql/plan/build_load.go b/pkg/sql/plan/build_load.go
index ae836c1a620c7..ec6a275aeebc3 100644
--- a/pkg/sql/plan/build_load.go
+++ b/pkg/sql/plan/build_load.go
@@ -176,6 +176,11 @@ func newReaderWithParam(param *tree.ExternParam, reader io.ReadCloser) (*csvpars
 		Comment: "",
 	}
 
+	if param.Format == tree.JSONLINE {
+		config.SkipWhitespaceRecords = true
+		config.Context = param.Ctx
+	}
+
 	return csvparser.NewCSVParser(&config, bufio.NewReader(reader), csvparser.ReadBlockSize, false)
 }
 
@@ -210,9 +215,8 @@ func IgnoredLines(param *tree.ExternParam, ctx CompilerContext) (offset int64, e
 		return 0, err
 	}
 
-	var lastRow []csvparser.Field
 	for skipLines > 0 {
-		lastRow, err = csvReader.Read(lastRow)
+		err = csvReader.ScanRecord()
 		if err != nil {
 			if err == io.EOF {
 				param.Tail.IgnoredLines = 0
diff --git a/pkg/sql/plan/build_load_prescan_test.go b/pkg/sql/plan/build_load_prescan_test.go
index 96d5e639db335..1c1d73b6abc75 100644
--- a/pkg/sql/plan/build_load_prescan_test.go
+++ b/pkg/sql/plan/build_load_prescan_test.go
@@ -137,3 +137,30 @@ func TestIgnoredLinesClosesPrescanReaders(t *testing.T) {
 		})
 	}
 }
+
+func TestIgnoredLinesJSONLineWhitespace(t *testing.T) {
+	for _, tc := range []struct {
+		name, data, format string
+		ignore             uint64
+		offset             int64
+	}{
+		{"JSONLINE LF", "\n \n{\"a\":2}\n{\"a\":3}\n", tree.JSONLINE, 1, 11},
+		{"JSONLINE CRLF", "\r\n \r\n[2]\r\n[3]", tree.JSONLINE, 1, 10},
+		{"JSONLINE empty EOF", " \n\u00a0", tree.JSONLINE, 1, 4},
+		{"JSONLINE ignore beyond EOF", " \n[2]", tree.JSONLINE, 2, 5},
+		{"CSV whitespace is a record", "\n \nx\n", tree.CSV, 1, 3},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			fs := newPrescanTrackingFS(t, tc.data)
+			param := &tree.ExternParam{ExParamConst: tree.ExParamConst{Filepath: "etl:/prescan.csv", Format: tc.format, ScanType: tree.INFILE, Tail: &tree.TailParameter{IgnoredLines: tc.ignore}}, ExParam: tree.ExParam{FileService: fs}}
+			compCtx, _, _ := newLoadGlobTestCtx(t)
+			offset, err := IgnoredLines(param, compCtx)
+			require.NoError(t, err)
+			require.Equal(t, tc.offset, offset)
+			require.Zero(t, param.Tail.IgnoredLines)
+			require.Nil(t, param.Ctx)
+			require.Len(t, fs.readers, 1)
+			require.Equal(t, 1, fs.readers[0].closes)
+		})
+	}
+}
diff --git a/pkg/sql/util/csvparser/csv_parser.go b/pkg/sql/util/csvparser/csv_parser.go
index 5c195e7aa927a..8ae768d38e0bc 100644
--- a/pkg/sql/util/csvparser/csv_parser.go
+++ b/pkg/sql/util/csvparser/csv_parser.go
@@ -16,6 +16,7 @@ package csvparser
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -77,6 +78,11 @@ type CSVConfig struct {
 	NotNull           bool
 
 	AllowEmptyLine bool
+	// SkipWhitespaceRecords keeps unquoted single-field whitespace records out
+	// of JSONLINE record accounting. CSV callers leave it disabled.
+	SkipWhitespaceRecords bool
+	// Context is borrowed from the reader owner; nil disables cancellation checks.
+	Context context.Context
 	// For non-empty FieldsEnclosedBy (for example quotes), null elements inside quotes are not considered as null except for
 	// `\N` (when escape-by is `\`). That is to say, `\N` is special for null because it always means null.
 	QuotedNullIsText bool
@@ -602,6 +608,11 @@ func (parser *CSVParser) readUntil(chars *byteSet) ([]byte, byte, error) {
 }
 
 func (parser *CSVParser) readRecord() error {
+	if parser.cfg.Context != nil {
+		if err := parser.cfg.Context.Err(); err != nil {
+			return err
+		}
+	}
 	parser.recordStartLine = parser.lineNo
 	parser.recordBuffer = parser.recordBuffer[:0]
 	parser.fieldIndexes = parser.fieldIndexes[:0]
@@ -746,6 +757,21 @@ outside:
 					}
 				}
 			}
+			if parser.cfg.SkipWhitespaceRecords && len(parser.fieldIndexes) == 0 &&
+				!fieldIsQuoted && len(parser.recordBuffer) > 0 && len(bytes.TrimSpace(parser.recordBuffer)) == 0 {
+				if err == io.EOF {
+					return io.EOF
+				}
+				if parser.cfg.Context != nil {
+					if err := parser.cfg.Context.Err(); err != nil {
+						return err
+					}
+				}
+				parser.recordBuffer = parser.recordBuffer[:0]
+				parser.recordStartLine = parser.lineNo
+				isEmptyLine = true
+				continue
+			}
 			parser.fieldIndexes = append(parser.fieldIndexes, len(parser.recordBuffer))
 			parser.fieldIsQuoted = append(parser.fieldIsQuoted, fieldIsQuoted)
 			// the loop is end, no need to reset fieldIsQuoted
@@ -884,6 +910,11 @@ func (parser *CSVParser) readUntilTerminator() ([]byte, int64, error) {
 }
 
 func (parser *CSVParser) readBlock() error {
+	if parser.cfg.Context != nil {
+		if err := parser.cfg.Context.Err(); err != nil {
+			return err
+		}
+	}
 	n, err := io.ReadFull(parser.reader, parser.blockBuf)
 
 	switch {
diff --git a/pkg/sql/util/csvparser/scan_record_test.go b/pkg/sql/util/csvparser/scan_record_test.go
index c3427e29e0fc1..e95cff963163f 100644
--- a/pkg/sql/util/csvparser/scan_record_test.go
+++ b/pkg/sql/util/csvparser/scan_record_test.go
@@ -15,6 +15,8 @@
 package csvparser
 
 import (
+	"context"
+	"fmt"
 	"io"
 	"strings"
 	"testing"
@@ -26,7 +28,8 @@ import (
 func TestScanRecordMatchesReadFraming(t *testing.T) {
 	for _, tc := range []struct {
 		name, data, terminator, comment, startingBy string
-		header, emptyLines                          bool
+		header, emptyLines, skipWhitespace          bool
+		separator                                   string
 		records                                     int
 		syntaxError                                 bool
 	}{
@@ -36,6 +39,13 @@ func TestScanRecordMatchesReadFraming(t *testing.T) {
 		{name: "starting prefix", data: "skip\nrow:1,\"a\nb\"\nrow:2,x\n", terminator: "\n", startingBy: "row:", records: 2},
 		{name: "empty lines skipped", data: "\n1,x\n\n2,y\n", records: 2},
 		{name: "empty lines retained", data: "\n1,x\n", emptyLines: true, records: 2},
+		{name: "whitespace default data", data: " \nx\n\t", records: 3},
+		{name: "JSONLINE blank records", data: "\n \n\u00a0\nx\n\t", records: 1, skipWhitespace: true},
+		{name: "JSONLINE CRLF and EOF", data: " \r\nx\r\n ", records: 1, skipWhitespace: true},
+		{name: "JSONLINE all whitespace", data: " \n\u2003", skipWhitespace: true},
+		{name: "JSONLINE quoted whitespace", data: "\"\"\n\" \"\n", records: 2, skipWhitespace: true},
+		{name: "JSONLINE tab fields", data: " \t\n", separator: "\t", records: 1, skipWhitespace: true},
+		{name: "JSONLINE header", data: " \nID\n \nx\n", header: true, records: 1, skipWhitespace: true},
 		{name: "empty file"},
 		{name: "no final terminator", data: "1,x", records: 1},
 		{name: "comment without final terminator", data: "1,x", comment: "#", records: 1},
@@ -51,6 +61,10 @@ func TestScanRecordMatchesReadFraming(t *testing.T) {
 				LinesStartingBy: tc.startingBy, AllowEmptyLine: tc.emptyLines,
 				HeaderSchemaMatch: tc.header, Null: []string{`\N`},
 			}
+			cfg.SkipWhitespaceRecords = tc.skipWhitespace
+			if tc.separator != "" {
+				cfg.FieldsTerminatedBy = tc.separator
+			}
 			// Small buffers exercise framing across input chunks too.
 			read, err := NewCSVParser(&cfg, strings.NewReader(tc.data), 7, tc.header)
 			require.NoError(t, err)
@@ -128,3 +142,23 @@ func TestCSVLookaheadPropagatesReadError(t *testing.T) {
 		}
 	}
 }
+
+func TestWhitespaceRecordCancellation(t *testing.T) {
+	for _, scan := range []bool{false, true} {
+		t.Run(fmt.Sprint(scan), func(t *testing.T) {
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+			parser, err := NewCSVParser(&CSVConfig{FieldsTerminatedBy: ",", SkipWhitespaceRecords: true, Context: ctx}, strings.NewReader("x\n \n \n"), 64, false)
+			require.NoError(t, err)
+			require.NoError(t, parser.ScanRecord())
+			cancel()
+			if scan {
+				require.ErrorIs(t, parser.ScanRecord(), context.Canceled)
+			} else {
+				_, err = parser.Read(nil)
+				require.ErrorIs(t, err, context.Canceled)
+			}
+			require.Equal(t, int64(2), parser.Pos(), "buffered whitespace must not consume records after cancellation")
+		})
+	}
+}
diff --git a/pkg/tests/sqlintegration/csv_multiline_load_test.go b/pkg/tests/sqlintegration/csv_multiline_load_test.go
index 54b97ad937241..f265c07198d42 100644
--- a/pkg/tests/sqlintegration/csv_multiline_load_test.go
+++ b/pkg/tests/sqlintegration/csv_multiline_load_test.go
@@ -116,3 +116,81 @@ func TestLoadCSVParallelNonStrictMultiline(t *testing.T) {
 		checkRows()
 	})
 }
+
+func TestLoadJSONLineIgnoreWhitespace(t *testing.T) {
+	runSQLIntegration(t, func(c embed.Cluster) {
+		ctx, cancel := context.WithTimeout(t.Context(), 90*time.Second)
+		defer cancel()
+		cn, err := c.GetCNService(0)
+		require.NoError(t, err)
+		defer cleanupSQLIntegration(t, cn, "drop database if exists jsonline_ignore_whitespace")
+		db, err := sql.Open("mysql", fmt.Sprintf("dump:111@tcp(127.0.0.1:%d)/", cn.GetServiceConfig().CN.Frontend.Port))
+		require.NoError(t, err)
+		defer db.Close()
+		conn, err := db.Conn(ctx)
+		require.NoError(t, err)
+		defer conn.Close()
+		exec := func(t *testing.T, query string) {
+			t.Helper()
+			_, err := conn.ExecContext(ctx, query)
+			require.NoError(t, err, query)
+		}
+		exec(t, "create database jsonline_ignore_whitespace")
+		exec(t, "use jsonline_ignore_whitespace")
+		exec(t, "set max_dop=2")
+		exec(t, "create table dst(a int)")
+		testingContext := moruntime.MustGetTestingContext(cn.ServiceID())
+		for _, shape := range []string{"object", "array"} {
+			t.Run(shape, func(t *testing.T) {
+				row := func(value int) string {
+					if shape == "object" {
+						return fmt.Sprintf(`{"a":%d}`, value)
+					}
+					return fmt.Sprintf("[%d]", value)
+				}
+				for _, split := range []bool{false, true} {
+					data := "\n \n" + row(2) + "\n" + row(3) + "\n"
+					want := "3"
+					if split {
+						data = "\n \n" + row(1) + "\n \n" + row(2) + "\n" + row(3) + "\n" + row(4) + "\n" + row(5) + "\n "
+						want = "2,3,4,5"
+					}
+					path := filepath.Join(t.TempDir(), "records.jsonl")
+					require.NoError(t, os.WriteFile(path, []byte(data), 0600))
+					var observed struct {
+						sync.Mutex
+						readers int
+						offsets []int64
+					}
+					previous := testingContext.SetLoadDataTestConfig(moruntime.LoadDataTestConfig{
+						Filepath: path, MinSize: 1,
+						OnRanges: func(readers int, offsets []int64) {
+							observed.Lock()
+							defer observed.Unlock()
+							observed.readers = readers
+							observed.offsets = append([]int64(nil), offsets...)
+						},
+					})
+					func() {
+						defer testingContext.SetLoadDataTestConfig(previous)
+						for _, parallel := range []bool{false, true} {
+							exec(t, "truncate table dst")
+							exec(t, fmt.Sprintf("load data infile{'filepath'='%s','format'='jsonline','jsondata'='%s'} into table dst ignore 1 lines parallel '%t' strict 'false'", strings.ReplaceAll(path, "'", "''"), shape, parallel))
+							var got string
+							require.NoError(t, conn.QueryRowContext(ctx, "select group_concat(a order by a) from dst").Scan(&got))
+							require.Equal(t, want, got, "shape=%s split=%t parallel=%t", shape, split, parallel)
+							if split && parallel {
+								observed.Lock()
+								readers, offsets := observed.readers, append([]int64(nil), observed.offsets...)
+								observed.Unlock()
+								require.Equal(t, 2, readers, "must exercise actual multiple readers")
+								require.NotEmpty(t, offsets)
+								require.Equal(t, int64(3+len(row(1))+1), offsets[0], "prescan skips exactly the first visible JSON record")
+							}
+						}
+					}()
+				}
+			})
+		}
+	})
+}

From d4ceca3805a471c20e44e4a71228426edd761119 Mon Sep 17 00:00:00 2001
From: iamlinjunhong <1030420200@qq.com>
Date: Mon, 19 Oct 2026 22:42:19 +0800
Subject: [PATCH 5/5] test(csvparser): preserve record lines under whitespace
 policies

---
 pkg/sql/util/csvparser/csv_parser_test.go | 19 +++++++++++++------
 1 file changed, 13 insertions(+), 6 deletions(-)

diff --git a/pkg/sql/util/csvparser/csv_parser_test.go b/pkg/sql/util/csvparser/csv_parser_test.go
index 8ba887635974e..3d0a6c2106839 100644
--- a/pkg/sql/util/csvparser/csv_parser_test.go
+++ b/pkg/sql/util/csvparser/csv_parser_test.go
@@ -727,12 +727,13 @@ func TestRecordLine(t *testing.T) {
 	// Each skip path must move the candidate record start forward, while a
 	// multi-line quoted record must keep the line where its first field began.
 	for _, tc := range []struct {
-		name       string
-		input      string
-		comment    string
-		startingBy string
-		lines      []int64
-		rows       [][]string
+		name           string
+		input          string
+		comment        string
+		startingBy     string
+		skipWhitespace bool
+		lines          []int64
+		rows           [][]string
 	}{
 		{
 			name:  "comments before and between records",
@@ -752,6 +753,11 @@ func TestRecordLine(t *testing.T) {
 		{
 			name:  "comments and blanks before multiline record",
 			input: "#h\n\n   \n#c\na,\"x\ny\"\n#d\nb,2", comment: "#",
+			lines: []int64{3, 5, 8}, rows: [][]string{{"   "}, {"a", "x\ny"}, {"b", "2"}},
+		},
+		{
+			name:  "opt-in whitespace skipping before multiline record",
+			input: "#h\n\n   \n#c\na,\"x\ny\"\n#d\nb,2", comment: "#", skipWhitespace: true,
 			lines: []int64{5, 8}, rows: [][]string{{"a", "x\ny"}, {"b", "2"}},
 		},
 		{
@@ -782,6 +788,7 @@ func TestRecordLine(t *testing.T) {
 					skipCfg.Comment = tc.comment
 					skipCfg.LinesStartingBy = tc.startingBy
 					skipCfg.LinesTerminatedBy = "\n"
+					skipCfg.SkipWhitespaceRecords = tc.skipWhitespace
 					parser, err := NewCSVParser(&skipCfg, NewStringReader(tc.input), blockSize, false)
 					require.NoError(t, err)
 					for i, want := range tc.lines {