GoのSpannerクライアントについて ReadOnlyTransaction編

QualiArts Advent Calendar 2019、20日目担当のs9iです。

弊社の新規タイトルでは、バックエンドのプログラミング言語にGo、データベースにCloud Spannerを採用しています。今回はGoでSpannerを使用する際に使う、公式クライアントライブラリの使い方について書いていきます。

Cloud Spannerとは

2017年にGoogleから公開されたフルマネージドのデータベースです。
詳細な説明はここでは割愛しますが、

• リージョンを跨いだ水平スケール
• RDBとしての特性（スキーマ、トランザクション、SQL ...）
• ノード数を増やすほどスループット向上

といった特徴があります。弊社でも、RDBの特性と自動で行われる水平スケールに特に魅力を感じ、採用に至りました。

Spannerのトランザクション

Spannerには2種類のトランザクションがあります。

• ReadOnlyTransaction
• ReadWriteTransaction

この2種類を処理に応じて使い分けることで、より効率の良いデータベース操作が可能となります。本記事では、主にReadOnlyTransactionの使用方法を記します。
トランザクションの詳細についてはこちらの記事が非常に参考になります。

詳解 google-cloud-go/spanner — トランザクション編

Spannerクライアントの使用方法

本記事では、

• go 1.13
• cloud.google.com/go/spanner 1.1.0

を使用しています。

スキーマ定義

今回説明に使用するテーブルとして、以下のItemテーブルを定義します。
（本来、Spannerではプライマリキーに基づいてデータのシャーディングが行われるので、連続値等の偏りがある値をプライマリキーに指定するべきではありません。その他についても突っ込みどころは多々ありますが、説明用のテーブルということでご了承ください。）

CREATE TABLE Item(
  ItemID INT64 NOT NULL,
  ItemType STRING(MAX) NOT NULL,
  Name STRING(MAX) NOT NULL,
  Effect INT64 NOT NULL,
  Description STRING(MAX),
) PRIMARY KEY(ItemID, ItemType);
CREATE INDEX IdxItemName ON Item(Name);

こちらのテーブルに以下のデータを入れておきます。

ItemID	ItemType	Name	Effect	Description
1	Type1	アイテム1	10	説明1-1
1	Type2	アイテム2	20	説明1-2
1	Type3	アイテム3	30	説明1-3
1	Type4	アイテム4	40	説明1-4
2	Type1	アイテム5	50	説明2-1
2	Type2	アイテム6	60	説明2-2
2	Type3	アイテム7	70	説明2-3
3	Type1	アイテム8	80	説明3-1
3	Type2	アイテム9	90	説明3-2
4	Type1	アイテム10	100	NULL

Spannerクライアントの生成

cloud.google.com/go/spannerパッケージのNewClient()でクライアントを生成します。
セッションの設定や認証情報等を指定する場合は、NewClientWithConfig()を使用します。

import "cloud.google.com/go/spanner"

// projectID：GCPプロジェクトのID
// instance：Spannerのインスタンス名
// db：Spannerのデータベース名
func NewClient(ctx context.Context, projectID, instance, db string) (*spanner.Client, error) {
	dbPath := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instance, db)
	client, err := spanner.NewClient(ctx, dbPath)
	if err != nil {
		return nil, errors.New("Failed to Create Spanner Client.")
	}
	return client, nil
}

Read

SpannerクライアントのReadOnlyTransaction()を用いて読み取り専用トランザクションを開始します。直後にdeferでクローズ処理を登録しておきましょう。
取得したトランザクションのRead()で検索処理が実行できます。こちらも忘れずにdeferでイテレータのStop()を実行しましょう。

func ReadOnlyTxRead(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	// コンテキスト, テーブル名, キー, カラムの順で指定
	iter := rtx.Read(ctx, "Item", spanner.AllKeys(), []string{"ItemID", "ItemType", "Name"})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		// Read()で指定したカラム順で取得できる
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}
	return nil
}

