From 2c8b0f6f5adcc61c5d5dd57eb9aaeb41648a1c34 Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Tue, 29 Sep 2026 14:45:22 +0530 Subject: [PATCH 1/2] PERF: Reduce initial buffer allocation for SQLColumns fetchall Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- mssql_python/pybind/ddbc_bindings.cpp | 80 +++++++++++++++---------- mssql_python/pybind/result_metadata.hpp | 7 ++- tests/test_004_cursor.py | 68 +++++++++++++++++++++ 3 files changed, 122 insertions(+), 33 deletions(-) diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index f410631d4..b774fdc1c 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -1886,7 +1886,9 @@ SQLRETURN SQLColumns_wrap(SqlHandlePtr StatementHandle, const py::object& catalo const py::object& schemaObj, const py::object& tableObj, const py::object& columnObj) { PERF_TIMER("SQLColumns_wrap"); - StatementHandle->resultMetadata.clear(); + StatementHandle->resultMetadata.clear(true); + SQLRETURN ret = SQL_ERROR; + ResultMetadataFailureGuard metadataFailure(StatementHandle->resultMetadata, ret); if (!SQLColumns_ptr) { ThrowStdException("SQLColumns function not loaded"); } @@ -1897,16 +1899,19 @@ SQLRETURN SQLColumns_wrap(SqlHandlePtr StatementHandle, const py::object& catalo std::u16string column = columnObj.is_none() ? u"" : columnObj.cast(); // Release the GIL during the blocking ODBC catalog call - py::gil_scoped_release release; - return SQLColumns_ptr(StatementHandle->get(), - catalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(catalog), - catalog.empty() ? 0 : SQL_NTS, - schema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(schema), - schema.empty() ? 0 : SQL_NTS, - table.empty() ? nullptr : reinterpretU16stringAsSqlWChar(table), - table.empty() ? 0 : SQL_NTS, - column.empty() ? nullptr : reinterpretU16stringAsSqlWChar(column), - column.empty() ? 0 : SQL_NTS); + { + py::gil_scoped_release release; + ret = SQLColumns_ptr(StatementHandle->get(), + catalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(catalog), + catalog.empty() ? 0 : SQL_NTS, + schema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(schema), + schema.empty() ? 0 : SQL_NTS, + table.empty() ? nullptr : reinterpretU16stringAsSqlWChar(table), + table.empty() ? 0 : SQL_NTS, + column.empty() ? nullptr : reinterpretU16stringAsSqlWChar(column), + column.empty() ? 0 : SQL_NTS); + } + return ret; } // Helper function to check for driver errors @@ -6232,32 +6237,45 @@ SQLRETURN FetchAll_wrap(SqlHandlePtr StatementHandle, py::list& rows, } else { fetchSize = 1000; } - LOG("FetchAll_wrap: Fetching data in batch sizes of %d", fetchSize); - - ColumnBuffers buffers(numCols, fetchSize); - SQLULEN numRowsFetched = 0; - FetchStateGuard fetchStateGuard(StatementHandle, messages); - - // Bind columns - ret = SQLBindColums(hStmt, buffers, columnNames, numCols, fetchSize, charCtype, messages); - if (!SQL_SUCCEEDED(ret)) { - LOG("FetchAll_wrap: Error when binding columns - SQLRETURN=%d", ret); - return ret; + const int maxFetchSize = fetchSize; + // SQLColumns declares wide fields even for small catalogs. Start at the + // existing 10-row tier, then grow through the same tiers as batches fill. + if (metadataSnapshot.catalogResult) { + fetchSize = std::min(fetchSize, 10); } - fetchStateGuard.configure(&numRowsFetched, fetchSize); - while (ret != SQL_NO_DATA) { - ret = FetchBatchData(hStmt, buffers, columnNames, rows, numCols, numRowsFetched, lobColumns, - charEncoding, charCtype, messages); - CheckFetchError(StatementHandle, ret); - if (!SQL_SUCCEEDED(ret) && ret != SQL_NO_DATA) { - LOG("FetchAll_wrap: Error when fetching data - SQLRETURN=%d", ret); + LOG("FetchAll_wrap: Fetching data in batch sizes of %d", fetchSize); + ColumnBuffers buffers(numCols, fetchSize); + SQLULEN numRowsFetched = 0; + FetchStateGuard fetchStateGuard(StatementHandle, messages); + + ret = SQLBindColums(hStmt, buffers, columnNames, numCols, fetchSize, charCtype, messages); + if (!SQL_SUCCEEDED(ret)) { + LOG("FetchAll_wrap: Error when binding columns - SQLRETURN=%d", ret); return ret; } - } - fetchStateGuard.close(); + fetchStateGuard.configure(&numRowsFetched, fetchSize); + + while (ret != SQL_NO_DATA) { + ret = FetchBatchData(hStmt, buffers, columnNames, rows, numCols, numRowsFetched, + lobColumns, charEncoding, charCtype, messages); + CheckFetchError(StatementHandle, ret); + if (!SQL_SUCCEEDED(ret) && ret != SQL_NO_DATA) { + LOG("FetchAll_wrap: Error when fetching data - SQLRETURN=%d", ret); + return ret; + } + if (SQL_SUCCEEDED(ret) && numRowsFetched == static_cast(fetchSize) && + fetchSize < maxFetchSize) { + break; + } + } + + // Unbind while these buffers are still alive, before allocating the next tier. + fetchStateGuard.close(); + fetchSize = std::min(fetchSize * 10, maxFetchSize); + } return ret; } diff --git a/mssql_python/pybind/result_metadata.hpp b/mssql_python/pybind/result_metadata.hpp index 3a9912860..3238cc4c3 100644 --- a/mssql_python/pybind/result_metadata.hpp +++ b/mssql_python/pybind/result_metadata.hpp @@ -30,11 +30,12 @@ class ResultMetadataCache { struct Snapshot { uint64_t generation; std::shared_ptr metadata; + bool catalogResult; // SQLColumns allocation hint, reset with the result generation. }; Snapshot snapshot() const { std::lock_guard lock(mutex_); - return {generation_, metadata_}; + return {generation_, metadata_, catalogResult_}; } void publish(uint64_t generation, std::shared_ptr metadata) { @@ -44,16 +45,18 @@ class ResultMetadataCache { } } - void clear() { + void clear(bool catalogResult = false) { std::lock_guard lock(mutex_); ++generation_; metadata_.reset(); + catalogResult_ = catalogResult; } private: mutable std::mutex mutex_; uint64_t generation_ = 0; std::shared_ptr metadata_; + bool catalogResult_ = false; }; class ResultMetadataFailureGuard { diff --git a/tests/test_004_cursor.py b/tests/test_004_cursor.py index f2050376f..95a49aaa0 100644 --- a/tests/test_004_cursor.py +++ b/tests/test_004_cursor.py @@ -13729,6 +13729,10 @@ def test_columns_specific_table(cursor, db_connection): # Verify we got results assert len(cols) == 9, "Should find exactly 9 columns in columns_test" + assert [col.ordinal_position for col in cols] == list(range(1, 10)) + assert cursor.rowcount == 9 + assert cursor.rownumber == 8 + assert cursor.fetchone() is None # Verify all column names are present (case insensitive) col_names = [col.column_name.lower() for col in cols] @@ -14055,7 +14059,24 @@ def test_columns_table_pattern(cursor): """Test columns with table name pattern""" try: # Get columns with table pattern + arraysize = cursor.arraysize cols = cursor.columns(table="columns_%", schema="pytest_cols_schema").fetchall() + description = cursor.description + assert len(cols) == 18 + assert cursor.rowcount == 18 + assert cursor.rownumber == 17 + assert cursor.fetchone() is None + assert cursor.arraysize == arraysize + + # Compare all ordered cells and types with the independent row-wise fetch path. + cursor.columns(table="columns_%", schema="pytest_cols_schema") + assert type(cursor.description) is type(description) + assert cursor.description == description + expected = [tuple(row) for row in cursor] + assert [tuple(row) for row in cols] == expected + assert [[type(value) for value in row] for row in cols] == [ + [type(value) for value in row] for row in expected + ] # Should find columns from both test tables tables_found = set() @@ -14068,6 +14089,33 @@ def test_columns_table_pattern(cursor): "columns_special_test" in tables_found ), "Should find columns_special_test with pattern columns_%" + # Cross both catalog batch-growth boundaries without creating another table. + extra_columns = [f"fetch_growth_{i:03d}" for i in range(100)] + cursor.execute( + "ALTER TABLE pytest_cols_schema.columns_special_test ADD " + + ", ".join(f"[{name}] INT NULL" for name in extra_columns) + ) + try: + expanded = cursor.columns(table="columns_%", schema="pytest_cols_schema").fetchall() + assert cursor.description == description + assert len(expanded) == 118 + assert cursor.rowcount == 118 + assert cursor.rownumber == 117 + assert cursor.fetchone() is None + assert cursor.arraysize == arraysize + + cursor.columns(table="columns_%", schema="pytest_cols_schema") + expected = [tuple(row) for row in cursor] + assert [tuple(row) for row in expanded] == expected + assert [[type(value) for value in row] for row in expanded] == [ + [type(value) for value in row] for row in expected + ] + finally: + cursor.execute( + "ALTER TABLE pytest_cols_schema.columns_special_test DROP COLUMN " + + ", ".join(f"[{name}]" for name in extra_columns) + ) + finally: # Clean up happens in test_columns_cleanup pass @@ -17649,6 +17697,26 @@ def test_columns_fetchone(cursor, db_connection, catalog_fetch_schema): assert row is not None, "fetchone() should return a row from columns()" assert hasattr(row, "column_name") assert row.table_name.lower() == "fetch_test" + rows = [row] + cursor.fetchmany(1) + cursor.fetchall() + assert [item.column_name for item in rows] == ["id", "name", "value", "ts"] + assert [item.ordinal_position for item in rows] == [1, 2, 3, 4] + assert cursor.rowcount == 4 + assert cursor.rownumber == 3 + assert cursor.fetchone() is None + statement = cursor.hstmt + operation = "SELECT ? AS ordinary; SELECT ? AS next_result" + cursor.execute(operation, (1, 2)) + assert cursor.hstmt is statement + assert cursor.fetchall()[0].ordinary == 1 + assert cursor.nextset() is True + assert cursor.fetchall()[0].next_result == 2 + assert cursor.nextset() is False + cursor.execute(operation, (3, 4), reset_cursor=False) + assert cursor.hstmt is statement + assert cursor.fetchall()[0].ordinary == 3 + assert cursor.nextset() is True + assert cursor.fetchall()[0].next_result == 4 + assert cursor.nextset() is False def test_primarykeys_fetchone(cursor, db_connection, catalog_fetch_schema): From a28f9db65112c558206422bcc198e2cc30ace580 Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Tue, 29 Sep 2026 17:18:47 +0530 Subject: [PATCH 2/2] PERF: Cover SQLColumns in PR performance benchmarks Add catalog workloads for 2, 118, and 2111 rows with untimed fixture setup, raw metadata and typed row validation, and rollback cleanup. Wire the workloads into the existing advisory report and cover their contracts with driver-free tests. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- eng/profiler_benchmarks/README.md | 22 ++- eng/profiler_benchmarks/report.py | 3 + eng/profiler_benchmarks/workloads.py | 77 +++++++++ tests/test_036_profiler_ci.py | 225 ++++++++++++++++++++++++++- 4 files changed, 325 insertions(+), 2 deletions(-) diff --git a/eng/profiler_benchmarks/README.md b/eng/profiler_benchmarks/README.md index ef2365adf..df8bfeff2 100644 --- a/eng/profiler_benchmarks/README.md +++ b/eng/profiler_benchmarks/README.md @@ -12,13 +12,33 @@ python -m eng.profiler_benchmarks.controller --base main --candidate HEAD \ python -m eng.profiler_benchmarks.report profiler-results/report.json ``` -The fixed registry has 21 tasks. `--scenarios` runs a local subset, but subset +The fixed registry has 24 tasks. `--scenarios` runs a local subset, but subset reports remain incomplete and cannot produce a verdict. `lob_varchar_256k_fetchall` fetches one 256 KiB `VARCHAR(MAX)` value to exercise multi-chunk streaming. Query setup and exact payload validation are outside the timed fetch window. +`catalog_columns_2`, `catalog_columns_118`, and `catalog_columns_2111` time +`columns()` plus the complete `fetchall()` drain on a fresh cursor. Each task +creates UUID-prefixed tables in the current database's `dbo` schema, with exactly +2, 118, or 2,111 nullable `INT` columns in total (at most 704 per table). It needs +permission to create tables and uses the profiler-owned connection with autocommit +off, not a caller's connection with pending writes. Its DDL is uncommitted and +rolled back on success or failure. Fixture setup, cursor creation, EOF checks, and +comparison of all 29 provider fields, raw descriptions, ordered cells and Python +types with an independent rowwise drain are outside the measurement window. +The rowwise oracle runs after timing, +so it does not prime the measured catalog allocation. Native call counters verify +one `SQLColumns` and one `FetchAll` call in the measured window. + +These tasks cover small results, growth through both initial tiers, and reuse of +the maximum tier. They measure instrumented latency, not allocation bytes, and +are not the same fixtures as the standalone catalog experiments. Ordinary SELECT +fetch-all remains a separate control. The unchanged advisory thresholds below can +leave smaller catalog slowdowns labeled "no signal"; that is not proof of no +regression. + ## Measurement contract CI uses the PR merge's first parent as the exact base. It reuses the diff --git a/eng/profiler_benchmarks/report.py b/eng/profiler_benchmarks/report.py index 6ac2723f2..e2a9fff38 100644 --- a/eng/profiler_benchmarks/report.py +++ b/eng/profiler_benchmarks/report.py @@ -38,6 +38,9 @@ "fetch_1_2m": "1.2-million-row fetching", "cte": "Common table expression queries", "lob_varchar_256k_fetchall": "256 KiB VARCHAR(MAX) / fetchall()", + "catalog_columns_2": "Catalog columns / 2 rows", + "catalog_columns_118": "Catalog columns / 118 rows", + "catalog_columns_2111": "Catalog columns / 2,111 rows", } CASES = tuple(TASK_NAMES) MAX_BYTES = 8 * 1024 * 1024 diff --git a/eng/profiler_benchmarks/workloads.py b/eng/profiler_benchmarks/workloads.py index c183f5d89..fa5bc52c7 100644 --- a/eng/profiler_benchmarks/workloads.py +++ b/eng/profiler_benchmarks/workloads.py @@ -2,6 +2,7 @@ from functools import partial import time +from uuid import uuid4 from profiler import scenarios @@ -163,6 +164,78 @@ def lob_fetch(conn, ctx): ctx.disable() +def catalog_columns(conn, ctx, row_count): + """Use the profiler-owned transaction; time only SQLColumns and its complete drain.""" + if conn.autocommit: + raise ValueError("Catalog benchmarks require a transactional connection") + prefix = "profcat" + uuid4().hex + tables = [ + (f"{prefix}{index:02d}", min(704, row_count - start)) + for index, start in enumerate(range(0, row_count, 704)) + ] + expected = [ + (table, f"c{column:04d}", column + 1) for table, count in tables for column in range(count) + ] + try: + with conn.cursor() as setup: + catalog = setup.execute("SELECT DB_NAME()").fetchone()[0] + for table, count in tables: + definitions = ", ".join(f"[c{column:04d}] INT NULL" for column in range(count)) + setup.execute(f"CREATE TABLE [dbo].[{table}] ({definitions})") + filters = dict(catalog=catalog, schema="dbo", table=prefix + "%", column=None) + with conn.cursor() as cursor: + try: + ctx.enable() + start = time.perf_counter() + cursor.columns(**filters) + rows = cursor.fetchall() + wall_ms = (time.perf_counter() - start) * 1000 + cpp, py = ctx.collect() + finally: + ctx.disable() + description = cursor.description + assert len(rows) == row_count + assert len(description) == 29 and all(len(row) == 29 for row in rows) + assert [ + (row.table_name, row.column_name, row.ordinal_position) for row in rows + ] == expected + assert all( + row.table_cat == catalog + and row.table_schem == "dbo" + and type(row.data_type) is int + and row.data_type == 4 # SQL_INTEGER + and row.column_size == 10 + and row.nullable == 1 + and row.column_def is None + for row in rows + ) + assert cursor.fetchone() is None + assert not cursor.messages, "Clean catalog fetch unexpectedly produced diagnostics" + # A separate rowwise oracle runs only after the measured first allocation. + with conn.cursor() as reference: + reference.columns(**filters) + assert type(reference.description) is type(description) + assert reference.description == description + expected_rows = [tuple(row) for row in reference] + assert [tuple(row) for row in rows] == expected_rows + assert [[type(value) for value in row] for row in rows] == [ + [type(value) for value in row] for row in expected_rows + ] + assert not reference.messages, "Rowwise catalog oracle produced diagnostics" + assert cpp["ddbc::SQLColumns_wrap"]["calls"] == 1 + assert cpp["ddbc::FetchAll_wrap"]["calls"] == 1 + return dict( + title="SQLColumns metadata", + wall_ms=wall_ms, + cpp=cpp, + py=py, + detail=f"Rows: {row_count}; tables: {len(tables)}; type: INT NULL; API: columns+fetchall", + ) + finally: + # Fixture DDL is never committed; rollback also removes partially created fixtures. + conn.rollback() + + def registry(): """Keep every PR #552 scenario, including its existing timing boundaries.""" result = dict(scenarios.SCENARIOS) @@ -176,4 +249,8 @@ def registry(): ) result.update((name, (partial(query, sql=sql), False)) for name, sql in QUERIES.items()) result["lob_varchar_256k_fetchall"] = (lob_fetch, False) + result.update( + (f"catalog_columns_{count}", (partial(catalog_columns, row_count=count), False)) + for count in (2, 118, 2111) + ) return result diff --git a/tests/test_036_profiler_ci.py b/tests/test_036_profiler_ci.py index eced790d5..1f23c3bd7 100644 --- a/tests/test_036_profiler_ci.py +++ b/tests/test_036_profiler_ci.py @@ -1,6 +1,7 @@ """Contract tests for paired performance comparisons and data-only PR reporting.""" import copy +from collections import namedtuple from http.client import IncompleteRead import importlib.util import io @@ -687,10 +688,232 @@ def __exit__(self, *args): def test_report_cases_match_the_executed_workload_registry(): _, workloads = controller.load_suite() assert tuple(workloads.registry()) == reporting.CASES - assert len(reporting.CASES) == 21 + assert len(reporting.CASES) == 24 assert [name for name in reporting.CASES if name.startswith("lob_")] == [ "lob_varchar_256k_fetchall", ] + assert [name for name in reporting.CASES if name.startswith("catalog_")] == [ + "catalog_columns_2", + "catalog_columns_118", + "catalog_columns_2111", + ] + + +@pytest.fixture +def catalog_workload(monkeypatch): + def build(row_count=118): + prefix = "profcat" + "a" * 32 + row_type = namedtuple( + "CatalogRow", + "table_cat table_schem table_name column_name data_type type_name column_size " + "buffer_length decimal_digits num_prec_radix nullable remarks column_def " + "sql_data_type sql_datetime_sub char_octet_length ordinal_position is_nullable " + + " ".join(f"provider_{i}" for i in range(11)), + ) + rows = [ + row_type( + "catalog_db", + "dbo", + f"{prefix}{index // 704:02d}", + f"c{index % 704:04d}", + 4, + "int", + 10, + 4, + 0, + 10, + 1, + None, + None, + 4, + None, + None, + index % 704 + 1, + "YES", + *([None] * 11), + ) + for index in range(row_count) + ] + setup, cursor, reference = MagicMock(), MagicMock(), MagicMock() + for item in (setup, cursor, reference): + item.__enter__.return_value = item + setup.execute.return_value.fetchone.return_value = ("catalog_db",) + cursor.fetchall.return_value = rows + cursor.fetchone.return_value = None + cursor.description = [ + (name, str, None, None, None, None, True) for name in row_type._fields + ] + cursor.messages = reference.messages = [] + reference.description = copy.deepcopy(cursor.description) + reference.__iter__.side_effect = lambda: iter(rows) + connection = MagicMock(autocommit=False) + connection.cursor.side_effect = [setup, cursor, reference] + context = MagicMock() + counters = { + name: dict(calls=1, total_us=10, min_us=10, max_us=10) + for name in ("ddbc::SQLColumns_wrap", "ddbc::FetchAll_wrap") + } + context.collect.return_value = (counters, {}) + clock = MagicMock(side_effect=[1, 1.1]) + events = MagicMock() + for name, mock in ( + ("cursor", connection.cursor), + ("setup", setup.execute), + ("enable", context.enable), + ("clock", clock), + ("columns", cursor.columns), + ("fetchall", cursor.fetchall), + ("collect", context.collect), + ("disable", context.disable), + ("eof", cursor.fetchone), + ("reference", reference.columns), + ("rollback", connection.rollback), + ): + events.attach_mock(mock, name) + monkeypatch.setattr(benchmark_workloads, "uuid4", lambda: SimpleNamespace(hex="a" * 32)) + monkeypatch.setattr(benchmark_workloads.time, "perf_counter", clock) + return SimpleNamespace( + connection=connection, + context=context, + setup=setup, + cursor=cursor, + reference=reference, + rows=rows, + prefix=prefix, + events=events, + ) + + return build + + +@pytest.mark.parametrize("row_count", (2, 118, 2111)) +def test_catalog_workload_measures_columns_and_drain_only(catalog_workload, row_count): + case = catalog_workload(row_count) + workload, needs_table = benchmark_workloads.registry()[f"catalog_columns_{row_count}"] + assert needs_table is False + result = workload(case.connection, case.context) + statements = [call.args[0] for call in case.setup.execute.call_args_list] + assert statements[0] == "SELECT DB_NAME()" + table_count = (row_count + 703) // 704 + assert len(statements) == table_count + 1 + assert [len(re.findall(r"\[c\d{4}\] INT NULL", sql)) for sql in statements[1:]] == [ + min(704, row_count - start) for start in range(0, row_count, 704) + ] + assert all( + sql.startswith(f"CREATE TABLE [dbo].[{case.prefix}{i:02d}] (") + for i, sql in enumerate(statements[1:]) + ) + case.cursor.columns.assert_called_once_with( + catalog="catalog_db", schema="dbo", table=case.prefix + "%", column=None + ) + assert case.reference.columns.call_args == case.cursor.columns.call_args + case.cursor.fetchall.assert_called_once_with() + case.cursor.fetchone.assert_called_once_with() + case.cursor.fetchmany.assert_not_called() + case.reference.fetchall.assert_not_called() + case.connection.commit.assert_not_called() + case.connection.rollback.assert_called_once_with() + names = [call[0] for call in case.events.mock_calls] + clocks = [index for index, name in enumerate(names) if name == "clock"] + assert names[clocks[0] + 1 : clocks[1]] == ["columns", "fetchall"] + assert names.index("enable") > max(i for i, name in enumerate(names) if name == "setup") + assert names[clocks[1] + 1 :] == [ + "collect", + "disable", + "eof", + "cursor", + "reference", + "rollback", + ] + assert result["wall_ms"] == pytest.approx(100) + assert result["detail"] == ( + f"Rows: {row_count}; tables: {table_count}; type: INT NULL; API: columns+fetchall" + ) + + +@pytest.mark.parametrize( + "problem", + ( + "missing", + "order", + "type", + "schema", + "width", + "facts", + "eof", + "warning", + "reference-warning", + "columns-error", + "fetch-error", + "counter", + ), +) +def test_catalog_workload_rejects_invalid_results_and_rolls_back(catalog_workload, problem): + case = catalog_workload() + if problem == "missing": + case.cursor.fetchall.return_value = case.rows[:-1] + elif problem == "order": + case.cursor.fetchall.return_value = list(reversed(case.rows)) + elif problem == "type": + case.reference.__iter__.side_effect = lambda: iter( + [case.rows[0]._replace(buffer_length=4.0), *case.rows[1:]] + ) + elif problem == "schema": + case.reference.description[0] = ("changed", *case.reference.description[0][1:]) + elif problem == "width": + case.cursor.description = case.cursor.description[:-1] + elif problem == "facts": + case.cursor.fetchall.return_value = [row._replace(nullable=0) for row in case.rows] + elif problem == "eof": + case.cursor.fetchone.return_value = case.rows[0] + elif problem == "warning": + case.cursor.messages = [("01000", "unexpected")] + elif problem == "reference-warning": + case.reference.messages = [("01000", "unexpected")] + elif problem in ("columns-error", "fetch-error"): + method = case.cursor.columns if problem == "columns-error" else case.cursor.fetchall + method.side_effect = RuntimeError("catalog failure") + else: + case.context.collect.return_value[0]["ddbc::SQLColumns_wrap"]["calls"] = 2 + with pytest.raises(RuntimeError if problem.endswith("-error") else AssertionError): + benchmark_workloads.catalog_columns(case.connection, case.context, 118) + case.context.disable.assert_called_once() + case.connection.rollback.assert_called_once() + + +def test_catalog_workload_rolls_back_partial_setup(catalog_workload): + case = catalog_workload(2111) + case.setup.execute.side_effect = [ + SimpleNamespace(fetchone=lambda: ("catalog_db",)), + None, + RuntimeError("DDL failed"), + ] + with pytest.raises(RuntimeError, match="DDL failed"): + benchmark_workloads.catalog_columns(case.connection, case.context, 2111) + assert case.setup.execute.call_count == 3 + case.context.enable.assert_not_called() + case.cursor.columns.assert_not_called() + case.connection.rollback.assert_called_once() + + +def test_catalog_workload_rejects_autocommit(catalog_workload): + case = catalog_workload() + case.connection.autocommit = True + with pytest.raises(ValueError, match="transactional"): + benchmark_workloads.catalog_columns(case.connection, case.context, 118) + case.connection.cursor.assert_not_called() + + +def test_catalog_tasks_appear_in_pr_report(report): + reporting.validate(report) + body = reporting.render([report], "c" * 40, 42) + for count in (2, 118, 2111): + name = f"catalog_columns_{count}" + assert f"| {reporting.TASK_NAMES[name]} |" in body + incomplete = copy.deepcopy(report) + del incomplete["pairs"][0]["candidate"]["scenarios"][name] + with pytest.raises(ValueError, match="Scenario set incomplete"): + reporting.validate(incomplete) def test_lob_workload_validates_payload_and_times_only_fetch(monkeypatch):