From 6ccadc8454ca0191ee8f19121a0ee0fff3677458 Mon Sep 17 00:00:00 2001
From: Carlos Crespo <carloshenrique.leonelcrespo@elastic.co>
Date: Thu, 19 Dec 2024 10:04:57 +0100
Subject: [PATCH] esql-composer ideas

---
 packages/kbn-esql-composer/index.ts           |   9 +-
 packages/kbn-esql-composer/jest.config.js     |   9 +-
 packages/kbn-esql-composer/src/builder.ts     |  56 ++++++
 .../kbn-esql-composer/src/commands/append.ts  |  26 ++-
 .../src/commands/drop.test.ts                 |  10 +-
 .../kbn-esql-composer/src/commands/drop.ts    |  23 +--
 .../src/commands/eval.test.ts                 |  57 +++++++
 .../kbn-esql-composer/src/commands/eval.ts    |  44 ++++-
 .../src/commands/from.test.ts                 |  10 +-
 .../kbn-esql-composer/src/commands/from.ts    |  11 +-
 .../src/commands/keep.test.ts                 |  10 +-
 .../kbn-esql-composer/src/commands/keep.ts    |  21 +--
 .../kbn-esql-composer/src/commands/limit.ts   |  11 +-
 .../src/commands/sort.test.ts                 |  65 +++++--
 .../kbn-esql-composer/src/commands/sort.ts    |  30 ++--
 .../src/commands/stats.test.ts                | 134 +++++++++++++++
 .../kbn-esql-composer/src/commands/stats.ts   |  65 ++++++-
 .../src/commands/where.test.ts                | 159 ++++++++++++++++++
 .../kbn-esql-composer/src/commands/where.ts   | 132 ++++++++++++++-
 .../kbn-esql-composer/src/create_pipeline.ts  |  36 +++-
 packages/kbn-esql-composer/src/index.test.ts  |  69 ++++++--
 packages/kbn-esql-composer/src/types.ts       |  39 ++++-
 .../src/utils/escape_identifier.ts            |   9 +-
 23 files changed, 904 insertions(+), 131 deletions(-)
 create mode 100644 packages/kbn-esql-composer/src/builder.ts
 create mode 100644 packages/kbn-esql-composer/src/commands/eval.test.ts
 create mode 100644 packages/kbn-esql-composer/src/commands/stats.test.ts
 create mode 100644 packages/kbn-esql-composer/src/commands/where.test.ts

diff --git a/packages/kbn-esql-composer/index.ts b/packages/kbn-esql-composer/index.ts
index 5c93e9a2fe70b..ce1709d720285 100644
--- a/packages/kbn-esql-composer/index.ts
+++ b/packages/kbn-esql-composer/index.ts
@@ -1,9 +1,10 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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".
  */
 
 export { from } from './src/commands/from';
