Aurora DSQL Storage

Aurora DSQL storage 實作透過 Amazon Aurora DSQL 及 IAM 驗證提供儲存功能。

Aurora DSQL 不支援 PostgreSQL 擴充功能（CREATE EXTENSION），包括 pgvector。如需向量儲存，請使用獨立的 vector store，例如 @mastra/s3vectors。

安裝

npm install @mastra/dsql@beta

先決條件

• Amazon Aurora DSQL 叢集
• 具備 DSQL 叢集存取權限的 AWS 憑證（IAM 驗證）

使用方式

import { DSQLStore } from '@mastra/dsql'

const storage = new DSQLStore({
  id: 'my-dsql-store',
  host: 'abc123.dsql.us-east-1.on.aws',
  // region is auto-detected from host, or specify explicitly:
  // region: 'us-east-1',
  // user: 'admin', // default
  // database: 'postgres', // default
})

// Initialize the store (creates tables if needed)
await storage.init()

參數

id:

string

此 store 實例的唯一識別碼

host:

string

DSQL 叢集端點（例如 abc123.dsql.us-east-1.on.aws）

pool?:

pg.Pool

預先設定的 pg.Pool 實例。需要直接控制連線池時使用。不可與 host 設定同時使用。

user?:

string

資料庫使用者。Aurora DSQL 的管理員角色是 'admin'。

database?:

string

資料庫名稱。每個 Aurora DSQL 叢集只提供一個名為 'postgres' 的資料庫。

region?:

string

AWS 區域。如未提供，則從 host 擷取。

schemaName?:

string

建立 Mastra 資料表及索引的 PostgreSQL schema 名稱。

customCredentialsProvider?:

AwsCredentialIdentityProvider

用於 IAM 驗證的自訂 AWS 憑證 Provider。

max?:

number

連線池中的連線數目上限。

min?:

number

連線池中的連線數目下限。

idleTimeoutMillis?:

number

閒置連線經過此毫秒數後關閉。

maxLifetimeSeconds?:

number

連線生命週期上限（秒）。由於 Aurora DSQL 的連線上限為 60 分鐘，此值必須小於 3600。

connectionTimeoutMillis?:

number

取得連線的逾時時間（毫秒）。

allowExitOnIdle?:

boolean

允許程序在所有連線均閒置時結束。

Constructor 範例

你可以透過以下方式建立 DSQLStore 實例：

import { DSQLStore } from '@mastra/dsql'

// Basic configuration (region auto-detected from host)
const store1 = new DSQLStore({
  id: 'my-dsql-store',
  host: 'abc123.dsql.us-east-1.on.aws',
})

// With explicit region and schema
const store2 = new DSQLStore({
  id: 'my-dsql-store',
  host: 'abc123.dsql.us-east-1.on.aws',
  region: 'us-east-1',
  schemaName: 'my_app',
})

// With custom credentials provider
import { fromNodeProviderChain } from '@aws-sdk/credential-providers'

const store3 = new DSQLStore({
  id: 'my-dsql-store',
  host: 'abc123.dsql.us-east-1.on.aws',
  customCredentialsProvider: fromNodeProviderChain(),
})

// With connection pool settings
const store4 = new DSQLStore({
  id: 'my-dsql-store',
  host: 'abc123.dsql.us-east-1.on.aws',
  max: 20,
  min: 2,
  idleTimeoutMillis: 300000,
  maxLifetimeSeconds: 3000,
  connectionTimeoutMillis: 10000,
})

// Using a pre-configured pg.Pool
import { Pool } from 'pg'
import { AuroraDSQLClient } from '@aws/aurora-dsql-node-postgres-connector'

const pool = new Pool({
  host: 'abc123.dsql.us-east-1.on.aws',
  Client: AuroraDSQLClient,
  region: 'us-east-1',
})

const store5 = new DSQLStore({
  id: 'my-dsql-store',
  pool,
})

補充說明

Schema 管理

storage 實作會自動處理 schema 的建立和更新，並建立以下資料表：

• mastra_workflow_snapshot：儲存 Workflow 狀態及執行資料
• mastra_threads：儲存對話 thread
• mastra_messages：儲存個別訊息
• mastra_ai_spans：儲存用於 observability 的 span 資料
• mastra_scorers：儲存評分及評估資料
• mastra_resources：儲存資源 working memory 資料
• mastra_agents：儲存 Agent 資料

初始化

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

