詳解 ROWS_PRODUCED

この記事は Snowflake Advent Calendar 2021 の 13 日目です。

誰

Snowflake で SQL や Query Processing などのデータベースコア機能のサポートエンジニアをやっている者です。

ROWS_PRODUCED とは

Snowflake の QUERY_HISTORY ビュー/テーブル関数には ROWS_PRODUCED というカラムがあり、そのクエリが生成した行数を表示しています。

https://docs.snowflake.com/ja/sql-reference/account-usage/query_history.html#columns

ROWS_PRODUCED NUMBER このステートメントによって生成された行の数。

このカラムの値は、Web UI のクエリ履歴やクエリ詳細画面に表示されている Rows (日本語 UI では「行」) の値と同じになります。

クエリ履歴の Rows 列

クエリ詳細画面の Rows 列

そんな ROWS_PRODUCED ですが、おそらく Snowflake を使っていると「DML の更新行数でも結果セットの行数でもない数字が Rows に出てておかしいな」と思うことが一度はきっとあると思います。

この記事では ROWS_PRODUCED が何を表す値なのかを説明していきます。

ROWS_PRODUCED の正体

結論から言うと ROWS_PRODUCED は、

• SELECT ... 結果セットの行数
• DML (INSERT, UPDATE, DELETE ...) ... 物理的にマイクロパーティションに書き込まれた行数

になります。

ここで問題になるのが、ここで言う「物理的にマイクロパーティションに書き込まれた行数」は「DML によって論理的に更新/削除された行数」と一致しません。

例えば、下記のようなケースでは ROWS_PRODUCED と更新/削除行数が一致していません。

create or replace table t1 (c1 int) as
select seq4() from table(generator(rowcount => 100));

update t1 set t1.c1 = 10000 where t1.c1 = 0;
/*
number of rows updated	number of multi-joined rows updated
1	0
*/

delete from t1 where t1.c1 = 10000;
/*
number of rows deleted
1
*/

select query_text, rows_produced
from table(information_schema.query_history());
/*
delete from t1 where t1.c1 = 10000;		99
update t1 set t1.c1 = 10000 where t1.c1 = 0;		100
*/

上記のクエリは下記のような処理を実行しています。

• SEQ4 関数と GENERATOR テーブル関数で 0 〜 99 までの異なる数字が格納された 100 行のテーブルを CREATE TABLE する
• UPDATE で c1 カラムが 0 の行を 10000 に更新する
• DELETE でその更新された 10000 の行を削除する
• 上記 2 クエリについて QUERY_HISTORY テーブル関数で ROWS_PRODUCED を確認する

当然 UPDATE で論理的に更新された行数は 1 行、DELETE で論理的に削除された行数も 1 行になります…が、ROWS_PRODUCED はそれぞれなぜか 100 と 99 を示しています。

なぜこのような結果になるのでしょうか。

物理的にマイクロパーティションに書き込まれた行数

「物理的にマイクロパーティションに書き込まれた行数」という概念を理解するにあたり、マイクロパーティションが何者で、どういう特性を持つのかから考えていきます。

マイクロパーティションは Snowflake のテーブルの実体である非圧縮で 50 〜 500 MB 程度のファイルで、カラムごとに作成されて複数の行の値を格納しています (カラムナーストレージ)。このマイクロパーティションが多数集合して Snowflake のテーブルを構成しています。

ここで、Snowflake のストレージは各クラウドサービスのオブジェクトストレージ (Amazon S3, Azure Blob Storage, Google Cloud Storage) 上に存在します。すなわち、マイクロパーティションはオブジェクトストレージのオブジェクトとして格納されています。

この「マイクロパーティションはオブジェクトストレージのオブジェクトである」という事実により、マイクロパーティションは「イミュータブルである」という重要な特性を得ることになります。下記は Google Cloud Storage のドキュメントですが、Amazon S3 も Azure Blob Storage も同様の特性を持っています。

https://cloud.google.com/storage/docs/key-terms#immutability

オブジェクトの不変性
オブジェクトは不変です。具体的には、アップロードされたオブジェクトは、その保存期間が終わるまで変わることはありません。オブジェクトの保存期間とは、オブジェクトが正常に作成され（たとえば、アップロードなど）、正常に削除されるまでの時間です。つまり、後からオブジェクトに付加や切り捨てなどのオペレーションで部分的な変更を加えることはできません。ただし、Google Cloud Storage に保存されているオブジェクトを置換することは可能で、これはアトミックに行われます。

