feat(Databricks Node): Refuse the Vector Store metadata filter on non-standard endpoints (no-changelog) (#41047)

packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreDatabricks/DatabricksVectorStore.test.ts

@@ -13,17 +13,21 @@ const managedDescribe = {
 	name: 'cat.sch.idx',
 	primary_key: 'id',
 	index_type: 'DELTA_SYNC',
+	endpoint_name: 'ep',
 	delta_sync_index_spec: {
 		source_table: 'cat.sch.docs',
 		embedding_source_columns: [{ name: 'text', embedding_model_endpoint_name: 'e5' }],
 	},
 };
 const ucTable = { columns: [{ name: 'id' }, { name: 'text' }, { name: 'source' }] };
+const standardEndpoint = { name: 'ep', endpoint_type: 'STANDARD' };
+const storageOptimizedEndpoint = { name: 'ep', endpoint_type: 'STORAGE_OPTIMIZED' };
 
 const directDescribe = {
 	name: 'cat.sch.idx',
 	primary_key: 'id',
 	index_type: 'DIRECT_ACCESS',
+	endpoint_name: 'ep',
 	direct_access_index_spec: {
 		embedding_vector_columns: [{ name: 'embedding', embedding_dimension: 3 }],
 		schema_json: '{"id":"string","text":"string","source":"string","embedding":"array<float>"}',
@@ -54,6 +58,13 @@ describe('DatabricksVectorStore', () => {
 	const bodyOf = (call: number) => JSON.parse(String(fetchMock.mock.calls[call][1]?.body));
 	const urlOf = (call: number) => String(fetchMock.mock.calls[call][0]);
 
+	// A filtered query reads the endpoint type before the query; STANDARD lets the filter through
+	const replyByUrl = (endpoint: unknown = standardEndpoint) =>
+		fetchMock.mockImplementation(async (url) =>
+			json(String(url).endsWith('/query') ? queryReply : endpoint),
+		);
+	const queryBody = () => bodyOf(fetchMock.mock.calls.length - 1);
+
 	const createStore = async (
 		describe: unknown,
 		overrides: Partial<DatabricksVectorStoreConfig> = {},
@@ -94,6 +105,7 @@ describe('DatabricksVectorStore', () => {
 				embedding: { kind: 'managed', sourceColumn: 'text', modelEndpoint: 'e5' },
 				schemaColumns: undefined,
 				sourceTable: 'cat.sch.docs',
+				endpointName: 'ep',
 			});
 		});
 
@@ -159,9 +171,11 @@ describe('DatabricksVectorStore', () => {
 
 			const info = await DatabricksVectorStore.describeIndex(fetchMock, host, 'cat.sch.idx');
 
+			expect(fetchMock).toHaveBeenCalledTimes(2);
 			expect(urlOf(0)).toBe(`${host}/api/2.0/vector-search/indexes/cat.sch.idx`);
 			expect(urlOf(1)).toBe(`${host}/api/2.1/unity-catalog/tables/cat.sch.docs`);
 			expect(info.schemaColumns).toEqual(['id', 'text', 'source']);
+			expect(info.endpointType).toBeUndefined();
 		});
 
 		it('leaves the columns unknown when Unity Catalog denies the table', async () => {
@@ -267,11 +281,11 @@ describe('DatabricksVectorStore', () => {
 		// The retriever calls similaritySearch without a filter, so the configured one must apply
 		it('applies the configured search mode and filter to a retriever search', async () => {
 			const store = await managedStore({ queryType: 'HYBRID', filter: { source: 'hr' } });
-			fetchMock.mockResolvedValue(json(queryReply));
+			replyByUrl();
 
 			const docs = await store.similaritySearch('hello', 3);
 
-			expect(bodyOf(0)).toEqual({
+			expect(queryBody()).toEqual({
 				columns: ['id', 'text', 'source'],
 				num_results: 3,
 				query_type: 'HYBRID',
@@ -366,12 +380,129 @@ describe('DatabricksVectorStore', () => {
 			},
 		])('sends $label as filters_json', async ({ config, call, expected }) => {
 			const store = await managedStore({ filter: config });
-			fetchMock.mockResolvedValue(json(queryReply));
+			replyByUrl();
 
 			await store.similaritySearchWithScore('hello', 2, call);
 
-			expect(bodyOf(0).filters_json).toEqual(expected);
+			expect(queryBody().filters_json).toEqual(expected);
 		});
+
+		it('reads the endpoint type once per index description, and only for a filtered query', async () => {
+			const store = await managedStore();
+			replyByUrl();
+
+			await store.similaritySearchWithScore('hello', 2);
+			await store.similaritySearchWithScore('hello', 2, { source: 'hr' });
+			await store.similaritySearchWithScore('hello', 2, { source: 'it' });
+
+			expect(fetchMock.mock.calls.map(([url]) => String(url))).toEqual([
+				`${host}/api/2.0/vector-search/indexes/cat.sch.idx/query`,
+				`${host}/api/2.0/vector-search/endpoints/ep`,
+				`${host}/api/2.0/vector-search/indexes/cat.sch.idx/query`,
+				`${host}/api/2.0/vector-search/indexes/cat.sch.idx/query`,
+			]);
+		});
+
+		it('shares one endpoint read between concurrent filtered queries', async () => {
+			const store = await managedStore();
+			replyByUrl();
+
+			await Promise.all([
+				store.similaritySearchWithScore('hello', 2, { source: 'hr' }),
+				store.similaritySearchWithScore('hello', 2, { source: 'it' }),
+			]);
+
+			expect(fetchMock.mock.calls.map(([url]) => String(url))).toEqual([
+				`${host}/api/2.0/vector-search/endpoints/ep`,
+				`${host}/api/2.0/vector-search/indexes/cat.sch.idx/query`,
+				`${host}/api/2.0/vector-search/indexes/cat.sch.idx/query`,
+			]);
+		});
+
+		it('remembers a refused endpoint read instead of repeating it', async () => {
+			const store = await managedStore();
+			fetchMock.mockResolvedValue(json({ error_code: 'PERMISSION_DENIED' }, 403));
+			const search = async () => await store.similaritySearch('hello', 2, { source: 'hr' });
+
+			await expect(search()).rejects.toThrow('Check that you can read endpoint ep');
+			await expect(search()).rejects.toThrow('Check that you can read endpoint ep');
+
+			expect(fetchMock).toHaveBeenCalledTimes(1);
+		});
+
+		it('retries the endpoint read after a transport error', async () => {
+			const store = await managedStore();
+			fetchMock.mockRejectedValueOnce(new Error('ECONNRESET'));
+
+			await expect(store.similaritySearch('hello', 2, { source: 'hr' })).rejects.toThrow(
+				'ECONNRESET',
+			);
+			replyByUrl();
+			await store.similaritySearch('hello', 2, { source: 'hr' });
+
+			expect(fetchMock).toHaveBeenCalledTimes(3);
+			expect(urlOf(1)).toBe(`${host}/api/2.0/vector-search/endpoints/ep`);
+		});
+
+		it.each([
+			{
+				label: 'the call filter on a STORAGE_OPTIMIZED endpoint',
+				endpoint: json(storageOptimizedEndpoint),
+				config: undefined,
+				search: async (store: DatabricksVectorStore) =>
+					await store.similaritySearchWithScore('hello', 2, { source: 'hr' }),
+				fragment: 'of type STORAGE_OPTIMIZED',
+			},
+			{
+				label: 'the config filter on a STORAGE_OPTIMIZED endpoint',
+				endpoint: json(storageOptimizedEndpoint),
+				config: { source: 'hr' },
+				search: async (store: DatabricksVectorStore) => await store.similaritySearch('hello', 2),
+				fragment: 'needs a SQL filter string (filter_string)',
+			},
+			{
+				label: 'a filter when the endpoint cannot be read',
+				endpoint: json({ error_code: 'PERMISSION_DENIED' }, 403),
+				config: undefined,
+				search: async (store: DatabricksVectorStore) =>
+					await store.similaritySearchWithScore('hello', 2, { source: 'hr' }),
+				fragment: 'Check that you can read endpoint ep',
+			},
+		])('refuses $label before the query', async ({ endpoint, config, search, fragment }) => {
+			const store = await managedStore({ filter: config });
+			fetchMock.mockResolvedValueOnce(endpoint);
+
+			const attempt = search(store);
+
+			await expect(attempt).rejects.toThrow(UserError);
+			await expect(attempt).rejects.toThrow(fragment);
+			expect(fetchMock).toHaveBeenCalledTimes(1);
+			expect(urlOf(0)).toBe(`${host}/api/2.0/vector-search/endpoints/ep`);
+		});
+
+		it('refuses a filter without any request when the index names no endpoint', async () => {
+			const { endpoint_name: _, ...noEndpoint } = directDescribe;
+			const store = await createStore(noEndpoint, { contentColumn: 'text' });
+
+			await expect(store.similaritySearch('hello', 2, { source: 'hr' })).rejects.toThrow(
+				'Check that you can read endpoint (unknown)',
+			);
+			expect(fetchMock).not.toHaveBeenCalled();
+		});
+
+		it.each([{ call: {} }, { call: undefined }])(
+			'queries without a filter ($call) and never reads the endpoint',
+			async ({ call }) => {
+				const store = await managedStore();
+				fetchMock.mockResolvedValue(json(queryReply));
+
+				await store.similaritySearchWithScore('hello', 2, call);
+
+				expect(fetchMock).toHaveBeenCalledTimes(1);
+				expect(urlOf(0)).toBe(`${host}/api/2.0/vector-search/indexes/cat.sch.idx/query`);
+				expect(bodyOf(0)).not.toHaveProperty('filters_json');
+			},
+		);
 	});
 
 	describe('result parsing', () => {

packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreDatabricks/DatabricksVectorStore.ts

@@ -20,6 +20,13 @@ export type DatabricksIndexInfo = {
 	/** DIRECT_ACCESS: keys of schema_json. DELTA_SYNC: columns_to_sync, or the Unity Catalog columns */
 	schemaColumns?: string[];
 	sourceTable?: string;
+	/** Vector search endpoint that serves the index */
+	endpointName?: string;
+	/**
+	 * STANDARD or STORAGE_OPTIMIZED. The first filtered query starts the GET and caches the promise,
+	 * so concurrent and later queries share one read. Resolves undefined when the GET is refused
+	 */
+	endpointType?: Promise<string | undefined>;
 };
 
 export interface DatabricksVectorStoreConfig {
@@ -123,6 +130,7 @@ export function parseIndexInfo(raw: unknown): DatabricksIndexInfo {
 		embedding,
 		schemaColumns,
 		sourceTable: optionalString(deltaSpec?.source_table),
+		endpointName: optionalString(raw.endpoint_name),
 	};
 }
 
@@ -391,6 +399,7 @@ export class DatabricksVectorStore extends VectorStore {
 			effectiveFilter && Object.keys(effectiveFilter).length > 0
 				? JSON.stringify(effectiveFilter)
 				: undefined;
+		if (filtersJson) await this.assertFilterable();
 
 		const response = await databricksRequest(
 			this.fetch,
@@ -439,4 +448,49 @@ export class DatabricksVectorStore extends VectorStore {
 			Number(row[scoreAt]),
 		]);
 	}
+
+	// Databricks accepts filters_json on STANDARD endpoints only; STORAGE_OPTIMIZED takes filter_string.
+	// An unknown type also refuses, so a filter never reaches an endpoint that can ignore it
+	private async assertFilterable(): Promise<void> {
+		const { name, endpointName } = this.index;
+		// The node shares one index description per run, so the GET happens once per run, and only
+		// for filtered queries: the column dropdown and unfiltered runs never pay for it. Caching the
+		// promise keeps concurrent tool calls on one GET; a refused read caches as unknown. A transport
+		// error evicts, like the describe cache in the node, so one blip does not poison the run
+		this.index.endpointType ??= endpointName
+			? this.readEndpointType(endpointName).catch((error: unknown) => {
+					this.index.endpointType = undefined;
+					throw error;
+				})
+			: Promise.resolve(undefined);
+		const endpointType = await this.index.endpointType;
+		if (endpointType === 'STANDARD') return;
+
+		const endpoint = sanitizeMessage(endpointName ?? '(unknown)');
+		if (!endpointType) {
+			throw new UserError(
+				`Could not read the type of the endpoint that serves index ${name}. The Metadata Filter works on STANDARD endpoints only. Check that you can read endpoint ${endpoint}`,
+			);
+		}
+		const hint =
+			endpointType === 'STORAGE_OPTIMIZED'
+				? ' A STORAGE_OPTIMIZED endpoint needs a SQL filter string (filter_string), which this node does not support yet.'
+				: '';
+		throw new UserError(
+			`Index ${name} runs on endpoint ${endpoint} of type ${sanitizeMessage(endpointType)}. The Metadata Filter works on STANDARD endpoints only.${hint} See https://docs.databricks.com/aws/en/ai-search/filtering-guide`,
+		);
+	}
+
+	private async readEndpointType(endpointName: string): Promise<string | undefined> {
+		const response = await this.fetch(
+			`${this.host}/api/2.0/vector-search/endpoints/${toPathSegment(endpointName)}`,
+		);
+		// No read on the endpoint: the type stays unknown and the filtered query fails
+		if (!response.ok) {
+			await response.body?.cancel();
+			return undefined;
+		}
+		const endpoint: unknown = await response.json();
+		return optionalString(isRecord(endpoint) ? endpoint.endpoint_type : undefined);
+	}
 }

packages/@n8n/nodes-langchain/nodes/vector_store/VectorStoreDatabricks/VectorStoreDatabricks.node.ts

@@ -114,7 +114,7 @@ const sharedFields: INodeProperties[] = [
 				default: '{}',
 				validateType: 'object',
 				description:
-					'Restrict results to rows matching this metadata condition, for example {"source": "handbook"}. Sent as filters_json.',
+					'Restrict results to rows matching this metadata condition, for example {"source": "handbook"}. Sent as filters_json. Works on STANDARD endpoints only.',
 			},
 		],
 	},