generate_documentation.py

# Copyright (c) 2025-2026 ADBC Drivers Contributors
#
# 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.

"""
Generate user documentation based on validation suite results.
"""

import collections
import dataclasses
import datetime
import functools
import html
import json
import typing
import xml.etree.ElementTree
import zoneinfo
from pathlib import Path

import bidict
import duckdb
import jinja2
import pyarrow

from . import model
from .utils import arrow_type_name


class GetQuirks(typing.Protocol):
    """Get the quirks for a driver given the vendor/version."""

    def __call__(self, version: str, *, vendor: str) -> model.DriverQuirks: ...


@dataclasses.dataclass
class Span:
    """A span of cells in a table."""

    span: int
    value: str

    def __iter__(self) -> typing.Iterator[int | str]:
        # Allow unpacking into (span, value) for the template
        yield self.span
        yield self.value


@dataclasses.dataclass
class CustomFeature:
    """A custom feature that a driver supports."""

    #: The name to display.
    name: str
    #: A short description.
    description: str
    #: Whether the test(s) for this feature passed.
    supported: bool


@dataclasses.dataclass
class CustomFeatures:
    """A set of custom features that a driver supports."""

    #: A mapping from categories to a list of subfeatures.
    groups: dict[str, list[CustomFeature]] = dataclasses.field(
        default_factory=lambda: collections.defaultdict(list)
    )


@dataclasses.dataclass(frozen=True)
class TypeTableEntry:
    lhs: str
    rhs: tuple[str, ...]
    result: typing.Literal["passed", "partial", "failed"]
    footnotes: tuple[str, ...] = dataclasses.field(default_factory=tuple)
    # e.g. for BigQuery, we may have both "ingest (default)" and "ingest (with
    # Storage Write API)" for the same type
    variant: str | None = None

    def render_rhs(self) -> str:
        rhs = ", ".join(self.rhs)
        if self.result == "failed":
            rhs = "❌"
        elif self.result == "partial":
            rhs += " ⚠️"
        for footnote in self.footnotes:
            rhs += footnote
        return rhs

    def merge(self, other: "TypeTableEntry") -> "TypeTableEntry":
        if self.lhs != other.lhs:
            raise ValueError(f"Cannot merge lhs {self.lhs} and {other.lhs}")
        if self.variant != other.variant:
            raise ValueError(f"Cannot merge variant {self.variant} and {other.variant}")

        # failed + failed = failed
        # failed + (other) = partial
        # partial + (anything) or (anything) + partial = partial
        # passed + passed = passed
        rhs = tuple(sorted(set(self.rhs) | set(other.rhs)))
        if self.result == "failed" and other.result == "failed":
            result = "failed"
        elif self.result == "failed":
            result = "partial"
            # we don't want the failed side to contribute to the RHS
            rhs = other.rhs
        elif other.result == "failed":
            result = "partial"
            rhs = self.rhs
        elif self.result == "partial" or other.result == "partial":
            result = "partial"
        else:
            result = "passed"

        return TypeTableEntry(
            lhs=self.lhs,
            rhs=rhs,
            result=result,
            footnotes=tuple(sorted(set(self.footnotes) | set(other.footnotes))),
            variant=self.variant,
        )


@dataclasses.dataclass
class DriverTypeTable:
    """A table of features supported by a driver."""

    quirks: model.DriverQuirks
    features: model.DriverFeatures

    custom_features: CustomFeatures = dataclasses.field(default_factory=CustomFeatures)

    type_select: list[TypeTableEntry] = dataclasses.field(default_factory=list)
    type_bind: list[TypeTableEntry] = dataclasses.field(default_factory=list)
    type_ingest: list[TypeTableEntry] = dataclasses.field(default_factory=list)

    get_objects: dict[str, bool] = dataclasses.field(default_factory=dict)
    get_table_schema: bool = False
    ingest: dict[str, bool] = dataclasses.field(default_factory=dict)
    vendor_version: str = "unknown"

    def pprint(self) -> str:
        # Slightly more friendly representation for debugging
        lines = []

        lines.append("Features")
        lines.append("~~~~~~~~")

        for field in self.features.__fields__:
            if field.startswith("_"):
                continue

            value = getattr(self.features, field)
            if isinstance(value, bool):
                value = "✅" if value else "❌"
            lines.append(f"- {field}: {value}")

        for group, features in self.custom_features.groups.items():
            lines.append(f"- {group}:")
            for feature in features:
                status = "✅" if feature.supported else "❌"
                lines.append(f"  - {feature.name}: {status} {feature.description}")

        lines.append("")
        lines.append("GetObjects")
        lines.append("~~~~~~~~~~")
        for name, supported in self.get_objects.items():
            status = "✅" if supported else "❌"
            lines.append(f"- {name}: {status}")

        status = "✅" if self.get_table_schema else "❌"
        lines.append("")
        lines.append(f"GetTableSchema: {status}")

        lines.append("")
        lines.append("Ingest Modes")
        lines.append("~~~~~~~~~~~~")
        for name, supported in self.ingest.items():
            status = "✅" if supported else "❌"
            lines.append(f"- {name}: {status}")

        def render_type_table(category: str, entries: list[TypeTableEntry]) -> None:
            if not entries:
                return
            lines.append("")
            lines.append(f"{category.capitalize()} Types")
            lines.append("~" * (len(category) + 6))
            max_lhs = max(len(entry.lhs) for entry in entries)
            for entry in entries:
                line = f"- {entry.lhs.ljust(max_lhs)} → {entry.render_rhs()}"
                if entry.variant:
                    line += f" ({entry.variant})"
                lines.append(line)

        render_type_table("select", self.type_select)
        render_type_table("bind", self.type_bind)
        render_type_table("ingest", self.type_ingest)

        return "\n".join(lines)


