diff --git a/packages/backend-core/src/sql/sql.ts b/packages/backend-core/src/sql/sql.ts
index 3585dacbed..ed8dc929d6 100644
--- a/packages/backend-core/src/sql/sql.ts
+++ b/packages/backend-core/src/sql/sql.ts
@@ -139,29 +139,61 @@ class InternalBuilder {
     return this.table.schema[column]
   }
 
-  // Takes a string like foo and returns a quoted string like [foo] for SQL Server
-  // and "foo" for Postgres.
-  private quote(str: string): string {
+  private quoteChars(): [string, string] {
     switch (this.client) {
-      case SqlClient.SQL_LITE:
       case SqlClient.ORACLE:
       case SqlClient.POSTGRES:
-        return `"${str}"`
+        return ['"', '"']
       case SqlClient.MS_SQL:
-        return `[${str}]`
+        return ["[", "]"]
       case SqlClient.MARIADB:
       case SqlClient.MY_SQL:
-        return `\`${str}\``
+      case SqlClient.SQL_LITE:
+        return ["`", "`"]
     }
   }
 
-  // Takes a string like a.b.c and returns a quoted identifier like [a].[b].[c]
-  // for SQL Server and `a`.`b`.`c` for MySQL.
-  private quotedIdentifier(key: string): string {
-    return key
-      .split(".")
-      .map(part => this.quote(part))
-      .join(".")
+  // Takes a string like foo and returns a quoted string like [foo] for SQL Server
+  // and "foo" for Postgres.
+  private quote(str: string): string {
+    const [start, end] = this.quoteChars()
+    return `${start}${str}${end}`
+  }
+
+  private isQuoted(key: string): boolean {
+    const [start, end] = this.quoteChars()
+    return key.startsWith(start) && key.endsWith(end)
+  }
+
+  // Takes a string like a.b.c or an array like ["a", "b", "c"] and returns a
+  // quoted identifier like [a].[b].[c] for SQL Server and `a`.`b`.`c` for
+  // MySQL.
+  private quotedIdentifier(key: string | string[]): string {
+    if (!Array.isArray(key)) {
+      key = this.splitIdentifier(key)
+    }
+    return key.map(part => this.quote(part)).join(".")
+  }
+
+  // Turns an identifier like a.b.c or `a`.`b`.`c` into ["a", "b", "c"]
+  private splitIdentifier(key: string): string[] {
+    const [start, end] = this.quoteChars()
+    if (this.isQuoted(key)) {
+      return key.slice(1, -1).split(`${end}.${start}`)
+    }
+    return key.split(".")
+  }
+
+  private qualifyIdentifier(key: string): string {
+    const tableName = this.getTableName()
+    const parts = this.splitIdentifier(key)
+    if (parts[0] !== tableName) {
+      parts.unshift(tableName)
+    }
+    if (this.isQuoted(key)) {
+      return this.quotedIdentifier(parts)
+    }
+    return parts.join(".")
   }
 
   private isFullSelectStatementRequired(): boolean {
@@ -231,8 +263,13 @@ class InternalBuilder {
   // OracleDB can't use character-large-objects (CLOBs) in WHERE clauses,
   // so when we use them we need to wrap them in to_char(). This function
   // converts a field name to the appropriate identifier.
-  private convertClobs(field: string): string {
-    const parts = field.split(".")
+  private convertClobs(field: string, opts?: { forSelect?: boolean }): string {
+    if (this.client !== SqlClient.ORACLE) {
+      throw new Error(
+        "you've called convertClobs on a DB that's not Oracle, this is a mistake"
+      )
+    }
+    const parts = this.splitIdentifier(field)
     const col = parts.pop()!
     const schema = this.table.schema[col]
     let identifier = this.quotedIdentifier(field)
@@ -244,7 +281,11 @@ class InternalBuilder {
       schema.type === FieldType.OPTIONS ||
       schema.type === FieldType.BARCODEQR
     ) {
-      identifier = `to_char(${identifier})`
+      if (opts?.forSelect) {
+        identifier = `to_char(${identifier}) as ${this.quotedIdentifier(col)}`
+      } else {
+        identifier = `to_char(${identifier})`
+      }
     }
     return identifier
   }
@@ -859,28 +900,58 @@ class InternalBuilder {
     const fields = this.query.resource?.fields || []
     const tableName = this.getTableName()
     if (fields.length > 0) {
-      query = query.groupBy(fields.map(field => `${tableName}.${field}`))
-      query = query.select(fields.map(field => `${tableName}.${field}`))
+      const qualifiedFields = fields.map(field => this.qualifyIdentifier(field))
+      if (this.client === SqlClient.ORACLE) {
+        const groupByFields = qualifiedFields.map(field =>
+          this.convertClobs(field)
+        )
+        const selectFields = qualifiedFields.map(field =>
+          this.convertClobs(field, { forSelect: true })
+        )
+        query = query
+          .groupByRaw(groupByFields.join(", "))
+          .select(this.knex.raw(selectFields.join(", ")))
+      } else {
+        query = query.groupBy(qualifiedFields).select(qualifiedFields)
+      }
     }
     for (const aggregation of aggregations) {
       const op = aggregation.calculationType
-      const field = `${tableName}.${aggregation.field} as ${aggregation.name}`
-      switch (op) {
-        case CalculationType.COUNT:
-          query = query.count(field)
-          break
-        case CalculationType.SUM:
-          query = query.sum(field)
-          break
-        case CalculationType.AVG:
-          query = query.avg(field)
-          break
-        case CalculationType.MIN:
-          query = query.min(field)
-          break
-        case CalculationType.MAX:
-          query = query.max(field)
-          break
+      if (op === CalculationType.COUNT) {
+        if ("distinct" in aggregation && aggregation.distinct) {
+          if (this.client === SqlClient.ORACLE) {
+            const field = this.convertClobs(`${tableName}.${aggregation.field}`)
+            query = query.select(
+              this.knex.raw(
+                `COUNT(DISTINCT ${field}) as ${this.quotedIdentifier(
+                  aggregation.name
+                )}`
+              )
+            )
+          } else {
+            query = query.countDistinct(
+              `${tableName}.${aggregation.field} as ${aggregation.name}`
+            )
+          }
+        } else {
+          query = query.count(`* as ${aggregation.name}`)
+        }
+      } else {
+        const field = `${tableName}.${aggregation.field} as ${aggregation.name}`
+        switch (op) {
+          case CalculationType.SUM:
+            query = query.sum(field)
+            break
+          case CalculationType.AVG:
+            query = query.avg(field)
+            break
+          case CalculationType.MIN:
+            query = query.min(field)
+            break
+          case CalculationType.MAX:
+            query = query.max(field)
+            break
+        }
       }
     }
     return query
diff --git a/packages/server/src/api/controllers/row/ExternalRequest.ts b/packages/server/src/api/controllers/row/ExternalRequest.ts
index 4ce326c35f..ed8836626c 100644
--- a/packages/server/src/api/controllers/row/ExternalRequest.ts
+++ b/packages/server/src/api/controllers/row/ExternalRequest.ts
@@ -696,9 +696,8 @@ export class ExternalRequest<T extends Operation> {
       const calculationFields = helpers.views.calculationFields(this.source)
       for (const [key, field] of Object.entries(calculationFields)) {
         aggregations.push({
+          ...field,
           name: key,
-          field: field.field,
-          calculationType: field.calculationType,
         })
       }
     }
diff --git a/packages/server/src/api/controllers/view/viewsV2.ts b/packages/server/src/api/controllers/view/viewsV2.ts
index 0257c86ded..a0ffc5c460 100644
--- a/packages/server/src/api/controllers/view/viewsV2.ts
+++ b/packages/server/src/api/controllers/view/viewsV2.ts
@@ -11,14 +11,40 @@ import {
   ViewCalculationFieldMetadata,
   RelationSchemaField,
   ViewFieldMetadata,
+  CalculationType,
 } from "@budibase/types"
 import { builderSocket, gridSocket } from "../../../websockets"
 import { helpers } from "@budibase/shared-core"
 
 function stripUnknownFields(
-  field: BasicViewFieldMetadata
-): RequiredKeys<BasicViewFieldMetadata> {
+  field: ViewFieldMetadata
+): RequiredKeys<ViewFieldMetadata> {
   if (helpers.views.isCalculationField(field)) {
+    if (field.calculationType === CalculationType.COUNT) {
+      if ("distinct" in field && field.distinct) {
+        return {
+          order: field.order,
+          width: field.width,
+          visible: field.visible,
+          readonly: field.readonly,
+          icon: field.icon,
+          distinct: field.distinct,
+          calculationType: field.calculationType,
+          field: field.field,
+          columns: field.columns,
+        }
+      } else {
+        return {
+          order: field.order,
+          width: field.width,
+          visible: field.visible,
+          readonly: field.readonly,
+          icon: field.icon,
+          calculationType: field.calculationType,
+          columns: field.columns,
+        }
+      }
+    }
     const strippedField: RequiredKeys<ViewCalculationFieldMetadata> = {
       order: field.order,
       width: field.width,
diff --git a/packages/server/src/api/routes/tests/viewV2.spec.ts b/packages/server/src/api/routes/tests/viewV2.spec.ts
index 83b26bc687..a1a4475ee5 100644
--- a/packages/server/src/api/routes/tests/viewV2.spec.ts
+++ b/packages/server/src/api/routes/tests/viewV2.spec.ts
@@ -24,8 +24,8 @@ import {
   RenameColumn,
   FeatureFlag,
   BBReferenceFieldSubType,
+  NumericCalculationFieldMetadata,
   ViewV2Schema,
-  ViewCalculationFieldMetadata,
 } from "@budibase/types"
 import { generator, mocks } from "@budibase/backend-core/tests"
 import { DatabaseName, getDatasource } from "../../../integrations/tests/utils"
@@ -557,13 +557,13 @@ describe.each([
 
         expect(Object.keys(view.schema!)).toHaveLength(1)
 
-        let sum = view.schema!.sum as ViewCalculationFieldMetadata
+        let sum = view.schema!.sum as NumericCalculationFieldMetadata
         expect(sum).toBeDefined()
         expect(sum.calculationType).toEqual(CalculationType.SUM)
         expect(sum.field).toEqual("Price")
 
         view = await config.api.viewV2.get(view.id)
-        sum = view.schema!.sum as ViewCalculationFieldMetadata
+        sum = view.schema!.sum as NumericCalculationFieldMetadata
         expect(sum).toBeDefined()
         expect(sum.calculationType).toEqual(CalculationType.SUM)
         expect(sum.field).toEqual("Price")
@@ -864,6 +864,185 @@ describe.each([
             }
           )
         })
+
+      !isLucene &&
+        describe("calculation views", () => {
+          let table: Table
+          let view: ViewV2
+
+          beforeEach(async () => {
+            table = await config.api.table.save(
+              saveTableRequest({
+                schema: {
+                  name: {
+                    name: "name",
+                    type: FieldType.STRING,
+                    constraints: {
+                      presence: true,
+                    },
+                  },
+                  country: {
+                    name: "country",
+                    type: FieldType.STRING,
+                  },
+                  age: {
+                    name: "age",
+                    type: FieldType.NUMBER,
+                  },
+                },
+              })
+            )
+
+            view = await config.api.viewV2.create({
+              tableId: table._id!,
+              name: generator.guid(),
+              schema: {
+                country: {
+                  visible: true,
+                },
+                age: {
+                  visible: true,
+                  calculationType: CalculationType.SUM,
+                  field: "age",
+                },
+              },
+            })
+
+            await config.api.row.bulkImport(table._id!, {
+              rows: [
+                {
+                  name: "Steve",
+                  age: 30,
+                  country: "UK",
+                },
+                {
+                  name: "Jane",
+                  age: 31,
+                  country: "UK",
+                },
+                {
+                  name: "Ruari",
+                  age: 32,
+                  country: "USA",
+                },
+                {
+                  name: "Alice",
+                  age: 33,
+                  country: "USA",
+                },
+              ],
+            })
+          })
+
+          it("returns the expected rows prior to modification", async () => {
+            const { rows } = await config.api.row.search(view.id)
+            expect(rows).toHaveLength(2)
+            expect(rows).toEqual(
+              expect.arrayContaining([
+                {
+                  country: "USA",
+                  age: 65,
+                },
+                {
+                  country: "UK",
+                  age: 61,
+                },
+              ])
+            )
+          })
+
+          it("can remove a group by field", async () => {
+            delete view.schema!.country
+            await config.api.viewV2.update(view)
+
+            const { rows } = await config.api.row.search(view.id)
+            expect(rows).toHaveLength(1)
+            expect(rows).toEqual(
+              expect.arrayContaining([
+                {
+                  age: 126,
+                },
+              ])
+            )
+          })
+
+          it("can remove a calculation field", async () => {
+            delete view.schema!.age
+            await config.api.viewV2.update(view)
+
+            const { rows } = await config.api.row.search(view.id)
+            expect(rows).toHaveLength(4)
+
+            // Because the removal of the calculation field actually makes this
+            // no longer a calculation view, these rows will now have _id and
+            // _rev fields.
+            expect(rows).toEqual(
+              expect.arrayContaining([
+                expect.objectContaining({ country: "UK" }),
+                expect.objectContaining({ country: "UK" }),
+                expect.objectContaining({ country: "USA" }),
+                expect.objectContaining({ country: "USA" }),
+              ])
+            )
+          })
+
+          it("can add a new group by field", async () => {
+            view.schema!.name = { visible: true }
+            await config.api.viewV2.update(view)
+
+            const { rows } = await config.api.row.search(view.id)
+            expect(rows).toHaveLength(4)
+            expect(rows).toEqual(
+              expect.arrayContaining([
+                {
+                  name: "Steve",
+                  age: 30,
+                  country: "UK",
+                },
+                {
+                  name: "Jane",
+                  age: 31,
+                  country: "UK",
+                },
+                {
+                  name: "Ruari",
+                  age: 32,
+                  country: "USA",
+                },
+                {
+                  name: "Alice",
+                  age: 33,
+                  country: "USA",
+                },
+              ])
+            )
+          })
+
+          it("can add a new calculation field", async () => {
+            view.schema!.count = {
+              visible: true,
+              calculationType: CalculationType.COUNT,
+            }
+            await config.api.viewV2.update(view)
+
+            const { rows } = await config.api.row.search(view.id)
+            expect(rows).toHaveLength(2)
+            expect(rows).toEqual(
+              expect.arrayContaining([
+                {
+                  country: "USA",
+                  age: 65,
+                  count: 2,
+                },
+                {
+                  country: "UK",
+                  age: 61,
+                  count: 2,
+                },
+              ])
+            )
+          })
+        })
     })
 
     describe("delete", () => {
@@ -2570,6 +2749,74 @@ describe.each([
               expect(actual).toEqual(expected)
             }
           })
+
+          it("should be able to do a COUNT(DISTINCT)", async () => {
+            const table = await config.api.table.save(
+              saveTableRequest({
+                schema: {
+                  name: {
+                    name: "name",
+                    type: FieldType.STRING,
+                  },
+                },
+              })
+            )
+
+            const view = await config.api.viewV2.create({
+              tableId: table._id!,
+              name: generator.guid(),
+              schema: {
+                count: {
+                  visible: true,
+                  calculationType: CalculationType.COUNT,
+                  distinct: true,
+                  field: "name",
+                },
+              },
+            })
+
+            await config.api.row.bulkImport(table._id!, {
+              rows: [
+                {
+                  name: "John",
+                },
+                {
+                  name: "John",
+                },
+                {
+                  name: "Sue",
+                },
+              ],
+            })
+
+            const { rows } = await config.api.row.search(view.id)
+            expect(rows).toHaveLength(1)
+            expect(rows[0].count).toEqual(2)
+          })
+
+          it("should not be able to COUNT(DISTINCT ...) against a non-existent field", async () => {
+            await config.api.viewV2.create(
+              {
+                tableId: table._id!,
+                name: generator.guid(),
+                schema: {
+                  count: {
+                    visible: true,
+                    calculationType: CalculationType.COUNT,
+                    distinct: true,
+                    field: "does not exist oh no",
+                  },
+                },
+              },
+              {
+                status: 400,
+                body: {
+                  message:
+                    'Calculation field "count" references field "does not exist oh no" which does not exist in the table schema',
+                },
+              }
+            )
+          })
         })
 
       !isLucene &&
diff --git a/packages/server/src/sdk/app/rows/search/internal/sqs.ts b/packages/server/src/sdk/app/rows/search/internal/sqs.ts
index f0e47ce2b7..3847eb8f31 100644
--- a/packages/server/src/sdk/app/rows/search/internal/sqs.ts
+++ b/packages/server/src/sdk/app/rows/search/internal/sqs.ts
@@ -1,5 +1,6 @@
 import {
   Aggregation,
+  CalculationType,
   Datasource,
   DocumentType,
   FieldType,
@@ -369,11 +370,27 @@ export async function search(
         continue
       }
 
-      aggregations.push({
-        name: key,
-        field: mapToUserColumn(field.field),
-        calculationType: field.calculationType,
-      })
+      if (field.calculationType === CalculationType.COUNT) {
+        if ("distinct" in field && field.distinct) {
+          aggregations.push({
+            name: key,
+            distinct: true,
+            field: mapToUserColumn(field.field),
+            calculationType: field.calculationType,
+          })
+        } else {
+          aggregations.push({
+            name: key,
+            calculationType: field.calculationType,
+          })
+        }
+      } else {
+        aggregations.push({
+          name: key,
+          field: mapToUserColumn(field.field),
+          calculationType: field.calculationType,
+        })
+      }
     }
   }
 
diff --git a/packages/server/src/sdk/app/views/index.ts b/packages/server/src/sdk/app/views/index.ts
index 83ee78e165..73785edd98 100644
--- a/packages/server/src/sdk/app/views/index.ts
+++ b/packages/server/src/sdk/app/views/index.ts
@@ -1,4 +1,5 @@
 import {
+  CalculationType,
   FieldType,
   PermissionLevel,
   RelationSchemaField,
@@ -65,6 +66,15 @@ async function guardCalculationViewSchema(
   const calculationFields = helpers.views.calculationFields(view)
   for (const calculationFieldName of Object.keys(calculationFields)) {
     const schema = calculationFields[calculationFieldName]
+    const isCount = schema.calculationType === CalculationType.COUNT
+    const isDistinct = isCount && "distinct" in schema && schema.distinct
+
+    // Count fields that aren't distinct don't need to reference another field,
+    // so we don't validate it.
+    if (isCount && !isDistinct) {
+      continue
+    }
+
     const targetSchema = table.schema[schema.field]
     if (!targetSchema) {
       throw new HTTPError(
@@ -73,7 +83,7 @@ async function guardCalculationViewSchema(
       )
     }
 
-    if (!helpers.schema.isNumeric(targetSchema)) {
+    if (!isCount && !helpers.schema.isNumeric(targetSchema)) {
       throw new HTTPError(
         `Calculation field "${calculationFieldName}" references field "${schema.field}" which is not a numeric field`,
         400
diff --git a/packages/types/src/documents/app/view.ts b/packages/types/src/documents/app/view.ts
index 77bbb3e4d1..5fcc9049c1 100644
--- a/packages/types/src/documents/app/view.ts
+++ b/packages/types/src/documents/app/view.ts
@@ -42,11 +42,31 @@ export interface RelationSchemaField extends UIFieldMetadata {
   readonly?: boolean
 }
 
-export interface ViewCalculationFieldMetadata extends BasicViewFieldMetadata {
-  calculationType: CalculationType
+export interface NumericCalculationFieldMetadata
+  extends BasicViewFieldMetadata {
+  calculationType:
+    | CalculationType.MIN
+    | CalculationType.MAX
+    | CalculationType.SUM
+    | CalculationType.AVG
   field: string
 }
 
+export interface CountCalculationFieldMetadata extends BasicViewFieldMetadata {
+  calculationType: CalculationType.COUNT
+}
+
+export interface CountDistinctCalculationFieldMetadata
+  extends CountCalculationFieldMetadata {
+  distinct: true
+  field: string
+}
+
+export type ViewCalculationFieldMetadata =
+  | NumericCalculationFieldMetadata
+  | CountCalculationFieldMetadata
+  | CountDistinctCalculationFieldMetadata
+
 export type ViewFieldMetadata =
   | BasicViewFieldMetadata
   | ViewCalculationFieldMetadata
diff --git a/packages/types/src/sdk/row.ts b/packages/types/src/sdk/row.ts
index 0b63e88229..ea46e82508 100644
--- a/packages/types/src/sdk/row.ts
+++ b/packages/types/src/sdk/row.ts
@@ -3,12 +3,33 @@ import { SearchFilters } from "./search"
 import { CalculationType, Row } from "../documents"
 import { WithRequired } from "../shared"
 
-export interface Aggregation {
+export interface BaseAggregation {
   name: string
-  calculationType: CalculationType
+}
+
+export interface NumericAggregation extends BaseAggregation {
+  calculationType:
+    | CalculationType.AVG
+    | CalculationType.MAX
+    | CalculationType.MIN
+    | CalculationType.SUM
   field: string
 }
 
+export interface CountAggregation extends BaseAggregation {
+  calculationType: CalculationType.COUNT
+}
+
+export interface CountDistinctAggregation extends CountAggregation {
+  distinct: true
+  field: string
+}
+
+export type Aggregation =
+  | NumericAggregation
+  | CountAggregation
+  | CountDistinctAggregation
+
 export interface SearchParams {
   tableId?: string
   viewId?: string