Try to make tests to pass

.clj-kondo/config/modules/config.edn

@@ -1768,6 +1768,7 @@
               metabase.query-processor.parameters.values
               metabase.query-processor.pipeline
               metabase.query-processor.pivot
+              metabase.query-processor.pivot.common
               metabase.query-processor.preprocess
               metabase.query-processor.schema
               metabase.query-processor.store

modules/drivers/redshift/src/metabase/driver/redshift.clj

@@ -45,6 +45,8 @@
 
 (doseq [[feature supported?] {:atomic-renames                   true
                               :connection-impersonation         true
+                              ;; Native pivot SQL is Postgres-only for now; inherited from :postgres parent otherwise.
+                              :native-pivot-tables              false
                               :database-routing                 true
                               :describe-default-expr            false
                               :describe-fields                  true

src/metabase/driver.clj

@@ -878,6 +878,20 @@
     ;; Does this driver support the workspace feature
     :workspace
     ;;
+    ;; Does this driver support computing pivot table subtotals in a single database-native query instead of the
+    ;; default multi-query path?
+    ;;
+    ;; Contract: the driver's SQL compilation must produce a result set with:
+    ;;   1. All breakout columns (NULLs for aggregated dimensions in subtotal rows).
+    ;;   2. A column named "pivot-grouping" containing an integer bitmask from the SQL `GROUPING()` function
+    ;;      (or equivalent), indicating which breakouts are active per row.
+    ;;   3. All aggregation columns.
+    ;;
+    ;; The QP-level logic in `metabase.query-processor.pivot/run-native-pivot-query` is driver-agnostic; only the
+    ;; SQL generation layer is driver-specific.  See `metabase.driver.postgres.pivot` for the reference
+    ;; implementation using `GROUPING SETS`.
+    :native-pivot-tables
+    ;;
     ;; Does this driver support table references in native queries -- for example, "select * from {{table}}" where
     ;; `{{table}}` gets replaced by a reference to a table.
     :parameters/table-reference})

src/metabase/driver/postgres.clj

@@ -16,6 +16,7 @@
    [metabase.driver.common :as driver.common]
    [metabase.driver.postgres.actions :as postgres.actions]
    [metabase.driver.postgres.ddl :as postgres.ddl]
+   [metabase.driver.postgres.pivot :as postgres.pivot]
    [metabase.driver.sql :as driver.sql]
    [metabase.driver.sql-jdbc :as sql-jdbc]
    [metabase.driver.sql-jdbc.common :as sql-jdbc.common]
@@ -54,6 +55,7 @@
   ;; method impls live in these namespaces.
   postgres.actions/keep-me
   postgres.ddl/keep-me
+  postgres.pivot/keep-me
   sql.pg-ops/keep-me)
 
 ;; Inherit from `::like-escape-char-built-in/like-escape-char-built-in` because Postgres's
@@ -94,9 +96,14 @@
                               :transforms/python        true
                               :transforms/index-ddl     true
                               :metadata/table-existence-check true
-                              :workspace                true}]
+                              :workspace                true
+                              :native-pivot-tables      true}]
   (defmethod driver/database-supports? [:postgres feature] [_driver _feature _db] supported?))
 
+(defmethod driver/database-supports? [:postgres-mbql5 :native-pivot-tables]
+  [_driver _feature _db]
+  false)
+
 (defmethod driver/database-supports? [:postgres :nested-field-columns]
   [_driver _feat db]
   (driver.common/json-unfolding-default db))
@@ -845,23 +852,30 @@
 ;; being described twice instead of using the alias.
 ;; Therefore, force the alias, but only for JSON fields to avoid ambiguity.
 ;; The alias names in JSON fields are unique wrt nfc path
