Fixing SQL Server flink CDC setup

.github/workflows/flink_cdc.yml

@@ -115,19 +115,8 @@ jobs:
       - name: Run license check
         run: gem install rubyzip -v 2.3.0 && ./tools/ci/license_check.rb
 
-  migration_test:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Check out repository code
-        uses: actions/checkout@v4
-        with:
-          submodules: true
-      - name: Compile snapshot CDC version
-        run: mvn --no-snapshot-updates -B install -DskipTests
-      - name: Run migration tests
-        run: cd flink-cdc-migration-tests && mvn clean verify
-
   compile_and_test:
+    needs: license_check
     # Only run the CI pipeline for the flink-cdc-connectors repository
 #    if: github.repository == 'apache/flink-cdc-connectors'
     runs-on: ubuntu-latest
@@ -263,3 +252,100 @@ jobs:
             done
           fi
           exit 0
+
+
+  migration_test_ut:
+    needs: license_check
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out repository code
+        uses: actions/checkout@v4
+        with:
+          submodules: true
+      - name: Compile snapshot CDC version
+        run: mvn --no-snapshot-updates -B install -DskipTests
+      - name: Run migration tests
+        run: cd flink-cdc-migration-tests && mvn clean verify
+
+  pipeline_migration_test:
+    needs: migration_test_ut
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        java-version: [ '8', '11' ]
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Ruby
+        uses: ruby/setup-ruby@v1
+        with:
+          ruby-version: 3.0
+          bundler-cache: true # runs 'bundle install' and caches installed gems automatically
+      - uses: actions/setup-java@v4
+        with:
+          java-version: ${{ matrix.java-version }}
+          distribution: temurin
+          cache: maven
+      - name: Install dependencies
+        run: gem install terminal-table
+      - name: Prepare CDC versions
+        run: CDC_SOURCE_HOME=$PWD ruby tools/mig-test/prepare_libs.rb
+      - name: Prepare Flink distro
+        run: wget https://dlcdn.apache.org/flink/flink-1.18.1/flink-1.18.1-bin-scala_2.12.tgz && tar -xzvf flink-1.18.1-bin-scala_2.12.tgz
+        working-directory: ./tools/mig-test
+      - name: Patch Flink configs
+        run: FLINK_HOME=./flink-1.18.1/ ruby misc/patch_flink_conf.rb
+        working-directory: ./tools/mig-test
+      - name: Start containers
+        run: cd conf && docker-compose up -d
+        working-directory: ./tools/mig-test
+      - name: Run migration tests
+        run: FLINK_HOME=./flink-1.18.1/ ruby run_migration_test.rb
+        working-directory: ./tools/mig-test
+      - name: Stop containers
+        if: always()
+        run: cd conf && docker-compose down
+        working-directory: ./tools/mig-test
+
+  data_stream_migration_test:
+    needs: migration_test_ut
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        java-version: [ '8', '11' ]
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Ruby
+        uses: ruby/setup-ruby@v1
+        with:
+          ruby-version: 3.0
+          bundler-cache: true # runs 'bundle install' and caches installed gems automatically
+      - uses: actions/setup-java@v4
+        with:
+          java-version: ${{ matrix.java-version }}
+          distribution: temurin
+          cache: maven
+      - name: Install dependencies
+        run: gem install terminal-table
+      - name: Prepare CDC versions
+        run: CDC_SOURCE_HOME=$PWD ruby tools/mig-test/prepare_libs.rb
+      - name: Prepare Flink distro
+        run: wget https://dlcdn.apache.org/flink/flink-1.18.1/flink-1.18.1-bin-scala_2.12.tgz && tar -xzvf flink-1.18.1-bin-scala_2.12.tgz
+        working-directory: ./tools/mig-test
+      - name: Patch Flink configs
+        run: FLINK_HOME=./flink-1.18.1/ ruby misc/patch_flink_conf.rb
+        working-directory: ./tools/mig-test
+      - name: Compile Dummy DataStream Jobs
+        run: cd datastream && ruby compile_jobs.rb
+        working-directory: ./tools/mig-test
+      - name: Start containers
+        run: cd conf && docker-compose up -d
+        working-directory: ./tools/mig-test
+      - name: Run migration tests
+        run: cd datastream && FLINK_HOME=../flink-1.18.1/ ruby run_migration_test.rb
+        working-directory: ./tools/mig-test
+      - name: Stop containers
+        if: always()
+        run: cd conf && docker-compose down
+        working-directory: ./tools/mig-test

README.md

@@ -53,19 +53,19 @@ full database synchronization, sharding table synchronization, schema evolution
       password: pass
 
     transform:
