TiFlow Documentation

TiFlow is a unified data replication platform designed for TiDB. It provides tools for both initial data movement and continuous change data replication through two primary components:

1. TiDB Data Migration (DM): Used for full data migration and incremental data replication from MySQL or MariaDB into TiDB.
2. TiCDC: Used to replicate change data from TiDB to downstream systems, including Kafka and other MySQL protocol-compatible databases.

Important Version Note: The TiCDC component included in this repository is only intended for use with v8.5.x and lower (the old architecture). For versions v8.5.4 and higher, you must use the new TiCDC architecture hosted at github.com/pingcap/ticdc.

The TiCDC Storage Sink is a design implementation that allows TiCDC to output incremental changelogs to external storage services. This enables scalable and cost-effective management of TiDB changelogs and supports building end-to-end data integration pipelines.

Supported storage backends include:

• NFS
• Amazon S3
• Google Cloud Platform (GCP) Blob Storage
• Azure Blob Storage

The TiCDC Dashboard provides visibility into the health and performance of TiCDC clusters through several metric categories. Monitoring these metrics allows you to track synchronization progress, resource usage, and internal data flow latency.

Key metric categories include:

• Server: Summary of TiKV and TiCDC node status (Uptime, CPU, Memory, Goroutines).
• Changefeed: Detailed information on synchronization tasks (Progress/Checkpoints, Lag, Sink write duration, Error counts).
• Events: Internal data flow details (RPC counts, buffer sizes, sorting/merging durations, unmarshal latency).
• Unified Sorter: Performance of the memory/disk hybrid sorter (Intake/Output rates, data sizes, flush/merge sizes).
• TiKV: Metrics related to TiKV's interaction with TiCDC (CDC endpoint CPU, resolved ts lag, initial scan duration).

The TiCDC Storage Sink feature is validated through four primary testing categories to ensure data consistency, stability, and performance:

1. Functional Tests: Ensures correctness of data replication using csv and canal-json protocols via unit and integration testing. It also includes manual verification of data synchronization across various external storage systems.
2. Scenario Tests: Focuses on stability and chaos testing under diverse workloads to ensure upstream and downstream data consistency and stable throughput/latency.
3. Compatibility Tests: Verifies compatibility with existing features and ensures there are no upgrade or downgrade compatibility issues (as it is a new feature).
4. Benchmark Tests: Evaluates performance across common, big data, multi-table, and wide-table scenarios using varying parameters.

The diff library provides functionality to compare data and structures between tables. It supports comparing tables with different names, comparing a single target table against multiple source tables, and generating SQL statements to fix data discrepancies in the target table.

To perform a comparison, you must construct a TableDiff struct and call its Equal method.

import "context"

// Assuming TableDiff and TableInstance are defined in the package
func Compare(ctx context.Context, td *TableDiff) {
    // Example usage pattern
    structEqual, dataEqual, err := td.Equal(ctx, func(sql string) error {
        // This callback is triggered to provide SQL for fixing data
        return nil
    })
    if err != nil {
        // handle error
    }
    // use structEqual and dataEqual to determine results
}

TiCDC supports four primary protocols for writing data changes to a message queue (like Kafka). A "protocol" in TiCDC refers to the serialization scheme combined with specific policies for message ordering and partitioning.

Ordering and Duplication Guarantees

• Ordering: All protocols guarantee that, if duplicate messages are ignored, data changes to a given table within a single partition are output in the order of their commit timestamps.
• Duplication: Duplicate messages may occur when a TiCDC node re-establishes a connection with TiKV or when a table's replication task is migrated between TiCDC nodes.

Old Value Support

• Avro: Does not support outputting the old value before a change.
• All other protocols: Support outputting the old value.

ChunkQueue is a memory-efficient, generic queue implementation for Go (1.18+). It is designed to be GC-friendly by using chunked memory allocation instead of a single large slice, which avoids redundant copies during push and pop operations.

Key Design Principles:

• Chunked Allocation: Memory is divided into equal-length segments called 'chunks'.
• Chunk Length: Determined at initialization as max{16, 1Kb / size of type T}.
• Efficiency: It minimizes GC pressure and provides optimized bulk operations using copy().

In TiFlow (DM), congestion occurs when the data import or replication rate exceeds the downstream TiDB's capacity. This leads to decreased service quality, characterized by:

• Increased Latency: Transaction execution time grows (often exponentially as concurrency increases).
• Decreased TPS (Transactions Per Second): Throughput plateaus or drops even as concurrency increases.
• Increased Transaction Failure Rate: Higher frequency of downstream resource busy errors (e.g., tmysql.ErrTiKVServerBusy, tmysql.ErrTiKVServerTimeout).

To prevent downstream overload, TiFlow implements a dynamic concurrency control framework that automatically adjusts the data flow speed based on detected congestion.

Avro is an Apache serialization format highly compatible with the Confluent Platform (Kafka).

Key Features:

• Compatibility: Native support in Confluent Platform; can be parsed by Kafka Connect.
• Limitations:
  • No Timestamps: Current version contains only row values in appropriate Avro types; it does not include commit timestamps.
  • No Transactions: Cannot be used to restore transactions.
  • No DDL: Does not provide DDL information.
• Old Values: Does not support outputting the old value before a change.