import { Mastra } from '@mastra/core'
import { DSQLStore } from '@mastra/dsql'

const storage = new DSQLStore({
  id: 'my-dsql-store',
  host: 'abc123.dsql.us-east-1.on.aws',
})

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

如果你不透過 Mastra 而直接使用 storage，便必須明確呼叫 init() 來建立資料表：

import { DSQLStore } from '@mastra/dsql'

const storage = new DSQLStore({
  id: 'my-dsql-store',
  host: 'abc123.dsql.us-east-1.on.aws',
})

// Required when using storage directly
await storage.init()

// Access domain-specific stores via getStore()
const memoryStore = await storage.getStore('memory')
const thread = await memoryStore?.getThreadById({ threadId: '...' })

注意

如果未呼叫 init()，系統便不會建立資料表，而 storage 操作會靜默失敗或擲回錯誤。

直接存取資料庫及連線池

DSQLStore 會將底層資料庫 client 及 pg.Pool 實例作為 public field 提供：

storage.db // Database client for executing queries
storage.pool // Underlying pg.Pool instance

它支援直接查詢及自訂 transaction 管理。使用這些 field 時：

• 你須負責妥善處理連線及 transaction。
• 如果連線池由 store 建立，關閉 store（storage.close()）時將會銷毀該連線池。
• 直接存取會略過 DSQLStore method 提供的任何額外邏輯或驗證。

此方式適用於需要低階存取的進階情境。

Aurora DSQL 特性

僅限 IAM 驗證

連線使用 IAM 驗證，無需資料庫密碼。@mastra/dsql 使用 @aws/aurora-dsql-node-postgres-connector 產生短期 auth token。你可以透過 customCredentialsProvider 提供自訂憑證 Provider。

單一資料庫、以 schema 隔離

每個叢集只提供一個 postgres 資料庫，並透過 schema 作邏輯分隔。schemaName 選項控制 Mastra 資料表的建立位置。

不支援 PostgreSQL 擴充功能

系統不支援 CREATE EXTENSION，包括 pgvector、PostGIS 等。如需向量儲存，請在 DSQLStore 之外另用獨立 store，例如 @mastra/s3vectors。

JSON 以文字儲存

JSON/JSONB 可用作查詢型別，但不可用作欄位型別。@mastra/dsql 會將結構化 field（metadata、content 等）儲存於 TEXT 欄位，並在查詢時轉換為 JSON。

Schema 及 DDL 限制

部分 PostgreSQL 功能無法使用：

• 外鍵約束
• TRUNCATE
• 同步 CREATE INDEX

系統使用 CREATE INDEX ASYNC 非同步建立索引。store 的 init() 及索引 helper API 均會遵守這些限制。

Transaction 及樂觀並行控制

Aurora DSQL 使用樂觀並行控制（OCC），在出現競爭時可能傳回可重試的 OCC 錯誤。transaction 的持續時間及大小均有限制。大型批次操作應在應用程式層分拆為較小的批次。

連線生命週期

個別連線的上限約為 60 分鐘。預設值 maxLifetimeSeconds: 3300 可確保連線在到達此上限前循環替換。

使用範例

為 Agent 加入 memory

要為 Agent 加入 Aurora DSQL memory，請使用 Memory class，並透過 DSQLStore 建立新的 storage key。host 應指向你的 Aurora DSQL 叢集端點。

src/mastra/agents/example-dsql-agent.ts

import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'
import { DSQLStore } from '@mastra/dsql'

export const dsqlAgent = new Agent({
  id: 'dsql-agent',
  name: 'DSQL Agent',
  instructions:
    'You are an AI agent with the ability to automatically recall memories from previous interactions.',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({
    storage: new DSQLStore({
      id: 'dsql-agent-storage',
      host: process.env.DSQL_HOST!,
    }),
    options: {
      generateTitle: true, // Explicitly enable automatic title generation
    },
  }),
})

使用 Agent

使用 memoryOptions 設定這次請求的 recall 範圍。設定 lastMessages: 5 以限制按近期程度 recall 的內容，並使用 semanticRecall 擷取最相關的 topK: 3 則訊息，當中包括 messageRange: 2 則相鄰訊息，提供每項配對結果附近的 context。

src/test-dsql-agent.ts

import 'dotenv/config'

import { mastra } from './mastra'

const threadId = '123'
const resourceId = 'user-456'

const agent = mastra.getAgent('dsql-agent')

