Corda5におけるVault-Named Queryの活用法

公開日

May 14, 2024

カテゴリ

Corda技術を知る

タグ

⭐Corda5 🔰Corda入門 🧑‍💻CorDapp開発

筆者

新井

この記事で学べること

Vault-Named Queryの基本的な使い方
Valut-Named Queryにおけるページネーションの実装例

⚠️

今回説明する Vault-Named Query は Corda API ver release-5.2.0.52に準拠しています． API はバージョンによってよって仕様が異なることご承知おきください．

Vault-Named Queryとは
API
Vault-Named Queries で使用する API
必須の準備
ContractStateVaultJsonFactory
VaultNamedQueryFactory
全てのクエリを組み合わせる
Vault-Named Queries を実行する
結果を取り出す
ページネーション
シーク法①
制約
課題
シーク法②
制約
課題
結論

Vault-Named Queryとは

一言で説明すると Corda5 から実装されたStateをSQLのように柔軟に検索するためのAPIです．Corda4 で言うところの Custom query にあたります．

通常，トランザクション ID とインデックス等で State を台帳からクエリしてくるところを，Stateをユーザが定義したフィールドを持つJSONオブジェクトにシリアライズし，jsonb型でDBに格納し，クエリに際しては，JSONの任意のフィールドに対し，Postgres演算子を用いて検索をかけることができます．（一部未対応の演算子あり）

API

任意のStateを自身のValutからクエリするためにCordapp開発者が使用する API（関数）は大きく分けて以下の二つがあります．

Vault-Named Query で使用する API
UtxoLedgerService で使用する API

Vault-Named Queries で使用する API

Vault-Named Query を利用するために以下の準備を行う必要があります．

必須の準備

以下のクラスをcontractsパッケージ配下に作成します．

ContractStateVaultJsonFactory を implements したクラスの定義
VaultNamedQueryFactory を implements したクラスの定義

ContractStateVaultJsonFactory

こちらは visible_states.custom_representation(jsonb 型)に記録される Json を定義するためのクラスです．

⚠️

visible_states.custom_representationは State を JSON オブジェクトに変換した値を保持するテーブルの列名です．Corda4 では「１State対１レコード」でマッピングしていましたが，Corda 5 からは JSON オブジェクト変換したものを「１State対１カラム」に挿入します．

Corda4 で言うところの MappedSchema を implements したスキーマクラスにあたります．ここで定義したフィールドでクエリできるようになります．

例：State の定義

package com.r3.corda.demo.contract

class TestContract : Contract {
    override fun verify(transaction: UtxoLedgerTransaction) {}
}

@BelongsToContract(TestContract::class)
class TestState(
    val testField: String,
    private val participants: List<PublicKey>
) : ContractState {
    override fun getParticipants(): List<PublicKey> = participants
}

例：JsonFactory の定義

class TestStateJsonFactory : ContractStateVaultJsonFactory<TestState> {
    override fun getStateType(): Class<TestState> = TestState::class.java
    override fun create(state: TestState, jsonMarshallingService: JsonMarshallingService): String {
        return jsonMarshallingService.format(state)
    }
}

⚠️

マッピングできる型には制限があります．com.fasterxml.jackson.databind.ObjectMapperがデフォルトでサポーしている型およびnet.corda.v5.crypto.SecureHash がマッピングできます．

VaultNamedQueryFactory

こちらは具体的な検索条件を定義するためのクラスです．例では State の testField がパラメータで渡された testFiled に一致するものを検索条件としてします．利用可能なオペレータは後述します．

例：

class DummyCustomQueryFactory : VaultNamedQueryFactory {
    override fun create(vaultNamedQueryBuilderFactory: VaultNamedQueryBuilderFactory) {
        vaultNamedQueryBuilderFactory.create("DUMMY_CUSTOM_QUERY")
            .whereJson(
                "WHERE visible_states.custom_representation -> 'com.r3.corda.demo.contract.TestState' " +
                        "->> 'testField' = :testField"
            )
            .register()
    }
}

whereJson()で指定した条件で DB からクエリしてきますが，その結果についてオンメモリで以下の操作を行うことができます．

filter()
map()
collect()

また以下の関数は DB に対して直接作用することができます．

orderBy()
selectUnconsumedStatesOnly()

それぞれの機能について見ていきます，

whereJson

