diff --git a/.changeset/calm-ducks-report.md b/.changeset/calm-ducks-report.md
new file mode 100644
index 000000000..7e4040242
--- /dev/null
+++ b/.changeset/calm-ducks-report.md
@@ -0,0 +1,5 @@
+---
+'@electric-sql/pglite-react': patch
+---
+
+Expose live query initialization errors through an optional `onError` callback.
diff --git a/.changeset/live-query-options.md b/.changeset/live-query-options.md
new file mode 100644
index 000000000..3851850ee
--- /dev/null
+++ b/.changeset/live-query-options.md
@@ -0,0 +1,6 @@
+---
+'@electric-sql/pglite': patch
+'@electric-sql/pglite-react': patch
+---
+
+Allow `live.query` and React's `useLiveQuery` hook to accept query options such as `rowMode: 'array'`.
diff --git a/packages/pglite-react/src/hooks.ts b/packages/pglite-react/src/hooks.ts
index 5bf90d313..130b727f3 100644
--- a/packages/pglite-react/src/hooks.ts
+++ b/packages/pglite-react/src/hooks.ts
@@ -1,8 +1,14 @@
 import type { LiveQuery, LiveQueryResults } from '@electric-sql/pglite/live'
+import type { QueryOptions } from '@electric-sql/pglite'
 import { query as buildQuery } from '@electric-sql/pglite/template'
 import { useEffect, useRef, useState } from 'react'
 import { usePGlite } from './provider'
 
