PLV8でPostgreSQLのテーブル分割をしてツイートを保存する

目的

TwitterのSearchAPIの結果をPostgreSQLに保存します。この時保存するテーブルは1日ごとにテーブル分割を行います。ツイートを保存するコードはplv8で記述しました。

テーブル分割

PostgreSQL10から入った新しいテーブル分割ですがPostgreSQL11から機能が向上して主キーやインデックスが子テーブルにも反映されるようになりました。ただし分割するキーを主キーにする必要があります。ツイートの場合本来ステータスIDのみが主キーになるはずですが、分割のためにはツイート日時を含める必要があります。

ステータスIDは将来BIGINTの範囲を超えそうなのでNUMERICにしています(BIGINT最大992京に対して現在のステータスIDは106京)。

CREATE TABLE public.tweets (
  status_twid NUMERIC NOT NULL
  ,tw_created_at TIMESTAMPTZ NOT NULL
  ,data_json JSONB NOT NULL
  ,PRIMARY KEY(status_twid, tw_created_at)
) PARTITION BY RANGE (tw_created_at);

ツイート日時

テーブル名をtweets_20181201のように日付を後ろにつけたいです。日付とフォーマットを与えて文字列を取得する関数を準備しました。

CREATE OR REPLACE FUNCTION tw_get_data_text_to_ymd (
  p_data TIMESTAMPTZ
  ,p_format TEXT
) RETURNS TEXT AS $FUNCTION$
  return p_format.replace(/yyyy/g, p_data.getUTCFullYear()).
    replace(/MM/g, ('0' + (p_data.getUTCMonth() + 1)).slice(-2)).
    replace(/dd/g, ('0' + p_data.getUTCDate()).slice(-2)).
    replace(/HH/g, ('0' + p_data.getUTCHours()).slice(-2)).
    replace(/mm/g, ('0' + p_data.getUTCMinutes()).slice(-2)).
    replace(/ss/g, ('0' + p_data.getUTCSeconds()).slice(-2)).
    replace(/SSS/g, ('00' + p_data.getUTCMilliseconds()).slice(-3))
$FUNCTION$
LANGUAGE plv8 IMMUTABLE STRICT;

ツイート登録

パラメーターp_jsonはTwitterのSearchAPIのレスポンスそのものが入ります。

子テーブルはあらかじめ用意せずに、とりあえず登録してみてテーブルが無かったらその場で子テーブルを作成するようにしています。エラーコード「23514」はテーブルが存在しなかった時に出るエラーです。このエラーが発生した時にはテーブルを作り再度登録しなおしています。prepare statementのおかげ同じSQLを書かなくて済んでいます。

CREATE OR REPLACE FUNCTION tw_set_add_tweets (
  p_json JSONB DEFAULT NULL
) RETURNS VOID AS $FUNCTION$
  const statuses = p_json.statuses
  const func = plv8.find_function("tw_get_data_text_to_ymd")
  const insert = plv8.prepare(`
    INSERT INTO public.tweets (
      status_twid
      ,tw_created_at
      ,data_json
    ) VALUES (
      $1
      ,$2
      ,$3
    ) ON CONFLICT (status_twid, tw_created_at)
    DO NOTHING
  `, ['numeric', 'timestamptz', 'jsonb'])
  for (let key in statuses) {
    if (statuses.hasOwnProperty(key)) {
      try {
        insert.execute([statuses[key].id_str, statuses[key].created_at, statuses[key]])
      } catch (error) {
        if (error.sqlerrcode != 23514) {
          throw error
        }
        const tw_created_at = new Date(statuses[key].created_at)
        const ymd = func(tw_created_at, 'yyyyMMdd')
        const from = func(tw_created_at, 'yyyy-MM-dd')
        tw_created_at.setDate(tw_created_at.getDate() + 1)
        const to = func(tw_created_at, 'yyyy-MM-dd')
        plv8.execute(`
          CREATE TABLE IF NOT EXISTS public.tweets_${ymd}
          PARTITION OF public.tweets FOR VALUES 
          FROM ('${from}')
          TO ('${to}')
        `)
        insert.execute([statuses[key].id_str, statuses[key].created_at, statuses[key]])
      }
    }
  }
  insert.free()
$FUNCTION$
LANGUAGE plv8;

まとめ

PLV8を使うことでJSONの取り回しが簡単にできました。レスポンスのほとんどがJSONのTwitterAPIとは相性がよさそうです。

PostgreSQLのテーブル分割で分割キーを主キーに含める必要があるのは、困りものです。今回の例のように主キーでは無いカラムで分割したい要件は色々あると思います。主キーですら論理設計がそのまま使えないのは不便なので将来改善してほしいところです。