SQL WHERE 句を定義します．ただし，現行バージョンで使用できる演算子は以下に限られています．

IN
LIKE
IS NOT NULL
IS NULL
AS
OR
AND
!=
>
<
>=
<=
>
>>
?
::

また，CAST()や to_timestamp()などの関数使用することができません．代わりにtimestamp で検索範囲を区切りたい場合，以下のようになります．

class DummyCustomQueryGetByCreatedTimeQueryFactory : VaultNamedQueryFactory {
    override fun create(vaultNamedQueryBuilderFactory: VaultNamedQueryBuilderFactory) {
        vaultNamedQueryBuilderFactory.create("DUMMY_CUSTOM_QUERY")
            .whereJson(
                "WHERE visible_states.custom_representation -> 'com.r3.corda.demo.contract.TestState' IS NOT NULL " +
                    "AND visible_states.created >= (:fromDateTime::timestamp) " +
                    "AND visible_states.created <= (:toDateTime::timestamp) " +
                    "AND visible_states.consumed IS NULL",
            )
            .map(DummyCustomQueryTransformer())
            .register()
    }
}

orderBy

任意カラムまたは JSON のフィールドで並び替えができます．以下はトランザクション ID の昇順で並び替えた例です．

例：

class DummyCustomQueryFactory : VaultNamedQueryFactory {
    override fun create(vaultNamedQueryBuilderFactory: VaultNamedQueryBuilderFactory) {
        vaultNamedQueryBuilderFactory.create("DUMMY_CUSTOM_QUERY")
            .whereJson(
                "WHERE visible_states.custom_representation -> 'com.r3.corda.demo.contract.TestState' " +
                        "->> 'testField' = :testField"
            )
            .orderBy("visible_states.transaction_id", "ASC")
            .register()
    }
}

selectUnconsumedStatesOnly

クエリ対象を未消費の State に絞り込むことができます．

⚠️

Github 上のサンプルなどでは，未消費の State に絞り込むときにvisible_states.consumed IS NULLを使用している例が多数ありますが，この条件句はクエリ中に State が消費された場合，返すべき State をスキップする可能性があるので，任意時点の断面を取得したい場合（任意時点の残高照会）は非推奨となっています． selectUnconsumedStatesOnlyではクエリが開始された時点のタイムスタンプがクエリの基準時点に設定されるため，ページネーションが完了する前に State が消費された場合でも，~~悪い出力~~クエリ結果の整合性が担保できます．

例：

class DummyCustomQueryFactory : VaultNamedQueryFactory {
    override fun create(vaultNamedQueryBuilderFactory: VaultNamedQueryBuilderFactory) {
        vaultNamedQueryBuilderFactory.create("DUMMY_CUSTOM_QUERY")
            .whereJson(
                "WHERE visible_states.custom_representation -> 'com.r3.corda.demo.contract.TestState' " +
                        "->> 'testField' = :testField"
            )
            .selectUnconsumedStatesOnly()
            .register()
    }
}

filter

クエリしてきた値をフィルタリングできます．ただしフィルタリングするためには事前にフィルタリング用のクラスを定義してしておく必要があります．

例：

class DummyCustomQueryFilter : VaultNamedQueryStateAndRefFilter<TestState> {
    override fun filter(data: StateAndRef<TestUtxoState>, parameters: MutableMap<String, Any>): Boolean {
        return data.state.contractState.participantNames.contains("Alice")
    }
}

上記の例では，取ってきた State について「フィールド participantNames が"Alice"を含む」という条件でフィルタリングしています．

🔔

フィルタリングするだけなら Java の Stream API の filter でいいんじゃないかと思ったのではないでしょうか．わざわざフィルタを定義しておく意味はメリットは何でしょうか？それはフィルタリングした上で，目的の件数まで API が値を取ってきてくれるからです．例えば，フィルタを使わず「フィールド participantNames が"Alice"を含む」というフィルタを Stream API で行った場合，クエリ結果が 100 件欲しいのにフィルタリングしたせいで 10 件しか取得できなかったということが発生すると思います．しかし，上記フィルタクラスを使用したフィルタリングでは，フィルタリングの結果足りなくなった件数分を再度クエリしてきてくれます．

map

クエリしてきた値を任意の型に変換できます．ただしフィルタリング同様，変換するためには事前に変換用のクラスを定義してしておく必要があります．

例：

