@@ -11,6 +11,9 @@ edition = "2021"
 asap-types = { path = "../types" }
 asap-sql-function-catalog = { path = "../sql-function-catalog" }
 datafusion = "43"
+# `AggIntent::Extension.payload` for ClickHouse's argMax/argMin (issue #232)
+# -- opaque JSON, matching how `asap-types` itself carries it.
+serde_json = "1"
 
 [dev-dependencies]
 tokio = { version = "1", features = ["rt", "macros", "rt-multi-thread"] }
 
@@ -1063,6 +1063,10 @@ impl FunctionRewrite for ClickHouseBuiltinRewrite {
             return Ok(Transformed::no(Expr::AggregateFunction(f)));
         };
         let rewritten = match builtin.rewrite {
+            // No native DataFusion shape to become — leave the call exactly
+            // as DataFusion's planner parsed it. `lower_agg_intent` handles
+            // the ClickHouse name (`argMax`/`argMin`) directly (issue #232).
+            RewriteKind::PassThrough => return Ok(Transformed::no(Expr::AggregateFunction(f))),
             // `f(args...)` -> `count(args...) DISTINCT` — `lower_agg_intent`
             // already maps `count` + `DISTINCT` to `AggIntent::Cardinality`.
             RewriteKind::CountDistinct => AggregateFunction::new_udf(
@@ -1120,6 +1124,14 @@ fn lower_agg_intent(expr: &Expr) -> Result<AggIntent<ColumnRef>, LoweringError>
         Expr::Alias(a) => lower_agg_intent(&a.expr),
         Expr::AggregateFunction(agg_fn) => {
             let name = agg_fn.func.name().to_lowercase();
+            // ClickHouse's row-selecting `argMax`/`argMin` — `RewriteKind::
+            // PassThrough` in the catalog, so the call reaches here under its
+            // own name rather than a native DataFusion aggregate. Handled
+            // before the `NATIVE_FUNCTIONS` lookup below since neither name
+            // is in that table (issue #232).
+            if let Some(intent) = lower_arg_selector(&name, &agg_fn.args)? {
+                return Ok(intent);
+            }
             let semantic = asap_sql_function_catalog::lookup_native(&name)
                 .ok_or_else(|| LoweringError::UnsupportedAggregate(name.clone()))?;
             // The canonical intent algebra has no DISTINCT modifier for the
@@ -1194,6 +1206,62 @@ fn lower_agg_intent(expr: &Expr) -> Result<AggIntent<ColumnRef>, LoweringError>
     }
 }
 
+/// ClickHouse's row-selecting `argMax(arg, val)` / `argMin(arg, val)` —
+/// "return `arg`'s value from the row where `val` is maximal/minimal".
+/// `Some(name)` for `"argmax"`/`"argmin"`, `None` for every other name (the
+/// caller falls through to the ordinary `NATIVE_FUNCTIONS` path).
+///
+/// Unlike every existing `AggIntent` reducer (`Sum`/`Min`/`Max`/`Avg`/…),
+/// which folds *one* column to a value derived from itself, this is a
+/// two-column, row-selecting aggregate: it returns a *different* column's
+/// value, selected by which row maximizes/minimizes a second column. No
+/// existing `AggIntent` shape fits, and — per its own doc comment's
+/// "core only grows for intents ≥2 deployment models actually use" bar —
+/// a repo-wide search (PromQL front end, the other SQL dialects, docs) found
+/// no second deployment model wanting this shape, so this lowers to
+/// `AggIntent::Extension` rather than earning a first-class `ArgMax`/`ArgMin`
+/// core variant (issue #232). Core treats `Extension` opaquely: both columns
+/// are kept only as validated bare-column names in `payload` (`reducer_col`'s
+/// same "no expression arguments" rule, issue #115) — they are **not** run
+/// through `resolve_agg_intent`'s positional `ColumnRef` -> `ColumnId`
+/// binding the way a real reducer's `col` is, since `Extension` carries no
+/// typed column field for core to resolve. A deployment model that wants to
+/// actually bind/execute `argMax`/`argMin` is expected to re-derive that
+/// itself from `payload`, per `AggIntent::Extension`'s own doc.
+///
+/// Known scope gap, not chased down here (matching issue #230's precedent
+/// for other structural-only gaps like `splitByChar` array-indexing):
+/// `DerivedCols::rewrite_agg` (this module) only materializes/passes through
+/// an aggregate's *first* argument when deriving columns beneath the
+/// `Aggregate` node, so `val` here would be silently dropped from a `Project`
+/// DataFusion's planner inserts for some *other* reason in the same query
+/// (e.g. a sibling `SUM(a * b)`). Every corpus `argMax` use has both
+/// arguments as bare columns of the scanned table and no such `Project` is
+/// inserted, so this does not affect `bgp_jan2024_workload` today.
+fn lower_arg_selector(
+    name: &str,
+    args: &[Expr],
+) -> Result<Option<AggIntent<ColumnRef>>, LoweringError> {
+    let ext_kind = match name {
+        "argmax" => "arg_max",
+        "argmin" => "arg_min",
+        _ => return Ok(None),
+    };
+    let [arg, val] = args else {
+        unreachable!(
+            "{name}'s stub signature (asap_sql_function_catalog::CLICKHOUSE_BUILTINS) fixes \
+             its arity at 2 -- the planner already rejected any other argument count before \
+             lower_agg_intent runs"
+        );
+    };
+    let arg_col = reducer_col(name, std::slice::from_ref(arg))?;
+    let val_col = reducer_col(name, std::slice::from_ref(val))?;
+    Ok(Some(AggIntent::Extension {
+        ext_kind: ext_kind.to_string(),
+        payload: serde_json::json!({ "arg_col": arg_col, "val_col": val_col }),
+    }))
+}
+
 /// The name an `IN (subquery)`'s key column is projected under, so the join
 /// predicate cannot bind it to a same-named column of the outer relation.
 const IN_SUBQUERY_KEY: &str = "__asap_in_key";
 
@@ -182,8 +182,13 @@ async fn corpus_lowering_matches_the_pinned_aggregate_tally() {
     // caveat" section) -- getting these functions' *names* to lower to a
     // structurally correct `FunctionCall` node is what's in scope here, not
     // array/map indexing or interval-literal conversion.
-    expect(Category::Lowered, 145);
-    expect(Category::Plan, 40);
+    // `argMax` support (issue #232 -- `AggIntent::Extension`, catalog-driven
+    // `RewriteKind::PassThrough`) clears the "unknown function: argmax" `Plan`
+    // failure for all 3 corpus occurrences: every one has both arguments as
+    // bare columns of the scanned table, so all 3 lower end to end with no
+    // companion gap (unlike `splitByChar`'s array-indexing gap, #230).
+    expect(Category::Lowered, 148);
+    expect(Category::Plan, 37);
     expect(Category::Schema, 0);
     expect(Category::Parse, 0);
     // One query that used to fail at `uniqExact` (`Plan`) now clears that
 
@@ -1466,3 +1466,91 @@ async fn count_if_composes_with_group_by() {
     assert_eq!(*by, GroupKeys::by(vec![0]));
     assert!(matches!(measures.as_slice(), [AggIntent::Sum { .. }]));
 }
+
+// ── Issue #232: argMax/argMin -- AggIntent::Extension, not a first-class
+// core variant. A repo-wide search (PromQL front end, other SQL dialects,
+// docs) turned up no second deployment model wanting this two-column,
+// row-selecting shape, so per `AggIntent::Extension`'s own "core only grows
+// for intents ≥2 deployment models actually use" bar, it stays an opaque
+// `Extension` rather than a new `ArgMax`/`ArgMin` core variant. Unlike
+// `countIf`/`uniqExact`, there is no native DataFusion aggregate shape to
+// rewrite to (`RewriteKind::PassThrough`) -- `lower_agg_intent` builds the
+// `AggIntent` directly from the ClickHouse name. ─────────────────────────
+
+#[tokio::test]
+async fn arg_max_lowers_to_an_extension_intent() {
+    // No existing `AggIntent` reducer fits: every one folds one column to a
+    // value derived from itself, while `argMax(arg, val)` returns a
+    // *different* column's value, selected by which row maximizes a second.
+    let qe = lower_clickhouse(
+        "SELECT service, argMax(service, latency) AS busiest FROM metrics GROUP BY service",
+    )
+    .await;
+    let (by, measures) = find_aggregate(&qe).expect("expected an Aggregate");
+    assert_eq!(
+        *by,
+        GroupKeys::by(vec![1]),
+        "grouped by `service` (schema index 1)"
+    );
+    assert!(
+        matches!(
+            measures.as_slice(),
+            [AggIntent::Extension { ext_kind, .. }] if ext_kind == "arg_max"
+        ),
+        "expected Extension {{ ext_kind: \"arg_max\", .. }}, got {measures:?}"
+    );
+}
+
+#[tokio::test]
+async fn arg_min_lowers_to_its_own_extension_kind() {
+    let qe = lower_clickhouse("SELECT argMin(service, latency) FROM metrics").await;
+    let (by, measures) = find_aggregate(&qe).expect("expected an Aggregate");
+    assert!(by.is_empty());
+    assert!(
+        matches!(
+            measures.as_slice(),
+            [AggIntent::Extension { ext_kind, .. }] if ext_kind == "arg_min"
+        ),
+        "expected Extension {{ ext_kind: \"arg_min\", .. }}, got {measures:?}"
+    );
+}
+
+#[tokio::test]
+async fn arg_max_payload_preserves_both_column_names() {
+    // Core never resolves an `Extension`'s payload, so both columns are kept
+    // as validated bare-column `ColumnRef`s in `payload`, not run through
+    // positional `ColumnId` binding -- see `lower_arg_selector`'s doc.
+    let qe = lower_clickhouse("SELECT argMax(service, latency) AS m FROM metrics").await;
+    let (_, measures) = find_aggregate(&qe).expect("expected an Aggregate");
+    let AggIntent::Extension { payload, .. } = &measures[0] else {
+        panic!("expected an Extension intent, got {:?}", measures[0]);
+    };
+    let named = |key: &str| {
+        payload
+            .get(key)
+            .and_then(|c| c.get("Named"))
+            .and_then(|n| n.as_str())
+            .map(str::to_string)
+    };
+    assert_eq!(named("arg_col"), Some("service".to_string()));
+    assert_eq!(named("val_col"), Some("latency".to_string()));
+}
+
+#[tokio::test]
+async fn arg_max_rejects_a_non_column_argument() {
+    // Same "bare column only" rule as every other reducer (`reducer_col`,
+    // issue #115) -- an expression argument is rejected, not silently
+    // dropped or materialized into the wrong column.
+    let err = lower_sql_dialect(
+        "SELECT argMax(service, latency * 2) FROM metrics",
+        &catalog(),
+        SqlDialect::ClickhouseSQL,
+        AccuracyTarget::Exact,
+    )
+    .await
+    .expect_err("argMax over a non-column expression must be rejected");
+    assert!(
+        format!("{err}").contains("non-column expression"),
+        "got {err}"
+    );
+}
@@ -26,14 +26,18 @@
 //!   must be a bare column" rule are call-site logic, not per-function data,
 //!   and stay in `asap-frontend-sql`.
 //! - [`CLICKHOUSE_BUILTINS`] -- ClickHouse-only *aggregate* names DataFusion
-//!   doesn't know at all (`uniqExact`, `countIf`). Each entry additionally
-//!   carries a [`RewriteKind`]: the native DataFusion aggregate shape the
-//!   call rewrites to before `lower_agg_intent` (or DataFusion's own physical
-//!   planner) ever has to understand the ClickHouse name itself. This is
-//!   what generalizes `uniqExact`'s old bespoke `UniqExactRewrite` +
-//!   `uniq_exact_udaf` pair (issue #221): a new builtin that rewrites to an
-//!   already-handled shape is a new entry in this table, not a new
-//!   `FunctionRewrite` impl and a new stub-`AggregateUDF` constructor.
+//!   doesn't know at all (`uniqExact`, `countIf`, `argMax`, `argMin`). Each
+//!   entry additionally carries a [`RewriteKind`]: either the native
+//!   DataFusion aggregate shape the call rewrites to before
+//!   `lower_agg_intent` (or DataFusion's own physical planner) ever has to
+//!   understand the ClickHouse name itself -- this is what generalizes
+//!   `uniqExact`'s old bespoke `UniqExactRewrite` + `uniq_exact_udaf` pair
+//!   (issue #221): a new builtin that rewrites to an already-handled shape is
+//!   a new entry in this table, not a new `FunctionRewrite` impl and a new
+//!   stub-`AggregateUDF` constructor -- or [`RewriteKind::PassThrough`] for a
+//!   builtin with no native shape to rewrite to at all (`argMax`/`argMin`,
+//!   issue #232): the call survives unchanged and `lower_agg_intent` handles
+//!   the ClickHouse name itself directly.
 //! - [`CLICKHOUSE_SCALAR_BUILTINS`] -- ClickHouse-only *scalar* names
 //!   DataFusion doesn't know at all (`splitByChar`, `toDate`, `match`,
 //!   the `toStartOf*` family, `startsWith`, `positionCaseInsensitive`). No
@@ -229,6 +233,16 @@ pub enum RewriteKind {
     /// for any reducer over an expression (issue #110) -- so no new
     /// `AggIntent` variant or lowering path is needed either.
     CountIfToSum,
+    /// No native DataFusion aggregate shape to rewrite to at all -- the call
+    /// survives unchanged (`ClickHouseBuiltinRewrite` is a no-op for it) and
+    /// reaches `lower_agg_intent`, which builds an `AggIntent` directly from
+    /// the ClickHouse name itself. `argMax`/`argMin` (issue #232): a
+    /// two-column, row-selecting aggregate ("return `arg`'s value from the
+    /// row where `val` is maximal") that fits no existing single-column
+    /// `AggIntent` reducer shape and has no native DataFusion equivalent to
+    /// borrow one from -- unlike `CountDistinct`/`CountIfToSum`, there is no
+    /// "already-handled shape" for this rewrite to target.
+    PassThrough,
 }
 
 /// One [`CLICKHOUSE_BUILTINS`] entry.
@@ -259,6 +273,19 @@ pub const CLICKHOUSE_BUILTINS: &[ClickHouseBuiltin] = &[
         arity: Arity::Exact(1),
         rewrite: RewriteKind::CountIfToSum,
     },
+    // argMax(arg, val) -- "arg's value from the row where val is maximal".
+    // No native DataFusion aggregate shape to rewrite to (issue #232), so
+    // `lower_agg_intent` builds `AggIntent::Extension` from these directly.
+    ClickHouseBuiltin {
+        name: "argmax",
+        arity: Arity::Exact(2),
+        rewrite: RewriteKind::PassThrough,
+    },
+    ClickHouseBuiltin {
+        name: "argmin",
+        arity: Arity::Exact(2),
+        rewrite: RewriteKind::PassThrough,
+    },
 ];
 
 /// Aggregate function names DataFusion's own planner resolves out of the box