@flowblade/sqlduck

@ flowblade/ sqlduck

目前试验

• 来自Zod schemas的 DuckDB 表创建.
• 容易地摄取来自发电机或动能的数据.

快开始

创建数据库连接

import { DuckDBInstance } from "@duckdb/node-api";
DuckDBInstance.create(undefined, {
  access_mode: "READ_WRITE",
  max_memory: "512M",
});
export const conn = await instance.connect();

将数据添加到数据库

import { DuckDatabaseManager, Table } from "@flowblade/sqlduck";
import { DuckTableLoader } from "@flowblade/sqlduck/loader";
import * as z from "zod";

import { conn } from "./db.config.ts";

const dbManager = new DuckDatabaseManager(conn);
const database = await dbManager.attach({
  type: "memory", // can be 'filesystem', ...
  alias: "mydb",
  options: { COMPRESS: "false" },
});

// Define a zod schema, it will be used to create the table
const userSchema = z.object({
  id: z.int32().min(1).meta({ primaryKey: true }),
  name: z.string(),
});

// Example of a datasource (can be an array, generator, async iterable...)
async function* getUsers(): AsyncIterableIterator<z.input<typeof userSchema>> {
  // database or api call
  yield { id: 1, name: "John" };
  yield { id: 2, name: "Jane" };
}

const loader = new DuckTableLoader({ conn });

// Create a table from the schema and load the rows
const result = await loader.load({
  table: new Table({ name: "user", database: database.alias }),
  schema: userSchema, // The schema to use to create the table
  rows: getUsers(), // The rows to load
  // 👇Optional:
  mode: "createOrReplace", // Default is "create"
  progress: {
    intervalMs: 1000, // Default, or everyRows: 100_000
    onProgress: ({ intervalMs, totalRows, rowsPerSecond }) => {
      console.log(
        `Loaded ${totalRows} rows, est: ${rowsPerSecond} rows/s over ${intervalMs}ms`
      );
    },
  },
});

console.log(`Inserted ${result.totalRows} rows in ${result.timeMs}ms`);
console.log(`Table created with DDL: ${result.createTableDDL}`);

const reader = await conn.runAndReadAll("select * from mydb.user");
const rows = reader.getRowObjectsJS();
// [{id: 1, name: 'John'}, {id: 2, name: 'Jane'}]]

装入策略(默认情况下的DuckDB附件、行附件、准备的有上行的插入)、模式、检查点、进度和调取在 装入器文档。 。 。 .

导出查询

QueryExporter 将参数化的 SQL-tag 查询导出到 DuckDB 的 CSV、 Parquet 或 Excel (XLSX) COPY,所以行从不通过JavaScript. 每个持久性战略都有一个方法:

import { DuckdbDatasource, sql } from "@flowblade/source-duckdb";
import { QueryExporter } from "@flowblade/sqlduck/export";

const exporter = new QueryExporter({
  datasource: new DuckdbDatasource({ connection: conn }),
});

导出为文件

const file = await exporter.toFile({
  sql: sql`SELECT id, name FROM user WHERE active = ${true}`,
  format: { type: "parquet", compression: "zstd" },
  file: "/data/exports/users.parquet",
});

console.log(`${file.path}: ${file.rowCount} rows, ${file.sizeBytes} bytes`);

目录 file 必须存在。 一个已存在的文件被替换, 但只有在导出完成后: 一个失败的导出没有留下部分文件, 并且保留了先前的文件 .

导出为对象存储

