diff --git a/src/platform/plugins/private/vis_types/vega/moon.yml b/src/platform/plugins/private/vis_types/vega/moon.yml
index 30185f78624cb..126905de0e2fa 100644
--- a/src/platform/plugins/private/vis_types/vega/moon.yml
+++ b/src/platform/plugins/private/vis_types/vega/moon.yml
@@ -50,6 +50,8 @@ dependsOn:
   - '@kbn/ui-actions-plugin'
   - '@kbn/css-utils'
   - '@kbn/visualizations-common'
+  - '@kbn/es-types'
+  - '@kbn/esql-utils'
 tags:
   - plugin
   - prod
diff --git a/src/platform/plugins/private/vis_types/vega/public/data_model/es_query_parser.ts b/src/platform/plugins/private/vis_types/vega/public/data_model/es_query_parser.ts
index 7f0cfb841aff3..a1262a03ad0cc 100644
--- a/src/platform/plugins/private/vis_types/vega/public/data_model/es_query_parser.ts
+++ b/src/platform/plugins/private/vis_types/vega/public/data_model/es_query_parser.ts
@@ -25,6 +25,7 @@ import type {
   Query,
   ContextVarsObject,
 } from './types';
+import { CONTEXT, getRequestName, TIMEFIELD } from './parser_utils';
 
 const TIMEFILTER: string = '%timefilter%';
 const AUTOINTERVAL: string = '%autointerval%';
@@ -34,15 +35,6 @@ const FILTER_CLAUSE: string = '%dashboard_context-filter_clause%';
 
 // These values may appear in the  'url': { ... }  object
 const LEGACY_CONTEXT: string = '%context_query%';
