Window TVF ステート互換性原则 - Realtime Compute for Apache Flink

デプロイメントの SQL ステートメントでウィンドウテーブル値関数 (TVF) を変更し、状態データを使用してデプロイメントを再起動する場合、互換性の結果は変更内容によって異なります。ウィンドウ TVF クエリの 3 つの部分は、保存された状態に永続的にバインドされます。それは、ウィンドウ定義 (タイプ、サイズ、時間属性)、グループキー、および DISTINCT 集計メトリックのセットです。これらのいずれかを変更すると、常にすべての状態が破棄されます。非 DISTINCT 集計メトリックはより柔軟で、部分的または完全な状態の再利用を伴って、追加、削除、または順序変更が可能です。

クイックリファレンス

変更	結果
非 DISTINCT 集計メトリックの追加	一部
非 DISTINCT 集計メトリックの削除	対応
非 DISTINCT 集計メトリックの同時追加と削除	部分的
非 DISTINCT 集計メトリックの変更	部分的
非 DISTINCT 集計メトリックの位置の変更	互換
非 DISTINCT 集計メトリック内のフィールドの計算ロジックの変更	部分的
ウィンドウ属性フィールド (window_start、window_end、window_time) の追加または削除	互換性あり
変更前後で集計メトリックが存在しない	互換性あり
ウィンドウ関連のグループキーのシーケンスの変更 (他のキーは変更なし)	互換性あり
ウィンドウタイプ、ウィンドウサイズ、または時間属性の変更	非互換
グループキーの追加、削除、または変更	非互換
非ウィンドウグループキーのシーケンスの変更	非互換
DISTINCT 集計メトリックの追加、削除、または変更	非互換
すべての集計メトリックの削除	非互換
元の SQL に集計メトリックがない場合に追加する	非互換
集計メトリックが 1 つしか存在せず、その計算ロジックが変更される	非互換
集計メトリックのセットが完全に変更される (変更前後で異なる)	非互換
ウィンドウ関数と GROUP BY の間にフィールド計算句を追加または削除する	非互換
GROUP BY に window_start と window_end の両方が含まれるかどうかを変更する	非互換
GROUPING SETS、CUBE、または ROLLUP の追加または削除	非互換
変更の前後両方でフィールド計算句が存在する	不明
ウィンドウフィールドに対するフィルター句の追加、削除、または保持	不明
ユーザー定義テーブル関数 (UDTF) の追加、削除、または保持	不明
Python ユーザー定義集計関数 (UDAF) の追加、削除、または保持	不明
window_start、window_end、または window_time に対する集計の追加、削除、または保持	不明
変更の前後両方で GROUPING SETS、CUBE、または ROLLUP が存在する	不明

完全な互換性を維持する変更

これらの変更により、デプロイメントは集計結果に影響を与えることなく、既存の状態データから再開できます。

非 DISTINCT 集計メトリックの削除。 削除されたメトリックの状態データは破棄され、残りのメトリックは中断することなく保存された値から継続します。

非 DISTINCT 集計メトリックの位置の変更。 SELECT リスト内の集計列の順序を変更しても、状態には影響しません。次の例では、max(c) を sum(b) の前に移動していますが、デプロイメントは完全な互換性を維持します。

-- 元のクエリ
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- 順序変更後 — 完全互換
SELECT a, max(c), sum(b)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

ウィンドウ属性フィールドの追加または削除。 SELECT リストに window_start、window_end、または window_time を追加または削除しても、状態には影響しません。

-- 元のクエリ
SELECT a, sum(b), max(c), window_start
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- window_end の追加 — 完全互換
SELECT a, sum(b), max(c), window_start, window_end
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- window_start の削除 — 完全互換
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

変更前後で集計メトリックが存在しない。 元の SQL と更新後の SQL の両方に集計メトリックが含まれていない場合、状態は完全に互換性があります。