+(defmethod sql.qp/preprocess :postgres
+  [_driver mbql5-query]
+  (let [inner-query ((get-method sql.qp/preprocess :sql) :sql mbql5-query)]
+    (postgres.pivot/merge-native-pivot-keys mbql5-query inner-query)))
+
 (defmethod sql.qp/apply-top-level-clause
   [:postgres :breakout]
   [driver clause honeysql-form {breakout-fields :breakout, _fields-fields :fields :as query}]
-  (let [stored-field-ids (map second breakout-fields)
-        stored-fields    (map #(when (integer? %)
-                                 (driver-api/field (driver-api/metadata-provider) %))
-                              stored-field-ids)
-        parent-method    (partial (get-method sql.qp/apply-top-level-clause [:sql :breakout])
-                                  driver clause honeysql-form)
-        qualified        (parent-method query)
-        unqualified      (parent-method (update query
-                                                :breakout
-                                                #(sql.qp/rewrite-fields-to-force-using-column-aliases % {:is-breakout true})))]
-    (if (some driver-api/json-field? stored-fields)
-      (merge qualified
-             (select-keys unqualified #{:group-by}))
-      qualified)))
+  (if (:qp.pivot/native-pivot? query)
+    (postgres.pivot/apply-native-pivot-breakout driver honeysql-form query)
+    (let [stored-field-ids (map second breakout-fields)
+          stored-fields    (map #(when (integer? %)
+                                   (driver-api/field (driver-api/metadata-provider) %))
+                                stored-field-ids)
+          parent-method    (partial (get-method sql.qp/apply-top-level-clause [:sql :breakout])
+                                    driver clause honeysql-form)
+          qualified        (parent-method query)
+          unqualified      (parent-method (update query
+                                                  :breakout
+                                                  #(sql.qp/rewrite-fields-to-force-using-column-aliases % {:is-breakout true})))]
+      (if (some driver-api/json-field? stored-fields)
+        (merge qualified
+               (select-keys unqualified #{:group-by}))
+        qualified))))
 
 (defmethod sql.qp/apply-top-level-clause
   [:postgres-mbql5 :breakout]

src/metabase/driver/postgres/pivot.clj

@@ -0,0 +1,178 @@
+(ns metabase.driver.postgres.pivot
+  "PostgreSQL-specific SQL generation for native pivot tables.
+
+  Fulfills the contract required by [[metabase.query-processor.pivot/run-native-pivot-query]]: when the
+  `:native-pivot-tables` feature is declared, the driver must produce a result set containing all breakout columns,
+  a `\"pivot-grouping\"` integer bitmask column, and the aggregation columns. This namespace achieves that by
+  rewriting the `GROUP BY` clause to use `GROUPING SETS` and appending a `GROUPING()` call to the `SELECT`.
+
+  This file serves as the reference implementation for adding `:native-pivot-tables` support to other drivers.
+  The `GROUPING()` function and `GROUPING SETS` clause are part of the SQL standard and are supported by
+  PostgreSQL, Redshift, MySQL 8+, Snowflake, BigQuery, Oracle, and SQL Server."
+  (:refer-clojure :exclude [mapv empty? not-empty select-keys])
+  (:require
+   [clojure.string :as str]
+   [honey.sql :as sql]
+   [honey.sql.helpers :as sql.helpers]
+   [metabase.driver :as driver]
+   [metabase.driver.sql.query-processor :as sql.qp]
+   [metabase.query-processor.pivot.common :as pivot.common]
+   [metabase.util.honey-sql-2 :as h2x]
+   [metabase.util.performance :refer [mapv empty? not-empty select-keys]]))
+
+(set! *warn-on-reflection* true)
+
+(def ^:private pivot-grouping-column-name
+  "pivot-grouping")
+
+(defn- format-grouping-fn
+  "HoneySQL formatter for the SQL `GROUPING()` function. Each arg is a separate column expression.
+  `GROUPING()` is standard SQL (not PostgreSQL-specific) and is supported by most analytical databases.
+  It returns an integer bitmask indicating which of the supplied columns are aggregated (NULL) in the current row."
+  [_fn args]
+  (let [sqls-and-params (map #(sql/format-expr % {:nested true}) args)
+        sqls            (map first sqls-and-params)
+        params          (mapcat rest sqls-and-params)]
+    (into [(str "GROUPING(" (str/join ", " sqls) ")")]
+          params)))
+
+(sql/register-fn! ::grouping-fn format-grouping-fn)
+
+(defn- grouping-fn-expr
+  "HoneySQL form for GROUPING(breakoutN, ..., breakout0).
+   Uses varargs so each column is passed as a separate arg to the formatter.
+   Includes every breakout expression so remapped columns are accounted for."
+  [breakout-exprs]
+  (into [::grouping-fn] (reverse breakout-exprs)))
+
+(defn- format-grouping-sets
+  "Format GROUP BY GROUPING SETS (...) clause. Each arg is a vector of expressions for one set."
+  [_fn args]
+  (let [format-set (fn [set-exprs]
+                     (if (empty? set-exprs)
+                       ["()"]
+                       (let [sqls-and-params (map #(sql/format-expr % {:nested true}) set-exprs)
+                             sqls            (map first sqls-and-params)
+                             params          (mapcat rest sqls-and-params)]
+                         (into [(str "(" (str/join ", " sqls) ")")]
+                               params))))
+        set-results (map format-set args)
+        sets-sqls   (map first set-results)
+        all-params  (mapcat rest set-results)]
+    (into [(str "GROUPING SETS (" (str/join ", " sets-sqls) ")")]
+          all-params)))
+
+(sql/register-fn! ::grouping-sets format-grouping-sets)
+
+(defn- grouping-set-combinations->honeysql
+  "Convert unremapped breakout combinations to HoneySQL GROUPING SET expressions."
+  [driver breakout-fields {:keys [qp.pivot/breakout-combinations
+                                  qp.pivot/remapped-indexes]}]
+  (let [breakout-exprs (mapv (partial sql.qp/->honeysql driver) breakout-fields)]
+    (mapv (fn [combination]
+            (let [full-combination (pivot.common/full-breakout-combination
+                                    {:qp.pivot/remapped-breakout-combination combination
+                                     :qp.pivot/remapped-indexes              remapped-indexes})]
+              (mapv #(nth breakout-exprs %) full-combination)))
+          breakout-combinations)))
+
+(defn- native-pivot-order-by-clauses
+  "Order-by clauses for native pivot queries: use any remaining MBQL `:order-by` on the inner query, else ASC on
+  each breakout (matches [[metabase.driver.sql.query-processor/apply-top-level-clause]] `[:sql :order-by]`)."
+  [{:keys [order-by breakout]}]
+  (or (not-empty order-by)
+      (mapv (fn [breakout-field] [:asc breakout-field]) breakout)))
+
+(defn- native-pivot-order-by->honeysql
+  "Compile a native-pivot order-by clause, using `NULLS LAST` for ASC (so subtotal/grand-total rows sort last)."
+  [driver order-by-clause]
+  (let [[expr direction :as compiled] (sql.qp/->honeysql driver order-by-clause)]
+    (if (= direction :asc)
+      [expr :asc-nulls-last]
+      compiled)))
+
+(defn- flatten-over-order-by-entry
+  "Normalize a single OVER `order-by` entry to `[expr direction]` (HoneySQL 2 shape)."
+  [entry]
+  (loop [e entry]
+    (if (and (vector? e) (= 2 (count e)) (keyword? (second e)))
+      (let [expr (first e)]
+        (if (and (vector? expr) (= 2 (count expr)) (keyword? (second expr)))
+          (recur expr)
+          [(if (vector? expr) (vec expr) expr) (second e)]))
+      e)))
+
+(defn- normalize-over-order-by-in-form
+  "Fix doubly-nested `:order-by` vectors inside `:over` expressions after the parent breakout pass."
+  [form]
+  (cond
+    (and (vector? form)
+         (= :over (first form))
+         (<= 2 (count form)))
+    (let [[tag inner & rest] form
+          inner' (if (map? inner)
+                   (update inner :order-by
+                           (fn [order-bys]
+                             (mapv flatten-over-order-by-entry order-bys)))
+                   inner)]
+      (into [tag inner'] rest))
+
+    (vector? form)
+    (mapv normalize-over-order-by-in-form form)
+
+    (map? form)
+    (into {} (map (fn [[k v]] [k (normalize-over-order-by-in-form v)]) form))
+
+    :else
+    form))
+
+(defn apply-native-pivot-breakout
+  "Apply breakout clause for native pivot queries using `GROUP BY GROUPING SETS`.
+
+  This function fulfills the native-pivot driver contract: the resulting SQL must produce a result set with:
+    1. All breakout columns (with NULLs for aggregated dimensions in subtotal rows).
+    2. A `\"pivot-grouping\"` column containing an integer bitmask from `GROUPING()`.
+    3. All aggregation columns.
+
+  It delegates to the parent `:sql` breakout method for proper `SELECT`/`GROUP BY` aliasing, then appends the
+  `GROUPING()` expression to `SELECT` and replaces the `GROUP BY` with `GROUPING SETS` derived from the breakout
+  combinations computed by [[metabase.query-processor.pivot/breakout-combinations]]."
+  [driver honeysql-form {breakout-fields :breakout, :as inner-query}]
+  (let [parent-method   (get-method sql.qp/apply-top-level-clause [:sql :breakout])
+        base-hsql       (parent-method driver :breakout honeysql-form inner-query)
+        select-fn       (if (:select-top base-hsql)
+                          sql.helpers/select-top
+                          sql.helpers/select)
+        breakout-exprs  (mapv (partial sql.qp/->honeysql driver) breakout-fields)
+        grouping-expr   (grouping-fn-expr breakout-exprs)
+        grouping-alias  (sql.qp/->honeysql driver
+                                           (h2x/identifier :field-alias
+                                                           (driver/escape-alias driver pivot-grouping-column-name)))
+        grouping-sets     (grouping-set-combinations->honeysql driver breakout-fields inner-query)
+        gs-expr           (into [::grouping-sets] grouping-sets)
+        order-by-clauses   (native-pivot-order-by-clauses inner-query)
+        honeysql-order-bys (into [[(sql.qp/compiled grouping-expr) :asc]]
+                                 (map (partial native-pivot-order-by->honeysql driver))
+                                 order-by-clauses)]
+    (let [hsql (-> base-hsql
+                   normalize-over-order-by-in-form
+                   ;; `grouping-expr` is itself a HoneySQL vector; wrap with `compiled` so `select` treats it as one
+                   ;; expression (same pattern as [[metabase.driver.sql.query-processor/apply-top-level-clause]] `[:sql :aggregation]`).
+                   (select-fn [(sql.qp/compiled grouping-expr) [grouping-alias]])
+                   (assoc :group-by [gs-expr]))]
+      (if (seq honeysql-order-bys)
+        (reduce sql.helpers/order-by hsql honeysql-order-bys)
+        hsql))))
+
+(def ^:private native-pivot-query-keys
+  #{:qp.pivot/native-pivot?
+    :qp.pivot/breakout-combinations
+    :qp.pivot/num-unremapped-breakouts
+    :qp.pivot/num-remapped-breakouts
+    :qp.pivot/num-remapped-cols
+    :qp.pivot/remapped-indexes})
+
+(defn merge-native-pivot-keys
+  "Copy native pivot keys from the outer MBQL 5 query onto the MBQL 4 inner query."
+  [mbql5-query target]
+  (merge target (select-keys mbql5-query native-pivot-query-keys)))

src/metabase/driver/sql/query_processor.clj

@@ -2083,9 +2083,11 @@
   "Return the keys present in an MBQL `inner-query` in the order they should be processed."
   [inner-query]
   ;; sort first by any known top-level clauses according to the `top-level-application-clause-order` defined above,
-  ;; then sort any unknown clauses by name.
-  (sort-by (fn [clause] [(get top-level-clause-application-order clause Integer/MAX_VALUE) clause])
-           (keys inner-query)))
+  ;; then sort any unknown clauses by name. Ignore namespaced keys (e.g. `:qp.pivot/*`) that are metadata for driver
+  ;; hooks but not top-level MBQL clauses.
+  (let [clause-keys (remove namespace (keys inner-query))]
+    (sort-by (fn [clause] [(get top-level-clause-application-order clause Integer/MAX_VALUE) clause])
+             clause-keys)))
 
 (defn- format-honeysql-2 [driver dialect honeysql-form]
   ;; make sure [[driver/*driver*]] is bound, we need it for [[sqlize-value]]

src/metabase/query_processor/pivot.clj

@@ -1,20 +1,28 @@
 (ns metabase.query-processor.pivot
-  "Pivot table query processor. Determines a bunch of different subqueries to run, then runs them one by one on the data
-  warehouse and concatenates the result rows together, sort of like the way [[clojure.core/lazy-cat]] works. This is
-  dumb, right? It's not just me? Why don't we just generate a big ol' UNION query so we can run one single query
-  instead of running like 10 separate queries? -- Cam
+  "Pivot table query processor. Two execution strategies are available:
 
-  Note that this namespace is mostly responsible for generating the series of different queries to run and doing QP
-  magic to combine the results together.
+  1. **Legacy multi-query path** ([[run-pivot-query-multi]]): generates one subquery per breakout combination
+     (powerset of pivot rows/cols), runs them sequentially, and stitches the results together with an
+     application-level `pivot-grouping` bitmask column. Works with every driver.
+
+  2. **Native single-query path** ([[run-native-pivot-query]]): for drivers that declare the
+     `:native-pivot-tables` feature, a single SQL query with `GROUPING SETS` (or the driver's equivalent) produces
+     all subtotal levels at once. The driver is responsible for returning a `\"pivot-grouping\"` column containing
+     an integer bitmask. This path is driver-agnostic at the QP level -- any driver that fulfills the contract can
+     use it.
+
+  [[run-pivot-query]] dispatches between the two paths based on [[native-pivot-supported?]].
 
   Post-processing middleware to add the `pivot-grouping` column to results and to massage result rows into a standard
   shape lives in [[metabase.query-processor.pivot.middleware]]."
   (:refer-clojure :exclude [every? mapv some select-keys update-keys empty? not-empty get-in])
   (:require
    [medley.core :as m]
+   [metabase.driver.util :as driver.u]
    [metabase.lib-be.core :as lib-be]
    [metabase.lib.core :as lib]
    [metabase.lib.equality :as lib.equality]
+   [metabase.lib.metadata :as lib.metadata]
    [metabase.lib.schema :as lib.schema]
    [metabase.lib.schema.id :as lib.schema.id]
    [metabase.lib.schema.info :as lib.schema.info]
@@ -26,6 +34,7 @@
    [metabase.query-processor.middleware.normalize-query :as qp.middleware.normalize]
    [metabase.query-processor.pipeline :as qp.pipeline]
    [metabase.query-processor.pivot.common :as pivot.common]
+   [metabase.query-processor.pivot.middleware :as pivot.middleware]
    [metabase.query-processor.reducible :as qp.reducible]
    [metabase.query-processor.schema :as qp.schema]
    [metabase.query-processor.setup :as qp.setup]
@@ -35,7 +44,9 @@
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
    [metabase.util.malli.registry :as mr]
-   [metabase.util.performance :refer [empty? every? get-in mapv not-empty select-keys some update-keys]]))
+   [metabase.util.performance :refer [empty? every? get-in mapv not-empty select-keys some update-keys]])
+  (:import
+   (java.util.concurrent ConcurrentLinkedQueue)))
 
 (set! *warn-on-reflection* true)
 
@@ -481,6 +492,10 @@
            :qp.pivot/num-remapped-breakouts   num-remapped-breakouts
            :qp.pivot/remapped-indexes         remap)))
 
+(def ^:dynamic *force-legacy-pivot*
+  "When true, [[run-pivot-query]] always uses the legacy multi-query path. Used by tests to compare native vs legacy results."
+  false)
+
 (def ^:dynamic ^:private *pivot-max-result-rows*
   "Maximum number of result rows for each pivot sub-query. Divided by the number of aggregations since each aggregation
   adds a column to the output, so fewer rows are needed to fill the pivot table."
@@ -495,9 +510,151 @@
       (quot *pivot-max-result-rows* num-aggs)
       *pivot-max-result-rows*)))
 
+(defn- has-window-fn-aggregation?
+  "Window function aggregations cannot be computed correctly in a single GROUPING SETS query; use legacy pivot."
+  [query]
+  (let [window-fn-tags #{:offset :cum-count :cum-sum}]
+    (some (fn [clause]
+            (and (vector? clause) (window-fn-tags (first clause))))
+          (lib/aggregations query))))
+
+(mu/defn native-pivot-supported? :- :boolean
+  "Does the database for this query support computing pivot subtotals in a single database-native query?
+  Returns true when the driver declares `:native-pivot-tables`. The actual SQL construct used (e.g. `GROUPING SETS`,
+  `WITH ROLLUP`) is driver-specific and opaque to this namespace."
+  [query :- [:map
+             [:database {:optional true} [:maybe ::lib.schema.id/database]]]]
+  (boolean
+   (when-let [database (u/ignore-exceptions (lib.metadata/database query))]
+     (when-let [driver (driver.u/database->driver database)]
+       (driver.u/supports? driver :native-pivot-tables database)))))
+
+(defn- find-pivot-grouping-index
+  "Return the index of the pivot-grouping column in `cols`, or nil if not found."
+  [cols]
+  (first (keep-indexed (fn [i col] (when (= (:name col) "pivot-grouping") i)) cols)))
+
+(defn- vec-remove-at
+  "Remove the element at `idx` from vector `v`."
+  [v idx]
+  (into (subvec v 0 idx) (subvec v (inc idx))))
+
+(defn- vec-insert-at
+  "Insert `val` at `idx` in vector `v`."
+  [v idx val]
+  (into (conj (subvec v 0 idx) val) (subvec v idx)))
+
+(defn- strip-pivot-grouping-from-row
+  "Remove the `pivot-grouping` column at `idx` from `row`, enqueueing its original database value."
+  [idx ^ConcurrentLinkedQueue grouping-queue row]
+  (let [row-v (vec row)]
+    (.offer grouping-queue (nth row-v idx))
+    (vec-remove-at row-v idx)))
+
+(mu/defn- run-native-pivot-query
+  "Run a pivot query using a single database-native query. This function is **driver-agnostic** -- it works for any
+  driver that declares the `:native-pivot-tables` feature. The contract is:
+
+  * The driver produces all subtotal levels in one query (e.g. via `GROUPING SETS` or equivalent).
+  * The driver returns one extra column named `\"pivot-grouping\"` containing an integer bitmask that encodes which
+    breakout columns are active for each row (see [[metabase.query-processor.pivot.common/group-bitmask]]).
+
+  **Why the `*reduce*` strip/restore dance is needed:**
+
+  The post-processing middleware chain (specifically `annotate/add-column-info`) compares driver-returned columns
+  against the columns Lib calculates from the query structure. Lib knows nothing about the synthetic
+  `pivot-grouping` column, so the count would mismatch and throw. To avoid this:
+
+  1. We bind [[qp.pipeline/*reduce*]] to intercept the driver's metadata and rows, stripping out the
+     `pivot-grouping` column before the post-processing middleware chain sees them.
+  2. We wrap the outermost `rff` so that after all middleware has processed, `pivot-grouping` is re-inserted into
+     the final metadata and every row at its original position.
+
+  Data flow:
+
+      Driver returns [breakouts... pivot-grouping aggs...]
+        -> *reduce* binding strips pivot-grouping
+        -> post-processing middleware sees [breakouts... aggs...] (matches Lib expectation)
+        -> wrapped-rff re-inserts pivot-grouping
+        -> final output: [breakouts... pivot-grouping aggs...]  (matches frontend expectation)"
+  [query        :- ::qp.schema/any-query
+   rff          :- ::qp.schema/rff
+   pivot-opts   :- ::pivot-opts
+   pivot-limit  :- [:maybe nat-int?]]
+  (let [original-query query
+        all-breakout-combinations (breakout-combinations (count (lib/breakouts query))
+                                                         (:pivot-rows pivot-opts)
+                                                         (:pivot-cols pivot-opts)
+                                                         (:show-row-totals pivot-opts)
+                                                         (:show-column-totals pivot-opts))
+        query        (-> query
+                         (assoc :qp.pivot/native-pivot? true
+                                :qp.pivot/breakout-combinations all-breakout-combinations)
+                         (assoc-in [:info :pivot/original-query] original-query)
+                         ;; GROUPING SETS returns many aggregated rows; do not apply the unaggregated row limit.
+                         (update :constraints dissoc :max-results-bare-rows)
+                         ;; A stage `:limit` applies to the single GROUPING SETS query and can drop subtotal rows;
+                         ;; legacy pivot runs one query per combination so each can return rows independently.
+                         (lib/limit nil)
+                         remove-non-aggregation-order-bys)
+        grouping-queue (ConcurrentLinkedQueue.)
+        pivot-col-info (volatile! nil)
+        row-count      (volatile! 0)
+        wrapped-rff  (fn [metadata]
+                       (if-let [{:keys [pivot-idx pivot-col]} @pivot-col-info]
+                         (let [restored-metadata (update metadata :cols #(vec-insert-at (vec %) pivot-idx pivot-col))
+                               rf (rff restored-metadata)]
+                           (fn native-pivot-restore-rf
+                             ([] (rf))
+                             ([result]
+                              (let [result (rf result)]
+                                (cond-> result
+                                  (and pivot-limit (>= @row-count pivot-limit) (map? result))
+                                  (assoc-in [:data :pivot_rows_truncated] @row-count))))
+                             ([result row]
+                              (vswap! row-count inc)
+                              (rf result
+                                  (vec-insert-at (vec row)
+                                                 pivot-idx
+                                                 (.poll grouping-queue))))))
+                         (rff metadata)))]
+    (log/debugf "Running native pivot query:\n%s" (u/pprint-to-str query))
+    (binding [qp.pipeline/*reduce*
+              (let [orig-reduce qp.pipeline/*reduce*]
+                (fn native-pivot-reduce [rff metadata reducible-rows]
+                  (let [cols (vec (:cols metadata))
+                        idx  (find-pivot-grouping-index cols)]
+                    (if (nil? idx)
+                      (orig-reduce rff metadata reducible-rows)
+                      (do
+                        (vreset! pivot-col-info {:pivot-idx idx
+                                                 :pivot-col (merge pivot.middleware/pivot-grouping-column-metadata
+                                                                   (nth cols idx))})
+                        (let [stripped-metadata (update metadata :cols #(vec-remove-at (vec %) idx))
+                              stripped-rows     (eduction
+                                                 (map (partial strip-pivot-grouping-from-row idx grouping-queue))
+                                                 reducible-rows)]
+                          (orig-reduce rff stripped-metadata stripped-rows)))))))]
+      (qp/process-query (cond-> query
+                          (seq (:info query)) qp/userland-query)
+                        wrapped-rff))))
+
+(mu/defn- run-pivot-query-multi
+  "Run a pivot query using multiple subqueries (legacy path). One subquery is generated per breakout combination
+  and they are executed sequentially; the `pivot-grouping` bitmask column is synthesized at the application level
+  by [[metabase.query-processor.pivot.middleware/add-pivot-grouping]]. Used when the driver does not declare
+  `:native-pivot-tables`."
+  [query        :- ::qp.schema/any-query
+   rff          :- ::qp.schema/rff
+   pivot-opts   :- ::pivot-opts
+   pivot-limit  :- [:maybe nat-int?]]
+  (let [all-queries (generate-queries query pivot-opts)]
+    (process-multiple-queries all-queries rff pivot-limit)))
+
 (mu/defn run-pivot-query
-  "Run the pivot query. You are expected to wrap this call in [[metabase.query-processor.streaming/streaming-response]]
-  yourself."
+  "Run the pivot query. Dispatches to [[run-native-pivot-query]] when the driver supports `:native-pivot-tables`,
+  otherwise falls back to [[run-pivot-query-multi]]. You are expected to wrap this call in
+  [[metabase.query-processor.streaming/streaming-response]] yourself."
   ([query]
    (run-pivot-query query nil))
 
@@ -517,6 +674,9 @@
                            (assoc-in [:constraints :max-results] pivot-limit)
                            (cond-> (get-in query [:constraints :max-results-bare-rows])
                              (update-in [:constraints :max-results-bare-rows] min pivot-limit))
-                           add-canonical-col-info)
-           all-queries (generate-queries query pivot-opts)]
-       (process-multiple-queries all-queries rff pivot-limit)))))
+                           add-canonical-col-info)]
+       (if (and (not *force-legacy-pivot*)
+                (native-pivot-supported? query)
+                (not (has-window-fn-aggregation? query)))
+         (run-native-pivot-query query rff pivot-opts pivot-limit)
+         (run-pivot-query-multi query rff pivot-opts pivot-limit))))))

src/metabase/query_processor/pivot/common.clj

@@ -1,8 +1,10 @@
 (ns metabase.query-processor.pivot.common
+  (:refer-clojure :exclude [empty?])
   (:require
    [clojure.math :as math]
    [metabase.util.malli :as mu]
-   [metabase.util.malli.registry :as mr]))
+   [metabase.util.malli.registry :as mr]
+   [metabase.util.performance :refer [empty?]]))
 
 ;;; these defs are just for readability, even tho they're all just ints >= 0
 (mr/def ::bitmask       nat-int?)
@@ -33,3 +35,35 @@
    (completing bit-xor)
    (long (dec (math/pow 2 num-breakouts)))
    indexes))
+
+(mu/defn full-breakout-combination :- ::breakout-combination
+  "Returns the breakout combination corresponding to `breakout-combination` belonging to the base query (the one without
+  remapped fields) accounting for the field remapping specified by `remap`.
+
+  To produce the breakout combination for the real query, the target indexes have to be included whenever a source
+  index is selected, we have to shift the indexes before which a mapped index is inserted."
+  [{breakout-combination :qp.pivot/remapped-breakout-combination
+    remap                :qp.pivot/remapped-indexes
+    :as                  _subquery} :- [:map
+                                        [:qp.pivot/remapped-breakout-combination ::breakout-combination]
+                                        [:qp.pivot/remapped-indexes              ::remapped-indexes]]]
+  (if (or (empty? remap)
+          (empty? breakout-combination))
+    breakout-combination
+    (let [limit    (apply max breakout-combination)
+          selected (set breakout-combination)
+          inserted (set (vals remap))]
+      (loop [index 0, offset 0, combination #{}]
+        (if (> index limit)
+          (-> combination sort vec)
+          (let [offset        (cond-> offset
+                                (inserted (+ index offset)) inc)
+                spliced-index (+ index offset)
+                selected?     (selected index)
+                mapped-index  (when selected?
+                                (remap spliced-index))]
+            (recur (inc index)
+                   offset
+                   (cond-> combination
+                     selected?    (conj spliced-index)
+                     mapped-index (into (take-while some? (iterate remap mapped-index)))))))))))

src/metabase/query_processor/pivot/middleware.clj

@@ -9,7 +9,8 @@
    [metabase.util.malli.registry :as mr]
    [metabase.util.performance :as perf :refer [empty? mapv]]))
 
-(def ^:private pivot-column-gropuing-metadata
+(def pivot-grouping-column-metadata
+  "Column metadata for the synthetic `pivot-grouping` column (legacy path and native restore)."
   {:name                     "pivot-grouping"
    :display_name             "pivot-grouping"
    :lib/desired-column-alias "pivot-grouping"
@@ -20,7 +21,7 @@
   (vec
    (concat
     (take num-breakouts cols)
-    [pivot-column-gropuing-metadata]
+    [pivot-grouping-column-metadata]
     (drop num-breakouts cols))))
 
 (defn add-pivot-grouping
@@ -31,16 +32,25 @@
   discarded.
 
   * Splice the [[metabase.query-processor.pivot.common/group-bitmask]] into each row in the same position as the pivot
-    column grouping metadata."
-  [{unremapped-breakout-combination :qp.pivot/unremapped-breakout-combination
-    num-remapped-breakouts          :qp.pivot/num-remapped-breakouts
-    num-unremapped-breakouts        :qp.pivot/num-unremapped-breakouts
-    :as                             _subquery}
-   rff]
-  (if-not unremapped-breakout-combination
+    column grouping metadata.
+
+  For native pivot queries (drivers that declare `:native-pivot-tables`), the `pivot-grouping` column is computed by
+  the database and the strip/restore logic in [[metabase.query-processor.pivot/run-native-pivot-query]] handles
+  metadata splicing, so this middleware is a no-op."
+  [subquery rff]
+  (cond
+    (:qp.pivot/native-pivot? subquery)
     rff
+
+    (not (:qp.pivot/unremapped-breakout-combination subquery))
+    rff
+
+    :else
     (fn rff' [metadata]
-      (let [metadata'     (m/update-existing metadata :cols add-column-grouping-metadata num-remapped-breakouts)
+      (let [num-remapped-breakouts   (:qp.pivot/num-remapped-breakouts subquery)
+            num-unremapped-breakouts (:qp.pivot/num-unremapped-breakouts subquery)
+            unremapped-breakout-combination (:qp.pivot/unremapped-breakout-combination subquery)
+            metadata'     (m/update-existing metadata :cols add-column-grouping-metadata num-remapped-breakouts)
             rf            (rff metadata')
             group-bitmask (pivot.common/group-bitmask num-unremapped-breakouts unremapped-breakout-combination)
             xform         (map (fn [row]
@@ -106,40 +116,8 @@
             (get canonical-index->subquery-index i))
           (range num-remapped-cols))))
 
-(mu/defn- full-breakout-combination :- ::pivot.common/breakout-combination
-  "Returns the breakout combination corresponding to `breakout-combination` belonging to the base query (the one without
-  remapped fields) accounting for the field remapping specified by `remap`.
-
-  To produce the breakout combination for the real query, the target indexes have to be included whenever a source
-  index is selected, we have to shift the indexes before which a mapped index is inserted."
-  [{breakout-combination :qp.pivot/remapped-breakout-combination
-    remap                :qp.pivot/remapped-indexes
-    :as                  _subquery} :- [:map
-                                        [:qp.pivot/remapped-breakout-combination ::pivot.common/breakout-combination]
-                                        [:qp.pivot/remapped-indexes              ::pivot.common/remapped-indexes]]]
-  (if (or (empty? remap)
-          (empty? breakout-combination))
-    breakout-combination
-    (let [limit    (apply max breakout-combination)
-          selected (set breakout-combination)
-          inserted (set (vals remap))]
-      (loop [index 0, offset 0, combination #{}]
-        (if (> index limit)
-          (-> combination sort vec)
-          (let [offset        (cond-> offset
-                                (inserted (+ index offset)) inc)
-                spliced-index (+ index offset)
-                selected?     (selected index)
-                mapped-index  (when selected?
-                                (remap spliced-index))]
-            (recur (inc index)
-                   offset
-                   (cond-> combination
-                     selected?    (conj spliced-index)
-                     mapped-index (into (take-while some? (iterate remap mapped-index)))))))))))
-
 (mu/defn- column-mapping [subquery :- :map]
-  (let [full-breakout-combination (full-breakout-combination subquery)]
+  (let [full-breakout-combination (pivot.common/full-breakout-combination subquery)]
     (column-mapping-for-subquery subquery full-breakout-combination)))
 
 (mu/defn- row-mapping-fn :- [:=> [:cat ::row] ::row]
@@ -162,9 +140,13 @@
 
 (defn project-pivot-subquery-rows
   "For Pivot QP subqueries, adjust the shape of the rows to match the shape of rows in the original query, splicing in
-  `nil` values as needed."
+  `nil` values as needed.
+
+  Not needed for native pivot queries (`:native-pivot-tables` drivers) where the database returns all breakout columns
+  in every row (NULLs included), so no row-shape adjustment is required."
   [query rff]
-  (if-not (:qp.pivot/remapped-breakout-combination query)
+  (if (or (:qp.pivot/native-pivot? query)
+          (not (:qp.pivot/remapped-breakout-combination query)))
     rff
     (fn rff' [metadata]
       (let [rf             (rff metadata)

test/metabase/query_processor/pivot/middleware_test.clj

@@ -1,11 +1,11 @@
 (ns metabase.query-processor.pivot.middleware-test
   (:require
    [clojure.test :refer :all]
-   [metabase.query-processor.pivot.middleware :as qp.pivot.middleware]))
+   [metabase.query-processor.pivot.common :as pivot.common]))
 
 (deftest ^:parallel full-breakout-combination-test
   (letfn [(combo [breakout-combination remaps]
-            (#'qp.pivot.middleware/full-breakout-combination
+            (pivot.common/full-breakout-combination
              {:qp.pivot/remapped-breakout-combination breakout-combination
               :qp.pivot/remapped-indexes              remaps}))]
     (is (= []

test/metabase/query_processor/pivot/native_test.clj

@@ -0,0 +1,259 @@
+(ns ^:mb/driver-tests metabase.query-processor.pivot.native-test
+  "Tests for native pivot tables (single-query path via `:native-pivot-tables`)."
+  {:clj-kondo/config '{:linters {:deprecated-var {:exclude {metabase.test.data/mbql-query {:namespaces [metabase.query-processor.pivot.native-test]}}}}}}
+  (:require
+   [clojure.test :refer :all]
+   [metabase.lib.core :as lib]
+   [metabase.query-processor :as qp]
+   [metabase.query-processor.middleware.constraints :as qp.constraints]
+   [metabase.query-processor.pivot :as qp.pivot]
+   [metabase.query-processor.pivot.common :as pivot.common]
+   [metabase.query-processor.pivot.test-util :as qp.pivot.test-util]
+   [metabase.query-processor.settings :as qp.settings]
+   [metabase.test :as mt]
+   [metabase.util.malli.registry :as mr]))
+
+(def ^:private native-pivot-drivers
+  "Drivers that implement the native pivot path."
+  #{:postgres})
+
+(defn- run-legacy-pivot-query!
+  "Run the legacy multi-query pivot path for comparison."
+  [query]
+  (binding [qp.pivot/*force-legacy-pivot* true]
+    (qp.pivot/run-pivot-query query)))
+
+(defn- sort-rows
+  "Sort rows deterministically for exact comparison."
+  [rows]
+  (sort (mapv vec rows)))
+
+(defn- pivot-grouping-index
+  [cols]
+  (first (keep-indexed (fn [i col] (when (= "pivot-grouping" (:name col)) i)) cols)))
+
+(defn- distinct-values
+  [table col]
+  (->> (mt/rows
+        (mt/dataset test-data
+          (qp/process-query
+           (mt/mbql-query nil
+             {:source-table (mt/id table)
+              :breakout     [[:field (mt/id table col) nil]]}))))
+       (map first)
+       set))
+
+(defn- breakout-excluded-at?
+  "Return true if breakout column at `breakout-idx` should be NULL for `bitmask`."
+  [bitmask breakout-idx]
+  (pos? (bit-and bitmask (bit-shift-left 1 breakout-idx))))
+
+(defn- mbql-breakout-count
+  [query]
+  (when-let [breakout (or (:breakout query) (:breakout (:query query)))]
+    (when (seq breakout) (count breakout))))
+
+(defn- expected-bitmasks
+  [query]
+  (let [pivot-opts (select-keys query [:pivot-rows :pivot-cols :show-row-totals :show-column-totals])
+        ;; Match [[metabase.query-processor.pivot/add-canonical-col-info]]: use unremapped breakout count, not
+        ;; remapped Lib breakouts (which can include extra columns and expect invalid bitmasks like 2 and 6).
+        num-breakouts (or (:qp.pivot/num-unremapped-breakouts query)
+                          (mbql-breakout-count query)
+                          (count (lib/breakouts (lib/query (mt/metadata-provider) query))))
+        combinations (#'qp.pivot/breakout-combinations
+                      num-breakouts
+                      (:pivot-rows pivot-opts)
+                      (:pivot-cols pivot-opts)
+                      (:show-row-totals pivot-opts)
+                      (:show-column-totals pivot-opts))]
+    (set (map #(pivot.common/group-bitmask num-breakouts %) combinations))))
+
+(defn- assert-native-matches-legacy
+  [query]
+  (let [native-rows (sort-rows (mt/rows (qp.pivot/run-pivot-query query)))
+        legacy-rows (sort-rows (mt/rows (run-legacy-pivot-query! query)))]
+    (is (= legacy-rows native-rows))))
+
+(deftest ^:parallel native-pivot-supported-driver-test
+  (mt/test-drivers native-pivot-drivers
+    (let [query (lib/query (mt/metadata-provider) (qp.pivot.test-util/pivot-query))]
+      (is (true? (qp.pivot/native-pivot-supported? query)))))
+  (mt/test-drivers #{:h2 :redshift}
+    (is (false? (qp.pivot/native-pivot-supported? (qp.pivot.test-util/pivot-query))))))
+
+(deftest ^:parallel native-vs-legacy-pivot-parity-test
+  (mt/test-drivers native-pivot-drivers
+    (assert-native-matches-legacy (qp.pivot.test-util/pivot-query))))
+
+(deftest native-pivot-uses-single-query-test
+  (mt/test-drivers native-pivot-drivers
+    (let [query-count (atom 0)
+          orig        @#'qp/process-query
+          query       (qp.pivot.test-util/pivot-query)]
+      (with-redefs [qp/process-query (fn [q rff]
+                                       (swap! query-count inc)
+                                       (orig q rff))]
+        (qp.pivot/run-pivot-query query))
+      (is (= 1 @query-count)))))
+
+(deftest ^:parallel native-pivot-row-schema-test
+  (mt/test-drivers native-pivot-drivers
+    (let [results   (qp.pivot/run-pivot-query (qp.pivot.test-util/pivot-query))
+          rows      (mt/rows results)
+          Row       [:cat
+                     [:maybe (into [:enum] (distinct-values :people :state))]
+                     [:maybe (into [:enum] (distinct-values :people :source))]
+                     [:maybe (into [:enum] (distinct-values :products :category))]
+                     [:enum 0 1 3 4 5 7]
+                     :int
+                     :int]
+          validator (mr/validator Row)]
+      (is (pos? (count rows)))
+      (is (=? {:status :success}
+              (reduce
+               (fn [_ row]
+                 (testing (pr-str row)
+                   (if (validator row)
+                     {:status :success}
+                     (reduced {:status :fail :bad-row row}))))
+               nil
+               rows))
+          "all rows match the Row schema above"))))
+
+(deftest ^:parallel native-pivot-bitmask-correctness-test
+  (mt/test-drivers native-pivot-drivers
+    (let [query         (qp.pivot.test-util/pivot-query)
+          pivot-opts    (select-keys query [:pivot-rows :pivot-cols :pivot_rows :pivot_cols
+                                            :show-row-totals :show-column-totals
+                                            :show_row_totals :show_column_totals])
+          pivot-rows    (or (:pivot-rows pivot-opts) (:pivot_rows pivot-opts))
+          pivot-cols    (or (:pivot-cols pivot-opts) (:pivot_cols pivot-opts))
+          row-totals?   (or (:show-row-totals pivot-opts) (:show_row_totals pivot-opts) true)
+          col-totals?   (or (:show-column-totals pivot-opts) (:show_column_totals pivot-opts) true)
+          num-breakouts 3
+          combinations  (#'qp.pivot/breakout-combinations
+                         num-breakouts pivot-rows pivot-cols row-totals? col-totals?)
+          results       (qp.pivot/run-pivot-query query)
+          cols          (mt/cols results)
+          rows          (mt/rows results)
+          pivot-idx     (pivot-grouping-index cols)
+          expected      (set (map #(pivot.common/group-bitmask num-breakouts %) combinations))
+          actual        (set (map #(nth % pivot-idx) rows))]
+      (is (= expected actual)
+          (str "unexpected bitmask values; expected " (pr-str (sort expected))
+               " but got " (pr-str (sort actual))))
+      (doseq [row rows
+              :let [bitmask (nth row pivot-idx)]
+              i (range num-breakouts)]
+        (is (= (breakout-excluded-at? bitmask i)
+               (nil? (nth row i)))
+            (str "breakout " i " null pattern for bitmask " bitmask " in row " (pr-str row)))))))
+
+(deftest ^:parallel native-pivot-column-metadata-test
+  (mt/test-drivers native-pivot-drivers
+    (let [results   (qp.pivot/run-pivot-query (qp.pivot.test-util/pivot-query))
+          cols      (mt/cols results)
+          pivot-idx (pivot-grouping-index cols)
+          pivot-col (nth cols pivot-idx)]
+      (is (= 6 (count cols)))
+      (is (= ["User → State"
+              "User → Source"
+              "Product → Category"
+              "pivot-grouping"
+              "Count"
+              "Sum of Quantity"]
+             (mapv :display_name cols)))
+      (is (= "pivot-grouping" (:name pivot-col)))
+      (is (= "pivot-grouping" (:display_name pivot-col)))
+      (is (= :type/Integer (:base_type pivot-col)))
+      (is (= :type/Integer (:effective_type pivot-col))))))
+
+(deftest ^:parallel native-pivot-filters-query-test
+  (mt/test-drivers native-pivot-drivers
+    (assert-native-matches-legacy (qp.pivot.test-util/filters-query))))
+
+(deftest ^:parallel native-pivot-parameters-query-test
+  (mt/test-drivers native-pivot-drivers
+    (assert-native-matches-legacy (qp.pivot.test-util/parameters-query))))
+
+(deftest ^:parallel native-pivot-expressions-in-breakout-test
+  (mt/test-drivers native-pivot-drivers
+    (assert-native-matches-legacy
+     (assoc (mt/mbql-query orders
+              {:expressions {"Product Rating + 1" [:+ $product_id->products.rating 1]}
+               :aggregation [[:count]]
+               :breakout    [$user_id->people.source [:expression "Product Rating + 1"]]})
+            :pivot-rows [0]
+            :pivot-cols [1]))))
+
+(deftest ^:parallel native-pivot-order-by-test
+  (mt/test-drivers native-pivot-drivers
+    (let [query (mt/mbql-query products
+                  {:breakout    [$category]
+                   :aggregation [[:count]]
+                   :order-by    [[:asc $category]]})]
+      (assert-native-matches-legacy query)
+      (is (= (sort-rows (mt/rows (run-legacy-pivot-query! query)))
+             (sort-rows (mt/rows (qp.pivot/run-pivot-query query))))))))
+
+(deftest ^:parallel native-pivot-single-breakout-test
+  (mt/test-drivers native-pivot-drivers
+    (let [query (mt/mbql-query products
+                  {:breakout    [$category]
+                   :aggregation [[:count]]
+                   :pivot-rows  [0]
+                   :pivot-cols  []})]
+      (assert-native-matches-legacy query)
+      (is (= #{0 1} (set (map second (mt/rows (qp.pivot/run-pivot-query query)))))))))
+
+(deftest ^:parallel native-pivot-totals-toggled-off-test
+  (mt/test-drivers native-pivot-drivers
+    (doseq [[_label opts] [["row totals disabled" {:show-row-totals false}]
+                           ["column totals disabled" {:show-column-totals false}]
+                           ["row and column totals disabled" {:show-row-totals    false
+                                                              :show-column-totals false}]]]
+      (let [query (merge (qp.pivot.test-util/pivot-query) opts)]
+        (assert-native-matches-legacy query)
+        (is (< (count (mt/rows (qp.pivot/run-pivot-query query)))
+               (count (mt/rows (qp.pivot/run-pivot-query (qp.pivot.test-util/pivot-query))))))))))
+
+(deftest ^:parallel native-pivot-custom-limit-test
+  (mt/test-drivers native-pivot-drivers
+    (let [query             (assoc (qp.pivot.test-util/pivot-query) :constraints {:max-results 50})
+          results           (qp.pivot/run-pivot-query query)
+          rows              (mt/formatted-rows [str str str int int int] results)
+          detail-rows       (filter #(zero? (nth % 3)) rows)
+          grand-total-row   (first (filter #(= 7 (nth % 3)) rows))
+          detail-count-sum  (reduce + (map #(nth % 4) detail-rows))
+          grand-total-count (nth grand-total-row 4)]
+      (is (> (count detail-rows) 50)
+          "All detail rows are returned (not truncated to caller's max-results)")
+      (is (some? grand-total-row)
+          "The grand total row exists")
+      (is (= detail-count-sum grand-total-count)
+          "Detail rows sum to the grand total"))))
+
+(deftest ^:parallel native-pivot-truncation-test
+  (mt/test-drivers native-pivot-drivers
+    (let [query           (qp.pivot.test-util/pivot-query)
+          unlimited-count (count (mt/rows (qp.pivot/run-pivot-query query)))]
+      (binding [qp.pivot/*pivot-max-result-rows* 20]
+        (let [results     (qp.pivot/run-pivot-query query)
+              rows        (mt/rows results)
+              pivot-limit 10]
+          (is (<= (count rows) pivot-limit)
+              "Results are truncated to the native pivot limit")
+          (is (< (count rows) unlimited-count))
+          (is (= (count rows) (get-in results [:data :pivot_rows_truncated]))
+              "Native pivot signals truncation with the row count, like legacy"))))))
+
+(deftest native-pivot-high-unaggregated-row-limit-test
+  (mt/test-drivers native-pivot-drivers
+    (binding [qp.pivot/*pivot-max-result-rows* 20]
+      (mt/with-temporary-setting-values [qp.settings/unaggregated-query-row-limit 15]
+        (let [query (-> (qp.pivot.test-util/pivot-query)
+                        (assoc :constraints (qp.constraints/default-query-constraints)))]
+          (is (not= :failed (:status (qp.pivot/run-pivot-query query)))
+              "Native pivot query should not fail with a constraint violation")
+          (assert-native-matches-legacy query))))))