Amazon Redshift SYSビューによる実行クエリの取得

石川覚
2024.10.15
AWS事業本部コンサルティング部の石川です。Amzon Redshift Serverless は、従来のSTLテーブルやSVLビューに代わりSYSビューが導入されました。SYSビュー導入の背景、Serverless と Provisioned の両方で利用できるクエリログを取得する方法をご紹介します。
 SYSビューへの移行の理由Amazon Redshiftが従来のSTLテーブルやSVLビューに代わってSYSビューを導入した背景は以下のとおりです。
https://aws.amazon.com/jp/blogs/news/simplify-amazon-redshift-monitoring-using-the-new-unified-sys-views/
https://docs.aws.amazon.com/ja_jp/redshift/latest/dg/cm_chap_system-tables.html
 モニタリングの簡素化SYSビューは、Redshift Provisioned と Redshift Serverless の両方で、クエリとワークロードのリソース使用状況を監視するために使用できるようにするため、再設計されました。
 整合性の向上従来のシステムテーブルは整合性モデルが通常のテーブルと異なっていましたが、SYSビューではこの問題に対処しています。
 データの可視性の改善SYSビューでは、ユーザーの権限に基づいてデータの可視性がより明確に制御されています。
 特長 機能的な整理SYSビューは、クエリの状態、パフォーマンスメトリクス、クエリタイプなどの機能的な観点から分類されています。
 新しいパフォーマンスメトリクスの導入planning_time、lock_wait_time、remote_read_io、local_read_ioなどの新しいメトリクスが追加され、パフォーマンスのトラブルシューティングに役立ちます。
 ユーザビリティの向上Redshiftオプティマイザによって書き換えられたクエリではなく、ユーザーが送信したクエリをログに記録することで、モニタリングビューの使いやすさが向上しました。
 外部クエリの詳細な洞察外部オブジェクトのリスティングや分割の時間、外部ファイルの場所、外部テーブル名など、外部クエリに関する詳細な情報が提供されます。
 ユーザークエリレベルのロギングユーザーが送信したクエリ（親クエリ）と、Redshiftクエリエンジンによって書き換えられたクエリ（子クエリ）を区別して記録できるようになりました。
これらの改善により、Amazon Redshiftのモニタリングとトラブルシューティングがより効率的かつ効果的になり、ユーザーエクスペリエンスが向上しました。
 SQLによるクエリの取得過去24時間のクエリを取得するSQLです。SYS_QUERY_HISTORYなら簡単にクエリを取得できるのですが、クエリが4000バイトに切り詰められてしまいます。
そこで以下のクエリでは、SYS_QUERY_HISTORYをジョインして、64000バイトまで出力できるように拡張しました。逆の言い方をすると、AND sequence < 16のように64000バイトまでに限定しています。また、Redshiftの内部ユーザーのクエリは、user_id >= 100のように不要ユーザーのクエリは排除しています。
WITH 
query_text AS (
    SELECT user_id, query_id, LISTAGG(text) WITHIN GROUP (ORDER BY sequence) AS query_text
    FROM 
        sys_query_text
    WHERE
        user_id >= 100
        AND sequence < 16 
    GROUP BY user_id, query_id
),
query_all AS (
    SELECT q.query_id, q.start_time, q.end_time, q.user_id, 
        REGEXP_REPLACE(u.query_text, 'declare "SQL_CUR[0-9]+" cursor with hold for ', '', 1, 'i') AS query_text
    FROM 
        SYS_QUERY_HISTORY q
    INNER JOIN query_text u ON q.query_id = u.query_id
    WHERE q.start_time >= CURRENT_TIMESTAMP - INTERVAL '1 hours'
        AND u.query_text NOT ILIKE '% pg_%'
        AND u.query_text NOT ILIKE '% svv_%'
        AND u.query_text NOT ILIKE '% sys_%'
        AND u.query_text NOT ILIKE '%SET %'
        AND u.query_text NOT ILIKE '%COPY %'
        AND u.query_text NOT ILIKE '%UNLOAD %'
        AND u.query_text NOT ILIKE '%CREATE %'
        AND u.query_text NOT ILIKE '%INSERT %'
        AND u.query_text NOT ILIKE '%UPDATE %'
        AND u.query_text NOT ILIKE '%DELETE %'
        AND u.query_text NOT ILIKE '%DROP %'
        AND u.query_text NOT ILIKE '%FETCH %'
        AND u.query_text NOT ILIKE '%CLOSE %'
        AND u.query_text NOT ILIKE '%TRUNCATE %'
        AND u.query_text NOT ILIKE '%SHOW %'
        AND u.query_text NOT ILIKE '%GRANT %'
        AND u.query_text NOT ILIKE '%ALTER %'
        AND u.query_text NOT ILIKE '%BEGIN%'
        AND u.query_text NOT ILIKE '%COMMIT%'
        AND u.query_text NOT ILIKE '%ROLLBACK%'
        AND u.query_text NOT ILIKE 'select 1'
        AND u.query_text NOT ILIKE 'select current_user'
    ORDER BY q.start_time ASC
)
SELECT * FROM query_all;