# 出力結果
1 Type1 アイテム1
1 Type2 アイテム2
1 Type3 アイテム3
1 Type4 アイテム4
2 Type1 アイテム5
2 Type2 アイテム6
2 Type3 アイテム7
3 Type1 アイテム8
3 Type2 アイテム9
4 Type1 アイテム10

今回の例ではrow.Columns()を用いて値をバインドしていますが、構造体に一括でバインドするrow.ToStruct()や、カラム名を指定してバインドするrow.ColumnByName()も使用できます。

Read（カラムがNullを含む場合）

指定したカラムがNullを含む場合は、NullXxx型で値を取得した後、フィールドのValidで値の有無を判定します。

func ReadOnlyTxRead(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	iter := rtx.Read(ctx, "Item", spanner.AllKeys(), []string{"ItemID", "ItemType", "Name", "Description"})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		var uncheckedDescription spanner.NullString
		if err := row.Columns(&itemID, &itemType, &name, &uncheckedDescription); err != nil {
			return err
		}
		var description string
		if uncheckedDescription.Valid {
			description = uncheckedDescription.String()
		}
		fmt.Printf("%d %s %s %s\n", itemID, itemType, name, description)
	}
	return nil
}

# 出力結果
1 Type1 アイテム1 説明1-1
1 Type2 アイテム2 説明1-2
1 Type3 アイテム3 説明1-3
1 Type4 アイテム4 説明1-4
2 Type1 アイテム5 説明2-1
2 Type2 アイテム6 説明2-2
2 Type3 アイテム7 説明2-3
3 Type1 アイテム8 説明3-1
3 Type2 アイテム9 説明3-2
4 Type1 アイテム10

Read（特定のキーを指定する場合）

上記の例で使用していたspanner.AllKeys()はすべてのキーを指定するものでした。特定のキーを指定したい場合は、spanner.Keyでキーの順序に合わせて値を指定してあげます。

func ReadOnlyTxRead(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	keySet := spanner.KeySets(spanner.Key{1, "Type4"}, spanner.Key{3, "Type2"})
	iter := rtx.Read(ctx, "Item", keySet, []string{"ItemID", "ItemType", "Name"})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}
	return nil
}

# 出力結果
1 Type4 アイテム4
3 Type2 アイテム9

Read（キー範囲を指定する場合）

キーの範囲を指定する場合はspanner.KeyRangeを使用します。
Start, Endでキーの境界を指定し、Kindで境界値を含むかどうかを指定できます。

func ReadOnlyTxRead(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	keyRange := spanner.KeyRange{
		Start: spanner.Key{1, "Type4"},
		End:   spanner.Key{3, "Type2"},
		Kind:  spanner.OpenOpen,
	}
	iter := rtx.Read(ctx, "Item", keyRange, []string{"ItemID", "ItemType", "Name"})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}
	return nil
}

# 出力結果
2 Type1 アイテム5
2 Type2 アイテム6
2 Type3 アイテム7
3 Type1 アイテム8

ReadUsingIndex

セカンダリインデックスを指定して検索を行う場合は、ReadUsingIndex()を使用します。
Spannerのセカンダリインデックスは、1つのテーブルとして扱われ、そのインデックステーブルに含まれるのは以下のカラムだけです。

• プライマリキー
• インデックスとして指定したカラム
• Storing句で指定したカラム

そのため、ReadUsingIndexにおいては上記のカラム以外を取得することはできません。（後述のQuery()を使用することで上記以外のカラムも取得できますが、これは元テーブルとのJoinが実行されているのでパフォーマンス面の考慮が必要です。）

func ReadOnlyTxReadUsingIndex(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	// PKとインデックス以外のカラム（Effect, Description）は取得できない
	iter := rtx.ReadUsingIndex(ctx, "Item", "IdxItemName", spanner.AllKeys(), []string{"ItemID", "ItemType", "Name"})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}
	return nil
}

# 出力結果
1 Type1 アイテム1
4 Type1 アイテム10
1 Type2 アイテム2
1 Type3 アイテム3
1 Type4 アイテム4
2 Type1 アイテム5
2 Type2 アイテム6
2 Type3 アイテム7
3 Type1 アイテム8
3 Type2 アイテム9