ウィンドウ関連のグループキーのシーケンスの変更。 他のすべてのグループキーの相対的な順序が同じである限り、GROUP BY 句内で window_start または window_end キーの順序を変更しても安全です。

-- 元のクエリ
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, c, window_end, window_start;

-- window_end と window_start の入れ替え — 完全互換
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_end, c, window_start;

部分的な互換性となる変更

これらの変更により、デプロイメントは変更されていないメトリックの状態データを再利用できます。新しく追加されたメトリックはデプロイメントの再起動時にゼロから集計を開始し、削除されたメトリックの状態は破棄されます。

説明

変更しないメトリックについては、状態を再利用した後の計算結果は、完全な履歴リプレイによって生成される結果と同一になります。

非 DISTINCT 集計メトリックの追加。 新しいメトリックは、デプロイメントの開始時にゼロからカウントを開始します。既存のメトリックには影響しません。
非 DISTINCT 集計メトリックの同時追加と削除。 追加されたメトリックはゼロから開始し、削除されたメトリックの状態は破棄され、変更されていないメトリックは通常どおり継続します。
非 DISTINCT 集計メトリックの変更 (集計関数の変更やその入力フィールドの計算ロジックの変更を含む)。元のメトリックは削除されたものとして扱われ、その状態は破棄されます。新しいメトリックはゼロから開始します。

次の例は、これらのケース (メトリックの追加、削除、変更) を示しています。追加を伴わない純粋な削除については、非 DISTINCT 集計メトリックの削除に関する互換性のある変更をご参照ください。