@dataclasses.dataclass(frozen=True)
class VendorVersion:
    vendor: str
    version: str


@dataclasses.dataclass
class ValidationReport:
    driver: str
    versions: dict[VendorVersion, DriverTypeTable]
    driver_version: str = "unknown"
    footnotes: bidict.bidict[int, str] = dataclasses.field(
        default_factory=bidict.bidict
    )

    def get_version(self, test_case: dict[str, typing.Any]) -> DriverTypeTable:
        return self.versions[
            VendorVersion(test_case["vendor"], test_case["vendor_version"])
        ]

    def pprint(self) -> str:
        # Slightly more friendly representation for debugging
        lines = []

        lines.append(f"Driver Version: {self.driver_version}")
        lines.append("")

        lines.append("Versions")
        lines.append("========")

        for version, table in self.versions.items():
            version_repr = f"{version.vendor} {version.version}"
            lines.append("")
            lines.append(version_repr)
            lines.append("-" * len(version_repr))
            lines.append(table.pprint())

        lines.append("")
        lines.append("Footnotes")
        lines.append("=========")
        for idx, footnote in self.footnotes.items():
            lines.append(f"[^{idx}]: {footnote}")

        return "\n".join(lines)

    def add_footnote(self, contents: str) -> str:
        if contents in self.footnotes.inverse:
            counter = self.footnotes.inverse[contents]
            return f" [^{counter}]"
        counter = len(self.footnotes) + 1
        self.footnotes[counter] = contents
        return f" [^{counter}]"

    def add_table_entry(
        self,
        vendor: str,
        vendor_version: str,
        category: typing.Literal["select", "bind", "ingest"],
        lhs: str,
        rhs: str | list[str],
        test_case: dict[str, typing.Any],
        *,
        extra_caveats: list[str] | None = None,
        variant: str | None = None,
    ) -> None:
        caveats = []
        passed = test_case["test_results"].count("passed")

        partial_support = False
        for raw_meta in test_case["metadata"]:
            meta = json.loads(raw_meta)
            tags = meta.get("tags", {})
            partial_support = partial_support or tags.get("partial-support", False)
            caveats.extend(tags.get("caveats", []))

            if caveat := tags.get("broken-driver"):
                caveats.append(caveat)
            if caveat := tags.get("broken-vendor"):
                caveats.append(caveat)

        result = "passed"
        if passed == 0:
            result = "failed"
        elif (
            partial_support or passed < len(test_case["test_results"]) or extra_caveats
        ):
            result = "partial"
            for i, (query_name, test_result) in enumerate(
                zip(test_case["query_names"], test_case["test_results"])
            ):
                if test_result == "passed":
                    continue
                elif test_result == "skipped":
                    if skip_reason := json.loads(test_case["metadata"][i]).get("skip"):
                        caveats.append(skip_reason)
                else:
                    query_kind = query_name.split("/")[1]
                    caveats.append(f"{query_kind} is not supported")

        caveats.extend(extra_caveats or [])

        footnotes = []
        for caveat in caveats:
            fn = self.add_footnote(caveat)
            if fn not in footnotes:
                footnotes.append(fn)
        footnotes = tuple(sorted(footnotes))

        version = self.versions[VendorVersion(vendor, vendor_version)]
        getattr(version, f"type_{category}").append(
            TypeTableEntry(
                lhs=lhs,
                rhs=tuple(rhs) if isinstance(rhs, list) else (rhs,),
                result=result,
                footnotes=footnotes,
                variant=variant,
            )
        )