ReadRow

1行だけを読み取る場合にはReadRow()が使用できます。
イテレータから読み取る処理が内部で行われています。

func ReadOnlyTxReadRow(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	row, err := rtx.ReadRow(ctx, "Item", spanner.Key{"2", "Type1"}, []string{"ItemID", "ItemType", "Name"})
	if err != nil {
		return err
	}

	var itemID int64
	var itemType, name string
	if err := row.Columns(&itemID, &itemType, &name); err != nil {
		return err
	}
	fmt.Printf("%d %s %s\n", itemID, itemType, name)
	return nil
}

# 出力結果
2 Type1 アイテム5

ReadWithOptions

ReadWithOptions()はspanner.ReadOptionsを引数に取り、使用するインデックスと取得件数を指定して検索処理を行います。

func ReadOnlyTxReadWithOptions(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	opts := &spanner.ReadOptions{
		Index: "IdxItemName",
		Limit: 2,
	}
	iter := rtx.ReadWithOptions(ctx, "Item", spanner.AllKeys(), []string{"ItemID", "ItemType", "Name"}, opts)
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}
	return nil
}

# 出力結果
1 Type1 アイテム1
4 Type1 アイテム10

Query

Query()を用いると任意の検索クエリを組み立てることができます。
キーバリューの形式でParamsを指定することで、プレースホルダーも使用できます。

func ReadOnlyTxQuery(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	sql := `SELECT ItemID, ItemType, Name FROM Item WHERE ItemID = @ItemID`
	params := map[string]interface{}{"ItemID": 2}
	iter := rtx.Query(ctx, spanner.Statement{SQL: sql, Params: params})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}
	return nil
}

# 出力結果
2 Type1 アイテム5
2 Type2 アイテム6
2 Type3 アイテム7

AnalyzeQuery

クエリの実行計画を取得したい場合は、AnalyzeQuery()を使用します。

func ReadOnlyTxAnalyzeQuery(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	sql := `SELECT ItemID, ItemType, Name FROM Item WHERE ItemID = @ItemID AND ItemType = @ItemType`
	params := map[string]interface{}{
		"ItemID":   2,
		"ItemType": "Type3",
	}
	plan, err := rtx.AnalyzeQuery(ctx, spanner.Statement{SQL: sql, Params: params})
	if err != nil {
		return err
	}

	for _, node := range plan.GetPlanNodes() {
		fmt.Printf("%v\n", node)
	}
	return nil
}

