Source變更相容性 - Realtime Compute for Apache Flink

本文為您介紹Source變更的可相容性和不可相容性詳情。

背景資訊

當前建表DDL的Schema部分的變更，需要根據使用該表的Query的變化來判斷整體的相容性。WITH中除了連接器類型，其他的屬性暫不進行相容性檢查。

可相容的變更

修改欄位，且不影響下遊運算元的相容性，屬於完全或部分相容變更。

-- 原始SQL。
create table MyTable (
  a int,
  b bigint,
  c varchar
) with (
  'connector'='datagen'
);

select a, sum(b), max(c) from MyTable group by a;


-- 新增欄位d，該修改屬於完全相容變更。
-- Query中未使用到欄位d。
create table MyTable (
  a int,
  b bigint,
  c varchar,
  d int
) with (
  'connector'='datagen'
);

select a, sum(b), max(c) from MyTable group by a;


-- 修改欄位：b -> b as d + 1，該修改屬於部分相容變更。
-- max(c) 的計算結果不受影響。b列發生變化，因此原始的sum(b)被認為刪除。
-- 其對應的狀態資料會被丟棄，新的sum(b)被認為新增指標，其值在作業啟動時開始計算。
create table MyTable (
  a int,
  d bigint,
  c varchar,
  b as d + 1
) with (
  'connector'='datagen'
);

select a, sum(b), max(c) from MyTable group by a;

修改Watermark間隔，屬於完全相容變更。
修改Primary Key，可能導致下遊相容性受到影響。例如，upsert key發生變化。

修改Primary Key或WITH參數中的分區列，可能導致source的狀態相容性受到影響。例如mysql-cdc參數中的scan.incremental.snapshot.chunk.key-column發生變化，Source全量階段的狀態可能無法使用。
Changelog模式發生變化，使得發送給下遊的訊息類型可能發生變化（例如Delete、Update_before、Update_after等），可能導致下遊有狀態節點的相容性受到影響。例如，Hologres連接器WTH參數cdcMode（是否採用CDC讀取Binlog）修改、MongoDB或Postgres以Upsert還是Retract方式讀取。

不相容的變更

修改連接器類型屬於不相容變更。

-- 原始SQL。
create table MyTable (
  a int,
  b bigint,
  c varchar
) with (
  'connector'='datagen'
);

select a, sum(b), max(c) from MyTable group by a;


-- Connnector類型變化：datagen -> kafka，該修改屬於不相容修改。
create table MyTable (
  a int,
  b bigint,
  c varchar
) with (
  'connector'='kafka'，
  ...
);

select a, sum(b), max(c) from MyTable group by a;

修改表名屬於不相容變更。

-- 原始SQL。
create table MyTable (
  a int,
  b bigint,
  c varchar
) with (
  'connector'='datagen'
);
select a, sum(b), max(c) from MyTable group by a;


-- 表名類型變化：MyTable -> MyTable2，該修改屬於不相容修改。
create table MyTable2 (
  a int,
  b bigint,
  c varchar
) with (
  'connector'='datagen'
);

select a, sum(b), max(c) from MyTable2 group by a;

修改欄位且影響下遊運算元的相容性，該修改屬於不相容變更。

-- 原始SQL。
create table MyTable1 (
  a int,
  b bigint
) with (
  'connector'='datagen'
);

create table MyTable2 (
  c int,
  d bigint
) with (
  'connector'='datagen'
);

select * from MyTable1 join MyTable2 on c = d;


-- MyTable2中新增e欄位，該修改屬於不相容變更。
-- join要求其輸入的欄位不能修改。
create table MyTable1 (
  a int,
  b bigint
) with (
  'connector'='datagen'
);

create table MyTable2 (
  c int,
  d bigint,
  e varchar
) with (
  'connector'='datagen'
);

-- 原始SQL。
select * from MyTable1 join MyTable2 on c = d;