refactor(integrations): simplify Snowflake safeguards

This commit is contained in:
Bill Leoutsakos
2026-08-08 09:58:10 -07:00
parent b6405b0480
commit 624b7373f5
9 changed files with 45 additions and 173 deletions
@@ -145,7 +145,7 @@ Cancel a running Snowflake SQL API statement.
### Snowflake Insert Rows
Insert up to 1000 structured JSON rows using bound values.
Insert structured JSON rows using bound values.
#### Input
+9 -16
View File
@@ -2,7 +2,6 @@ import { SnowflakeIcon } from '@/components/icons'
import type { BlockConfig, BlockMeta } from '@/blocks/types'
import { AuthMode, IntegrationType } from '@/blocks/types'
import type { SnowflakeStatementResponse } from '@/tools/snowflake/types'
import { addSnowflakeRequestBytes } from '@/tools/snowflake/utils'
const statementOperations = [
'execute_sql',
@@ -41,10 +40,9 @@ const contextOnlyOperations = [
const dataOperations = ['insert_rows', 'update_rows', 'upsert_rows', 'delete_rows', 'load_data']
const taskDefinitionOperations = ['list_tasks', 'get_task', 'run_task']
function parseJson(value: unknown, label: string, budget: { bytes: number }): unknown {
function parseJson(value: unknown, label: string): unknown {
if (value === undefined || value === null || value === '') return undefined
if (typeof value !== 'string') return value
budget.bytes = addSnowflakeRequestBytes(budget.bytes, value)
try {
return JSON.parse(value)
} catch {
@@ -236,7 +234,7 @@ export const SnowflakeBlock: BlockConfig<SnowflakeStatementResponse> = {
wandConfig: {
enabled: true,
prompt:
'Generate a non-empty JSON array of flat row objects. Every row must have the same keys and the batch must contain at most 1000 rows. Return ONLY the JSON array - no explanations, no extra text.',
'Generate a non-empty JSON array of flat row objects. Every row must have the same keys. Use Load Data instead for bulk ingestion from staged files. Return ONLY the JSON array - no explanations, no extra text.',
placeholder: 'Describe the records to write...',
},
},
@@ -598,7 +596,6 @@ export const SnowflakeBlock: BlockConfig<SnowflakeStatementResponse> = {
config: {
tool: (params) => `snowflake_${params.operation}`,
params: (params) => {
const jsonBudget = { bytes: 0 }
const statementParams = () => ({
timeout: optionalNumber(params.timeout),
maxRows: optionalNumber(params.maxRows),
@@ -619,7 +616,7 @@ export const SnowflakeBlock: BlockConfig<SnowflakeStatementResponse> = {
return {
...contextParams(),
async: optionalBoolean(params.async),
bindings: parseJson(params.bindings, 'Bindings', jsonBudget),
bindings: parseJson(params.bindings, 'Bindings'),
}
case 'get_statement':
return {
@@ -629,19 +626,19 @@ export const SnowflakeBlock: BlockConfig<SnowflakeStatementResponse> = {
case 'insert_rows':
return {
...objectParams(),
rows: parseJson(params.rows, 'Rows', jsonBudget),
rows: parseJson(params.rows, 'Rows'),
}
case 'update_rows':
case 'upsert_rows':
return {
...objectParams(),
rows: parseJson(params.rows, 'Rows', jsonBudget),
matchColumns: parseJson(params.matchColumns, 'Match columns', jsonBudget),
rows: parseJson(params.rows, 'Rows'),
matchColumns: parseJson(params.matchColumns, 'Match columns'),
}
case 'delete_rows':
return {
...objectParams(),
filters: parseJson(params.filters, 'Filters', jsonBudget),
filters: parseJson(params.filters, 'Filters'),
}
case 'load_data':
return {
@@ -689,11 +686,7 @@ export const SnowflakeBlock: BlockConfig<SnowflakeStatementResponse> = {
case 'call_procedure':
return {
...objectParams(),
procedureArguments: parseJson(
params.procedureArguments,
'Procedure arguments',
jsonBudget
),
procedureArguments: parseJson(params.procedureArguments, 'Procedure arguments'),
}
default:
return {}
@@ -853,7 +846,7 @@ export const SnowflakeBlockMeta = {
name: 'sync-snowflake-rows',
description: 'Insert, update, or upsert structured records safely in Snowflake.',
content:
'# Synchronize Snowflake Rows\n\n## Steps\n1. Confirm the target table and record keys.\n2. Keep batches at or below 1000 records.\n3. Choose insert, update, or upsert and provide match columns when needed.\n4. Report Snowflake DML statistics.\n\n## Output\nReturn inserted, updated, deleted, and total affected row counts.',
'# Synchronize Snowflake Rows\n\n## Steps\n1. Confirm the target table and record keys.\n2. Keep the structured request within Sim’s request-size limit, and use Load Data for bulk ingestion.\n3. Choose insert, update, or upsert and provide match columns when needed.\n4. Report Snowflake DML statistics.\n\n## Output\nReturn inserted, updated, deleted, and total affected row counts.',
},
{
name: 'load-snowflake-stage',
+1 -1
View File
@@ -18175,7 +18175,7 @@
},
{
"name": "Insert Rows",
"description": "Insert up to 1000 structured JSON rows using bound values."
"description": "Insert structured JSON rows using bound values."
},
{
"name": "Update Rows",
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -18,7 +18,7 @@ export const insertRowsTool: ToolConfig<SnowflakeInsertRowsParams, SnowflakeInse
id: 'snowflake_insert_rows',
version: '1.0.0',
name: 'Snowflake Insert Rows',
description: 'Insert up to 1000 structured JSON rows using bound values.',
description: 'Insert structured JSON rows using bound values.',
params: {
...snowflakeBaseParams,
...snowflakeContextParams,
+21 -43
View File
@@ -23,7 +23,6 @@ import {
normalizeBindings,
qualifiedIdentifier,
} from '@/tools/snowflake/sql'
import { MAX_REQUEST_BYTES, MAX_WRITE_ROWS } from '@/tools/snowflake/utils'
const context = { host: 'acme.snowflakecomputing.com', apiKey: 'secret' }
const table = { ...context, database: 'ANALYTICS', schema: 'PUBLIC', table: 'EVENTS' }
@@ -45,31 +44,25 @@ describe('Snowflake SQL builders', () => {
expect(() => normalizeBindings({ '1': { type: 'NOPE', value: 'x' } } as never)).toThrow(
'Unsupported'
)
expect(() =>
normalizeBindings({ '1': { type: 'TEXT', value: 'x'.repeat(MAX_REQUEST_BYTES) } })
).toThrow('exceeds')
const largeValue = 'x'.repeat(1024 * 1024 + 1)
expect(
normalizeBindings({ '1': { type: 'TEXT', value: largeValue } })?.['1'].value
).toHaveLength(largeValue.length)
expect(SnowflakeBlock.inputs.bindings.description).toContain(
'object keyed by 1-based positions'
)
expect(SnowflakeBlock.inputs.procedureArguments.description).toContain('ordered JSON array')
})
it('rejects oversized JSON block inputs before parsing', () => {
it('parses JSON block inputs above the former Snowflake-specific byte limit', () => {
const mapParams = SnowflakeBlock.tools.config.params
if (!mapParams) throw new Error('Snowflake block must map tool parameters')
expect(() =>
mapParams({
operation: 'insert_rows',
rows: `[{"payload":"${'x'.repeat(MAX_REQUEST_BYTES)}"}]`,
})
).toThrow('exceeds')
expect(() =>
mapParams({
operation: 'update_rows',
rows: `[{"payload":"${'x'.repeat(MAX_REQUEST_BYTES / 2)}"}]`,
matchColumns: `["${'x'.repeat(MAX_REQUEST_BYTES / 2)}"]`,
})
).toThrow('exceeds')
const payload = 'x'.repeat(1024 * 1024 + 1)
const result = mapParams({
operation: 'insert_rows',
rows: `[{"payload":"${payload}"}]`,
}) as { rows: Array<{ payload: string }> }
expect(result.rows[0].payload).toHaveLength(payload.length)
})
it('only coerces fields used by the selected block operation', () => {
@@ -171,23 +164,11 @@ describe('Snowflake SQL builders', () => {
)
})
it('rejects malformed or oversized structured writes', () => {
it('rejects malformed structured writes', () => {
expect(() => buildInsertRows({ ...table, rows: [] })).toThrow('non-empty')
expect(() => buildInsertRows({ ...table, rows: [{ id: 1 }, { other: 2 }] })).toThrow(
'same columns'
)
expect(() =>
buildInsertRows({
...table,
rows: Array.from({ length: MAX_WRITE_ROWS + 1 }, (_, id) => ({ id })),
})
).toThrow('cannot exceed')
expect(() =>
buildInsertRows({
...table,
rows: [{ payload: { value: 'x'.repeat(MAX_REQUEST_BYTES) } }],
})
).toThrow('exceeds')
expect(() => buildInsertRows({ ...table, rows: [{ id: 1 }, { ID: 2 }] })).toThrow(
'same columns'
)
@@ -202,11 +183,17 @@ describe('Snowflake SQL builders', () => {
).toThrow('safe integers')
})
it('builds structured writes above the former 1000-row limit', () => {
const result = buildInsertRows({
...table,
rows: Array.from({ length: 1001 }, (_, id) => ({ id })),
})
expect(Object.keys(result.bindings ?? {})).toHaveLength(1001)
expect(result.statement).toContain('VALUES (?)')
})
it('requires delete filters and binds every filter value', () => {
expect(() => buildDeleteRows({ ...table, filters: {} })).toThrow('cannot be empty')
expect(() =>
buildDeleteRows({ ...table, filters: { payload: 'x'.repeat(MAX_REQUEST_BYTES) } })
).toThrow('exceeds')
expect(buildDeleteRows({ ...table, filters: { id: 7, deleted_at: null } })).toEqual({
statement: 'DELETE FROM ANALYTICS.PUBLIC.EVENTS WHERE id = ? AND deleted_at IS NULL',
bindings: { '1': { type: 'FIXED', value: '7' } },
@@ -380,14 +367,5 @@ describe('Snowflake SQL builders', () => {
procedureArguments: { type: 'TEXT', value: 'x' } as never,
})
).toThrow('JSON array')
expect(() =>
buildCallProcedure({
...context,
database: 'ANALYTICS',
schema: 'PUBLIC',
procedureName: 'REFRESH_MODEL',
procedureArguments: [{ type: 'TEXT', value: 'x'.repeat(MAX_REQUEST_BYTES) }],
})
).toThrow('exceeds')
})
})
+1 -59
View File
@@ -14,13 +14,7 @@ import {
type SnowflakeUpdateRowsParams,
type SnowflakeWarehouseParams,
} from '@/tools/snowflake/types'
import {
addSnowflakeRequestBytes,
addSnowflakeRequestOverhead,
MAX_WRITE_ROWS,
normalizeMaxRows,
type SnowflakeStatementSpec,
} from '@/tools/snowflake/utils'
import { normalizeMaxRows, type SnowflakeStatementSpec } from '@/tools/snowflake/utils'
const UNQUOTED_IDENTIFIER = /^[A-Za-z_][A-Za-z0-9_$]*$/
const QUOTED_IDENTIFIER = /^"(?:[^"]|"")+"$/
@@ -96,7 +90,6 @@ export function normalizeBindings(
throw new Error('bindings must be a JSON object keyed by 1-based positions')
}
const normalized: Record<string, SnowflakeBinding> = {}
let requestBytes = 0
let hasBindings = false
for (const position in input) {
if (!Object.hasOwn(input, position)) continue
@@ -114,8 +107,6 @@ export function normalizeBindings(
if (typeof binding.value !== 'string') {
throw new Error(`binding ${position} value must be a string`)
}
requestBytes = addSnowflakeRequestOverhead(requestBytes, 32)
requestBytes = addSnowflakeRequestBytes(requestBytes, position, binding.type, binding.value)
normalized[position] = { type: binding.type, value: binding.value }
}
return hasBindings ? normalized : undefined
@@ -124,13 +115,10 @@ export function normalizeBindings(
class BindingsBuilder {
readonly bindings: Record<string, SnowflakeBinding> = {}
private position = 0
private requestBytes = 0
private addBinding(type: SnowflakeBinding['type'], value: string): string {
this.position += 1
const key = String(this.position)
this.requestBytes = addSnowflakeRequestOverhead(this.requestBytes, 32)
this.requestBytes = addSnowflakeRequestBytes(this.requestBytes, key, type, value)
this.bindings[key] = { type, value }
return '?'
}
@@ -153,7 +141,6 @@ class BindingsBuilder {
return this.addBinding('TEXT', value)
}
if (Array.isArray(value) || (typeof value === 'object' && value !== null)) {
assertJsonValueWithinRequestBudget(value)
this.addBinding('TEXT', JSON.stringify(value))
return 'PARSE_JSON(?)'
}
@@ -161,51 +148,8 @@ class BindingsBuilder {
}
}
function assertJsonValueWithinRequestBudget(value: unknown): void {
const pending: Array<{ value: unknown; leave?: boolean }> = [{ value }]
const active = new WeakSet<object>()
let requestBytes = 0
while (pending.length > 0) {
const item = pending.pop()
const current = item?.value
if (item?.leave && current && typeof current === 'object') {
active.delete(current)
continue
}
if (typeof current === 'string') {
requestBytes = addSnowflakeRequestOverhead(requestBytes, 4)
requestBytes = addSnowflakeRequestBytes(requestBytes, current)
continue
}
if (!current || typeof current !== 'object') {
requestBytes = addSnowflakeRequestOverhead(requestBytes, 2)
requestBytes = addSnowflakeRequestBytes(requestBytes, String(current))
continue
}
if (active.has(current)) throw new Error('Snowflake row JSON values cannot be circular')
active.add(current)
pending.push({ value: current, leave: true })
requestBytes = addSnowflakeRequestOverhead(requestBytes, 2)
if (Array.isArray(current)) {
for (const nestedValue of current) {
requestBytes = addSnowflakeRequestOverhead(requestBytes, 1)
pending.push({ value: nestedValue })
}
continue
}
for (const key in current) {
if (!Object.hasOwn(current, key)) continue
requestBytes = addSnowflakeRequestOverhead(requestBytes, 4)
requestBytes = addSnowflakeRequestBytes(requestBytes, key)
pending.push({ value: (current as Record<string, unknown>)[key] })
}
}
}
function validateRows(rows: Array<Record<string, unknown>>): string[] {
if (!Array.isArray(rows) || rows.length === 0) throw new Error('rows must be a non-empty array')
if (rows.length > MAX_WRITE_ROWS) throw new Error(`rows cannot exceed ${MAX_WRITE_ROWS} items`)
assertJsonValueWithinRequestBudget(rows)
const columns = Object.keys(rows[0] ?? {})
if (columns.length === 0) throw new Error('rows must contain at least one column')
const signature = [...columns].sort().join('\u0000')
@@ -305,7 +249,6 @@ export function buildDeleteRows(params: SnowflakeDeleteRowsParams): SnowflakeSta
if (!params.filters || Array.isArray(params.filters) || typeof params.filters !== 'object') {
throw new Error('filters must be a JSON object')
}
assertJsonValueWithinRequestBudget(params.filters)
const filters = Object.entries(params.filters)
if (filters.length === 0) throw new Error('filters cannot be empty')
const binds = new BindingsBuilder()
@@ -503,7 +446,6 @@ export function buildCallProcedure(params: SnowflakeCallProcedureParams): Snowfl
if (!Array.isArray(procedureArguments)) {
throw new Error('procedureArguments must be a JSON array')
}
assertJsonValueWithinRequestBudget(procedureArguments)
const bindings: Record<string, SnowflakeBinding> = {}
const placeholders = procedureArguments.map((argument, index) => {
if (!SNOWFLAKE_BINDING_TYPES.includes(argument.type)) {
+8 -7
View File
@@ -9,7 +9,6 @@ import { SNOWFLAKE_STATEMENT_OUTPUTS } from '@/tools/snowflake/types'
import {
buildSnowflakeStatementBody,
getSnowflakeHeaders,
MAX_REQUEST_BYTES,
MAX_RESPONSE_BYTES,
normalizeMaxRows,
normalizeSnowflakeHost,
@@ -181,19 +180,20 @@ describe('Snowflake SQL API transport', () => {
)
})
it('enforces result row and request byte limits', () => {
it('enforces result row limits without a Snowflake-specific request cap', () => {
expect(normalizeMaxRows()).toBe(1000)
expect(normalizeMaxRows(10_000)).toBe(10_000)
expect(() => normalizeMaxRows(10_001)).toThrow('between 1 and 10000')
expect(() =>
const statement = 'x'.repeat(1024 * 1024 + 1)
expect(
buildSnowflakeStatementBody(
{ host: 'acme.snowflakecomputing.com', apiKey: 'secret' },
{ statement: 'x'.repeat(MAX_REQUEST_BYTES) }
)
).toThrow('exceeds')
{ statement }
).statement
).toBe(statement)
})
it('builds a bounded SQL API request body with execution context and bindings', () => {
it('builds a SQL API request body with execution context and bindings', () => {
expect(
buildSnowflakeStatementBody(
{
@@ -442,6 +442,7 @@ describe('Snowflake SQL API transport', () => {
})
it('rejects HTTP and SQL-level failures', async () => {
expect(MAX_RESPONSE_BYTES).toBe(10 * 1024 * 1024)
await expect(
transformSnowflakeResponse(jsonResponse({ message: 'Forbidden', code: '390100' }, 401))
).rejects.toThrow('Forbidden')
+2 -44
View File
@@ -12,9 +12,7 @@ import type { ToolConfig } from '@/tools/types'
export const DEFAULT_MAX_ROWS = 1_000
export const MAX_RESULT_ROWS = 10_000
export const MAX_WRITE_ROWS = 1_000
export const MAX_REQUEST_BYTES = 1024 * 1024
export const MAX_RESPONSE_BYTES = 32 * 1024 * 1024
export const MAX_RESPONSE_BYTES = 10 * 1024 * 1024
const SNOWFLAKE_HOST_SUFFIXES = ['.snowflakecomputing.com', '.snowflakecomputing.cn']
@@ -38,7 +36,6 @@ interface SnowflakeApiResponse {
sqlState?: string
message?: string
statementHandle?: string
statementStatusUrl?: string
data?: Array<Array<string | null>>
resultSetMetaData?: {
numRows?: number
@@ -200,45 +197,11 @@ function normalizeContextName(value: string): string {
return trimmed
}
export function addSnowflakeRequestBytes(total: number, ...values: string[]): number {
let next = total
for (const value of values) {
const remaining = MAX_REQUEST_BYTES - next
if (remaining < 0 || value.length > remaining) {
throw new Error(`Snowflake request body exceeds ${MAX_REQUEST_BYTES} bytes`)
}
next += new TextEncoder().encode(value).byteLength
if (next > MAX_REQUEST_BYTES) {
throw new Error(`Snowflake request body exceeds ${MAX_REQUEST_BYTES} bytes`)
}
}
return next
}
export function addSnowflakeRequestOverhead(total: number, bytes: number): number {
if (total > MAX_REQUEST_BYTES - bytes) {
throw new Error(`Snowflake request body exceeds ${MAX_REQUEST_BYTES} bytes`)
}
return total + bytes
}
export function buildSnowflakeStatementBody(
params: SnowflakeContextParams,
spec: SnowflakeStatementSpec
): Record<string, unknown> {
if (!/\S/.test(spec.statement)) throw new Error('Snowflake statement is required')
let requestBytes = addSnowflakeRequestBytes(256, spec.statement)
for (const value of [params.warehouse, params.database, params.schema, params.role]) {
if (value !== undefined) requestBytes = addSnowflakeRequestBytes(requestBytes, value)
}
let hasBindings = false
for (const position in spec.bindings) {
if (!Object.hasOwn(spec.bindings, position)) continue
hasBindings = true
const binding = spec.bindings[position]
requestBytes = addSnowflakeRequestOverhead(requestBytes, 32)
requestBytes = addSnowflakeRequestBytes(requestBytes, position, binding.type, binding.value)
}
const body: Record<string, unknown> = {
statement: spec.statement,
@@ -250,12 +213,7 @@ export function buildSnowflakeStatementBody(
if (params.database?.trim()) body.database = normalizeContextName(params.database)
if (params.schema?.trim()) body.schema = normalizeContextName(params.schema)
if (params.role?.trim()) body.role = normalizeContextName(params.role)
if (hasBindings) body.bindings = spec.bindings
const size = new TextEncoder().encode(JSON.stringify(body)).byteLength
if (size > MAX_REQUEST_BYTES) {
throw new Error(`Snowflake request body exceeds ${MAX_REQUEST_BYTES} bytes`)
}
if (spec.bindings && Object.keys(spec.bindings).length > 0) body.bindings = spec.bindings
return body
}