+export interface UseLiveQueryOptions extends QueryOptions {
+  /** Called when the initial live query setup rejects. */
+  onError?: (error: Error) => void
+}
+
 function paramsEqual(
   a1: unknown[] | undefined | null,
   a2: unknown[] | undefined | null,
@@ -17,13 +23,56 @@ function paramsEqual(
   return true
 }
 
+function shallowRecordsEqual(a: object | undefined, b: object | undefined) {
+  if (!a && !b) return true
+  if (!a || !b) return false
+  const aRecord = a as Record<PropertyKey, unknown>
+  const bRecord = b as Record<PropertyKey, unknown>
+  const aKeys = Reflect.ownKeys(a)
+  const bKeys = Reflect.ownKeys(b)
+  return (
+    aKeys.length === bKeys.length &&
+    aKeys.every((key) => Object.is(aRecord[key], bRecord[key]))
+  )
+}
+
+function queryOptionsEqual(
+  a: QueryOptions | undefined,
+  b: QueryOptions | undefined,
+) {
+  if (!a && !b) return true
+  if (!a || !b) return false
+  return (
+    a.rowMode === b.rowMode &&
+    shallowRecordsEqual(a.parsers, b.parsers) &&
+    shallowRecordsEqual(a.serializers, b.serializers) &&
+    Object.is(a.blob, b.blob) &&
+    Object.is(a.onNotice, b.onNotice) &&
+    paramsEqual(a.paramTypes, b.paramTypes)
+  )
+}
+
+function getQueryOptions(
+  options: UseLiveQueryOptions | undefined,
+): QueryOptions | undefined {
+  if (!options) return undefined
+  const queryOptions = { ...options }
+  delete queryOptions.onError
+  return Reflect.ownKeys(queryOptions).length > 0 ? queryOptions : undefined
+}
+
 function useLiveQueryImpl<T = { [key: string]: unknown }>(
   query: string | LiveQuery<T> | Promise<LiveQuery<T>>,
   params: unknown[] | undefined | null,
   key?: string,
+  options?: UseLiveQueryOptions,
 ): Omit<LiveQueryResults<T>, 'affectedRows'> | undefined {
   const db = usePGlite()
   const paramsRef = useRef(params)
+  const queryOptions = getQueryOptions(options)
+  const optionsRef = useRef(queryOptions)
+  const onErrorRef = useRef(options?.onError)
+  onErrorRef.current = options?.onError
   const liveQueryRef = useRef<LiveQuery<T> | undefined>(undefined)
   let liveQuery: LiveQuery<T> | undefined
   let liveQueryChanged = false
@@ -42,6 +91,12 @@ function useLiveQueryImpl<T = { [key: string]: unknown }>(
     currentParams = params
   }
 
+  let currentOptions = optionsRef.current
+  if (!queryOptionsEqual(optionsRef.current, queryOptions)) {
+    optionsRef.current = queryOptions
+    currentOptions = queryOptions
+  }
+
   /* eslint-disable @eslint-react/hooks-extra/no-direct-set-state-in-use-effect */
   useEffect(() => {
     let cancelled = false
@@ -50,14 +105,36 @@ function useLiveQueryImpl<T = { [key: string]: unknown }>(
       setResults(results)
     }
     if (typeof query === 'string') {
+      let unsubscribe: (() => Promise<void>) | undefined
       const ret =
         key !== undefined
           ? db.live.incrementalQuery<T>(query, currentParams, key, cb)
-          : db.live.query<T>(query, currentParams, cb)
+          : currentOptions
+            ? db.live.query<T>(query, currentParams, currentOptions, cb)
+            : db.live.query<T>(query, currentParams, cb)
+
+      void ret.then(
+        ({ unsubscribe: resolvedUnsubscribe }) => {
+          if (cancelled) {
+            void resolvedUnsubscribe()
+            return
+          }
+          unsubscribe = resolvedUnsubscribe
+        },
+        (error: Error) => {
+          if (cancelled) return
+          const onError = onErrorRef.current
+          if (onError) {
+            onError(error)
+          } else {
+            throw error
+          }
+        },
+      )
 
       return () => {
         cancelled = true
-        ret.then(({ unsubscribe }) => unsubscribe())
+        void unsubscribe?.()
       }
     } else if (query instanceof Promise) {
       query.then((liveQuery) => {
@@ -80,7 +157,7 @@ function useLiveQueryImpl<T = { [key: string]: unknown }>(
     } else {
       throw new Error('Should never happen')
     }
-  }, [db, key, query, currentParams, liveQuery])
+  }, [db, key, query, currentParams, currentOptions, liveQuery])
   /* eslint-enable @eslint-react/hooks-extra/no-direct-set-state-in-use-effect */
 
   if (liveQueryChanged && liveQuery) {
@@ -101,6 +178,7 @@ function useLiveQueryImpl<T = { [key: string]: unknown }>(
 export function useLiveQuery<T = { [key: string]: unknown }>(
   query: string,
   params?: unknown[] | null,
+  options?: UseLiveQueryOptions,
 ): LiveQueryResults<T> | undefined
 
 export function useLiveQuery<T = { [key: string]: unknown }>(
@@ -114,8 +192,9 @@ export function useLiveQuery<T = { [key: string]: unknown }>(
 export function useLiveQuery<T = { [key: string]: unknown }>(
   query: string | LiveQuery<T> | Promise<LiveQuery<T>>,
   params?: unknown[] | null,
+  options?: UseLiveQueryOptions,
 ): LiveQueryResults<T> | undefined {
-  return useLiveQueryImpl<T>(query, params)
+  return useLiveQueryImpl<T>(query, params, undefined, options)
 }
 
 useLiveQuery.sql = function <T = { [key: string]: unknown }>(
diff --git a/packages/pglite-react/test/hooks-options.test.tsx b/packages/pglite-react/test/hooks-options.test.tsx
new file mode 100644
index 000000000..887dbc881
--- /dev/null
+++ b/packages/pglite-react/test/hooks-options.test.tsx
@@ -0,0 +1,94 @@
+import { act, renderHook } from '@testing-library/react'
+import { waitFor } from '@testing-library/dom'
+import { describe, expect, it, vi } from 'vitest'
+import type { LiveQueryResults } from '@electric-sql/pglite/live'
+import { useLiveQuery } from '../src/hooks'
+
+const { usePGliteMock } = vi.hoisted(() => ({
+  usePGliteMock: vi.fn(),
+}))
+
+vi.mock('../src/provider', () => ({
+  usePGlite: usePGliteMock,
+}))
+
+describe('useLiveQuery query options', () => {
+  it('passes options to live.query for initial and updated results', async () => {
+    type Row = [number, string]
+    let callback: ((results: LiveQueryResults<Row>) => void) | undefined
+    const initialResults: LiveQueryResults<Row> = {
+      rows: [[1, 'initial']],
+      fields: [
+        { name: 'id', dataTypeID: 23 },
+        { name: 'name', dataTypeID: 25 },
+      ],
+    }
+    const query = vi.fn(async (...args: unknown[]) => {
+      callback = args.find(
+        (arg): arg is (results: LiveQueryResults<Row>) => void =>
+          typeof arg === 'function',
+      )
+      callback?.(initialResults)
+      return {
+        initialResults,
+        subscribe: vi.fn(),
+        unsubscribe: vi.fn(),
+        refresh: vi.fn(),
+      }
+    })
+    usePGliteMock.mockReturnValue({ live: { query } })
+
+    const { result } = renderHook(() =>
+      useLiveQuery<Row>('SELECT id, name FROM test', [], {
+        rowMode: 'array',
+      }),
+    )
+
+    await waitFor(() => expect(result.current).toEqual(initialResults))
+    expect(query).toHaveBeenCalledTimes(1)
+    expect(query).toHaveBeenCalledWith(
+      'SELECT id, name FROM test',
+      [],
+      { rowMode: 'array' },
+      expect.any(Function),
+    )
+
+    act(() => {
+      callback?.({
+        ...initialResults,
+        rows: [
+          [1, 'initial'],
+          [2, 'updated'],
+        ],
+      })
+    })
+
+    expect(result.current?.rows).toEqual([
+      [1, 'initial'],
+      [2, 'updated'],
+    ])
+  })
+
+  it('reports query initialization errors through onError', async () => {
+    const queryError = new Error('syntax error at or near "table"')
+    const onError = vi.fn()
+    const query = vi.fn(() => Promise.reject(queryError))
+    usePGliteMock.mockReturnValue({ live: { query } })
+
+    const { unmount } = renderHook(() =>
+      useLiveQuery('SELECT FROM table', [], { onError }),
+    )
+
+    await waitFor(() => expect(onError).toHaveBeenCalledOnce())
+    expect(onError).toHaveBeenCalledWith(queryError)
+    expect(query).toHaveBeenCalledWith(
+      'SELECT FROM table',
+      [],
+      expect.any(Function),
+    )
+
+    unmount()
+    await Promise.resolve()
+    expect(onError).toHaveBeenCalledOnce()
+  })
+})
diff --git a/packages/pglite-react/test/hooks.test-d.tsx b/packages/pglite-react/test/hooks.test-d.tsx
new file mode 100644
index 000000000..258825fb8
--- /dev/null
+++ b/packages/pglite-react/test/hooks.test-d.tsx
@@ -0,0 +1,26 @@
+import { describe, it, expectTypeOf } from 'vitest'
+import type { QueryOptions } from '@electric-sql/pglite'
+import type { LiveQueryOptions } from '@electric-sql/pglite/live'
+import { useLiveQuery } from '../src'
+
+describe('useLiveQuery types', () => {
+  it('accepts exported query options in object and positional APIs', () => {
+    const queryOptions: QueryOptions = { rowMode: 'array' }
+    const liveOptions: LiveQueryOptions<[number, string]> = {
+      query: 'SELECT id, name FROM test',
+      ...queryOptions,
+    }
+
+    expectTypeOf(liveOptions.rowMode).toEqualTypeOf<QueryOptions['rowMode']>()
+    ;() =>
+      useLiveQuery<[number, string]>(
+        'SELECT id, name FROM test',
+        [],
+        queryOptions,
+      )
+    ;() =>
+      useLiveQuery('SELECT FROM table', [], {
+        onError: (error) => expectTypeOf(error).toEqualTypeOf<Error>(),
+      })
+  })
+})
diff --git a/packages/pglite/src/live/index.ts b/packages/pglite/src/live/index.ts
index ca65f2cb8..e059184df 100644
--- a/packages/pglite/src/live/index.ts
+++ b/packages/pglite/src/live/index.ts
@@ -3,6 +3,7 @@ import type {
   PGliteInterface,
   Results,
   Transaction,
+  QueryOptions,
 } from '../interface'
 import type {
   LiveQueryOptions,
@@ -23,6 +24,7 @@ export type {
   LiveChanges,
   Change,
   LiveQueryResults,
+  LiveQueryOptions,
 } from './interface.js'
 
 const MAX_RETRIES = 5
@@ -36,18 +38,44 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
     async query<T>(
       query: string | LiveQueryOptions<T>,
       params?: any[] | null,
+      optionsOrCallback?: QueryOptions | ((results: Results<T>) => void),
       callback?: (results: Results<T>) => void,
     ) {
       let signal: AbortSignal | undefined
       let offset: number | undefined
       let limit: number | undefined
+      let options: QueryOptions | undefined
       if (typeof query !== 'string') {
-        signal = query.signal
-        params = query.params
-        callback = query.callback
-        offset = query.offset
-        limit = query.limit
-        query = query.query
+        const {
+          signal: querySignal,
+          params: queryParams,
+          callback: queryCallback,
+          offset: queryOffset,
+          limit: queryLimit,
+          query: queryString,
+          ...queryOptions
+        } = query
+        signal = querySignal
+        params = queryParams
+        callback = queryCallback
+        offset = queryOffset
+        limit = queryLimit
+        options = queryOptions
+        query = queryString
+      } else if (typeof optionsOrCallback === 'function') {
+        callback = optionsOrCallback
+      } else {
+        options = optionsOrCallback
+      }
+
+      // The prepared EXECUTE has no bind parameters, and live-query metadata
+      // queries rely on object rows. Only pass options that apply to the user's
+      // result rows or notices.
+      const resultOptions: QueryOptions | undefined = options && {
+        rowMode: options.rowMode,
+        parsers: options.parsers,
+        blob: options.blob,
+        onNotice: options.onNotice,
       }
 
       // Offset and limit must be provided together
@@ -82,7 +110,7 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
           // Create a temporary view with the query
           const formattedQuery =
             params && params.length > 0
-              ? await formatQuery(pg, query, params, tx)
+              ? await formatQuery(pg, query, params, tx, options)
               : query
           await tx.exec(
             `CREATE OR REPLACE TEMP VIEW live_query_${id}_view AS ${formattedQuery}`,
@@ -110,6 +138,8 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
             results = {
               ...(await tx.query<T>(
                 `EXECUTE live_query_${id}_get(${limit}, ${offset});`,
+                undefined,
+                resultOptions,
               )),
               offset,
               limit,
@@ -120,7 +150,11 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
               PREPARE live_query_${id}_get AS
               SELECT * FROM live_query_${id}_view;
             `)
-            results = await tx.query<T>(`EXECUTE live_query_${id}_get;`)
+            results = await tx.query<T>(
+              `EXECUTE live_query_${id}_get;`,
+              undefined,
+              resultOptions,
+            )
           }
           // Setup the listeners
           unsubList = await Promise.all(
@@ -178,13 +212,19 @@ const setup = async (pg: PGliteInterface, _emscriptenOpts: any) => {
                 results = {
                   ...(await pg.query<T>(
                     `EXECUTE live_query_${id}_get(${limit}, ${offset});`,
+                    undefined,
+                    resultOptions,
                   )),
                   offset,
                   limit,
                   totalCount, // This is the old total count
                 }
               } else {
-                results = await pg.query<T>(`EXECUTE live_query_${id}_get;`)
+                results = await pg.query<T>(
+                  `EXECUTE live_query_${id}_get;`,
+                  undefined,
+                  resultOptions,
+                )
               }
             } catch (e) {
               const msg = (e as Error).message
diff --git a/packages/pglite/src/live/interface.ts b/packages/pglite/src/live/interface.ts
index f39c9c2df..9973c0960 100644
--- a/packages/pglite/src/live/interface.ts
+++ b/packages/pglite/src/live/interface.ts
@@ -1,6 +1,7 @@
-import type { Results } from '../interface'
+import type { QueryOptions, Results } from '../interface'
 
-export interface LiveQueryOptions<T = { [key: string]: any }> {
+export interface LiveQueryOptions<T = { [key: string]: any }>
+  extends QueryOptions {
   query: string
   params?: any[] | null
   offset?: number
@@ -40,6 +41,22 @@ export interface LiveNamespace {
     callback?: (results: Results<T>) => void,
   ): Promise<LiveQuery<T>>
 
+  /**
+   * Create a live query with query options
+   * @param query - The query to run
+   * @param params - The parameters to pass to the query
+   * @param options - The options to apply to the query results
+   * @param callback - A callback to run when the query is updated
+   * @returns A promise that resolves to an object with the initial results,
+   * an unsubscribe function, and a refresh function
+   */
+  query<T = { [key: string]: any }>(
+    query: string,
+    params: any[] | undefined | null,
+    options: QueryOptions,
+    callback?: (results: Results<T>) => void,
+  ): Promise<LiveQuery<T>>
+
   /**
    * Create a live query
    * @param options - The options to pass to the query
diff --git a/packages/pglite/src/utils.ts b/packages/pglite/src/utils.ts
index aa688a8cd..d552c91f5 100644
--- a/packages/pglite/src/utils.ts
+++ b/packages/pglite/src/utils.ts
@@ -1,4 +1,4 @@
-import type { PGliteInterface, Transaction } from './interface.js'
+import type { PGliteInterface, QueryOptions, Transaction } from './interface.js'
 import { serialize as serializeProtocol } from '@electric-sql/pg-protocol'
 import { parseDescribeStatementResults } from './parse.js'
 import { TEXT } from './types.js'
@@ -19,6 +19,7 @@ export async function formatQuery(
   query: string,
   params?: any[] | null,
   tx?: Transaction | PGliteInterface,
+  options?: Pick<QueryOptions, 'paramTypes' | 'serializers'>,
 ) {
   if (!params || params.length === 0) {
     // no params so no formatting needed
@@ -30,9 +31,10 @@ export async function formatQuery(
   // Get the types of the parameters
   const messages = []
   try {
-    await pg.execProtocol(serializeProtocol.parse({ text: query }), {
-      syncToFs: false,
-    })
+    await pg.execProtocol(
+      serializeProtocol.parse({ text: query, types: options?.paramTypes }),
+      { syncToFs: false },
+    )
 
     messages.push(
       ...(
@@ -50,20 +52,55 @@ export async function formatQuery(
 
   const dataTypeIDs = parseDescribeStatementResults(messages)
 
+  let parameterTypeNames: string[] | undefined
+  if (options?.paramTypes?.length) {
+    const typeNameResult = await tx.query<Record<string, string>>(
+      `SELECT ${dataTypeIDs
+        .map(
+          (_, index) => `format_type($${index + 1}::oid, NULL) AS "${index}"`,
+        )
+        .join(', ')}`,
+      dataTypeIDs,
+    )
+    parameterTypeNames = dataTypeIDs.map(
+      (_, index) => typeNameResult.rows[0][index],
+    )
+  }
+
   // replace $1, $2, etc with %1$L, %2$L, etc
   // The `$` in `%n$L` is required for positional arguments; a bare `%nL` is
   // "min width n" and makes format() consume arguments sequentially, binding
   // the wrong values when placeholders are out of order or repeated.
   const subbedQuery = query.replace(/\$([0-9]+)/g, (_, num) => {
-    return '%' + num + '$L'
+    const typeName = parameterTypeNames?.[Number(num) - 1]
+    const placeholder = '%' + num + '$L'
+    return typeName ? `(${placeholder})::${typeName}` : placeholder
+  })
+
+  const serializedParams = params.map((param, index) => {
+    const serialize = options?.serializers?.[dataTypeIDs[index]]
+    return serialize && param !== null && param !== undefined
+      ? serialize(param)
+      : param
   })
+  const passthroughSerializers = options?.serializers
+    ? Object.fromEntries(
+        Object.keys(options.serializers).map((oid) => [
+          oid,
+          (param: any) => String(param),
+        ]),
+      )
+    : undefined
 
   const ret = await tx.query<{
     query: string
   }>(
     `SELECT format($1, ${params.map((_, i) => `$${i + 2}`).join(', ')}) as query`,
-    [subbedQuery, ...params],
-    { paramTypes: [TEXT, ...dataTypeIDs] },
+    [subbedQuery, ...serializedParams],
+    {
+      paramTypes: [TEXT, ...dataTypeIDs],
+      serializers: passthroughSerializers,
+    },
   )
   return ret.rows[0].query
 }
diff --git a/packages/pglite/tests/live.test.ts b/packages/pglite/tests/live.test.ts
index afd3f3cf3..2ecd1e4e9 100644
--- a/packages/pglite/tests/live.test.ts
+++ b/packages/pglite/tests/live.test.ts
@@ -128,6 +128,123 @@ await testEsmCjsAndDTC(async (importType) => {
       ])
     })
 
+    it('applies query options to initial and updated live query results', async () => {
+      await db.exec(`
+        CREATE TABLE live_query_options_test (
+          id SERIAL PRIMARY KEY,
+          number INT
+        );
+        INSERT INTO live_query_options_test (number) VALUES (10), (20);
+      `)
+
+      let updatedResults
+      const eventTarget = new EventTarget()
+      const { initialResults, unsubscribe } = await db.live.query<
+        [number, number]
+      >({
+        query: 'SELECT * FROM live_query_options_test ORDER BY number',
+        offset: 0,
+        limit: 2,
+        rowMode: 'array',
+        paramTypes: [23],
+        callback: (results) => {
+          updatedResults = results
+          eventTarget.dispatchEvent(new Event('change'))
+        },
+      })
+
+      expect(initialResults.rows).toEqual([
+        [1, 10],
+        [2, 20],
+      ])
+      expect(initialResults.totalCount).toBe(2)
+
+      const updated = new Promise((resolve) =>
+        eventTarget.addEventListener('change', resolve, { once: true }),
+      )
+      await db.exec('INSERT INTO live_query_options_test (number) VALUES (15)')
+      await updated
+
+      expect(updatedResults.rows).toEqual([
+        [1, 10],
+        [3, 15],
+      ])
+
+      await unsubscribe()
+    })
+
+    it('uses query option serializers for live query params', async () => {
+      const value = { text: 'serialized' }
+      const { initialResults, unsubscribe } = await db.live.query<{
+        value: string
+      }>({
+        query: 'SELECT $1::text AS value',
+        params: [value],
+        serializers: {
+          25: (param: typeof value) => param.text,
+        },
+      })
+
+      expect(initialResults.rows).toEqual([{ value: 'serialized' }])
+
+      await unsubscribe()
+    })
+
+    it('lets query option serializers override default serializers', async () => {
+      const serializerDb = await PGlite.create({
+        extensions: { live },
+        serializers: {
+          1700: (param) => String(Number(param) + 1),
+        },
+      })
+
+      try {
+        const { initialResults, unsubscribe } = await serializerDb.live.query<{
+          value: string
+        }>({
+          query: 'SELECT ($1::numeric)::text AS value',
+          params: [3],
+          serializers: {
+            1700: (param) => String(Number(param) * 2),
+          },
+        })
+
+        expect(initialResults.rows).toEqual([{ value: '6' }])
+
+        await unsubscribe()
+      } finally {
+        await serializerDb.close()
+      }
+    })
+
+    it('uses explicit param types when formatting live query params', async () => {
+      const { initialResults, unsubscribe } = await db.live.query<{
+        type: string
+        value: number
+      }>({
+        query: 'SELECT pg_typeof($1)::text AS type, $1 AS value',
+        params: [42],
+        paramTypes: [23],
+      })
+
+      expect(initialResults.rows).toEqual([{ type: 'integer', value: 42 }])
+
+      await unsubscribe()
+    })
+
+    it('attaches a query option blob to live query execution', async () => {
+      const { initialResults, unsubscribe } = await db.live.query<{
+        value: string
+      }>({
+        query: "SELECT pg_read_file('/dev/blob') AS value",
+        blob: new Blob(['blob value']),
+      })
+
+      expect(initialResults.rows).toEqual([{ value: 'blob value' }])
+
+      await unsubscribe()
+    })
+
     it('live query on view', async () => {
       await db.exec(`
         CREATE TABLE IF NOT EXISTS testTable (