# 出力結果
kind:RELATIONAL display_name:"Distributed Union" child_links:<child_index:1 > child_links:<child_index:18 type:"Split Range" > metadata:<fields:<key:"subquery_cluster_node" value:<string_value:"1" > > > 
index:1 kind:RELATIONAL display_name:"Distributed Union" child_links:<child_index:2 > metadata:<fields:<key:"call_type" value:<string_value:"Local" > > fields:<key:"subquery_cluster_node" value:<string_value:"2" > > > 
index:2 kind:RELATIONAL display_name:"Serialize Result" child_links:<child_index:3 > child_links:<child_index:15 > child_links:<child_index:16 > child_links:<child_index:17 > 
index:3 kind:RELATIONAL display_name:"FilterScan" child_links:<child_index:4 > child_links:<child_index:14 type:"Seek Condition" > 
index:4 kind:RELATIONAL display_name:"Scan" child_links:<child_index:5 variable:"ItemID" > child_links:<child_index:6 variable:"ItemType" > child_links:<child_index:7 variable:"Name" > metadata:<fields:<key:"scan_target" value:<string_value:"Item" > > fields:<key:"scan_type" value:<string_value:"TableScan" > > > 
index:5 kind:SCALAR display_name:"Reference" short_representation:<description:"ItemID" > 
index:6 kind:SCALAR display_name:"Reference" short_representation:<description:"ItemType" > 
index:7 kind:SCALAR display_name:"Reference" short_representation:<description:"Name" > 
index:8 kind:SCALAR display_name:"Function" child_links:<child_index:9 > child_links:<child_index:10 > short_representation:<description:"($ItemID = @itemid)" > 
index:9 kind:SCALAR display_name:"Reference" short_representation:<description:"$ItemID" > 
index:10 kind:SCALAR display_name:"Parameter" short_representation:<description:"@itemid" > metadata:<fields:<key:"name" value:<string_value:"itemid" > > fields:<key:"type" value:<string_value:"scalar" > > > 
index:11 kind:SCALAR display_name:"Function" child_links:<child_index:12 > child_links:<child_index:13 > short_representation:<description:"($ItemType = @itemtype)" > 
index:12 kind:SCALAR display_name:"Reference" short_representation:<description:"$ItemType" > 
index:13 kind:SCALAR display_name:"Parameter" short_representation:<description:"@itemtype" > metadata:<fields:<key:"name" value:<string_value:"itemtype" > > fields:<key:"type" value:<string_value:"scalar" > > > 
index:14 kind:SCALAR display_name:"Function" child_links:<child_index:8 > child_links:<child_index:11 > short_representation:<description:"($ItemID = @itemid) AND ($ItemType = @itemtype)" > 
index:15 kind:SCALAR display_name:"Parameter" short_representation:<description:"@itemid" > metadata:<fields:<key:"name" value:<string_value:"itemid" > > fields:<key:"type" value:<string_value:"scalar" > > > 
index:16 kind:SCALAR display_name:"Parameter" short_representation:<description:"@itemtype" > metadata:<fields:<key:"name" value:<string_value:"itemtype" > > fields:<key:"type" value:<string_value:"scalar" > > > 
index:17 kind:SCALAR display_name:"Reference" short_representation:<description:"$Name" > 
index:18 kind:SCALAR display_name:"Function" child_links:<child_index:19 > child_links:<child_index:22 > short_representation:<description:"(($ItemID = @itemid) AND ($ItemType = @itemtype))" > 
index:19 kind:SCALAR display_name:"Function" child_links:<child_index:20 > child_links:<child_index:21 > short_representation:<description:"($ItemID = @itemid)" > 
index:20 kind:SCALAR display_name:"Reference" short_representation:<description:"$ItemID" > 
index:21 kind:SCALAR display_name:"Parameter" short_representation:<description:"@itemid" > metadata:<fields:<key:"name" value:<string_value:"itemid" > > fields:<key:"type" value:<string_value:"scalar" > > > 
index:22 kind:SCALAR display_name:"Function" child_links:<child_index:23 > child_links:<child_index:24 > short_representation:<description:"($ItemType = @itemtype)" > 
index:23 kind:SCALAR display_name:"Reference" short_representation:<description:"$ItemType" > 
index:24 kind:SCALAR display_name:"Parameter" short_representation:<description:"@itemtype" > metadata:<fields:<key:"name" value:<string_value:"itemtype" > > fields:<key:"type" value:<string_value:"scalar" > > > 

QueryWithStats

QueryWithStats()を使用することで、クエリの実行統計を取得できます。（RowIteratorのNext()がDoneを返した後に使用できます。）また、AnalyzeQuery()同様にQueryPlanフィールドで実行計画も取得できます。

func ReadOnlyTxQueryWithStats(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	sql := `SELECT ItemID, ItemType, Name FROM Item WHERE ItemID = @ItemID AND ItemType = @ItemType`
	params := map[string]interface{}{
		"ItemID":   2,
		"ItemType": "Type3",
	}
	iter := rtx.QueryWithStats(ctx, spanner.Statement{SQL: sql, Params: params})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}

	// クエリの実行統計
	for k, v := range iter.QueryStats {
		fmt.Printf("%s: %v\n", k, v)
	}
	// 実行計画も取得できる（省略）
	// fmt.Printf("QueryPlan: %v\n", iter.QueryPlan)

	return nil
}

