diff --git a/lib/sea/SeaAuth.ts b/lib/sea/SeaAuth.ts
index 0c393430..c6fd5178 100644
--- a/lib/sea/SeaAuth.ts
+++ b/lib/sea/SeaAuth.ts
@@ -49,26 +49,47 @@ const U2M_DEFAULT_REDIRECT_PORT = 8030;
  * incompatible with `isolatedModules` and a runtime-coupling hazard.
  * The Rust source of truth lives at `native/sea/src/database.rs`.
  */
-export type SeaNativeConnectionOptions =
-  | {
-      hostName: string;
-      httpPath: string;
-      authMode: 'Pat';
-      token: string;
-    }
-  | {
-      hostName: string;
-      httpPath: string;
-      authMode: 'OAuthM2m';
-      oauthClientId: string;
-      oauthClientSecret: string;
-    }
-  | {
-      hostName: string;
-      httpPath: string;
-      authMode: 'OAuthU2m';
-      oauthRedirectPort: number;
-    };
+/**
+ * Session-level defaults shared across all auth-mode variants.
+ *
+ * Mirrors `ConnectionOptions.catalog` / `.schema` / `.sessionConf` on
+ * the napi binding (kernel `Session::builder().defaults(DefaultOpts)`
+ * and `.session_conf(HashMap)` — the routes that actually populate SEA
+ * `CreateSession.catalog` / `.schema` / `.session_confs`).
+ *
+ * Per-statement overrides do not exist on the kernel surface; both
+ * pyo3 and napi expose catalog / schema / sessionConf only at session
+ * creation. Mirror that here so the adapter doesn't promise a
+ * capability the binding can't honour.
+ */
+export interface SeaSessionDefaults {
+  catalog?: string;
+  schema?: string;
+  sessionConf?: Record<string, string>;
+}
+
+export type SeaNativeConnectionOptions = SeaSessionDefaults &
+  (
+    | {
+        hostName: string;
+        httpPath: string;
+        authMode: 'Pat';
+        token: string;
+      }
+    | {
+        hostName: string;
+        httpPath: string;
+        authMode: 'OAuthM2m';
+        oauthClientId: string;
+        oauthClientSecret: string;
+      }
+    | {
+        hostName: string;
+        httpPath: string;
+        authMode: 'OAuthU2m';
+        oauthRedirectPort: number;
+      }
+  );
 
 function prependSlash(str: string): string {
   if (str.length > 0 && str.charAt(0) !== '/') {
diff --git a/lib/sea/SeaBackend.ts b/lib/sea/SeaBackend.ts
index 998796fa..61b1a333 100644
--- a/lib/sea/SeaBackend.ts
+++ b/lib/sea/SeaBackend.ts
@@ -89,28 +89,36 @@ export default class SeaBackend implements IBackend {
       throw new HiveDriverError('SeaBackend: not connected. Call connect() first.');
     }
 
+    // Fold session-level defaults from the OpenSessionRequest into the
+    // napi `ConnectionOptions`. The kernel routes these through
+    // `Session::builder().defaults(DefaultOpts)` + `.session_conf(...)`
+    // so they land on the SEA `CreateSession` wire fields, not on each
+    // per-statement request. Matches pyo3's `Session.__new__` shape.
+    //
+    // Only set the optional keys when present so the napi call shape
+    // stays minimal — keeps wire snapshots / test assertions stable
+    // for callers who pass no defaults.
+    const sessionOptions: SeaNativeConnectionOptions = { ...this.nativeOptions };
+    if (request.initialCatalog !== undefined) {
+      sessionOptions.catalog = request.initialCatalog;
+    }
+    if (request.initialSchema !== undefined) {
+      sessionOptions.schema = request.initialSchema;
+    }
+    if (request.configuration !== undefined) {
+      sessionOptions.sessionConf = { ...request.configuration };
+    }
+
     let nativeConnection: SeaNativeConnection;
     try {
-      nativeConnection = (await this.binding.openSession(this.nativeOptions)) as SeaNativeConnection;
+      nativeConnection = (await this.binding.openSession(sessionOptions)) as SeaNativeConnection;
     } catch (err) {
       throw decodeNapiKernelError(err);
     }
 
-    // Merge `request.configuration` (the existing public field for Spark
-    // conf) with any backend-specific session config. The SEA wire
-    // protocol applies these per-statement, but we capture them at
-    // session-open time and forward with every executeStatement to
-    // preserve session-config semantics.
-    const sessionConfig = request.configuration ? { ...request.configuration } : undefined;
-
     return new SeaSessionBackend({
       connection: nativeConnection!,
       context: this.context,
-      defaults: {
-        initialCatalog: request.initialCatalog,
-        initialSchema: request.initialSchema,
-        sessionConfig,
-      },
     });
   }
 
diff --git a/lib/sea/SeaNativeLoader.ts b/lib/sea/SeaNativeLoader.ts
index b2b60e6f..a68caf90 100644
--- a/lib/sea/SeaNativeLoader.ts
+++ b/lib/sea/SeaNativeLoader.ts
@@ -37,18 +37,6 @@ import type { SeaNativeConnectionOptions } from './SeaAuth';
 // eslint-disable-next-line @typescript-eslint/no-var-requires, import/no-dynamic-require, global-require
 const native = require('../../native/sea/index.js');
 
-/**
- * JS-visible per-execute options carried over the napi binding boundary.
- * Mirrors the `ExecuteOptions` shape generated by napi-rs into
- * `native/sea/index.d.ts`. Re-declared here so the JS adapter layer
- * isn't tied to the binding-generated types.
- */
-export interface SeaExecuteOptions {
-  initialCatalog?: string;
-  initialSchema?: string;
-  sessionConfig?: Record<string, string>;
-}
-
 /**
  * Arrow IPC payload returned by `Statement.fetchNextBatch()`. Carries a
  * complete Arrow IPC stream (schema header + 1 record-batch message).
@@ -77,10 +65,93 @@ export interface SeaNativeStatement {
 }
 
 /**
- * Typed surface for the opaque napi `Connection` handle.
+ * Typed surface for the opaque napi `Connection` handle. Signatures
+ * match `native/sea/index.d.ts` exactly as generated by napi-rs from
+ * `msrathore/krn-napi-metadata-exposure` (tip fd5e866).
+ *
+ * napi-rs emits `string | undefined | null` for every Rust `Option<String>`
+ * parameter — both `undefined` and `null` are accepted at the call site.
  */
 export interface SeaNativeConnection {
-  executeStatement(sql: string, options: SeaExecuteOptions): Promise<SeaNativeStatement>;
+  /**
+   * Execute a SQL statement. Catalog / schema / sessionConf are
+   * session-level — set on `openSession`, applied to every statement
+   * executed on the resulting `Connection`. No per-statement options.
+   */
+  executeStatement(sql: string): Promise<SeaNativeStatement>;
+
+  // ── Metadata methods ──────────────────────────────────────────────────
+  /** All catalogs visible to the session. */
+  listCatalogs(): Promise<SeaNativeStatement>;
+
+  /** Schemas filtered by catalog (exact) and schema name LIKE pattern. */
+  listSchemas(
+    catalog?: string | undefined | null,
+    schemaPattern?: string | undefined | null,
+  ): Promise<SeaNativeStatement>;
+
+  /** Tables filtered by catalog (exact), schema and table LIKE patterns, and optional type list. */
+  listTables(
+    catalog?: string | undefined | null,
+    schemaPattern?: string | undefined | null,
+    tablePattern?: string | undefined | null,
+    tableTypes?: Array<string> | undefined | null,
+  ): Promise<SeaNativeStatement>;
+
+  /** Columns filtered by catalog (exact), schema/table/column LIKE patterns. */
+  listColumns(
+    catalog?: string | undefined | null,
+    schemaPattern?: string | undefined | null,
+    tablePattern?: string | undefined | null,
+    columnPattern?: string | undefined | null,
+  ): Promise<SeaNativeStatement>;
+
+  /** Functions filtered by catalog (exact), schema and name LIKE patterns. */
+  listFunctions(
+    catalog?: string | undefined | null,
+    schemaPattern?: string | undefined | null,
+    functionPattern?: string | undefined | null,
+  ): Promise<SeaNativeStatement>;
+
+  // NOTE: `listProcedures(catalog?, schemaPattern?, procedurePattern?)`
+  // is exposed on the napi binding for pyo3 parity, but intentionally
+  // not surfaced on IDBSQLSession in this M1 pass — the existing thrift
+  // NodeJS driver doesn't expose `getProcedures` either, and extending
+  // the public driver interface needs a separate spec entry + parity
+  // decision. Re-enable here when that decision is made.
+
+  /** All supported table types. No wire call — static result. */
+  listTableTypes(): Promise<SeaNativeStatement>;
+
+  /** All supported SQL data types. No wire call — static result. */
+  listTypeInfo(): Promise<SeaNativeStatement>;
+
+  /**
+   * Primary keys for the given table. All three arguments are required
+   * exact-match identifiers (napi rejects `undefined`/`null` with
+   * `InvalidArgument`). Callers must supply non-empty strings for all
+   * three — the JS adapter maps absent optional JS-layer fields to empty
+   * string or rejects early if the contract demands them.
+   */
+  getPrimaryKeys(
+    catalog: string,
+    schema: string,
+    table: string,
+  ): Promise<SeaNativeStatement>;
+
+  /**
+   * Foreign-key relationships. Parent side is optional (pass
+   * `undefined`/`null` to omit); foreign side is required.
+   */
+  getCrossReference(
+    parentCatalog: string | undefined | null,
+    parentSchema: string | undefined | null,
+    parentTable: string | undefined | null,
+    foreignCatalog: string,
+    foreignSchema: string,
+    foreignTable: string,
+  ): Promise<SeaNativeStatement>;
+
   close(): Promise<void>;
 }
 
diff --git a/lib/sea/SeaSessionBackend.ts b/lib/sea/SeaSessionBackend.ts
index de63191f..512958aa 100644
--- a/lib/sea/SeaSessionBackend.ts
+++ b/lib/sea/SeaSessionBackend.ts
@@ -31,33 +31,15 @@ import {
 import Status from '../dto/Status';
 import InfoValue from '../dto/InfoValue';
 import HiveDriverError from '../errors/HiveDriverError';
-import { SeaNativeConnection, SeaExecuteOptions } from './SeaNativeLoader';
+import { SeaNativeConnection } from './SeaNativeLoader';
 import { decodeNapiKernelError } from './SeaErrorMapping';
 import SeaOperationBackend from './SeaOperationBackend';
-
-/**
- * Per-session defaults that apply to every `executeStatement` issued
- * through this backend. Captured at `SeaBackend.openSession()` time from
- * the `OpenSessionRequest` — `initialCatalog` / `initialSchema` /
- * `sessionConfig`.
- *
- * The napi binding routes these to the kernel's `statement_conf` map,
- * which the SEA wire treats as session-scoped parameters. They are
- * forwarded with every `executeStatement` call so the JDBC-style
- * "session config" semantics are preserved even though SEA's wire
- * protocol is statement-scoped.
- */
-export interface SeaSessionDefaults {
-  initialCatalog?: string;
-  initialSchema?: string;
-  sessionConfig?: Record<string, string>;
-}
+import SeaTableTypeFilter from './SeaTableTypeFilter';
 
 export interface SeaSessionBackendOptions {
   /** The opaque napi `Connection` handle returned by `openSession`. */
   connection: SeaNativeConnection;
   context: IClientContext;
-  defaults?: SeaSessionDefaults;
   /** Optional override for `id`. Defaults to a fresh UUIDv4. */
   id?: string;
 }
@@ -65,37 +47,42 @@ export interface SeaSessionBackendOptions {
 /**
  * SEA-backed implementation of `ISessionBackend`.
  *
- * **M0 scope:** `executeStatement` + `close`. Metadata methods
- * (`getCatalogs`, `getSchemas`, etc.) defer to M1 — they throw a clear
- * `HiveDriverError` so consumers using SEA against metadata APIs get an
- * actionable message instead of silently falling back. The Thrift
- * backend continues to handle the metadata path by default (callers
- * opt into SEA via `ConnectionOptions.useSEA`).
+ * **M1 scope:** `executeStatement`, 9 of the 10 `IDBSQLSession`
+ * metadata methods, and `close`. The implemented nine are:
+ * `getTypeInfo`, `getCatalogs`, `getSchemas`, `getTables`,
+ * `getTableTypes`, `getColumns`, `getFunctions`, `getPrimaryKeys`,
+ * `getCrossReference`. `getInfo` is a stub-throw (deferred — the
+ * kernel `Metadata` API does not expose it yet); `getProcedures` is
+ * not on `IDBSQLSession` (see `SeaNativeLoader.ts` NOTE comment).
+ *
+ * All implemented metadata methods delegate directly to the corresponding
+ * napi `Connection` method — the kernel performs SHOW/information_schema
+ * queries and returns JDBC-shaped Arrow batches through a `Statement`
+ * handle identical to `executeStatement`. The JS layer wraps each
+ * returned `Statement` in a `SeaOperationBackend` so callers consume
+ * metadata results through the same `IOperationBackend` interface they
+ * use for SQL results.
  *
- * **Session config flow:** the SEA wire protocol is statement-scoped,
- * so "session config" semantics (Spark conf, `initialCatalog`,
- * `initialSchema`) are emulated by forwarding the same defaults with
- * every `executeStatement` call. Per-statement overrides on
- * `ExecuteStatementOptions` are reserved for M1; M0 carries only the
- * defaults captured at session-open time plus the `useCloudFetch`
- * boolean projected onto `sessionConfig.use_cloud_fetch` for the
- * kernel.
+ * **Session config flow:** catalog / schema / sessionConf are applied
+ * once at session creation (kernel `Session::builder().defaults()` +
+ * `.session_conf()` → SEA `CreateSession.catalog` / `.schema` /
+ * `.session_confs`) and remain in effect for every statement run on
+ * the resulting napi `Connection`. No per-statement forwarding is
+ * needed — that pattern was removed when the napi binding moved these
+ * onto `openSession` to match pyo3.
  */
 export default class SeaSessionBackend implements ISessionBackend {
   private readonly connection: SeaNativeConnection;
 
   private readonly context: IClientContext;
 
-  private readonly defaults: SeaSessionDefaults;
-
   private readonly _id: string;
 
   private closed = false;
 
-  constructor({ connection, context, defaults, id }: SeaSessionBackendOptions) {
+  constructor({ connection, context, id }: SeaSessionBackendOptions) {
     this.connection = connection;
     this.context = context;
-    this.defaults = defaults ?? {};
     this._id = id ?? uuidv4();
   }
 
@@ -108,13 +95,21 @@ export default class SeaSessionBackend implements ISessionBackend {
   }
 
   /**
-   * Execute a SQL statement through the napi binding. Merges the
-   * session-level defaults (`initialCatalog` / `initialSchema` /
-   * `sessionConfig`) with the per-call `useCloudFetch` override.
+   * Execute a SQL statement through the napi binding.
+   *
+   * Catalog / schema / sessionConf were applied at session open, so
+   * there are no per-statement options to thread through.
    *
    * M0 intentionally rejects `queryTimeout`, `namedParameters`, and
-   * `ordinalParameters` with explicit deferred-to-M1 errors. The Thrift
-   * backend remains the path for consumers that need any of those today.
+   * `ordinalParameters` with explicit deferred-to-M1 errors. `useCloudFetch`
+   * is a no-op on the SEA path — the kernel hardcodes the SEA
+   * `disposition` to `INLINE_OR_EXTERNAL_LINKS`, and per-statement
+   * conf overrides have no reader on the kernel; cloud-fetch behaviour
+   * is governed entirely by the kernel's `ResultConfig` (M1 binding
+   * surface).
+   *
+   * The Thrift backend remains the path for consumers that need any
+   * of those today.
    */
   public async executeStatement(statement: string, options: ExecuteStatementOptions): Promise<IOperationBackend> {
     this.failIfClosed();
@@ -131,67 +126,169 @@ export default class SeaSessionBackend implements ISessionBackend {
       );
     }
 
-    // Merge session-level sessionConfig with per-statement useCloudFetch.
-    // The kernel accepts only string-valued conf values; booleans are
-    // String()'d to "true"/"false" matching the existing Thrift conf
-    // convention.
-    const sessionConfig: Record<string, string> = { ...(this.defaults.sessionConfig ?? {}) };
-    if (options.useCloudFetch !== undefined) {
-      sessionConfig.use_cloud_fetch = String(options.useCloudFetch);
-    }
-
-    const executeOptions: SeaExecuteOptions = {
-      initialCatalog: this.defaults.initialCatalog,
-      initialSchema: this.defaults.initialSchema,
-      sessionConfig: Object.keys(sessionConfig).length > 0 ? sessionConfig : undefined,
-    };
-
     let nativeStatement;
     try {
-      nativeStatement = await this.connection.executeStatement(statement, executeOptions);
+      nativeStatement = await this.connection.executeStatement(statement);
     } catch (err) {
       throw decodeNapiKernelError(err);
     }
     return new SeaOperationBackend({
-      statement: nativeStatement!,
+      statement: nativeStatement,
       context: this.context,
     });
   }
 
   public async getTypeInfo(_request: TypeInfoRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError('SeaSessionBackend.getTypeInfo: not implemented yet (deferred to M1)');
+    this.failIfClosed();
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.listTypeInfo();
+    } catch (err) {
+      throw decodeNapiKernelError(err);
+    }
+    return new SeaOperationBackend({ statement: nativeStatement, context: this.context });
   }
 
   public async getCatalogs(_request: CatalogsRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError('SeaSessionBackend.getCatalogs: not implemented yet (deferred to M1)');
+    this.failIfClosed();
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.listCatalogs();
+    } catch (err) {
+      throw decodeNapiKernelError(err);
+    }
+    return new SeaOperationBackend({ statement: nativeStatement, context: this.context });
   }
 
-  public async getSchemas(_request: SchemasRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError('SeaSessionBackend.getSchemas: not implemented yet (deferred to M1)');
+  public async getSchemas(request: SchemasRequest): Promise<IOperationBackend> {
+    this.failIfClosed();
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.listSchemas(
+        request.catalogName,
+        request.schemaName,
+      );
+    } catch (err) {
+      throw decodeNapiKernelError(err);
+    }
+    return new SeaOperationBackend({ statement: nativeStatement, context: this.context });
   }
 
-  public async getTables(_request: TablesRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError('SeaSessionBackend.getTables: not implemented yet (deferred to M1)');
+  public async getTables(request: TablesRequest): Promise<IOperationBackend> {
+    this.failIfClosed();
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.listTables(
+        request.catalogName,
+        request.schemaName,
+        request.tableName,
+        request.tableTypes,
+      );
+    } catch (err) {
+      throw decodeNapiKernelError(err);
+    }
+    const backend = new SeaOperationBackend({ statement: nativeStatement, context: this.context });
+    // The server does not honour tableTypes server-side (advisory only).
+    // Apply client-side filter when the caller supplied a non-null list.
+    if (request.tableTypes != null) {
+      return new SeaTableTypeFilter(backend, new Set(request.tableTypes));
+    }
+    return backend;
   }
 
   public async getTableTypes(_request: TableTypesRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError('SeaSessionBackend.getTableTypes: not implemented yet (deferred to M1)');
+    this.failIfClosed();
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.listTableTypes();
+    } catch (err) {
+      throw decodeNapiKernelError(err);
+    }
+    return new SeaOperationBackend({ statement: nativeStatement, context: this.context });
   }
 
-  public async getColumns(_request: ColumnsRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError('SeaSessionBackend.getColumns: not implemented yet (deferred to M1)');
+  public async getColumns(request: ColumnsRequest): Promise<IOperationBackend> {
+    this.failIfClosed();
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.listColumns(
+        request.catalogName,
+        request.schemaName,
+        request.tableName,
+        request.columnName,
+      );
+    } catch (err) {
+      throw decodeNapiKernelError(err);
+    }
+    return new SeaOperationBackend({ statement: nativeStatement, context: this.context });
   }
 
-  public async getFunctions(_request: FunctionsRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError('SeaSessionBackend.getFunctions: not implemented yet (deferred to M1)');
+  public async getFunctions(request: FunctionsRequest): Promise<IOperationBackend> {
+    this.failIfClosed();
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.listFunctions(
+        request.catalogName,
+        request.schemaName,
+        request.functionName,
+      );
+    } catch (err) {
+      throw decodeNapiKernelError(err);
+    }
+    return new SeaOperationBackend({ statement: nativeStatement, context: this.context });
   }
 
-  public async getPrimaryKeys(_request: PrimaryKeysRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError('SeaSessionBackend.getPrimaryKeys: not implemented yet (deferred to M1)');
+  public async getPrimaryKeys(request: PrimaryKeysRequest): Promise<IOperationBackend> {
+    this.failIfClosed();
+    if (!request.catalogName) {
+      throw new HiveDriverError(
+        'SeaSessionBackend.getPrimaryKeys: catalogName is required on the SEA path (kernel rejects empty identifiers; Thrift backend silently resolves to session default)',
+      );
+    }
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.getPrimaryKeys(
+        request.catalogName,
+        request.schemaName,
+        request.tableName,
+      );
+    } catch (err) {
+      throw decodeNapiKernelError(err);
+    }
+    return new SeaOperationBackend({ statement: nativeStatement, context: this.context });
   }
 
-  public async getCrossReference(_request: CrossReferenceRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError('SeaSessionBackend.getCrossReference: not implemented yet (deferred to M1)');
+  public async getCrossReference(request: CrossReferenceRequest): Promise<IOperationBackend> {
+    this.failIfClosed();
+    if (!request.foreignCatalogName) {
+      throw new HiveDriverError(
+        'SeaSessionBackend.getCrossReference: foreignCatalogName is required on the SEA path (kernel rejects empty identifiers)',
+      );
+    }
+    if (!request.foreignSchemaName) {
+      throw new HiveDriverError(
+        'SeaSessionBackend.getCrossReference: foreignSchemaName is required on the SEA path',
+      );
+    }
+    if (!request.foreignTableName) {
+      throw new HiveDriverError(
+        'SeaSessionBackend.getCrossReference: foreignTableName is required on the SEA path',
+      );
+    }
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.getCrossReference(
+        request.parentCatalogName,
+        request.parentSchemaName,
+        request.parentTableName,
+        request.foreignCatalogName,
+        request.foreignSchemaName,
+        request.foreignTableName,
+      );
+    } catch (err) {
+      throw decodeNapiKernelError(err);
+    }
+    return new SeaOperationBackend({ statement: nativeStatement, context: this.context });
   }
 
   public async close(): Promise<Status> {
diff --git a/lib/sea/SeaTableTypeFilter.ts b/lib/sea/SeaTableTypeFilter.ts
new file mode 100644
index 00000000..35243a05
--- /dev/null
+++ b/lib/sea/SeaTableTypeFilter.ts
@@ -0,0 +1,93 @@
+// Copyright (c) 2026 Databricks, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import {
+  TGetOperationStatusResp,
+  TGetResultSetMetadataResp,
+} from '../../thrift/TCLIService_types';
+import IOperationBackend from '../contracts/IOperationBackend';
+import Status from '../dto/Status';
+import { WaitUntilReadyOptions } from '../contracts/IOperation';
+
+/**
+ * Wraps an `IOperationBackend` and filters rows by the `TABLE_TYPE` column.
+ *
+ * The Databricks server does not honour the `table_types` filter in the
+ * `listTables` kernel call — the field is advisory only (see pyo3 metadata.rs
+ * and matrix-audit-python.md).  This adapter applies the filter client-side on
+ * every `fetchChunk` call.
+ *
+ * Semantics:
+ * - `allowedTypes` is a `Set` built from `request.tableTypes` before this
+ *   wrapper is constructed.  The caller decides what to pass in.
+ * - An empty `Set` means "keep nothing" — the caller passed `tableTypes: []`,
+ *   which explicitly requests zero table types.
+ * - Matching is case-sensitive exact equality on the `TABLE_TYPE` column value.
+ *   Databricks returns upper-case values (`TABLE`, `VIEW`, `EXTERNAL TABLE`, …);
+ *   callers should pass upper-case strings.
+ * - Rows whose `TABLE_TYPE` column is absent or non-string are dropped.
+ *
+ * All lifecycle methods (`cancel`, `close`, `status`, `waitUntilReady`,
+ * `getResultMetadata`) delegate unchanged to the inner backend.
+ */
+export default class SeaTableTypeFilter implements IOperationBackend {
+  private readonly inner: IOperationBackend;
+
+  private readonly allowedTypes: Set<string>;
+
+  constructor(inner: IOperationBackend, allowedTypes: Set<string>) {
+    this.inner = inner;
+    this.allowedTypes = allowedTypes;
+  }
+
+  public get id(): string {
+    return this.inner.id;
+  }
+
+  public get hasResultSet(): boolean {
+    return this.inner.hasResultSet;
+  }
+
+  public async fetchChunk(options: { limit: number; disableBuffering?: boolean }): Promise<Array<object>> {
+    const rows = await this.inner.fetchChunk(options);
+    return rows.filter((row) => {
+      const tableType = (row as Record<string, unknown>).TABLE_TYPE;
+      return typeof tableType === 'string' && this.allowedTypes.has(tableType);
+    });
+  }
+
+  public async hasMore(): Promise<boolean> {
+    return this.inner.hasMore();
+  }
+
+  public async waitUntilReady(options?: WaitUntilReadyOptions): Promise<void> {
+    return this.inner.waitUntilReady(options);
+  }
+
+  public async status(progress: boolean): Promise<TGetOperationStatusResp> {
+    return this.inner.status(progress);
+  }
+
+  public async getResultMetadata(): Promise<TGetResultSetMetadataResp> {
+    return this.inner.getResultMetadata();
+  }
+
+  public async cancel(): Promise<Status> {
+    return this.inner.cancel();
+  }
+
+  public async close(): Promise<Status> {
+    return this.inner.close();
+  }
+}
diff --git a/native/sea/index.d.ts b/native/sea/index.d.ts
index 97257895..85b82edf 100644
--- a/native/sea/index.d.ts
+++ b/native/sea/index.d.ts
@@ -4,61 +4,16 @@
 /* auto-generated by NAPI-RS */
 
 /**
- * JS-visible per-execute options. M0 only carries
- * initialCatalog / initialSchema / sessionConfig — parameters and
- * per-statement overrides land in M1.
- */
-export interface ExecuteOptions {
-  /** Default catalog applied to this statement via session conf. */
-  initialCatalog?: string
-  /** Default schema applied to this statement via session conf. */
-  initialSchema?: string
-  /**
-   * Per-statement session conf overrides (forwarded to SEA
-   * `parameters` / Thrift `confOverlay`).
-   */
-  sessionConfig?: Record<string, string>
-}
-/**
- * JS-visible auth-mode discriminant.
- *
- * Crosses the FFI as the variant name verbatim — napi-rs's
- * `#[napi(string_enum)]` without an explicit case option emits the
- * Rust variant identifier as-is, so this enum becomes
- * `AuthMode.Pat = 'Pat'` / `AuthMode.OAuthM2m = 'OAuthM2m'` /
- * `AuthMode.OAuthU2m = 'OAuthU2m'` in the auto-generated
- * `native/sea/index.d.ts`. The JS adapter
- * (`SeaNativeConnectionOptions` in `lib/sea/SeaAuth.ts`) must use the
- * PascalCase literals verbatim.
+ * JS-visible options for opening a Databricks SQL session over PAT.
  *
- * Keeping the discriminant explicit instead of inferring it from
- * "which Option is set" makes the napi-side validation single-pass
- * and the JS-side schema typed.
+ * M0 supports PAT only — `token` is required. OAuth M2M / U2M variants
+ * land in M1 along with a discriminated-union shape on the JS side.
  *
- * Note: adding a variant here requires extending `open_session()`'s
- * `match` — Rust will fail the build if the match is non-exhaustive,
- * but the cross-reference shortens the review loop.
- */
-export const enum AuthMode {
-  Pat = 'Pat',
-  OAuthM2m = 'OAuthM2m',
-  OAuthU2m = 'OAuthU2m'
-}
-/**
- * JS-visible options for opening a Databricks SQL session.
- *
- * Discriminated by `auth_mode`:
- * - `AuthMode::Pat`      → requires `token`; ignores oauth_*.
- * - `AuthMode::OAuthM2m` → requires `oauth_client_id` + `oauth_client_secret`.
- * - `AuthMode::OAuthU2m` → kernel handles the auth-code flow with
- *   default client_id (`databricks-cli`), scopes
- *   (`["all-apis","offline_access"]`), and OIDC discovery; the JS
- *   adapter hardcodes `oauth_redirect_port` to 8030 to override the
- *   kernel default of 8020 (thrift uses 8030 — preserves parity).
- *
- * Scopes / token_url_override / client_id / callback_timeout are not
- * exposed — kernel defaults match thrift parity and the public driver
- * surface has no demand to override them.
+ * Catalog / schema / sessionConf are applied once at session creation
+ * and remain in effect for every statement run on the resulting
+ * `Connection`. The SEA wire protocol carries them on
+ * `CreateSession`, not on `ExecuteStatement` — so there is no
+ * per-statement override path in either this binding or pyo3.
  */
 export interface ConnectionOptions {
   /**
@@ -71,24 +26,33 @@ export interface ConnectionOptions {
    * kernel parses out the warehouse id.
    */
   httpPath: string
-  /** Auth-mode discriminant. Required. */
-  authMode: AuthMode
-  /** Personal access token. Required iff `auth_mode == Pat`. */
-  token?: string
-  /** OAuth client id. Required iff `auth_mode == OAuthM2m`. */
-  oauthClientId?: string
-  /** OAuth client secret. Required iff `auth_mode == OAuthM2m`. */
-  oauthClientSecret?: string
-  /**
-   * Local listener port for the U2M authorization-code callback.
-   * Forwarded verbatim to the kernel; the JS adapter hardcodes 8030
-   * for thrift parity.
-   */
-  oauthRedirectPort?: number
+  /**
+   * Personal access token. Must be non-empty (the kernel rejects
+   * empty PATs at session construction).
+   */
+  token: string
+  /**
+   * Default catalog for statements executed on this session.
+   * Routed through the kernel's `DefaultOpts` and onto the SEA
+   * `CreateSession.catalog` wire field.
+   */
+  catalog?: string
+  /**
+   * Default schema for statements executed on this session.
+   * Routed through the kernel's `DefaultOpts` and onto the SEA
+   * `CreateSession.schema` wire field.
+   */
+  schema?: string
+  /**
+   * Server-bound session conf (Spark conf, `ANSI_MODE`, `TIMEZONE`,
+   * query-tag presets, …). Forwarded verbatim to SEA
+   * `session_confs`. Unknown keys are rejected server-side.
+   */
+  sessionConf?: Record<string, string>
 }
 /**
- * Open a Databricks SQL session and return an opaque `Connection`
- * wrapping the kernel `Session`. Supports PAT, OAuth M2M, and OAuth U2M.
+ * Open a Databricks SQL session over PAT auth and return an opaque
+ * `Connection` wrapping the kernel `Session`.
  *
  * The JS-visible name is `openSession` (napi-rs converts snake_case
  * to camelCase for free functions).
@@ -130,8 +94,12 @@ export declare class Connection {
   /**
    * Execute a SQL statement and return a Statement handle that
    * streams batches via `fetchNextBatch()`.
+   *
+   * No per-statement options: catalog / schema / sessionConf are
+   * session-level (`openSession`). Positional / named parameters
+   * land in M1 via `Statement::spec().param(…)` on the kernel.
    */
-  executeStatement(sql: string, options: ExecuteOptions): Promise<Statement>
+  executeStatement(sql: string): Promise<Statement>
   /**
    * Explicit close. Marks the connection wrapper as closed so
    * subsequent calls on this `Connection` return `InvalidArg`, then
@@ -151,6 +119,67 @@ export declare class Connection {
    * Round 3 findings.
    */
   close(): Promise<void>
+  /**
+   * All catalogs visible to the session.
+   *
+   * JDBC `getCatalogs` shape: `TABLE_CAT: Utf8`.
+   */
+  listCatalogs(): Promise<Statement>
+  /**
+   * Schemas filtered by catalog (exact) and schema name pattern.
+   *
+   * JDBC `getSchemas` shape: `TABLE_SCHEM, TABLE_CATALOG`.
+   */
+  listSchemas(catalog?: string | undefined | null, schemaPattern?: string | undefined | null): Promise<Statement>
+  /**
+   * Tables filtered by catalog (exact), schema (pattern), table (pattern).
+   *
+   * JDBC `getTables` shape: 10 columns. `tableTypes`, when provided,
+   * filters rows by `TABLE_TYPE` kernel-side.
+   *
+   * `tableTypes` is an advisory filter. Databricks `SHOW TABLES` does
+   * NOT honour the table-type filter server-side; the kernel applies
+   * it client-side after the result returns. Callers expecting
+   * server-side rejection of off-type tables should not rely on this.
+   */
+  listTables(catalog?: string | undefined | null, schemaPattern?: string | undefined | null, tablePattern?: string | undefined | null, tableTypes?: Array<string> | undefined | null): Promise<Statement>
+  /**
+   * Columns of tables matching the filter.
+   *
+   * JDBC `getColumns` shape: 23 columns.
+   */
+  listColumns(catalog?: string | undefined | null, schemaPattern?: string | undefined | null, tablePattern?: string | undefined | null, columnPattern?: string | undefined | null): Promise<Statement>
+  /**
+   * Functions visible to the session. `catalog` is exact;
+   * `schemaPattern` and `functionPattern` are SQL LIKE.
+   */
+  listFunctions(catalog?: string | undefined | null, schemaPattern?: string | undefined | null, functionPattern?: string | undefined | null): Promise<Statement>
+  /**
+   * Procedures visible to the session. `catalog` is exact;
+   * `schemaPattern` and `procedurePattern` are SQL LIKE.
+   */
+  listProcedures(catalog?: string | undefined | null, schemaPattern?: string | undefined | null, procedurePattern?: string | undefined | null): Promise<Statement>
+  /**
+   * All table types (`TABLE`, `VIEW`, `SYSTEM TABLE`, …).
+   * No wire call — static in-memory result.
+   */
+  listTableTypes(): Promise<Statement>
+  /**
+   * SQL data types supported by the workspace.
+   * No wire call — static in-memory result.
+   */
+  listTypeInfo(): Promise<Statement>
+  /**
+   * Primary keys for the given table. All three identifiers are
+   * exact — ODBC `SQLPrimaryKeys` does not support patterns.
+   */
+  getPrimaryKeys(catalog: string, schema: string, table: string): Promise<Statement>
+  /**
+   * Foreign-key relationships. The foreign side must be fully
+   * specified (catalog + schema + table); the parent side is
+   * optional. All identifiers are exact — no LIKE patterns.
+   */
+  getCrossReference(parentCatalog: string | undefined | null, parentSchema: string | undefined | null, parentTable: string | undefined | null, foreignCatalog: string, foreignSchema: string, foreignTable: string): Promise<Statement>
 }
 /**
  * Opaque executed-statement handle.
@@ -178,11 +207,16 @@ export declare class Statement {
    * fetched.
    */
   schema(): Promise<ArrowSchema>