すなわち、マイクロパーティションはイミュータブルなオブジェクトであるため、部分的な書き換えができず、UPDATE や DELETE でマイクロパーティションの一部が更新/削除される場合、そのマイクロパーティションごと更新/削除後のデータで再作成される形となります。

したがって、この例における UPDATE は、

1. 100 行のマイクロパーティション A を S3 から読み込み
2. c1 = 0 の行を 10000 に更新した 100 行のマイクロパーティション B をローカルディスクに作成
3. マイクロパーティション B を S3 にアップロードすると同時に、マイクロパーティション A を削除済みとしてマーク

という流れで処理されます。

ここで、マイクロパーティション B の行数は、マイクロパーティション A のデータを UPDATE で論理的に更新した後の行数になり、当然 UPDATE では行数は変化しないため、100 行になります。

上記を前提に、「物理的にマイクロパーティションに書き込まれた行数」を考えると、これは S3 にアップロードされたマイクロパーティションの行数となるので、ここではマイクロパーティション B の行数、すなわち 100 行となり、ROWS_PRODUCED も 100 となります。

同様に DELETE の場合は、

1. 100 行のマイクロパーティション B を S3 から読み込み
2. c1 = 10000 の行を削除した 99 行のマイクロパーティション C をローカルに作成
3. マイクロパーティション C を S3 にアップロードすると同時に、マイクロパーティション B を削除済みとしてマーク

という流れで処理されるため、ROWS_PRODUCED はマイクロパーティション C の行数である 99 となります。

特殊な例: INSERT

さて、今まで触れてこなかった INSERT についても考えてみましょう。

INSERT は挿入されたデータで新しいマイクロパーティションを作成するだけのため、ROWS_PRODUCED も挿入された行数と一致するはずです。

しかし実は特定の条件下ではそうなりません。例えば、下記のようなケースでは ROWS_PRODUCED と挿入行数が一致していません。

create or replace table t1 (c1 int) as
select seq4() from table(generator(rowcount => 100));

insert into t1 values (10000);
/*
number of rows inserted
1
*/

select query_text, rows_produced
from table(information_schema.query_history());
/*
insert into t1 values (10000);
101
*/

INSERT は 1 行しか挿入していないのに、ROWS_PRODUCED は挿入後の行数である 101 を示しています。

なるほど、INSERT だと ROWS_PRODUCED は挿入後の行数を表示するのか…と思いきや、例えば下記のケース (1,000,000 行のテーブルに 1,000,000 行の INSERT) では挿入した行数を表示しています。

create or replace table t1 (c1 int) as
select seq4() from table(generator(rowcount => 1000000));

insert into t1
select seq4() from table(generator(rowcount => 1000000));
/*
number of rows inserted
1000000
*/

select query_text, rows_produced
from table(information_schema.query_history());
/*
insert into t1 select seq4() from table(generator(rowcount => 1000000));
1000000
*/

なぜこんなことが起きるかというと、実は INSERT 実行時に内部的に適用されている最適化の影響があります。

前述の通り、基本的には INSERT や COPY を実行した場合「挿入されたデータで新しいマイクロパーティションを作成しアップロードする」という処理になります。

しかし、例えば 1 行の INSERT が 1,000,000 回実行されたときのことを考えると、この処理には少々問題があることがわかります。

上記のようなワークロードを普通にそのまま処理すると、1,000,000 個のマイクロパーティションが新規に作成されることになりますが、前述の通りマイクロパーティションは通常非圧縮で 50 〜 500 MB 程度のサイズであるため、1,000,000 行程度であれば (行長や圧縮効率にもよりますが) 1 つのマイクロパーティションで格納できるデータになります。

つまり、普通であれば 1 つのマイクロパーティションで収まるデータが 1,000,000 個のマイクロパーティションに格納されてしまうことになるわけです。

このような過剰なマイクロパーティション数は、テーブルスキャン時のスキャン対象マイクロパーティションをリストアップする処理や、マイクロパーティションのメタデータをダウンロードする処理などにおいて、無視できないレベルのオーバーヘッドになりパフォーマンス劣化を起こすため、Snowflake としてはこういった状況になることは避けたいと考えています。