# 出力結果
2 Type3 アイテム7
elapsed_time: 4.25 msecs
data_bytes_read: 0
query_text: SELECT ItemID, ItemType, Name FROM Item WHERE ItemID = @ItemID AND ItemType = @ItemType
rows_returned: 1
filesystem_delay_seconds: 0 msecs
query_plan_creation_time: 2.34 msecs
deleted_rows_scanned: 0
bytes_returned: 30
cpu_time: 2.34 msecs
remote_server_calls: 0/0
rows_scanned: 1
runtime_creation_time: 0 msecs

Single

Single()を使用することで明示的にReadOnlyTransaction()を実行せずとも、テンポラリなトランザクション上でクエリを実行できます。単一の検索クエリを投げるだけであればこちらの方が簡単に使用できます。

func ReadOnlyTxSingleQuery(ctx context.Context, client *spanner.Client) error {
	sql := `SELECT ItemID, ItemType, Name FROM Item WHERE ItemID = @ItemID AND ItemType = @ItemType`
	params := map[string]interface{}{
		"ItemID":   2,
		"ItemType": "Type3",
	}
	iter := client.Single().Query(ctx, spanner.Statement{SQL: sql, Params: params})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			fmt.Println(err)
			return err
		}
		var itemID int64
		var itemType, name string
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}
	return nil
}

# 出力結果
2 Type3 アイテム7

WithTimestampBound

Spannerの読み取り時に使用するタイムスタンプには、次の3つのモードが存在します。（デフォルトはStrong reads）

• Strong reads（最新データを読み取り）
• Exact staleness（指定タイムスタンプより古くないデータを読み取り）
• Bounded staleness（指定タイムスタンプと一致するデータを読み取り）

WithTimestampBound()を用いて、トランザクション内での読み取りモードを指定することができます。（この指定は主に、マルチリージョンで効果を発揮します。）

func ReadOnlyTxQueryWithTimestampBound(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	sql := `SELECT ItemID, ItemType, Name FROM Item WHERE ItemID = @ItemID AND ItemType = @ItemType`
	params := map[string]interface{}{
		"ItemID":   2,
		"ItemType": "Type3",
	}
	iter := rtx.WithTimestampBound(spanner.ExactStaleness(15*time.Second)).Query(ctx, spanner.Statement{SQL: sql, Params: params})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}

	return nil
}

# 出力結果
2 Type3 アイテム7

Timestamp

Timestamp()を実行すると、トランザクションにおけるタイムスタンプを取得できます。
タイムスタンプはクエリの実行後に取得可能です。

func ReadOnlyTxTimestamp(ctx context.Context, client *spanner.Client) error {
	rtx := client.ReadOnlyTransaction()
	defer rtx.Close()

	t, _ := rtx.Timestamp()
	fmt.Println("Timestamp: ", t)

	sql := `SELECT ItemID, ItemType, Name FROM Item WHERE ItemID = @ItemID AND ItemType = @ItemType`
	params := map[string]interface{}{
		"ItemID":   2,
		"ItemType": "Type3",
	}
	iter := rtx.Query(ctx, spanner.Statement{SQL: sql, Params: params})
	defer iter.Stop()

	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var itemID int64
		var itemType, name string
		if err := row.Columns(&itemID, &itemType, &name); err != nil {
			return err
		}
		fmt.Printf("%d %s %s\n", itemID, itemType, name)
	}

	iter = rtx.WithTimestampBound(spanner.ExactStaleness(15*time.Second)).Query(ctx, spanner.Statement{SQL: sql, Params: params})
	t, _ = rtx.Timestamp()
	fmt.Println("Timestamp: ", t)

	return nil
}

# 出力結果
Timestamp:  0001-01-01 00:00:00 +0000 UTC
2 Type3 アイテム7
Timestamp:  2019-12-19 20:57:24.615294 +0900 JST

まとめ

今回はGoのSpannerクライアントでReadOnlyTransactionを扱う方法をまとめました。
ReadWriteTransactionの読み取り処理も、ReadOnlyTransactionを埋め込んで実装されているので、今回のポイントを抑えておけば問題なく使用できると思います。また機会があれば、ReadWriteTransactionの書き込み処理についても触れたいと思います。
最後までお付き合いいただきありがとうございました。