Aurora DSQL 存储

Aurora DSQL 存储实现使用带 IAM 身份验证的 Amazon Aurora DSQL 提供存储功能。

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 凭据提供程序。

max?:

number

连接池中的最大连接数。

min?:

number

连接池中的最小连接数。

idleTimeoutMillis?:

number

经过这么多毫秒后关闭空闲连接。

maxLifetimeSeconds?:

number

最大连接生命周期（秒）。由于 Aurora DSQL 的连接上限为 60 分钟，该值必须小于 3600。

connectionTimeoutMillis?:

number

获取连接的超时时间（毫秒）。

allowExitOnIdle?:

boolean

所有连接均空闲时允许进程退出。

构造函数示例

可以通过以下方式实例化 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 管理

该存储实现会自动处理 schema 的创建和更新。它会创建以下表：

• mastra_workflow_snapshot：存储 Workflow 状态和执行数据
• mastra_threads：存储对话 thread
• mastra_messages：存储单条消息
• mastra_ai_spans：存储用于 observability 的 span 数据
• mastra_scorers：存储评分和评估数据
• mastra_resources：存储资源工作记忆数据
• mastra_agents：存储 Agent 数据

初始化

将 storage 传递给 Mastra class 时，任何存储操作之前都会自动调用 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()，将不会创建表，存储操作会静默失败或抛出错误。

直接访问数据库和连接池

DSQLStore 将底层数据库 client 和 pg.Pool 实例作为公共字段公开：

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

它支持直接查询和自定义事务管理。使用这些字段时：

• 你负责正确处理连接和事务。
• 如果连接池由 store 创建，关闭 store（storage.close()）将销毁该连接池。
• 直接访问会绕过 DSQLStore 方法提供的任何附加逻辑或验证。

此方法适用于需要低级访问的高级场景。

Aurora DSQL 特性

仅 IAM 身份验证

连接通过 IAM 进行身份验证，无需数据库密码。@mastra/dsql 使用 @aws/aurora-dsql-node-postgres-connector 生成短期身份验证令牌。你可以通过 customCredentialsProvider 提供自定义凭据提供程序。

单一数据库，基于 schema 的隔离

每个集群提供一个 postgres 数据库。逻辑隔离通过 schema 实现。schemaName 选项控制 Mastra 表的创建位置。

不支持 PostgreSQL 扩展

不支持 CREATE EXTENSION，包括 pgvector、PostGIS 等。对于向量存储，请将 @mastra/s3vectors 等独立 store 与 DSQLStore 一起使用。

以文本存储 JSON

JSON/JSONB 可作为查询类型使用，但不可作为列类型。@mastra/dsql 将结构化字段（metadata、content 等）存储在 TEXT 列中，并在查询时转换为 JSON。

Schema 和 DDL 限制

部分 PostgreSQL 功能不可用：

• 外键约束
• TRUNCATE
• 同步 CREATE INDEX

索引通过 CREATE INDEX ASYNC 异步创建。store 的 init() 和索引辅助 API 遵守这些限制。

事务和乐观并发

Aurora DSQL 使用乐观并发控制（OCC），在竞争情况下可能返回可重试的 OCC 错误。事务的持续时间和大小均有限制。应在应用层将大型批量操作拆分为较小批次。

连接生命周期

单个连接限制约为 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 条相邻消息。

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[]

列名数组。为兼容 Aurora DSQL，会自动移除 ASC/DESC 修饰符。

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 功能
• 支持的数据类型
• 异步索引