This repository has been archived by the owner on Dec 8, 2021. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 66
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
*: parse the data source directly into data and skip the KV encoder (#…
…145) * *: parse the data source directly into data and skip the KV encoder This skips the more complex pingcap/parser, and speeds up parsing speed by 50%. We have also refactored the KV delivery mechanism to use channels directly, and revamped metrics: - Make the metrics about engines into its own `engines` counter. The `tables` counter is exclusively about tables now. - Removed `block_read_seconds`, `block_read_bytes`, `block_encode_seconds` since the concept of "block" no longer applies. Replaced by the equivalents named `row_***`. - Removed `chunk_parser_read_row_seconds` for being overlapping with `row_read_seconds`. - Changed `block_deliver_bytes` into a histogram vec, with kind=index or kind=data. Introduced `block_deliver_kv_pairs`. * tests,restore: prevent spurious error in checkpoint_chunks test Only kill Lightning if the whole chunk is imported exactly. The chunk checkpoint may be recorded before a chunk is fully written, and this will hit the failpoint more than 5 times. * kv: use composed interface to simplify some types * kv: properly handle the SQL mode * common: disable IsContextCanceledError() when log level = debug This helps debugging some mysterious cancellation where the log is inhibited. Added IsReallyContextCanceledError() for code logic affected by error type. * restore: made some log more detailed * restore: made the SlowDownImport failpoint apply to index engines too * restore: do not open a write stream when there are no KV pairs to send * tests: ensure we drop the checkpoints DB before re-run * mydump: fixed various off-by-one errors in the CSV parser * *: rename `!IsContextCanceledError` to `ShouldLogError` * *: addressed comments * restore: zero the checksums and column permutations on initialization * *: addressed comments * tests: add back a missing license header * tests: improve a comment.
- Loading branch information
Showing
42 changed files
with
4,715 additions
and
3,091 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| // Copyright 2019 PingCAP, Inc. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package kv | ||
|
|
||
| import ( | ||
| "github.com/pingcap/parser/mysql" | ||
| "github.com/pingcap/tidb/kv" | ||
| "github.com/pingcap/tidb/sessionctx" | ||
| "github.com/pingcap/tidb/sessionctx/variable" | ||
| kvec "github.com/pingcap/tidb/util/kvencoder" | ||
| ) | ||
|
|
||
| // transaction is a trimmed down Transaction type which only supports adding a | ||
| // new KV pair. | ||
| type transaction struct { | ||
| kv.Transaction | ||
| kvPairs []kvec.KvPair | ||
| } | ||
|
|
||
| // Set implements the kv.Transaction interface | ||
| func (t *transaction) Set(k kv.Key, v []byte) error { | ||
| t.kvPairs = append(t.kvPairs, kvec.KvPair{ | ||
| Key: k.Clone(), | ||
| Val: append([]byte{}, v...), | ||
| }) | ||
| return nil | ||
| } | ||
|
|
||
| // SetOption implements the kv.Transaction interface | ||
| func (t *transaction) SetOption(opt kv.Option, val interface{}) {} | ||
|
|
||
| // DelOption implements the kv.Transaction interface | ||
| func (t *transaction) DelOption(kv.Option) {} | ||
|
|
||
| // session is a trimmed down Session type which only wraps our own trimmed-down | ||
| // transaction type and provides the session variables to the TiDB library | ||
| // optimized for Lightning. | ||
| type session struct { | ||
| sessionctx.Context | ||
| txn transaction | ||
| vars *variable.SessionVars | ||
| } | ||
|
|
||
| func newSession(sqlMode mysql.SQLMode) *session { | ||
| vars := variable.NewSessionVars() | ||
| vars.LightningMode = true | ||
| vars.SkipUTF8Check = true | ||
| vars.StmtCtx.InInsertStmt = true | ||
| vars.StmtCtx.BadNullAsWarning = !sqlMode.HasStrictMode() | ||
| vars.StmtCtx.TruncateAsWarning = !sqlMode.HasStrictMode() | ||
| vars.StmtCtx.OverflowAsWarning = !sqlMode.HasStrictMode() | ||
| vars.StmtCtx.AllowInvalidDate = sqlMode.HasAllowInvalidDatesMode() | ||
| vars.StmtCtx.IgnoreZeroInDate = !sqlMode.HasStrictMode() || sqlMode.HasAllowInvalidDatesMode() | ||
| vars.StmtCtx.TimeZone = vars.Location() | ||
| return &session{ | ||
| txn: transaction{}, | ||
| vars: vars, | ||
| } | ||
| } | ||
|
|
||
| func (se *session) takeKvPairs() []kvec.KvPair { | ||
| pairs := se.txn.kvPairs | ||
| se.txn.kvPairs = make([]kvec.KvPair, 0, len(pairs)) | ||
| return pairs | ||
| } | ||
|
|
||
| // Txn implements the sessionctx.Context interface | ||
| func (se *session) Txn(active bool) (kv.Transaction, error) { | ||
| return &se.txn, nil | ||
| } | ||
|
|
||
| // GetSessionVars implements the sessionctx.Context interface | ||
| func (se *session) GetSessionVars() *variable.SessionVars { | ||
| return se.vars | ||
| } |
Oops, something went wrong.