const message = await agent.stream('My name is Mastra', {
  memory: {
    thread: threadId,
    resource: resourceId,
  },
})

await message.textStream.pipeTo(new WritableStream())

const stream = await agent.stream("What's my name?", {
  memory: {
    thread: threadId,
    resource: resourceId,
  },
  memoryOptions: {
    lastMessages: 5,
    semanticRecall: {
      topK: 3,
      messageRange: 2,
    },
  },
})

for await (const chunk of stream.textStream) {
  process.stdout.write(chunk)
}

索引管理

Aurora DSQL storage 提供索引管理功能，以改善查詢效能。

自動效能索引

Aurora DSQL storage 會在初始化期間，針對常見查詢模式自動建立複合索引：

• mastra_threads_resourceid_createdat_idx：（resourceId, createdAt）
• mastra_messages_thread_id_createdat_idx：（thread_id, createdAt）
• mastra_ai_spans_traceid_startedat_idx：（traceId, startedAt）
• mastra_ai_spans_parentspanid_startedat_idx：（parentSpanId, startedAt）
• mastra_ai_spans_name_idx：（name）
• mastra_ai_spans_spantype_startedat_idx：（spanType, startedAt）
• mastra_scores_trace_id_span_id_created_at_idx：（traceId, spanId, createdAt）

Aurora DSQL 使用 CREATE INDEX ASYNC 非同步建立這些索引。由於索引建立屬非同步操作，新索引在 init() 後未必可即時使用。store 在沒有這些索引的情況下仍可繼續運作，但查詢速度可能較慢，直至索引建立完成。

建立自訂索引

建立額外索引以改善特定查詢模式：

await storage.createIndex({
  name: 'idx_threads_resource',
  table: 'mastra_threads',
  columns: ['resourceId'],
})

await storage.createIndex({
  name: 'idx_messages_composite',
  table: 'mastra_messages',
  columns: ['thread_id', 'createdAt'],
})

Aurora DSQL 不允許在 CREATE INDEX ASYNC 中使用 ASC/DESC。如有加入，系統會自動移除。

索引選項

name:

string

索引的唯一名稱

table:

string

資料表名稱（例如 'mastra_threads'）

columns:

string[]

欄位名稱陣列。系統會自動移除 ASC/DESC 修飾符，以兼容 Aurora DSQL。

unique?:

boolean

建立唯一索引。

concurrent?:

boolean

Aurora DSQL 會忽略此選項。索引一律以非同步方式建立。

where?:

string

部分索引條件。

method?:

string

Aurora DSQL 會忽略此選項。只支援 btree 索引。

opclass?:

string

Aurora DSQL 會忽略此選項。

storage?:

Record<string, any>

Aurora DSQL 會忽略此選項。

tablespace?:

string

Aurora DSQL 會忽略此選項。不支援 tablespace。

管理索引

列出及監察現有索引：

// List all indexes
const allIndexes = await storage.listIndexes()
console.log(allIndexes)
// [
//   {
//     name: 'mastra_threads_pkey',
//     table: 'mastra_threads',
//     columns: ['id'],
//     unique: true,
//     size: '16 KB',
//     definition: 'CREATE UNIQUE INDEX...'
//   },
//   ...
// ]

// List indexes for a specific table
const threadIndexes = await storage.listIndexes('mastra_threads')

// Get detailed statistics for an index
const stats = await storage.describeIndex('idx_threads_resource')
console.log(stats)
// {
//   name: 'idx_threads_resource',
//   table: 'mastra_threads',
//   columns: ['resourceId'],
//   unique: false,
//   size: '128 KB',
//   definition: 'CREATE INDEX idx_threads_resource...',
//   method: 'btree',
//   scans: 1542,
//   tuples_read: 45230,
//   tuples_fetched: 12050
// }

// Drop an index
await storage.dropIndex('idx_threads_status')

指定 Schema 的索引

使用自訂 schema 時，索引會以 schema 前綴建立：

const storage = new DSQLStore({
  id: 'my-dsql-store',
  host: 'abc123.dsql.us-east-1.on.aws',
  schemaName: 'custom_schema',
})

// Creates index as: custom_schema_idx_threads_status
await storage.createIndex({
  name: 'idx_threads_status',
  table: 'mastra_threads',
  columns: ['status'],
})

相關資源

• Aurora DSQL 說明文件
• SQL 參考
• 支援的 SQL 功能
• 不支援的 PostgreSQL 功能
• 支援的資料型別
• 非同步索引