toObjectStorage 将导出直接上传到 Azure Blob 存储器(az://, (中文(简体) ), abfss://页:1s3://),谷歌云存储器(gs://)或 Cloudflare R2(r2://) (中文(简体) ). DuckDB 写给存储器本身: 本地没有写出任何内容 。 其他urls和本地路径被拒绝.

import { DuckSecretsManager } from "@flowblade/sqlduck";

await new DuckSecretsManager(conn).create({
  name: "exports_azure",
  type: "azure",
  connectionString: process.env.AZURE_STORAGE_CONNECTION_STRING,
  scope: "az://exports/",
});

const exported = await exporter.toObjectStorage({
  sql: sql`SELECT id, name FROM user WHERE active = ${true}`,
  format: { type: "parquet" },
  url: "az://exports/2026/users.parquet",
});

console.log(
  `${exported.url}: ${exported.rowCount} rows in ${exported.timeMs}ms`
);

出口商不取证书:DuckDB使用其范围与url相匹配的秘密(见 管理机密) (中文(简体) ). 容器或桶必须存在,并替换一个已存在的物体。 所需的 DuckDB 扩展名(azure 或者说 httpfs)被安装并加载在第一个导出上,这可能需要网络访问. 在 Azure 上, 失败的导出不会留下局部的blob .

导出为响应

toResponse 返回网络标准 Response 与 Content-Type, (中文(简体) ), Content-Length 财务报告和审定财务报表 Content-Disposition 信头设定 。 从Hono、Next.js 路线处理器、 Bun 或 Deno 也可以返回:

return await exporter.toResponse({
  sql: sql`SELECT id, name FROM user WHERE active = ${true}`,
  format: { type: "csv", header: true },
  filename: "users.csv", // optional, defaults to export.<extension>
  signal: request.signal,
});

与 node:http 或 Express, 把它插入服务器响应中 :

import { Readable } from "node:stream";
import { pipeline } from "node:stream/promises";

const response = await exporter.toResponse({ sql, format });
res.writeHead(response.status, Object.fromEntries(response.headers));
await pipeline(Readable.fromWeb(response.body), res);

首先将查询导出为临时文件, 因此在发送任何字节前失败的查询拒绝 。 当主体读取完毕后, 临时文件被删除, 取消( 客户端断开) , 中止通过 signal 或者失败。 作为一个安全网,一个其身体无人阅读的反应 idleTimeoutMs,在其第一次读取或两次读取之前,失败,其临时文件也被删除.

压缩 CSV 作为压缩文件发送 (application/gzip 或者说 application/zstd, (中文(简体) ), export.csv.gz),无 Content-Encoding。 。 。 .

导出 Kysely 查询

DuckDB上运行的导出,但行可以通过一个 Kysely 查询从另一个数据库(msql, postgres...)中来. 说吧 source 改为 sql:它的行被流到一个临时 DuckDB 表格中,导出,然后不管结果如何,该表格都会被丢出. 和我们一起工作 toFile, (中文(简体) ), toObjectStorage 财务报告和审定财务报表 toResponse。 。 。 .

import { KyselyQueryWithZodSchema } from "@flowblade/sqlduck/kysely";
import * as z from "zod";

const file = await exporter.toFile({
  source: new KyselyQueryWithZodSchema({
    query: db.selectFrom("user").select(["id", "name"]),
    // Creates the temporary table, must match the selected columns
    schema: z.strictObject({ id: z.int32(), name: z.string() }),
  }),
  format: { type: "parquet" },
  file: "/data/exports/users.parquet",
  // Optional: where the temporary table is created
  materialization: { database: "scratch" }, // e.g. ATTACH 'scratch.duckdb' AS scratch
});

mapQuery 财务报告和审定财务报表 selectColumns 适用于已实现行: mapQuery 收到一个 SELECT 页:1 各行一次通过JavaScript,从源数据库到DuckDB;到CSV,Parquet或Excel的写入会留在DuckDB内部. materialization.database 财务报告和审定财务报表 materialization.schema 必须是有效的、非保留的 DuckDB 标识符,而且 计划必须存在.

格式

这个 format 选项是受歧视的结合:

Excel使用DuckDB的Excel扩展,安装并加载在第一个Excel导出上,这可能需要网络访问. 出口商本身要求 DuckdbDatasource;通过 Kysely 输出查询其他数据库 source。 。 。 .

映射导出查询

DuckDB内部有两种选择,每种方法都有:

• mapQuery 重写查询,一般通过将查询包入过滤、排序或添加计算出列来进行.
• selectColumns 选择、重新排序和重命名导出列。 密钥是查询的列名( 以 mapQuery页:1 列按照按键的顺序排列.

const file = await exporter.toFile({
  sql: sql`SELECT id, name, password FROM user`,
  format: { type: "csv", header: true },
  file: "/data/exports/users.csv",
  mapQuery: (query) =>
    sql`SELECT *, upper(name) AS name_uc FROM (${query}) ORDER BY id`,
  selectColumns: { id: "User ID", name_uc: "Name" },
});

日志

QueryExporter 日志为 日志 下划线 ["flowblade", "sqlduck"] 类别,或与 logger 给构造者:

管理机密

DuckSecretsManager 创建 DuckDB 秘密,允许访问对象存储(Azure,S3,GCS,R2)或HTTP端点. 它跟随其他管理器:它是由与logtape的连接和日志创建的.

import { DuckSecretsManager } from "@flowblade/sqlduck";

const secrets = new DuckSecretsManager(conn);

await secrets.create({
  name: "exports_azure",
  type: "azure",
  connectionString: process.env.AZURE_STORAGE_CONNECTION_STRING,
  scope: "az://exports/",
});

// Without keys: managed identity, cli, env...
await secrets.create({
  name: "lake_azure",
  type: "azure",
  provider: "credential_chain",
  accountName: "mylake",
});

• 秘密值被绑定为查询参数,因此它们从未出现在已登录的语句或出错中.
• 秘密默认是暂时的:他们一生生活在DuckDB实例的记忆中. persistent: true 在 DuckDB 的秘密目录中保存它们.
• 秘密属于 DuckDB 实例: 它的所有关系都能看到它们,包括 a withSecret 在函数运行时隐藏。 使用 scope 以限制一个秘密所适用的路径.
• azure, (中文(简体) ), s3, (中文(简体) ), r2, (中文(简体) ), gcs 财务报告和审定财务报表 http 秘密被输入。 其他选项进入 options (骆驼案件密钥),其他类型 { type: "custom", customType }。 。 。 .
• 创建扩展类型的秘密( N)azure, (中文(简体) ), s3...)使 DuckDB 安装并加载扩展,这可能需要网络第一次访问.

查询已实现的 Kysely 查询

⚠️ 实验:下面的API可能会改变一个小的发布.

withMaterializedKyselyQuery 将 Kysely 查询的行( 例如从 mssql 或 postgres) 流入一个临时 DuckDB 表格, 将您的查询运行在它上, 然后丢弃表格, 无论查询成功、 失败或丢弃 .

import { DuckDBInstance } from "@duckdb/node-api";
import {
  createKyselyMssqlDialect,
  TediousConnUtils,
} from "@flowblade/source-kysely";
import { sql } from "@flowblade/sql-tag";
import {
  KyselyQueryWithZodSchema,
  withMaterializedKyselyQuery,
} from "@flowblade/sqlduck/kysely";
import { Kysely } from "kysely";
import * as z from "zod";

// Source database: any Kysely instance (here mssql)
type DB = {
  user: { id: number; name: string };
};
const db = new Kysely<DB>({
  dialect: createKyselyMssqlDialect({
    tediousConfig: TediousConnUtils.fromJdbcDsn(
      "sqlserver://localhost:1433;database=db;user=sa;password=pwd"
    ),
  }),
});

// Target duckdb connection (a DuckDBConnection or a DuckdbDatasource)
const instance = await DuckDBInstance.create();
const conn = await instance.connect();

const table = new KyselyQueryWithZodSchema({
  query: db.selectFrom("user").select(["id", "name"]),
  // used to create the temporary table, must match the selected columns
  schema: z.strictObject({ id: z.int32(), name: z.string() }),
});

const result = await withMaterializedKyselyQuery({
  duckConn: conn,
  table,
  // 👇Optional: where the temporary table is created.
  // Defaults to the connection's current database and schema.
  database: "scratch", // e.g. ATTACH 'scratch.duckdb' AS scratch
  schema: "staging", // must already exist
  query: ({ dsDuck, table }) =>
    dsDuck.query(
      sql<{ id: number; name: string }>`
                SELECT id, name FROM ${sql.raw(table.getFullName())} WHERE id < 1000
            `
    ),
});

if (result.error) {
  // materialization or query error
} else {
  console.log(result.data);
  // the first span holds the materialization info (ddl, affectedRows, timeMs, tableName)
  console.log(result.meta.getSpansByType("materialization"));
}

database 财务报告和审定财务报表 schema 必须是有效的、非保留的 DuckDB 标识符,否则返回出错结果,查询不运行。 以所附文件数据库为目标,可使大型材料化不包含在内.

交易

斯克尔达克参加 DuckdbDatasource 同一连接的交易(参见 @ flowblade/ source-duckdb (英语):

• 交易等待运行的负载( e)toTable()或管理者查询, 他们等待一个运行的交易,
• 在一次交易召回中,他们在交易中运行:失败的负载会阻止承诺.

const ds = new DuckdbDatasource({ connection: conn });
await ds.transactionOrThrow(async (tx) => {
  await tx.queryOrThrow(sql`DELETE FROM product`);
  // rolled back with the delete if anything fails
  await sqlDuck.toTable({ table, schema, rowStream, autoCheckpoint: false });
});

说明: DuckDB在交易中无法检查,在交易回调中运行的负载的检查点被跳过( DuckDB) :autoCheckpoint, (中文(简体) ), checkpointChunksFrequency,则 checkpoint 选项 DuckTableLoader) (中文(简体) ). 使用 isInConnectionTransaction(conn) 从 @flowblade/source-duckdb 在你的代码里做同样的事情.

兼容性

基准

节点 24.21

 RUN  v5.0.1 /home/sebastien/github/flowblade/packages/sqlduck

 ✓  bench  bench/appender.bench.ts (1 test) 7533ms
   ✓ appender benches 7530ms
     name                                                             hz     min     max    mean     p75     p99    p995    p999      rme  samples
     duckdb appender, count: 100000, chunk size 1024              5.1751  180.11  231.60  194.56  192.86  229.70  230.65  231.41  ±10.12%        6   fastest
     duckdb appender memory, count: 100000, chunk size 2048       5.0295  172.58  244.98  202.55  226.73  244.25  244.62  244.91  ±19.34%        5
     duckdb appender file no wal, count: 100000, chunk size 1024  4.2547  210.67  317.28  240.39  237.64  314.09  315.69  316.96  ±22.80%        5
     duckdb appender file no wal, count: 100000, chunk size 2048  3.8371  200.49  323.42  268.76  295.48  322.31  322.87  323.31  ±30.51%        4
     duckdb appender file, count: 100000, chunk size 2048         3.7207  257.80  287.20  269.21  272.99  286.63  286.91  287.14   ±7.53%        4   slowest
 ✓  bench  bench/stream.bench.ts (1 test) 3986ms
   ✓ Bench rowsToColumnsChunks 3985ms
     name                                                                         hz      min      max     mean      p75      p99     p995     p999      rme  samples
     rowToColumnsChunk with chunkSize 2048 (count: 100000)                   16.4982  53.5190  78.3487  61.5746  70.6733  78.0419  78.1953  78.3180   ±7.00%       17   fastest
     rowToColumnsChunk with transformer with chunkSize 2048 (count: 100000)  13.4073  63.2078   105.13  76.1133  81.6002   102.92   104.03   104.91   ±8.98%       14
     mapFakeRowStream with chunkSize 2048 (count: 100000)                    10.3533  82.5579   120.58  98.5407   113.23   120.25   120.42   120.54  ±10.12%       11   slowest
 ✓  bench  bench/table-create.bench.ts (1 test) 1286ms
   ✓ Bench getTableCreateFromZod 1284ms
     name                          hz     min      max    mean     p75     p99    p995    p999     rme  samples
     getTableCreateFromZod  20,993.53  0.0343  10.9287  0.0597  0.0619  0.2081  0.3518  0.6929  ±4.00%    16755

1.4.2号舱

 RUN  v5.0.1 /home/sebastien/github/flowblade/packages/sqlduck

 ✓  bench  bench/appender.bench.ts (1 test) 7353ms
   ✓ appender benches 7351ms
     name                                                             hz     min     max    mean     p75     p99    p995    p999      rme  samples
     duckdb appender memory, count: 100000, chunk size 2048       6.5010  141.93  167.82  154.19  158.35  167.29  167.55  167.77   ±4.92%        7   fastest
     duckdb appender, count: 100000, chunk size 1024              5.8972  158.33  185.98  170.04  174.01  185.43  185.70  185.92   ±6.09%        6
     duckdb appender file no wal, count: 100000, chunk size 2048  5.1876  181.29  250.19  195.34  189.16  247.18  248.69  249.89  ±14.55%        6
     duckdb appender file no wal, count: 100000, chunk size 1024  4.8410  187.20  264.48  209.80  208.05  262.22  263.35  264.26  ±18.63%        5
     duckdb appender file, count: 100000, chunk size 2048         4.1319  229.30  261.16  242.64  253.36  260.85  261.00  261.13   ±7.09%        5   slowest
 ✓  bench  bench/stream.bench.ts (1 test) 3933ms
   ✓ Bench rowsToColumnsChunks 3931ms
     name                                                                         hz      min      max     mean      p75      p99     p995     p999     rme  samples
     rowToColumnsChunk with chunkSize 2048 (count: 100000)                   24.5296  34.3957  51.2149  41.1676  43.4825  50.3109  50.7629  51.1245  ±4.18%       25   fastest
     rowToColumnsChunk with transformer with chunkSize 2048 (count: 100000)  23.5864  36.3807  57.7801  42.8068  43.2094  56.3888  57.0845  57.6410  ±4.56%       24
     mapFakeRowStream with chunkSize 2048 (count: 100000)                    21.5954  42.7406  51.6524  46.4448  48.1176  51.5462  51.5993  51.6418  ±2.52%       22   slowest
 ✓  bench  bench/table-create.bench.ts (1 test) 1310ms
   ✓ Bench getTableCreateFromZod 1308ms
     name                          hz     min      max    mean     p75     p99    p995    p999     rme  samples
     getTableCreateFromZod  37,765.77  0.0185  10.1286  0.0360  0.0346  0.1235  0.1771  2.3683  ±4.27%    27781

本地脚本