class DummyCustomQueryTransformer : VaultNamedQueryStateAndRefTransformer<TestState, String> {
    override fun transform(data: StateAndRef<TestState>, parameters: MutableMap<String, Any>): String {
        return data.ref.transactionId.toString()
    }
}

上記の例では，取ってきた State について「その State を含むトランザクションの ID」に変換しています．

collect

クエリしてきた値の件数を Int で返します．例によって変換するためには事前に変換用のクラスを定義してしておく必要があります．

例：

class DummyCustomQueryCollector : VaultNamedQueryCollector<String, Int> {
    override fun collect(
        resultSet: MutableList<String>,
        parameters: MutableMap<String, Any>
    ): VaultNamedQueryCollector.Result<Int> {
        return VaultNamedQueryCollector.Result(
            listOf(resultSet.size),
            true
        )
    }
}

全てのクエリを組み合わせる

上で紹介した全てのクエリ条件を組み合わせると以下のようになります．

例：

class DummyCustomQueryFactory : VaultNamedQueryFactory {
    override fun create(vaultNamedQueryBuilderFactory: VaultNamedQueryBuilderFactory) {
        vaultNamedQueryBuilderFactory.create("DUMMY_CUSTOM_QUERY")
            .whereJson(
                "WHERE visible_states.custom_representation -> 'com.r3.corda.demo.contract.TestState' " +
                        "->> 'testField' = :testField"
            )
            .orderBy("visible_states.transaction_id", "ASC")
            .filter(DummyCustomQueryFilter())
            .map(DummyCustomQueryTransformer())
            .collect(DummyCustomQueryCollector())
            .register()
    }
}

⚠️

filter(), map(), collect()は順序関係があり，map()→filter()→collect()の順に実行する必要があります．

Vault-Named Queries を実行する

Vault-Named Query は UtxoLedgerService を使って以下のように実行できます．

例：

val resultSet = utxoLedgerService.query("DUMMY_CUSTOM_QUERY", Integer::class.java) // DocではIntになっていますが，エラーになるのでIntegerを指定してください．
            .setLimit(1000)
            .setOffSet(0) // 5.1から廃止されました．使用すると実行時エラーになります．
            .setParameter("testField", "dummy")
            .setCreatedTimestampLimit(Instant.now())
            .execute()

それぞれの関数について説明していきます．

setLimit(): 一度に返す結果の件数を設定できます．最大値はInt.MAX(2,147,483,647)です．
setParameter(): DummyCustomQueryFactory で定義した SQL に渡すパラメータを設定します．全ての定義済みパラメータを渡さないと例外を投げます．また，setParameters()を用いて，一度に複数のパラメータを kotlin.collections.Map で渡すことができます．
setCreatedTimestampLimit():クエリ対象を任意の時点以前に作成された State に絞り込むことができます．

参考 setOffset()の実装

    override fun setOffset(offset: Int): VaultNamedParameterizedQuery<T> {
        throw UnsupportedOperationException("This query does not support offset functionality.")
    }

結果を取り出す

上で実行したクエリの結果を取り出しましょう．以下のように取り出すことができます．

例：

val resultSet = utxoLedgerService.query("DUMMY_CUSTOM_QUERY", Int::class.java)
            .setLimit(1000)
            .setParameter("testField", "dummy")
            .setCreatedTimestampLimit(Instant.now())
            .execute()
var results = resultSet.results
while (resultSet.hasNext()) {
    results = resultSet.next()
}

Vault-Named Query の使い方は以上になります．お疲れ様でした.

ページネーション

上記のサンプルコードで示した通り，既存のオフセット法のによるページネーションでは，一連のクエリ操作中に検索対象のStateが消費されてしまうと，本来検索されるべきState がクエリ結果から漏れてしまう可能性があるため，5.1 からオフセットが廃止されてしました．

クエリ結果の整合性を担保するためにはこの仕様を受け入れざるを得ませんが，このままでは Corda を利用するフロントエンドから n ページ目のクエリがあった場合，Corda では毎回 n 回の問い合わせを DB にする必要が出てきます．それには以下の問題があります．

Corda 側の負担が大きく，件数が増えるとレスポンスが著しく劣化する恐れがある．
結局ページネーションの度にFlowをCallするので，結局フロントエンドから見ると整合性が担保されていない．(整合性が保たれるのは Flow のライフサイクル内のみ)

