Advanced Methods - 🌲 Sonamu

Puri provides various advanced query methods including aggregation, modification, and vector search.

Aggregation Queries

groupBy

Groups results.

const stats = await puri
  .table("orders")
  .select({
    user_id: "orders.user_id",
    total_orders: Puri.count(),
    total_amount: Puri.sum("orders.amount"),
  })
  .groupBy("orders.user_id");

// GROUP BY orders.user_id

distinct

Returns results with duplicates removed. Uses the SQL DISTINCT clause.

// Remove duplicate rows
const categories = await puri
  .table("products")
  .distinct("products.category")
  .select({ category: "products.category" });

// Multiple columns
const pairs = await puri.table("orders").distinct("orders.user_id", "orders.status").select({
  user_id: "orders.user_id",
  status: "orders.status",
});

having

Filters grouped results.

const activeUsers = await puri
  .table("orders")
  .select({
    user_id: "orders.user_id",
    order_count: Puri.count(),
  })
  .groupBy("orders.user_id")
  .having("order_count", ">", 10);

// HAVING order_count > 10

Data Modification

insert

Inserts new records.

// Single insert
const [id] = await puri.table("users").insert({
  email: "john@example.com",
  name: "John Doe",
});

// Batch insert
const ids = await puri.table("users").insert([
  { email: "john@example.com", name: "John" },
  { email: "jane@example.com", name: "Jane" },
]);

update

Updates records.

const count = await puri.table("users").where("users.id", 1).update({
  name: "John Smith",
  updated_at: new Date(),
});

console.log(`${count} rows updated`);

delete

Deletes records.

const count = await puri.table("users").where("users.status", "inactive").delete();

console.log(`${count} rows deleted`);

increment / decrement

Increments/decrements numeric columns.

// Increment views
await puri.table("posts").where("posts.id", 123).increment("posts.views", 1);

// Decrement stock
await puri.table("products").where("products.id", 456).decrement("products.stock", 5);

Row Locking

Acquire locks on SELECTed rows within a transaction to prevent concurrent modifications by other transactions.

forUpdate

Acquires an exclusive lock on selected rows, preventing other transactions from modifying or locking those rows.

@transactional()
async deductBalance(userId: number, amount: number) {
  const wdb = this.getPuri("w");

  const user = await wdb
    .table("users")
    .select({ id: "users.id", balance: "users.balance" })
    .where("users.id", userId)
    .forUpdate()
    .first();

  if (!user || user.balance < amount) {
    throw new Error("Insufficient balance");
  }

  await wdb.table("users").where("users.id", userId).update({
    balance: user.balance - amount,
  });
}

// SELECT users.id, users.balance FROM users WHERE users.id = ? FOR UPDATE

forShare

Acquires a shared lock on selected rows, preventing other transactions from modifying those rows while still allowing reads.

@transactional()
async getBalanceSnapshot(userId: number) {
  const wdb = this.getPuri("w");

  const user = await wdb
    .table("users")
    .select({ id: "users.id", balance: "users.balance" })
    .where("users.id", userId)
    .forShare()
    .first();

  return user;
}

// SELECT users.id, users.balance FROM users WHERE users.id = ? FOR SHARE

forUpdate vs forShare:

Lock Type	Plain SELECT	Other Lock Requests (FOR UPDATE / FOR SHARE)	Writes (UPDATE / DELETE)	When to Use
`forUpdate()`	Allowed	Blocked	Blocked	When reading then modifying
`forShare()`	Allowed	FOR UPDATE only	Blocked	When only read consistency is needed

Under PostgreSQL MVCC, row-level locks (FOR UPDATE / FOR SHARE) do not block plain SELECT reads. Only other lock requests and write operations on the same rows are blocked.

forUpdate() and forShare() are only meaningful within a transaction. Without a transaction, the lock is released immediately.

Fetching Results

first

Fetches only the first record.

const user = await puri
  .table("users")
  .where("users.email", "john@example.com")
  .select({ id: "users.id", name: "users.name" })
  .first();