-    - source-table: adb.web_order01
-      projection: \*, UPPER(product_name) as product_name
-      filter: id > 10 AND order_id > 100
-      description: project fields and filter
-    - source-table: adb.web_order02
-      projection: \*, UPPER(product_name) as product_name
-      filter: id > 20 AND order_id > 200
-      description: project fields and filter  
+      - source-table: adb.web_order01
+        projection: \*, UPPER(product_name) as product_name
+        filter: id > 10 AND order_id > 100
+        description: project fields and filter
+      - source-table: adb.web_order02
+        projection: \*, UPPER(product_name) as product_name
+        filter: id > 20 AND order_id > 200
+        description: project fields and filter  
 
     route:
-    - source-table: adb.web_order\.*
-      sink-table: adb.ods_web_orders
-      description: sync sharding tables to one destination table
+      - source-table: adb.web_order\.*
+        sink-table: adb.ods_web_orders
+        description: sync sharding tables to one destination table
 
     pipeline:
        name: MySQL to Doris Pipeline

docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md

@@ -261,7 +261,7 @@ Flink SQL> SELECT * FROM orders;
       <td style="word-wrap: break-word;">initial</td>
       <td>String</td>
       <td> MySQL CDC 消费者可选的启动模式，
-         合法的模式为 "initial"，"earliest-offset"，"latest-offset"，"specific-offset" 和 "timestamp"。
+         合法的模式为 "initial"，"earliest-offset"，"latest-offset"，"specific-offset"，"timestamp" 和 "snapshot"。
            请查阅 <a href="#a-name-id-002-a">启动模式</a> 章节了解更多详细信息。</td>
     </tr>
     <tr>