そこで Snowflake は、対象テーブルにサイズの小さいマイクロパーティションが存在するときに、そのマイクロパーティションから読み込んだデータを挿入行とマージして、マージした結果を新しいマイクロパーティションとして作成しアップロードする、という形で INSERT を処理します。

したがって、このセクションの最初の例における INSERT は、

1. 100 行のマイクロパーティション A を S3 から読み込み
2. 挿入行 1 行 (c1 = 10000) と読み込んだ 100 行をマージし、101 行のマイクロパーティション B をローカルディスクに作成
3. マイクロパーティション B を S3 にアップロードすると同時に、マイクロパーティション A を削除済みとしてマーク

という流れで処理されます。

つまり、サイズの小さいマイクロパーティションを含むテーブルへの INSERT については、更新処理がマージ処理になることを除けば、ほぼ UPDATE と同じ手順で処理されることになり、ROWS_PRODUCED も UPDATE と同様にアップロードされたマイクロパーティション B の行数、101 になります。

その一方で、このセクションの 2 つ目の例については、既存のマイクロパーティションがサイズの小さいマイクロパーティションであると判断されなかったため、

1. 1,000,000 行の挿入行からマイクロパーティション C をローカルディスクに作成
2. マイクロパーティション C を S3 にアップロード

というシンプルな手順で処理され、ROWS_PRODUCED も挿入行だけを格納したマイクロパーティション C の行数である 1,000,000 になります。

論理的に挿入/更新/削除された行数が知りたいときはどうしたら

とはいえ、Snowflake を使っている側の立場に立つと、物理的にマイクロパーティションに書き込まれた行数は正直そんなに重要ではなく、だいたいのケースで実際に論理的に挿入/更新/削除された行数のほうが重要になることが多いと思います。

Snowflake ももちろんそこは理解しており、昨年 2020 年の初め頃に ROWS_INSERTED, ROWS_UPDATED, ROWS_DELETED という極めて直感的なメトリクスを追加しました。

https://docs.snowflake.com/ja/sql-reference/account-usage/query_history.html#columns

ROWS_INSERTED NUMBER クエリによって挿入された行の数。
ROWS_UPDATED NUMBER クエリによって更新された行の数。
ROWS_DELETED NUMBER クエリによって削除された行の数。

しかし、これにも少しだけ注意点があります。これらのメトリクスは ACCOUNT_USAGE の QUERY_HISTORY ビューにのみ存在し、INFORMATION_SCHEMA の QUERY_HISTORY テーブル関数には存在しないのです。

ACCOUNT_USAGE ビューは INFORMATION_SCHEMA テーブル関数と異なりタイムラグがあるため、もちろん実行中のクエリについては(無いとは思いますが)確認できませんし、直近数分の実行でも表示されない可能性があります。

ACCOUNT_USAGE ビューと INFORMATION_SCHEMA テーブル関数の違いは、ドキュメントにわかりやすくまとめられているので、こちらを見ていただくといいと思います。

https://docs.snowflake.com/ja/sql-reference/account-usage.html#differences-between-account-usage-and-information-schema

ちなみに、どのカラムが ACCOUNT_USAGE の QUERY_HISTORY ビューだけにあって、どのカラムが INFORMATION_SCHEMA の QUERY_HISTORY テーブル関数だけにあるかは、こんな感じのクエリで確認することができます。

with
account_usage_cols as (
    select k.value::varchar name
    from table(flatten(
        select object_keys(object_construct_keep_null(*))
        from snowflake.account_usage.query_history limit 1
    )) k
),
information_schema_cols as (
    select k.value::varchar name
    from table(flatten(
        select object_keys(object_construct_keep_null(*))
        from table(information_schema.query_history(result_limit => 1))
    )) k
)
select
    coalesce(a.name, i.name) column_name,
    iff(
        a.name is not null,
        iff(
            i.name is not null,
            'Both',
            'View only'
        ),
        'Function only'
    ) existence
from account_usage_cols a
full join information_schema_cols i on a.name = i.name
order by existence, column_name
;

関係ない話ですが、私の知る限り OBJECT_CONSTRUCT 関数による集約が Snowflake において SQL ネイティブで結果セットのカラム名を文字列化する唯一の方法なので、カラム名であれこれしたいときは便利です。前回の記事でも使ってます。

結びに

この記事によって、クエリ履歴の Rows の値がわけわからなくて夜も眠れない Snowflake ユーザが安眠できるようになることを祈っております。