// user: { id: number, name: string } | undefined

pluck

Fetches only specific column values as an array.

const userIds = await puri.table("users").where("users.status", "active").pluck("users.id");

// userIds: number[]
// [1, 2, 3, 4, 5]

const emails = await puri.table("users").pluck("users.email");

// emails: string[]
// ["john@example.com", "jane@example.com", ...]

Vector Similarity Search

vectorSimilarity

Vector similarity search using pgvector.

const queryEmbedding = [0.1, 0.2, 0.3 /* ... */]; // 1536 dimensions

const results = await puri
  .table("documents")
  .vectorSimilarity("documents.embedding", queryEmbedding, {
    method: "cosine", // or "l2", "inner_product"
    threshold: 0.7, // Only similarity >= 0.7
  })
  .select({
    id: "documents.id",
    title: "documents.title",
    similarity: "similarity", // Automatically added
  })
  .limit(10);

// ORDER BY documents.embedding <=> '[0.1,0.2,0.3,...]' ASC

Options:

method: Similarity measurement method
- cosine: Cosine similarity (0~1, higher is more similar)
- l2: Euclidean distance (lower is more similar)
- inner_product: Inner product (higher is more similar)
threshold: Similarity filtering threshold
distinctOn: Returns only the most similar result per unique column value

distinctOn Option

The distinctOn option allows you to retrieve only the most similar result for each unique value in a specified column. It leverages PostgreSQL’s DISTINCT ON clause.

// Get the most similar document per category
const results = await puri
  .table("documents")
  .vectorSimilarity("documents.embedding", queryEmbedding, {
    method: "cosine",
    distinctOn: "documents.category_id", // One most similar document per category
  })
  .select({
    id: "documents.id",
    title: "documents.title",
    category_id: "documents.category_id",
    similarity: "similarity",
  })
  .limit(10);

You can combine distinctOn with threshold:

// Get the most similar document per author with similarity >= 0.6
const results = await puri
  .table("documents")
  .vectorSimilarity("documents.embedding", queryEmbedding, {
    method: "cosine",
    distinctOn: "documents.author_id",
    threshold: 0.6,
  })
  .limit(5);

When using the distinctOn option, the query is internally wrapped in a subquery and sorted by similarity in descending order in the outer query.

Upsert (INSERT or UPDATE)

onConflict

Specifies action on conflict.

// DO NOTHING
await puri
  .table("users")
  .insert({ email: "john@example.com", name: "John" })
  .onConflict(["email"], "nothing");

// INSERT ... ON CONFLICT (email) DO NOTHING

// DO UPDATE
await puri
  .table("users")
  .insert({ email: "john@example.com", name: "John Smith" })
  .onConflict(["email"], {
    update: ["name"], // Update only name
  });

// INSERT ... ON CONFLICT (email) DO UPDATE SET name = EXCLUDED.name

// Object form
await puri
  .table("users")
  .insert({ email: "john@example.com", count: 1 })
  .onConflict(["email"], {
    update: {
      count: Puri.rawNumber("users.count + 1"), // SQL expression
      updated_at: new Date(),
    },
  });

returning

Returns inserted/updated records.

// All columns
const users = await puri
  .table("users")
  .insert({ email: "john@example.com", name: "John" })
  .returning("*");

// Specific columns
const [user] = await puri
  .table("users")
  .insert({ email: "john@example.com", name: "John" })
  .returning(["id", "email"]);

console.log(user.id, user.email);

Utility Methods

clone

Clones a query.

const baseQuery = puri.table("users").where("users.status", "active");

// Clone and add different conditions
const admins = await baseQuery
  .clone()
  .where("users.role", "admin")
  .select({ id: "users.id", name: "users.name" });

const editors = await baseQuery
  .clone()
  .where("users.role", "editor")
  .select({ id: "users.id", name: "users.name" });

debug

Prints generated SQL to console.