def render_to(
    sink: Path, template: jinja2.Template, kwargs: dict[str, typing.Any]
) -> None:
    rendered = render_part(template, kwargs)
    sink.parent.mkdir(parents=True, exist_ok=True)
    with sink.open("w") as f:
        f.write(rendered)
        f.write("\n")
    print("Generated", sink)


def render_part(template: jinja2.Template, kwargs: dict[str, typing.Any]) -> str:
    rendered = template.render(**kwargs)
    # Eliminate trailing whitespace from lines
    lines = [line.rstrip() for line in rendered.splitlines()]
    # Remove empty lines
    while lines and not lines[-1]:
        lines = lines[:-1]
    while lines and not lines[0]:
        lines = lines[1:]
    rendered = "\n".join(lines)
    return rendered


def load_testcases(get_quirks: GetQuirks, results_path: Path) -> None:
    """Load test case data into DuckDB."""
    report = xml.etree.ElementTree.parse(results_path).getroot()

    testcases = []
    for testcase in report.findall(".//testsuite[@name='validation']/testcase"):
        module = testcase.get("classname")
        name = testcase.get("name")
        if name is None:
            raise ValueError("Testcase without a name")
        if "[" in name:
            name, _, _ = name.partition("[")
        failure = testcase.find("failure")
        error = testcase.find("error")
        skipped = testcase.find("skipped")

        properties = {}
        for prop in testcase.findall(".//properties/property"):
            properties[prop.get("name")] = prop.get("value")

        if "driver" not in properties or not isinstance(properties["driver"], str):
            print(
                "Warning: testcase is missing driver property, skipping:",
                f"{module}.{name}",
            )
            continue

        # XXX: for historical reasons this param was the "driver", but to
        # generalize validation suites for drivers that support multiple
        # vendors/backends (e.g. Spark, MySQL), now it should be
        # "vendor:version".
        driver, _, version = properties["driver"].partition(":")
        quirks = get_quirks(version, vendor=driver)
        query_set = quirks.query_set
        driver = quirks.name
        version = quirks.short_version

        query_name = properties.get("query")
        if query_name is None:
            query = None
            metadata = {}
        else:
            query = query_set.queries[query_name]
            metadata = query.metadata().model_dump(by_alias=True)

        if failure is not None or error is not None:
            test_result = "failed"
        elif skipped is not None:
            if skipped.get("type") == "pytest.xfail":
                test_result = "xfail"
            else:
                test_result = "skipped"
        else:
            test_result = "passed"

        arrow_type = None
        if query_name:
            query = query_set.queries[query_name]
            arrow_type = query.arrow_type_name

        testcases.append(
            {
                "test_module": module,
                "test_name": name,
                "test_result": test_result,
                "vendor": driver,
                "vendor_version": version,
                "query_name": query_name,
                "arrow_type_name": arrow_type,
                "properties": json.dumps(properties),
                "metadata": json.dumps(metadata),
            }
        )

    schema = pyarrow.schema(
        [
            pyarrow.field("test_module", pyarrow.string()),
            pyarrow.field("test_name", pyarrow.string()),
            pyarrow.field("test_result", pyarrow.string()),
            pyarrow.field("vendor", pyarrow.string()),
            pyarrow.field("vendor_version", pyarrow.string()),
            pyarrow.field("query_name", pyarrow.string()),
            pyarrow.field("arrow_type_name", pyarrow.string()),
            # Actually JSON
            pyarrow.field("properties", pyarrow.string()),
            pyarrow.field("metadata", pyarrow.string()),
        ]
    )
    duckdb.register(
        "testcases_raw", pyarrow.Table.from_pylist(testcases, schema=schema)
    )
    duckdb.sql(
        """
        CREATE TABLE IF NOT EXISTS testcases (
              test_module STRING,
              test_name STRING,
              test_result STRING,
              vendor STRING,
              vendor_version STRING,
              query_name STRING,
              arrow_type_name STRING,
              properties JSON,
              metadata JSON,
        );
        INSERT INTO testcases
        SELECT
          test_module,
          test_name,
          test_result,
          vendor,
          vendor_version,
          query_name,
          arrow_type_name,
          CAST(properties AS JSON) AS properties,
          CAST(metadata AS JSON) AS properties,
        FROM testcases_raw
        """
    )


