リレーショナルデータベースのテーブルでリストを扱う例（SQLite 3、Python を使用）

1. 前準備

SQLite 3の詳細は別ページ »にまとめている。

2. テストデータ生成用 Python プログラム

CSV形式のテストデータを生成し、標準出力へ出力するPythonプログラムである。指定された数のリストと各リストの要素数に基づいてデータを生成する。

生成されるデータの各行は、次の属性を持つ。

• id: 一意の識別子
• list_num: リスト番号
• item_num: リスト内の要素番号
• x: ランダムな実数値（0.0から100.0の範囲）
• y: ランダムな実数値（0.0から100.0の範囲）
• price: ランダムな整数値（1から1000の範囲）
• name: ランダムな文字列（長さ8文字）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# usage: python hoge.py 10 5

import sys
import random
import string

LEN = 8
c = 0
TOTAL_LIST_NUM = int(sys.argv[1])
LIST_LEN = int(sys.argv[2])

print("# id, list_num, item_num, x, y, price, name")

for i in range(1, TOTAL_LIST_NUM + 1):
    for j in range(1, LIST_LEN + 1):
        x = 100 * random.random()
        y = 100 * random.random()
        price = int(1000 * random.random()) + 1
        name = ''.join(random.choices(string.ascii_letters + string.digits, k=LEN))
        print(f"{c}, {i}, {j}, {x:.6f}, {y:.6f}, {price}, {name}")
        c += 1

3. テストデータベースの生成手順

上記のPythonプログラムをhoge.pyとして保存し、以下の手順でテストデータベースを生成する。この例では、10個のリストを生成し、各リストは5個の要素を持つ。

生成されるデータベースの仕様は次のとおりである。

• データベースファイル名: /tmp/1.db
• テーブル名: dat
• リスト数: 10
• 各リストの要素数: 5

CSVの先頭行は列名なので、テーブルを先に作成したうえで、.import --skip 1で先頭行を読み飛ばす。SQLite 3の.importは、対象テーブルが存在する場合、先頭行もデータとして取り込むため、この指定が必要である。

#!/bin/bash
rm -f /tmp/1.csv
python /tmp/hoge.py 10 5 > /tmp/1.csv

rm -f /tmp/1.$$.sql
cat >/tmp/1.$$.sql << SQL
create table dat (
  id       integer primary key not null,
  list_num integer,
  item_num integer,
  x        real,
  y        real,
  price    integer,
  name     text );
.mode csv
.import --skip 1 /tmp/1.csv dat
.exit
SQL

rm -f /tmp/1.db
cat /tmp/1.$$.sql | sqlite3 /tmp/1.db

データベースの内容を確認するには、次のコマンドを実行する。

sqlite3 /tmp/1.db "select * from dat;"

4. エッジリストの生成

ノードリストの集合 {L1, L2, ..., Lk} において、各集合 Li は一連のノードで構成される。これらをテーブル dat に写像し、自己結合を行う。この結果、各行が1つのエッジ（2つのノードを接続する辺）を表すテーブルが得られる。

エッジリストの生成には、次のSQL文を使用する。この文は、同一リスト内で連続する要素のペアを抽出する。WHERE句の条件 A.item_num + 1 = B.item_num and A.list_num = B.list_num により、リスト内で隣接する要素のみが結合される。

select A.id, A.list_num, A.item_num, B.id, B.list_num, B.item_num from dat A, dat B where A.item_num + 1 = B.item_num and A.list_num = B.list_num;

5. 大規模テストデータベース生成と性能評価

エッジリスト生成の性能を検証するため、大規模なテストデータベースで性能評価を行う。リスト数を 500000、1000000、2000000、5000000 と変化させ、各リストの要素数は5で固定する。

評価に使用するデータベースの仕様は次のとおりである。

• データベースファイル名: /tmp/500000.db、/tmp/1000000.db、/tmp/2000000.db、/tmp/5000000.db
• テーブル名: dat
• リスト数: 500000、1000000、2000000、5000000
• 各リストの要素数: 5

以下のPythonスクリプトは、テストデータベースの作成、データ投入、性能評価を実行する。

#!/usr/bin/env python3
import sqlite3
import subprocess
import time

def create_database(db_name):
    sql = '''
    create table dat (
        id       integer primary key not null,
        list_num integer,
        item_num integer,
        x        real,
        y        real,
        price    integer,
        name     text
    );
    '''
    conn = sqlite3.connect(f'/tmp/{db_name}.db')
    conn.execute(sql)
    conn.close()

def populate_database(db_name, list_num, list_len):
    csv_file = f'/tmp/{db_name}.csv'
    subprocess.run(
        ['python', '/tmp/hoge.py', str(list_num), str(list_len)],
        stdout=open(csv_file, 'w')
    )
    conn = sqlite3.connect(f'/tmp/{db_name}.db')
    with open(csv_file) as f:
        next(f)  # 先頭行（ヘッダ）を読み飛ばす
        conn.executemany(
            'insert into dat values (?,?,?,?,?,?,?)',
            [line.strip().split(',') for line in f]
        )
    conn.commit()
    conn.close()

def run_performance_test(db_name):
    print(f'Testing {db_name}.db')
    start_time = time.time()
    conn = sqlite3.connect(f'/tmp/{db_name}.db')
    conn.execute('drop table if exists T')
    conn.execute('''
        create table T as
        select A.id, A.list_num, A.item_num, B.id, B.list_num, B.item_num
        from dat A, dat B
        where A.item_num + 1 = B.item_num and A.list_num = B.list_num
    ''')
    count = conn.execute('select count(*) from T').fetchone()[0]
    conn.close()
    end_time = time.time()
    print(f'Count: {count}')
    print(f'Time taken: {end_time - start_time:.2f} seconds\n')

def main():
    test_sizes = [500000, 1000000, 2000000, 5000000]
    for size in test_sizes:
        create_database(str(size))
        populate_database(str(size), size, 5)
    for size in test_sizes:
        run_performance_test(str(size))

if __name__ == '__main__':
    main()

6. 二次索引の適用と性能評価

性能を最適化するため、二次索引（検索を速くするための補助的なデータ構造）を適用し、その効果を評価する。索引は item_num と list_num の組み合わせに対して作成する。これにより、エッジリスト生成時の検索性能の向上が期待される。

以下のPythonスクリプトは、索引を作成し、性能評価を行う。

#!/usr/bin/env python3
import sqlite3
import time

def create_index_and_test(db_name):
    print(f'Testing {db_name}.db with index')
    conn = sqlite3.connect(f'/tmp/{db_name}.db')
    conn.execute('create index idx001 on dat(item_num, list_num)')
    conn.close()

    start_time = time.time()
    conn = sqlite3.connect(f'/tmp/{db_name}.db')
    conn.execute('drop table if exists T')
    conn.execute('''
        create table T as
        select A.id, A.list_num, A.item_num, B.id, B.list_num, B.item_num
        from dat A, dat B
        where A.item_num + 1 = B.item_num and A.list_num = B.list_num
    ''')
    count = conn.execute('select count(*) from T').fetchone()[0]
    conn.close()
    end_time = time.time()
    print(f'Count: {count}')
    print(f'Time taken: {end_time - start_time:.2f} seconds\n')

def main():
    test_sizes = ['500000', '1000000', '2000000', '5000000']
    for size in test_sizes:
        create_index_and_test(size)

if __name__ == '__main__':
    main()