const users = await puri
  .table("users")
  .where("users.status", "active")
  .select({ id: "users.id", name: "users.name" })
  .debug(); // Print SQL

// [Puri Debug] SELECT users.id, users.name FROM users WHERE users.status = 'active'

toQuery

Returns SQL query string.

const query = puri
  .table("users")
  .where("users.status", "active")
  .select({ id: "users.id", name: "users.name" });

const sql = query.toQuery();
console.log(sql);
// "SELECT users.id, users.name FROM users WHERE users.status = 'active'"

raw

Executes raw SQL.

const result = await puri.raw(
  `
  SELECT * FROM users WHERE status = ?
`,
  ["active"],
);

Real-World Examples

async function getUserStatistics() {
  return puri.table("users")
    .leftJoin("orders", "users.id", "orders.user_id")
    .select({
      user_id: "users.id",
      user_name: "users.name",

      // Aggregation
      order_count: Puri.count("orders.id"),
      total_spent: Puri.sum("orders.amount"),
      avg_order: Puri.avg("orders.amount"),

      // First/last order
      first_order: Puri.min("orders.created_at"),
      last_order: Puri.max("orders.created_at")
    })
    .groupBy("users.id", "users.name")
    .having("order_count", ">", 0)
    .orderBy("total_spent", "desc");
}

async function updateInactivUsers(days: number = 30) {
  const cutoffDate = new Date();
  cutoffDate.setDate(cutoffDate.getDate() - days);

  const count = await puri.table("users")
    .where("users.last_login", "<", cutoffDate)
    .where("users.status", "active")
    .update({
      status: "inactive",
      updated_at: new Date()
    });

  console.log(`${count} users marked as inactive`);
  return count;
}

async function upsertUserStats(userId: number, stats: {
  login_count: number;
  last_login: Date;
}) {
  await puri.table("user_stats")
    .insert({
      user_id: userId,
      login_count: stats.login_count,
      last_login: stats.last_login,
      created_at: new Date()
    })
    .onConflict(["user_id"], {
      update: {
        login_count: Puri.rawNumber("user_stats.login_count + 1"),
        last_login: stats.last_login,
        updated_at: new Date()
      }
    })
    .returning("*");
}

async function semanticSearch(query: string, limit: number = 10) {
  // Convert query to embedding (e.g., OpenAI API)
  const queryEmbedding = await getEmbedding(query);

  return puri.table("documents")
    // Vector similarity search
    .vectorSimilarity("documents.embedding", queryEmbedding, {
      method: "cosine",
      threshold: 0.5  // Similarity >= 0.5
    })
    .where("documents.published", true)
    .select({
      id: "documents.id",
      title: "documents.title",
      content: "documents.content",
      similarity: "similarity"
    })
    .limit(limit);
}

async function transferPoints(
  fromUserId: number,
  toUserId: number,
  points: number
) {
  const wdb = getPuri("w");

  await wdb.transaction(async (trx) => {
    // Deduct points
    await trx.table("users")
      .where("users.id", fromUserId)
      .decrement("users.points", points);

    // Add points
    await trx.table("users")
      .where("users.id", toUserId)
      .increment("users.points", points);

    // Record history
    await trx.table("point_history")
      .insert({
        from_user_id: fromUserId,
        to_user_id: toUserId,
        points,
        created_at: new Date()
      });
  });
}

async function searchProducts(filters: {
  category?: string;
  minPrice?: number;
  maxPrice?: number;
  inStock?: boolean;
  query?: string;
}) {
  let query = puri.table("products")
    .select({
      id: "products.id",
      name: "products.name",
      price: "products.price",
      stock: "products.stock"
    });

  // Conditional filtering
  if (filters.category) {
    query = query.where("products.category", filters.category);
  }

  if (filters.minPrice !== undefined) {
    query = query.where("products.price", ">=", filters.minPrice);
  }

  if (filters.maxPrice !== undefined) {
    query = query.where("products.price", "<=", filters.maxPrice);
  }

  if (filters.inStock) {
    query = query.where("products.stock", ">", 0);
  }

  if (filters.query) {
    query = query.where("products.name", "like", `%${filters.query}%`);
  }

  return query
    .orderBy("products.created_at", "desc")
    .limit(50);
}