def render(
    report: ValidationReport,
    vendor_mapping: list[tuple[str, str]],
    driver_template_path: Path,
    output_directory: Path,
) -> None:
    # do not sort; use the given order
    vendor_sort: dict[str, tuple[int, str]] = {
        vendor: (idx, friendly) for idx, (vendor, friendly) in enumerate(vendor_mapping)
    }
    print(vendor_sort)

    env = jinja2.Environment(
        loader=jinja2.PackageLoader("adbc_drivers_validation"),
        autoescape=jinja2.select_autoescape(),
        trim_blocks=True,
    )
    env.globals["len"] = len  # ty: ignore[invalid-assignment]
    with driver_template_path.open("r") as source:
        driver_template = env.from_string(source.read())

    driver = report.driver
    template_vars: typing.Dict[str, typing.Any] = {
        "driver": report.driver,
    }

    # ======================================================================
    # Type Support Table
    # ======================================================================

    # Combine select into a single table. The table has a variable number of
    # columns, one for each vendor tested.

    # vendor : { SQL type name : Arrow type name }
    columns = collections.defaultdict(lambda: collections.defaultdict(set))
    for version, type_table in report.versions.items():
        for entry in type_table.type_select:
            columns[version.vendor][entry.lhs].add(entry)

    column_order = list(sorted(columns.keys(), key=lambda v: vendor_sort[v][0]))
    row_order = list(
        sorted(functools.reduce(lambda a, b: a | b, (set(c) for c in columns.values())))
    )
    type_select: list[list[Span]] = []
    for k in row_order:
        all_cells = []
        for c in column_order:
            entries = columns[c].get(k)
            if entries:
                entry = functools.reduce(lambda a, b: a.merge(b), entries)
                all_cells.append(entry.render_rhs())
            else:
                all_cells.append("(NA/not tested)")

        span_cells = [Span(1, k)]
        for i, cell in enumerate(all_cells):
            if i > 0 and cell == span_cells[-1].value:
                span_cells[-1].span += 1
            else:
                span_cells.append(Span(1, cell))
        type_select.append(span_cells)

    template_vars["type_select"] = type_select
    template_vars["type_select_columns"] = column_order

    # Combine bind and ingest into a single type_bind_ingest table. The table
    # has a variable number of columns since ingest may have multiple modes.
    # Because the logic gets complicated, render it entirely in Python rather
    # than in the template

    # vendor: category (bind/ingest/ingest variant...) : { Arrow type name : SQL type names }
    columns = collections.defaultdict(
        lambda: collections.defaultdict(lambda: collections.defaultdict(set))
    )

    for version, type_table in report.versions.items():
        if type_table.features.statement_bind:
            for entry in type_table.type_bind:
                columns[version.vendor]["Bind"][entry.lhs].add(entry)
        for entry in type_table.type_ingest:
            column = "Ingest"
            if entry.variant:
                column += f" ({entry.variant})"
            columns[version.vendor][column][entry.lhs].add(entry)

    vendor_order = list(sorted(columns.keys(), key=lambda v: vendor_sort[v][0]))
    column_order = {
        vendor: list(sorted(columns[vendor].keys())) for vendor in vendor_order
    }
    row_order = []
    if columns:
        row_order = list(
            sorted(
                functools.reduce(
                    lambda a, b: a | b,
                    (set(c) for v in columns.values() for c in v.values()),
                )
            )
        )
    type_bind_ingest: list[list[Span]] = []
    for k in row_order:
        all_cells = []
        for v in vendor_order:
            for c in column_order[v]:
                entries = columns[v][c].get(k)
                if entries:
                    entry = functools.reduce(lambda a, b: a.merge(b), entries)
                    all_cells.append(entry.render_rhs())
                else:
                    all_cells.append("(NA/not tested)")

        span_cells = [Span(1, k)]
        for i, cell in enumerate(all_cells):
            if i > 0 and cell == span_cells[-1].value:
                span_cells[-1].span += 1
            else:
                span_cells.append(Span(1, cell))
        type_bind_ingest.append(span_cells)

    template_vars["type_bind_ingest"] = type_bind_ingest
    template_vars["type_bind_ingest_columns"] = column_order
    template_vars["type_bind_ingest_vendors"] = vendor_order
    template_vars["vendor_friendly_name"] = {
        vendor: vendor_sort[vendor][1] for vendor in vendor_order
    }

    types = render_part(env.get_template("types.md"), template_vars)

    # ======================================================================
    # Feature Support Table
    # ======================================================================

    rows = []
    for version, type_table in report.versions.items():
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Bind Parameters",
                "subfeature": None,
                "suborder": None,
                "supported": type_table.features.statement_bind,
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Bulk Ingestion",
                "subfeature": "Create",
                "suborder": 1,
                "supported": type_table.ingest.get("create", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Bulk Ingestion",
                "subfeature": "Append",
                "suborder": 2,
                "supported": type_table.ingest.get("append", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Bulk Ingestion",
                "subfeature": "Create/Append",
                "suborder": 3,
                "supported": type_table.ingest.get("createappend", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Bulk Ingestion",
                "subfeature": "Replace",
                "suborder": 4,
                "supported": type_table.ingest.get("replace", False)
                and type_table.ingest.get("replace_noop", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Bulk Ingestion",
                "subfeature": "Temporary Table",
                "suborder": 5,
                "supported": type_table.ingest.get("temporary", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Bulk Ingestion",
                "subfeature": "Target Catalog",
                "suborder": 6,
                "supported": type_table.ingest.get("catalog", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Bulk Ingestion",
                "subfeature": "Target Schema",
                "suborder": 7,
                "supported": type_table.ingest.get("schema", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Bulk Ingestion",
                "subfeature": "Non-nullable fields are marked NOT NULL",
                "suborder": 8,
                "supported": type_table.ingest.get("not_null", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Catalog (GetObjects)",
                "subfeature": "depth=catalogs",
                "suborder": 1,
                "supported": type_table.get_objects.get("catalog", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Catalog (GetObjects)",
                "subfeature": "depth=db_schemas",
                "suborder": 2,
                "supported": type_table.get_objects.get("schema", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Catalog (GetObjects)",
                "subfeature": "depth=tables",
                "suborder": 3,
                "supported": type_table.get_objects.get("table", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Catalog (GetObjects)",
                "subfeature": "depth=columns (all)",
                "suborder": 4,
                "supported": type_table.get_objects.get("column", False),
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Get Parameter Schema",
                "subfeature": None,
                "suborder": None,
                "supported": type_table.features.statement_get_parameter_schema,
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Get Table Schema",
                "subfeature": None,
                "suborder": None,
                "supported": type_table.get_table_schema,
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Prepared Statements",
                "subfeature": None,
                "suborder": None,
                "supported": type_table.features.statement_prepare,
            }
        )
        rows.append(
            {
                "vendor": version.vendor,
                "version": version.version,
                "feature": "Transactions",
                "subfeature": None,
                "suborder": None,
                "supported": type_table.features.connection_transactions,
            }
        )

    duckdb.register("features", pyarrow.Table.from_pylist(rows))
    duckdb.sql(
        """
        CREATE VIEW features_agg AS
        FROM features
        SELECT
          feature,
          subfeature,
          suborder,
          vendor,
          CASE
            WHEN BOOL_AND(supported) THEN 'supported'
            WHEN BOOL_OR(supported) THEN 'inconsistent'
            ELSE 'unsupported'
          END AS support
        GROUP BY feature, subfeature, suborder, vendor
        ORDER BY feature, suborder
        """
    )
    raw_features = duckdb.sql(
        """
        PIVOT features_agg
        ON vendor
        USING ANY_VALUE(support)
        GROUP BY feature, subfeature, suborder
        ORDER BY feature, suborder
        """
    )

    features = []
    _raw_vendor_names: list[str] = raw_features.columns[3:]
    # list of (index, ord, vendor_name)
    vendors: list[tuple[int, int, str]] = [
        (i, *vendor_sort[v]) for (i, v) in enumerate(_raw_vendor_names)
    ]
    vendors.sort(key=lambda v: v[1])
    for row in raw_features.fetchall():
        group = row[0]
        subgroup = row[1] or ""
        if not features or features[-1]["feature"] != group:
            features.append(
                {
                    "feature": group,
                    "subfeatures": [],
                }
            )

        span_cells: list[tuple[int, str]] = []
        for i, (index, _, _) in enumerate(vendors):
            cell = row[3 + index]
            if cell == "inconsistent":
                cell = "⚠️" + report.add_footnote("Support varies based on version")
            elif cell == "supported":
                cell = "✅"
            elif cell == "unsupported":
                cell = "❌"
            else:
                raise ValueError(f"Unexpected support value: {cell}")

            if i > 0 and cell == span_cells[-1][1]:
                span_cells[-1] = (span_cells[-1][0] + 1, span_cells[-1][1])
            else:
                span_cells.append((1, cell))

        features[-1]["subfeatures"].append(
            {
                "subfeature": subgroup,
                "support": span_cells,
            }
        )

    # TODO: restore support for driver-specific features (we don't really use
    # this right now)
    features = render_part(
        env.get_template("features.md"),
        {
            "features": features,
            "vendors": [v[2] for v in vendors],
        },
    )

    # ======================================================================
    # Misc.
    # ======================================================================

    footnotes = render_part(
        env.get_template("footnotes.md"),
        {**template_vars, "footnotes": report.footnotes},
    )

    # Assemble the version header/warnings/etc
    is_prerelease = report.driver_version.endswith("-dirty") or any(
        x in report.driver_version for x in ("-dev", "-alpha", "-beta", "-rc", "-pre")
    )

    if is_prerelease:
        ref = f"driver-{driver}-prerelease"
        heading = f"{{badge-primary}}`Driver Version|{report.driver_version}`"
    else:
        ref = f"driver-{driver}-{report.driver_version}"
        heading = f'[{{badge-primary}}`Driver Version|{report.driver_version}`](#{ref} "Permalink")'

    # Assume the release date is the date that the docs are being
    # generated. We've generally tagged the release and uploaded the same day;
    # if not, then it can be edited by hand before merging the docs PR. Use
    # PST as a reference explicitly.
    tz = zoneinfo.ZoneInfo("America/Los_Angeles")
    release_date = datetime.datetime.now(tz).strftime("%Y-%m-%d")
    heading += f" {{badge-secondary}}`Release Date|{release_date}`"

    # TODO: Improve this display for drivers tested with many versions. We
    # probably want to show one badge with a range rather than a badge for every
    # version
    for version in sorted(
        report.versions, key=lambda v: (vendor_sort[v.vendor][0], v.version)
    ):
        # TODO: allow omitting version when it's meaningless (e.g. cloud services)
        friendly_vendor = vendor_sort[version.vendor][1]
        heading += (
            f" {{badge-success}}`Tested With|{friendly_vendor} {version.version}`"
        )

    compatibility_info = "This driver was tested on:\n"
    # Sort and deduplicate by the rendered string not by dev-facing identifier
    compat_versions = set()
    for version in report.versions:
        vendor_name = report.versions[version].quirks.vendor_name
        full_version = report.versions[version].vendor_version
        compat_versions.add(f"\n- {vendor_name} `{full_version}`")
    compatibility_info += "\n".join(sorted(compat_versions))

    if is_prerelease:
        heading += (
            "\n\n:::{warning}\nThis is documentation for a prerelease version.\n:::"
        )

    render_to(
        output_directory / f"{driver}.md",
        driver_template,
        {
            **template_vars,
            "types": types,
            "features": features,
            "footnotes": footnotes,
            "cross_reference": f"({ref})=",
            "heading": heading,
            "version": report.driver_version,
            "compatibility_info": compatibility_info,
        },
    )


def generate_includes(driver: str, get_quirks: GetQuirks) -> ValidationReport:
    # Handle different vendors, multiple versions
    versions = {}
    for row in duckdb.sql(
        "FROM testcases SELECT DISTINCT vendor, vendor_version"
    ).fetchall():
        vendor = row[0]
        version = row[1]
        print("Found suite:", vendor, version)
        quirks = get_quirks(version, vendor=vendor)
        versions[VendorVersion(vendor, version)] = DriverTypeTable(
            quirks=quirks, features=quirks.features
        )

    report = ValidationReport(driver=driver, versions=versions)

    # Version
    version = (
        duckdb.sql("""
    FROM testcases
    SELECT
      vendor,
      vendor_version,
      properties->>'driver_version' AS driver_version,
      properties->>'short_version' AS short_version,
      properties->>'vendor_version' AS full_version,
    WHERE test_name = 'test_get_info'
    """)
        .arrow()
        .read_all()
        .to_pylist()
    )
    if version:
        driver_version = list(
            set(v["driver_version"] for v in version if v["driver_version"])
        )
        if len(driver_version) == 0:
            report.driver_version = "(unknown)"
        elif len(driver_version) != 1:
            raise ValueError(f"Expected one driver version, got {driver_version}")
        else:
            report.driver_version = driver_version[0]
        for v in version:
            short_version = v["short_version"] or "(unknown)"
            full_version = v["full_version"] or "(unknown)"
            report.get_version(v).vendor_version = full_version
    else:
        # No version info available (test_get_info didn't run or failed)
        report.driver_version = "(unknown)"
        for short_version in report.versions:
            report.versions[short_version].vendor_version = "(unknown)"

    # Select type support
    type_tests = (
        duckdb.sql("""
        FROM testcases
        SELECT
          vendor,
          vendor_version,
          metadata->>'tags'->>'sql-type-name' AS sql_type,
          ARRAY_AGG(test_result ORDER BY query_name ASC) AS test_results,
          ARRAY_AGG(query_name ORDER BY query_name ASC) AS query_names,
          ARRAY_AGG(metadata ORDER BY query_name ASC) as metadata,
        WHERE
          test_name = 'test_query'
          AND query_name NOT LIKE 'type/bind/%'
          AND (metadata->>'tags'->>'sql-type-name') IS NOT NULL
        GROUP BY vendor, vendor_version, metadata->>'tags'->>'sql-type-name'
        ORDER BY vendor, vendor_version, metadata->>'tags'->>'sql-type-name'
        """)
        .arrow()
        .read_all()
        .to_pylist()
    )
    for test_case in type_tests:
        query_set = get_quirks(
            test_case["vendor_version"], vendor=test_case["vendor"]
        ).query_set
        arrow_type_names = set()
        for query_name in test_case["query_names"]:
            query = query_set.queries[query_name]
            show_type_parameters = query.metadata().tags.show_arrow_type_parameters

            # Take the first field; some queries may select additional things
            # like nested types to test how a type behaves in different
            # contexts
            field = query.query.expected_schema()[0]
            arrow_type_names.add(
                arrow_type_name(
                    field.type,
                    field.metadata,
                    show_type_parameters=show_type_parameters,
                )
            )
        extra_caveats = []
        if len(arrow_type_names) != 1:
            extra_caveats.append(
                "Return type is inconsistent depending on how the query was written"
            )
        arrow_types = [
            html.escape(arrow_type) for arrow_type in sorted(arrow_type_names)
        ]
        sql_type = html.escape(test_case["sql_type"])
        report.add_table_entry(
            test_case["vendor"],
            test_case["vendor_version"],
            "select",
            sql_type,
            arrow_types,
            test_case,
            extra_caveats=extra_caveats,
        )

    # Bind type support
    type_tests = (
        duckdb.sql("""
        FROM testcases
        SELECT
          vendor,
          vendor_version,
          arrow_type_name,
          metadata->>'tags'->>'sql-type-name' AS sql_type,
          ARRAY_AGG(test_result ORDER BY query_name ASC) AS test_results,
          ARRAY_AGG(query_name ORDER BY query_name ASC) AS query_names,
          ARRAY_AGG(metadata ORDER BY query_name ASC) as metadata,
        WHERE
          test_name = 'test_query'
          AND query_name LIKE 'type/bind/%'
          AND (metadata->>'tags'->>'sql-type-name') IS NOT NULL
        GROUP BY vendor, vendor_version, arrow_type_name, metadata->>'tags'->>'sql-type-name'
        ORDER BY vendor, vendor_version, arrow_type_name, metadata->>'tags'->>'sql-type-name'
        """)
        .arrow()
        .read_all()
        .to_pylist()
    )
    for test_case in type_tests:
        query_set = get_quirks(
            test_case["vendor_version"], vendor=test_case["vendor"]
        ).query_set
        arrow_type_names = set()
        for query_name in test_case["query_names"]:
            arrow_type_names.add(query_set.queries[query_name].arrow_type_name)
        sql_type = html.escape(test_case["sql_type"])
        arrow_type = html.escape(test_case["arrow_type_name"])
        report.add_table_entry(
            test_case["vendor"],
            test_case["vendor_version"],
            "bind",
            arrow_type,
            sql_type,
            test_case,
        )

    # Ingest type support
    type_tests = (
        duckdb.sql("""
        FROM testcases
        SELECT
          vendor,
          vendor_version,
          arrow_type_name,
          ARRAY_AGG(metadata->>'tags'->>'sql-type-name' ORDER BY query_name ASC) AS sql_types,
          ARRAY_AGG(test_result ORDER BY query_name ASC) AS test_results,
          ARRAY_AGG(query_name ORDER BY query_name ASC) as query_names,
          ARRAY_AGG(metadata ORDER BY query_name ASC) as metadata,
        WHERE
          test_module LIKE '%TestIngest'
          AND test_name = 'test_create'
          AND (metadata->>'tags'->>'sql-type-name') IS NOT NULL
        GROUP BY vendor, vendor_version, arrow_type_name, metadata->>'tags'->>'variant'
        ORDER BY vendor, vendor_version, arrow_type_name, metadata->>'tags'->>'variant'
        """)
        .arrow()
        .read_all()
        .to_pylist()
    )
    for test_case in type_tests:
        query_set = get_quirks(
            test_case["vendor_version"], vendor=test_case["vendor"]
        ).query_set
        arrow_type_names = set()
        for query_name in test_case["query_names"]:
            arrow_type_names.add(query_set.queries[query_name].arrow_type_name)
        arrow_type = html.escape(test_case["arrow_type_name"])
        sql_types = {html.escape(sql_type) for sql_type in test_case["sql_types"]}
        sql_type = ", ".join(sorted(sql_types))

        report.add_table_entry(
            test_case["vendor"],
            test_case["vendor_version"],
            "ingest",
            arrow_type,
            sql_type,
            test_case,
            variant=query_set.queries[query_name].metadata().tags.variant,
        )

    # GetObjects
    get_objects = (
        duckdb.sql("""
    WITH get_objects_cases AS (
      FROM testcases
      SELECT
        vendor,
        vendor_version,
        regexp_extract(test_name, 'test_get_objects_([a-z]+)', 1) AS test_name,
        test_result,
      WHERE test_name LIKE 'test_get_objects_%'
    )
    FROM get_objects_cases
    SELECT vendor, vendor_version, test_name, BOOL_AND(test_result = 'passed') AS supported
    GROUP BY vendor, vendor_version, test_name
    """)
        .arrow()
        .read_all()
        .to_pylist()
    )
    for test_case in get_objects:
        report.get_version(test_case).get_objects[test_case["test_name"]] = test_case[
            "supported"
        ]

    # Get table schema
    get_table_schema = (
        duckdb.sql("""
    FROM testcases
    SELECT vendor, vendor_version, CAST(COUNTIF(test_result = 'passed') AS BIGINT) AS supported_cases, COUNT() AS total_cases
    WHERE test_name = 'test_get_table_schema' AND test_result != 'skipped'
    GROUP BY vendor, vendor_version, test_name
    """)
        .arrow()
        .read_all()
        .to_pylist()
    )
    for test_case in get_table_schema:
        version = report.get_version(test_case)
        supported = test_case["supported_cases"] == test_case["total_cases"]
        version.get_table_schema = supported

    # Ingest modes
    ingest_types = (
        duckdb.sql("""
    WITH ingest_cases AS (
      FROM testcases
      SELECT
        vendor,
        vendor_version,
        test_name,
        test_result,
      WHERE test_module LIKE '%TestIngest'
    )
    FROM ingest_cases
    SELECT vendor, vendor_version, test_name, BOOL_OR(test_result = 'passed') AS supported
    GROUP BY vendor, vendor_version, test_name
    """)
        .arrow()
        .read_all()
        .to_pylist()
    )
    for test_case in ingest_types:
        ingest = report.get_version(test_case).ingest
        name = test_case["test_name"][5:]  # Strip 'test_' prefix
        ingest[name] = test_case["supported"]

    # Custom features
    custom_features = (
        duckdb.sql("""
        FROM testcases
        SELECT
          vendor,
          vendor_version,
          properties->>'feature:group' AS feature_group,
          properties->>'feature:name' AS feature_name,
          ANY_VALUE(properties->>'doc') AS description,
          ANY_VALUE(test_result) = 'passed' AS supported
        WHERE
          (properties->>'feature:group' IS NOT NULL) AND
          (properties->>'feature:name' IS NOT NULL)
        GROUP BY
          vendor,
          vendor_version,
          properties->>'feature:group',
          properties->>'feature:name'
        ORDER BY
          vendor,
          vendor_version,
          properties->>'feature:group',
          properties->>'feature:name'
        """)
        .arrow()
        .read_all()
        .to_pylist()
    )
    for test_case in custom_features:
        custom_features = report.get_version(test_case).custom_features
        custom_features.groups[test_case["feature_group"]].append(
            CustomFeature(
                name=test_case["feature_name"],
                description=test_case["description"],
                supported=test_case["supported"],
            )
        )

    return report


def generate(
    driver: str,
    get_quirks: GetQuirks,
    vendor_mapping: list[tuple[str, str]],
    test_results: list[Path],
    driver_template: Path,
    output: Path,
) -> None:
    """Generate documentation for a driver based on test results.

    Parameters
    ----------
    get_quirks
      A function that takes a vendor version and returns the corresponding
      driver quirks.

    version_mapping
      A mapping from vendor strings ("mysql") to display names
      ("MySQL"). Additionally, the order of keys will be used to determine the
      order of display (so "MySQL" can be sorted before "MariaDB").

    test_results
      A list of paths to JUnit XML files.

    driver_template
      The Markdown template to use.

    output
      Where to write the result.

    """
    for results in test_results:
        load_testcases(get_quirks, results)
    report = generate_includes(driver, get_quirks)
    print(report.pprint())
    render(report, vendor_mapping, driver_template, output)