feat: support LOAD DATA INFILE with IGNORE to skip error rows

pkg/frontend/status_stmt.go

@@ -15,6 +15,7 @@
 package frontend
 
 import (
+	"math"
 	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
@@ -285,6 +286,12 @@ func (resper *MysqlResp) respStatus(ses *Session,
 				isIssue3482 = true
 				localFileName = st.Param.Filepath
 			}
+			if w := execCtx.proc.Base.Warnings.Load(); w > 0 {
+				if w > math.MaxUint16 {
+					w = math.MaxUint16
+				}
+				res.warnings = uint16(w)
+			}
 		}
 
 		if err2 := resper.mysqlRrWr.WriteResponse(execCtx.reqCtx, res); err2 != nil {

pkg/pb/pipeline/pipeline.go

@@ -69,6 +69,10 @@ func (m *Message) SetProcData(data []byte) {
 	m.ProcInfoData = data
 }
 
+func (m *Message) SetWarningCount(n uint32) {
+	m.WarningCount = n
+}
+
 func (m *Message) DebugString() string {
 	errInfo := "none"
 	if len(m.Err) > 0 {

pkg/sql/colexec/external/reader_csv.go

@@ -122,6 +122,7 @@ func (r *CsvReader) makeBatchRows(proc *process.Process, bat *batch.Batch) (file
 	var finish bool
 	var row []csvparser.Field
 	var unexpectEOF bool
+	ignoreError := param.IgnoreError
 
 	for i := 0; i < OneBatchMaxRow; i++ {
 		select {
@@ -135,6 +136,14 @@ func (r *CsvReader) makeBatchRows(proc *process.Process, bat *batch.Batch) (file
 			if err != nil {
 				if err == io.EOF {
 					finish = true
+				} else if ignoreError {
+					if recoverErr := csvReader.RecoverAfterReadError(); recoverErr != nil {
+						return false, recoverErr
+					}
+					logutil.Warnf("load data ignore error: read csv row failed: %v", err)
+					proc.Base.Warnings.Add(1)
+					i--
+					continue
 				} else {
 					return false, err
 				}
@@ -161,10 +170,6 @@ func (r *CsvReader) makeBatchRows(proc *process.Process, bat *batch.Batch) (file
 			}
 		}
 
-		for j := 0; j < len(row); j++ {
-			curBatchSize += uint64(len(row[j].Val))
-		}
-
 		rowIdx := i
 		if param.Extern.Format == tree.JSONLINE {
 			row, err = r.transJson2Lines(proc.Ctx, row[0].Val, param.Attrs, param.Cols, param.Extern.JsonData)
@@ -181,6 +186,9 @@ func (r *CsvReader) makeBatchRows(proc *process.Process, bat *batch.Batch) (file
 		if err = getOneRowData(proc, bat, row, rowIdx, param); err != nil {
 			return false, err
 		}
+		for j := 0; j < len(row); j++ {
+			curBatchSize += uint64(len(row[j].Val))
+		}
 
 		if curBatchSize >= param.maxBatchSize {
 			break

pkg/sql/colexec/external/types.go

@@ -52,6 +52,7 @@ type ExternalParam struct {
 type ExParamConst struct {
 	ParallelLoad  bool
 	StrictSqlMode bool
+	IgnoreError   bool
 	Close         byte
 	maxBatchSize  uint64
 	Idx           int

pkg/sql/compile/operator.go

@@ -350,6 +350,7 @@ func dupOperator(sourceOp vm.Operator, index int, maxParallel int) vm.Operator {
 					FileOffsetTotal: t.Es.FileOffsetTotal,
 					Extern:          t.Es.Extern,
 					StrictSqlMode:   t.Es.StrictSqlMode,
+					IgnoreError:     t.Es.IgnoreError,
 				},
 				ExParam: external.ExParam{
 					Filter: &external.FilterParam{
@@ -898,6 +899,7 @@ func constructExternal(node *plan.Node, param *tree.ExternParam, ctx context.Con
 				FileSize:        FileSize,
 				ClusterTable:    node.GetClusterTable(),
 				StrictSqlMode:   strictSqlMode,
+				IgnoreError:     param.IgnoreError,
 			},
 			ExParam: external.ExParam{
 				Fileparam: new(external.ExFileparam),

pkg/sql/compile/remoterun.go

@@ -659,6 +659,7 @@ func convertToPipelineInstruction(op vm.Operator, proc *process.Process, ctx *sc
 			FileList:        t.Es.FileList,
 			Filter:          t.Es.Filter.FilterExpr,
 			StrictSqlMode:   t.Es.StrictSqlMode,
+			IgnoreError:     t.Es.IgnoreError,
 		}
 		in.ProjectList = t.ProjectList
 	case *source.Source:
@@ -1104,6 +1105,7 @@ func convertToVmOperator(opr *pipeline.Instruction, ctx *scopeContext, eng engin
 					CreateSql:       t.CreateSql,
 					FileList:        t.FileList,
 					StrictSqlMode:   t.StrictSqlMode,
+					IgnoreError:     t.IgnoreError,
 				},
 				ExParam: external.ExParam{
 					Fileparam: new(external.ExFileparam),

pkg/sql/compile/remoterunClient.go

@@ -224,9 +224,13 @@ func receiveMessageFromCnServerIfOnlyRun(s *Scope, sender *messageSenderOnClient
 	// However, I have used a loop here to ensure that the query can still be executed normally even if this situation occurs.
 	for {
 		bat, end, err = sender.receiveBatch()
-		if err != nil || end || bat == nil {
+		if err != nil {
 			return err
 		}
+		if end || bat == nil {
+			sender.mergeRemoteWarnings(s.Proc)
+			return nil
+		}
 		bat.Clean(mp)
 	}
 }
@@ -244,9 +248,13 @@ func receiveMessageFromCnServerIfConnector(s *Scope, sender *messageSenderOnClie
 	nextChannel := s.RootOp.(*connector.Connector).Reg.Ch2
 	for {
 		bat, end, err = sender.receiveBatch()
-		if err != nil || end || bat == nil {
+		if err != nil {
 			return err
 		}
+		if end || bat == nil {
+			sender.mergeRemoteWarnings(s.Proc)
+			return nil
+		}
 		connectorAnalyze.Network(bat)
 
 		if err = forwardRemoteBatchWithContext(sender, nextChannel, bat, mp); err != nil {
@@ -282,9 +290,13 @@ func receiveMessageFromCnServerIfDispatch(s *Scope, sender *messageSenderOnClien
 	mp := s.Proc.Mp()
 	for {
 		bat, end, err = sender.receiveBatch()
-		if err != nil || end || bat == nil {
+		if err != nil {
 			return err
 		}
+		if end || bat == nil {
+			sender.mergeRemoteWarnings(s.Proc)
+			return nil
+		}
 
 		dispatchAnalyze.Network(bat)
 		fakeValueScanOperator.Batchs = append(fakeValueScanOperator.Batchs, bat)
@@ -363,6 +375,9 @@ type messageSenderOnClient struct {
 	// gaugeDecOnce ensures PipelineMessageSenderGauge.Dec() is called at most once when close() runs
 	// (including when close() returns early because alreadyClose is true, so the gauge still decrements).
 	gaugeDecOnce sync.Once
+
+	// remoteWarningCount accumulates warnings returned by remote MessageEnd packets.
+	remoteWarningCount uint32
 }
 
 func newMessageSenderOnClient(
@@ -484,6 +499,7 @@ func (sender *messageSenderOnClient) receiveBatch() (bat *batch.Batch, over bool
 		}
 		if m.IsEndMessage() {
 			sender.safeToClose = true
+			sender.remoteWarningCount += m.GetWarningCount()
 
 			anaData := m.GetAnalyse()
 			if len(anaData) > 0 {
@@ -518,6 +534,14 @@ func (sender *messageSenderOnClient) receiveBatch() (bat *batch.Batch, over bool
 	}
 }
 
+func (sender *messageSenderOnClient) mergeRemoteWarnings(proc *process.Process) {
+	if sender.remoteWarningCount == 0 || proc == nil {
+		return
+	}
+	proc.Base.Warnings.Add(sender.remoteWarningCount)
+	sender.remoteWarningCount = 0
+}
+
 func (sender *messageSenderOnClient) contextDoneError() error {
 	if sender.ctx == nil {
 		return nil

pkg/sql/compile/remoterunClient_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/defines"
+	"github.com/matrixorigin/matrixone/pkg/pb/pipeline"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec/connector"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
@@ -93,6 +94,39 @@ func Test_newMessageSenderOnClient(t *testing.T) {
 	client.waitingTheStopResponse()
 }
 
+func TestMessageSenderMergeRemoteWarnings(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	sender := &messageSenderOnClient{remoteWarningCount: 7}
+
+	sender.mergeRemoteWarnings(proc)
+	assert.Equal(t, uint32(7), proc.Base.Warnings.Load())
+	assert.Equal(t, uint32(0), sender.remoteWarningCount)
+
+	// idempotent when no remote warnings left.
+	sender.mergeRemoteWarnings(proc)
+	assert.Equal(t, uint32(7), proc.Base.Warnings.Load())
+}
+
+func TestMessageSenderReceiveBatchCapturesWarningCount(t *testing.T) {
+	ctx := context.Background()
+	sender := &messageSenderOnClient{
+		ctx:       ctx,
+		receiveCh: make(chan morpc.Message, 1),
+		mp:        mpool.MustNewZero(),
+	}
+
+	endMsg := &pipeline.Message{}
+	endMsg.SetSid(pipeline.Status_MessageEnd)
+	endMsg.SetWarningCount(5)
+	sender.receiveCh <- endMsg
+
+	bat, over, err := sender.receiveBatch()
+	assert.NoError(t, err)
+	assert.Nil(t, bat)
+	assert.True(t, over)
+	assert.Equal(t, uint32(5), sender.remoteWarningCount)
+}
+
 func TestRemoteRun(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()

pkg/sql/compile/remoterunServer.go

@@ -220,6 +220,7 @@ func handlePipelineMessage(receiver *messageReceiverOnServer) error {
 		if err == nil {
 			runCompile.GenPhyPlan(runCompile)
 			receiver.phyPlan = runCompile.anal.GetPhyPlan()
+			receiver.warningCount = runCompile.proc.Base.Warnings.Load()
 		}
 		return err
 
@@ -299,6 +300,9 @@ type messageReceiverOnServer struct {
 
 	// result.
 	phyPlan *models.PhyPlan
+
+	// warningCount records warnings generated during remote execution.
+	warningCount uint32
 }
 
 func newMessageReceiverOnServer(
@@ -488,6 +492,7 @@ func (receiver *messageReceiverOnServer) sendEndMessage() error {
 	message.SetSid(pipeline.Status_MessageEnd)
 	message.SetID(receiver.messageId)
 	message.SetMessageType(receiver.messageTyp)
+	message.SetWarningCount(receiver.warningCount)
 
 	jsonData, err := json.MarshalIndent(receiver.phyPlan, "", "  ")
 	if err != nil {

pkg/sql/compile/remoterun_test.go

@@ -866,7 +866,7 @@ func TestReceiveMsgAndForward_ReturnsOnBlockedReceiverCancel(t *testing.T) {
 
 	done := make(chan error, 1)
 	go func() {
-		done <- receiveMsgAndForward(sender, forwardCh)
+		done <- receiveMsgAndForward(sender, forwardCh, nil)
 	}()
 
 	cancel()
@@ -881,6 +881,25 @@ func TestReceiveMsgAndForward_ReturnsOnBlockedReceiverCancel(t *testing.T) {
 	}
 }
 
+func TestReceiveMsgAndForward_MergesRemoteWarningsOnEnd(t *testing.T) {
+	proc := testutil.NewProcess(t)
+	forwardCh := make(chan process.PipelineSignal, 1)
+
+	sender := &messageSenderOnClient{
+		ctx:       context.Background(),
+		mp:        proc.Mp(),
+		receiveCh: make(chan morpc.Message, 1),
+	}
+	endMsg := &pipeline.Message{}
+	endMsg.SetSid(pipeline.Status_MessageEnd)
+	endMsg.SetWarningCount(6)
+	sender.receiveCh <- endMsg
+
+	err := receiveMsgAndForward(sender, forwardCh, proc)
+	require.NoError(t, err)
+	require.Equal(t, uint32(6), proc.Base.Warnings.Load())
+}
+
 func TestReceiveMessageFromCnServerIfDispatch_PreservesCleanupOnOriginalRoot(t *testing.T) {
 	proc := testutil.NewProcess(t)
 

pkg/sql/compile/scope.go

@@ -864,7 +864,7 @@ func (s *Scope) sendNotifyMessage(wg *sync.WaitGroup, resultChan chan notifyMess
 				sender.safeToClose = false
 				sender.alreadyClose = false
 
-				err = receiveMsgAndForward(sender, s.Proc.Reg.MergeReceivers[receiverIdx].Ch2)
+				err = receiveMsgAndForward(sender, s.Proc.Reg.MergeReceivers[receiverIdx].Ch2, s.Proc)
 				closeWithError(err, s.Proc.Reg.MergeReceivers[receiverIdx], sender)
 			},
 		)
@@ -885,12 +885,20 @@ func suppressRemoteRunCancelError(procCtx context.Context, err error) error {
 	return err
 }
 
-func receiveMsgAndForward(sender *messageSenderOnClient, forwardCh chan process.PipelineSignal) error {
+func receiveMsgAndForward(
+	sender *messageSenderOnClient,
+	forwardCh chan process.PipelineSignal,
+	proc *process.Process,
+) error {
 	for {
 		bat, end, err := sender.receiveBatch()
-		if err != nil || end || bat == nil {
+		if err != nil {
 			return err
 		}
+		if end || bat == nil {
+			sender.mergeRemoteWarnings(proc)
+			return nil
+		}
 
 		if err = forwardRemoteBatchWithContext(sender, forwardCh, bat, sender.mp); err != nil {
 			return err

pkg/sql/parsers/tree/update.go

@@ -173,6 +173,7 @@ type ExParam struct {
 	Local       bool
 	Parallel    bool
 	Strict      bool
+	IgnoreError bool
 }
 
 type S3Parameter struct {

pkg/sql/plan/bind_load.go

@@ -50,6 +50,9 @@ func (builder *QueryBuilder) bindExternalScan(
 
 	ctx := builder.compCtx
 	stmt.Param.Local = stmt.Local
+	if _, ok := stmt.DuplicateHandling.(*tree.DuplicateKeyIgnore); ok {
+		stmt.Param.IgnoreError = true
+	}
 	fileName, err := checkFileExist(stmt.Param, ctx)
 	if err != nil {
 		return -1, nil, err

pkg/sql/plan/build_load.go

@@ -223,6 +223,9 @@ func buildLoad(stmt *tree.Load, ctx CompilerContext, isPrepareStmt bool) (*Plan,
 	}
 
 	stmt.Param.Local = stmt.Local
+	if _, ok := stmt.DuplicateHandling.(*tree.DuplicateKeyIgnore); ok {
+		stmt.Param.IgnoreError = true
+	}
 	fileName, err := checkFileExist(stmt.Param, ctx)
 	if err != nil {
 		return nil, err

pkg/sql/util/csvparser/csv_parser.go

@@ -240,6 +240,20 @@ func (parser *CSVParser) Read(lastRow []Field) (row []Field, err error) {
 	return row, err
 }
 
+// RecoverAfterReadError discards bytes until the next record terminator so the
+// next Read starts from a new physical line after a parse error.
+func (parser *CSVParser) RecoverAfterReadError() error {
+	parser.recordBuffer = parser.recordBuffer[:0]
+	parser.fieldIndexes = parser.fieldIndexes[:0]
+	parser.fieldIsQuoted = parser.fieldIsQuoted[:0]
+
+	_, _, err := parser.readUntilTerminator()
+	if err == io.EOF {
+		return nil
+	}
+	return err
+}
+
 func (parser *CSVParser) Pos() int64 {
 	return parser.pos
 }