Make stale read and history read compatible with DDL

[djshow832]

Background

This is a subtask of #21094.

The executor and coprocessor always read the newest schema, even if in a staleness transaction or it's a history read. If a schema change happens after the specified timestamp, some cases may happen:

DDL without data reorganization

In this case, the schema change needs no data reorganization, which means only the metadata changes but not table data. Most of the DDL is in this case.

Since the table data format stays the same, the data can still be parsed with the newer schema.

E.g.

ALTER TABLE tbl ADD COLUMN x INT;
START TRANSACTION READ ONLY WITH TIMESTAMP BOUND EXACT STALENESS '00:00:30';
SELECT * FROM tbl LIMIT 1;         # Returns `x` because the schema is the newest

In this case, the result contains the latest table structure even if the data is older. This is acceptable in most cases because the user applies staleness transactions to reduce cross-region latency and release read hotspot rather than reading history data.

There may be some DDL that affects the read result but it doesn't occur to me for now.

DDL with data reorganization

In this case, the schema change needs to reorganize table data, which means some or all of the table data will be reformated.
So far there are only 3 kinds of such DDL:

1. adding indexes
2. changing column types
3. dropping/exchanging partitions from a table with a global index

E.g.

ALTER TABLE tbl ADD INDEX idx(x);
START TRANSACTION READ ONLY WITH TIMESTAMP BOUND EXACT STALENESS '00:00:30';
SELECT * FROM tbl WHERE x=1;             # The optimizer may use the index `idx` to query but the transaction won't read the index

ALTER TABLE tbl MODIFY x VARCHAR(10);          # Change from JSON to VARCHAR, which needs to modify all of the data
START TRANSACTION READ ONLY WITH TIMESTAMP BOUND EXACT STALENESS '00:00:30';
SELECT x FROM tbl WHERE id=1;            # The old stored data format is in JSON type but it will be parsed as VARCHAR type

In these cases, some of these problems will occur:

1. TiDB will panic or report an error
2. The returned result is wrong
3. The returned result is in the latest schema

Solutions

For the DDL without data reorganization, as the result is acceptable, we just need to declare in the document that the schema is always the latest.

For the DDL with data reorganization, there are some possible solutions:

1. Cache the past schema in both TiDB and TiKV and use the past schema in a stale read or history head. This is almost impossible because the workload is huge.
2. Forbid reading the tables affected by DDL that satisfies the following conditions:
   • A DDL is executed after the specified timestamp in the staleness transaction or history read
   • The DDL involves data reorganization
   • The DDL affects the tables to be read

E.g.

ALTER TABLE tbl ADD INDEX idx(x);         # Executed at 2021-01-01 00:01:00
START TRANSACTION READ ONLY WITH TIMESTAMP BOUND READ TIMESTAMP '2021-01-01 00:00:00';
SELECT * FROM tbl WHERE x=1;             # Returns an error because a DDL was executed

Implementations

Firstly, we need to collect the DDL info. Secondly, we need to look up the tables against the DDL in staleness transactions or history reads.

Collect the DDL which needs data reorganization

DDL info of each DDL needs to be cached in a list. The DDL info includes the schema version, the DDL type, and affected table ids.

schemaValidator.deltaSchemaInfos is a similar list that contains recent schema changes. It is mainly used to validate that the schema of tables affected by one transaction is not changed during the transaction. See schemaValidator.isRelatedTablesChanged.

However, its capacity is 1024 by default and it contains all schema changes, not only those with data reorganization. So there may a possibility when the transaction is too old and the DDL info list runs out, just like what schemaValidator.isRelatedTablesChanged reports.

Get the schema version for the start ts of the staleness transaction

We need to compare the time order of transaction start ts (or the snapshot time of a history read) with schema changes.

For normal transactions, it is done by comparing the schema version when the transaction starts (TransactionContext.SchemaVersion) with the schema version of each schema change, just like schemaValidator.isRelatedTablesChanged.

However, for staleness transactions, the schema version recorded in the transaction context should be the one corresponding to the transaction start ts, rather than the one when the transaction really starts.

One way is to get the DDL job history from the metadata, like what GetDDLJobs does. In this way, we can get the start time for each schema version, but it needs to read the metadata on TiKV, which is slow. What's more, the start time is not accurate.

Check the DDL info list when reading tables

Each time the staleness transaction or history read reads a table, check the DDL info list. If there exists any DDL that affected the table, report an error.

Just like validating the transaction scopes in local transactions, we can also validate the schema of tables in all operators that reading tables directly. For example, validate the tables in RequestBuilder.Build to cover TableReader, IndexMergeReader and IndexReader.

[Yisaer]

/assign

[Yisaer]

For Get the schema version for the start ts of the staleness transaction, I think a possible way is to maintain a queue like deltaSchemaInfos in schemaValidator and it only receives the RelatedSchemaChange which need to cause the data reorgnization. The capacity of this queue is fixed, if the startTS of the stale read transaction is too old, tidb-server will forbid the query because it can't judge whether there exists the uncompatible ddl change.

[Yisaer]

maintain a queue like deltaSchemaInfos in schemaValidator

To maintain the queue, there are 2 problems we need to solve:

1. How to compare the schemaDiff and stale read startTS
2. If the tidb-server is restarted, the queue would be empty and stale read couldn't check the startTS with the queue.

For the second problem, I think the tidb-server could load the schemaDiff from the storage as the init job during being started.

For the first problem, I think it can be solved in the following way:

1. SchemaDiff adding a attributor as startTS to record the startTS of the transaction when running ddl job (which is not equal to the startTS of ddl job itself)
2. As each DDL job will be running in a new transaction, we can pass the startTS in the following ways: RunInNewTxn -> runDDLJob -> updateSchemaVersion

In this way, during loadInfoSchema, we can push the schema change with the startTS into the queue, then camparing the schema change and the stale read startTS during reading data phase.

Though recording the committs is better in this case, it is hard to pass the committs during do DDL job. As comparing the startTS could also guarantee the compatibility, I choose to record the startTS

[AilinKid]

One question:
How does the executor and coprocessor always read the newest schema comes from.

I think stale read is quite like a snapshot read, the latter will form a new informationchema with the specified timestamp in session context; then the following the select statement's start ts will be also set as the specified timestamp and its schema will be built as the cached one. (is here is session scope)

So as regards stale read syntax, seems we can form a new informationchema with the specified timestamp for a statement like START TRANSACTION READ ONLY WITH TIMESTAMP BOUND READ TIMESTAMP '2021-01-01 00:00:00';; then the latter select statement in this txn should also use the specified timestamp as start ts and its schema will be built as the cached one too. (is here is not session scope)

[AilinKid]

PTAL @djshow832 @Yisaer

[xhebox]

Should be closed by #24285.