wip: validate-query-for-view

* Post /transform endpoint stub

* WIP Hook in POST and add test

* Uniqueness check, query-type, schema  in endpoint

* Tests and assertions for post

* PUT and further tweaks

* Migration 2.0 including metabase_table col

* Adjust model

* Migration adj

* model

* Adjust tests

* Adjust PUT

* Delete enpoint

* GET endpoint

* Cleanup

* Migration

* Serialization for transforms will be handled later

* Migration ids

* Tests

* top schemas query

* Docstrings

* Test

* Modules kondo config

* Modules interfaces update

* Transform model requires

* API cleanup

* tests

* Sort nss

* Review remarks and cleanup

* Restore view replacement because of postgres

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

@@ -1245,6 +1245,21 @@
            models
            util}}
 
+  transform
+  {:team "Semantic Layer"
+   :api #{}
+   :uses #{api
+           driver
+           driver-api
+           lib
+           lib-be
+           models
+           queries
+           query-permissions
+           query-processor
+           sync
+           util}}
+
   types
   {:team "Semantic Layer"
    :api  #{metabase.types.core

docs/developers-guide/driver-changelog.md

@@ -4,6 +4,16 @@ title: Driver interface changelog
 
 # Driver Interface Changelog
 
+## Metabase 0.57.0
+
+- Add the feature `:view` for drivers that support views
+
+- Add the multi-method `metabase.driver/view-name-length-limit`, like `table-name-length-limit` but for views
+
+- Add the multi-method `metabase.driver/create-view!` that allows the driver to create or replace a view
+
+- Add the multi-method `metabase.driver/drop-view!` that allows the driver to delete a view
+
 ## Metabase 0.56.0
 
 - Add the testing multi-method `tx/track-dataset` for shared cloud dbs to track loaded datasets for more efficient sharing.

enterprise/backend/src/metabase_enterprise/serialization/v2/models.clj

@@ -90,6 +90,7 @@
    "Secret"
    "Session"
    "TaskHistory"
+   "TransformView"
    "User"
    "UserKeyValue"
    "UserParameterValue"

enterprise/backend/test/metabase_enterprise/models/entity_id_test.clj

@@ -79,6 +79,7 @@
     :model/Session
     :model/TaskHistory
     :model/TimelineEvent
+    :model/TransformView
     :model/User
     :model/UserParameterValue
     :model/UserKeyValue

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

@@ -42,7 +42,9 @@
                               :uuid-type                 false
                               :nested-field-columns      false
                               :test/jvm-timezone-setting false
-                              :database-routing          false}]
+                              :database-routing          false
+                              ;; will eventually be turned on
+                              :view                      false}]
   (defmethod driver/database-supports? [:redshift feature] [_driver _feat _db] supported?))
 
 ;;; +----------------------------------------------------------------------------------------------------------------+

resources/migrations/057_update_migrations.yaml

@@ -0,0 +1,228 @@
+databaseChangeLog:
+  - objectQuotingStrategy: QUOTE_ALL_OBJECTS
+
+  - changeSet:
+      id: v57.2025-07-08T11:37:31
+      author: lbrdnk
+      comment: Add transform_view table
+      changes:
+        - createTable:
+            tableName: transform_view
+            remarks: Metabase managed views in user databases
+            columns:
+              - column:
+                  name: id
+                  type: int
+                  remarks: Transform view id
+                  autoIncrement: true
+                  constraints:
+                    primaryKey: true
+                    nullable: false
+              - column:
+                  name: creator_id
+                  type: int
+                  remarks: Id of user who created or updated the transform
+              - column:
+                  name: database_id
+                  type: int
+                  remarks: Id of metabase_database the view is created in
+              - column:
+                  name: status
+                  type: varchar(16)
+                  remarks: Status of the entity the table represents, useful eg. for error handling
+              - column:
+                  name: view_schema
+                  type: varchar(256)
+                  remarks: The schema view is created in
+              - column:
+                  name: view_name
+                  type: varchar(256)
+                  remarks: Name of the view, unique per database and schema
+              - column:
+                  name: dataset_query
+                  type: ${text.type}
+                  remarks: Query as stored in report_card.dataset_query
+              - column:
+                  name: dataset_query_schema
+                  type: int
+                  remarks: Schema version similar to report_card.card_schema
+              - column:
+                  name: dataset_query_type
+                  type: varchar(16)
+                  remarks: Type of dataset_query
+              - column:
+                  name: created_at
+                  type: ${timestamp_type}
+                  remarks: Timestamp of transform view creation
+                  defaultValueComputed: current_timestamp
+                  constraints:
+                    nullable: false
+              - column:
+                  name: updated_at
+                  type: ${timestamp_type}
+                  remarks: Timestamp of transform view update
+                  defaultValueComputed: current_timestamp
+                  constraints:
+                    nullable: false
+
+  - changeSet:
+      id: v57.2025-07-08T17:13:14
+      author: lbrdnk
+      comment: Add unique constraint onto transform_view table
+      changes:
+        - addUniqueConstraint:
+            tableName: transform_view
+            columnNames: database_id, view_schema, view_name
+            constraintName: unique_transform_view_database_id_view_schema_view_name
+
+  - changeSet:
+      id: v57.2025-07-09T15:18:38
+      author: lbrdnk
+      comment: Create index for database_id of transform_view table
+      preConditions:
+        - onFail: MARK_RAN
+        - not:
+            - indexExists:
+                tableName: transform_view
+                indexName: idx_transform_view_database_id
+      changes:
+        - createIndex:
+            tableName: transform_view
+            indexName: idx_transform_view_database_id
+            columns:
+              - column:
+                  name: database_id
+      rollback: # Handled in rollback of "Add transform_view table" changeset
+
+  - changeSet:
+      id: v57.2025-07-10T16:56:28
+      author: lbrdnk
+      comment: Create index for creator_id of transform_view table
+      preConditions:
+        - onFail: MARK_RAN
+        - not:
+            - indexExists:
+                tableName: transform_view
+                indexName: idx_transform_view_creator_id
+      changes:
+        - createIndex:
+            tableName: transform_view
+            indexName: idx_transform_view_creator_id
+            columns:
+              - column:
+                  name: creator_id
+      rollback: # Handled in rollback of "Add transform_view table" changeset
+
+  - changeSet:
+      id: v57.2025-07-10T16:58:15
+      author: lbrdnk
+      comment: Create fk constraint on core_user.id
+      changes:
+        - addForeignKeyConstraint:
+            constraintName: fk_transform_view_creator_id
+            baseTableName: transform_view
+            baseColumnNames: creator_id
+            referencedTableName: core_user
+            referencedColumnNames: id
+            onDelete: SET NULL
+
+  - changeSet:
+      id: v57.2025-07-10T16:59:52
+      author: lbrdnk
+      comment: Create fk constraint on metabase_database.id
+      changes:
+        - addForeignKeyConstraint:
+            constraintName: fk_transform_view_database_id
+            baseTableName: transform_view
+            baseColumnNames: database_id
+            referencedTableName: metabase_database
+            referencedColumnNames: id
+            onDelete: SET NULL
+
+  - changeSet:
+      id: v57.2025-07-13T12:30:22
+      author: lbrdnk
+      comment: Add transform_id to metabase_table to reference transform_view.id
+      changes:
+        - addColumn:
+            tableName: metabase_table
+            columns:
+              - column:
+                  name: transform_id
+                  type: int
+                  remarks: fk to tranform_view table
+
+  - changeSet:
+      id: v57.2025-07-13T12:35:10
+      author: lbrdnk
+      comment: Index metabase_table.transform_id
+      changes:
+        - createIndex:
+            tableName: metabase_table
+            indexName: idx_metabase_table_transform_id
+            columns:
+              - column:
+                  name: transform_id
+
+  - changeSet:
+      id: v57.2025-07-13T12:39:23
+      author: lbrdnk
+      comment: Add metabase_table.transform_id fk constraint
+      changes:
+        - addForeignKeyConstraint:
+            constraintName: fk_metabase_table_transform_id
+            baseTableName: metabase_table
+            baseColumnNames: transform_id
+            referencedTableName: transform_view
+            referencedColumnNames: id
+            onDelete: SET NULL
+
+  # >>>>>>>>>> DO NOT ADD NEW MIGRATIONS BELOW THIS LINE! ADD THEM ABOVE <<<<<<<<<<
+
+########################################################################################################################
+#
+# ADVICE:
+#
+# 1) Think through some of these points when writing a migration, learn from our past mistakes:
+#    - Do you really need a migration? Could the feature work without it? Prefer not to if possible.
+#      Data in the wild can be vastly different from what you expect, and it's hard to get right.
+#    - Make sure your migration is backward compatible: it might not be possible to add a constraint back
+#      if you drop it in a migration.
+#    - Postgres, MySQL and H2 have their differences. Make sure your migration works for all.
+#    - Database encryption is a major issue:
+#      - Fields like `metabase_database.details` or `setting.value` can be encrypted, so you need to decrypt them in
+#        your migration. See #42617, #44048.
+#      - Database could be partially encrypted, see https://www.notion.so/72575933ef2a446bafd16909e05a7387
+#    - Custom migrations:
+#      - Prefer SQL migrations when possible.
+#      - Never use application code: it can change and *will* break your migration.
+#      - Do not use Toucan models - refer table names directly.
+#      - If it's a big change, consider using Quartz, see #42279
+#      - More in `metabase.app_db.custom_migrations` namespace doc.
+#    - Never delete a migration: users won't be able to downgrade. If you really need to, see #44908
+#    - Migration id (`vXX.<date>`) must match its earliest released version:
+#      - Do not backport `v51....` to version 50, Metabase will try to downgrade it.
+#      - Instead, write a migration with an oldest target you plan to backport to in mind.
+#
+# 2) Migrations should go in the ###_update_migrations.yaml file for the target version.
+#
+# 3) Run mage lint-migrations-file to run core.spec checks against any changes you make here. Liquibase is pretty
+#    forgiving and won't complain if you accidentally mix up things like deleteCascade and onDelete: CASCADE. CI runs
+#    this check but it's nicer to know now instead of waiting for CI.
+#
+# 3) Migration IDs should follow the format
+#
+#    vMM.TIMESTAMP
+#
+#    Where
+#
+#    M         = major version
+#    TIMESTAMP = the current timestamp with format `yyyy-MM-dd'T'HH:mm:ss`
+#                To get this timestamp, evaluate this in your REPL: `(dev/migration-timestamp)`
+#
+#    E.g: You're adding a new migration for version 49, And it's 10:30:00AM on April 1, 2023 (UTC),
+#    your migration id should be: `v49.2023-04-01T10:30:00`.
+#
+# PLEASE KEEP THIS MESSAGE AT THE BOTTOM OF THIS FILE!!!!! Add new migrations above the message.
+#
+########################################################################################################################