diff --git a/packages/kbn-esql-composer/jest.config.js b/packages/kbn-esql-composer/jest.config.js
index 088320f1a5671..af98fd0083e4d 100644
--- a/packages/kbn-esql-composer/jest.config.js
+++ b/packages/kbn-esql-composer/jest.config.js
@@ -1,9 +1,10 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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".
  */
 
 module.exports = {
diff --git a/packages/kbn-esql-composer/src/builder.ts b/packages/kbn-esql-composer/src/builder.ts
new file mode 100644
index 0000000000000..ecbef9e63eb5d
--- /dev/null
+++ b/packages/kbn-esql-composer/src/builder.ts
@@ -0,0 +1,56 @@
+/*
+ * 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 { append } from './commands/append';
+import {
+  QueryOperator,
+  BuilderCommand,
+  Params,
+  ChainedCommand,
+  QueryBuilderToOperator,
+} from './types';
+
+export abstract class QueryBuilder implements QueryBuilderToOperator {
+  protected readonly commands: BuilderCommand[] = [];
+
+  public abstract build(): ChainedCommand;
+
+  public toQueryOperator(): QueryOperator {
+    return append(this.build());
+  }
+
+  protected buildChain(): ChainedCommand {
+    const commandParts: string[] = [];
+    const bindingParts: Params[] = [];
+
+    for (let i = 0; i < this.commands.length; i++) {
+      const currentCondition = this.commands[i];
+
+      if (i > 0) {
+        commandParts.push(currentCondition.type);
+      }
+
+      if (typeof currentCondition.command === 'function') {
+        const innerCommand = currentCondition.command().buildChain();
+        commandParts.push(
+          currentCondition.nested ? `(${innerCommand.command})` : innerCommand.command
+        );
+        bindingParts.push(innerCommand.bindings ?? []);
+      } else {
+        commandParts.push(currentCondition.command);
+        bindingParts.push(currentCondition.bindings ?? []);
+      }
+    }
+
+    return {
+      command: commandParts.join(' '),
+      bindings: bindingParts.flatMap((binding) => binding),
+    };
+  }
+}
diff --git a/packages/kbn-esql-composer/src/commands/append.ts b/packages/kbn-esql-composer/src/commands/append.ts
index 4d45bba13c63d..55aa7957a9c3b 100644
--- a/packages/kbn-esql-composer/src/commands/append.ts
+++ b/packages/kbn-esql-composer/src/commands/append.ts
@@ -1,19 +1,31 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { Command, QueryOperator } from '../types';
+import { isObject } from 'lodash';
+import { Command, QueryOperator, Params, Query } from '../types';
 
-export function append(command: Command | string): QueryOperator {
-  return (source) => {
+export function append({
+  command,
+  bindings,
+}: {
+  command: Command | string;
+  bindings?: Params;
+}): QueryOperator {
+  return (source): Query => {
     const nextCommand = typeof command === 'string' ? { body: command } : command;
+
     return {
       ...source,
       commands: source.commands.concat(nextCommand),
+      bindings: !!bindings
+        ? source.bindings.concat(isObject(bindings) ? bindings : [bindings])
+        : source.bindings,
     };
   };
 }
diff --git a/packages/kbn-esql-composer/src/commands/drop.test.ts b/packages/kbn-esql-composer/src/commands/drop.test.ts
index 9e2eb4c214cd7..660dfb464c951 100644
--- a/packages/kbn-esql-composer/src/commands/drop.test.ts
+++ b/packages/kbn-esql-composer/src/commands/drop.test.ts
@@ -1,10 +1,12 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { drop } from './drop';
 import { from } from './from';
 
diff --git a/packages/kbn-esql-composer/src/commands/drop.ts b/packages/kbn-esql-composer/src/commands/drop.ts
index ccc2a043e2976..e8077aa1886a8 100644
--- a/packages/kbn-esql-composer/src/commands/drop.ts
+++ b/packages/kbn-esql-composer/src/commands/drop.ts
@@ -1,19 +1,22 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { escapeIdentifier } from '../utils/escape_identifier';
 import { append } from './append';
 
 export function drop(...columns: Array<string | string[]>) {
-  return append(
-    `DROP ${columns
-      .flatMap((column) => column)
-      .map((column) => escapeIdentifier(column))
-      .join(', ')}`
-  );
+  const command = `DROP ${columns
+    .flatMap((column) => column)
+    .map((column) => escapeIdentifier(column))
+    .join(', ')}`;
+
+  return append({
+    command,
+  });
 }
diff --git a/packages/kbn-esql-composer/src/commands/eval.test.ts b/packages/kbn-esql-composer/src/commands/eval.test.ts
new file mode 100644
index 0000000000000..ee4bb959b264e
--- /dev/null
+++ b/packages/kbn-esql-composer/src/commands/eval.test.ts
@@ -0,0 +1,57 @@
+/*
+ * 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 { evaluate } from './eval';
+import { from } from './from';
+
+describe('evaluate', () => {
+  const source = from('logs-*');
+
+  it('handles single strings', () => {
+    const pipeline = source.pipe(
+      evaluate('type = CASE(languages <= 1, "monolingual",languages <= 2, "bilingual","polyglot")')
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| EVAL type = CASE(languages <= 1, "monolingual",languages <= 2, "bilingual","polyglot")'
+    );
+    expect(pipeline.getBindings()).toEqual([]);
+  });
+
+  it('handles chained EVAL', () => {
+    const ids = ['host1', 'host2', 'host3'];
+    const pipeline = source.pipe(
+      evaluate('entity.type = ?', 'host')
+        .concat('entity.display_name = COALESCE(?, entity.id)', 'some_host')
+        .concat(`entity.id = CONCAT(${ids.map(() => '?').join()})`, ids)
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| EVAL entity.type = ?, entity.display_name = COALESCE(?, entity.id), entity.id = CONCAT(?,?,?)'
+    );
+    expect(pipeline.getBindings()).toEqual(['host', 'some_host', 'host1', 'host2', 'host3']);
+  });
+
+  it('handles EVAL with params', () => {
+    const pipeline = source.pipe(
+      evaluate('hour = DATE_TRUNC(1 hour, ?ts)', {
+        ts: {
+          identifier: '@timestamp',
+        },
+      })
+    );
+
+    expect(pipeline.asString()).toEqual('FROM `logs-*`\n\t| EVAL hour = DATE_TRUNC(1 hour, ?ts)');
+    expect(pipeline.getBindings()).toEqual([
+      {
+        ts: {
+          identifier: '@timestamp',
+        },
+      },
+    ]);
+  });
+});
diff --git a/packages/kbn-esql-composer/src/commands/eval.ts b/packages/kbn-esql-composer/src/commands/eval.ts
index ea10e1403366a..516ee490db763 100644
--- a/packages/kbn-esql-composer/src/commands/eval.ts
+++ b/packages/kbn-esql-composer/src/commands/eval.ts
@@ -1,14 +1,42 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { QueryOperator } from '../types';
-import { append } from './append';
+import { QueryBuilder } from '../builder';
+import { ChainedCommand, Params } from '../types';
 
-export function evaluate(body: string): QueryOperator {
-  return append(`EVAL ${body}`);
+const EVAL = 'EVAL';
+
+class EvalBuilder extends QueryBuilder {
+  private constructor(body: string, bindings?: Params) {
+    super();
+    this.commands.push({ command: body, bindings, type: EVAL });
+  }
+
+  public static create(body: string, bindings?: Params) {
+    return new EvalBuilder(body, bindings);
+  }
+
+  public concat(body: string, bindings?: Params) {
+    this.commands.push({ command: body, bindings, type: EVAL });
+    return this;
+  }
+
+  public build(): ChainedCommand {
+    const { command, bindings } = this.buildChain();
+
+    return {
+      command: `${EVAL} ${command.replace(/\s+EVAL/g, ',')}`,
+      bindings,
+    };
+  }
+}
+
+export function evaluate(body: string, bindings?: Params) {
+  return EvalBuilder.create(body, bindings);
 }
diff --git a/packages/kbn-esql-composer/src/commands/from.test.ts b/packages/kbn-esql-composer/src/commands/from.test.ts
index e15809238aeab..a3cab67c45466 100644
--- a/packages/kbn-esql-composer/src/commands/from.test.ts
+++ b/packages/kbn-esql-composer/src/commands/from.test.ts
@@ -1,10 +1,12 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { from } from './from';
 
 describe('from', () => {
diff --git a/packages/kbn-esql-composer/src/commands/from.ts b/packages/kbn-esql-composer/src/commands/from.ts
index e82cac51ac16f..f4dc153b056d4 100644
--- a/packages/kbn-esql-composer/src/commands/from.ts
+++ b/packages/kbn-esql-composer/src/commands/from.ts
@@ -1,10 +1,12 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { createPipeline } from '../create_pipeline';
 import type { QueryPipeline } from '../types';
 import { escapeIdentifier } from '../utils/escape_identifier';
@@ -18,5 +20,6 @@ export function from(...patterns: Array<string | string[]>): QueryPipeline {
         body: `FROM ${allPatterns.map((pattern) => escapeIdentifier(pattern)).join(',')}`,
       },
     ],
+    bindings: [],
   });
 }
diff --git a/packages/kbn-esql-composer/src/commands/keep.test.ts b/packages/kbn-esql-composer/src/commands/keep.test.ts
index b95920e89fcb6..8c727b9030421 100644
--- a/packages/kbn-esql-composer/src/commands/keep.test.ts
+++ b/packages/kbn-esql-composer/src/commands/keep.test.ts
@@ -1,10 +1,12 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { from } from './from';
 import { keep } from './keep';
 
diff --git a/packages/kbn-esql-composer/src/commands/keep.ts b/packages/kbn-esql-composer/src/commands/keep.ts
index ff4c872d7f185..755817265f69f 100644
--- a/packages/kbn-esql-composer/src/commands/keep.ts
+++ b/packages/kbn-esql-composer/src/commands/keep.ts
@@ -1,19 +1,20 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { escapeIdentifier } from '../utils/escape_identifier';
 import { append } from './append';
 
 export function keep(...columns: Array<string | string[]>) {
-  return append(
-    `KEEP ${columns
-      .flatMap((column) => column)
-      .map((column) => escapeIdentifier(column))
-      .join(', ')}`
-  );
+  const command = `KEEP ${columns
+    .flatMap((column) => column)
+    .map((column) => escapeIdentifier(column))
+    .join(', ')}`;
+
+  return append({ command });
 }
diff --git a/packages/kbn-esql-composer/src/commands/limit.ts b/packages/kbn-esql-composer/src/commands/limit.ts
index 7d17f81174257..f9c061669ea99 100644
--- a/packages/kbn-esql-composer/src/commands/limit.ts
+++ b/packages/kbn-esql-composer/src/commands/limit.ts
@@ -1,13 +1,14 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { append } from './append';
 
 export function limit(value: number) {
-  return append(`LIMIT ${value}`);
+  return append({ command: `LIMIT ${value}` });
 }
diff --git a/packages/kbn-esql-composer/src/commands/sort.test.ts b/packages/kbn-esql-composer/src/commands/sort.test.ts
index 20504fb39a7bc..e6be6c14a881c 100644
--- a/packages/kbn-esql-composer/src/commands/sort.test.ts
+++ b/packages/kbn-esql-composer/src/commands/sort.test.ts
@@ -1,32 +1,67 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { from } from './from';
-import { sort, SortOrder } from './sort';
+import { sort, SortOrder, sortRaw } from './sort';
 
 describe('sort', () => {
   const source = from('logs-*');
 
   it('handles single strings', () => {
-    expect(source.pipe(sort('@timestamp', 'log.level')).asString()).toEqual(
-      'FROM `logs-*`\n\t| SORT @timestamp ASC, log.level ASC'
-    );
+    const pipeline = source.pipe(sort('@timestamp', 'log.level'));
+    expect(pipeline.asString()).toEqual('FROM `logs-*`\n\t| SORT @timestamp ASC, log.level ASC');
+    expect(pipeline.getBindings()).toEqual([]);
   });
 
-  it('handles an array of strings', () => {
-    expect(source.pipe(sort(['@timestamp', 'log.level'])).asString()).toEqual(
-      'FROM `logs-*`\n\t| SORT @timestamp ASC, log.level ASC'
-    );
+  it('handles SORT with SortOrder', () => {
+    const pipeline = source.pipe(sort({ '@timestamp': SortOrder.Desc }));
+    expect(pipeline.asString()).toEqual('FROM `logs-*`\n\t| SORT @timestamp DESC');
+    expect(pipeline.getBindings()).toEqual([]);
+  });
+
+  it('handles a mix of strings and SortOrder instructions', () => {
+    const pipeline = source.pipe(sort('@timestamp', { 'log.level': SortOrder.Desc }));
+
+    expect(pipeline.asString()).toEqual('FROM `logs-*`\n\t| SORT @timestamp ASC, log.level DESC');
+    expect(pipeline.getBindings()).toEqual([]);
+  });
+
+  it('handles nested sort arrays', () => {
+    const pipeline = source.pipe(sort(['@timestamp', { 'log.level': SortOrder.Asc }]));
+    expect(pipeline.asString()).toEqual('FROM `logs-*`\n\t| SORT @timestamp ASC, log.level ASC');
+    expect(pipeline.getBindings()).toEqual([]);
   });
 
-  it('handles sort instructions', () => {
-    expect(source.pipe(sort({ '@timestamp': SortOrder.Desc })).asString()).toEqual(
-      'FROM `logs-*`\n\t| SORT @timestamp DESC'
+  it('handles SORT with params', () => {
+    const pipeline = source.pipe(
+      sortRaw('?timestamp DESC, ?logLevel ASC', {
+        env: {
+          identifier: '@timestamp',
+        },
+        logLevel: {
+          identifier: 'log.level',
+        },
+      })
     );
+
+    expect(pipeline.asString()).toEqual('FROM `logs-*`\n\t| SORT ?timestamp DESC, ?logLevel ASC');
+    expect(pipeline.getBindings()).toEqual([
+      {
+        env: {
+          identifier: '@timestamp',
+        },
+      },
+      {
+        logLevel: {
+          identifier: 'log.level',
+        },
+      },
+    ]);
   });
 });
diff --git a/packages/kbn-esql-composer/src/commands/sort.ts b/packages/kbn-esql-composer/src/commands/sort.ts
index 18dc0c44e9097..d353978da1b3c 100644
--- a/packages/kbn-esql-composer/src/commands/sort.ts
+++ b/packages/kbn-esql-composer/src/commands/sort.ts
@@ -1,12 +1,13 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { QueryOperator } from '../types';
+import { NamedParameterWithIdentifier, QueryOperator } from '../types';
 import { append } from './append';
 
 export enum SortOrder {
@@ -16,7 +17,14 @@ export enum SortOrder {
 
 type Sort = Record<string, SortOrder>;
 
-export function sort(...sorts: Array<string | Sort | Array<string | Sort>>): QueryOperator {
+type SortArgs = Sort | string | Array<Sort | string>;
+
+// TODO: a better name?
+export function sortRaw(body: string, bindings?: NamedParameterWithIdentifier): QueryOperator {
+  return append({ command: `SORT ${body}`, bindings });
+}
+
+export function sort(...sorts: SortArgs[]): QueryOperator {
   const allSorts = sorts
     .flatMap((sortInstruction) => sortInstruction)
     .map((sortInstruction): { column: string; order: 'ASC' | 'DESC' } => {
@@ -31,9 +39,9 @@ export function sort(...sorts: Array<string | Sort | Array<string | Sort>>): Que
       };
     });
 
-  return append(
-    `SORT ${allSorts
-      .map((sortInstruction) => `${sortInstruction.column} ${sortInstruction.order}`)
-      .join(', ')}`
-  );
+  const command = `SORT ${allSorts
+    .map((sortInstruction) => `${sortInstruction.column} ${sortInstruction.order}`)
+    .join(', ')}`;
+
+  return append({ command });
 }
diff --git a/packages/kbn-esql-composer/src/commands/stats.test.ts b/packages/kbn-esql-composer/src/commands/stats.test.ts
new file mode 100644
index 0000000000000..aa50e72896c22
--- /dev/null
+++ b/packages/kbn-esql-composer/src/commands/stats.test.ts
@@ -0,0 +1,134 @@
+/*
+ * 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 { from } from './from';
+import { stats } from './stats';
+
+describe('stats', () => {
+  const source = from('logs-*');
+
+  it('handles a basic STATS command', () => {
+    const pipeline = source.pipe(
+      stats('avg_duration = AVG(transaction.duration.us) BY service.name')
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| STATS avg_duration = AVG(transaction.duration.us) BY service.name'
+    );
+    expect(pipeline.getBindings()).toEqual([]);
+  });
+
+  it('handles STATS with bindings', () => {
+    const pipeline = source.pipe(
+      stats('AVG(?duration), COUNT(?svcName) BY ?env', {
+        duration: {
+          identifier: 'transaction.duration.us',
+        },
+        svcName: {
+          identifier: 'service.name',
+        },
+        env: {
+          identifier: 'service.environment',
+        },
+      })
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| STATS AVG(?duration), COUNT(?svcName) BY ?env'
+    );
+    expect(pipeline.getBindings()).toEqual([
+      {
+        duration: {
+          identifier: 'transaction.duration.us',
+        },
+      },
+      {
+        svcName: {
+          identifier: 'service.name',
+        },
+      },
+      {
+        env: {
+          identifier: 'service.environment',
+        },
+      },
+    ]);
+  });
+
+  it('handles STATS with WHERE and BY', () => {
+    const pipeline = source.pipe(
+      stats('avg_duration = AVG(transaction.duration.us)')
+        .where({
+          'log.level': 'error',
+        })
+        .by('service.name')
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| STATS avg_duration = AVG(transaction.duration.us) WHERE log.level == ? BY service.name'
+    );
+    expect(pipeline.getBindings()).toEqual(['error']);
+  });
+
+  it('handles STATS and BY with bindings', () => {
+    const pipeline = source.pipe(
+      stats('avg_duration = AVG(transaction.duration.us)').by('?svcName', {
+        svcName: { identifier: 'service.name' },
+      })
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| STATS avg_duration = AVG(transaction.duration.us) BY ?svcName'
+    );
+    expect(pipeline.getBindings()).toEqual([
+      {
+        svcName: {
+          identifier: 'service.name',
+        },
+      },
+    ]);
+  });
+
+  it('handles STATS and BY with multiple fields', () => {
+    const pipeline = source.pipe(
+      stats('avg_duration = AVG(transaction.duration.us)').by(['?svcName', '?svcEnv'], {
+        svcName: { identifier: 'service.name' },
+        svcEnv: { identifier: 'service.environment' },
+      })
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| STATS avg_duration = AVG(transaction.duration.us) BY ?svcName, ?svcEnv'
+    );
+    expect(pipeline.getBindings()).toEqual([
+      {
+        svcName: {
+          identifier: 'service.name',
+        },
+      },
+      {
+        svcEnv: {
+          identifier: 'service.environment',
+        },
+      },
+    ]);
+  });
+
+  it('handles multiple chained STATS', () => {
+    const pipeline = source.pipe(
+      stats('avg_duration = AVG(transaction.duration.us)')
+        .concat('max_duration = MAX(transaction.duration.us)')
+        .where('@timestamp > ?', '2021-01-01')
+        .concat('min_duration = MIN(transaction.duration.us)')
+        .where({
+          'service.name': 'service2',
+        })
+        .by('service.environment')
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| STATS avg_duration = AVG(transaction.duration.us), max_duration = MAX(transaction.duration.us) WHERE @timestamp > ?, min_duration = MIN(transaction.duration.us) WHERE service.name == ? BY service.environment'
+    );
+    expect(pipeline.getBindings()).toEqual(['2021-01-01', 'service2']);
+  });
+});
diff --git a/packages/kbn-esql-composer/src/commands/stats.ts b/packages/kbn-esql-composer/src/commands/stats.ts
index 4c91c0fd1dff0..3ac6e1ec3200e 100644
--- a/packages/kbn-esql-composer/src/commands/stats.ts
+++ b/packages/kbn-esql-composer/src/commands/stats.ts
@@ -1,13 +1,64 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { append } from './append';
+import { QueryBuilder } from '../builder';
+import { ChainedCommand, NamedParameterWithIdentifier } from '../types';
+import { where, WhereBuilder } from './where';
 
-export function stats(body: string) {
-  return append(`STATS ${body}`);
+const STATS = 'STATS';
+
+type StatsAfterWhere = Omit<StatsBuilder, 'where'>;
+type StatsAfterBy = Omit<StatsBuilder, 'where' | 'by' | 'concat'>;
+class StatsBuilder extends QueryBuilder {
+  private constructor(body: string, bindings?: NamedParameterWithIdentifier) {
+    super();
+    this.commands.push({ command: body, bindings, type: STATS });
+  }
+
+  public static create(body: string, bindings?: NamedParameterWithIdentifier) {
+    return new StatsBuilder(body, bindings);
+  }
+
+  public concat(body: string, bindings?: NamedParameterWithIdentifier) {
+    this.commands.push({ command: body, bindings, type: STATS });
+    return this;
+  }
+
+  public by(criteria: string | string[], bindings?: NamedParameterWithIdentifier) {
+    this.commands.push({
+      command: Array.isArray(criteria) ? criteria.join(', ') : criteria,
+      bindings,
+      type: 'BY',
+    });
+
+    return this as StatsAfterBy;
+  }
+
+  public where(...args: Parameters<typeof where>) {
+    this.commands.push({
+      command: () => WhereBuilder.create(...args),
+      type: 'WHERE',
+    });
+
+    return this as StatsAfterWhere;
+  }
+
+  public build(): ChainedCommand {
+    const { command, bindings } = this.buildChain();
+
+    return {
+      command: `${STATS} ${command.replace(/\s+STATS/g, ',')}`,
+      bindings,
+    };
+  }
+}
+
+export function stats(body: string, bindings?: NamedParameterWithIdentifier) {
+  return StatsBuilder.create(body, bindings);
 }
diff --git a/packages/kbn-esql-composer/src/commands/where.test.ts b/packages/kbn-esql-composer/src/commands/where.test.ts
new file mode 100644
index 0000000000000..81df38ec91263
--- /dev/null
+++ b/packages/kbn-esql-composer/src/commands/where.test.ts
@@ -0,0 +1,159 @@
+/*
+ * 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 { where } from './where';
+import { from } from './from';
+
+describe('where', () => {
+  const source = from('logs-*');
+
+  it('appends a basic WHERE clause', () => {
+    const pipeline = source.pipe(
+      where(`@timestamp <= NOW() AND @timestamp > NOW() - 24 hours`),
+      where('log.level', '==', 'error')
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| WHERE @timestamp <= NOW() AND @timestamp > NOW() - 24 hours\n\t| WHERE log.level == ?'
+    );
+    expect(pipeline.getBindings()).toEqual(['error']);
+  });
+
+  it('appends a WHERE clause with positional parameters', () => {
+    const pipeline = source.pipe(where('host.id == ?', 'host'));
+    expect(pipeline.asString()).toEqual('FROM `logs-*`\n\t| WHERE host.id == ?');
+    expect(pipeline.getBindings()).toEqual(['host']);
+  });
+
+  it('appends a WHERE clause with named parameters', () => {
+    const params = { hostName: 'host', serviceName: 'service' };
+    const pipeline = source.pipe(
+      where('host.name = ?hostName AND service.name = ?serviceName', params)
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| WHERE host.name = ?hostName AND service.name = ?serviceName'
+    );
+    expect(pipeline.getBindings()).toEqual([{ hostName: 'host' }, { serviceName: 'service' }]);
+  });
+
+  it('handles WHERE clause with IN operator and positional parameters', () => {
+    const hosts = ['host1', 'host2', 'host3'];
+    const pipeline = source.pipe(where(`host.name IN (${hosts.map(() => '?').join(',')})`, hosts));
+    expect(pipeline.asString()).toEqual('FROM `logs-*`\n\t| WHERE host.name IN (?,?,?)');
+    expect(pipeline.getBindings()).toEqual(['host1', 'host2', 'host3']);
+  });
+
+  it('handles nested WHERE clauses with OR conditions', () => {
+    const pipeline = source.pipe(
+      where(() => where('host.name == ?', 'host4').or('host.name', '==', 'host5'))
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| WHERE (host.name == ? OR host.name == ?)'
+    );
+    expect(pipeline.getBindings()).toEqual(['host4', 'host5']);
+  });
+
+  it('handles nested WHERE clauses with AND and OR combinations', () => {
+    const hosts = ['host1', 'host2', 'host3'];
+    const pipeline = source.pipe(
+      where(() => where('host.name == ?', 'host4').or('host.name', '==', 'host5')).and(
+        `host.name IN (${hosts.map(() => '?').join(',')})`,
+        hosts
+      )
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| WHERE (host.name == ? OR host.name == ?) AND host.name IN (?,?,?)'
+    );
+    expect(pipeline.getBindings()).toEqual(['host4', 'host5', 'host1', 'host2', 'host3']);
+  });
+
+  it('handles multiple nested WHERE clauses', () => {
+    const pipeline = source.pipe(
+      where(() => where('host.name == ?', 'host4').or('host.name', '==', 'host5')).and(
+        'service.name == ?',
+        'service1'
+      )
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| WHERE (host.name == ? OR host.name == ?) AND service.name == ?'
+    );
+    expect(pipeline.getBindings()).toEqual(['host4', 'host5', 'service1']);
+  });
+
+  it('handles empty parameters gracefully', () => {
+    const pipeline = source.pipe(where('host.name == ?', ''));
+    expect(pipeline.asString()).toEqual('FROM `logs-*`\n\t| WHERE host.name == ?');
+    expect(pipeline.getBindings()).toEqual(['']);
+  });
+
+  it('handles deeply nested queries', () => {
+    const pipeline = source.pipe(
+      where(() => where('host.name == ?', 'host1').or('host.name == ?', 'host2')).and(() =>
+        where('service.name == ?', 'service1').or('service.name == ?', 'service2')
+      )
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| WHERE (host.name == ? OR host.name == ?) AND (service.name == ? OR service.name == ?)'
+    );
+    expect(pipeline.getBindings()).toEqual(['host1', 'host2', 'service1', 'service2']);
+  });
+
+  it('handles WHERE clause with object', () => {
+    const pipeline = source.pipe(where({ 'host.name': 'host', 'service.name': 'service' }));
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| WHERE host.name == ? AND service.name == ?'
+    );
+    expect(pipeline.getBindings()).toEqual(['host', 'service']);
+  });
+
+  it('handles WHERE clause with object with  AND and OR combinations', () => {
+    const pipeline = source.pipe(
+      where({
+        'log.level': 'error',
+      })
+        .and(() => where('host.name == ?', 'host4').or('host.name', '==', 'host5'))
+        .and({
+          'log.message': 'error message',
+        })
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| WHERE log.level == ? AND (host.name == ? OR host.name == ?) AND log.message == ?'
+    );
+    expect(pipeline.getBindings()).toEqual(['error', 'host4', 'host5', 'error message']);
+  });
+
+  it('handles WHERE clause with with many OR groups and nested conditions', () => {
+    const pipeline = source.pipe(
+      where('host.name == ?', 'host2')
+        .and(() =>
+          where({ 'log.level': 'warning' })
+            .or({ 'log.message': 'debug' })
+            .or({ 'log.message': 'info' })
+            .or({ 'log.level': 'error' })
+        )
+        .or(() =>
+          where(() => where('host.name == ?', 'host1').or('host.name == ?', 'host2')).and(() =>
+            where('service.name == ?', 'service1').or('service.name == ?', 'service2')
+          )
+        )
+    );
+    expect(pipeline.asString()).toEqual(
+      'FROM `logs-*`\n\t| WHERE host.name == ? AND (log.level == ? OR log.message == ? OR log.message == ? OR log.level == ?) OR ((host.name == ? OR host.name == ?) AND (service.name == ? OR service.name == ?))'
+    );
+    expect(pipeline.getBindings()).toEqual([
+      'host2',
+      'warning',
+      'debug',
+      'info',
+      'error',
+      'host1',
+      'host2',
+      'service1',
+      'service2',
+    ]);
+  });
+});
diff --git a/packages/kbn-esql-composer/src/commands/where.ts b/packages/kbn-esql-composer/src/commands/where.ts
index 683c663869ae4..1c2dde757a150 100644
--- a/packages/kbn-esql-composer/src/commands/where.ts
+++ b/packages/kbn-esql-composer/src/commands/where.ts
@@ -1,14 +1,130 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { QueryOperator } from '../types';
-import { append } from './append';
+import { QueryBuilder } from '../builder';
+import { Params, FieldValue, ChainedCommand } from '../types';
 
-export function where(body: string): QueryOperator {
-  return append(`WHERE ${body}`);
+const operators = ['==', '>', '<', '!=', '>=', '<='] as const;
+
+const WHERE = 'WHERE';
+
+type WhereCriteria = string | (() => WhereBuilder) | Record<string, FieldValue>;
+type Operators = (typeof operators)[number];
+type LogicalOperator = 'AND' | 'OR';
+
+function isOperator(value?: Params | Operators): value is Operators {
+  return !!value && operators.includes(value as Operators);
+}
+
+export class WhereBuilder extends QueryBuilder {
+  private constructor(
+    criteria: WhereCriteria,
+    operatorOrBindings?: Operators | Params,
+    params?: Params
+  ) {
+    super();
+    this.push(criteria, operatorOrBindings, params);
+  }
+
+  public static create(
+    criteria: WhereCriteria,
+    operatorOrBindings?: Operators | Params,
+    params?: Params
+  ) {
+    return new WhereBuilder(criteria, operatorOrBindings, params);
+  }
+
+  public and(
+    criteria: WhereCriteria,
+    operatorOrParams?: Operators | Params,
+    bindings?: Params
+  ): WhereBuilder {
+    return this.addCondition('AND', criteria, operatorOrParams, bindings);
+  }
+
+  public or(
+    criteria: WhereCriteria,
+    operatorOrParams?: Operators | Params,
+    bindings?: Params
+  ): WhereBuilder {
+    return this.addCondition('OR', criteria, operatorOrParams, bindings);
+  }
+
+  public build(): ChainedCommand {
+    const { command, bindings } = this.buildChain();
+
+    return {
+      command: `${WHERE} ${command}`,
+      bindings,
+    };
+  }
+
+  private addCondition(
+    logicalOperator: LogicalOperator,
+    body: WhereCriteria,
+    operatorOrParams?: Operators | Params,
+    bindings?: Params
+  ): WhereBuilder {
+    return this.push(body, operatorOrParams, bindings, logicalOperator);
+  }
+
+  private push(
+    criteria: WhereCriteria,
+    operatorOrBindings?: Operators | Params,
+    params?: Params,
+    type: LogicalOperator | typeof WHERE = 'WHERE'
+  ) {
+    if (typeof criteria === 'function') {
+      // Handle nested conditions
+      this.commands.push({
+        command: criteria,
+        type,
+        nested: true,
+      });
+    } else if (typeof criteria === 'object') {
+      // Handle object with named parameters
+      const keys = Object.keys(criteria);
+      this.commands.push({
+        command: keys.map((key) => `${key} == ?`).join(' AND '),
+        bindings: keys.map((key) => criteria[key]),
+        type,
+      });
+    } else if (
+      !Array.isArray(operatorOrBindings) &&
+      isOperator(operatorOrBindings) &&
+      params !== undefined
+    ) {
+      // Handle explicit operator and parameter
+      const operator = operatorOrBindings;
+      this.commands.push({
+        command: `${criteria} ${operator} ?`,
+        bindings: params,
+        type,
+      });
+    } else {
+      // Handle raw and nested conditions
+      const bindings = operatorOrBindings;
+      this.commands.push({
+        command: criteria,
+        bindings,
+        type,
+      });
+    }
+
+    return this;
+  }
+}
+
+export function where(
+  criteria: WhereCriteria,
+  operatorOrParams?: Operators | Params,
+  bindings?: Params
+): WhereBuilder {
+  return WhereBuilder.create(criteria, operatorOrParams, bindings);
 }
diff --git a/packages/kbn-esql-composer/src/create_pipeline.ts b/packages/kbn-esql-composer/src/create_pipeline.ts
index 76b861e55e18d..ee88ec2e0ef77 100644
--- a/packages/kbn-esql-composer/src/create_pipeline.ts
+++ b/packages/kbn-esql-composer/src/create_pipeline.ts
@@ -1,17 +1,29 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { Query, QueryPipeline } from './types';
+import { isObject } from 'lodash';
+import { Query, QueryBuilderToOperator, QueryOperator, QueryPipeline } from './types';
+
+function isQueryBuilderOperator(
+  value: QueryOperator | QueryBuilderToOperator
+): value is QueryBuilderToOperator {
+  return (value as QueryBuilderToOperator).toQueryOperator !== undefined;
+}
 
 export function createPipeline(source: Query): QueryPipeline {
   return {
     pipe: (...operators) => {
       const nextSource = operators.reduce((previousQuery, operator) => {
+        if (isQueryBuilderOperator(operator)) {
+          return operator.toQueryOperator()(previousQuery);
+        }
+
         return operator(previousQuery);
       }, source);
 
@@ -20,5 +32,19 @@ export function createPipeline(source: Query): QueryPipeline {
     asString: () => {
       return source.commands.map((command) => command.body).join('\n\t| ');
     },
+    getBindings: () => {
+      return source.bindings.flatMap((binding) => {
+        if (isObject(binding)) {
+          return Object.entries(binding).map(([key, value]) => ({
+            [key]: value,
+          }));
+        }
+        if (Array.isArray(binding)) {
+          return binding.map((p) => p);
+        }
+
+        return binding;
+      });
+    },
   };
 }
diff --git a/packages/kbn-esql-composer/src/index.test.ts b/packages/kbn-esql-composer/src/index.test.ts
index b3df929e24550..bb51e97230fed 100644
--- a/packages/kbn-esql-composer/src/index.test.ts
+++ b/packages/kbn-esql-composer/src/index.test.ts
@@ -1,25 +1,70 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { from } from './commands/from';
+import { keep } from './commands/keep';
+import { SortOrder, sort } from './commands/sort';
 import { stats } from './commands/stats';
 import { where } from './commands/where';
 
 describe('composer', () => {
+  const source = from('logs-*');
+
   it('applies operators in order', () => {
-    expect(
-      from('logs-*')
-        .pipe(
-          where(`@timestamp <= NOW() AND @timestamp > NOW() - 24 hours`),
-          stats(`avg_duration = AVG(transaction.duration.us) BY service.name`)
-        )
-        .asString()
-    ).toEqual(
+    const pipeline = source.pipe(
+      where(`@timestamp <= NOW() AND @timestamp > NOW() - 24 hours`),
+      stats(`avg_duration = AVG(transaction.duration.us) BY service.name`)
+    );
+
+    expect(pipeline.asString()).toEqual(
       `FROM \`logs-*\`\n\t| WHERE @timestamp <= NOW() AND @timestamp > NOW() - 24 hours\n\t| STATS avg_duration = AVG(transaction.duration.us) BY service.name`
     );
   });
+
+  it('applies many operators', () => {
+    const pipeline = source.pipe(
+      where(`@timestamp <= NOW() AND @timestamp > NOW() - 24 hours`),
+      stats(`avg_duration = AVG(transaction.duration.us) BY service.name`),
+      keep('@timestamp', 'avg_duration', 'service.name'),
+      sort('avg_duration', { '@timestamp': SortOrder.Desc })
+    );
+
+    expect(pipeline.asString()).toEqual(
+      `FROM \`logs-*\`\n\t| WHERE @timestamp <= NOW() AND @timestamp > NOW() - 24 hours\n\t| STATS avg_duration = AVG(transaction.duration.us) BY service.name\n\t| KEEP \`@timestamp\`, \`avg_duration\`, \`service.name\`\n\t| SORT avg_duration ASC, @timestamp DESC`
+    );
+  });
+
+  it('applies many operators with fluent', () => {
+    const pipeline = source.pipe(
+      where('host.name == ?', 'host2')
+        .and(() =>
+          where({ 'log.level': 'warning' })
+            .or({ 'log.message': 'debug' })
+            .or({ 'log.message': 'info' })
+            .or({ 'log.level': 'error' })
+        )
+        .or(() =>
+          where(() => where('host.name == ?', 'host1').or('host.name == ?', 'host2')).and(() =>
+            where('service.name == ?', 'service1').or('service.name == ?', 'service2')
+          )
+        ),
+      stats('avg_duration = AVG(transaction.duration.us)')
+        .where({
+          'log.level': 'error',
+        })
+        .concat('min_duration = MIN(transaction.duration.us)')
+        .by('service.name'),
+      sort('avg_duration', { '@timestamp': SortOrder.Desc })
+    );
+
+    expect(pipeline.asString()).toEqual(
+      `FROM \`logs-*\`\n\t| WHERE host.name == ? AND (log.level == ? OR log.message == ? OR log.message == ? OR log.level == ?) OR ((host.name == ? OR host.name == ?) AND (service.name == ? OR service.name == ?))\n\t| STATS avg_duration = AVG(transaction.duration.us) WHERE log.level == ?, min_duration = MIN(transaction.duration.us) BY service.name\n\t| SORT avg_duration ASC, @timestamp DESC`
+    );
+  });
 });
diff --git a/packages/kbn-esql-composer/src/types.ts b/packages/kbn-esql-composer/src/types.ts
index 9f0fe693b0100..b704d85a4d032 100644
--- a/packages/kbn-esql-composer/src/types.ts
+++ b/packages/kbn-esql-composer/src/types.ts
@@ -1,21 +1,50 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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 { QueryBuilder } from './builder';
+
 export interface Command {
   body: string;
 }
 
+// Replace
+export type FieldValue = number | string | boolean | null;
+export type NamedParameterWithIdentifier = Record<
+  string,
+  { identifier: string } | { pattern: string }
+>;
+export type NamedParameter = Record<string, string> | NamedParameterWithIdentifier;
+
+export type Params = NamedParameter | FieldValue | Array<FieldValue | NamedParameter>;
 export interface QueryPipeline {
-  pipe: (...args: QueryOperator[]) => QueryPipeline;
+  pipe: (...args: Array<QueryOperator | QueryBuilderToOperator>) => QueryPipeline;
   asString: () => string;
+  getBindings: () => Params[];
 }
 export interface Query {
   commands: Command[];
+  bindings: Params[];
 }
 
 export type QueryOperator = (sourceQuery: Query) => Query;
+export interface BuilderCommand<TType extends string = string> {
+  command: string | (() => QueryBuilder);
+  bindings?: Params;
+  type: TType;
+  nested?: boolean;
+}
+
+export interface ChainedCommand {
+  command: string;
+  bindings?: Params;
+}
+
+export interface QueryBuilderToOperator {
+  toQueryOperator(): QueryOperator;
+}
diff --git a/packages/kbn-esql-composer/src/utils/escape_identifier.ts b/packages/kbn-esql-composer/src/utils/escape_identifier.ts
index c0ca065198489..12e3ab7099299 100644
--- a/packages/kbn-esql-composer/src/utils/escape_identifier.ts
+++ b/packages/kbn-esql-composer/src/utils/escape_identifier.ts
@@ -1,9 +1,10 @@
 /*
  * 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 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 or the Server
- * Side Public License, v 1.
+ * 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".
  */
 
 export function escapeIdentifier(identifier: string) {