Add dynamic SQL-view tools for schema discovery: `search_schema` and `inspect_table`

Summary

Add a documentation-only example for two dynamic MCP tools implemented entirely as ClickHouse SQL views:

• search_schema: broad metadata search across databases, tables, columns, types, and comments.
• inspect_table: compact structured inspection for one selected table.

No Go changes are required. The implementation should be delivered as a separate Markdown file containing ready-to-use YAML and SQL.

Suggested file:

docs/examples/dynamic-schema-tools.md

Motivation

Agents currently can use execute_query to query system.tables, system.columns, and SHOW CREATE TABLE, but that forces the model to repeatedly generate ad hoc metadata SQL.

For agentic SQL analysis, the recurring workflow is usually:

find relevant tables -> inspect selected table -> write SQL

Dynamic view-backed tools are a good fit for this because Altinity MCP can expose ClickHouse views as read-only MCP tools using server.tools[].view_regexp.

The goal is to provide a reusable pattern that gives agents structured metadata without adding static Go tools or increasing the built-in tool surface.

Proposed dynamic tools

1. search_schema

Purpose:

Find candidate tables and columns relevant to a natural-language or keyword query.

This tool should be broad and shallow. It should not return full schemas or DDL.

It should search:

• database name
• table name
• table comment
• column name
• column type
• column comment

Expected use:

User asks: "Show revenue by customer for last month."

Agent calls:
search_schema(query = "revenue customer amount invoice payment", database = "", limit = 10)

Expected result shape:

{
  "database": "prod",
  "table": "payments",
  "engine": "MergeTree",
  "table_comment": "Customer payment events",
  "matched_on": ["table_name", "column_name"],
  "matched_column_count": 3,
  "matched_columns": [
    {
      "position": "2",
      "name": "customer_id",
      "type": "UInt64",
      "comment": "Customer identifier"
    },
    {
      "position": "5",
      "name": "amount",
      "type": "Decimal(18, 2)",
      "comment": "Payment amount"
    }
  ],
  "partition_by": "toYYYYMM(paid_at)",
  "order_by": "customer_id, paid_at",
  "total_rows": 100000000,
  "total_bytes": 1234567890,
  "relevance_score": 180
}

2. inspect_table

Purpose:

Return compact structured metadata for one selected table.

This tool should be narrow and precise. It replaces the need for separate get_table_schema and get_table_overview tools or writing targeted SQL queries.

It should return:

• engine
• partition key
• sorting key
• primary key
• sampling key
• row/byte estimates
• part/partition summary
• structured columns
• likely time columns
• likely dimension columns
• likely measure columns
• optional raw DDL only when requested
• sample SELECT statement, but not actual sample data

Important: actual arbitrary table sampling is intentionally not included here because a generic SQL view should keep a stable output shape. The tool can return a sample_select_sql string that the agent may run through execute_query if needed.

Expected use:

Agent calls:
inspect_table(database = "prod", table = "payments", include_ddl = 0)

Expected result shape:

{
  "database": "prod",
  "table": "payments",
  "engine": "MergeTree",
  "partition_by": "toYYYYMM(paid_at)",
  "order_by": "customer_id, paid_at",
  "primary_key": "customer_id, paid_at",
  "column_count": 12,
  "columns": [
    {
      "position": "1",
      "name": "customer_id",
      "type": "UInt64",
      "nullable": "false",
      "in_sorting_key": "true",
      "in_primary_key": "true"
    }
  ],
  "likely_time_columns": ["paid_at"],
  "likely_dimension_columns": ["customer_id", "currency", "status"],
  "likely_measure_columns": ["amount"],
  "sample_select_sql": "SELECT `customer_id`, `amount`, `paid_at` FROM `prod`.`payments` LIMIT 10",
  "ddl": ""
}

Proposed Markdown file content

Create:

docs/examples/dynamic-schema-tools.md

with the following content.

# Dynamic schema tools

This example adds two read-only MCP tools using ClickHouse SQL views:

- `ch_mcp_search_schema`
- `ch_mcp_inspect_table`

No Go code changes are required.

## MCP configuration

Add the dynamic views under `server.tools`.

```yaml
clickhouse:
  host: localhost
  port: 8123
  username: default
  password: ""
  database: default
  read_only: true

  # Optional but recommended guardrails for agent-facing tools.
  max_result_rows: 500
  max_result_bytes: 50000

server:
  tools:
    - type: read
      name: execute_query

    - type: read
      view_regexp: "mcp\\.(search_schema|inspect_table)"
      prefix: "ch_"
```

