Google Cloud Spanner 儲存

Google Cloud Spanner 儲存實作為 Mastra 提供可水平擴展至高容量、具強一致性的儲存 backend。此實作以 Cloud Spanner 的 GoogleSQL dialect 為目標。

安裝

npm

npm install @mastra/spanner@latest

pnpm

pnpm add @mastra/spanner@latest

Yarn

yarn add @mastra/spanner@latest

Bun

bun add @mastra/spanner@latest

用法

import { SpannerStore } from '@mastra/spanner'

const storage = new SpannerStore({
  id: 'spanner-storage',
  projectId: process.env.SPANNER_PROJECT_ID!,
  instanceId: process.env.SPANNER_INSTANCE_ID!,
  databaseId: process.env.SPANNER_DATABASE_ID!,
})

instance 及資料庫必須已經存在。adapter 會在首次使用時建立所需資料表，因此提供予 Spanner client 的 credential 必須有權執行 schema 變更（或在部署步驟中以較高權限的 credential 執行一次 storage.init()）。

參數

id:

string

此 storage instance 的唯一識別碼。

projectId?:

string

Google Cloud project ID。除非已提供 database，否則此項為必填。

instanceId?:

string

Cloud Spanner instance ID。除非已提供 database，否則此項為必填。

databaseId?:

string

Cloud Spanner database ID。除非已提供 database，否則此項為必填。

database?:

@google-cloud/spanner Database

預先設定的 Spanner Database handle。當你在其他位置管理 Spanner client 時使用（例如跨服務共用驗證或連線選項）。

spannerOptions?:

object

轉交予 @google-cloud/spanner client constructor 的選項。可用於設定 credential、自訂 endpoint，或指向本機 emulator。

disableInit?:

boolean

= false

設為 true 時，首次使用不會自動建立資料表。你必須在另一個部署步驟中明確呼叫 storage.init()。

skipDefaultIndexes?:

boolean

= false

設為 true 時，初始化期間不會建立預設 index。

indexes?:

CreateIndexOptions[]

要建立的自訂 secondary index。每個 index 都必須指定所屬資料表，系統會根據資料表名稱將 index 路由至相應 domain。

initMode?:

'sync' | 'validate'

= 'sync'

控制 schema 初始化行為。'sync' 會在 init() 期間建立缺少的資料表、欄位及 index（以往的行為）。'validate' 不會發出任何 DDL，而是驗證每個預期的資料表、欄位及預設／自訂 index 是否已存在；如有任何項目缺少，便會拋出具類型的使用者錯誤。當 schema 由外部程序（Terraform、Liquibase、release pipeline 等）擁有，而 Mastra 只應加以驗證時，這個選項非常適合。

Constructor 範例

你可以透過多種方式建立 SpannerStore instance：

import { Spanner } from '@google-cloud/spanner'
import { SpannerStore } from '@mastra/spanner'

// Using projectId / instanceId / databaseId
const store1 = new SpannerStore({
  id: 'spanner-storage-1',
  projectId: 'my-gcp-project',
  instanceId: 'my-instance',
  databaseId: 'mastra',
})

// Reusing an existing Spanner Database handle
const spanner = new Spanner({ projectId: 'my-gcp-project' })
const database = spanner.instance('my-instance').database('mastra')

const store2 = new SpannerStore({
  id: 'spanner-storage-2',
  database,
})

// Using the local Spanner emulator (set the SPANNER_EMULATOR_HOST env var)
process.env.SPANNER_EMULATOR_HOST = 'localhost:9010'
const store3 = new SpannerStore({
  id: 'spanner-storage-emulator',
  projectId: 'test-project',
  instanceId: 'test-instance',
  databaseId: 'test-db',
  spannerOptions: { servicePath: 'localhost', port: 9010, sslCreds: undefined },
})

補充說明

Schema 管理

儲存 adapter 會建立以下資料表，全部使用 GoogleSQL dialect：

• mastra_workflow_snapshot：Workflow 狀態及執行資料
• mastra_threads：對話 thread
• mastra_messages：個別訊息
• mastra_resources：resource working memory
• mastra_scorers：評估分數
• mastra_background_tasks：背景 Tool 執行狀態
• mastra_agents：精簡 Agent 記錄（id、狀態、使用中版本）
• mastra_agent_versions：設有版本的 Agent 設定 snapshot
• mastra_mcp_clients / mastra_mcp_client_versions：MCP client 設定及其版本歷史
• mastra_mcp_servers / mastra_mcp_server_versions：MCP server 設定及其版本歷史
• mastra_skills / mastra_skill_versions：Skill 記錄及設有版本的 Skill snapshot（指示、參考資料、script、asset、內容樹）
• mastra_skill_blobs：以 SHA-256 hash 為 key 的 content-addressable blob store，用於 Skill 版本內容
• mastra_prompt_blocks / mastra_prompt_block_versions：prompt block 記錄及設有版本的內容 snapshot（template 內容、規則、request-context schema）
• mastra_scorer_definitions / mastra_scorer_definition_versions：scorer definition 記錄及設有版本的設定 snapshot（judge 指示、model、分數範圍、preset 設定、預設 sampling）
• mastra_schedules / mastra_schedule_triggers：由 cron 驅動的 Workflow 排程及觸發歷史，由 Mastra 內置的 WorkflowScheduler 使用
• mastra_workspaces / mastra_workspace_versions：Workspace 記錄及設有版本的設定 snapshot（檔案系統、Sandbox、mount、搜尋、Skill、Tool）
• mastra_datasets / mastra_dataset_items / mastra_dataset_versions：評估 dataset、設有 SCD-2 版本的項目及版本 snapshot
• mastra_experiments / mastra_experiment_results：實驗執行及各項目結果
• mastra_favorites：每位使用者收藏的 Agent 及 Skill，並在 parent record 上維護 denormalized favoriteCount
• mastra_channel_installations / mastra_channel_config：多平台 channel 安裝及各平台設定
• mastra_ai_spans：用於可觀測性的 AI tracing span（每個 Trace 及每個 span 的記錄，用於支援 Studio Trace UI）