-  /** Server-side cancel. No-op if already finished. */
+  /**
+   * Server-side cancel. No-op if already finished or if this
+   * `Statement` wraps a metadata `ResultStream` (metadata calls have
+   * no in-flight cancellation surface in the kernel today).
+   */
   cancel(): Promise<void>
   /**
    * Explicit close. Awaits the server-side close so the JS caller
-   * can observe failures.
+   * can observe failures. For metadata streams, drops the stream
+   * (no server round-trip needed).
    */
   close(): Promise<void>
 }
diff --git a/tests/unit/sea/execution.test.ts b/tests/unit/sea/execution.test.ts
index d093a756..638f76c4 100644
--- a/tests/unit/sea/execution.test.ts
+++ b/tests/unit/sea/execution.test.ts
@@ -21,7 +21,6 @@ import {
   SeaNativeBinding,
   SeaNativeConnection,
   SeaNativeStatement,
-  SeaExecuteOptions,
 } from '../../../lib/sea/SeaNativeLoader';
 import IClientContext, { ClientConfig } from '../../../lib/contracts/IClientContext';
 import IDBSQLLogger, { LogLevel } from '../../../lib/contracts/IDBSQLLogger';
@@ -61,21 +60,64 @@ class FakeNativeConnection implements SeaNativeConnection {
 
   public lastSql?: string;
 
-  public lastOptions?: SeaExecuteOptions;
-
   public throwOnExecute: Error | null = null;
 
   public statementToReturn: FakeNativeStatement = new FakeNativeStatement();
 
-  public async executeStatement(sql: string, options: SeaExecuteOptions): Promise<SeaNativeStatement> {
+  public async executeStatement(sql: string): Promise<SeaNativeStatement> {
     if (this.throwOnExecute) {
       throw this.throwOnExecute;
     }
     this.lastSql = sql;
-    this.lastOptions = options;
     return this.statementToReturn;
   }
 
+  // Metadata stubs — return a fresh statement so callers can test wrapping.
+  public async listCatalogs() { return new FakeNativeStatement(); }
+
+  public async listSchemas(_catalog: string | undefined, _schemaPattern: string | undefined) {
+    return new FakeNativeStatement();
+  }
+
+  public async listTables(
+    _catalog: string | undefined,
+    _schemaPattern: string | undefined,
+    _tablePattern: string | undefined,
+    _tableTypes: string[] | undefined,
+  ) { return new FakeNativeStatement(); }
+
+  public async listColumns(
+    _catalog: string | undefined,
+    _schemaPattern: string | undefined,
+    _tablePattern: string | undefined,
+    _columnPattern: string | undefined,
+  ) { return new FakeNativeStatement(); }
+
+  public async listFunctions(
+    _catalog: string | undefined,
+    _schemaPattern: string | undefined,
+    _functionPattern: string | undefined,
+  ) { return new FakeNativeStatement(); }
+
+  public async listTableTypes() { return new FakeNativeStatement(); }
+
+  public async listTypeInfo() { return new FakeNativeStatement(); }
+
+  public async getPrimaryKeys(
+    _catalog: string,
+    _schema: string,
+    _table: string,
+  ) { return new FakeNativeStatement(); }
+
+  public async getCrossReference(
+    _parentCatalog: string | undefined | null,
+    _parentSchema: string | undefined | null,
+    _parentTable: string | undefined | null,
+    _foreignCatalog: string,
+    _foreignSchema: string,
+    _foreignTable: string,
+  ) { return new FakeNativeStatement(); }
+
   public async close(): Promise<void> {
     this.closed = true;
   }
@@ -240,7 +282,7 @@ describe('SeaBackend', () => {
     expect(sessionBackend.id).to.be.a('string').and.have.length.greaterThan(0);
   });
 
-  it('openSession() propagates initialCatalog / initialSchema / sessionConfig through to executeStatement', async () => {
+  it('openSession() forwards initialCatalog / initialSchema / configuration to the napi openSession call (not per-statement)', async () => {
     const connection = new FakeNativeConnection();
     const binding = makeBinding(connection);
     const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
@@ -257,14 +299,22 @@ describe('SeaBackend', () => {
       configuration: { 'spark.sql.execution.arrow.enabled': 'true' },
     });
 
-    await session.executeStatement('SELECT 1', {});
+    // The defaults reach the kernel via `Session::builder().defaults()` +
+    // `.session_conf()`, applied on `CreateSession`. Assert they were
+    // folded into the napi `openSession` arg.
+    expect(binding.openSessionStub.calledOnce).to.equal(true);
+    expect(binding.openSessionStub.firstCall.args[0]).to.deep.include({
+      authMode: 'Pat',
+      token: 't',
+      catalog: 'main',
+      schema: 'default',
+      sessionConf: { 'spark.sql.execution.arrow.enabled': 'true' },
+    });
 
+    // And the SQL still threads through executeStatement (now with no
+    // per-statement options).
+    await session.executeStatement('SELECT 1', {});
     expect(connection.lastSql).to.equal('SELECT 1');
-    expect(connection.lastOptions).to.deep.equal({
-      initialCatalog: 'main',
-      initialSchema: 'default',
-      sessionConfig: { 'spark.sql.execution.arrow.enabled': 'true' },
-    });
   });
 
   it('close() clears connection state without throwing', async () => {
@@ -285,8 +335,8 @@ describe('SeaBackend', () => {
 });
 
 describe('SeaSessionBackend', () => {
-  function makeSession(connection: SeaNativeConnection, defaults = {}) {
-    return new SeaSessionBackend({ connection, context: makeContext(), defaults });
+  function makeSession(connection: SeaNativeConnection) {
+    return new SeaSessionBackend({ connection, context: makeContext() });
   }
 
   it('executeStatement passes sql through verbatim', async () => {
@@ -304,21 +354,6 @@ describe('SeaSessionBackend', () => {
     expect(op.id).to.be.a('string').and.have.length.greaterThan(0);
   });
 
-  it('executeStatement merges session defaults into ExecuteOptions', async () => {
-    const connection = new FakeNativeConnection();
-    const session = makeSession(connection, {
-      initialCatalog: 'main',
-      initialSchema: 'default',
-      sessionConfig: { foo: 'bar' },
-    });
-    await session.executeStatement('SELECT 1', {});
-    expect(connection.lastOptions).to.deep.equal({
-      initialCatalog: 'main',
-      initialSchema: 'default',
-      sessionConfig: { foo: 'bar' },
-    });
-  });
-
   it('executeStatement rejects namedParameters (M1)', async () => {
     const connection = new FakeNativeConnection();
     const session = makeSession(connection);
@@ -357,31 +392,13 @@ describe('SeaSessionBackend', () => {
     expect((thrown as Error).message).to.match(/queryTimeout/);
   });
 
-  it('metadata methods throw deferred-M1 errors', async () => {
+  // Metadata-method happy-path and arg-routing coverage lives in
+  // tests/unit/sea/metadata.test.ts (sea-execution-metadata milestone).
+  it('metadata methods return SeaOperationBackend (wired to napi)', async () => {
     const connection = new FakeNativeConnection();
     const session = makeSession(connection);
-    for (const method of [
-      'getInfo',
-      'getTypeInfo',
-      'getCatalogs',
-      'getSchemas',
-      'getTables',
-      'getTableTypes',
-      'getColumns',
-      'getFunctions',
-      'getPrimaryKeys',
-      'getCrossReference',
-    ] as const) {
-      let thrown: unknown;
-      try {
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        await (session as any)[method]({});
-      } catch (err) {
-        thrown = err;
-      }
-      expect(thrown, `expected ${method} to throw`).to.be.instanceOf(HiveDriverError);
-      expect((thrown as Error).message).to.match(/M1|not implemented/);
-    }
+    const op = await session.getCatalogs({});
+    expect(op).to.be.instanceOf(SeaOperationBackend);
   });
 
   it('close() forwards to the native connection', async () => {
diff --git a/tests/unit/sea/metadata.test.ts b/tests/unit/sea/metadata.test.ts
new file mode 100644
index 00000000..8afdfccd
--- /dev/null
+++ b/tests/unit/sea/metadata.test.ts
@@ -0,0 +1,605 @@
+// Copyright (c) 2026 Databricks, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import { expect } from 'chai';
+import SeaSessionBackend from '../../../lib/sea/SeaSessionBackend';
+import SeaOperationBackend from '../../../lib/sea/SeaOperationBackend';
+import SeaTableTypeFilter from '../../../lib/sea/SeaTableTypeFilter';
+import {
+  SeaNativeConnection,
+  SeaNativeStatement,
+} from '../../../lib/sea/SeaNativeLoader';
+import IOperationBackend from '../../../lib/contracts/IOperationBackend';
+import IClientContext, { ClientConfig } from '../../../lib/contracts/IClientContext';
+import IDBSQLLogger, { LogLevel } from '../../../lib/contracts/IDBSQLLogger';
+import HiveDriverError from '../../../lib/errors/HiveDriverError';
+
+// ─── Fakes ───────────────────────────────────────────────────────────────────
+
+class FakeNativeStatement implements SeaNativeStatement {
+  public async fetchNextBatch() { return null; }
+  public async schema() { return { ipcBytes: Buffer.alloc(0) }; }
+  public async cancel() {}
+  public async close() {}
+}
+
+interface RecordedMetadataCall {
+  method: string;
+  args: unknown[];
+  returnStatement: FakeNativeStatement;
+}
+
+/**
+ * Connection fake that records every metadata call and the args passed
+ * so tests can assert on both the argument routing and the return-value
+ * wrapping path.
+ */
+class FakeMetadataConnection implements SeaNativeConnection {
+  public readonly calls: RecordedMetadataCall[] = [];
+
+  public throwNextCall: unknown = null;
+
+  private record(method: string, args: unknown[]): FakeNativeStatement {
+    if (this.throwNextCall !== null) {
+      const err = this.throwNextCall;
+      this.throwNextCall = null;
+      throw err;
+    }
+    const returnStatement = new FakeNativeStatement();
+    this.calls.push({ method, args, returnStatement });
+    return returnStatement;
+  }
+
+  public async executeStatement(_sql: string): Promise<SeaNativeStatement> {
+    return this.record('executeStatement', [_sql]);
+  }
+
+  public async listCatalogs(): Promise<SeaNativeStatement> {
+    return this.record('listCatalogs', []);
+  }
+
+  public async listSchemas(
+    catalog: string | undefined,
+    schemaPattern: string | undefined,
+  ): Promise<SeaNativeStatement> {
+    return this.record('listSchemas', [catalog, schemaPattern]);
+  }
+
+  public async listTables(
+    catalog: string | undefined,
+    schemaPattern: string | undefined,
+    tablePattern: string | undefined,
+    tableTypes: string[] | undefined,
+  ): Promise<SeaNativeStatement> {
+    return this.record('listTables', [catalog, schemaPattern, tablePattern, tableTypes]);
+  }
+
+  public async listColumns(
+    catalog: string | undefined,
+    schemaPattern: string | undefined,
+    tablePattern: string | undefined,
+    columnPattern: string | undefined,
+  ): Promise<SeaNativeStatement> {
+    return this.record('listColumns', [catalog, schemaPattern, tablePattern, columnPattern]);
+  }
+
+  public async listFunctions(
+    catalog: string | undefined,
+    schemaPattern: string | undefined,
+    functionPattern: string | undefined,
+  ): Promise<SeaNativeStatement> {
+    return this.record('listFunctions', [catalog, schemaPattern, functionPattern]);
+  }
+
+  public async listTableTypes(): Promise<SeaNativeStatement> {
+    return this.record('listTableTypes', []);
+  }
+
+  public async listTypeInfo(): Promise<SeaNativeStatement> {
+    return this.record('listTypeInfo', []);
+  }
+
+  public async getPrimaryKeys(
+    catalog: string,
+    schema: string,
+    table: string,
+  ): Promise<SeaNativeStatement> {
+    return this.record('getPrimaryKeys', [catalog, schema, table]);
+  }
+
+  public async getCrossReference(
+    parentCatalog: string | undefined | null,
+    parentSchema: string | undefined | null,
+    parentTable: string | undefined | null,
+    foreignCatalog: string,
+    foreignSchema: string,
+    foreignTable: string,
+  ): Promise<SeaNativeStatement> {
+    return this.record('getCrossReference', [
+      parentCatalog, parentSchema, parentTable,
+      foreignCatalog, foreignSchema, foreignTable,
+    ]);
+  }
+
+  public async close(): Promise<void> {}
+}
+
+function makeContext(): IClientContext {
+  const logger: IDBSQLLogger = { log(_level: LogLevel, _message: string): void {} };
+  const config = {} as ClientConfig;
+  return {
+    getConfig: () => config,
+    getLogger: () => logger,
+    getConnectionProvider: async () => { throw new Error('unused'); },
+    getClient: async () => { throw new Error('unused'); },
+    getDriver: async () => { throw new Error('unused'); },
+  };
+}
+
+function makeSession(connection: SeaNativeConnection): SeaSessionBackend {
+  return new SeaSessionBackend({ connection, context: makeContext() });
+}
+
+// ─── Tests ───────────────────────────────────────────────────────────────────
+
+describe('SeaSessionBackend metadata methods', () => {
+  // ── getCatalogs ──────────────────────────────────────────────────────────
+
+  describe('getCatalogs', () => {
+    it('calls listCatalogs() with no args and returns SeaOperationBackend', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      const op = await session.getCatalogs({});
+      expect(op).to.be.instanceOf(SeaOperationBackend);
+      expect(conn.calls).to.have.length(1);
+      expect(conn.calls[0].method).to.equal('listCatalogs');
+      expect(conn.calls[0].args).to.deep.equal([]);
+    });
+
+    it('rejects when session is closed', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.close();
+      let thrown: unknown;
+      try { await session.getCatalogs({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+      expect((thrown as Error).message).to.match(/closed/);
+    });
+
+    it('wraps kernel error via decodeNapiKernelError', async () => {
+      const conn = new FakeMetadataConnection();
+      conn.throwNextCall = 'napi-err';
+      const session = makeSession(conn);
+      let thrown: unknown;
+      try { await session.getCatalogs({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+  });
+
+  // ── getSchemas ───────────────────────────────────────────────────────────
+
+  describe('getSchemas', () => {
+    it('routes catalogName and schemaName to listSchemas', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      const op = await session.getSchemas({ catalogName: 'main', schemaName: 'info%' });
+      expect(op).to.be.instanceOf(SeaOperationBackend);
+      expect(conn.calls[0].method).to.equal('listSchemas');
+      expect(conn.calls[0].args).to.deep.equal(['main', 'info%']);
+    });
+
+    it('passes undefined when request fields are absent', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.getSchemas({});
+      expect(conn.calls[0].args).to.deep.equal([undefined, undefined]);
+    });
+
+    it('rejects when session is closed', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.close();
+      let thrown: unknown;
+      try { await session.getSchemas({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+
+    it('wraps kernel error via decodeNapiKernelError', async () => {
+      const conn = new FakeMetadataConnection();
+      conn.throwNextCall = 'napi-err';
+      let thrown: unknown;
+      try { await makeSession(conn).getSchemas({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+  });
+
+  // ── getTables ────────────────────────────────────────────────────────────
+
+  describe('getTables', () => {
+    it('routes all four args to listTables', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.getTables({
+        catalogName: 'cat',
+        schemaName: 'sch%',
+        tableName: 'tbl%',
+        tableTypes: ['TABLE', 'VIEW'],
+      });
+      expect(conn.calls[0].method).to.equal('listTables');
+      expect(conn.calls[0].args).to.deep.equal(['cat', 'sch%', 'tbl%', ['TABLE', 'VIEW']]);
+    });
+
+    it('passes undefined for absent fields', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.getTables({});
+      expect(conn.calls[0].args).to.deep.equal([undefined, undefined, undefined, undefined]);
+    });
+
+    it('returns SeaOperationBackend when tableTypes is absent', async () => {
+      const conn = new FakeMetadataConnection();
+      const op = await makeSession(conn).getTables({});
+      expect(op).to.be.instanceOf(SeaOperationBackend);
+    });
+
+    it('wraps in SeaTableTypeFilter when tableTypes is provided', async () => {
+      const conn = new FakeMetadataConnection();
+      const op = await makeSession(conn).getTables({ tableTypes: ['TABLE'] });
+      expect(op).to.be.instanceOf(SeaTableTypeFilter);
+    });
+
+    it('wraps in SeaTableTypeFilter when tableTypes is empty array', async () => {
+      const conn = new FakeMetadataConnection();
+      const op = await makeSession(conn).getTables({ tableTypes: [] });
+      expect(op).to.be.instanceOf(SeaTableTypeFilter);
+    });
+
+    it('rejects when session is closed', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.close();
+      let thrown: unknown;
+      try { await session.getTables({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+
+    it('wraps kernel error via decodeNapiKernelError', async () => {
+      const conn = new FakeMetadataConnection();
+      conn.throwNextCall = 'napi-err';
+      let thrown: unknown;
+      try { await makeSession(conn).getTables({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+  });
+
+  // ── getTableTypes ────────────────────────────────────────────────────────
+
+  describe('getTableTypes', () => {
+    it('calls listTableTypes() with no args', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      const op = await session.getTableTypes({});
+      expect(op).to.be.instanceOf(SeaOperationBackend);
+      expect(conn.calls[0].method).to.equal('listTableTypes');
+      expect(conn.calls[0].args).to.deep.equal([]);
+    });
+
+    it('rejects when session is closed', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.close();
+      let thrown: unknown;
+      try { await session.getTableTypes({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+
+    it('wraps kernel error via decodeNapiKernelError', async () => {
+      const conn = new FakeMetadataConnection();
+      conn.throwNextCall = 'napi-err';
+      let thrown: unknown;
+      try { await makeSession(conn).getTableTypes({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+  });
+
+  // ── getTypeInfo ──────────────────────────────────────────────────────────
+
+  describe('getTypeInfo', () => {
+    it('calls listTypeInfo() and returns SeaOperationBackend', async () => {
+      const conn = new FakeMetadataConnection();
+      const op = await makeSession(conn).getTypeInfo({});
+      expect(op).to.be.instanceOf(SeaOperationBackend);
+      expect(conn.calls[0].method).to.equal('listTypeInfo');
+    });
+
+    it('rejects when session is closed', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.close();
+      let thrown: unknown;
+      try { await session.getTypeInfo({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+
+    it('wraps kernel error via decodeNapiKernelError', async () => {
+      const conn = new FakeMetadataConnection();
+      conn.throwNextCall = 'napi-err';
+      let thrown: unknown;
+      try { await makeSession(conn).getTypeInfo({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+  });
+
+  // ── getColumns ───────────────────────────────────────────────────────────
+
+  describe('getColumns', () => {
+    it('routes all four args to listColumns', async () => {
+      const conn = new FakeMetadataConnection();
+      await makeSession(conn).getColumns({
+        catalogName: 'c',
+        schemaName: 's',
+        tableName: 't',
+        columnName: 'col%',
+      });
+      expect(conn.calls[0].method).to.equal('listColumns');
+      expect(conn.calls[0].args).to.deep.equal(['c', 's', 't', 'col%']);
+    });
+
+    it('passes undefined for absent fields', async () => {
+      const conn = new FakeMetadataConnection();
+      await makeSession(conn).getColumns({});
+      expect(conn.calls[0].args).to.deep.equal([undefined, undefined, undefined, undefined]);
+    });
+
+    it('rejects when session is closed', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.close();
+      let thrown: unknown;
+      try { await session.getColumns({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+
+    it('wraps kernel error via decodeNapiKernelError', async () => {
+      const conn = new FakeMetadataConnection();
+      conn.throwNextCall = 'napi-err';
+      let thrown: unknown;
+      try { await makeSession(conn).getColumns({}); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+  });
+
+  // ── getFunctions ─────────────────────────────────────────────────────────
+
+  describe('getFunctions', () => {
+    it('routes catalogName, schemaName, functionName to listFunctions', async () => {
+      const conn = new FakeMetadataConnection();
+      await makeSession(conn).getFunctions({
+        catalogName: 'c',
+        schemaName: 's%',
+        functionName: 'fn%',
+      });
+      expect(conn.calls[0].method).to.equal('listFunctions');
+      expect(conn.calls[0].args).to.deep.equal(['c', 's%', 'fn%']);
+    });
+
+    it('passes undefined catalogName when absent', async () => {
+      const conn = new FakeMetadataConnection();
+      await makeSession(conn).getFunctions({ functionName: 'myfn' });
+      expect(conn.calls[0].args).to.deep.equal([undefined, undefined, 'myfn']);
+    });
+
+    it('rejects when session is closed', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.close();
+      let thrown: unknown;
+      try { await session.getFunctions({ functionName: 'f' }); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+
+    it('wraps kernel error via decodeNapiKernelError', async () => {
+      const conn = new FakeMetadataConnection();
+      conn.throwNextCall = 'napi-err';
+      let thrown: unknown;
+      try { await makeSession(conn).getFunctions({ functionName: 'f' }); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+  });
+
+  // ── getPrimaryKeys ───────────────────────────────────────────────────────
+
+  describe('getPrimaryKeys', () => {
+    it('routes catalogName, schemaName, tableName to getPrimaryKeys', async () => {
+      const conn = new FakeMetadataConnection();
+      await makeSession(conn).getPrimaryKeys({
+        catalogName: 'cat',
+        schemaName: 'myschema',
+        tableName: 'orders',
+      });
+      expect(conn.calls[0].method).to.equal('getPrimaryKeys');
+      expect(conn.calls[0].args).to.deep.equal(['cat', 'myschema', 'orders']);
+    });
+
+    it('throws HiveDriverError when catalogName is absent', async () => {
+      const conn = new FakeMetadataConnection();
+      let thrown: unknown;
+      try { await makeSession(conn).getPrimaryKeys({ schemaName: 'sch', tableName: 'tbl' }); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+      expect((thrown as Error).message).to.match(/catalogName is required/);
+    });
+
+    it('returns SeaOperationBackend when all args present', async () => {
+      const conn = new FakeMetadataConnection();
+      const op = await makeSession(conn).getPrimaryKeys({ catalogName: 'cat', schemaName: 's', tableName: 't' });
+      expect(op).to.be.instanceOf(SeaOperationBackend);
+    });
+
+    it('rejects when session is closed', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.close();
+      let thrown: unknown;
+      try { await session.getPrimaryKeys({ catalogName: 'cat', schemaName: 's', tableName: 't' }); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+
+    it('wraps kernel error via decodeNapiKernelError', async () => {
+      const conn = new FakeMetadataConnection();
+      conn.throwNextCall = 'kernel-pk-error';
+      let thrown: unknown;
+      try { await makeSession(conn).getPrimaryKeys({ catalogName: 'cat', schemaName: 's', tableName: 't' }); } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+  });
+
+  // ── getCrossReference ────────────────────────────────────────────────────
+
+  describe('getCrossReference', () => {
+    it('routes all 6 fields to getCrossReference in the right order', async () => {
+      const conn = new FakeMetadataConnection();
+      await makeSession(conn).getCrossReference({
+        parentCatalogName: 'pc',
+        parentSchemaName: 'ps',
+        parentTableName: 'pt',
+        foreignCatalogName: 'fc',
+        foreignSchemaName: 'fs',
+        foreignTableName: 'ft',
+      });
+      expect(conn.calls[0].method).to.equal('getCrossReference');
+      expect(conn.calls[0].args).to.deep.equal(['pc', 'ps', 'pt', 'fc', 'fs', 'ft']);
+    });
+
+    it('returns SeaOperationBackend', async () => {
+      const conn = new FakeMetadataConnection();
+      const op = await makeSession(conn).getCrossReference({
+        parentCatalogName: 'pc',
+        parentSchemaName: 'ps',
+        parentTableName: 'pt',
+        foreignCatalogName: 'fc',
+        foreignSchemaName: 'fs',
+        foreignTableName: 'ft',
+      });
+      expect(op).to.be.instanceOf(SeaOperationBackend);
+    });
+
+    it('throws HiveDriverError when foreignCatalogName is absent', async () => {
+      const conn = new FakeMetadataConnection();
+      let thrown: unknown;
+      try {
+        await makeSession(conn).getCrossReference({
+          parentCatalogName: 'pc', parentSchemaName: 'ps', parentTableName: 'pt',
+          foreignCatalogName: '', foreignSchemaName: 'fs', foreignTableName: 'ft',
+        });
+      } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+      expect((thrown as Error).message).to.match(/foreignCatalogName is required/);
+    });
+
+    it('throws HiveDriverError when foreignSchemaName is absent', async () => {
+      const conn = new FakeMetadataConnection();
+      let thrown: unknown;
+      try {
+        await makeSession(conn).getCrossReference({
+          parentCatalogName: 'pc', parentSchemaName: 'ps', parentTableName: 'pt',
+          foreignCatalogName: 'fc', foreignSchemaName: '', foreignTableName: 'ft',
+        });
+      } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+      expect((thrown as Error).message).to.match(/foreignSchemaName is required/);
+    });
+
+    it('throws HiveDriverError when foreignTableName is absent', async () => {
+      const conn = new FakeMetadataConnection();
+      let thrown: unknown;
+      try {
+        await makeSession(conn).getCrossReference({
+          parentCatalogName: 'pc', parentSchemaName: 'ps', parentTableName: 'pt',
+          foreignCatalogName: 'fc', foreignSchemaName: 'fs', foreignTableName: '',
+        });
+      } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+      expect((thrown as Error).message).to.match(/foreignTableName is required/);
+    });
+
+    it('rejects when session is closed', async () => {
+      const conn = new FakeMetadataConnection();
+      const session = makeSession(conn);
+      await session.close();
+      let thrown: unknown;
+      try {
+        await session.getCrossReference({
+          parentCatalogName: 'pc',
+          parentSchemaName: 'ps',
+          parentTableName: 'pt',
+          foreignCatalogName: 'fc',
+          foreignSchemaName: 'fs',
+          foreignTableName: 'ft',
+        });
+      } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+
+    it('wraps kernel error via decodeNapiKernelError', async () => {
+      const conn = new FakeMetadataConnection();
+      conn.throwNextCall = 'napi-err';
+      let thrown: unknown;
+      try {
+        await makeSession(conn).getCrossReference({
+          parentCatalogName: 'pc', parentSchemaName: 'ps', parentTableName: 'pt',
+          foreignCatalogName: 'fc', foreignSchemaName: 'fs', foreignTableName: 'ft',
+        });
+      } catch (e) { thrown = e; }
+      expect(thrown).to.be.instanceOf(HiveDriverError);
+    });
+  });
+});
+
+// ─── SeaTableTypeFilter behavior ─────────────────────────────────────────────
+
+describe('SeaTableTypeFilter fetchChunk row reduction', () => {
+  const MIXED_ROWS = [
+    { TABLE_TYPE: 'TABLE', TABLE_NAME: 't1' },
+    { TABLE_TYPE: 'VIEW', TABLE_NAME: 'v1' },
+    { TABLE_TYPE: 'TABLE', TABLE_NAME: 't2' },
+    { TABLE_TYPE: 'SYSTEM TABLE', TABLE_NAME: 's1' },
+  ];
+
+  function makeInner(rows: object[]): IOperationBackend {
+    return {
+      id: 'fake',
+      hasResultSet: true,
+      async fetchChunk() { return rows; },
+      async hasMore() { return false; },
+      async waitUntilReady() {},
+      async status() { return {} as any; },
+      async getResultMetadata() { return {} as any; },
+      async cancel() { return {} as any; },
+      async close() { return {} as any; },
+    };
+  }
+
+  it('keeps only rows whose TABLE_TYPE is in the allowed set', async () => {
+    const filter = new SeaTableTypeFilter(makeInner(MIXED_ROWS), new Set(['TABLE']));
+    const rows = await filter.fetchChunk({ limit: 100 });
+    expect(rows).to.have.length(2);
+    expect(rows.every((r) => (r as Record<string, unknown>).TABLE_TYPE === 'TABLE')).to.be.true;
+  });
+
+  it('returns empty array when allowedTypes is an empty set', async () => {
+    const filter = new SeaTableTypeFilter(makeInner(MIXED_ROWS), new Set());
+    const rows = await filter.fetchChunk({ limit: 100 });
+    expect(rows).to.have.length(0);
+  });
+});