Tsurugi DBにデータを入力する

> tgctl start
> tgsql  -c ipc://tsurugi
tgsql> create table test_table (id int primary key, value DOUBLE,name varchar(20));
tgsql> insert into test_table ( id ,value ,name) values (1,2.3,'aaa');

~$ cat test.sql
create table test_table (id int primary key, value DOUBLE,name varchar(20));
insert into test_table ( id ,value ,name) values (1,1.3,'a'),(2,2.3,'b');
>tgsql  -c ipc://tsurugi --script test.sql

> export emb_ver=0.11.5
> curl -L https://github.com/embulk/embulk/releases/download/v${emb_ver}/embulk-${emb_ver}.jar -o embulk-${emb_ver}.jar
> java -jar embulk-${emb_ver}.jar install io.github.hishidama.embulk:embulk-output-tsurugidb:1.1.0
> tgsql  -c ipc://tsurugi
tgsql> create table test_table (id int primary key)
tgsql> \exit
> cat 1.yml
in:
  type: file
  path_prefix: input/input.csv
  parser:
    type: csv
    columns:
    - {name: id, type: long}

out:
  type:
    source: maven
    group: io.github.hishidama.embulk
    name: tsurugidb
    version: 1.1.0
  endpoint: ipc://tsurugi
  table: test_table
  tx_type: OCC
  method: put
  method_option: put_overwrite
> cat input/input.csv
id
1
2
3
4
> java  -jar embulk-${emb_ver}.jar run 1.yml
> tgsql  -c ipc://tsurugi
tgsql> select * from test_table;

[id: INT4]
[1]
[2]
[3]
[4]

> gzip input.csv
> file input.csv.gz
input.csv.gz: gzip compressed data, was "input.csv"
> cat 1.yml
    in:
      type: file
      path_prefix: input.csv.gz
      decoders:
      - {type: gzip}
      parser:
        type: csv
...
> java  -jar embulk-${emb_ver}.jar run 1.yml

> bzip2 input.csv
> file input.csv.bz2
input.csv.bz2: bzip2 compressed data, block size = 900
> cat 1.yml
    in:
      type: file
      path_prefix: input.csv.bz2
      decoders:
      - {type: bzip2}
      parser:
        type: csv
...
> java  -jar embulk-${emb_ver}.jar run 1.yml

import csv
import subprocess
import time
import os

table_name = "test_table"
columns = [
    {"name": "id", "type": "INT PRIMARY KEY", "csv_type": "long"},
    {"name": "name", "type": "INT", "csv_type": "long"},
    {"name": "note", "type": "INT", "csv_type": "long"}
]
start_row = 1
end_row = 20

def check_tsurugi_status():
    """Check the status of the Tsurugi OLTP database and ensure it is running."""
    while True:
        try:
            # Run the "tgctl status" command and capture the output
            result = subprocess.run(["tgctl", "status"], capture_output=True, text=True)
            status_output = result.stdout.strip()

            if "Tsurugi OLTP database is RUNNING" in status_output:
                print("ok")
                break  # Exit the loop if the database is running
            elif "Tsurugi OLTP database is INACTIVE" in status_output:
                print("Database is INACTIVE. Starting the database...")
                subprocess.run(["tgctl", "start"], check=True)
            else:
                print("Unexpected status. Retrying in 10 seconds...")
        except subprocess.CalledProcessError as e:
            print(f"Error while executing command: {e}")
        except Exception as e:
            print(f"An unexpected error occurred: {e}")

        time.sleep(10)  # Wait 10 seconds before rechecking

def execute_sql_script():
    """Create and execute an SQL script to set up the test table."""
    column_definitions = ", ".join([f"{col['name']} {col['type']}" for col in columns])
    sql_content = f"""
    DROP TABLE IF EXISTS {table_name};
    CREATE TABLE {table_name} ({column_definitions});
    """

    script_filename = "test_table_script.sql"
    with open(script_filename, "w") as script_file:
        script_file.write(sql_content)

    try:
        result = subprocess.run([
            "tgsql", "-c", "ipc://tsurugi", "--script", script_filename
        ], capture_output=True, text=True, check=True)

        print("Execution Output:")
        print(result.stdout)
    except subprocess.CalledProcessError as e:
        print("Error executing tgsql command:")
        print(e.stderr)
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

def create_yml_file():
    """Create a YAML configuration file for Embulk."""
    column_definitions = "\n".join([
        f"        - {{name: {col['name']}, type: {col['csv_type']}}}" for col in columns
    ])
    yml_content = f"""
    in:
      type: file
      path_prefix: input/input.csv
      parser:
        type: csv
        columns:
{column_definitions}

    out:
      type:
        source: maven
        group: io.github.hishidama.embulk
        name: tsurugidb
        version: 1.1.0
      endpoint: ipc://tsurugi
      table: {table_name}
      tx_type: OCC
      method: put
      method_option: put_overwrite
    """
    yml_filename = "1.yml"
    with open(yml_filename, "w") as yml_file:
        yml_file.write(yml_content)
    return yml_filename

def create_csv_file(start, end):
    """Create a CSV file with the specified range of rows."""
    csv_filename = "input/input.csv"
    header = [col['name'] for col in columns]
    rows = [[i, 1, 1] for i in range(start, end + 1)]

    os.makedirs(os.path.dirname(csv_filename), exist_ok=True)

    with open(csv_filename, "w", newline="") as csv_file:
        writer = csv.writer(csv_file)
        writer.writerow(header)
        writer.writerows(rows)
    return csv_filename

def run_embulk(yml_filename):
    """Run the Embulk command with the given YAML configuration file."""
    try:
        result = subprocess.run([
            "java", "-jar", "embulk-0.11.5.jar", "run", yml_filename
        ], capture_output=True, text=True, check=True)
        print("Embulk Execution Output:")
        print(result.stdout)
    except subprocess.CalledProcessError as e:
        print("Error executing embulk command:")
        print(e.stderr)
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

if __name__ == "__main__":
    check_tsurugi_status()
    execute_sql_script()
    yml_file = create_yml_file()
    create_csv_file(start_row, end_row)
    run_embulk(yml_file)