-const CONTEXT: string = '%context%';
-const TIMEFIELD: string = '%timefield%';
-
-const getRequestName = (request: EsQueryRequest, index: number) =>
-  request.dataObject.name ||
-  i18n.translate('visTypeVega.esQueryParser.unnamedRequest', {
-    defaultMessage: 'Unnamed request #{index}',
-    values: { index },
-  });
 
 /**
  * This class parses ES requests specified in the data.url objects.
@@ -207,7 +199,7 @@ export class EsQueryParser {
   async populateData(requests: EsQueryRequest[]) {
     const esSearches = requests.map((r: EsQueryRequest, index: number) => ({
       ...r.url,
-      name: getRequestName(r, index),
+      name: getRequestName(r.dataObject.name, index),
     }));
 
     const data$ = this._searchAPI.search(esSearches);
@@ -215,7 +207,9 @@ export class EsQueryParser {
     const results = await data$.toPromise();
 
     results.forEach((data, index) => {
-      const requestObject = requests.find((item) => getRequestName(item, index) === data.name);
+      const requestObject = requests.find(
+        (item) => getRequestName(item.dataObject.name, index) === data.name
+      );
 
       if (requestObject) {
         requestObject.dataObject.url = requestObject.url;
diff --git a/src/platform/plugins/private/vis_types/vega/public/data_model/esql_query_parser.test.js b/src/platform/plugins/private/vis_types/vega/public/data_model/esql_query_parser.test.js
new file mode 100644
index 0000000000000..f77bc9b63b456
--- /dev/null
+++ b/src/platform/plugins/private/vis_types/vega/public/data_model/esql_query_parser.test.js
@@ -0,0 +1,610 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { of } from 'rxjs';
+import { EsqlQueryParser } from './esql_query_parser';
+
+const rangeStart = 1000000;
+const rangeEnd = 2000000;
+
+const mockFilters = {
+  bool: {
+    must: [{ match_all: {} }],
+    filter: [{ range: { '@timestamp': { gte: '2024-01-01', lte: '2024-12-31' } } }],
+  },
+};
+
+function createParser(min = rangeStart, max = rangeEnd, dashboardCtx = {}) {
+  const timeCache = {
+    getTimeBounds: () => ({ min, max }),
+  };
+
+  const searchAPI = {
+    searchEsql: jest.fn(() => of([])),
+  };
+
+  const onWarning = jest.fn();
+
+  const parser = new EsqlQueryParser(timeCache, searchAPI, dashboardCtx, onWarning);
+  parser.$$$warnCount = 0;
+  parser._onWarning = (...args) => {
+    parser.$$$warnCount++;
+    onWarning(...args);
+  };
+
+  return { parser, searchAPI, onWarning };
+}
+
+jest.mock('../services');
+
+describe('EsqlQueryParser.parseUrl', () => {
+  test('should parse basic ES|QL query', () => {
+    const { parser } = createParser();
+    const dataObject = { name: 'test' };
+    const url = {
+      '%type%': 'esql',
+      query: 'FROM logs-* | STATS count=COUNT()',
+    };
+
+    const result = parser.parseUrl(dataObject, url);
+
+    expect(result.dataObject).toBe(dataObject);
+    expect(result.url.query).toBe('FROM logs-* | STATS count=COUNT()');
+    expect(result.url.dropNullColumns).toBe(true);
+  });
+
+  test('should throw error when query is missing', () => {
+    const { parser } = createParser();
+    const dataObject = { name: 'test' };
+    const url = {
+      '%type%': 'esql',
+    };
+
+    expect(() => parser.parseUrl(dataObject, url)).toThrow(/requires a.*query.*parameter/);
+  });
+
+  test('should throw error when query is not a string', () => {
+    const { parser } = createParser();
+    const dataObject = { name: 'test' };
+    const url = {
+      '%type%': 'esql',
+      query: { invalid: 'object' },
+    };
+
+    expect(() => parser.parseUrl(dataObject, url)).toThrow(/requires a.*query.*parameter/);
+  });
+
+  test('should throw error when query is empty', () => {
+    const { parser } = createParser();
+    const dataObject = { name: 'test' };
+    const url = {
+      '%type%': 'esql',
+      query: '   ',
+    };
+
+    expect(() => parser.parseUrl(dataObject, url)).toThrow(/cannot be empty/);
+  });
+
+  test('should handle %context% flag', () => {
+    const { parser } = createParser(rangeStart, rangeEnd, mockFilters);
+    const dataObject = { name: 'test' };
+    const url = {
+      '%type%': 'esql',
+      '%context%': true,
+      query: 'FROM logs-* | STATS count=COUNT()',
+    };
+
+    const result = parser.parseUrl(dataObject, url);
+
+    expect(result.url.filter).toEqual(mockFilters);
+    expect(result.url['%context%']).toBeUndefined();
+  });
+
+  test('should handle %timefield% parameter', () => {
+    const { parser } = createParser();
+    const dataObject = { name: 'test' };
+    const url = {
+      '%type%': 'esql',
+      '%timefield%': '@timestamp',
+      query: 'FROM logs-* | WHERE @timestamp >= ?_tstart',
+    };
+
+    const result = parser.parseUrl(dataObject, url);
+
+    expect(result.url._useTimeParams).toBe(true);
+    expect(result.url['%timefield%']).toBeUndefined();
+  });
+
+  test('should set dropNullColumns to true by default', () => {
+    const { parser } = createParser();
+    const dataObject = { name: 'test' };
+    const url = {
+      query: 'FROM logs-* | STATS count=COUNT()',
+    };
+
+    const result = parser.parseUrl(dataObject, url);
+
+    expect(result.url.dropNullColumns).toBe(true);
+  });
+
+  test('should preserve explicit dropNullColumns value', () => {
+    const { parser } = createParser();
+    const dataObject = { name: 'test' };
+    const url = {
+      query: 'FROM logs-* | STATS count=COUNT()',
+      dropNullColumns: false,
+    };
+
+    const result = parser.parseUrl(dataObject, url);
+
+    expect(result.url.dropNullColumns).toBe(false);
+  });
+});
+
+describe('EsqlQueryParser.populateData', () => {
+  test('should execute ES|QL query and populate data', async () => {
+    const { parser, searchAPI } = createParser();
+
+    const mockResponse = [
+      {
+        name: 'test_query',
+        rawResponse: {
+          columns: [
+            { name: 'country', type: 'keyword' },
+            { name: 'count', type: 'long' },
+          ],
+          values: [
+            ['US', 100],
+            ['UK', 50],
+          ],
+        },
+      },
+    ];
+
+    searchAPI.searchEsql.mockReturnValue(of(mockResponse));
+
+    const requests = [
+      {
+        url: { query: 'FROM logs-* | STATS count=COUNT() BY country' },
+        dataObject: { name: 'test_query' },
+      },
+    ];
+
+    await parser.populateData(requests);
+
+    expect(searchAPI.searchEsql).toHaveBeenCalled();
+    expect(requests[0].dataObject.values).toEqual([
+      { country: 'US', count: 100 },
+      { country: 'UK', count: 50 },
+    ]);
+  });
+
+  test('should handle multiple requests', async () => {
+    const { parser, searchAPI } = createParser();
+
+    const mockResponse = [
+      {
+        name: 'query1',
+        rawResponse: {
+          columns: [{ name: 'total', type: 'long' }],
+          values: [[100]],
+        },
+      },
+      {
+        name: 'query2',
+        rawResponse: {
+          columns: [{ name: 'total', type: 'long' }],
+          values: [[200]],
+        },
+      },
+    ];
+
+    searchAPI.searchEsql.mockReturnValue(of(mockResponse));
+
+    const requests = [
+      {
+        url: { query: 'FROM logs-* | STATS total=COUNT()' },
+        dataObject: { name: 'query1' },
+      },
+      {
+        url: { query: 'FROM metrics-* | STATS total=COUNT()' },
+        dataObject: { name: 'query2' },
+      },
+    ];
+
+    await parser.populateData(requests);
+
+    expect(requests[0].dataObject.values).toEqual([{ total: 100 }]);
+    expect(requests[1].dataObject.values).toEqual([{ total: 200 }]);
+  });
+
+  test('should inject time parameters when %timefield% is set', async () => {
+    const { parser, searchAPI } = createParser(1000000, 2000000);
+
+    const mockResponse = [
+      {
+        name: 'time_query',
+        rawResponse: {
+          columns: [{ name: 'count', type: 'long' }],
+          values: [[42]],
+        },
+      },
+    ];
+
+    searchAPI.searchEsql.mockReturnValue(of(mockResponse));
+
+    const requests = [
+      {
+        url: {
+          query: 'FROM logs-* | WHERE @timestamp >= ?_tstart AND @timestamp <= ?_tend',
+          _useTimeParams: true,
+        },
+        dataObject: { name: 'time_query' },
+      },
+    ];
+
+    await parser.populateData(requests);
+
+    const callArgs = searchAPI.searchEsql.mock.calls[0][0][0];
+    expect(callArgs.params).toHaveLength(2);
+    expect(callArgs.params[0]).toHaveProperty('_tstart');
+    expect(callArgs.params[1]).toHaveProperty('_tend');
+    expect(callArgs.params[0]._tstart).toBe(new Date(1000000).toISOString());
+    expect(callArgs.params[1]._tend).toBe(new Date(2000000).toISOString());
+  });
+
+  test('should apply dashboard filters when %context% is true', async () => {
+    const { parser, searchAPI } = createParser(rangeStart, rangeEnd, mockFilters);
+
+    const mockResponse = [
+      {
+        name: 'filtered_query',
+        rawResponse: {
+          columns: [{ name: 'count', type: 'long' }],
+          values: [[10]],
+        },
+      },
+    ];
+
+    searchAPI.searchEsql.mockReturnValue(of(mockResponse));
+
+    const requests = [
+      {
+        url: {
+          query: 'FROM logs-* | STATS count=COUNT()',
+          filter: mockFilters,
+        },
+        dataObject: { name: 'filtered_query' },
+      },
+    ];
+
+    await parser.populateData(requests);
+
+    const callArgs = searchAPI.searchEsql.mock.calls[0][0][0];
+    expect(callArgs.filter).toEqual(mockFilters);
+  });
+
+  test('should handle empty results', async () => {
+    const { parser, searchAPI } = createParser();
+
+    const mockResponse = [
+      {
+        name: 'empty_query',
+        rawResponse: {
+          columns: [{ name: 'count', type: 'long' }],
+          values: [],
+        },
+      },
+    ];
+
+    searchAPI.searchEsql.mockReturnValue(of(mockResponse));
+
+    const requests = [
+      {
+        url: { query: 'FROM logs-* | STATS count=COUNT() | WHERE count > 999999' },
+        dataObject: { name: 'empty_query' },
+      },
+    ];
+
+    await parser.populateData(requests);
+
+    expect(requests[0].dataObject.values).toEqual([]);
+  });
+
+  test('should pass dropNullColumns parameter', async () => {
+    const { parser, searchAPI } = createParser();
+
+    const mockResponse = [
+      {
+        name: 'test',
+        rawResponse: {
+          columns: [{ name: 'count', type: 'long' }],
+          values: [[1]],
+        },
+      },
+    ];
+
+    searchAPI.searchEsql.mockReturnValue(of(mockResponse));
+
+    const requests = [
+      {
+        url: { query: 'FROM logs-*', dropNullColumns: false },
+        dataObject: { name: 'test' },
+      },
+    ];
+
+    await parser.populateData(requests);
+
+    const callArgs = searchAPI.searchEsql.mock.calls[0][0][0];
+    expect(callArgs.dropNullColumns).toBe(false);
+  });
+});
+
+describe('EsqlQueryParser._injectNamedParams', () => {
+  test('should inject time parameters for ?_tstart', () => {
+    const { parser } = createParser(1000000, 2000000);
+
+    const query = 'FROM logs-* | WHERE @timestamp >= ?_tstart';
+    const url = { query, _useTimeParams: true };
+
+    const result = parser._injectNamedParams(query, url);
+
+    expect(result.query).toBe(query);
+    expect(result.params).toHaveLength(1);
+    expect(result.params[0]).toHaveProperty('_tstart');
+    expect(result.params[0]._tstart).toBe(new Date(1000000).toISOString());
+  });
+
+  test('should inject time parameters for ?_tend', () => {
+    const { parser } = createParser(1000000, 2000000);
+
+    const query = 'FROM logs-* | WHERE @timestamp <= ?_tend';
+    const url = { query, _useTimeParams: true };
+
+    const result = parser._injectNamedParams(query, url);
+
+    expect(result.query).toBe(query);
+    expect(result.params).toHaveLength(1);
+    expect(result.params[0]).toHaveProperty('_tend');
+    expect(result.params[0]._tend).toBe(new Date(2000000).toISOString());
+  });
+
+  test('should inject both time parameters', () => {
+    const { parser } = createParser(1000000, 2000000);
+
+    const query = 'FROM logs-* | WHERE @timestamp >= ?_tstart AND @timestamp <= ?_tend';
+    const url = { query, _useTimeParams: true };
+
+    const result = parser._injectNamedParams(query, url);
+
+    expect(result.params).toHaveLength(2);
+    expect(result.params[0]._tstart).toBe(new Date(1000000).toISOString());
+    expect(result.params[1]._tend).toBe(new Date(2000000).toISOString());
+  });
+
+  test('should return empty params when no time parameters in query', () => {
+    const { parser } = createParser(1000000, 2000000);
+
+    const query = 'FROM logs-* | STATS count=COUNT()';
+    const url = { query };
+
+    const result = parser._injectNamedParams(query, url);
+
+    expect(result.params).toHaveLength(0);
+  });
+
+  test('should warn when %timefield% set but no time params in query', () => {
+    const { parser } = createParser(1000000, 2000000);
+
+    const query = 'FROM logs-* | STATS count=COUNT()';
+    const url = { query, _useTimeParams: true };
+
+    parser._injectNamedParams(query, url);
+
+    expect(parser.$$$warnCount).toBe(1);
+  });
+
+  test('should include custom params from url', () => {
+    const { parser } = createParser();
+
+    const query = 'FROM logs-* | WHERE level = ?level';
+    const url = {
+      query,
+      params: [{ level: 'ERROR' }],
+    };
+
+    const result = parser._injectNamedParams(query, url);
+
+    expect(result.params).toHaveLength(1);
+    expect(result.params[0]).toEqual({ level: 'ERROR' });
+  });
+
+  test('should combine time params and custom params', () => {
+    const { parser } = createParser(1000000, 2000000);
+
+    const query = 'FROM logs-* | WHERE @timestamp >= ?_tstart AND level = ?level';
+    const url = {
+      query,
+      _useTimeParams: true,
+      params: [{ level: 'ERROR' }],
+    };
+
+    const result = parser._injectNamedParams(query, url);
+
+    expect(result.params).toHaveLength(2);
+    expect(result.params[0]).toHaveProperty('_tstart');
+    expect(result.params[1]).toEqual({ level: 'ERROR' });
+  });
+
+  test('should handle case-insensitive time parameter detection', () => {
+    const { parser } = createParser(1000000, 2000000);
+
+    const query = 'FROM logs-* | WHERE @timestamp >= ?_TSTART AND @timestamp <= ?_TEND';
+    const url = { query, _useTimeParams: true };
+
+    const result = parser._injectNamedParams(query, url);
+
+    expect(result.params).toHaveLength(2);
+    expect(result.params[0]).toHaveProperty('_tstart');
+    expect(result.params[1]).toHaveProperty('_tend');
+  });
+});
+
+describe('EsqlQueryParser._transformEsqlRowsToVegaRows', () => {
+  test('should transform columnar data to row objects', () => {
+    const { parser } = createParser();
+
+    const response = {
+      columns: [
+        { name: 'country', type: 'keyword' },
+        { name: 'count', type: 'long' },
+      ],
+      values: [
+        ['US', 100],
+        ['UK', 50],
+        ['DE', 75],
+      ],
+    };
+
+    const result = parser._transformEsqlRowsToVegaRows(response);
+
+    expect(result).toEqual([
+      { country: 'US', count: 100 },
+      { country: 'UK', count: 50 },
+      { country: 'DE', count: 75 },
+    ]);
+  });
+
+  test('should handle empty values array', () => {
+    const { parser } = createParser();
+
+    const response = {
+      columns: [{ name: 'count', type: 'long' }],
+      values: [],
+    };
+
+    const result = parser._transformEsqlRowsToVegaRows(response);
+
+    expect(result).toEqual([]);
+  });
+
+  test('should preserve null values', () => {
+    const { parser } = createParser();
+
+    const response = {
+      columns: [
+        { name: 'country', type: 'keyword' },
+        { name: 'count', type: 'long' },
+      ],
+      values: [
+        ['US', 100],
+        [null, 50],
+        ['DE', null],
+      ],
+    };
+
+    const result = parser._transformEsqlRowsToVegaRows(response);
+
+    expect(result).toEqual([
+      { country: 'US', count: 100 },
+      { country: null, count: 50 },
+      { country: 'DE', count: null },
+    ]);
+  });
+
+  test('should handle single column', () => {
+    const { parser } = createParser();
+
+    const response = {
+      columns: [{ name: 'total', type: 'long' }],
+      values: [[42]],
+    };
+
+    const result = parser._transformEsqlRowsToVegaRows(response);
+
+    expect(result).toEqual([{ total: 42 }]);
+  });
+
+  test('should handle many columns', () => {
+    const { parser } = createParser();
+
+    const response = {
+      columns: [
+        { name: 'a', type: 'keyword' },
+        { name: 'b', type: 'long' },
+        { name: 'c', type: 'double' },
+        { name: 'd', type: 'boolean' },
+        { name: 'e', type: 'keyword' },
+      ],
+      values: [['val1', 1, 1.5, true, 'val2']],
+    };
+
+    const result = parser._transformEsqlRowsToVegaRows(response);
+
+    expect(result).toEqual([{ a: 'val1', b: 1, c: 1.5, d: true, e: 'val2' }]);
+  });
+
+  test('should handle transformation errors gracefully', () => {
+    const { parser } = createParser();
+
+    // Invalid response - missing columns
+    const response = {
+      values: [['US', 100]],
+    };
+
+    const result = parser._transformEsqlRowsToVegaRows(response);
+
+    expect(result).toEqual([]);
+    expect(parser.$$$warnCount).toBe(1);
+  });
+
+  test('should handle multi-value fields (arrays)', () => {
+    const { parser } = createParser();
+
+    const response = {
+      columns: [
+        { name: 'tags', type: 'keyword' },
+        { name: 'count', type: 'long' },
+      ],
+      values: [
+        [['tag1', 'tag2'], 100],
+        [['tag3'], 50],
+      ],
+    };
+
+    const result = parser._transformEsqlRowsToVegaRows(response);
+
+    expect(result).toEqual([
+      { tags: ['tag1', 'tag2'], count: 100 },
+      { tags: ['tag3'], count: 50 },
+    ]);
+  });
+
+  test('should handle complex nested objects', () => {
+    const { parser } = createParser();
+
+    const response = {
+      columns: [
+        { name: 'location', type: 'geo_point' },
+        { name: 'count', type: 'long' },
+      ],
+      values: [
+        [{ lat: 40.7128, lon: -74.006 }, 100],
+        [{ lat: 51.5074, lon: -0.1278 }, 50],
+      ],
+    };
+
+    const result = parser._transformEsqlRowsToVegaRows(response);
+
+    expect(result).toEqual([
+      { location: { lat: 40.7128, lon: -74.006 }, count: 100 },
+      { location: { lat: 51.5074, lon: -0.1278 }, count: 50 },
+    ]);
+  });
+});
diff --git a/src/platform/plugins/private/vis_types/vega/public/data_model/esql_query_parser.ts b/src/platform/plugins/private/vis_types/vega/public/data_model/esql_query_parser.ts
new file mode 100644
index 0000000000000..b455a14eb7e14
--- /dev/null
+++ b/src/platform/plugins/private/vis_types/vega/public/data_model/esql_query_parser.ts
@@ -0,0 +1,243 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { i18n } from '@kbn/i18n';
+import type { ESQLSearchResponse } from '@kbn/es-types';
+import { hasStartEndParams, getStartEndParams } from '@kbn/esql-utils';
+import type { TimeCache } from './time_cache';
+import type { SearchAPI } from './search_api';
+import type { Data, EsqlUrlObject, Bool } from './types';
+import { CONTEXT, getRequestName, TIMEFIELD } from './parser_utils';
+
+/**
+ * Internal type for EsqlUrlObject with runtime properties added during parsing
+ */
+interface InternalEsqlUrlObject extends EsqlUrlObject {
+  _useTimeParams?: boolean;
+}
+
+/**
+ * Internal request type using InternalEsqlUrlObject
+ */
+interface InternalEsqlQueryRequest {
+  url: InternalEsqlUrlObject;
+  dataObject: Data;
+}
+
+/**
+ * Named parameters injected into ES|QL query
+ */
+interface InjectedParams {
+  query: string;
+  params: Record<string, unknown>[];
+}
+
+/**
+ * Default value for dropNullColumns parameter
+ */
+const DEFAULT_DROP_NULL_COLUMNS = true;
+
+/**
+ * Localized strings for ESQL query parser
+ */
+const strings = {
+  missingQueryErrorMessage: () =>
+    i18n.translate('visTypeVega.esqlQueryParser.missingQueryErrorMessage', {
+      defaultMessage:
+        '{dataUrlParam} with {typeParam} requires a {queryParam} parameter with an ES|QL query',
+      values: {
+        dataUrlParam: '"data.url"',
+        typeParam: '"%type%": "esql"',
+        queryParam: '"query"',
+      },
+    }),
+  emptyQueryErrorMessage: () =>
+    i18n.translate('visTypeVega.esqlQueryParser.emptyQueryErrorMessage', {
+      defaultMessage: 'ES|QL {queryParam} cannot be empty',
+      values: {
+        queryParam: '"query"',
+      },
+    }),
+  timeFieldWithoutParamsWarning: () =>
+    i18n.translate('visTypeVega.esqlQueryParser.timeFieldWithoutParamsWarning', {
+      defaultMessage:
+        '{timefieldParam} was specified but query does not contain {tstartParam} or {tendParam} parameters',
+      values: {
+        timefieldParam: '"%timefield%"',
+        tstartParam: '?_tstart',
+        tendParam: '?_tend',
+      },
+    }),
+  transformationWarningMessage: (message: string) =>
+    i18n.translate('visTypeVega.esqlQueryParser.transformationWarningMessage', {
+      defaultMessage: 'Failed to transform ES|QL response: {message}. Returning empty data.',
+      values: { message },
+    }),
+};
+
+/**
+ * This class parses ES|QL requests specified in the data.url objects.
+ */
+export class EsqlQueryParser {
+  readonly _timeCache: TimeCache;
+  readonly _searchAPI: SearchAPI;
+  readonly _filters: Bool;
+  readonly _onWarning: (...args: string[]) => void;
+
+  constructor(
+    timeCache: TimeCache,
+    searchAPI: SearchAPI,
+    filters: Bool,
+    onWarning: (...args: string[]) => void
+  ) {
+    this._timeCache = timeCache;
+    this._searchAPI = searchAPI;
+    this._filters = filters;
+    this._onWarning = onWarning;
+  }
+
+  /**
+   * Update request object, validating ES|QL configuration
+   */
+  parseUrl(dataObject: Data, url: EsqlUrlObject): { dataObject: Data; url: InternalEsqlUrlObject } {
+    // Extract special parameters before creating internal URL
+    const context = url[CONTEXT];
+    const timefield = url[TIMEFIELD];
+
+    // Create internal URL without mutating the original
+    const { [CONTEXT]: _, [TIMEFIELD]: __, ...cleanUrl } = url;
+    const internalUrl = cleanUrl as InternalEsqlUrlObject;
+
+    // Validate that query exists
+    if (!internalUrl.query || typeof internalUrl.query !== 'string') {
+      throw new Error(strings.missingQueryErrorMessage());
+    }
+
+    if (internalUrl.query.trim().length === 0) {
+      throw new Error(strings.emptyQueryErrorMessage());
+    }
+
+    // Handle context
+    if (context === true) {
+      internalUrl.filter = this._filters;
+    }
+
+    // Mark that we need time parameter injection
+    if (timefield) {
+      internalUrl._useTimeParams = true;
+    }
+
+    if (internalUrl.dropNullColumns === undefined) {
+      internalUrl.dropNullColumns = DEFAULT_DROP_NULL_COLUMNS;
+    }
+
+    return { dataObject, url: internalUrl };
+  }
+
+  /**
+   * Process items generated by parseUrl()
+   */
+  async populateData(requests: InternalEsqlQueryRequest[]) {
+    const esqlSearches = requests.map((r, index) => {
+      const { query, params: urlParams } = this._injectNamedParams(r.url.query, r.url);
+
+      return {
+        query,
+        filter: r.url.filter,
+        params: urlParams,
+        dropNullColumns: r.url.dropNullColumns ?? DEFAULT_DROP_NULL_COLUMNS,
+        name: getRequestName(r.dataObject.name, index),
+      };
+    });
+
+    const data$ = this._searchAPI.searchEsql(esqlSearches);
+    const results = await data$.toPromise();
+
+    results.forEach((data) => {
+      const requestObject = requests.find(
+        (item, idx) => getRequestName(item.dataObject.name, idx) === data.name
+      );
+
+      if (requestObject) {
+        const esqlResponse = data.rawResponse as unknown as ESQLSearchResponse;
+        const rowData = this._transformEsqlRowsToVegaRows(esqlResponse);
+
+        requestObject.dataObject.url = requestObject.url;
+        requestObject.dataObject.values = rowData;
+      }
+    });
+  }
+
+  /**
+   * Inject named parameters for time range
+   */
+  private _injectNamedParams(query: string, url: InternalEsqlUrlObject): InjectedParams {
+    const params: Record<string, unknown>[] = [];
+
+    // Check if we should inject time parameters
+    if (url._useTimeParams) {
+      if (hasStartEndParams(query)) {
+        const bounds = this._timeCache.getTimeBounds();
+
+        // Convert numeric bounds to TimeRange format for shared utility
+        const timeRange = {
+          from: new Date(bounds.min).toISOString(),
+          to: new Date(bounds.max).toISOString(),
+        };
+
+        params.push(...getStartEndParams(query, timeRange));
+      } else {
+        // Warn user that timefield was specified but no parameters in query
+        this._onWarning(strings.timeFieldWithoutParamsWarning());
+      }
+    }
+
+    // Include any params that were explicitly provided in the URL object
+    if (url.params && Array.isArray(url.params)) {
+      params.push(...url.params);
+    }
+
+    return { query, params };
+  }
+
+  /**
+   * Transform ES|QL columnar response to Vega row-based format
+   */
+  private _transformEsqlRowsToVegaRows(response: ESQLSearchResponse): Record<string, unknown>[] {
+    try {
+      // Check response validity before destructuring
+      if (!response || !response.columns) {
+        throw new Error('Invalid response: missing columns');
+      }
+
+      const { columns, values } = response;
+
+      // Handle empty results
+      if (!values || values.length === 0) {
+        return [];
+      }
+
+      // Transform each row from array to object
+      const rows = values.map((row) => {
+        const rowObject: Record<string, unknown> = {};
+
+        columns.forEach((column, columnIndex) => {
+          rowObject[column.name] = row[columnIndex];
+        });
+
+        return rowObject;
+      });
+
+      return rows;
+    } catch (error) {
+      this._onWarning(strings.transformationWarningMessage((error as Error).message));
+      return [];
+    }
+  }
+}
diff --git a/src/platform/plugins/private/vis_types/vega/public/data_model/parser_utils.ts b/src/platform/plugins/private/vis_types/vega/public/data_model/parser_utils.ts
new file mode 100644
index 0000000000000..6df00d6fa6e4e
--- /dev/null
+++ b/src/platform/plugins/private/vis_types/vega/public/data_model/parser_utils.ts
@@ -0,0 +1,21 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { i18n } from '@kbn/i18n';
+
+// These values may appear in the  'url': { ... }  object
+export const CONTEXT = '%context%';
+export const TIMEFIELD = '%timefield%';
+
+export const getRequestName = (name: string | undefined, index: number) =>
+  name ??
+  i18n.translate('visTypeVega.queryParser.unnamedRequest', {
+    defaultMessage: 'Unnamed request #{index}',
+    values: { index },
+  });
diff --git a/src/platform/plugins/private/vis_types/vega/public/data_model/search_api.ts b/src/platform/plugins/private/vis_types/vega/public/data_model/search_api.ts
index f3b4a404eac4f..8b6110519f390 100644
--- a/src/platform/plugins/private/vis_types/vega/public/data_model/search_api.ts
+++ b/src/platform/plugins/private/vis_types/vega/public/data_model/search_api.ts
@@ -124,6 +124,66 @@ export class SearchAPI {
     }
   }
 
