feat: Distinct operator for queries (#244)

Co-authored-by: Sam Willis <sam.willis@gmail.com>

Author: kevin-dp

.changeset/slick-ghosts-shine.md

@@ -0,0 +1,5 @@
+---
+"@tanstack/db": patch
+---
+
+New distinct operator for queries.

packages/db/package.json

@@ -3,7 +3,7 @@
   "description": "A reactive client store for building super fast apps on sync",
   "version": "0.0.21",
   "dependencies": {
-    "@electric-sql/d2mini": "^0.1.6",
+    "@electric-sql/d2mini": "^0.1.7",
     "@standard-schema/spec": "^1.0.0"
   },
   "devDependencies": {

packages/db/src/query/builder/index.ts

@@ -480,6 +480,27 @@ export class BaseQueryBuilder<TContext extends Context = Context> {
     }) as any
   }
 
+  /**
+   * Specify that the query should return distinct rows.
+   * Deduplicates rows based on the selected columns.
+   * @returns A QueryBuilder with distinct enabled
+   *
+   * @example
+   * ```ts
+   * // Get countries our users are from
+   * query
+   *   .from({ users: usersCollection })
+   *   .select(({users}) => users.country)
+   *   .distinct()
+   * ```
+   */
+  distinct(): QueryBuilder<TContext> {
+    return new BaseQueryBuilder({
+      ...this.query,
+      distinct: true,
+    }) as any
+  }
+
   // Helper methods
   private _getCurrentAliases(): Array<string> {
     const aliases: Array<string> = []

packages/db/src/query/compiler/index.ts

@@ -1,4 +1,4 @@
-import { filter, map } from "@electric-sql/d2mini"
+import { distinct, filter, map } from "@electric-sql/d2mini"
 import { compileExpression } from "./evaluators.js"
 import { processJoins } from "./joins.js"
 import { processGroupBy } from "./group-by.js"
@@ -98,8 +98,12 @@ export function compileQuery(
     }
   }
 
+  if (query.distinct && !query.fnSelect && !query.select) {
+    throw new Error(`DISTINCT requires a SELECT clause.`)
+  }
+
   // Process the SELECT clause early - always create __select_results
-  // This eliminates duplication and allows for future DISTINCT implementation
+  // This eliminates duplication and allows for DISTINCT implementation
   if (query.fnSelect) {
     // Handle functional select - apply the function to transform the row
     pipeline = pipeline.pipe(
@@ -190,6 +194,11 @@ export function compileQuery(
     }
   }
 
+  // Process the DISTINCT clause if it exists
+  if (query.distinct) {
+    pipeline = pipeline.pipe(distinct(([_key, row]) => row.__select_results))
+  }
+
   // Process orderBy parameter if it exists
   if (query.orderBy && query.orderBy.length > 0) {
     const orderedPipeline = processOrderBy(

packages/db/src/query/ir.ts

@@ -15,6 +15,7 @@ export interface QueryIR {
   orderBy?: OrderBy
   limit?: Limit
   offset?: Offset
+  distinct?: true
 
   // Functional variants
   fnSelect?: (row: NamespacedRow) => any