README
¶
About Chunkers
Chunkers are used to split a table into multiple chunks for copying. The downside of a chunk being too large is as follows:
- Replicas fall behind the master (we don't support read-replicas, but at a certain point it impacts DR).
- Locks are held for the duration of a chunk copy, so a large chunk can block other operations on the table.
All chunkers support "dynamic chunking" which is our way of saying that from a configuration perspective you specify the ideal chunk size in time-based units (e.g. 500ms) and the chunker will adjust the chunk size to try to meet that target. This tends to be a better approach than specifying a fixed chunk size, because the chunk size can vary wildly depending on the table. As the _new table gets larger, we typically see the chunk size reduce significantly to compensate for larger insert times. We believe this is more likely to occur on Aurora than MySQL because on IO bound workloads it does not have the change buffer.
Our thoughts are that spirit should be aggressive in copying, but there should only be minimal elevation in p99 response times. If you consider that a table regularly has DML queries that take 1-5ms, then it is reasonable to assume a chunk-time of 500ms will elevate some query to 505ms. Assuming this contention is limited, it may only be observed by the pMax and not the p99. It is usually application-dependent how much of a latency hit is acceptable. Our belief is that 500ms is on the high-end of acceptable for defaults, and users will typically lower it rather than increase it. We limit the maximum chunk time to 5s because it is unlikely that users can tolerate larger than a 5s latency hit for a single query on an OLTP system. Since we also adjust various lock wait timeouts based on the assumption that chunks are about this size, increasing beyond 5s would require additional tuning.
Chunking becomes a complicated problem because data can have an uneven distribution, and some tables have composite or strange data types for PRIMARY KEYs. We have chosen to solve the chunking problem by not using a one-size-fits-all approach, but rather an interface that has two current implementations: composite and optimistic.
Composite Chunker
The composite chunker is our newest chunker, and it is selected unless the table has an AUTO_INCREMENT single-column PRIMARY KEY.
Its implementation is very similar to how the chunker in gh-ost works:
- A
SELECTstatement is performed to find the exact PK value of the row that ischunkSizelarger than the current chunk pointer:
SELECT pk FROM table WHERE pk > chunkPointer ORDER BY pk LIMIT 1 OFFSET {chunkSize}
- An
INSERT .. SELECTstatement is run on the table to copy between the last chunk pointer and the new value.
The composite chunker is very good at dividing the chunks up equally, since sparing a brief race condition each chunk will match exactly the chunkSize value. The main downside is that it becomes a little bit wasteful when you have an auto_increment PRIMARY KEYs and rarely delete data. In this case, you waste the initial SELECT statement, since the client could easily calculate the next chunk pointer by adding chunkSize to the previous chunk pointer. A second issue is that the composite chunker always returns FALSE for the KeyAboveHighWatermark optimization. It is possible that it could be implemented correctly in future, but for code-simplicity we have chosen not to for now.
Many of our use-cases have auto_increment PRIMARY KEYs, so despite the composite chunker also being able to support non-composite PRIMARY KEYs, we have no plans to switch to it entirely.
Optimistic Chunker
The optimistic chunker was our first chunker, and it's ideal for one of our main use case: AUTO_INCREMENT PRIMARY KEYs.
Its basic implementation is as follows:
- Find the min/max of the
PRIMARY KEYcolumn. - Have a special chunk retrieve values less than the min value*
- Advance by chunk size until you reach the max value.
- Have a special chunk retrieve values greater than the max value*
* Because the value is cached, it's possible a small number of rows exist, particularly at the end of the table.
To a certain extent, the chunk-size will automatically adjust to small gaps in the table as dynamic chunking adjusts to compensate for slightly faster copies. However, this is intentionally limited with dynamic chunking having a hard limit on the chunk size of 100K rows. It can also only expand the chunk-size by 50% at a time. This helps prevent the scenario that quickly processed chunks (likely caused by table gaps) expand the chunk size too quickly, causing future chunks to be too large and causing QoS issues.
To deal with large gaps, the optimistic chunker also supports a special "prefetching mode". The prefetching mode is enabled when the chunk size has already reached the 100K limit, and each chunk is still only taking 20% of the target time for chunk copying. Prefetching was first developed when we discovered a user with ~20 million rows in the table but a big gap between the auto_increment value of 20 million and the end of the table (300 billion). You can think of the prefetching mode as not that much different from how the composite chunker works, as it will perform a SELECT query to find the next PK value it should use as a pointer. Prefetching is automatically disabled again if the chunk size is ever reduced below the 100K limit.
Documentation
¶
Overview ¶
Package table contains some common utilities for working with tables such as a 'Chunker' feature.
Index ¶
- Constants
- Variables
- func LazyFindP90(a []time.Duration) time.Duration
- func QuoteColumns(cols []string) string
- type Boundary
- type Chunk
- type Chunker
- type Datum
- type JSONBoundary
- type JSONChunk
- type Operator
- type TableInfo
- func (t *TableInfo) AutoUpdateStatistics(ctx context.Context, interval time.Duration, logger loggers.Advanced)
- func (t *TableInfo) Close() error
- func (t *TableInfo) DescIndex(keyName string) ([]string, error)
- func (t *TableInfo) MaxValue() Datum
- func (t *TableInfo) PrimaryKeyIsMemoryComparable() error
- func (t *TableInfo) PrimaryKeyValues(row interface{}) ([]interface{}, error)
- func (t *TableInfo) SetInfo(ctx context.Context) error
- func (t *TableInfo) WrapCastType(col string) string
Constants ¶
const ( // StartingChunkSize is the initial chunkSize StartingChunkSize = 1000 // MaxDynamicStepFactor is the maximum amount each recalculation of the dynamic chunkSize can // increase by. For example, if the newTarget is 5000 but the current target is 1000, the newTarget // will be capped back down to 1500. Over time the number 5000 will be reached, but not straight away. MaxDynamicStepFactor = 1.5 // MinDynamicRowSize is the minimum chunkSize that can be used when dynamic chunkSize is enabled. // This helps prevent a scenario where the chunk size is too small (it can never be less than 1). MinDynamicRowSize = 10 // MaxDynamicRowSize is the max allowed chunkSize that can be used when dynamic chunkSize is enabled. // This seems like a safe upper bound for now MaxDynamicRowSize = 100000 // DynamicPanicFactor is the factor by which the feedback process takes immediate action when // the chunkSize appears to be too large. For example, if the PanicFactor is 5, and the target *time* // is 50ms, an actual time 250ms+ will cause the dynamic chunk size to immediately be reduced. DynamicPanicFactor = 5 // ChunkerDefaultTarget is the default chunker target ChunkerDefaultTarget = 100 * time.Millisecond )
Variables ¶
Functions ¶
func LazyFindP90 ¶
LazyFindP90 finds the second to last value in a slice. This is the same as a p90 if there are 10 values, but if there were 100 values it would technically be a p99 etc.
func QuoteColumns ¶
Types ¶
type Chunk ¶
type Chunk struct {
Key []string
ChunkSize uint64
LowerBound *Boundary
UpperBound *Boundary
AdditionalConditions string
}
Chunk is returned by chunk.Next() Applications can use it to iterate over the rows.
type Chunker ¶
type Chunker interface {
Open() error
OpenAtWatermark(string, Datum) error
IsRead() bool
Close() error
Next() (*Chunk, error)
Feedback(*Chunk, time.Duration)
GetLowWatermark() (string, error)
KeyAboveHighWatermark(interface{}) bool
}
func NewChunker ¶
type Datum ¶
type Datum struct {
Val interface{}
Tp datumTp // signed, unsigned, binary
}
Datum could be a binary string, uint64 or int64.
func NewNilDatum ¶
func NewNilDatum(tp datumTp) Datum
func (Datum) GreaterThanOrEqual ¶
type JSONBoundary ¶
type JSONChunk ¶
type JSONChunk struct {
Key []string
ChunkSize uint64
LowerBound JSONBoundary
UpperBound JSONBoundary
}
type TableInfo ¶
type TableInfo struct {
sync.Mutex
EstimatedRows uint64
SchemaName string
TableName string
QuotedName string
Columns []string // all the column names
KeyColumns []string // the column names of the primaryKey
KeyIsAutoInc bool // if pk[0] is an auto_increment column
DisableAutoUpdateStatistics atomic.Bool
// contains filtered or unexported fields
}
func (*TableInfo) AutoUpdateStatistics ¶
func (t *TableInfo) AutoUpdateStatistics(ctx context.Context, interval time.Duration, logger loggers.Advanced)
AutoUpdateStatistics runs a loop that updates the table statistics every interval. This will continue until Close() is called on the tableInfo, or t.DisableAutoUpdateStatistics is set to true.
func (*TableInfo) PrimaryKeyIsMemoryComparable ¶
PrimaryKeyIsMemoryComparable checks that the PRIMARY KEY type is compatible. We no longer need this check for the chunker, since it can handle any type of key in the composite chunker. But the migration still needs to verify this, because of the delta map feature, which requires binary comparable keys.
func (*TableInfo) PrimaryKeyValues ¶
PrimaryKeyValues helps extract the PRIMARY KEY from a row image. It uses our knowledge of the ordinal position of columns to find the position of primary key columns (there might be more than one). For minimal row image, you need to send the before image to extract the PK. This is because in the after image, the PK might be nil.
func (*TableInfo) SetInfo ¶
SetInfo reads from MySQL metadata (usually infoschema) and sets the values in TableInfo.
Source Files