Expected tool names:

```text
ch_mcp_search_schema
ch_mcp_inspect_table
```

## SQL setup

Run the following SQL in ClickHouse.

```sql
CREATE DATABASE IF NOT EXISTS mcp;
```

## Tool 1: `search_schema`

```sql
CREATE OR REPLACE VIEW mcp.search_schema AS
WITH
    {query: String} AS p_query,
    {database: String} AS p_database,
    greatest(toUInt32(1), least({limit: UInt32}, toUInt32(50))) AS p_limit
SELECT
    database,
    table,
    any(engine) AS engine,
    nullIf(any(table_comment), '') AS table_comment,

    nullIf(any(partition_by), '') AS partition_by,
    nullIf(any(order_by), '') AS order_by,
    nullIf(any(primary_key), '') AS primary_key,

    any(total_rows) AS total_rows,
    any(total_bytes) AS total_bytes,

    arrayFilter(
        x -> x != '',
        [
            if(max(table_name_match) = 1, 'table_name', ''),
            if(max(table_comment_match) = 1, 'table_comment', ''),
            if(countIf(column_name_match) > 0, 'column_name', ''),
            if(countIf(column_type_match) > 0, 'column_type', ''),
            if(countIf(column_comment_match) > 0, 'column_comment', '')
        ]
    ) AS matched_on,

    countIf(column_match) AS matched_column_count,

    arrayMap(
        x -> map(
            'position', toString(x.1),
            'name', x.2,
            'type', x.3,
            'comment', x.4
        ),
        arraySlice(
            arraySort(
                x -> x.1,
                groupArrayIf(
                    tuple(
                        position,
                        column_name,
                        column_type,
                        column_comment
                    ),
                    column_match
                )
            ),
            1,
            20
        )
    ) AS matched_columns,

    (
        100 * max(table_name_match)
      + 80  * max(table_comment_match)
      + 40  * countIf(column_name_match)
      + 20  * countIf(column_comment_match)
      + 10  * countIf(column_type_match)
    ) AS relevance_score

FROM
(
    SELECT
        c.database,
        c.table,
        c.position,
        c.name AS column_name,
        c.type AS column_type,
        ifNull(c.comment, '') AS column_comment,

        t.engine AS engine,
        ifNull(t.comment, '') AS table_comment,
        ifNull(t.partition_key, '') AS partition_by,
        ifNull(t.sorting_key, '') AS order_by,
        ifNull(t.primary_key, '') AS primary_key,
        t.total_rows AS total_rows,
        t.total_bytes AS total_bytes,

        p_query != ''
            AND positionCaseInsensitive(c.table, p_query) > 0
            AS table_name_match,

        p_query != ''
            AND positionCaseInsensitive(ifNull(t.comment, ''), p_query) > 0
            AS table_comment_match,

        p_query != ''
            AND positionCaseInsensitive(c.name, p_query) > 0
            AS column_name_match,

        p_query != ''
            AND positionCaseInsensitive(c.type, p_query) > 0
            AS column_type_match,

        p_query != ''
            AND positionCaseInsensitive(ifNull(c.comment, ''), p_query) > 0
            AS column_comment_match,

        (
            p_query = ''
            OR table_name_match
            OR table_comment_match
            OR column_name_match
            OR column_type_match
            OR column_comment_match
        ) AS column_match

    FROM system.columns AS c
    INNER JOIN system.tables AS t
        ON t.database = c.database
       AND t.name = c.table

    WHERE
        (p_database = '' OR c.database = p_database)
        AND c.database NOT IN ('system', 'INFORMATION_SCHEMA', 'information_schema')
        AND (
            p_query = ''
            OR positionCaseInsensitive(c.database, p_query) > 0
            OR positionCaseInsensitive(c.table, p_query) > 0
            OR positionCaseInsensitive(ifNull(t.comment, ''), p_query) > 0
            OR positionCaseInsensitive(c.name, p_query) > 0
            OR positionCaseInsensitive(c.type, p_query) > 0
            OR positionCaseInsensitive(ifNull(c.comment, ''), p_query) > 0
        )
)
GROUP BY
    database,
    table
ORDER BY
    relevance_score DESC,
    database ASC,
    table ASC
LIMIT p_limit
COMMENT '{
  "title": "Search ClickHouse schema",
  "description": "Find candidate tables and columns by keyword across table names, column names, types, and comments. Use this before inspect_table when the relevant table is not known.",
  "params": {
    "query": "Keyword or phrase to search for. Use an empty string to list tables.",
    "database": "Exact database name to restrict search. Use an empty string to search all non-system databases.",
    "limit": "Maximum number of candidate tables to return. Values above 50 are capped."
  },
  "annotations": {
    "openWorldHint": false
  }
}';
```

