*: Support batch load data #2394
Conversation
| @@ -540,7 +552,8 @@ func (e *LoadData) Close() error { | |||
|
|
|||
| // InsertValues is the data to insert. | |||
| type InsertValues struct { | |||
| currRow int | |||
| currRow int64 | |||
There was a problem hiding this comment.
I think don't it's necessary.
| @@ -630,8 +630,9 @@ func (s *testSuite) TestLoadData(c *C) { | |||
| deleteSQL := "delete from load_data_test" | |||
| selectSQL := "select * from load_data_test;" | |||
| // data1 = nil, data2 = nil, fields and lines is default | |||
| _, err = ld.InsertData(nil, nil) | |||
| _, reachLimit, err := ld.InsertData(nil, nil) | |||
There was a problem hiding this comment.
add test for reach limit is true
There was a problem hiding this comment.
This test is in the file of server/tidb_test.go.
|
|
||
| txn := loadDataInfo.Ctx.Txn() | ||
| if err != nil { | ||
| if err1 := txn.Rollback(); err1 != nil { |
| break | ||
| } | ||
| loadDataInfo.Ctx.NewTxn() | ||
| shouldCommit = false |
| @@ -411,12 +417,18 @@ func (e *LoadDataInfo) InsertData(prevData, curData []byte) ([]byte, error) { | |||
| cols = escapeCols(rawCols) | |||
| e.insertData(cols) | |||
| e.insertVal.currRow++ | |||
| if e.insertVal.batchRows != 0 && e.insertVal.currRow%e.insertVal.batchRows == 0 { | |||
There was a problem hiding this comment.
I think use mod operator is not efficiently. You can let currRow = 0 every time it reaches batchRows, and then check currRow >= batchRows.
There was a problem hiding this comment.
The currRow is useful to other place.
| @@ -562,15 +550,38 @@ func (cc *clientConn) handleLoadData(loadDataInfo *executor.LoadDataInfo) error | |||
| break | |||
| } | |||
| } | |||
| prevData, err = loadDataInfo.InsertData(prevData, curData) | |||
| for { | |||
There was a problem hiding this comment.
This for loop can be extracted to a function.
| @@ -562,15 +550,38 @@ func (cc *clientConn) handleLoadData(loadDataInfo *executor.LoadDataInfo) error | |||
| break | |||
| } | |||
| } | |||
| prevData, err = loadDataInfo.InsertData(prevData, curData) | |||
| for { | |||
| prevData, shouldCommit, err = loadDataInfo.InsertData(prevData, curData) | |||
There was a problem hiding this comment.
shouldCommit or reachLimit? We better keep it consistent.
| @@ -603,6 +614,7 @@ func (cc *clientConn) handleQuery(sql string) (err error) { | |||
| } else { | |||
| loadDataInfo := cc.ctx.Value(executor.LoadDataVarKey) | |||
| if loadDataInfo != nil { | |||
| defer cc.ctx.SetValue(executor.LoadDataVarKey, nil) | |||
| if err = cc.handleLoadData(loadDataInfo.(*executor.LoadDataInfo)); err != nil { | |||
| return errors.Trace(err) | |||
There was a problem hiding this comment.
How about rollback or commit here, so we can return instead of break in handlLoadData
There was a problem hiding this comment.
I think it's OK to use break. And I want to put this logic in the same function.
…muxia/split-load-data Conflicts: server/conn.go
| @@ -0,0 +1,834 @@ | |||
| // Copyright 2016 PingCAP, Inc. | |||
There was a problem hiding this comment.
Why this file name is ":" ?
| return nil, errors.Trace(err) | ||
| } | ||
| loadDataInfo.Ctx.NewTxn() | ||
| reachLimit = false |
|
LGTM |
| if err = loadDataInfo.Ctx.Txn().Commit(); err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
| loadDataInfo.Ctx.NewTxn() |
There was a problem hiding this comment.
error is not checked.
And we can call NewTxn only, it will commit current transaction.
|
LGTM |
Fix #1771