src/metabase/api_routes/routes.clj

@@ -48,6 +48,7 @@
    [metabase.testing-api.core]
    [metabase.tiles.api]
    [metabase.timeline.api]
+   [metabase.transform.api]
    [metabase.upload.api]
    [metabase.user-key-value.api]
    [metabase.users.api]
@@ -177,6 +178,7 @@
    "/tiles"                (+auth 'metabase.tiles.api)
    "/timeline"             (+auth metabase.timeline.api/timeline-routes)
    "/timeline-event"       (+auth metabase.timeline.api/timeline-event-routes)
+   "/transform"            (+auth metabase.transform.api/routes)
    "/upload"               (+auth 'metabase.upload.api)
    "/user"                 (+auth 'metabase.users.api)
    "/user-key-value"       (+auth 'metabase.user-key-value.api)

src/metabase/driver.clj

@@ -708,7 +708,10 @@
     :test/uuids-in-create-table-statements
 
     ;; Does this driver support Metabase's database routing feature?
-    :database-routing})
+    :database-routing
+
+    ;; Does this driver support views?
+    :view})
 
 (defmulti database-supports?
   "Does this driver and specific instance of a database support a certain `feature`?
@@ -1122,6 +1125,41 @@
   ;; For most databases, the same limit is used for all identifier types.
   (table-name-length-limit driver))
 
+(defmulti view-name-length-limit
+  "Return the maximum number of bytes allowed in a view name, or `nil` if there is no limit."
+  {:changelog-test/ignore true, :added "0.57.0", :arglists '([driver])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
+(defmethod view-name-length-limit :default [driver]
+  ;; For most databases, the same limit is used for all identifier types.
+  (table-name-length-limit driver))
+
+(defmulti create-view!
+  "Create (or replace) a view named `view-name`. If the view already exists it will throw an error, unless `:replace?`
+  is true in the opts map.
+  The view definition should be a native query form, which is expected to have been compiled against the specific driver
+  via e.g. [[mbql->native]]."
+  {:added "0.57.0", :arglists '([driver database-id view-name view-definition & {:keys [replace?] :as opts}])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
+(defmulti drop-view!
+  "Drop a view named `view-name`. If the view doesn't exist it will not be dropped. `view-name` may be qualified
+  by schema e.g.
+
+    schema.view"
+  {:added "0.57.0", :arglists '([driver db-id ^String view-name])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
+(defmulti validate-query-for-view
+  "Validate an MBQL query for use as a view definition.
+   Results should match the [[metabase.transform.interface/view-validation-result]] schema."
+  {:arglists '([driver query])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
 (defmulti create-table!
   "Create a table named `table-name`. If the table already exists it will throw an error.
   `args` is an optional map with an optional entry `primary-key`. The `primary-key` value is a vector of column names

src/metabase/driver/h2.clj

@@ -69,7 +69,8 @@
                               :test/jvm-timezone-setting false
                               :uuid-type                 true
                               :uploads                   true
-                              :database-routing          true}]
+                              :database-routing          true
+                              :view                      true}]
   (defmethod driver/database-supports? [:h2 feature]
     [_driver _feature _database]
     supported?))
@@ -222,6 +223,8 @@
                     CommandInterface/DROP_TABLE
                     CommandInterface/CREATE_SCHEMA
                     CommandInterface/DROP_SCHEMA
+                    CommandInterface/CREATE_VIEW
+                    CommandInterface/DROP_VIEW
                     CommandInterface/ALTER_TABLE_RENAME
                     CommandInterface/ALTER_TABLE_ADD_COLUMN
                     CommandInterface/ALTER_TABLE_DROP_COLUMN

src/metabase/driver/postgres.clj

@@ -76,7 +76,8 @@
                               :expressions/integer      true
                               :expressions/float        true
                               :expressions/date         true
-                              :database-routing         true}]
+                              :database-routing         true
+                              :view                     true}]
   (defmethod driver/database-supports? [:postgres feature] [_driver _feature _db] supported?))
 
 (defmethod driver/database-supports? [:postgres :nested-field-columns]

src/metabase/driver/sql_jdbc.clj

@@ -132,6 +132,25 @@
   [_driver _semantic_type expr]
   (h2x/->timestamp expr))
 