## Tool 2: `inspect_table`

```sql
CREATE OR REPLACE VIEW mcp.inspect_table AS
WITH
    {database: String} AS p_database,
    {table: String} AS p_table,
    {include_ddl: UInt8} AS p_include_ddl
SELECT
    t.database,
    t.name AS table,
    t.engine,

    nullIf(t.engine_full, '') AS engine_full,
    nullIf(t.partition_key, '') AS partition_by,
    nullIf(t.sorting_key, '') AS order_by,
    nullIf(t.primary_key, '') AS primary_key,
    nullIf(t.sampling_key, '') AS sampling_key,

    coalesce(ps.active_rows, t.total_rows, toUInt64(0)) AS row_count_estimate,
    coalesce(ps.active_bytes_on_disk, t.total_bytes, toUInt64(0)) AS bytes_estimate,

    ifNull(ps.active_parts, toUInt64(0)) AS active_parts,
    ifNull(ps.active_partitions, toUInt64(0)) AS active_partitions,

    nullIf(t.comment, '') AS table_comment,

    ci.column_count,
    ci.columns,

    ci.likely_time_columns,
    ci.likely_dimension_columns,
    ci.likely_measure_columns,

    concat(
        'SELECT ',
        arrayStringConcat(
            arrayMap(
                x -> concat('`', replaceAll(x, '`', '``'), '`'),
                ci.first_column_names
            ),
            ', '
        ),
        ' FROM `',
        replaceAll(t.database, '`', '``'),
        '`.`',
        replaceAll(t.name, '`', '``'),
        '` LIMIT 10'
    ) AS sample_select_sql,

    if(p_include_ddl = 1, t.create_table_query, '') AS ddl

FROM system.tables AS t

LEFT JOIN
(
    SELECT
        database,
        table,

        count() AS column_count,

        arrayMap(
            x -> map(
                'position', toString(x.1),
                'name', x.2,
                'type', x.3,

                'nullable',
                if(position(x.3, 'Nullable(') > 0, 'true', 'false'),

                'low_cardinality',
                if(position(x.3, 'LowCardinality(') > 0, 'true', 'false'),

                'default_kind', x.4,
                'default_expression', x.5,
                'comment', x.6,

                'in_partition_key',
                if(x.7 = 1, 'true', 'false'),

                'in_sorting_key',
                if(x.8 = 1, 'true', 'false'),

                'in_primary_key',
                if(x.9 = 1, 'true', 'false'),

                'in_sampling_key',
                if(x.10 = 1, 'true', 'false'),

                'compression_codec', x.11,

                'compressed_bytes', toString(x.12),
                'uncompressed_bytes', toString(x.13)
            ),
            arraySort(
                x -> x.1,
                groupArray(
                    tuple(
                        position,
                        name,
                        type,
                        ifNull(default_kind, ''),
                        ifNull(default_expression, ''),
                        ifNull(comment, ''),
                        is_in_partition_key,
                        is_in_sorting_key,
                        is_in_primary_key,
                        is_in_sampling_key,
                        ifNull(compression_codec, ''),
                        data_compressed_bytes,
                        data_uncompressed_bytes
                    )
                )
            )
        ) AS columns,

        arrayMap(
            x -> x.2,
            arraySlice(
                arraySort(x -> x.1, groupArray(tuple(position, name))),
                1,
                20
            )
        ) AS first_column_names,

        groupArrayIf(
            name,
            position(type, 'Date') > 0
            OR position(type, 'Time') > 0
        ) AS likely_time_columns,

        groupArrayIf(
            name,
            (
                position(type, 'String') > 0
                OR position(type, 'Enum') > 0
                OR position(type, 'LowCardinality') > 0
                OR positionCaseInsensitive(name, 'id') > 0
                OR is_in_primary_key = 1
                OR is_in_sorting_key = 1
            )
            AND position(type, 'Date') = 0
            AND position(type, 'Time') = 0
        ) AS likely_dimension_columns,

        groupArrayIf(
            name,
            (
                position(type, 'Int') > 0
                OR position(type, 'Float') > 0
                OR position(type, 'Decimal') > 0
            )
            AND is_in_primary_key = 0
            AND is_in_sorting_key = 0
            AND positionCaseInsensitive(name, 'id') = 0
        ) AS likely_measure_columns

    FROM system.columns
    WHERE database = p_database
      AND table = p_table
    GROUP BY
        database,
        table
) AS ci
    ON ci.database = t.database
   AND ci.table = t.name

