生成AI で PostgreSQL の SQL を作成

グリムの昔話に、「はちの女王」というのがあります。
その中に、「こけの中から1000粒の王女の真珠を探せ」という課題に、3人の兄弟の王子が取り組むというシーンがあります。
同様のことを、生成 AI にやらせてみます。

## 課題

```text
PostgreSQL で次の SQL を作成して下さい。

テーブルA とスキーマが与えられます。
テーブルA の総てのコラムを含むスキーマ内のテーブルB を見つける。
見つからない場合は、False として完了。
見つかった場合は、テーブルB の Not Null のコラムが総て、テーブルA に
含まれるかを判定する。
```

## 対象システム

1. [command-r-plus](https://huggingface.co/spaces/CohereForAI/c4ai-command-r-plus)
2. [wrtn](https://wrtn.jp)</li>
3. [ChatGPT](https://chatgpt.com/)</li>
4. [Gemini](https://gemini.google.com/app)</li>
5. [Groq](https://groq.com)</li>

## テスト環境の構築

データベース: city
ユーザー: scott
パスワード: tiger1233

```sql:prepare.sql
create schema sch001;
SET search_path TO sch001, public;
create table table_a (id varchar(10) primary key, name text not null, population int, date_mod date);
create table table_b (id varchar(10) primary key, latitude double precision, longtitude double precision);
create table table_c (id varchar(10) primary key, name text not null, height real);
create table table_p (id varchar(10) primary key, name text not null, area double precision, date_mod date);
create table table_q (id varchar(10) primary key, name text not null, temperature real);
create table table_r (id varchar(10) primary key, name text not null, population int, area double precision, date_mod date);
create table table_s (id varchar(10) primary key, name text not null, latitude double precision, longtitude double precision);
```

実行

```bash
psql -U scott city < prepare.sql
```

作成されたテーブル

```text
city=# set search_path=sch001;
SET
city=# \dt
        List of relations
 Schema |  Name   | Type  | Owner 
--------+---------+-------+-------
 sch001 | table_a | table | scott
 sch001 | table_b | table | scott
 sch001 | table_c | table | scott
 sch001 | table_p | table | scott
 sch001 | table_q | table | scott
 sch001 | table_r | table | scott
 sch001 | table_s | table | scott
(7 rows)
```

## テストシナリオ

スキーマ sch001 テーブル table_a を入れた場合　True

スキーマ sch001 テーブル table_b を入れた場合　False

スキーマ sch001 テーブル table_c を入れた場合　False

## シナリオを検証するプログラム

```py:find_table_b.py
#! /usr/bin/python
#
#	find_table_b.py
#
#					Jul/09/2024
#
# ------------------------------------------------------------------
import	sys
import	os
import	psycopg2
from dotenv import load_dotenv
#
from get_columns import get_columns_proc
from get_columns import get_tables_proc
from get_columns import not_null_pickup_proc
# ------------------------------------------------------------------
def not_null_check_proc(cursor_aa,schema,table_b,list_a):
	array_bb = not_null_pickup_proc(cursor_aa,schema,table_b)
#	print(array_bb)
#	print(list_a)
	hantei = all(elem in list_a for elem in array_bb)
#	print(hantei)
	return hantei
# ------------------------------------------------------------------
def get_table_columns_proc(cursor_aa,schema):
	array_tables = get_tables_proc(cursor_aa,schema)
#
	dict_columns = {}
	for table in array_tables:
		dict_columns[table] = get_columns_proc(cursor_aa,schema,table)
#
	return dict_columns
# ------------------------------------------------------------------
schema = sys.argv[1]
table_a = sys.argv[2]
#
dotenv_path = '.env'
load_dotenv(dotenv_path)
USER=os.environ.get("user")
PASSWORD=os.environ.get("password")
DATA_BASE=os.environ.get("data_base")
#
dsn="dbname=" + DATA_BASE + " user=" + USER + " password=" + PASSWORD
conn = psycopg2.connect(dsn)
cursor_aa = conn.cursor()
#
dict_columns=get_table_columns_proc(cursor_aa,schema)
#
list_a = dict_columns[table_a]
result = False
for table,vv in dict_columns.items():
	if table != table_a:
		hantei = all(elem in vv for elem in list_a)
		if hantei:
			if not_null_check_proc(cursor_aa,schema,table,list_a):
				result = True
#		print(table,hantei)
#
print(schema,table_a,result)
conn.commit ()
cursor_aa.close ()
conn.close ()
#
# ------------------------------------------------------------------
```

```py:get_columns.py
#
#	get_columns.py
#
#				Jul/09/2024
#
# --------------------------------------------------------
import sys
import psycopg2
#
# --------------------------------------------------------
# [2]:
def get_columns_proc(cursor_aa,schema,table_src):
	sql_str = "select column_name from information_schema.columns where table_name = '" + table_src + "' "
	sql_str += "AND table_schema = '" + schema + "' "
	cursor_aa.execute(sql_str)
#	print(sql_str)
	rows = cursor_aa.fetchall()
#	print(rows)
	array_aa = []
	for row in rows:
		array_aa.append(row[0])
#
	return array_aa
# --------------------------------------------------------
# [4]:
def not_null_pickup_proc(cursor_aa,schema,table_target):
	sql_str = "SELECT column_name FROM information_schema.columns "
	sql_str += "WHERE table_schema = '" + schema + "' "
	sql_str += "AND table_name = '" + table_target + "' "
	sql_str += "AND is_nullable = 'NO'"
#	print(sql_str)
	array_bb = []
	try:
		cursor_aa.execute(sql_str)
		rows = cursor_aa.fetchall()
#		print(len(rows))
		for row in rows:
#			print(row[0])
			array_bb.append(row[0])
	except Exception as ee:
		sys.stderr.write("*** error *** in cursor_aa.execute ***\n")
		sys.stderr.write(str(ee) + "\n")
	return array_bb
#
# --------------------------------------------------------
def get_tables_proc(cursor_aa,schema):
	sql_str = "select table_name from information_schema.tables where table_schema = '" + schema + "'"
	cursor_aa.execute(sql_str)
#	print(sql_str)
	rows = cursor_aa.fetchall()
#	print(rows)
	array_aa = []
	for row in rows:
		array_aa.append(row[0])
#
	return array_aa
# --------------------------------------------------------
```

```text:.env
user='scott'
password='******'
data_base='city'
```

スクリプト

```bash:go_check.sh
./find_table_b.py sch001 table_a
./find_table_b.py sch001 table_b
./find_table_b.py sch001 table_c
```

実行結果

```text
$ ./go_check.sh 
sch001 table_a True
sch001 table_b False
sch001 table_c False
```