diff --git a/packages/nodes-base/nodes/Databricks/Databricks.node.ts b/packages/nodes-base/nodes/Databricks/Databricks.node.ts index 500cdf83929..cd84b395598 100644 --- a/packages/nodes-base/nodes/Databricks/Databricks.node.ts +++ b/packages/nodes-base/nodes/Databricks/Databricks.node.ts @@ -29,6 +29,7 @@ import { router } from './actions/router'; import { authenticationProperty, databricksCredentials } from './authentication'; import { DATABRICKS_NODE_VERSION } from './constants'; import * as loadOptions from './methods/loadOptions'; +import * as resourceMapping from './methods/resourceMapping'; import * as listSearch from './methods/listSearch'; export class Databricks implements INodeType { @@ -120,7 +121,7 @@ export class Databricks implements INodeType { ], }; - methods = { listSearch, loadOptions }; + methods = { listSearch, loadOptions, resourceMapping }; async execute(this: IExecuteFunctions): Promise { return await router.call(this); diff --git a/packages/nodes-base/nodes/Databricks/actions/lakebase/Lakebase.resource.ts b/packages/nodes-base/nodes/Databricks/actions/lakebase/Lakebase.resource.ts index 7c02c191947..6edd16edcf9 100644 --- a/packages/nodes-base/nodes/Databricks/actions/lakebase/Lakebase.resource.ts +++ b/packages/nodes-base/nodes/Databricks/actions/lakebase/Lakebase.resource.ts @@ -1 +1,2 @@ export * as getAll from './getAll.operation'; +export * as insert from './insert.operation'; diff --git a/packages/nodes-base/nodes/Databricks/actions/lakebase/conditions.ts b/packages/nodes-base/nodes/Databricks/actions/lakebase/conditions.ts index fd9d5730ffc..c83c804296b 100644 --- a/packages/nodes-base/nodes/Databricks/actions/lakebase/conditions.ts +++ b/packages/nodes-base/nodes/Databricks/actions/lakebase/conditions.ts @@ -61,9 +61,18 @@ export function buildLakebaseQuery(input: LakebaseQueryInput): IDataObject { } const column = quotePostgrestComponent(rule.column); - return VALUELESS_OPERATORS.has(condition) - ? `${column}.${condition}` - : `${column}.${condition}.${quotePostgrestComponent(rule.value ?? '')}`; + if (VALUELESS_OPERATORS.has(condition)) return `${column}.${condition}`; + + // An expression resolving to null would read as `col.eq.`, which selects rows + // holding an empty string rather than rows holding nothing. An untouched Value + // field is '', so only an expression or a hand-authored workflow gets here. + if (rule.value === null || rule.value === undefined) { + throw new UserError(`The filter on "${String(rule.column)}" has no value`, { + description: 'Use the Is Null or Is Not Null condition to match rows with no value.', + }); + } + + return `${column}.${condition}.${quotePostgrestComponent(rule.value)}`; }); if (terms.length > 0) { diff --git a/packages/nodes-base/nodes/Databricks/actions/lakebase/getAll.operation.ts b/packages/nodes-base/nodes/Databricks/actions/lakebase/getAll.operation.ts index c00bfc63324..6fcaf4fd470 100644 --- a/packages/nodes-base/nodes/Databricks/actions/lakebase/getAll.operation.ts +++ b/packages/nodes-base/nodes/Databricks/actions/lakebase/getAll.operation.ts @@ -1,4 +1,5 @@ import type { IDataObject, IExecuteFunctions, INodeExecutionData } from 'n8n-workflow'; +import { NodeOperationError } from 'n8n-workflow'; import { LAKEBASE_MAX_PAGES, LAKEBASE_PAGE_SIZE } from '../../constants'; import { lakebaseApiRequest } from '../../transport'; @@ -12,28 +13,21 @@ import { resolveLakebaseSchemaUrl, resolveLakebaseTableUrl } from './helpers'; import { fetchLakebaseColumns } from './schema'; /** - * Offset paging needs a stable order, otherwise a row can arrive twice or not at - * all. When the user set no sort rule, order by the primary key instead. Best - * effort: a table without one, or a project that does not serve the schema - * document, still pages. + * Offset paging needs a total order, otherwise a row can arrive twice or not at + * all. A user sort is not necessarily unique, so the primary key goes on the end + * as a tie-breaker rather than only when no sort was set. Best effort: a table + * without one, or a project that does not serve the schema document, still pages. */ -async function stableOrderFallback( - context: IExecuteFunctions, - i: number, -): Promise { +async function primaryKeyColumns(context: IExecuteFunctions, i: number): Promise { try { const schemaUrl = await resolveLakebaseSchemaUrl(context, i); const table = context.getNodeParameter('lakebaseTable', i, '', { extractValue: true, }) as string; - const primaryKeys = (await fetchLakebaseColumns(context, schemaUrl, table)).filter( - (column) => column.isPrimaryKey, - ); - if (primaryKeys.length === 0) return undefined; - - return primaryKeys.map((column) => `${quotePostgrestComponent(column.name)}.asc`).join(','); + const columns = await fetchLakebaseColumns(context, schemaUrl, table); + return columns.filter((column) => column.isPrimaryKey).map((column) => column.name); } catch { - return undefined; + return []; } } @@ -50,9 +44,16 @@ export async function execute(this: IExecuteFunctions, i: number): Promise LAKEBASE_PAGE_SIZE)) { - const order = await stableOrderFallback(this, i); - if (order) qs.order = order; + if (returnAll || limit > LAKEBASE_PAGE_SIZE) { + const alreadySorted = new Set( + sort.filter((rule) => rule.column).map((rule) => String(rule.column)), + ); + const tieBreakers = (await primaryKeyColumns(this, i)) + .filter((name) => !alreadySorted.has(name)) + .map((name) => `${quotePostgrestComponent(name)}.asc`); + if (tieBreakers.length > 0) { + qs.order = [qs.order, ...tieBreakers].filter(Boolean).join(','); + } } const rows: IDataObject[] = []; @@ -61,12 +62,16 @@ export async function execute(this: IExecuteFunctions, i: number): Promise= requested) apiHasFilledAPage = true; - else if (apiHasFilledAPage) break; + else if (apiHasFilledAPage) { + reachedTheEnd = true; + break; + } - if (!returnAll && rows.length >= limit) break; + if (!returnAll && rows.length >= limit) { + reachedTheEnd = true; + break; + } + } + + // Handing back part of the table as though it were all of it would be read as the + // table being that size, so stop instead of truncating in silence. + if (!reachedTheEnd) { + throw new NodeOperationError( + this.getNode(), + `The table holds more rows than one run can read (${LAKEBASE_MAX_PAGES * LAKEBASE_PAGE_SIZE})`, + { + itemIndex: i, + description: + 'Narrow the result with a condition under Select Rows, or turn off Return All and set a Limit.', + }, + ); } return rows.map((row) => ({ json: row, pairedItem: { item: i } })); diff --git a/packages/nodes-base/nodes/Databricks/actions/lakebase/helpers.ts b/packages/nodes-base/nodes/Databricks/actions/lakebase/helpers.ts index e3b81b75b25..e5683191434 100644 --- a/packages/nodes-base/nodes/Databricks/actions/lakebase/helpers.ts +++ b/packages/nodes-base/nodes/Databricks/actions/lakebase/helpers.ts @@ -20,6 +20,26 @@ export function getSelectedLakebaseTarget(context: ILoadOptionsFunctions) { }; } +/** + * Resolves the REST URL of the schema the editor currently has selected, or null + * when the user has not finished choosing a table. Shared by the column dropdown + * and the column mapper, which differ only in how they render and how they report + * a missing schema document. + */ +export async function resolveSelectedLakebaseSchema( + context: ILoadOptionsFunctions, +): Promise<{ schemaUrl: string; table: string } | null> { + const { project, branch, database, schema, table } = getSelectedLakebaseTarget(context); + // An unset locator means the user has not finished choosing a table yet + if (!project || !branch || !database || !schema || !table) return null; + + const base = await resolveLakebaseRestBase(context, project, branch); + return { + schemaUrl: `${base}/${encodeURIComponent(database)}/${encodeURIComponent(schema)}`, + table, + }; +} + function readLocator( context: IExecuteFunctions, itemIndex: number, diff --git a/packages/nodes-base/nodes/Databricks/actions/lakebase/insert.operation.ts b/packages/nodes-base/nodes/Databricks/actions/lakebase/insert.operation.ts new file mode 100644 index 00000000000..7dbb014fef2 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/actions/lakebase/insert.operation.ts @@ -0,0 +1,46 @@ +import type { IDataObject, IExecuteFunctions, INodeExecutionData } from 'n8n-workflow'; + +import { lakebaseApiRequest } from '../../transport'; +import { resolveLakebaseTableUrl } from './helpers'; + +/** + * The Data API rejects the whole row for a field the table does not have, so an + * item carrying anything extra from an earlier node would never insert. + */ +function columnsPresentInTable(context: IExecuteFunctions, i: number): IDataObject { + const item = context.getInputData()[i].json; + // getNodeParameter returns the fallback only for a missing parameter. A workflow + // written outside the editor can save null, or a value that is not an array. + const schema = context.getNodeParameter('columns.schema', i, []) as Array<{ id?: string }>; + const known = new Set(Array.isArray(schema) ? schema.map((field) => field.id) : []); + if (known.size === 0) return item; + + return Object.fromEntries(Object.entries(item).filter(([column]) => known.has(column))); +} + +export async function execute(this: IExecuteFunctions, i: number): Promise { + const url = await resolveLakebaseTableUrl(this, i); + + const mappingMode = this.getNodeParameter('columns.mappingMode', i) as string; + const row = + mappingMode === 'autoMapInputData' + ? columnsPresentInTable(this, i) + : // Core validates and coerces the mapped values only for a path ending in `value`. + // An all-defaults row leaves that value null rather than absent. + ((this.getNodeParameter('columns.value', i, {}) as IDataObject | null) ?? {}); + + const response = await lakebaseApiRequest(this, { + method: 'POST', + url, + body: row, + json: true, + headers: { Accept: 'application/json', Prefer: 'return=representation' }, + }); + + // `return=representation` echoes the created rows, including values the + // database filled in. A project that does not honour it answers empty. + const created = Array.isArray(response) ? (response as IDataObject[]) : []; + if (created.length === 0) return [{ json: { success: true }, pairedItem: { item: i } }]; + + return created.map((created_row) => ({ json: created_row, pairedItem: { item: i } })); +} diff --git a/packages/nodes-base/nodes/Databricks/actions/lakebase/openApiDocument.ts b/packages/nodes-base/nodes/Databricks/actions/lakebase/openApiDocument.ts new file mode 100644 index 00000000000..5849a0e64a9 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/actions/lakebase/openApiDocument.ts @@ -0,0 +1,17 @@ +import { isRecord } from '@n8n/utils/is-record'; +import { NodeApiError } from 'n8n-workflow'; + +/** The Data API serves the schema document here, so a missing-table code on this path means something else */ +export const OPENAPI_PATH = /\/openapi\.json$/; + +/** The Databricks UI path, so every message that names the setting stays in step */ +export const OPENAPI_SETTING_PATH = 'Data API > API > Advanced settings > OpenAPI specification'; + +export const OPENAPI_DISABLED_NOTICE = `Turn on ${OPENAPI_SETTING_PATH} for this Lakebase project, so the node can list its columns.`; + +export function isOpenApiUnavailable(error: unknown, url: string): boolean { + if (!OPENAPI_PATH.test(url)) return false; + + const body = error instanceof NodeApiError ? error.context.data : undefined; + return isRecord(body) && body.code === 'PGRST205'; +} diff --git a/packages/nodes-base/nodes/Databricks/methods/loadOptions.ts b/packages/nodes-base/nodes/Databricks/methods/loadOptions.ts index dd49edca16b..f540c11e7bd 100644 --- a/packages/nodes-base/nodes/Databricks/methods/loadOptions.ts +++ b/packages/nodes-base/nodes/Databricks/methods/loadOptions.ts @@ -1,22 +1,15 @@ import type { ILoadOptionsFunctions, INodePropertyOptions } from 'n8n-workflow'; -import { getSelectedLakebaseTarget } from '../actions/lakebase/helpers'; +import { resolveSelectedLakebaseSchema } from '../actions/lakebase/helpers'; import { fetchLakebaseColumns } from '../actions/lakebase/schema'; -import { explainLakebaseSchemaError, resolveLakebaseRestBase } from '../transport'; +import { explainLakebaseSchemaError } from '../transport'; async function listColumns(context: ILoadOptionsFunctions): Promise { - const { project, branch, database, schema, table } = getSelectedLakebaseTarget(context); - // An unset locator means the user has not finished choosing a table yet - if (!project || !branch || !database || !schema || !table) return []; - try { - const base = await resolveLakebaseRestBase(context, project, branch); - const columns = await fetchLakebaseColumns( - context, - `${base}/${encodeURIComponent(database)}/${encodeURIComponent(schema)}`, - table, - ); + const target = await resolveSelectedLakebaseSchema(context); + if (!target) return []; + const columns = await fetchLakebaseColumns(context, target.schemaUrl, target.table); return columns.map((column) => ({ name: column.name, value: column.name })); } catch (error) { explainLakebaseSchemaError( diff --git a/packages/nodes-base/nodes/Databricks/methods/resourceMapping.ts b/packages/nodes-base/nodes/Databricks/methods/resourceMapping.ts new file mode 100644 index 00000000000..0d2afba2d56 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/methods/resourceMapping.ts @@ -0,0 +1,66 @@ +import type { FieldType, ILoadOptionsFunctions, ResourceMapperFields } from 'n8n-workflow'; + +import { resolveSelectedLakebaseSchema } from '../actions/lakebase/helpers'; +import { isOpenApiUnavailable, OPENAPI_DISABLED_NOTICE } from '../actions/lakebase/openApiDocument'; +import { fetchLakebaseColumns, type LakebaseColumn } from '../actions/lakebase/schema'; + +const TIME_FORMATS = new Set(['time', 'time without time zone', 'time with time zone']); + +const DATE_FORMATS = new Set([ + 'date', + 'date-time', + 'timestamp with time zone', + 'timestamp without time zone', +]); + +function fieldTypeOf(column: LakebaseColumn): FieldType { + if (column.enum) return 'options'; + if (column.format && DATE_FORMATS.has(column.format)) return 'dateTime'; + if (column.format && TIME_FORMATS.has(column.format)) return 'time'; + + switch (column.type) { + case 'integer': + case 'number': + return 'number'; + case 'boolean': + return 'boolean'; + case 'array': + return 'array'; + case 'object': + return 'object'; + default: + return 'string'; + } +} + +/** Lists a table's columns for the Columns form. */ +export async function getLakebaseMappingColumns( + this: ILoadOptionsFunctions, +): Promise { + const target = await resolveSelectedLakebaseSchema(this); + if (!target) return { fields: [] }; + + let columns: LakebaseColumn[]; + try { + columns = await fetchLakebaseColumns(this, target.schemaUrl, target.table); + } catch (error) { + if (!isOpenApiUnavailable(error, `${target.schemaUrl}/openapi.json`)) throw error; + return { fields: [], emptyFieldsNotice: OPENAPI_DISABLED_NOTICE }; + } + + return { + fields: columns.map((column) => ({ + id: column.name, + displayName: column.name, + // The document marks every NOT NULL column required, including ones the + // database fills in, so asking for those would demand a value twice. + required: column.isRequired && !column.hasDefault && !column.isReadOnly, + display: true, + defaultMatch: false, + canBeUsedToMatch: false, + readOnly: column.isReadOnly, + type: fieldTypeOf(column), + options: column.enum?.map((value) => ({ name: String(value), value: value as string })), + })), + }; +} diff --git a/packages/nodes-base/nodes/Databricks/resources/lakebase/operations.ts b/packages/nodes-base/nodes/Databricks/resources/lakebase/operations.ts index 00efc219f3f..0e139f68f6d 100644 --- a/packages/nodes-base/nodes/Databricks/resources/lakebase/operations.ts +++ b/packages/nodes-base/nodes/Databricks/resources/lakebase/operations.ts @@ -17,6 +17,12 @@ export const lakebaseOperations: INodeProperties = { description: 'Get many rows from a table', action: 'Get many rows', }, + { + name: 'Insert', + value: 'insert', + description: 'Insert a row into a table', + action: 'Insert a row', + }, ], default: 'getAll', }; diff --git a/packages/nodes-base/nodes/Databricks/resources/lakebase/parameters.ts b/packages/nodes-base/nodes/Databricks/resources/lakebase/parameters.ts index c3a619ec2fd..eea048e831f 100644 --- a/packages/nodes-base/nodes/Databricks/resources/lakebase/parameters.ts +++ b/packages/nodes-base/nodes/Databricks/resources/lakebase/parameters.ts @@ -362,4 +362,39 @@ export const lakebaseParameters: INodeProperties[] = [ }, ], }, + { + displayName: 'Columns', + name: 'columns', + type: 'resourceMapper', + noDataExpression: true, + required: true, + default: { + mappingMode: 'defineBelow', + value: null, + }, + typeOptions: { + loadOptionsDependsOn: [ + 'lakebaseProject.value', + 'lakebaseBranch.value', + 'lakebaseDatabase.value', + 'lakebaseSchema.value', + 'lakebaseTable.value', + ], + resourceMapper: { + resourceMapperMethod: 'getLakebaseMappingColumns', + mode: 'add', + fieldWords: { + singular: 'column', + plural: 'columns', + }, + addAllFields: true, + multiKeyMatch: false, + // The saved schema is what core validates against, not a fresh read + refreshStaleSchemaOnOpen: true, + }, + }, + displayOptions: { + show: { resource: ['lakebase'], operation: ['insert'] }, + }, + }, ]; diff --git a/packages/nodes-base/nodes/Databricks/test/lakebaseConditions.test.ts b/packages/nodes-base/nodes/Databricks/test/lakebaseConditions.test.ts index 3aab15a728a..d137ff34830 100644 --- a/packages/nodes-base/nodes/Databricks/test/lakebaseConditions.test.ts +++ b/packages/nodes-base/nodes/Databricks/test/lakebaseConditions.test.ts @@ -64,6 +64,29 @@ describe('Lakebase -> condition translation', () => { ); }); + // `col.eq.` selects rows holding an empty string, so a null that slipped through + // would quietly return the wrong rows rather than the rows with no value. + it.each([ + ['null', null], + ['undefined', undefined], + ])('should reject a filter value of %s', (_name, value) => { + expect(() => query({ where: [{ column: 'sku', condition: 'eq', value }] })).toThrow( + UserError, + ); + }); + + it('should still accept an empty string, which matches an empty value', () => { + expect(query({ where: [{ column: 'sku', condition: 'eq', value: '' }] })).toEqual({ + and: '(sku.eq.)', + }); + }); + + it.each(['is.null', 'not.is.null'])('should allow %s with no value', (condition) => { + expect(query({ where: [{ column: 'sku', condition }] })).toEqual({ + and: `(sku.${condition})`, + }); + }); + it('skips a rule with no column', () => { expect(query({ where: [{ condition: 'eq', value: 'alice' }] })).toEqual({}); }); diff --git a/packages/nodes-base/nodes/Databricks/test/lakebaseGetMany.test.ts b/packages/nodes-base/nodes/Databricks/test/lakebaseGetMany.test.ts index b45653c6d32..e6cb7c40a46 100644 --- a/packages/nodes-base/nodes/Databricks/test/lakebaseGetMany.test.ts +++ b/packages/nodes-base/nodes/Databricks/test/lakebaseGetMany.test.ts @@ -140,10 +140,10 @@ describe('Lakebase -> Get Many', () => { expect(qsOf(context, 1)).toEqual(expect.objectContaining({ offset: capped })); }); - it('stops at the page cap rather than looping forever', async () => { + it('fails rather than return a truncated table as the whole table', async () => { const { context, result } = feedPages([rows(1, LAKEBASE_PAGE_SIZE)], { returnAll: true }); - expect(await result).toHaveLength(LAKEBASE_PAGE_SIZE * LAKEBASE_MAX_PAGES); + await expect(result).rejects.toThrow('more rows than one run can read'); expect(apiMock(context)).toHaveBeenCalledTimes(LAKEBASE_MAX_PAGES); }); @@ -193,7 +193,9 @@ describe('Lakebase -> Get Many', () => { expect(qsOf(context, 1).order).toBeUndefined(); }); - it('keeps the user sort rule instead of the primary key', async () => { + // A user sort need not be unique, so without the key as a tie-breaker two rows + // sharing a sku can swap between offset requests and be sent twice or skipped. + it('adds the primary key after the user sort rule', async () => { vi.mocked(fetchLakebaseColumns).mockResolvedValue([ { name: 'id', hasDefault: true, isRequired: true, isReadOnly: false, isPrimaryKey: true }, ]); @@ -203,8 +205,20 @@ describe('Lakebase -> Get Many', () => { }); await result; - expect(qsOf(context, 1).order).toBe('sku.asc'); - expect(fetchLakebaseColumns).not.toHaveBeenCalled(); + expect(qsOf(context, 1).order).toBe('sku.asc,id.asc'); + }); + + it('does not repeat a primary key the user already sorted on', async () => { + vi.mocked(fetchLakebaseColumns).mockResolvedValue([ + { name: 'id', hasDefault: true, isRequired: true, isReadOnly: false, isPrimaryKey: true }, + ]); + const { context, result } = feedPages([rows(1, 1000), rows(1001, 1)], { + returnAll: true, + 'sort.values': [{ column: 'id', direction: 'desc' }], + }); + await result; + + expect(qsOf(context, 1).order).toBe('id.desc'); }); it('reads the item index it was handed', async () => { diff --git a/packages/nodes-base/nodes/Databricks/test/lakebaseInsert.test.ts b/packages/nodes-base/nodes/Databricks/test/lakebaseInsert.test.ts new file mode 100644 index 00000000000..0e66548d673 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/test/lakebaseInsert.test.ts @@ -0,0 +1,159 @@ +import type { IDataObject, IExecuteFunctions, INode, NodeParameterValueType } from 'n8n-workflow'; +import { mockDeep } from 'vitest-mock-extended'; + +import { execute as insert } from '../actions/lakebase/insert.operation'; + +const TABLE_URL = 'https://host.example/api/2.0/workspace/7/rest/app/public/orders'; + +vi.mock('../actions/lakebase/helpers', () => ({ + resolveLakebaseTableUrl: vi.fn(async () => TABLE_URL), + resolveLakebaseSchemaUrl: vi.fn(async () => TABLE_URL.replace(/\/orders$/, '')), +})); + +const node: INode = { + id: '1', + name: 'Databricks', + type: 'n8n-nodes-base.databricks', + typeVersion: 1, + position: [0, 0], + parameters: {}, +}; + +describe('Lakebase -> Insert', () => { + const setupContext = ( + overrides: Record = {}, + inputJson: IDataObject[] = [{ sku: 'from-input' }], + itemIndex = 0, + ) => { + const parameters: Record = { + authentication: 'oAuth2', + 'columns.mappingMode': 'defineBelow', + 'columns.value': { sku: 'WF-1', price: 10 }, + ...overrides, + }; + const context = mockDeep(); + context.getNode.mockReturnValue(node); + context.getInputData.mockReturnValue(inputJson.map((json) => ({ json }))); + context.getExecutionCancelSignal.mockReturnValue(undefined); + context.getNodeParameter.mockImplementation((name, index, fallback) => { + if (index !== itemIndex && name !== 'authentication') return fallback; + // Core takes the fallback only for a parameter it does not have, so a stored + // null has to arrive as null. + return name in parameters ? parameters[name] : fallback; + }); + context.getCredentials.mockResolvedValue({ host: 'https://host.example' }); + return context; + }; + const apiMock = (context: ReturnType) => + context.helpers.httpRequestWithAuthentication; + const run = ( + response: unknown, + overrides: Record = {}, + inputJson?: IDataObject[], + itemIndex = 0, + ) => { + const context = setupContext(overrides, inputJson, itemIndex); + apiMock(context).mockResolvedValue(response); + return { context, result: insert.call(context, itemIndex) }; + }; + + it('posts the mapped columns to the table', async () => { + const { context, result } = run([{ id: 1, sku: 'WF-1', price: 10 }]); + + expect(await result).toEqual([ + { json: { id: 1, sku: 'WF-1', price: 10 }, pairedItem: { item: 0 } }, + ]); + expect(apiMock(context).mock.calls[0][0]).toBe('databricksOAuth2Api'); + expect(apiMock(context).mock.calls[0][1]).toEqual( + expect.objectContaining({ + method: 'POST', + url: TABLE_URL, + body: { sku: 'WF-1', price: 10 }, + }), + ); + }); + + it('asks the Data API to echo the created row', async () => { + const { context, result } = run([{ id: 1 }]); + await result; + + const options = apiMock(context).mock.calls[0][1] as { headers: Record }; + expect(options.headers.Prefer).toBe('return=representation'); + }); + + it('sends the input item when the mapping mode is automatic', async () => { + const { context, result } = run([{ id: 2 }], { 'columns.mappingMode': 'autoMapInputData' }, [ + { sku: 'from-input' }, + ]); + await result; + + expect(apiMock(context).mock.calls[0][1]).toEqual( + expect.objectContaining({ body: { sku: 'from-input' } }), + ); + }); + + it('drops input fields the table does not have', async () => { + const { context, result } = run( + [{ id: 1 }], + { + 'columns.mappingMode': 'autoMapInputData', + 'columns.schema': [{ id: 'sku' }, { id: 'price' }], + }, + [{ sku: 'WF-1', price: 10, upstreamNoise: 'x' }], + ); + await result; + + expect(apiMock(context).mock.calls[0][1]).toEqual( + expect.objectContaining({ body: { sku: 'WF-1', price: 10 } }), + ); + }); + + it.each<[string, Record]>([ + ['no schema is saved', {}], + ['the saved schema is null', { 'columns.schema': null }], + ['the saved schema is not an array', { 'columns.schema': 'sku,price' }], + ])('sends the whole item when %s', async (_name, overrides) => { + const { context, result } = run( + [{ id: 1 }], + { 'columns.mappingMode': 'autoMapInputData', ...overrides }, + [{ sku: 'WF-1' }], + ); + await result; + + expect(apiMock(context).mock.calls[0][1]).toEqual( + expect.objectContaining({ body: { sku: 'WF-1' } }), + ); + }); + + it('sends an empty row when every column is left to the database', async () => { + const { context, result } = run([{ id: 1 }], { 'columns.value': null }); + await result; + + expect(apiMock(context).mock.calls[0][1]).toEqual(expect.objectContaining({ body: {} })); + }); + + it('returns every row the Data API echoes back', async () => { + const { result } = run([{ id: 1 }, { id: 2 }]); + + expect(await result).toEqual([ + { json: { id: 1 }, pairedItem: { item: 0 } }, + { json: { id: 2 }, pairedItem: { item: 0 } }, + ]); + }); + + it.each([ + ['an empty array', []], + ['no body at all', undefined], + ['something that is not an array', { ok: true }], + ])('reports success when the response is %s', async (_name, response) => { + const { result } = run(response); + + expect(await result).toEqual([{ json: { success: true }, pairedItem: { item: 0 } }]); + }); + + it('pairs the row to the item index it was handed', async () => { + const { result } = run([{ id: 9 }], {}, [{ a: 1 }, { b: 2 }, { c: 3 }], 2); + + expect(await result).toEqual([{ json: { id: 9 }, pairedItem: { item: 2 } }]); + }); +}); diff --git a/packages/nodes-base/nodes/Databricks/test/lakebaseOpenApiDocument.test.ts b/packages/nodes-base/nodes/Databricks/test/lakebaseOpenApiDocument.test.ts new file mode 100644 index 00000000000..746243a7ea0 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/test/lakebaseOpenApiDocument.test.ts @@ -0,0 +1,41 @@ +import type { IDataObject, INode, JsonObject } from 'n8n-workflow'; +import { NodeApiError } from 'n8n-workflow'; + +import { isOpenApiUnavailable } from '../actions/lakebase/openApiDocument'; + +const node: INode = { + id: '1', + name: 'Databricks', + type: 'n8n-nodes-base.databricks', + typeVersion: 1, + position: [0, 0], + parameters: {}, +}; + +const SCHEMA_URL = 'https://host.example/api/2.0/workspace/7/rest/app/public'; +const DOCUMENT_URL = `${SCHEMA_URL}/openapi.json`; + +const errorWith = (data: unknown) => { + const error = new NodeApiError(node, { message: 'failed' } as JsonObject); + error.context.data = data as IDataObject; + return error; +}; + +describe('Lakebase -> schema document availability', () => { + it('recognises a project that serves no schema document', () => { + expect(isOpenApiUnavailable(errorWith({ code: 'PGRST205' }), DOCUMENT_URL)).toBe(true); + }); + + it.each([ + ['a missing table on a table path', { code: 'PGRST205' }, `${SCHEMA_URL}/orders`], + ['a different code on the document path', { code: '42501' }, DOCUMENT_URL], + ['no code at all', { message: 'nope' }, DOCUMENT_URL], + ['a body that is not an object', 'nope', DOCUMENT_URL], + ])('does not claim the document is off for %s', (_name, data, url) => { + expect(isOpenApiUnavailable(errorWith(data), url)).toBe(false); + }); + + it('ignores an error that did not come from the API', () => { + expect(isOpenApiUnavailable(new Error('socket hang up'), DOCUMENT_URL)).toBe(false); + }); +}); diff --git a/packages/nodes-base/nodes/Databricks/test/lakebasePickers.test.ts b/packages/nodes-base/nodes/Databricks/test/lakebasePickers.test.ts index c4b4b5aeb41..ddc2d97f929 100644 --- a/packages/nodes-base/nodes/Databricks/test/lakebasePickers.test.ts +++ b/packages/nodes-base/nodes/Databricks/test/lakebasePickers.test.ts @@ -10,6 +10,7 @@ import type { import { mockDeep } from 'vitest-mock-extended'; import { resolveLakebaseSchemaUrl, resolveLakebaseTableUrl } from '../actions/lakebase/helpers'; +import { OPENAPI_SETTING_PATH } from '../actions/lakebase/openApiDocument'; // The shared Lakebase transport waits before its schema-cache retry vi.mock('@n8n/utils/sleep', () => ({ @@ -501,9 +502,8 @@ describe('listSearch -> getLakebaseTables', () => { expect(error.message).toBe( 'Turn on the "OpenAPI specification" setting of the Data API to list tables, or enter the table name By ID', ); - expect(error.description).toBe( - 'In Databricks open the project, then Data API > API > Advanced settings, and enable OpenAPI specification.', - ); + // Pins that the user is told where the setting lives, not the wording around it + expect(error.description).toContain(OPENAPI_SETTING_PATH); }); it('rethrows other Data API errors untouched', async () => { diff --git a/packages/nodes-base/nodes/Databricks/test/lakebaseResourceMapping.test.ts b/packages/nodes-base/nodes/Databricks/test/lakebaseResourceMapping.test.ts new file mode 100644 index 00000000000..f0b64c8d353 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/test/lakebaseResourceMapping.test.ts @@ -0,0 +1,179 @@ +import type { + ILoadOptionsFunctions, + INode, + JsonObject, + NodeParameterValueType, +} from 'n8n-workflow'; +import { NodeApiError } from 'n8n-workflow'; +import { mockDeep } from 'vitest-mock-extended'; + +import { OPENAPI_DISABLED_NOTICE } from '../actions/lakebase/openApiDocument'; +import { fetchLakebaseColumns } from '../actions/lakebase/schema'; +import type { LakebaseColumn } from '../actions/lakebase/schema'; +import { getLakebaseMappingColumns } from '../methods/resourceMapping'; + +vi.mock('../actions/lakebase/schema', async (importOriginal) => ({ + ...(await importOriginal()), + fetchLakebaseColumns: vi.fn(async () => []), +})); + +// The real resolver reads the branch's compute host from the management API, +// which these tests have no reason to stand up. +vi.mock('../transport', async (importOriginal) => ({ + ...(await importOriginal()), + resolveLakebaseRestBase: vi.fn(async () => 'https://host.example/api/2.0/workspace/7/rest'), +})); + +const node: INode = { + id: '1', + name: 'Databricks', + type: 'n8n-nodes-base.databricks', + typeVersion: 1, + position: [0, 0], + parameters: {}, +}; + +const column = (overrides: Partial = {}): LakebaseColumn => ({ + name: 'sku', + type: 'string', + hasDefault: false, + isRequired: false, + isReadOnly: false, + isPrimaryKey: false, + ...overrides, +}); + +describe('Lakebase -> column mapping', () => { + const setup = (locators: Record = {}) => { + const values: Record = { + lakebaseProject: 'proj', + lakebaseBranch: 'main', + lakebaseDatabase: 'app', + lakebaseSchema: 'public', + lakebaseTable: 'orders', + authentication: 'oAuth2', + ...locators, + }; + const context = mockDeep(); + context.getNode.mockReturnValue(node); + context.getNodeParameter.mockReturnValue('oAuth2'); + context.getCurrentNodeParameter.mockImplementation( + (name: string) => values[name] as unknown as NodeParameterValueType, + ); + context.getCredentials.mockResolvedValue({ host: 'https://host.example' }); + return context; + }; + + it.each([ + ['time without time zone', 'time'], + ['time with time zone', 'time'], + ['timestamp without time zone', 'dateTime'], + ['date', 'dateTime'], + ])('maps the %s format to the %s field type', async (format, expected) => { + vi.mocked(fetchLakebaseColumns).mockResolvedValue([column({ name: 'at', format })]); + + const { fields } = await getLakebaseMappingColumns.call(setup()); + + expect(fields[0]).toEqual(expect.objectContaining({ type: expected })); + }); + + it('lists the table columns as mappable fields', async () => { + vi.mocked(fetchLakebaseColumns).mockResolvedValue([column({ name: 'sku' })]); + + const { fields } = await getLakebaseMappingColumns.call(setup()); + + expect(fields).toEqual([ + expect.objectContaining({ id: 'sku', displayName: 'sku', display: true, type: 'string' }), + ]); + }); + + it.each([ + ['plain names', 'app', 'public', '/app/public'], + ['names that need escaping', 'my db', 'a/b', '/my%20db/a%2Fb'], + ])('builds the schema URL for %s', async (_name, database, schema, path) => { + vi.mocked(fetchLakebaseColumns).mockResolvedValue([column()]); + + await getLakebaseMappingColumns.call( + setup({ lakebaseDatabase: database, lakebaseSchema: schema }), + ); + + expect(fetchLakebaseColumns).toHaveBeenCalledWith( + expect.anything(), + expect.stringContaining(path), + 'orders', + ); + }); + + it.each([ + ['a NOT NULL column with no default', { isRequired: true }, true], + ['a NOT NULL column the database defaults', { isRequired: true, hasDefault: true }, false], + ['a read-only NOT NULL column', { isRequired: true, isReadOnly: true }, false], + ['a nullable column', {}, false], + ])('marks %s required=%s', async (_name, overrides, expected) => { + vi.mocked(fetchLakebaseColumns).mockResolvedValue([column(overrides)]); + + const { fields } = await getLakebaseMappingColumns.call(setup()); + + expect(fields[0].required).toBe(expected); + }); + + it.each([ + ['integer', { type: 'integer' }, 'number'], + ['number', { type: 'number' }, 'number'], + ['boolean', { type: 'boolean' }, 'boolean'], + ['array', { type: 'array' }, 'array'], + ['object', { type: 'object' }, 'object'], + ['text', { type: 'string' }, 'string'], + ['a timestamp', { type: 'string', format: 'timestamp with time zone' }, 'dateTime'], + ['a date', { type: 'string', format: 'date' }, 'dateTime'], + ['an enum', { type: 'string', enum: ['a', 'b'] }, 'options'], + ['an unknown type', { type: 'inet' }, 'string'], + ])('maps %s to %s', async (_name, overrides, expected) => { + vi.mocked(fetchLakebaseColumns).mockResolvedValue([column(overrides)]); + + const { fields } = await getLakebaseMappingColumns.call(setup()); + + expect(fields[0].type).toBe(expected); + }); + + it('offers the enum values as options', async () => { + vi.mocked(fetchLakebaseColumns).mockResolvedValue([column({ enum: ['new', 'paid'] })]); + + const { fields } = await getLakebaseMappingColumns.call(setup()); + + expect(fields[0].options).toEqual(['new', 'paid'].map((value) => ({ name: value, value }))); + }); + + it('explains how to switch the schema document on instead of failing', async () => { + const unavailable = new NodeApiError(node, { + message: 'Could not find the table', + } as JsonObject); + unavailable.context.data = { code: 'PGRST205' }; + vi.mocked(fetchLakebaseColumns).mockRejectedValue(unavailable); + + const result = await getLakebaseMappingColumns.call(setup()); + + expect(result).toEqual({ fields: [], emptyFieldsNotice: OPENAPI_DISABLED_NOTICE }); + }); + + it('still fails on an error that is not a missing schema document', async () => { + const denied = new NodeApiError(node, { message: 'Forbidden' } as JsonObject); + denied.context.data = { code: '42501' }; + vi.mocked(fetchLakebaseColumns).mockRejectedValue(denied); + + await expect(getLakebaseMappingColumns.call(setup())).rejects.toThrow(); + }); + + it.each([ + 'lakebaseProject', + 'lakebaseBranch', + 'lakebaseDatabase', + 'lakebaseSchema', + 'lakebaseTable', + ])('returns no fields while %s is unset', async (missing) => { + const { fields } = await getLakebaseMappingColumns.call(setup({ [missing]: '' })); + + expect(fields).toEqual([]); + expect(fetchLakebaseColumns).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/nodes-base/nodes/Databricks/transport/lakebase.ts b/packages/nodes-base/nodes/Databricks/transport/lakebase.ts index 6a6db9561fc..6637364c18f 100644 --- a/packages/nodes-base/nodes/Databricks/transport/lakebase.ts +++ b/packages/nodes-base/nodes/Databricks/transport/lakebase.ts @@ -10,6 +10,7 @@ import { makePermissionErrorLegible, type DatabricksContext, } from '../actions/helpers'; +import { OPENAPI_SETTING_PATH } from '../actions/lakebase/openApiDocument'; // ponytail: single fixed delay; lengthen it if a real schema reload is observed // to take more than 1 s. @@ -35,8 +36,7 @@ function isExpiredToken(error: unknown, body: Record | undefine return /expired/i.test(text) && /jwt|token/i.test(text); } -const OPENAPI_SETTING_PATH = - 'In Databricks open the project, then Data API > API > Advanced settings, and enable OpenAPI specification.'; +const OPENAPI_SETTING_HINT = `In Databricks open the project, then turn on ${OPENAPI_SETTING_PATH}.`; /** * A dropdown shows only an error's message, so a raw PostgREST code leaves the user @@ -46,7 +46,7 @@ const OPENAPI_SETTING_PATH = export function explainLakebaseSchemaError(error: unknown, openApiMessage: string): never { if (error instanceof NodeApiError && errorBody(error)?.code === 'PGRST205') { error.message = openApiMessage; - error.description = OPENAPI_SETTING_PATH; + error.description = OPENAPI_SETTING_HINT; throw error; }