LEFT JOIN
(
    SELECT
        database,
        table,
        count() AS active_parts,
        uniqExact(partition) AS active_partitions,
        sum(rows) AS active_rows,
        sum(bytes_on_disk) AS active_bytes_on_disk
    FROM system.parts
    WHERE active
      AND database = p_database
      AND table = p_table
    GROUP BY
        database,
        table
) AS ps
    ON ps.database = t.database
   AND ps.table = t.name

WHERE t.database = p_database
  AND t.name = p_table
COMMENT '{
  "title": "Inspect ClickHouse table",
  "description": "Return compact structured metadata for one ClickHouse table, including engine, keys, row and byte estimates, columns, likely time/dimension/measure columns, and an optional raw CREATE TABLE statement.",
  "params": {
    "database": "Database name containing the table.",
    "table": "Table name to inspect.",
    "include_ddl": "Set to 1 to include raw CREATE TABLE DDL. Use 0 by default to keep context small."
  },
  "annotations": {
    "openWorldHint": false
  }
}';
```

## Ready-to-use examples

### Example 1: find candidate tables

Tool call:

```json
{
  "tool": "ch_mcp_search_schema",
  "arguments": {
    "query": "revenue customer amount payment",
    "database": "",
    "limit": 10
  }
}
```

Expected agent behavior:

```text
Use this when the user asks a question but the relevant table is unknown.
Pick the strongest candidate table, then call ch_mcp_inspect_table.
```

### Example 2: inspect a selected table

Tool call:

```json
{
  "tool": "ch_mcp_inspect_table",
  "arguments": {
    "database": "prod",
    "table": "payments",
    "include_ddl": 0
  }
}
```

Expected agent behavior:

```text
Use this before writing analytical SQL.
Prefer likely_time_columns for time filters.
Prefer likely_measure_columns for SUM/AVG metrics.
Use order_by and primary_key to avoid inefficient full scans when possible.
```

### Example 3: request raw DDL only when needed

Tool call:

```json
{
  "tool": "ch_mcp_inspect_table",
  "arguments": {
    "database": "prod",
    "table": "payments",
    "include_ddl": 1
  }
}
```

Expected agent behavior:

```text
Use include_ddl = 1 only for debugging, documentation, unusual engines, TTLs, projections, or exact schema auditing.
Keep include_ddl = 0 for normal query generation.
```

### Example 4: preview actual rows

`inspect_table` intentionally does not fetch arbitrary sample rows. It returns `sample_select_sql`.

Example returned value:

```sql
SELECT `customer_id`, `amount`, `paid_at` FROM `prod`.`payments` LIMIT 10
```

The agent can run that through `execute_query` only when actual data values are needed.

## Acceptance criteria

- A new Markdown file is added with YAML and SQL only.
- No Go code changes are required.
- `server.tools` example exposes both views as read-only dynamic tools.
- The example creates:
  - `mcp.search_schema`
  - `mcp.inspect_table`
- The generated tool names are documented:
  - `ch_mcp_search_schema`
  - `ch_mcp_inspect_table`
- `inspect_table` does not include raw DDL by default.
- `include_ddl` is documented as an explicit context-expanding option.
- `inspect_table` returns a sample query string, not arbitrary sample data.
- Result sizes are bounded by existing `max_result_rows` and `max_result_bytes` settings.

## Non-goals

- Do not add static Go tools.
- Do not implement generic arbitrary-table sampling inside a dynamic view.
- Do not add charting or result merging.
- Do not replace `execute_query`.
- Do not expose raw DDL by default.

## Notes

This keeps the MCP tool surface small:

```text
search_schema -> inspect_table -> execute_query
```

That gives agents enough structure for reliable ClickHouse SQL generation without adding many narrowly overlapping tools.

Create docs/examples/dynamic-schema-tools.md as a documentation-only change. Start by reviewing the supplied MCP YAML configuration and the SQL definitions for mcp.search_schema and mcp.inspect_table; done means the file documents both read-only tools, their parameters, expected outputs, and setup without requiring Go changes.