+(defn- create-view!-sql [driver view-name view-definition & {:keys [replace?]}]
+  (with-quoting driver
+    (first (sql/format {(if replace? :create-or-replace-view :create-view) (quote-table view-name)
+                        :raw view-definition}
+                       :quoted true
+                       :dialect (sql.qp/quote-style driver)))))
+
+(defmethod driver/create-view! :sql-jdbc
+  [driver database-id view-name view-definition & {:keys [replace?] :or {replace? false}}]
+  (let [sql (create-view!-sql driver view-name view-definition :replace? replace?)]
+    (driver-api/execute-write-sql! database-id sql)))
+
+(defmethod driver/drop-view! :sql-jdbc
+  [driver db-id view-name]
+  (let [sql (first (sql/format {:drop-view [:if-exists (keyword view-name)]}
+                               :quoted true
+                               :dialect (sql.qp/quote-style driver)))]
+    (driver-api/execute-write-sql! db-id sql)))
+
 (defn- create-table!-sql
   [driver table-name column-definitions & {:keys [primary-key]}]
   (with-quoting driver

src/metabase/driver_api/core.clj

@@ -132,6 +132,8 @@
  nest-query/nest-expressions
  premium-features/is-hosted?
  qp.compile/compile
+ qp.compile/compile-preprocessed
+ qp.compile/compile-with-inline-parameters
  qp.debug/debug>
  qp.i/*disable-qp-logging*
  qp.preprocess/preprocess

src/metabase/lib/core.cljc

@@ -388,5 +388,7 @@
   temporal-bucket
   with-temporal-bucket]
  [lib.util
+  native-stage?
+  query-stage
   normalized-query-type
   source-table-id])

src/metabase/models/resolution.clj

@@ -83,6 +83,7 @@
     :model/TaskHistory                       metabase.task-history.models.task-history
     :model/Timeline                          metabase.timeline.models.timeline
     :model/TimelineEvent                     metabase.timeline.models.timeline-event
+    :model/TransformView                     metabase.transform.models.transform
     :model/User                              metabase.users.models.user
     :model/UserKeyValue                      metabase.user-key-value.models.user-key-value
     :model/UserParameterValue                metabase.users.models.user-parameter-value

src/metabase/queries/core.clj

@@ -33,6 +33,7 @@
    ;; TODO -- not convinced whether this belongs here or in `permissions`
   with-can-run-adhoc-query]
  [metabase.queries.models.card.metadata
+  normalize-dataset-query
   refresh-metadata]
  [metabase.queries.models.parameter-card]
  [metabase.queries.models.query

src/metabase/query_processor/middleware/add_implicit_clauses.clj

@@ -5,6 +5,7 @@
    [metabase.legacy-mbql.schema :as mbql.s]
    [metabase.legacy-mbql.util :as mbql.u]
    [metabase.lib.metadata :as lib.metadata]
+   [metabase.lib.options :as lib.options]
    [metabase.lib.schema.id :as lib.schema.id]
    [metabase.lib.util.match :as lib.util.match]
    [metabase.query-processor.error-type :as qp.error-type]
@@ -212,6 +213,7 @@
      ;; like `:fields` to source queries *inside* joins)
      (if (and (map? form)
               ((some-fn :source-table :source-query) form)
+              (not (:for-transform (lib.options/options form)))
               (not (:condition form)))
        (-> form add-implicit-breakout-order-by add-implicit-fields)
        form))

src/metabase/query_processor/util/transformations/nest_breakouts.clj

@@ -137,7 +137,8 @@
 
 (defn- add-implicit-breakouts
   [stage]
-  (if-let [breakouts (not-empty (:breakout stage))]
+  (if-let [breakouts (and (not (:for-transform (lib.options/options stage)))
+                          (not-empty (:breakout stage)))]
     (let [finest-temp-breakout (finest-temporal-breakout-index breakouts 1)
           breakout-exprs (if finest-temp-breakout
                            (concat (m/remove-nth finest-temp-breakout breakouts)

src/metabase/sync/core.clj

@@ -29,7 +29,9 @@
   sync-database!
   sync-table!]
  [metabase.sync.sync-metadata
-  sync-db-metadata!]
+  sync-db-metadata!
+  sync-table-metadata!
+  sync-new-table-metadata!]
  [metabase.sync.sync-metadata.fields
   sync-fields-for-table!]
  [metabase.sync.sync-metadata.tables

src/metabase/sync/fetch_metadata.clj

@@ -45,10 +45,9 @@
   (log-if-error "table-fields-metadata"
     (let [driver (driver.u/database->driver database)
           result (if (driver.u/supports? driver :describe-fields database)
-                   (set (driver/describe-fields driver
-                                                database
-                                                :table-names [(:name table)]
-                                                :schema-names [(:schema table)]))
+                   (let [schema-names (when (driver.u/supports? driver :schemas database)
+                                        [(:schema table)])]
+                     (set (driver/describe-fields driver database :table-names [(:name table)] :schema-names schema-names)))
                    (:fields (driver/describe-table driver database table)))]
       result)))
 

src/metabase/sync/sync_metadata.clj

@@ -7,6 +7,7 @@
    3.  Sync FKs    (`metabase.sync.sync-metadata.fks`)
    4.  Sync Metabase Metadata table (`metabase.sync.sync-metadata.metabase-metadata`)"
   (:require
+   [metabase.driver.util :as driver.u]
    [metabase.sync.fetch-metadata :as fetch-metadata]
    [metabase.sync.interface :as i]
    [metabase.sync.sync-metadata.dbms-version :as sync-dbms-ver]
@@ -18,6 +19,7 @@
    [metabase.sync.sync-metadata.tables :as sync-tables]
    [metabase.sync.util :as sync-util]
    [metabase.util :as u]
+   [metabase.util.log :as log]
    [metabase.util.malli :as mu]
    [metabase.warehouse-schema.models.table :as table]))
 
@@ -76,3 +78,17 @@
     (sync-fields/sync-fields-for-table! database table)
     (sync-fks/sync-fks-for-table! database table)
     (sync-indexes/maybe-sync-indexes-for-table! database table)))
+
+;; NOTE: won't perform full analysis, so async we should `(-> table sync/sync-table! sync-util/set-initial-table-sync-complete!)`
+(mu/defn sync-new-table-metadata!
+  "Sync the metadata for a new table in `database`"
+  [database :- i/DatabaseInstance {:keys [table-name schema]}]
+  (let [schema (not-empty schema)
+        table (sync-tables/create-or-reactivate-table!
+               database
+               ;; TODO: ideally we should get this from [[driver/describe-table]], once that has `:schema` and `:table` filter support
+               {:name table-name :schema schema})]
+    (when (and (driver.u/supports? (driver.u/database->driver database) :schemas database)
+               (not schema))
+      (log/warnf "Schema not provided for table ''%s'' during sync-new-table-metadata!" table-name))
+    (doto table sync-table-metadata!)))

src/metabase/sync/sync_metadata/fields.clj

@@ -132,6 +132,10 @@
            ;; TODO: decouple nested field columns sync from field sync. This will allow
            ;; describe-fields to be used for field sync for databases with nested field columns
            ;; Also this should be a driver method, not a sql-jdbc.sync method
-           db-metadata (fetch-metadata/include-nested-fields-for-table db-metadata database table)]
-       {:total-fields   (count db-metadata)
+           db-metadata (fetch-metadata/include-nested-fields-for-table db-metadata database table)
+           total-fields (count db-metadata)]
+       (when (zero? total-fields)
+         ;; there can legit exist tables with no fields, so we can't warn about it
+         (log/debugf "Zero total fields sync'd for table ''%s''" (sync-util/name-for-logging table)))
+       {:total-fields   total-fields
         :updated-fields (sync-and-update! database table db-metadata)}))))

src/metabase/transform/api.clj

@@ -0,0 +1,73 @@
+(ns metabase.transform.api
+  (:require
+   [metabase.api.common :as api]
+   [metabase.api.macros :as api.macros]
+   [metabase.query-permissions.core :as query-perms]
+   [metabase.transform.models.transform :as models.transform]
+   [metabase.util.i18n :as i18n]
+   [metabase.util.malli.registry :as mr]
+   [metabase.util.malli.schema :as ms]))
+
+(mr/def ::post-body
+  [:map
+   [:display_name ms/NonBlankString]
+   [:dataset_query ms/Map]
+   [:schema {:optional true} ms/NonBlankString]])
+
+(api.macros/defendpoint :post "/"
+  "Create transform"
+  [_ _
+   {dataset-query :dataset_query
+    display-name :display_name
+    schema :schema
+    :as _body} :- ::post-body]
+  ;; TODO: check whether user is eligible to create transform view -- admin?
+  (query-perms/check-run-permissions-for-query dataset-query)
+  (models.transform/insert-returning-instance!
+   ;; Temporary, api callers should pick the schema
+   (or schema
+       (models.transform/top-schema (:database (models.transform/dataset-query->query dataset-query))))
+   display-name
+   dataset-query
+   @api/*current-user*))
+
+;; TODO: pagination, sort, filter, ?perms hydration.
+(api.macros/defendpoint :get "/"
+  "Get all transforms"
+  []
+  (models.transform/get-transforms))
+
+;; TODO: pagination, sort, filter, ?perms hydration.
+(api.macros/defendpoint :get "/:id"
+  "Get all transforms"
+  [{:keys [id]} :- [:map [:id ms/Int]]]
+  (models.transform/transform-by-id id))
+
+(mr/def ::update-body
+  [:map
+   [:dataset_query {:optional true} ms/Map]])
+
+(api.macros/defendpoint :put "/:id"
+  "Update transform"
+  [{:keys [id]} :- [:map [:id ms/Int]]
+   _
+   body :- ::update-body]
+  (let [dataset-query (:dataset_query body)]
+    (when (empty? (:dataset_query body))
+      (throw (ex-info (i18n/tru "Query is empty.")
+                      {:status-code 400
+                       :transform_id id})))
+    (query-perms/check-run-permissions-for-query dataset-query)
+    (models.transform/update-transform!
+     id dataset-query
+     @api/*current-user*)))
+
+(api.macros/defendpoint :delete "/:id"
+  "Delete transform"
+  [{:keys [id]} :- [:map [:id ms/Int]]]
+  ;; TODO: perms.
+  (models.transform/delete-transform! id))
+
+(def ^{:arglists '([request respond raise])} routes
+  "`/api/transform` routes."
+  (api.macros/ns-handler))

src/metabase/transform/interface.clj

@@ -0,0 +1,14 @@
+(ns metabase.transform.interface
+  (:require
+   [metabase.util.malli.registry :as mr]))
+
+(mr/def ::view-validation-issue
+  [:map
+   [:type :keyword]                    ; e.g. :order-by-without-limit
+   [:severity [:enum :warning :error]]
+   [:message :string]])
+
+(mr/def ::view-validation-result
+  [:map
+   [:status [:enum :valid :warning :error]]
+   [:issues [:sequential ::view-validation-issue]]])

src/metabase/transform/models/transform.clj

@@ -0,0 +1,230 @@
+(ns metabase.transform.models.transform
+  (:require
+   [metabase.driver :as driver]
+   [metabase.driver-api.core :as driver-api]
+   [metabase.driver.util :as driver.u]
+   [metabase.lib-be.metadata.jvm :as lib.metadata.jvm]
+   [metabase.lib.core :as lib]
+   [metabase.models.interface :as mi]
+   [metabase.queries.core :as queries]
+   [metabase.query-processor.compile :as qp.compile]
+   [metabase.sync.core :as sync]
+   [metabase.util.i18n :as i18n]
+   [methodical.core :as methodical]
+   [toucan2.core :as t2]))
+
+(doto :model/TransformView
+  (derive :metabase/model)
+  (derive :hook/timestamped?))
+
+(methodical/defmethod t2/table-name :model/TransformView [_model] :transform_view)
+
+(t2/deftransforms :model/TransformView
+  {:dataset_query mi/transform-metabase-query
+   :dataset_query_type mi/transform-keyword
+   :status mi/transform-keyword})
+
+(defn namespaced-view-name
+  "Generate view name"
+  [schema-name view-name]
+  (if schema-name
+    (str schema-name "." view-name)
+    view-name))
+
+;; TODO (lbrdnk 2025-07-15): Presumably, the schema should be inferred from mbql queries and required
+;;                           for native queries.
+(defn top-schema
+  "Get most used schema"
+  [database-id]
+  ;; Following is temporary. Later do it nicer way
+  (-> (t2/query {:select [[[:count :id] :c] [:schema :s]]
+                 :from [[:metabase_table :t]]
+                 :where [:= :t.db_id database-id]
+                 :group-by [:schema]
+                 :order-by [[[:count :id] :desc]]
+                 :limit 1})
+      first :s))
+
+;; TODO (lbrdnk 2025-07-15):
+;;       This is borrowed. Ideally dataset_query submodule should be created in metabase.queries and this and related
+;;       functions provided there. Gist is that not only functionality metabase.queries is supposed to manipulate this.
+(defn dataset-query->query
+  "Convert the `dataset_query` to pMBQL query."
+  ([dataset-query]
+   (if (-> dataset-query :lib/type #{:mbql/query})
+     dataset-query
+     (some-> (:database dataset-query)
+             lib.metadata.jvm/application-database-metadata-provider
+             (dataset-query->query dataset-query))))
+  ([metadata-provider dataset-query]
+   (if (-> dataset-query :lib/type #{:mbql/query})
+     dataset-query
+     (some->> dataset-query queries/normalize-dataset-query (lib/query metadata-provider)))))
+
+(def view-name-prefix
+  "Prefix for view names. Names `mb_transform_\\d+` in user schemas are reserved for managed views."
+  "mb_transform_")
+
+(defn transform-view-name
+  "Generate name for transform view. Id should be unique."
+  [id]
+  (assert (pos-int? id))
+  (str view-name-prefix id))
+
+;; TODO: test for version matching, stub now
+#_{:clj-kondo/ignore [:missing-docstring :unresolved-namespace]}
+(def dataset-query-current-schema @#'metabase.queries.models.card/current-schema-version)
+
+(defn- query-type
+  [query]
+  (or (and (lib/native-stage? query -1)
+           :native)
+      :query))
+
+(defn- assert-unique-name!
+  "Assert there is no same `name` :model/Table in the `schema`."
+  [database schema name]
+  ;; Deliberately not checking :model/TransformView. It is guaranteed the name is unique across that table
+  ;; as the transform_view.id is used to generate name.
+  (when-some [table-ids (not-empty (t2/select-fn-vec :id :model/Table
+                                                     :active true
+                                                     :db_id (:id database)
+                                                     :schema schema
+                                                     :name name))]
+    (throw (ex-info (i18n/tru "View name in use. `{0}<numbers>+` names are reserved." view-name-prefix)
+                    {:status-code 400
+                     :database-id (:id database)
+                     :database-name (:name database)
+                     :schema schema
+                     :name name
+                     :same-name-table-ids table-ids}))))
+
+;; TODO (lbrdnk 2025-07-15): Remove in favor of models' behavior, ie. display names can be duplicate. However
+;;                           adequate logic to differentiate that on FE should be put in place.
+(defn- assert-unique-display-name!
+  [database schema display-name]
+  ;; Deliberately not checking TransformView -- TODO: remove view_display_name from that table
+  (when-some [table-ids (not-empty (t2/select-fn-vec :id :model/Table
+                                                     :active true
+                                                     :db_id (:id database)
+                                                     :schema schema
+                                                     :display_name display-name))]
+    (throw (ex-info (i18n/tru "View display name in use. Display name of a transform must be unique.")
+                    {:status-code 400
+                     :database-id (:id database)
+                     :database-name (:name database)
+                     :schema schema
+                     :display-name display-name
+                     :same-display-name-table-ids table-ids}))))
+;; WIP
+(defmethod driver/validate-query-for-view :postgres [_ query]
+  (let [last-stage (lib/query-stage query -1)
+        has-limit (contains? last-stage :limit)
+        has-order-by (seq (:order-by last-stage))]
+
+    (if (and has-order-by (not has-limit))
+      {:status :warning
+       :issues [{:type :order-by-without-limit
+                 :severity :warning
+                 :message "ORDER BY without limit cannot be used in a view, will be ignored"}]}
+      {:status :valid})))
+
+(defn- validate-for-view [driver dataset-query]
+  (let [query (dataset-query->query dataset-query)
+        query-type (query-type query)]
+    (when (not= query-type :native)
+      (driver/validate-query-for-view driver query))))
+
+(defn insert-returning-instance!
+  "Create new transform."
+  [schema display-name dataset-query creator]
+  (let [query (dataset-query->query dataset-query)
+        query-type (query-type query)
+        creator-id (:id creator)
+        database-id (:database query)
+        database (doto (t2/select-one :model/Database :id database-id)
+                   (as-> $ (assert (some? $))))
+        driver (driver.u/database->driver database)
+        native (:query (driver-api/compile-with-inline-parameters query))]
+    (t2/with-transaction [_conn]
+      (assert-unique-display-name! database schema display-name)
+      (let [transform (t2/insert-returning-instance! :model/TransformView
+                                                     {:creator_id creator-id
+                                                      :database_id database-id
+                                                      :dataset_query dataset-query
+                                                      :dataset_query_schema dataset-query-current-schema
+                                                      :dataset_query_type query-type
+                                                      :status :record_created})
+            transform-id (doto (:id transform)
+                           (as-> $ (assert (pos-int? $))))
+            view-name (transform-view-name transform-id)]
+        (assert-unique-name! database schema view-name)
+        (driver/create-view! driver database-id (namespaced-view-name schema view-name) native)
+        ;; TODO: Is it reasonable to do the following in transaction?
+        ;; NB: Following could take a long time or fail, the same way as analysis
+        ;; TODO: All of that should go off thread.
+        (sync/sync-new-table-metadata! database {:table-name view-name
+                                                 :schema schema})
+        (let [view-table (t2/select-one :model/Table
+                                        :db_id database-id
+                                        :schema schema
+                                        :name view-name)]
+          (t2/update! :model/TransformView :id transform-id
+                      {:view_name view-name
+                       :view_schema (:schema view-table)
+                       :status :view_synced})
+          (t2/update! :model/Table :id (:id view-table)
+                      {:display_name display-name
+                       ;; TODO: This should be handled by analysis!!! (and should be async!!!)
+                       :initial_sync_status "complete"
+                       :transform_id transform-id})
+          (t2/select-one :model/TransformView :id transform-id))))))
+
+(defn update-transform!
+  "Update transform's query"
+  [transform-id dataset-query creator]
+  (let [transform (t2/select-one :model/TransformView :id transform-id)
+        view-name (:view_name transform)
+        schema-name (:view_schema transform)
+        namespaced-view-name* (namespaced-view-name schema-name view-name)
+        query (dataset-query->query dataset-query)
+        database-id (:database query)
+        compiled (:query (qp.compile/compile-with-inline-parameters query))
+        database (t2/select-one :model/Database :id database-id)
+        driver (driver.u/database->driver database)]
+    ;; TODO (lbrdnk 2025-07-15): (driver/create-view! ... :replace? true) can not be used because of postgres
+    ;;                           expects same replaced view having columns. We do not. Should we? Imo no. Should we
+    ;;                           ajdust the implementation of create-view! for postgres? Probably.
+    (driver/drop-view! driver database-id namespaced-view-name*)
+    (driver/create-view! driver database-id namespaced-view-name* compiled)
+    (t2/with-transaction [_conn]
+      (t2/update! :model/TransformView :id transform-id
+                  {:dataset_query dataset-query
+                   :creator_id (:id creator)})
+      ;; TODO (lbrdnk 2025-07-15): Add async analysis.
+      (sync/sync-table-metadata! (t2/select-one :model/Table :transform_id transform-id)))))
+
+(defn delete-transform!
+  "Delete transform and view."
+  [id]
+  (let [transform (t2/select-one :model/TransformView :id id)
+        schema-name (:schema transform)
+        view-name (:view_name transform)
+        database-id (:database_id transform)
+        database (t2/select-one :model/Database :id database-id)
+        driver (driver.u/database->driver database)
+        namespaced-view-name* (namespaced-view-name schema-name view-name)]
+    (driver/drop-view! driver database-id namespaced-view-name*)
+    (t2/delete! :model/Table :transform_id id)
+    (t2/delete! :model/TransformView :id id)))
+
+;; TODO (lbrdnk 2025-07-15): pagination, sort, filter, ?perms, table hydration?
+(defn get-transforms
+  "Get transforms"
+  []
+  (t2/select :model/TransformView))
+
+(defn transform-by-id
+  "Transform by id."
+  [id]
+  (t2/select-one :model/TransformView :id id))

src/metabase/warehouse_schema/models/table.clj

@@ -299,7 +299,8 @@
   {:copy      [:name :description :entity_type :active :display_name :visibility_type :schema
                :points_of_interest :caveats :show_in_getting_started :field_order :initial_sync_status :is_upload
                :database_require_filter :is_defective_duplicate :unique_table_helper]
-   :skip      [:estimated_row_count :view_count]
+   ;; TODO (lbrdnk 2025-07-15): Enable serialization of transforms and managed views.
+   :skip      [:estimated_row_count :view_count :transform_id]
    :transform {:created_at (serdes/date)
                :db_id      (serdes/fk :model/Database :name)}})
 

test/metabase/cmd/copy_test.clj

@@ -39,6 +39,7 @@
     :model/QueryTable
     :model/SearchIndexMetadata
     :model/TaskHistory
+    :model/TransformView
     :model/UserKeyValue})
 
 (defn- all-model-names []

test/metabase/sync/sync_test.clj

@@ -123,6 +123,7 @@
     :db_id       true
     :entity_type :entity/GenericTable
     :id          true
+    :transform_id false
     :updated_at  true}))
 
 (defn- field-defaults []

test/metabase/transform/transform_test.clj

@@ -0,0 +1,432 @@
+(ns ^:mb/driver-tests metabase.transform.transform-test
+  "Test for transforms"
+  (:require
+   [clojure.string :as str]
+   [clojure.test :refer :all]
+   [metabase.driver :as driver]
+   [metabase.lib.convert :as lib.convert]
+   [metabase.lib.core :as lib]
+   [metabase.lib.metadata :as lib.metadata]
+   [metabase.query-processor :as qp]
+   [metabase.request.core :as request]
+   [metabase.sync.core :as sync]
+   [metabase.test :as mt]
+   [metabase.test.data.sql :as sql.tx]
+   [metabase.transform.models.transform :as models.transform]
+   [metabase.util :as u]
+   [toucan2.core :as t2]))
+
+(mt/defdataset users-db
+  [["users"
+    [{:field-name "name" :base-type :type/Text}
+     {:field-name "age" :base-type :type/Integer}]
+    [["Foo" 10]
+     ["Bar" 20]
+     ["Baz" 30]]]])
+
+(mt/defdataset products-db
+  [["products"
+    [{:field-name "name" :base-type :type/Text}
+     {:field-name "price" :base-type :type/Decimal}]
+    [["Widget" 10.99]
+     ["Gadget" 25.50]]]])
+
+(mt/defdataset users-departments-db
+  [["users"
+    [{:field-name "name" :base-type :type/Text}
+     {:field-name "department_id" :base-type :type/Integer}]
+    [["Alice" 10]
+     ["Bob" 20]
+     ["Charlie" 10]]]
+   ["departments"
+    [{:field-name "idx" :base-type :type/Integer}
+     {:field-name "name" :base-type :type/Text}]
+    [[10 "Engineering"]
+     [20 "Sales"]]]])
+
+(defn- native-query-rows [query]
+  (set (mt/rows (qp/process-query (mt/native-query {:query query})))))
+
+(deftest create-view-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (testing "Can create and replace a new view"
+      (mt/dataset users-db
+
+        (driver/drop-view! driver/*driver* (u/the-id (mt/db)) "young_users")
+
+        (is (driver/create-view! driver/*driver* (u/the-id (mt/db))
+                                 "young_users"
+                                 "SELECT * FROM users where age < 25"))
+        (is (= #{["Foo"] ["Bar"]}
+               (native-query-rows "SELECT name FROM \"young_users\"")))
+        (is (driver/create-view! driver/*driver* (u/the-id (mt/db))
+                                 "young_users"
+                                 "SELECT * FROM users where age < 35"
+                                 :replace? true))
+        (is (= #{["Foo"] ["Bar"] ["Baz"]}
+               (native-query-rows "SELECT name FROM \"young_users\"")))))))
+
+(deftest drop-view-nonexistent-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (testing "Dropping a non-existent view should not throw an error"
+      (mt/dataset users-db
+        ;; Should not throw even if view doesn't exist
+        (is (driver/drop-view! driver/*driver* (u/the-id (mt/db)) "nonexistent_view"))))))
+
+(deftest create-view-without-replace-fails-if-exists-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (testing "Creating a view that already exists should fail without :replace? true"
+      (mt/dataset products-db
+
+        (driver/drop-view! driver/*driver* (u/the-id (mt/db)) "expensive_products")
+
+        (is (driver/create-view! driver/*driver* (u/the-id (mt/db))
+                                 "expensive_products"
+                                 "SELECT * FROM products WHERE price > 20"))
+
+        (is (thrown? Exception
+                     (driver/create-view! driver/*driver* (u/the-id (mt/db))
+                                          "expensive_products"
+                                          "SELECT * FROM products WHERE price > 15")))))))
+
+(deftest create-view-with-complex-query-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (testing "Creating a view with complex SQL"
+      (mt/dataset users-departments-db
+
+        (driver/drop-view! driver/*driver* (u/the-id (mt/db)) "user_departments")
+
+        (is (driver/create-view! driver/*driver* (u/the-id (mt/db))
+                                 "user_departments"
+                                 "SELECT u.name as user_name, d.name as dept_name FROM users u JOIN departments d ON u.department_id = d.idx"))
+
+        (is (= #{["Alice" "Engineering"] ["Bob" "Sales"] ["Charlie" "Engineering"]}
+               (native-query-rows "SELECT user_name, dept_name FROM \"user_departments\"")))))))
+
+(deftest view-name-edge-cases-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (testing "View names with special characters and edge cases"
+      (mt/dataset users-db
+        (doseq [view-name ["view_with_underscores"
+                           "ViewWithCamelCase"
+                           "view123"
+                           "view with spaces"]]
+          (driver/drop-view! driver/*driver* (u/the-id (mt/db)) view-name)
+
+          (testing (str "View name: " view-name)
+            (is (driver/create-view! driver/*driver* (u/the-id (mt/db)) view-name "SELECT * FROM users"))
+
+            (is (= #{[1 "Foo" 10] [2 "Bar" 20] [3 "Baz" 30]}
+                   (native-query-rows (format "SELECT * FROM \"%s\"" view-name))))))))))
+
+(deftest view-with-invalid-sql-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (testing "Creating a view with invalid SQL should fail"
+      (mt/dataset users-db
+        (is (thrown? Exception
+                     (driver/create-view! driver/*driver* (u/the-id (mt/db))
+                                          "invalid_view"
+                                          "SELECT * FROM nonexistent_table")))))))
+
+(deftest sync-view-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (testing "Can create a new view and immediately sync it"
+      (mt/dataset users-db
+        (mt/with-temp-copy-of-db
+
+          (let [{:keys [schema]} (t2/select-one :model/Table :name "users")
+                base-name "young_users"
+                view-name (if schema
+                            (str schema "." base-name)
+                            base-name)]
+            (driver/drop-view! driver/*driver* (u/the-id (mt/db)) view-name)
+
+            (is (driver/create-view! driver/*driver* (u/the-id (mt/db))
+                                     view-name
+                                     "SELECT name, age FROM users where age < 25"))
+
+            ;; make sure we're indeed creating it anew
+            (t2/delete! :model/Table :name view-name)
+
+            (sync/sync-new-table-metadata! (mt/db) {:table-name base-name :schema schema})
+
+            (let [table (t2/select-one :model/Table :name base-name :schema schema)]
+              (is (=? {:schema schema,
+                       :name base-name,
+                       :active true,
+                       :display_name "Young Users"} table))
+              (is (=? [{:semantic_type nil,
+                        :effective_type :type/Text,
+                        :active true,
+                        :position 0,
+                        :visibility_type :normal,
+                        :display_name "Name",
+                        :database_position 0,
+                        :base_type :type/Text}
+                       {:semantic_type nil,
+                        :effective_type :type/Integer,
+                        :active true,
+                        :position 1,
+                        :visibility_type :normal,
+                        :display_name "Age",
+                        :database_position 1,
+                        :base_type :type/Integer}]
+                      (sort-by :position
+                               (t2/select :model/Field :table_id (u/the-id table))))))))))))
+
+(deftest schema-qualified-view-names-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view :schemas)
+    (testing "Can create views with schema-qualified names"
+      (mt/dataset users-db
+        (let [schema-name (sql.tx/session-schema driver/*driver*)
+              view-name "test_view"
+              qualified-view-name (str schema-name "." view-name)]
+
+          (driver/drop-view! driver/*driver* (u/the-id (mt/db)) qualified-view-name)
+          (is (driver/create-view! driver/*driver* (u/the-id (mt/db)) qualified-view-name "SELECT * FROM users"))
+
+          (is (= #{["Foo"] ["Bar"] ["Baz"]} (native-query-rows (format "SELECT name FROM \"%s\".\"%s\"" schema-name view-name)))))))))
+
+;; Use this only with with-no-transform-views! and in non-parallel tests
+(mt/defdataset users-departments-with-score
+  [["users"
+    [{:field-name "name" :base-type :type/Text}
+     {:field-name "department_id" :base-type :type/Integer}
+     {:field-name "score" :base-type :type/Integer}]
+    [["Alice" 10 100]
+     ["Bob" 20 200]
+     ["Charlie" 10 300]]]
+   ["departments"
+    [{:field-name "idx" :base-type :type/Integer}
+     {:field-name "name" :base-type :type/Text}]
+    [[10 "Engineering"]
+     [20 "Sales"]]]])
+
+(defmacro with-no-transform-views!
+  "Clean up the appdb and dwh prior test exec"
+  [& body]
+  `(do (doseq [table# (:tables (driver/describe-database driver/*driver* (mt/db)))]
+         (let [schema# (:schema table#)
+               name# (:name table#)]
+           (when (str/starts-with? name# "mb_transform")
+             (request/with-current-user (mt/user->id :crowberto)
+               (driver/drop-view! driver/*driver* (mt/id) (str/join "." (filter some? [schema# name#])))))))
+       (t2/delete! :model/Table :name [:like "mb_transform%"])
+       (t2/delete! :model/TransformView :view_name [:like "mb_transform%"])
+       ~@body))
+
+(deftest view-name-length-limit-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (testing "View name length limits are respected"
+      (when-let [limit (driver/view-name-length-limit driver/*driver*)]
+        (is (driver/create-view! driver/*driver* (u/the-id (mt/db))
+                                 (apply str (repeat limit "a"))
+                                 "SELECT * FROM users"
+                                 :replace? true))))))
+
+;; API tests
+
+(deftest can-create-transform-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (mt/dataset
+      users-departments-with-score
+      (with-no-transform-views!
+        (let [dataset-query (let [mp (mt/metadata-provider)]
+                              (-> (lib/query mp (lib.metadata/table mp (mt/id :users)))
+                                  (lib/aggregate (lib/sum (lib.metadata/field mp (mt/id :users :score))))
+                                  (lib/breakout (lib.metadata/field mp (mt/id :users :score)))
+                                  (lib.convert/->legacy-MBQL)))
+              display-name "Test view 1"
+              resp (mt/user-http-request :rasta :post 200 "transform"
+                                         {:display_name display-name
+                                          :dataset_query dataset-query})]
+          (testing "POST /transform creates new view in app db"
+            (is (=? {:id pos-int?
+                     :database_id (mt/id)
+                     :creator_id (mt/user->id :rasta)
+                     :status "view_synced"}
+                    resp)))
+          (let [transform-id (:id resp)
+                transform (t2/select-one :model/TransformView :id transform-id)
+                table (t2/select-one :model/Table :transform_id transform-id)
+                table-id (:id table)]
+            (testing "Transform generated view has user set display name"
+              (is (= (:view_name transform) (:name table)))
+              (is (= display-name (:display_name table))))
+            (testing "Transform generated view can be queried"
+              (let [mp (mt/metadata-provider)]
+                (is (=? {:status :completed}
+                        (-> (lib/query mp (lib.metadata/table mp table-id))
+                            (qp/process-query))))))))))))
+
+(deftest transform-table-name-clash-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (mt/dataset
+      users-departments-with-score
+      (with-no-transform-views!
+        (let [clashing-name "mb_transform_333"
+              clashing-schema "clashing_schema_333"]
+          (mt/with-temp [:model/Table _table {:db_id (mt/id) :schema clashing-schema :name clashing-name}]
+            (with-redefs [models.transform/transform-view-name (constantly clashing-name)]
+              (is (=? {:data {:database-id (mt/id)
+                              :database-name (:name (mt/db))
+                              :name clashing-name
+                              :schema clashing-schema
+                              :same-name-table-ids (comp pos-int? first)}}
+                      (mt/user-http-request :rasta :post 400 "transform"
+                                            {:schema clashing-schema
+                                             :display_name "hello"
+                                             :dataset_query
+                                             (let [mp (mt/metadata-provider)]
+                                               (-> (lib/query mp (lib.metadata/table mp (mt/id :users)))
+                                                   (lib/aggregate (lib/sum (lib.metadata/field mp (mt/id :users :score))))
+                                                   (lib/breakout (lib.metadata/field mp (mt/id :users :department_id)))
+                                                   (lib.convert/->legacy-MBQL)))}))))))))))
+
+(deftest transform-display-name-clash-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (mt/dataset
+      users-departments-with-score
+      (with-no-transform-views!
+        (let [clashing-name "T T T"]
+          (testing "Base: view creation is successfull"
+            (mt/user-http-request :rasta :post 200 "transform"
+                                  {:display_name clashing-name
+                                   :dataset_query
+                                   (let [mp (mt/metadata-provider)]
+                                     (-> (lib/query mp (lib.metadata/table mp (mt/id :users)))
+                                         (lib.convert/->legacy-MBQL)))}))
+          (testing "Creation of duplicate display name view fails"
+            (is (=? {:database-id (mt/id)
+                     :database-name (:name (mt/db))
+                     :display-name clashing-name
+                     :same-display-name-table-ids (comp pos-int? first)}
+                    (mt/user-http-request :rasta :post 400 "transform"
+                                          {:display_name clashing-name
+                                           :dataset_query
+                                           (let [mp (mt/metadata-provider)]
+                                             (-> (lib/query mp (lib.metadata/table mp (mt/id :users)))
+                                                 (lib.convert/->legacy-MBQL)))})))
+            (is (= 1 (t2/count :model/Table :display_name clashing-name)))))))))
+
+;; schemas -- ensure that picking of existing schema creates in that schema
+
+(deftest put-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (mt/dataset
+      users-departments-with-score
+      (with-no-transform-views!
+        (let [display-name "t r a n s"
+              transform (mt/user-http-request :rasta :post 200 "transform"
+                                              {:display_name display-name
+                                               :dataset_query
+                                               (let [mp (mt/metadata-provider)]
+                                                 (-> (lib/query mp (lib.metadata/table mp (mt/id :users)))
+                                                     (lib.convert/->legacy-MBQL)))})
+              table (t2/select-one :model/Table :display_name display-name)
+              table-id (:id table)]
+          (testing "Base: view creation and sync is successful"
+            (is (= 1 (t2/count :model/Table :display_name display-name)))
+            (is (= #{"id" "name" "department_id" "score"}
+                   (t2/select-fn-set (comp u/lower-case-en :name) :model/Field
+                                     :table_id table-id
+                                     :active true))))
+          (let [mp (mt/metadata-provider)]
+            (is (= [[1 "Alice" 10 100] [2 "Bob" 20 200] [3 "Charlie" 10 300]]
+                   (mt/rows (-> (lib/query mp (lib.metadata/table mp table-id))
+                                (qp/process-query))))))
+          (testing "PUT / can modify the query"
+            (mt/user-http-request :rasta :put 200 (str "transform/" (:id transform))
+                                  {:dataset_query
+                                   (let [mp (mt/metadata-provider)]
+                                     (-> (lib/query mp (lib.metadata/table mp (mt/id :departments)))
+                                         (lib.convert/->legacy-MBQL)))})
+            (is (= 1 (t2/count :model/Table :display_name display-name)))
+            (is (= #{"idx" "name" "id"}
+                   (t2/select-fn-set (comp u/lower-case-en :name) :model/Field
+                                     :table_id table-id
+                                     :active true)))
+            (let [mp (mt/metadata-provider)]
+              (is (= [[1 10 "Engineering"] [2 20 "Sales"]]
+                     (mt/rows (-> (lib/query mp (lib.metadata/table mp table-id))
+                                  (qp/process-query))))))))))))
+
+(deftest delete-transform-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (mt/dataset
+      users-departments-with-score
+      (with-no-transform-views!
+        (let [mp (mt/metadata-provider)
+              query (-> (lib/query mp (lib.metadata/table mp (mt/id :users)))
+                        (lib/aggregate (lib/sum (lib.metadata/field mp (mt/id :users :score))))
+                        (lib/breakout (lib.metadata/field mp (mt/id :users :department_id))))
+              display-name "My shiny transfrom"
+              resp (mt/user-http-request :rasta :post 200 "transform"
+                                         {:display_name display-name
+                                          :dataset_query (lib.convert/->legacy-MBQL query)})
+              transform-id (:id resp)]
+          (testing "Base: Transform was successfully created"
+            (is (= 1 (t2/count :model/TransformView :id transform-id)))
+            (is (= 1 (t2/count :model/Table :transform_id transform-id))))
+          (testing "Transform was deleted"
+            (mt/user-http-request :rasta :delete 200 (str "transform/" transform-id))
+            ;; Ensure transform was removed from dwh, hence is not picked up by sync that follows
+            (sync/sync-database! (mt/db))
+            (is (empty? (filter #(str/starts-with? % models.transform/view-name-prefix)
+                                (t2/select-fn-vec :name :model/Table :db_id (mt/id)))))
+            (is (empty? (t2/select :model/TransformView :id transform-id)))))))))
+
+(deftest get-transform-test
+  (mt/test-drivers
+    (mt/normal-drivers-with-feature :view)
+    (mt/dataset
+      users-departments-with-score
+      (with-no-transform-views!
+        (let [mp (mt/metadata-provider)
+              query (-> (lib/query mp (lib.metadata/table mp (mt/id :users)))
+                        (lib/aggregate (lib/sum (lib.metadata/field mp (mt/id :users :score))))
+                        (lib/breakout (lib.metadata/field mp (mt/id :users :department_id))))
+              display-name "Transform strikes back"
+              resp (mt/user-http-request :rasta :post 200 "transform"
+                                         {:display_name display-name
+                                          :dataset_query (lib.convert/->legacy-MBQL query)})
+              transform-id (:id resp)
+              query-2 (-> (lib/query mp (lib.metadata/table mp (mt/id :users)))
+                          (lib/aggregate (lib/sum (lib.metadata/field mp (mt/id :users :score))))
+                          (lib/breakout (lib.metadata/field mp (mt/id :users :department_id))))
+              display-name-2 "Another one"
+              resp-2 (mt/user-http-request :rasta :post 200 "transform"
+                                           {:display_name display-name-2
+                                            :dataset_query (lib.convert/->legacy-MBQL query-2)})
+              transform-id-2 (:id resp-2)]
+          (testing "Base: Transforms were successfully created"
+            (is (= 1 (t2/count :model/TransformView :id transform-id)))
+            (is (= 1 (t2/count :model/Table :transform_id transform-id)))
+            (is (= 1 (t2/count :model/TransformView :id transform-id-2)))
+            (is (= 1 (t2/count :model/Table :transform_id transform-id-2))))
+          (testing "GET /"
+            (let [resp (mt/user-http-request :rasta :get 200 "transform")
+                  tables (t2/select :model/Table :transform_id [:in (mapv :id resp)])]
+              (is (= 2 (count resp)))
+              (is (= 2 (count tables)))
+              (is (some (comp #{display-name} :display_name) tables))
+              (is (some (comp #{display-name-2} :display_name) tables))))
+          (testing "GET /:id"
+            (is (= transform-id
+                   (:id (mt/user-http-request :rasta :get 200 (str "transform/" transform-id)))))
+            (is (= transform-id-2
+                   (:id (mt/user-http-request :rasta :get 200 (str "transform/" transform-id-2)))))))))))