※ 15〜34行目は、お好みでフィルタしてください。私はユーザーが実行していないクエリはいらないので、余計なクエリは排除しています。
 クエリごとにSQLファイル出力するクエリごとにSQLファイル分割して出力するプログラムが以下になります。Redshift Provisioned と Redshift Serverless の両方で、動くようにDataAPIは使わず、psycopg2で接続してクエリを実行しています。クエリの取得する期間（hours_to_fetch）や認証情報はSecretManagerから取得するなどお好みで修正してください。
import psycopg2
import os

# Redshift接続情報
redshift_config = {
    'host': '<endpoint>',
    'port': 5439,
    'dbname': '<database_name>',
    'user': '<user>',
    'password': '<password>'
}

# 出力ディレクトリ
output_dir = 'redshift_queries'
os.makedirs(output_dir, exist_ok=True)

# 過去n時間のクエリを取得
hours_to_fetch = 24

def fetch_queries(cursor):
    query = f"""
    WITH 
    query_text AS (
        SELECT user_id, query_id, LISTAGG(text) WITHIN GROUP (ORDER BY sequence) AS query_text
        FROM 
            sys_query_text
        WHERE
            user_id >= 100
            AND sequence < 16 
        GROUP BY user_id, query_id
    ),
    query_all AS (
        SELECT q.query_id, q.start_time, q.end_time, q.user_id, 
            REGEXP_REPLACE(u.query_text, 'declare "SQL_CUR[0-9]+" cursor with hold for ', '', 1, 'i') AS query_text
        FROM 
            SYS_QUERY_HISTORY q
        INNER JOIN query_text u ON q.query_id = u.query_id
        WHERE q.start_time >= CURRENT_TIMESTAMP - INTERVAL '{hours_to_fetch} hours'
            AND u.query_text NOT ILIKE '% pg_%'
            AND u.query_text NOT ILIKE '% svv_%'
            AND u.query_text NOT ILIKE '% sys_%'
            AND u.query_text NOT ILIKE '%SET %'
            AND u.query_text NOT ILIKE '%COPY %'
            AND u.query_text NOT ILIKE '%UNLOAD %'
            AND u.query_text NOT ILIKE '%CREATE %'
            AND u.query_text NOT ILIKE '%INSERT %'
            AND u.query_text NOT ILIKE '%UPDATE %'
            AND u.query_text NOT ILIKE '%DELETE %'
            AND u.query_text NOT ILIKE '%DROP %'
            AND u.query_text NOT ILIKE '%FETCH %'
            AND u.query_text NOT ILIKE '%CLOSE %'
            AND u.query_text NOT ILIKE '%TRUNCATE %'
            AND u.query_text NOT ILIKE '%SHOW %'
            AND u.query_text NOT ILIKE '%GRANT %'
            AND u.query_text NOT ILIKE '%ALTER %'
            AND u.query_text NOT ILIKE '%BEGIN%'
            AND u.query_text NOT ILIKE '%COMMIT%'
            AND u.query_text NOT ILIKE '%ROLLBACK%'
            AND u.query_text NOT ILIKE 'select 1'
            AND u.query_text NOT ILIKE 'select current_user'
        ORDER BY q.start_time ASC
    )
    SELECT * FROM query_all;
    """

    cursor.execute(query)
    return cursor.fetchall()

