feat(Databricks Node): Add the Lakebase Insert operation (no-changelog) (#40520)

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<INodeExecutionData[][]> {
 		return await router.call(this);

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';

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) {

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<string | undefined> {
+async function primaryKeyColumns(context: IExecuteFunctions, i: number): Promise<string[]> {
 	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<INode
 	const qs = buildLakebaseQuery({ where, combineConditions, sort, outputColumns });
 
 	// Every page has to share one ordering, so this is settled before the first request
-	if (qs.order === undefined && (returnAll || limit > 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<INode
 	// have to be the last one. Only once the API has filled a request can a shorter
 	// answer be read as the end of the table.
 	let apiHasFilledAPage = false;
+	let reachedTheEnd = false;
 
 	for (let page = 0; page < LAKEBASE_MAX_PAGES; page++) {
 		const requested = returnAll
 			? LAKEBASE_PAGE_SIZE
 			: Math.min(limit - rows.length, LAKEBASE_PAGE_SIZE);
-		if (requested <= 0) break;
+		if (requested <= 0) {
+			reachedTheEnd = true;
+			break;
+		}
 
 		const response = await lakebaseApiRequest(this, {
 			method: 'GET',
@@ -78,12 +83,35 @@ export async function execute(this: IExecuteFunctions, i: number): Promise<INode
 
 		const batch = Array.isArray(response) ? (response as IDataObject[]) : [];
 		rows.push(...batch);
-		if (batch.length === 0) break;
+		if (batch.length === 0) {
+			reachedTheEnd = true;
+			break;
+		}
 
 		if (batch.length >= 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 } }));

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,

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<INodeExecutionData[]> {
+	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 } }));
+}

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';
+}

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<INodePropertyOptions[]> {
-	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(

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<ResourceMapperFields> {
+	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 })),
+		})),
+	};
+}

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',
 };

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'] },
+		},
+	},
 ];

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({});
 		});

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 () => {

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<string, NodeParameterValueType | object> = {},
+		inputJson: IDataObject[] = [{ sku: 'from-input' }],
+		itemIndex = 0,
+	) => {
+		const parameters: Record<string, NodeParameterValueType | object> = {
+			authentication: 'oAuth2',
+			'columns.mappingMode': 'defineBelow',
+			'columns.value': { sku: 'WF-1', price: 10 },
+			...overrides,
+		};
+		const context = mockDeep<IExecuteFunctions>();
+		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<typeof setupContext>) =>
+		context.helpers.httpRequestWithAuthentication;
+	const run = (
+		response: unknown,
+		overrides: Record<string, NodeParameterValueType | object> = {},
+		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<string, string> };
+		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<string, NodeParameterValueType | object>]>([
+		['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 } }]);
+	});
+});

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);
+	});
+});

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 () => {

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<typeof import('../actions/lakebase/schema')>()),
+	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<typeof import('../transport')>()),
+	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> = {}): LakebaseColumn => ({
+	name: 'sku',
+	type: 'string',
+	hasDefault: false,
+	isRequired: false,
+	isReadOnly: false,
+	isPrimaryKey: false,
+	...overrides,
+});
+
+describe('Lakebase -> column mapping', () => {
+	const setup = (locators: Record<string, string> = {}) => {
+		const values: Record<string, string> = {
+			lakebaseProject: 'proj',
+			lakebaseBranch: 'main',
+			lakebaseDatabase: 'app',
+			lakebaseSchema: 'public',
+			lakebaseTable: 'orders',
+			authentication: 'oAuth2',
+			...locators,
+		};
+		const context = mockDeep<ILoadOptionsFunctions>();
+		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();
+	});
+});

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<string, unknown> | 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;
 	}