+  searchEsql(
+    esqlRequests: Array<{
+      query: string;
+      filter?: unknown;
+      params?: Array<Record<string, unknown>>;
+      dropNullColumns?: boolean;
+      name: string;
+    }>
+  ) {
+    const { search } = this.dependencies;
+    const requestResponders: any = {};
+
+    return combineLatest(
+      esqlRequests.map((request) => {
+        const { name: requestId, ...restRequest } = request;
+
+        return from(Promise.resolve()).pipe(
+          tap(() => {
+            /** inspect request data **/
+            if (this.inspectorAdapters) {
+              requestResponders[requestId] = this.inspectorAdapters.requests.start(requestId, {
+                ...request,
+                searchSessionId: this.searchSessionId,
+              });
+              requestResponders[requestId].json(restRequest);
+            }
+          }),
+          switchMap(() => {
+            return search
+              .search(
+                { params: restRequest },
+                {
+                  strategy: 'esql_async',
+                  abortSignal: this.abortSignal,
+                  sessionId: this.searchSessionId,
+                  executionContext: this.executionContext,
+                }
+              )
+              .pipe(
+                tap(
+                  (data) => this.inspectSearchResult(data, requestResponders[requestId]),
+                  (err) =>
+                    this.inspectSearchResult(
+                      {
+                        rawResponse: err?.err,
+                      },
+                      requestResponders[requestId]
+                    )
+                ),
+                map((data) => ({
+                  name: requestId,
+                  rawResponse: structuredClone(data.rawResponse),
+                }))
+              );
+          })
+        );
+      })
+    );
+  }
+
   private inspectSearchResult(response: IEsSearchResponse, requestResponder: RequestResponder) {
     if (requestResponder) {
       requestResponder
diff --git a/src/platform/plugins/private/vis_types/vega/public/data_model/types.ts b/src/platform/plugins/private/vis_types/vega/public/data_model/types.ts
index 35dd7421512a2..9a69a0c3d5cb0 100644
--- a/src/platform/plugins/private/vis_types/vega/public/data_model/types.ts
+++ b/src/platform/plugins/private/vis_types/vega/public/data_model/types.ts
@@ -11,6 +11,7 @@ import type { estypes } from '@elastic/elasticsearch';
 import type { Assign } from '@kbn/utility-types';
 import type { Spec } from 'vega';
 import type { EsQueryParser } from './es_query_parser';
+import type { EsqlQueryParser } from './esql_query_parser';
 import type { EmsFileParser } from './ems_file_parser';
 import type { UrlParser } from './url_parser';
 
@@ -167,6 +168,13 @@ export interface UrlObject {
   timeout?: string;
 }
 
+export interface EsqlUrlObject extends UrlObject {
+  query: string;
+  filter?: unknown;
+  dropNullColumns?: boolean;
+  params?: Array<Record<string, unknown>>;
+}
+
 export interface Data {
   [index: string]: any;
   url?: UrlObject;
@@ -186,6 +194,7 @@ interface Requests<TUrlData = UrlObject, TRequestDataObject = RequestDataObject<
 }
 
 export type EsQueryRequest = Requests;
+export type EsqlQueryRequest = Requests<EsqlUrlObject, RequestDataObject<EsqlUrlObject>>;
 export type EmsQueryRequest = Requests & {
   obj: UrlObject;
 };
@@ -231,6 +240,7 @@ export interface VegaConfig extends DstObj {
 export interface UrlParserConfig {
   [index: string]: any;
   elasticsearch: EsQueryParser;
+  esql: EsqlQueryParser;
   emsfile: EmsFileParser;
   url: UrlParser;
 }
diff --git a/src/platform/plugins/private/vis_types/vega/public/data_model/vega_parser.ts b/src/platform/plugins/private/vis_types/vega/public/data_model/vega_parser.ts
index de040e80e23e3..d194c9d1ac1a4 100644
--- a/src/platform/plugins/private/vis_types/vega/public/data_model/vega_parser.ts
+++ b/src/platform/plugins/private/vis_types/vega/public/data_model/vega_parser.ts
@@ -20,6 +20,7 @@ import { compile, version as vegaLiteVersion } from 'vega-lite';
 
 import type { CoreTheme } from '@kbn/core/public';
 import { EsQueryParser } from './es_query_parser';
+import { EsqlQueryParser } from './esql_query_parser';
 import { Utils, getVegaThemeColors } from './utils';
 import { EmsFileParser } from './ems_file_parser';
 import { UrlParser } from './url_parser';
@@ -605,6 +606,7 @@ The URL is an identifier only. Kibana and your browser will never access this UR
       const onWarn = this._onWarning.bind(this);
       this._urlParsers = {
         elasticsearch: new EsQueryParser(this.timeCache, this.searchAPI, this.filters, onWarn),
+        esql: new EsqlQueryParser(this.timeCache, this.searchAPI, this.filters, onWarn),
         emsfile: new EmsFileParser(serviceSettings),
         url: new UrlParser(onWarn),
       };
diff --git a/src/platform/plugins/private/vis_types/vega/tsconfig.json b/src/platform/plugins/private/vis_types/vega/tsconfig.json
index 3ca1960bcdcf6..798e47b69682f 100644
--- a/src/platform/plugins/private/vis_types/vega/tsconfig.json
+++ b/src/platform/plugins/private/vis_types/vega/tsconfig.json
@@ -46,6 +46,8 @@
     "@kbn/ui-actions-plugin",
     "@kbn/css-utils",
     "@kbn/visualizations-common",
+    "@kbn/es-types",
+    "@kbn/esql-utils",
   ],
   "exclude": [
     "target/**/*",
diff --git a/x-pack/platform/plugins/private/translations/translations/de-DE.json b/x-pack/platform/plugins/private/translations/translations/de-DE.json
index c76da4970209e..14dabea839384 100644
--- a/x-pack/platform/plugins/private/translations/translations/de-DE.json
+++ b/x-pack/platform/plugins/private/translations/translations/de-DE.json
@@ -8892,7 +8892,7 @@
     "visTypeVega.esQueryParser.shiftMustValueTypeErrorMessage": "{shiftParam} muss ein numerischer Wert sein",
     "visTypeVega.esQueryParser.timefilterValueErrorMessage": "Die Eigenschaft {timefilter} muss auf {trueValue}, {minValue} oder {maxValue} gesetzt werden",
     "visTypeVega.esQueryParser.unknownUnitValueErrorMessage": "Unbekannter {unitParamName}-Wert. Muss eines der folgenden sein: [{unitParamValues}]",
-    "visTypeVega.esQueryParser.unnamedRequest": "Unbenannte Anfrage #{index}",
+    "visTypeVega.queryParser.unnamedRequest": "Unbenannte Anfrage #{index}",
     "visTypeVega.esQueryParser.urlBodyValueTypeErrorMessage": "{configName} muss ein Objekt sein",
     "visTypeVega.esQueryParser.urlContextAndUrlTimefieldMustNotBeUsedErrorMessage": "{urlContext} und {timefield} dürfen nicht verwendet werden, wenn {queryParam} gesetzt ist.",
     "visTypeVega.function.help": "Vega-Visualisierung",
diff --git a/x-pack/platform/plugins/private/translations/translations/fr-FR.json b/x-pack/platform/plugins/private/translations/translations/fr-FR.json
index fac44c46912b0..b09d78619cad3 100644
--- a/x-pack/platform/plugins/private/translations/translations/fr-FR.json
+++ b/x-pack/platform/plugins/private/translations/translations/fr-FR.json
@@ -9041,7 +9041,7 @@
     "visTypeVega.esQueryParser.shiftMustValueTypeErrorMessage": "{shiftParam} doit être une valeur numérique",
     "visTypeVega.esQueryParser.timefilterValueErrorMessage": "La propriété {timefilter} doit être définie sur {trueValue}, {minValue} ou {maxValue}",
     "visTypeVega.esQueryParser.unknownUnitValueErrorMessage": "Valeur {unitParamName} inconnue. Doit être l'une des valeurs suivantes : [{unitParamValues}]",
-    "visTypeVega.esQueryParser.unnamedRequest": "Requête sans nom #{index}",
+    "visTypeVega.queryParser.unnamedRequest": "Requête sans nom #{index}",
     "visTypeVega.esQueryParser.urlBodyValueTypeErrorMessage": "{configName} doit être un objet",
     "visTypeVega.esQueryParser.urlContextAndUrlTimefieldMustNotBeUsedErrorMessage": "{urlContext} et {timefield} ne doivent pas être utilisés lorsque {queryParam} est défini",
     "visTypeVega.function.help": "Visualisation Vega",
diff --git a/x-pack/platform/plugins/private/translations/translations/ja-JP.json b/x-pack/platform/plugins/private/translations/translations/ja-JP.json
index 5a9dfb478cf60..a99bb74a98e27 100644
--- a/x-pack/platform/plugins/private/translations/translations/ja-JP.json
+++ b/x-pack/platform/plugins/private/translations/translations/ja-JP.json
@@ -9052,7 +9052,7 @@
     "visTypeVega.esQueryParser.shiftMustValueTypeErrorMessage": "{shiftParam} は数値でなければなりません",
     "visTypeVega.esQueryParser.timefilterValueErrorMessage": "{timefilter} のプロパティは {trueValue}、{minValue}、または {maxValue} に設定する必要があります",
     "visTypeVega.esQueryParser.unknownUnitValueErrorMessage": "不明な {unitParamName} 値。次のいずれかでなければなりません。[{unitParamValues}]",
-    "visTypeVega.esQueryParser.unnamedRequest": "無題のリクエスト#{index}",
+    "visTypeVega.queryParser.unnamedRequest": "無題のリクエスト#{index}",
     "visTypeVega.esQueryParser.urlBodyValueTypeErrorMessage": "{configName} はオブジェクトでなければなりません",
     "visTypeVega.esQueryParser.urlContextAndUrlTimefieldMustNotBeUsedErrorMessage": "{urlContext} と {timefield} は {queryParam} が設定されている場合使用できません",
     "visTypeVega.function.help": "Vega ビジュアライゼーション",
diff --git a/x-pack/platform/plugins/private/translations/translations/zh-CN.json b/x-pack/platform/plugins/private/translations/translations/zh-CN.json
index 7495d8e319952..502aeef9c2287 100644
--- a/x-pack/platform/plugins/private/translations/translations/zh-CN.json
+++ b/x-pack/platform/plugins/private/translations/translations/zh-CN.json
@@ -9042,7 +9042,7 @@
     "visTypeVega.esQueryParser.shiftMustValueTypeErrorMessage": "{shiftParam} 必须为数值",
     "visTypeVega.esQueryParser.timefilterValueErrorMessage": "{timefilter} 属性必须设置为 {trueValue}、{minValue} 或 {maxValue}",
     "visTypeVega.esQueryParser.unknownUnitValueErrorMessage": "{unitParamName} 值未知。必须是以下值之一：[{unitParamValues}]",
-    "visTypeVega.esQueryParser.unnamedRequest": "未命名的请求 #{index}",
+    "visTypeVega.queryParser.unnamedRequest": "未命名的请求 #{index}",
     "visTypeVega.esQueryParser.urlBodyValueTypeErrorMessage": "{configName} 必须为对象",
     "visTypeVega.esQueryParser.urlContextAndUrlTimefieldMustNotBeUsedErrorMessage": "设置了 {queryParam} 时，不得使用 {urlContext} 和 {timefield}",
     "visTypeVega.function.help": "Vega 可视化",