def write_query_to_file(query_data):
    query_id, start_time, user_id, username, query_text = query_data
    if not bool(re.search(r'\$\d+', query_text)): # Skip Parameterized Query
        filename = f"{output_dir}/{start_time.strftime('%Y%m%d_%H%M%S')}_query_{query_id}.sql"
        with open(filename, 'w') as f:
            f.write(f"-- Start Time: {start_time}\n")
            f.write(f"-- Query ID: {query_id}\n")
            f.write(query_text.replace('\\n', '\n').replace('\\r', '\r'))

def main():
    conn = psycopg2.connect(**redshift_config)
    try:
        with conn.cursor() as cursor:
            queries = fetch_queries(cursor)
            for query in queries:
                write_query_to_file(query)
        print(f"Successfully exported {len(queries)} queries to {output_dir}")
    finally:
        conn.close()

if __name__ == "__main__":
    main()
 補足: パッケージのインストール上記のプログラムを動かすには、psycopg2の導入が必要です。そのためには、必要なパッケージとpsycopg2のインストールが必要です。
$ sudo yum install  libpq-devel gcc python3-devel postgresql-devel　
Last metadata expiration check: 0:00:44 ago on Tue 15 Oct 2024 07:36:30 AM UTC.
Dependencies resolved.
================================================================================
 Package                   Arch   Version                     Repository   Size
================================================================================
Installing:
 gcc                       x86_64 11.4.1-2.amzn2023.0.2       amazonlinux  32 M
 libpq-devel               x86_64 16.4-1.amzn2023.0.1         amazonlinux 105 k
 python3-devel             x86_64 3.9.16-1.amzn2023.0.9       amazonlinux 205 k
Installing dependencies:
 amazon-rpm-config         noarch 228-3.amzn2023.0.2          amazonlinux  74 k
 :
 :
  python-srpm-macros-3.9-41.amzn2023.0.5.noarch                                 
  python3-devel-3.9.16-1.amzn2023.0.9.x86_64                                    
  rust-srpm-macros-21-42.amzn2023.0.2.noarch                                    

Complete!
$ sudo pip install psycopg2
Collecting psycopg2
  Downloading psycopg2-2.9.9.tar.gz (384 kB)
     |████████████████████████████████| 384 kB 5.3 MB/s            
  Preparing metadata (setup.py) ... done
Building wheels for collected packages: psycopg2
  Building wheel for psycopg2 (setup.py) ... done
：
：
/root/.cache/pip/wheels/3a/06/25/adb124afd8c8346e45c455f6586f7289cde2b4e339dfbcd9e9
Successfully built psycopg2
Installing collected packages: psycopg2
Successfully installed psycopg2-2.9.9
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv
 課題 クエリの最大サイズが64000バイトVARCHARの最大長が65535 バイト (64K -1)のため、最大64000バイトです。SYS_QUERY_HISTORYの4000バイトと比較すると、16倍の64000バイトですがこのサイズを超えることも稀にあります。
もし、制限をかけないと、VARCHARの最大長を超えると以下のエラーが起こり、全てのクエリが失敗してしまいます。
mydb=# WITH 
    query_text AS (
        SELECT user_id, query_id, LISTAGG(text) WITHIN GROUP (ORDER BY sequence) AS query_text
        FROM 
            sys_query_text
        WHERE
            user_id >= 100
        GROUP BY user_id, query_id
    )
select * from query_text;
ERROR:  Result size exceeds LISTAGG limit
DETAIL:  
  -----------------------------------------------
  error:  Result size exceeds LISTAGG limit
  code:      8001
  context:   LISTAGG limit: 65535
  query:     544300989
  location:  string_ops.cpp:131
  process:   query2_913_544300989 [pid=111325]
  -----------------------------------------------
 最後に現状は従来のシステムテーブルを使用したサンプルが多いため、情報が混乱しています。SYSビューは、Redshift Provisioned と Redshift Serverless の両方でサポートされているため、今後はこちらを使うのが良いでしょう。
今回作ったクエリのリストは、負荷試験用に出力したものです。簡易な負荷試験ツールも作る予定なのでまた紹介します。