【次世代高速RDB 劔"Tsurugi"】Tsurugi UDFを利用してリモート環境のAI処理をSQL一本で実行する

create table prediction_results (
  file_name varchar(*) not null,
  sequence int not null,
  bbox_x real not null,
  bbox_y real not null,
  bbox_width real not null,
  bbox_height real not null,
  confidence real not null,
  predicted_image blob not null,
  primary key (file_name, sequence)
);

create table images (
  file_name varchar(*) not null,
  image blob not null,
  primary key (file_name)
);

$ python -m venv udf-sample
$ source udf-sample/bin/activate

> python -m venv udf-sample
> udf-sample\Scripts\activate

$ pip install grpcio-tools
$ pip install ultralytics

$ curl -OL https://github.com/project-tsurugi/tsurugi-udf/archive/refs/tags/0.2.0.tar.gz
$ tar xf 0.2.0.tar.gz
$ cd tsurugi-udf-0.2.0/udf-library/python
$ pip install .

udf-sample
├── tsurugidb
│   └── udf
│       └── tsurugi_types.proto
└── udf-sample.proto

syntax = "proto3";

import "tsurugidb/udf/tsurugi_types.proto";

package udf_sample;

service UdfSample {
  // 1
  rpc predict_image (ImageData) returns (stream PredictionResult);
}

// 2
message ImageData {
  string file_name = 1;
  tsurugidb.udf.BlobReference image_data = 2;
}

// 3
message PredictionResult {
  int32 sequence = 1;
  float bbox_x = 2;
  float bbox_y = 3;
  float bbox_width = 4;
  float bbox_height = 5;
  float confidence = 6;
  tsurugidb.udf.BlobReference predicted_image = 7;
}

$ python -m grpc_tools.protoc -I. -I./tsurugidb/udf --python_out=. --grpc_python_out=. udf-sample.proto

from concurrent import futures
import os
from pathlib import Path

import grpc
from grpc import ServicerContext
from tsurugidb.udf import create_blob_client
from udf_sample_pb2 import ImageData, PredictionResult
from udf_sample_pb2_grpc import UdfSampleServicer, add_UdfSampleServicer_to_server
from ultralytics import YOLO


class SampleServer(UdfSampleServicer):

    def __init__(self):
        # 1
        self.model = YOLO("yolo26n.pt")

    def predict_image(self, request: ImageData, context: ServicerContext):
        file_name = request.file_name
        print(f"ファイル {file_name} に対して推論を実行します")
        with create_blob_client(context) as blob_client:
            # 2
            download_path = Path(file_name)
            if download_path.exists():
                os.remove(download_path)
            blob_client.download_blob(request.image_data, download_path)

            # 3
            results = self.model(file_name)
            for i, result in enumerate(results):
                # 4
                result_file_path = f"{file_name}_{i}.jpg"
                result.save(result_file_path)

                # 5
                boxes = result.boxes
                for j, conf in enumerate(boxes.conf):
                    sequence = i + j
                    xywh = boxes.xyxy[j]
                    bbox_x = xywh[0]
                    bbox_y = xywh[1]
                    bbox_width = xywh[2]
                    bbox_height = xywh[3]

                    # 6
                    upload_path = Path(result_file_path)
                    blob_ref = blob_client.upload_blob(upload_path)

                    # 7
                    yield PredictionResult(sequence=sequence, bbox_x=bbox_x, bbox_y=bbox_y,
                                           bbox_width=bbox_width, bbox_height=bbox_height, confidence=conf,
                                           predicted_image=blob_ref)


def serve():
    try:
        server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
        add_UdfSampleServicer_to_server(SampleServer(), server)
        # 8
        server.add_insecure_port("[::]:50051")
        server.start()
        server.wait_for_termination()
    except KeyboardInterrupt:
        print("停止します")
    finally:
        server.stop(0)


if __name__ == "__main__":
    serve()

$ python udf_server.py

$ curl -OL https://github.com/ultralytics/assets/releases/download/v0.0.0/coco8.zip
$ unzip coco8.zip

coco8
├── LICENSE
├── README.md
├── images
│   ├── train
│   └── val
└── labels
    ├── train
    └── val

package org.example;

import com.tsurugidb.iceaxe.TsurugiConnector;
import com.tsurugidb.iceaxe.sql.parameter.TgBindParameters;
import com.tsurugidb.iceaxe.sql.parameter.TgBindVariables;
import com.tsurugidb.iceaxe.sql.parameter.TgParameterMapping;
import com.tsurugidb.iceaxe.sql.type.TgBlob;
import com.tsurugidb.iceaxe.transaction.manager.TgTmSetting;
import com.tsurugidb.iceaxe.transaction.option.TgTxOption;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;

public class Uploader {
    private final TsurugiConnector connector;

    public Uploader() {
        this.connector = TsurugiConnector.of("ipc:tsurugi");
    }

    public static void main(String[] args) throws Exception {
        if (args.length < 1) {
            System.out.println("アップロードする画像のディレクトリを指定してください");
            System.exit(1);
        }
        new Uploader().execute(Path.of(args[0]));
    }

    public void execute(Path dir) throws IOException, InterruptedException {
        var sql = "insert or replace into images (file_name, image) values (:file_name, :image)";
        var variables = TgBindVariables.of().addString("file_name").addBlob("image");
        var mapping = TgParameterMapping.of(variables);
        try (var session = connector.createSession();
             var stmt = session.createStatement(sql, mapping);
             var fileListStream = Files.list(dir)) {
            var filePathList = fileListStream.toList();
            var tgTmSetting = TgTmSetting.of(TgTxOption.ofOCC());
            var tm = session.createTransactionManager(tgTmSetting);
            tm.execute(tx -> {
                for (Path filePath : filePathList) {
                    var fileName = filePath.getFileName().toString();
                    var blob = TgBlob.of(filePath);
                    var params = TgBindParameters.of().addString("file_name", fileName).addBlob("image", blob);
                    tx.executeAndGetCount(stmt, params);
                }
            });
        }
    }
}