@@ -500,7 +500,7 @@ CREATE TABLE products (
 * （3）在快照读取之前，Source 不需要数据库锁权限。
 
 如果希望 source 并行运行，则每个并行 reader 都应该具有唯一的 server id，因此`server id`的范围必须类似于 `5400-6400`，
-且范围必须大于并行度。在增量快照读取过程中，MySQL CDC Source 首先通过表的主键将表划分成多个块（chunk），
+且范围必须大于并行度。在增量快照读取过程中，MySQL CDC Source 源首先会根据您指定的表块键将表分块(chunk）
 然后 MySQL CDC Source 将多个块分配给多个 reader 以并行读取表的数据。
 
 #### 并发读取
@@ -550,7 +550,7 @@ MySQL 集群中你监控的服务器出现故障后, 你只需将受监视的服
 
 当 MySQL CDC Source 启动时，它并行读取表的快照，然后以单并行度的方式读取表的 binlog。
 
-在快照阶段，根据表的主键和表行的大小将快照切割成多个快照块。
+在快照阶段，快照会根据表的分块键和表行的大小切割成多个快照块。
 快照块被分配给多个快照读取器。每个快照读取器使用 [区块读取算法](#snapshot-chunk-reading) 并将读取的数据发送到下游。
 Source 会管理块的进程状态（完成或未完成），因此快照阶段的 Source 可以支持块级别的 checkpoint。
 如果发生故障，可以恢复 Source 并继续从最后完成的块中读取块。
@@ -565,7 +565,9 @@ Flink 定期为 Source 执行 checkpoint，在故障转移的情况下，作业
 
 在执行增量快照读取时，MySQL CDC source 需要一个用于分片的的算法。
 MySQL CDC Source 使用主键列将表划分为多个分片（chunk）。 默认情况下，MySQL CDC source 会识别表的主键列，并使用主键中的第一列作为用作分片列。
-如果表中没有主键， 增量快照读取将失败，你可以禁用 `scan.incremental.snapshot.enabled` 来回退到旧的快照读取机制。
+如果表中没有主键，用户必须指定 `scan.incremental.snapshot.chunk.key-column`、
+否则增量快照读取将失败，你可以禁用 `scan.incremental.snapshot.enabled` 恢复到旧的快照读取机制。
+请注意，使用不在主键中的列作为分块键可能会降低表的查询性能。
 
 对于数值和自动增量拆分列，MySQL CDC Source 按固定步长高效地拆分块。
 例如，如果你有一个主键列为`id`的表，它是自动增量 BIGINT 类型，最小值为`0`，最大值为`100`，
@@ -629,6 +631,7 @@ MySQLSource.builder()
     .startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 4L) // 从指定 binlog 文件名和位置启动
     .startupOptions(StartupOptions.specificOffset("24DA167-0C0C-11E8-8442-00059A3C7B00:1-19")) // 从 GTID 集合启动
     .startupOptions(StartupOptions.timestamp(1667232000000L) // 从时间戳启动
+    .startupOptions(StartupOptions.snapshot()) // 仅读取快照
     ...
     .build()
 ```
@@ -642,6 +645,7 @@ CREATE TABLE mysql_source (...) WITH (
     'scan.startup.mode' = 'latest-offset', -- 从最晚位点启动
     'scan.startup.mode' = 'specific-offset', -- 从特定位点启动
     'scan.startup.mode' = 'timestamp', -- 从特定位点启动
+    'scan.startup.mode' = 'snapshot', -- 仅读取快照
     'scan.startup.specific-offset.file' = 'mysql-bin.000003', -- 在特定位点启动模式下指定 binlog 文件名
     'scan.startup.specific-offset.pos' = '4', -- 在特定位点启动模式下指定 binlog 位置
     'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- 在特定位点启动模式下指定 GTID 集合

docs/content.zh/docs/connectors/flink-sources/overview.md

@@ -37,7 +37,7 @@ You can also read [tutorials]({{< ref "docs/connectors/flink-sources/tutorials/b
 
 | Connector                                                                  | Database                                                                                                                                                                                                                                                                                                                                                                                               | Driver                    |
 |----------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------|
-| [mongodb-cdc]({{< ref "docs/connectors/flink-sources/mongodb-cdc" >}})     | <li> [MongoDB](https://www.mongodb.com): 3.6, 4.x, 5.0                                                                                                                                                                                                                                                                                                                                                 | MongoDB Driver: 4.3.4     |
+| [mongodb-cdc]({{< ref "docs/connectors/flink-sources/mongodb-cdc" >}})     | <li> [MongoDB](https://www.mongodb.com): 3.6, 4.x, 5.0, 6.0, 6.1, 7.0                                                                                                                                                                                                                                                                                                                                  | MongoDB Driver: 4.11.2    |
 | [mysql-cdc]({{< ref "docs/connectors/flink-sources/mysql-cdc" >}})         | <li> [MySQL](https://dev.mysql.com/doc): 5.6, 5.7, 8.0.x <li> [RDS MySQL](https://www.aliyun.com/product/rds/mysql): 5.6, 5.7, 8.0.x <li> [PolarDB MySQL](https://www.aliyun.com/product/polardb): 5.6, 5.7, 8.0.x <li> [Aurora MySQL](https://aws.amazon.com/cn/rds/aurora): 5.6, 5.7, 8.0.x <li> [MariaDB](https://mariadb.org): 10.x <li> [PolarDB X](https://github.com/ApsaraDB/galaxysql): 2.0.1 | JDBC Driver: 8.0.28       |
 | [oceanbase-cdc]({{< ref "docs/connectors/flink-sources/oceanbase-cdc" >}}) | <li> [OceanBase CE](https://open.oceanbase.com): 3.1.x, 4.x <li> [OceanBase EE](https://www.oceanbase.com/product/oceanbase): 2.x, 3.x, 4.x                                                                                                                                                                                                                                                            | OceanBase Driver: 2.4.x   |
 | [oracle-cdc]({{< ref "docs/connectors/flink-sources/oracle-cdc" >}})       | <li> [Oracle](https://www.oracle.com/index.html): 11, 12, 19, 21                                                                                                                                                                                                                                                                                                                                       | Oracle Driver: 19.3.0.0   |

docs/content.zh/docs/connectors/pipeline-connectors/mysql.md

@@ -172,7 +172,7 @@ pipeline:
       <td style="word-wrap: break-word;">initial</td>
       <td>String</td>
       <td> MySQL CDC 消费者可选的启动模式，
-         合法的模式为 "initial"，"earliest-offset"，"latest-offset"，"specific-offset" 和 "timestamp"。</td>
+         合法的模式为 "initial"，"earliest-offset"，"latest-offset"，"specific-offset"，"timestamp" 和 ""snapshot"。</td>
     </tr>
     <tr>
       <td>scan.startup.specific-offset.file</td>
@@ -280,6 +280,7 @@ pipeline:
 - `latest-offset`：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
 - `specific-offset`：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。
 - `timestamp`：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。
+- `snapshot`: 只进行快照阶段，跳过增量阶段，快照阶段读取结束后退出。
 
 例如，可以在 YAML 配置文件中这样指定启动模式：
 
@@ -290,6 +291,7 @@ source:
   scan.startup.mode: latest-offset                      # Start from latest offset
   scan.startup.mode: specific-offset                    # Start from specific offset
   scan.startup.mode: timestamp                          # Start from timestamp
+  scan.startup.mode: snapshot                          # Read snapshot only
   scan.startup.specific-offset.file: 'mysql-bin.000003' # Binlog filename under specific offset startup mode
   scan.startup.specific-offset.pos: 4                   # Binlog position under specific offset mode
   scan.startup.specific-offset.gtid-set: 24DA167-...    # GTID set under specific offset startup mode

docs/content.zh/docs/core-concept/data-pipeline.md

@@ -79,6 +79,17 @@ We could use following yaml file to define a complicated Data Pipeline describin
      fenodes: 127.0.0.1:8030
      username: root
      password: ""
+
+   transform:
+     - source-table: adb.web_order01
+       projection: \*, UPPER(product_name) as product_name
+       filter: id > 10 AND order_id > 100
+       description: project fields and filter
+     - source-table: adb.web_order02
+       projection: \*, UPPER(product_name) as product_name
+       filter: id > 20 AND order_id > 200
+       description: project fields and filter
+
    route:
      - source-table: app_db.orders
        sink-table: ods_db.ods_orders