Handle models with joins (#29014)

* Handle models with joins

We had been using the metadata from the query to get column names, and
then `select (column-names) from persisted-table`.

But joins affect how columns are emitted.

For this tests, using orders join to products, selecting order.total and
product.category. And the metadata from running this query uses the
normal names `:name "total"`, and `:name "category"`. _BUT_ the emitted
sql does not use these names:

```sql
SELECT "public"."orders"."total" AS "total",
       "products"."category" AS "products__category" -- <--------
FROM "public"."orders"
LEFT JOIN "public"."products" AS "products"
       ON "public"."orders"."product_id" = "products"."id"
```

When we persist, we do `create table as <(qp/compile query)> so we were
creating a column named `products__category` _NOT_ "category". But later
on we would select "category" and this would blow up.

```sql
select "total", "category" -- <- category doesn't exist in this table
from "metabase_cache_424a9_8"."model_1105_txbhmkrnoy"
```

Now a query of `{:source-table "card__<model-id>"}' emits the following
query when the persisted cache is substituted:

```sql
SELECT "source"."total" AS "total",
       "source"."products__category" AS "products__category"
FROM (select * from "metabase_cache_424a9_8"."model_1152_lecqfzcjke") AS
"source"
LIMIT 1048575
```

* Test and lint fixes

src/metabase/models/persisted_info.clj

@@ -16,7 +16,7 @@
   of a field."
   [{field-name :name :keys [base_type effective_type]}]
   {:field-name field-name
-     :base-type (or effective_type base_type)})
+   :base-type (or effective_type base_type)})
 
 (def ^:private Metadata
   "Spec for metadata. Just asserting we have base types and names, not the full metadata of the qp."

src/metabase/query_processor/util/persisted_cache.clj

 (ns metabase.query-processor.util.persisted-cache
   (:require
-   [clojure.string :as str]
    [metabase.driver :as driver]
    [metabase.driver.ddl.interface :as ddl.i]
    [metabase.driver.sql.util :as sql.u]
@@ -26,20 +25,15 @@
 (defn persisted-info-native-query
   "Returns a native query that selects from the persisted cached table from `persisted-info`. Does not check if
   persistence is appropriate. Use [[can-substitute?]] for that check."
-  [persisted-info]
-  (let [database-id (:database_id persisted-info)
-        driver      (or driver/*driver* (driver.u/database->driver database-id))]
-    (format "select %s from %s.%s"
-            (str/join ", " (map #(sql.u/quote-name
-                                  driver
-                                  :field
-                                  (:field-name %))
-                                (get-in persisted-info [:definition :field-definitions])))
+  [{:keys [database_id table_name] :as _persisted-info}]
+  (let [driver      (or driver/*driver* (driver.u/database->driver database_id))]
+    ;; select * because we don't actually know the name of the fields when in the actual query. See #28902
+    (format "select * from %s.%s"
             (sql.u/quote-name
              driver
              :table
-             (ddl.i/schema-name {:id database-id} (public-settings/site-uuid)))
+             (ddl.i/schema-name {:id database_id} (public-settings/site-uuid)))
             (sql.u/quote-name
              driver
              :table
-             (:table_name persisted-info)))))
+             table_name))))

test/metabase/driver/common/parameters/values_test.clj

@@ -342,7 +342,7 @@
                 (testing "tag uses persisted table"
                   (let [pi (db/select-one 'PersistedInfo :card_id (u/the-id model))]
                     (is (= "persisted" (:state pi)))
-                    (is (re-matches #"select \"id\", \"name\" from \"metabase_cache_[a-z0-9]+_[0-9]+\".\"model_[0-9]+_model\""
+                    (is (re-matches #"select \* from \"metabase_cache_[a-z0-9]+_[0-9]+\".\"model_[0-9]+_model\""
                                     (:query
                                      (value-for-tag
                                       {:name         "card-template-tag-test"

test/metabase/query_processor/persistence_test.clj

@@ -107,4 +107,39 @@
                 (testing "Was persisted"
                   (is (str/includes? (-> results :data :native_form :query) persisted-schema)))
                 (testing "Did not find bad field clauses"
-                  (is (= [] @bad-refs)))))))))))
+                  (is (= [] @bad-refs))))))))))
+
+  (testing "Can use joins with persisted models (#28902)"
+    (mt/test-drivers (mt/normal-drivers-with-feature :persist-models)
+      (mt/dataset sample-dataset
+        (mt/with-persistence-enabled [persist-models!]
+          (mt/with-temp* [Card [model {:dataset true
+                                       :database_id (mt/id)
+                                       :query_type :query
+                                       :dataset_query
+                                       (mt/mbql-query orders
+                                         {:fields [$total &products.products.category]
+                                          :joins [{:source-table $$products
+                                                   :condition [:= $product_id &products.products.id]
+                                                   :strategy :left-join
+                                                   :alias "products"}]})}]]
+            (persist-models!)
+            (let [query   {:type :query
+                           :database (mt/id)
+                           :query {:source-table (str "card__" (:id model))}}
+                  results (qp/process-query query)
+                  persisted-schema (ddl.i/schema-name (mt/db) (public-settings/site-uuid))]
+              (testing "Was persisted"
+                (is (str/includes? (-> results :data :native_form :query) persisted-schema))))
+            (let [query {:type :query
+                         :database (mt/id)
+                         :query {:source-table (str "card__" (:id model))
+                                 :aggregation [[:count]]
+                                 :breakout [(mt/$ids $products.category)]}}
+                  results (qp/process-query query)
+                  persisted-schema (ddl.i/schema-name (mt/db) (public-settings/site-uuid))]
+              (testing "Was persisted"
+                (is (str/includes? (-> results :data :native_form :query) persisted-schema)))
+              (is (= {"Doohickey" 3976, "Gadget" 4939,
+                      "Gizmo"     4784, "Widget" 5061}
+                     (->> results :data :rows (into {})))))))))))