-- 元のクエリ
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- count(c) の追加: 部分的互換。
-- sum(b) と max(c) は保存された状態から継続。count(c) は 0 から開始。
SELECT a, sum(b), max(c), count(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- count(c) の追加と sum(b) の同時削除: 部分的互換。
-- max(c) は保存された状態から継続。sum(b) の状態は破棄。count(c) は 0 から開始。
SELECT a, max(c), count(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- max(c) を min(c) に変更: 部分的互換。
-- sum(b) は保存された状態から継続。
-- max(c) の状態は破棄。min(c) は 0 から開始。
SELECT a, sum(b), min(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

入力フィールドの計算ロジックの変更も変更とみなされます。次の例では、一時ビューを使用して max(c) を max(substring(c, 1, 5)) に変更しています。元の max(c) の状態は破棄され、新しいクエリの max(c) (ここで c は元の文字列の最初の 5 文字) はゼロから開始します。

-- 元のクエリ
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- 一時ビューを介した入力フィールドロジックの変更: 部分的互換。
-- sum(b) は保存された状態から継続。max(c) の状態 (元の) は破棄。
-- 新しいビューの max(c) は 0 から開始。
CREATE TEMPORARY VIEW MyView AS SELECT a, b, substring(c, 1, 5) AS c, ts FROM MyTable;
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyView, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

完全な非互換性となる変更

これらの変更では、デプロイメントはすべての状態データを破棄する必要があります。デプロイメントは、以前の状態が存在しないかのようにゼロから再起動します。

ウィンドウ属性の変更。 ウィンドウタイプ (例: TUMBLE から HOP へ)、ウィンドウサイズ、または時間属性を変更すると、完全な非互換性が発生します。

-- 元のクエリ
SELECT a, sum(b), max(c), window_start
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- ウィンドウタイプを HOP に変更: 非互換
SELECT a, sum(b), max(c),
  hop_start(ts, INTERVAL '1' MINUTE, INTERVAL '2' MINUTE) AS window_start
FROM MyTable
GROUP BY a, hop(ts, INTERVAL '1' MINUTE, INTERVAL '2' MINUTE);

-- ウィンドウサイズを 1 分から 2 分に変更: 非互換
SELECT a, sum(b), max(c),
  tumble_start(ts, INTERVAL '2' MINUTE) AS window_start
FROM MyTable
GROUP BY a, tumble(ts, INTERVAL '2' MINUTE);

-- 時間属性を ts から proctime に変更: 非互換
SELECT a, sum(b), max(c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, tumble(proctime, INTERVAL '1' MINUTE);

グループキーの追加、削除、または変更 (グループキーフィールドの計算ロジックの変更を含む)。

-- 元のクエリ
SELECT a, sum(b), max(c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);

-- グループキー d の追加: 非互換
SELECT a, sum(b), max(c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, d, tumble(ts, INTERVAL '1' MINUTE);

非ウィンドウグループキーのシーケンスの変更。 ウィンドウ関数に関連しないグループキーの順序を変更すると、完全な非互換性が発生します。

-- 元のクエリ
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, c, window_end, window_start;

-- a と c の入れ替え: 非互換
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY c, a, window_end, window_start;

DISTINCT 集計メトリックの追加、削除、または変更。 DISTINCT 集計メトリック (COUNT(DISTINCT ...) などの DISTINCT 集計関数を使用して計算) は、追加、削除、または変更されると、常に完全な非互換性を引き起こします。

-- 元のクエリ
SELECT a, sum(b), count(DISTINCT b), max(c), count(DISTINCT c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);

-- count(DISTINCT b) の削除: 非互換
SELECT a, sum(b), max(c), count(DISTINCT c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);

すべての集計メトリックの削除。 すべての状態が破棄され、状態データは再利用されません。

-- 元のクエリ
SELECT a, sum(b), count(DISTINCT b), max(c), count(DISTINCT c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);

-- すべての集計メトリックの削除: 非互換
SELECT a, tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);

元の SQL に集計メトリックがない場合に追加する。 集計メトリックがゼロの状態から 1 つ以上になると、完全な非互換性が発生します。

-- 元のクエリ (集計メトリックなし)
SELECT a, b
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, b, window_end, window_start;

-- count(a) の追加: 非互換
SELECT a, b, count(a)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, b, window_end, window_start;

集計メトリックが 1 つしか存在せず、その計算ロジックが変更される。

-- 元のクエリ (単一メトリック)
INSERT INTO MySink SELECT a, sum(b)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' SECOND))
GROUP BY a, window_start, window_end;

-- 一時ビューを介した入力計算の変更: 非互換
CREATE TEMPORARY VIEW MyView AS SELECT a, b + 1 AS b, ts FROM MyTable;
INSERT INTO MySink SELECT a, sum(b)
FROM TABLE(TUMBLE(TABLE MyView, DESCRIPTOR(ts), INTERVAL '1' SECOND))
GROUP BY a, window_start, window_end;

集計メトリックのセットが変更前後で完全に異なる。

-- 元のクエリ
INSERT INTO MySink SELECT a, sum(b)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' SECOND))
GROUP BY a, c, window_start, window_end;

-- sum(b) を min(b) に置換: 非互換
INSERT INTO MySink SELECT a, min(b)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' SECOND))
GROUP BY a, c, window_start, window_end;

ウィンドウ関数と GROUP BY の間にフィールド計算句を追加または削除する。 フィールド計算句は、TVF と GROUP BY ステップの間で window_start、window_end、または window_time を変換します。

-- 元のクエリ
SELECT a, sum(b), max(c), window_start, window_end
FROM (SELECT a, b, c, window_start, window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY a, window_start, window_end;

-- フィールド計算句の追加: 非互換
SELECT a, sum(b), max(c), window_start, window_end
FROM (SELECT a, b, c,
             window_start + (INTERVAL '1' SECOND) AS window_start,
             window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY a, window_start, window_end;

GROUP BY に window_start と window_end の両方が含まれるかどうかを変更する。 GROUP BY 句が両方のフィールドを含む状態から一方を除外する状態に変わる (またはその逆) 場合、デプロイメントは非互換になります。

-- 元のクエリ (GROUP BY には window_start のみ含まれる)
SELECT a, sum(b), max(c), window_start
FROM (SELECT a, b, c, window_start
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY a, window_start;

-- GROUP BY に window_end を追加: 非互換
SELECT a, sum(b), max(c), window_start, window_end
FROM (SELECT a, b, c, window_start, window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY a, window_start, window_end;

GROUPING SETS、CUBE、または ROLLUP の追加または削除。 これらの句は、window_start と window_end によってデータを個別にグループ化するため、内部状態のレイアウトが変更されます。

-- 元のクエリ
SELECT a, sum(b), max(c), window_start, window_end
FROM (SELECT a, b, c, window_start, window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY a, window_start, window_end;

-- GROUPING SETS の追加: 非互換
SELECT a, sum(b), max(c), window_start, window_end
FROM (SELECT a, b, c, window_start, window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY GROUPING SETS ((a), (window_start), (window_end));

互換性が不明な変更

これらのパターンでは、エンジンは状態データに互換性があるかどうかを事前に判断できません。これらの変更は潜在的に非互換であるとみなし、再起動後に動作を検証してください。

変更の前後両方でフィールド計算句が存在する。 元の SQL と変更後の SQL の両方に (たとえ変更がなくても) フィールド計算句が存在する場合、互換性は不明です。

ウィンドウフィールド (window_start、window_end、window_time) に対するフィルター句。 ウィンドウフィールドに対するフィルターを追加、削除、または保持すると、いずれも互換性が不明になります。

-- 元のクエリ
SELECT a, sum(b), max(c), window_start, window_end
FROM (SELECT a, b, c, window_start, window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY a, window_start, window_end;

-- フィルター句の追加: 互換性不明
SELECT a, sum(b), max(c), window_start, window_end
FROM (SELECT a, b, c, window_start, window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
      WHERE window_start >= TIMESTAMP '2024-04-15 08:06:00.000')
GROUP BY a, window_start, window_end;

ユーザー定義テーブル関数 (UDTF)。 SQL に UDTF を追加、削除、または保持すると、いずれも互換性が不明になります。

-- 元のクエリ
SELECT a, sum(b), length(c), window_start, window_end
FROM (SELECT a, b, c, window_start, window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY a, c, window_start, window_end;

-- UDTF の追加: 互換性不明
SELECT a, sum(b), length(c), window_start, window_end, c1, c2
FROM (SELECT a, b, c, window_start, window_end, c1, c2
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)),
           LATERAL TABLE(split(c)) AS T(c1, c2))
GROUP BY a, c, window_start, window_end, c1, c2;

Python ユーザー定義集計関数 (UDAF)。 Python UDAF を追加、削除、または保持すると、いずれも互換性が不明になります。

-- 元のクエリ
SELECT a, sum(b), max(c), window_start
FROM (SELECT a, b, c, window_start
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY a, window_start;

-- Python UDAF の追加: 互換性不明
SELECT a, sum(b), c, window_start
FROM (SELECT a, b, weighted_avg(c) AS c, window_start
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
      GROUP BY a, b, window_start)
GROUP BY a, c, window_start;

window_start、window_end、または window_time に対する集計。 ウィンドウフィールドに集計関数を適用する (追加、削除、または保持する) と、互換性が不明になります。

-- 元のクエリ
SELECT a, sum(b), max(c), window_start, window_end
FROM (SELECT a, b, c, window_start, window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY a, window_start, window_end;

-- window_start に対する集計の追加: 互換性不明
SELECT a, sum(b), max(c), MAX(window_start) AS ag, window_end
FROM (SELECT a, b, c, window_start, window_end
      FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
GROUP BY (a, window_start, window_end);

変更の前後両方で GROUPING SETS、CUBE、または ROLLUP が存在する。 変更の前後両方で GROUP BY 句が GROUPING SETS、CUBE、または ROLLUP を使用している場合、互換性は不明です。