$ java -cp "/path/to/lib/*" org.example.Uploader /path/to/coco8/images/train

tgsql> select * from images;
start transaction implicitly. option=[
  type: OCC
  label: "tgsql-implicit-transaction2026-03-05 20:40:21.985+09:00"
]
Time: 33.951 ms
[file_name: VARCHAR(*), image: BLOB]
[000000000009.jpg, blob@0]
[000000000025.jpg, blob@1]
[000000000030.jpg, blob@2]
[000000000034.jpg, blob@3]
(4 rows)

$ cd tsurugi-udf-0.2.0/udf-plugin
$ pip install .

$ udf-plugin-builder --proto udf-sample.proto

$ ls -l
total 1828
-rwxr-xr-x 1 aoe aoe  180176 Mar  5 21:04 libtsurugi_types.so
-rw-r--r-- 1 aoe aoe      81 Mar  5 21:04 libudf-sample.ini
-rwxr-xr-x 1 aoe aoe 1674200 Mar  5 21:04 libudf-sample.so
drwxr-xr-x 9 aoe aoe    4096 Mar  5 21:02 tmp
drwxr-xr-x 3 aoe aoe    4096 Mar  5 20:58 tsurugidb
-rwxr-xr-x 1 aoe aoe     491 Mar  5 20:57 udf-sample.proto

[udf]
enabled=true
endpoint=dns:///localhost:50051
secure=false
transport=stream

[udf]
enabled=true
endpoint=dns:///192.168.201.3:50051 # リモート接続の場合はgRPCサーバーのアドレスに変更
secure=false
transport=stream

[grpc_server]
#enabled=true
endpoint=dns:///192.168.201.10:52345
#listen_address=0.0.0.0:52345

insert or replace into prediction_results
select
  i.file_name,
  p.sequence,
  p.bbox_x,
  p.bbox_y,
  p.bbox_width,
  p.bbox_height,
  p.confidence,
  p.predicted_image
from images as i
apply predict_image(i.file_name, i.image) as p;

tgsql> insert or replace into prediction_results
     | select
     |   i.file_name,
     |   p.sequence,
     |   p.bbox_x,
     |   p.bbox_y,
     |   p.bbox_width,
     |   p.bbox_height,
     |   p.confidence,
     |   p.predicted_image
     | from images as i
     | apply predict_image(i.file_name, i.image) as p;
start transaction implicitly. option=[
  type: OCC
  label: "tgsql-implicit-transaction2026-03-06 17:09:26.065+09:00"
]
Time: 1.034 ms
(9 rows merged)
Time: 550.167 ms
transaction commit(DEFAULT) finished implicitly.
Time: 5.568 ms

tgsql> select * from prediction_results;
start transaction implicitly. option=[
  type: OCC
  label: "tgsql-implicit-transaction2026-03-06 17:09:35.389+09:00"
]
Time: 0.803 ms
[file_name: VARCHAR(*), sequence: INT, bbox_x: REAL, bbox_y: REAL, bbox_width: REAL, bbox_height: REAL, confidence: REAL, predicted_image: BLOB]
[000000000009.jpg, 0, 311.58472, 0.0, 630.928, 238.57057, 0.91180897, blob@18]
[000000000009.jpg, 1, 0.0, 190.48422, 619.58685, 480.0, 0.7473526, blob@19]
[000000000009.jpg, 2, 263.9724, 231.55132, 571.2544, 479.85583, 0.6965864, blob@20]
[000000000009.jpg, 3, 0.0, 8.265778, 439.20117, 345.323, 0.6897867, blob@21]
[000000000009.jpg, 4, 325.6366, 408.4717, 424.6326, 479.71454, 0.48934183, blob@22]
[000000000025.jpg, 0, 378.4063, 68.28476, 601.26447, 356.92383, 0.934395, blob@23]
[000000000030.jpg, 0, 239.04587, 154.69746, 406.92102, 350.6156, 0.88299567, blob@24]
[000000000030.jpg, 1, 200.10843, 31.734947, 461.2537, 352.72113, 0.333447, blob@25]
[000000000034.jpg, 0, 0.77467346, 21.598114, 439.7422, 400.88147, 0.95875555, blob@26]
(9 rows)
Time: 3.264 ms
transaction commit(DEFAULT) finished implicitly.
Time: 3.372 ms

tgsql> begin;
transaction started. option=[
  type: OCC
]
Time: 0.698 ms

tgsql> select * from prediction_results where file_name = '000000000009.jpg';
[file_name: VARCHAR(*), sequence: INT, bbox_x: REAL, bbox_y: REAL, bbox_width: REAL, bbox_height: REAL, confidence: REAL, predicted_image: BLOB]
[000000000009.jpg, 0, 311.58472, 0.0, 630.928, 238.57057, 0.91180897, blob@36]
[000000000009.jpg, 1, 0.0, 190.48422, 619.58685, 480.0, 0.7473526, blob@37]
[000000000009.jpg, 2, 263.9724, 231.55132, 571.2544, 479.85583, 0.6965864, blob@38]
[000000000009.jpg, 3, 0.0, 8.265778, 439.20117, 345.323, 0.6897867, blob@39]
[000000000009.jpg, 4, 325.6366, 408.4717, 424.6326, 479.71454, 0.48934183, blob@40]
(5 rows)
Time: 3.35 ms

tgsql> \store blob@36 /path/to/000000000009_result.jpg
tgsql> commit;
transaction commit(DEFAULT) finished.
Time: 3.997 ms