Performance Optimization

Batch INSERT

// ✅ Good: Batch at once
await puri.table("users").insert([
  { email: "user1@example.com", name: "User 1" },
  { email: "user2@example.com", name: "User 2" },
  // ... 1000 items
]);

// ❌ Bad: One by one in loop
for (const user of users) {
  await puri.table("users").insert(user);
}

increment vs UPDATE

// ✅ Good: increment (atomic, fast)
await puri.table("posts").where("posts.id", 123).increment("posts.views", 1);

// ❌ Bad: SELECT → UPDATE (race condition, slow)
const post = await puri.table("posts").where("posts.id", 123).first();
await puri
  .table("posts")
  .where("posts.id", 123)
  .update({ views: post.views + 1 });

HAVING vs WHERE

// ✅ Good: Filter with WHERE first
await puri
  .table("orders")
  .where("orders.status", "completed") // Filter before GROUP BY
  .select({
    user_id: "orders.user_id",
    total: Puri.sum("orders.amount"),
  })
  .groupBy("orders.user_id")
  .having("total", ">", 1000); // Filter after GROUP BY

// ❌ Inefficient: All HAVING
await puri
  .table("orders")
  .select({
    user_id: "orders.user_id",
    total: Puri.sum("orders.amount"),
  })
  .groupBy("orders.user_id")
  .having("total", ">", 1000)
  .having(/* status condition */); // Inefficient

Cautions

1. increment/decrement Values

// ✅ Correct: Positive only
.increment("column", 1)
.decrement("column", 5)

// ❌ Error: 0 or below
.increment("column", 0)   // Error
.increment("column", -1)  // Error

2. onConflict Constraints

// Table needs UNIQUE constraint or PRIMARY KEY
CREATE UNIQUE INDEX idx_users_email ON users(email);

// For onConflict to work
.onConflict(["email"], ...)

3. returning

.insert({ ... })
.returning("*")

4. vectorSimilarity Requires pgvector

-- PostgreSQL extension required
CREATE EXTENSION vector;

-- Create vector column
ALTER TABLE documents
ADD COLUMN embedding vector(1536);

-- Create HNSW index
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops);

Next Steps

select

Select fields to retrieve

where

Filter conditions

UpsertBuilder

Handle bulk upserts

Transactions

Use transactions

​Aggregation Queries

​groupBy

​distinct

​having

​Data Modification

​insert

​update

​delete

​increment / decrement

​Row Locking

​forUpdate

​forShare

​Fetching Results

​first

​pluck

​Vector Similarity Search

​vectorSimilarity

​distinctOn Option

​Upsert (INSERT or UPDATE)

​onConflict

​returning

​Utility Methods

​clone

​debug

​toQuery

​raw

​Real-World Examples

​Performance Optimization

​Batch INSERT

​increment vs UPDATE

​HAVING vs WHERE

​Cautions

​1. increment/decrement Values

​2. onConflict Constraints

​3. returning

​4. vectorSimilarity Requires pgvector

​Next Steps

select

where

UpsertBuilder

Transactions

Aggregation Queries

groupBy

distinct

having

Data Modification

insert

update

delete

increment / decrement

Row Locking

forUpdate

forShare

Fetching Results

first

pluck

Vector Similarity Search

vectorSimilarity

distinctOn Option

Upsert (INSERT or UPDATE)

onConflict

returning

Utility Methods

clone

debug

toQuery

raw

Real-World Examples

Performance Optimization

Batch INSERT

increment vs UPDATE

HAVING vs WHERE

Cautions

1. increment/decrement Values

2. onConflict Constraints

3. returning

4. vectorSimilarity Requires pgvector

Next Steps