Spaces:
Running
Running
File size: 4,920 Bytes
aa89cb5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | import { UserContext } from '../src/core/context';
import { OrderBy, SelectQuery } from '../src/core/ast';
import { EntitySchema, SQLExecutionEvidenceStore } from '../src/sql/core';
import { QueryRequest } from '../src/core/request-intent';
import { SQLiteTeaQLClient } from '../src/sql/sqlite';
import { PostgreSQLDriver } from '../src/sql/postgres';
import { MySQLDriver } from '../src/sql/mysql';
const schemas: Record<string, EntitySchema> = {
Order: {
table: 'orders',
columns: {
id: { columnName: 'id', logicalType: 'integer', decode: 'string' },
version: { columnName: 'version', logicalType: 'integer', decode: 'number' },
},
},
};
describe('SQLite true streaming query', () => {
it.each([true, false])('preserves a native named-count owner on a real cursor, logging=%p', async logging => {
const client = new SQLiteTeaQLClient(':memory:', schemas).setDiagnosticSQLLogSink(undefined);
try {
await new UserContext().insertResource('dataService', client).ensureSchema();
for (const id of ['1', '2']) await client.executeMutation({entity:'Order', action:'Create', id,
payload:{}, comment:'seed native cursor fixture'});
const evidence = new SQLExecutionEvidenceStore(), diagnostics: unknown[] = [];
client.setRuntimeTelemetrySink(evidence).setDiagnosticSQLLogSink({write: entry => diagnostics.push(entry)})
.setQueryLoggingEnabled(logging);
const root = new QueryRequest(new SelectQuery('Order').comment('load order totals').purpose('render totals'));
const counted = root.deriveOperation(root.query.forExactCount('totalOrders'),
{kind:'named-count',method:'countAs',alias:'totalOrders'});
const chunks: any[][] = [];
for await (const chunk of client.executeForStream(counted.query, 1)) chunks.push(chunk);
expect(chunks.flat().map(row=>row.totalOrders)).toEqual([2]);
const entries = evidence.snapshot();
expect(entries).toHaveLength(1); expect(diagnostics).toHaveLength(logging ? 1 : 0);
expect(entries[0].traceChain?.map(node=>node.detail)).toEqual(['render totals','load order totals',"countAs('totalOrders')"]);
expect(entries[0].executionOutcome).toBe('success');
expect((entries[0].tracePath ?? []).some(node=>node.kind==='relation')).toBe(false);
expect(root.traceChain).toHaveLength(2);
} finally { await client.close(); }
});
it('yields bounded chunks from the statement iterator', async () => {
const client = new SQLiteTeaQLClient(':memory:', schemas);
await new UserContext().insertResource('dataService', client).ensureSchema();
for (let id = 1; id <= 5; id++) {
await client.executeMutation({
entity: 'Order', action: 'Create', id: String(id), payload: {}, comment: 'seed stream fixture',
});
}
const query = new SelectQuery('Order')
.comment('stream orders')
.purpose('verify cursor streaming')
.order(OrderBy.asc('id'));
const chunks: any[][] = [];
for await (const chunk of client.executeForStream(query, 2)) chunks.push(chunk);
expect(chunks.map(chunk => chunk.length)).toEqual([2, 2, 1]);
expect(chunks.flat().map(row => row.id)).toEqual(['1', '2', '3', '4', '5']);
await client.close();
});
it('rejects missing governance and invalid chunk sizes at iteration time', async () => {
const client = new SQLiteTeaQLClient(':memory:', schemas);
await new UserContext().insertResource('dataService', client).ensureSchema();
const consume = async (stream: AsyncIterable<any[]>) => {
for await (const _chunk of stream) { /* consume */ }
};
await expect(consume(client.executeForStream(new SelectQuery('Order'), 10)))
.rejects.toMatchObject({ code: 'REQUEST_COMMENT_REQUIRED', field: 'comment', requestKind: 'query' });
const query = new SelectQuery('Order').comment('invalid size').purpose('verify validation');
await expect(consume(client.executeForStream(query, 0))).rejects.toThrow(/positive integer/);
await client.close();
});
});
describe('real SQL driver streams', () => {
for (const [name, envName, createDriver, sql] of [
['PostgreSQL', 'TEAQL_TEST_POSTGRES_URL', (url: string) => new PostgreSQLDriver(url),
'SELECT id FROM (VALUES (1), (2), (3), (4), (5)) AS fixture(id) ORDER BY id'],
['MySQL', 'TEAQL_TEST_MYSQL_URL', (url: string) => new MySQLDriver(url),
'SELECT id FROM (SELECT 1 id UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL SELECT 4 UNION ALL SELECT 5) fixture ORDER BY id'],
] as const) (process.env[envName] ? it : it.skip)(`streams rows from ${name}`, async () => {
const url = process.env[envName];
const driver = createDriver(url!);
const ids: number[] = [];
try {
for await (const row of driver.stream(sql)) ids.push(Number(row.id));
} finally {
await driver.close();
}
expect(ids).toEqual([1, 2, 3, 4, 5]);
});
});
|