次章ではこれらの解決方法について考えてみます．

シーク法①

シーク法の基本的な考え方はページネーションの開始点と終了点を決定し，それらの間の範囲内のデータを取得することです．具体的には、次の手順に従います：

ページネーションの開始点と終了点を決定します．開始点は前のページの最後の要素に基づいており，終了点は現在のページの最後の要素に基づいています。
インデックスを使用して開始点と終了点の間のデータを検索します．これにより，必要なページのデータのみが取得されます．
次のページを取得する際には，前のページの終了点を新しい開始点として使用し，同様の手順で次の範囲のデータを取得します．

ここでは created を一意な値とみなし，created をカーソルとしてページネーションを実現します．

created を使用したカーソルページネーションの実装例を示します：

class DummyCustomQueryFactory : VaultNamedQueryFactory {
    override fun create(vaultNamedQueryBuilderFactory: VaultNamedQueryBuilderFactory) {
        vaultNamedQueryBuilderFactory.create("DUMMY_PAGINATION")
            .whereJson(
                "WHERE visible_states.custom_representation -> 'com.r3.corda.demo.contract.TestState' IS NOT NULL " +
                    "AND visible_states.created > (:created::timestamp) " + // ・・・①
                    "AND (visible_states.consumed < (:start::timestamp) " + // ・・・②
                    "OR visible_states.consumed IS NULL)" // ・・・③
            )
            .orderBy("visible_states.created", "ASC") // ・・・④
            .register()
    }
}

① クエリ開始位置を指定します．
② クエリ開始時点の基準地点を指定します．これを指定することで一連のクエリ中に消費されてしまった State も検索対象に含めることができます．
③ 未使用の State のみをクエリするように指定します．この時，上で「visible_states.consumed IS NULLは使わないようにしましょう」と言いましたが，selectUnconsumedStatesOnlyでは Flow をまたぐクエリで整合性を担保できないのでここでは使用しません．② と ③ を組み合わせることで一連のクエリ結果の整合性を保っています．
クエリ開始位置となるカラムでソートします．

次に上で定義した Vault-Named Query を呼ぶ時の実装を見てみます：

初回のクエリ

val resultSet = utxoLedgerService.query("DUMMY_PAGINATION", StateAndRef::class.java)
            .setLimit(1000)
            .setParameter("created", 0)
            .setParameter("start", 1700000000)
            .execute()

2 回のクエリ（初回クエリ結果の最後のオブジェクトの created が 1600000000 だった場合）

val resultSet = utxoLedgerService.query("DUMMY_PAGINATION", StateAndRef::class.java)
            .setLimit(1000)
            .setParameter("created", 1600000000)
            .setParameter("start", 1700000000)
            .execute()

制約

ページ番号

この方法はページ番号を表示するページネーションはサポートしていません．対応可能なのは『前』と『次』のページネーションのみです．

データベースインデックス

この方法では，オフセット法よりパフォーマンスで勝るには order by で使用するカラムにインデックスを貼る必要があります．つまり，呼び出しもとに複数のソートのオプションを提供する場合，複数のカラムにインデックスを貼る必要がでてきます．従って，インデックスの数が増えると書き込みのパフォーマンスは落ちてきます．（またパフォーマンスを気にせずクエリ結果の整合性のみを重視するならインデックスなしでの運用も考えられます．）

ユニークネス

この方法では，クエリの開始点となるカラムはユニークである必要があります．

select * from visible_states where `created` > 1600000000 order by `created` asc limit 100;

上で示した実装は created が重複しない前提ですが，例えば同じ created のオブジェクトが複数あり，一度のクエリで created が同じオブジェクト群の途中までしか返せなかった場合，次のクエリで開始位置が動くので同じ created の残りのオブジェクトがスキップされます．したがって，created が重複する可能性がある Cordapp は別の方法を考える必要があります．逆に言えば，created が重複する恐れがない Cordapp がこの方法は有用と考えられます．

課題

結局ページ番号指定はできない
クエリ開始位置を毎回再計算する必要がある．
utxo_visible_transaction_outputのプライマリキーはtransaction_id, group_idx, leaf_idxであり，created にはインデックスが貼られていないためパフォーマンスはオフセット法と変わらない．（ただし今後のアップデートで任意のカラムにインデックスを貼れるようになる可能性はあります．）

シーク法②