文字及 JSON payload 使用 STRING(MAX) 建立資料表，另亦使用 INT64、FLOAT64、BOOL 及 TIMESTAMP。

以下資料表包含 Spanner 特有的 STORED generated column。adapter 會從 JSON payload 填入這些欄位，讓常用 filter 可以使用一般 secondary index，而毋須執行 JSON_VALUE scan：

• mastra_workflow_snapshot.snapshotStatus：從 snapshot 擷取 $.status，支援 listWorkflowRuns({ status })。
• mastra_schedules.target_workflow_id：從 target 擷取 $.workflowId，支援 listSchedules({ workflowId })。

兩者均會在 init() 期間透過 ALTER TABLE ... ADD COLUMN IF NOT EXISTS 加入；在 initMode: 'validate' 下則會略過（此時 schema 由外部擁有）。缺少欄位時，adapter 會在 runtime 改用 JSON_VALUE filter。

adapter 不會建立或使用 schema。請使用專用資料庫作隔離。

初始化

將 storage 傳入 Mastra class 時，系統會在任何儲存操作前自動呼叫 init()：

import { Mastra } from '@mastra/core'
import { SpannerStore } from '@mastra/spanner'

const storage = new SpannerStore({
  id: 'spanner-storage',
  projectId: process.env.SPANNER_PROJECT_ID!,
  instanceId: process.env.SPANNER_INSTANCE_ID!,
  databaseId: process.env.SPANNER_DATABASE_ID!,
})

const mastra = new Mastra({
  storage, // init() is called automatically
})

如直接使用 storage，請在首次操作前呼叫一次 init()。Spanner 不允許同時變更 schema，因此 SpannerStore.init() 會按順序執行每個 domain 的設定。

const storage = new SpannerStore({
  id: 'spanner-storage',
  projectId: process.env.SPANNER_PROJECT_ID!,
  instanceId: process.env.SPANNER_INSTANCE_ID!,
  databaseId: process.env.SPANNER_DATABASE_ID!,
})

await storage.init()
const memory = await storage.getStore('memory')
const thread = await memory?.getThreadById({ threadId: '...' })

注意

如未呼叫 init() 而 disableInit 為 true，所需資料表將不存在，儲存操作亦會失敗。

GoogleSQL 特點

部分行為與其他 relational adapter 不同：

• Upsert 使用 INSERT OR UPDATE。Spanner 不為 upsert 提供 RETURNING clause，因此需要取得寫入後狀態的 caller 必須重新讀取。
• 不設 TRUNCATE。dangerouslyClearAll() 會發出 DELETE WHERE TRUE。
• 識別符以 backtick 括起。
• DDL 透過 database.updateSchema(...) 套用，這是非同步的長時間操作。
• 不支援 NULLS FIRST/LAST。使用 IS NULL ordering key 模擬包含 NULL 處理的排序。
• 不原生支援 JSON containment。listTraces 的 metadata 及 scope filter 會編譯為逐一 key 的 JSON_VALUE(...) = @v 等值檢查，而 tags filter 則會編譯為對 JSON_QUERY_ARRAY(...) 使用 EXISTS。這與 Postgres 的 @> containment operator 不同（後者可以在一次 index scan 中比對巢狀結構）：大部分單次 lookup 仍然適用，但無法表達深層巢狀結構比對。

直接存取資料庫

SpannerStore 會公開底層 Spanner client object：

store.database // @google-cloud/spanner Database
store.instance // @google-cloud/spanner Instance (when created internally)
store.spanner // @google-cloud/spanner Spanner client (when created internally)

這些 object 適用於自訂交易或 schema introspection 等進階情境。直接重用資料庫時，會繞過 adapter 的驗證及 JSON 轉換邏輯。

使用 emulator 作本機開發

使用 Docker 在本機執行 Cloud Spanner emulator：

docker run -p 9010:9010 -p 9020:9020 gcr.io/cloud-spanner-emulator/emulator

設定 SPANNER_EMULATOR_HOST=localhost:9010，並在執行應用程式前建立 instance 及資料庫：

gcloud spanner instances create test-instance --config=emulator-config --nodes=1
gcloud spanner databases create test-db --instance=test-instance

然後在 Node.js process 中設定相同的 env var 並連線。@google-cloud/spanner client 會自動偵測 emulator。