次に，created が重複する可能性がある Cordapp について，created が重複した場合でも整合性を担保できる方法を考えます．

至って当然の考え方ですが， visible_states テーブルの複合主キーと created を基準点としクエリを実装します：．

class DummyCustomQueryFactory : VaultNamedQueryFactory {
    override fun create(vaultNamedQueryBuilderFactory: VaultNamedQueryBuilderFactory) {
        vaultNamedQueryBuilderFactory.create("DUMMY_PAGINATION")
            .whereJson(
                "WHERE visible_states.custom_representation -> 'com.r3.corda.demo.contract.TestState' IS NOT NULL " +
                    "AND visible_states.created > (:created::timestamp) " + // ・・・①
                    "AND visible_states.transaction_id > :transaction_id " + // ・・・②
                    "AND visible_states.group_idx > (:group_idx::int) " + // ・・・③
                    "AND visible_states.leaf_idx > (:leaf_idx::int) " + // ・・・④
                    "AND (visible_states.consumed < (:start::timestamp) " + // ・・・⑤
                    "OR visible_states.consumed IS NULL)" // ・・・⑥
            )
            .orderBy("visible_states.created", "ASC") // ・・・⑦
            .orderBy("visible_states.transaction_id", "ASC") // ・・・⑦
            .orderBy("visible_states.group_idx", "ASC") // ・・・⑦
            .orderBy("visible_states.leaf_idx", "ASC") // ・・・⑦
            .register()
    }
}

①②③④ クエリの開始点を指定します
⑤ クエリ開始時点の基準地点を指定します．これを指定することで一連のクエリ中に消費されてしまった State も検索対象に含めることができます．
⑥ 未使用の State のみをクエリするように指定します．この時上で「visible_states.consumed IS NULLは使わないようにしましょう」と言いましたが，selectUnconsumedStatesOnlyでは Flow をまたぐクエリで整合性を担保できないのでここでは使用しません．⑤ と ⑥ を組み合わせることで一連のクエリ結果の整合性を保っています．
⑦ クエリ開始点でのカラムでソートします．

次に上で定義した Vault-Named Query を呼ぶ時の実装を見てみます：

初回のクエリ

val resultSet = utxoLedgerService.query("DUMMY_PAGINATION", StateAndRef::class.java)
            .setLimit(1000)
            .setParameter("created", 0)
            .setParameter("transaction_id", "SHA-256D:0000000000000000000000000000000000000000000000000000000000000000")
            .setParameter("group_idx", 0)
            .setParameter("leaf_idx", 0)
            .setParameter("start", 1700000000)
            .execute()

2 回目のクエリ初回クエリ結果の最後のオブジェクトの created が 1600000000，transaction_id が SHA-256D:1111111111111111111111111111111111111111111111111111111111111111，group_idx が 5，leaf_idx が 2 だった場合）

val resultSet = utxoLedgerService.query("DUMMY_PAGINATION", StateAndRef::class.java)
            .setLimit(1000)
            .setParameter("created", 1600000000)
            .setParameter("transaction_id", "SHA-256D:1111111111111111111111111111111111111111111111111111111111111111")
            .setParameter("group_idx", 5)
            .setParameter("leaf_idx", 2)
            .setParameter("start", 1700000000)
            .execute()

制約

シーク法①と同じです．

課題

結局ページ番号指定はできない
クエリ開始位置を毎回再計算する必要がある．

結論

ページネーション全般に言えることですが銀の弾丸はありませんでした．沸きらない答えとなってしまいますが，Cordapp の要件に応じて適切なアルゴリズムを選択して行くことが肝要でしょう．

📬

最後までお読みいただきありがとうございます。当社へのご質問・ご要望がございましたら、📪SBI R3 Japan お問い合わせフォーム📪よりお気軽にお問い合わせください！

＜ご質問・ご要望の例＞

Corda Portalの記事について質問したい
ブロックチェーンを活用した新規事業を相談したい
企業でのブロックチェーン活用方法を教えて欲しい　等々

📢

また、厳選されたCordaに関する最新情報をお伝えるするメールマガジンやX、当社主催のイベントコミュニティを運営しております。ぜひご登録ください。

✍️

Written by 新井 (ARAI)

SBI R3 Japan エンジニアリング部

アプリケーションアーキテクト/PoC支援

刃牙シリーズが好きです

→筆者の記事一覧