diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py
index 85701a408..6806f1742 100644
--- a/mssql_python/cursor.py
+++ b/mssql_python/cursor.py
@@ -29,6 +29,7 @@
     DatabaseError,
 )
 from mssql_python.row import Row
+from mssql_python.perf_timer import perf_phase, perf_start, perf_stop
 from mssql_python import get_settings
 from mssql_python.parameter_helper import (
     detect_and_convert_parameters,
@@ -1455,34 +1456,37 @@ def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-state
         # it will be unwrapped for parameter binding. This means you cannot
         # pass a tuple as a single parameter value (but SQL Server doesn't
         # support tuple types as parameter values anyway).
-        if parameters:
-            # Check if single parameter is a nested container that should be unwrapped
-            # e.g., execute("SELECT ?", (value,)) vs execute("SELECT ?, ?", ((1, 2),))
-            if isinstance(parameters, tuple) and len(parameters) == 1:
-                if isinstance(parameters[0], (tuple, list, dict)):
-                    actual_params = parameters[0]
-                elif isinstance(parameters[0], Row):
-                    # A Row (e.g. from fetchone()) is a sequence of column values.
-                    # Normalize it to a tuple so the downstream binding logic, which
-                    # only handles tuple/list/dict, can unwrap it into individual
-                    # parameters instead of treating the whole Row as one value.
-                    actual_params = tuple(parameters[0])
+        with perf_phase("py::execute::param_unpack"):
+            if parameters:
+                # Check if single parameter is a nested container that should be unwrapped
+                # e.g., execute("SELECT ?", (value,)) vs execute("SELECT ?, ?", ((1, 2),))
+                if isinstance(parameters, tuple) and len(parameters) == 1:
+                    if isinstance(parameters[0], (tuple, list, dict)):
+                        actual_params = parameters[0]
+                    elif isinstance(parameters[0], Row):
+                        # A Row (e.g. from fetchone()) is a sequence of column values.
+                        # Normalize it to a tuple so the downstream binding logic, which
+                        # only handles tuple/list/dict, can unwrap it into individual
+                        # parameters instead of treating the whole Row as one value.
+                        actual_params = tuple(parameters[0])
+                    else:
+                        actual_params = parameters
                 else:
                     actual_params = parameters
-            else:
-                actual_params = parameters
 
-            # Skip detect_and_convert_parameters when re-executing the same SQL —
-            # the parameter style (qmark vs pyformat) won't change between calls.
-            if operation == self.last_executed_stmt and isinstance(actual_params, (tuple, list)):
-                parameters = list(actual_params)
+                # Skip detect_and_convert_parameters when re-executing the same SQL —
+                # the parameter style (qmark vs pyformat) won't change between calls.
+                if operation == self.last_executed_stmt and isinstance(
+                    actual_params, (tuple, list)
+                ):
+                    parameters = list(actual_params)
+                else:
+                    operation, converted_params = detect_and_convert_parameters(
+                        operation, actual_params
+                    )
+                    parameters = list(converted_params)
             else:
-                operation, converted_params = detect_and_convert_parameters(
-                    operation, actual_params
-                )
-                parameters = list(converted_params)
-        else:
-            parameters = []
+                parameters = []
 
         # Getting encoding setting
         encoding_settings = self._get_encoding_settings()
@@ -1503,10 +1507,11 @@ def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-state
                     Warning,
                 )
 
-        if parameters:
-            for i, param in enumerate(parameters):
-                paraminfo = self._create_parameter_types_list(param, param_info, parameters, i)
-                parameters_type.append(paraminfo)
+        with perf_phase("py::execute::param_type_detection"):
+            if parameters:
+                for i, param in enumerate(parameters):
+                    paraminfo = self._create_parameter_types_list(param, param_info, parameters, i)
+                    parameters_type.append(paraminfo)
 
         # Prepare caching: skip SQLPrepare when re-executing the same SQL
         # with parameters. The HSTMT is reused via _soft_reset_cursor, so the
@@ -1531,15 +1536,16 @@ def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-state
                     parameters_type[i].inputOutputType,
                 )
 
-        ret = ddbc_bindings.DDBCSQLExecute(
-            self.hstmt,
-            operation,
-            parameters,
-            parameters_type,
-            self.is_stmt_prepared,
-            effective_use_prepare,
-            encoding_settings,
-        )
+        with perf_phase("py::execute::cpp_call"):
+            ret = ddbc_bindings.DDBCSQLExecute(
+                self.hstmt,
+                operation,
+                parameters,
+                parameters_type,
+                self.is_stmt_prepared,
+                effective_use_prepare,
+                encoding_settings,
+            )
         # Check return code
         try:
 
@@ -1550,23 +1556,26 @@ def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-state
             self._reset_cursor()
             raise
 
-        self._capture_diagnostics(ret)
+        # Capture any diagnostic messages (SQL_SUCCESS_WITH_INFO, etc.)
+        with perf_phase("py::execute::diag_records"):
+            self._capture_diagnostics(ret)
 
         self.last_executed_stmt = operation
 
-        # Update rowcount after execution
-        # TODO: rowcount return code from SQL needs to be handled
-        self.rowcount = ddbc_bindings.DDBCSQLRowCount(self.hstmt)
+        with perf_phase("py::execute::post_execute"):
+            # Update rowcount after execution
+            # TODO: rowcount return code from SQL needs to be handled
+            self.rowcount = ddbc_bindings.DDBCSQLRowCount(self.hstmt)
 
-        # Initialize description after execution
-        # After successful execution, initialize description if there are results
-        column_metadata = []
-        try:
-            ddbc_bindings.DDBCSQLDescribeCol(self.hstmt, column_metadata)
-            self._initialize_description(column_metadata)
-        except Exception as e:  # pylint: disable=broad-exception-caught
-            # If describe fails, it's likely there are no results (e.g., for INSERT)
-            self.description = None
+            # Initialize description after execution
+            # After successful execution, initialize description if there are results
+            column_metadata = []
+            try:
+                ddbc_bindings.DDBCSQLDescribeCol(self.hstmt, column_metadata)
+                self._initialize_description(column_metadata)
+            except Exception as e:  # pylint: disable=broad-exception-caught
+                # If describe fails, it's likely there are no results (e.g., for INSERT)
+                self.description = None
 
         # Reset rownumber for new result set (only for SELECT statements)
         if self.description:  # If we have column descriptions, it's likely a SELECT
@@ -2210,6 +2219,7 @@ def executemany(  # pylint: disable=too-many-locals,too-many-branches,too-many-s
                 )
 
         # Prepare parameter type information
+        _t0 = perf_start()
         for col_index in range(param_count):
             column = (
                 [row[col_index] for row in seq_of_parameters]
@@ -2359,6 +2369,7 @@ def executemany(  # pylint: disable=too-many-locals,too-many-branches,too-many-s
                 parameters_type.append(paraminfo)
                 if paraminfo.isDAE:
                     any_dae = True
+        perf_stop("py::executemany::param_type_detection", _t0)
 
         if any_dae:
             logger.debug(
@@ -2402,7 +2413,10 @@ def executemany(  # pylint: disable=too-many-locals,too-many-branches,too-many-s
             processed_parameters.append(processed_row)
 
         # Now transpose the processed parameters
-        columnwise_params, row_count = self._transpose_rowwise_to_columnwise(processed_parameters)
+        with perf_phase("py::executemany::param_processing"):
+            columnwise_params, row_count = self._transpose_rowwise_to_columnwise(
+                processed_parameters
+            )
 
         # Get encoding settings
         encoding_settings = self._get_encoding_settings()
@@ -2417,13 +2431,20 @@ def executemany(  # pylint: disable=too-many-locals,too-many-branches,too-many-s
             ),  # Limit to first 5 rows for large batches
         )
 
-        ret = ddbc_bindings.SQLExecuteMany(
-            self.hstmt, operation, columnwise_params, parameters_type, row_count, encoding_settings
-        )
+        with perf_phase("py::executemany::cpp_call"):
+            ret = ddbc_bindings.SQLExecuteMany(
+                self.hstmt,
+                operation,
+                columnwise_params,
+                parameters_type,
+                row_count,
+                encoding_settings,
+            )
 
         # Capture any diagnostic messages after execution
-        if self.hstmt:
-            self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))
+        with perf_phase("py::executemany::diag_records"):
+            if self.hstmt:
+                self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))
 
         try:
             check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
@@ -2477,16 +2498,18 @@ def fetchone(self) -> Union[None, Row]:
         # Fetch raw data
         row_data = []
         try:
-            ret = ddbc_bindings.DDBCSQLFetchOne(
-                self.hstmt,
-                row_data,
-                char_decoding.get("encoding", "utf-16le"),
-                wchar_decoding.get("encoding", "utf-16le"),
-                char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value),
-            )
+            with perf_phase("py::fetchone::cpp_call"):
+                ret = ddbc_bindings.DDBCSQLFetchOne(
+                    self.hstmt,
+                    row_data,
+                    char_decoding.get("encoding", "utf-16le"),
+                    wchar_decoding.get("encoding", "utf-16le"),
+                    char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value),
+                )
 
-            if self.hstmt:
-                self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))
+            with perf_phase("py::fetchone::diag_records"):
+                if self.hstmt:
+                    self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))
 
             if ret == ddbc_sql_const.SQL_NO_DATA.value:
                 # No more data available
@@ -2506,14 +2529,15 @@ def fetchone(self) -> Union[None, Row]:
 
             # Get column and converter maps
             column_map, converter_map, column_map_lower = self._get_column_and_converter_maps()
-            return Row(
-                row_data,
-                column_map,
-                cursor=self,
-                converter_map=converter_map,
-                uuid_str_indices=self._uuid_str_indices,
-                column_map_lower=column_map_lower,
-            )
+            with perf_phase("py::fetchone::row_wrap"):
+                return Row(
+                    row_data,
+                    column_map,
+                    cursor=self,
+                    converter_map=converter_map,
+                    uuid_str_indices=self._uuid_str_indices,
+                    column_map_lower=column_map_lower,
+                )
         except Exception as e:
             # On error, don't increment rownumber - rethrow the error
             raise e
@@ -2544,17 +2568,19 @@ def fetchmany(self, size: Optional[int] = None) -> List[Row]:
         # Fetch raw data
         rows_data = []
         try:
-            ret = ddbc_bindings.DDBCSQLFetchMany(
-                self.hstmt,
-                rows_data,
-                size,
-                char_decoding.get("encoding", "utf-16le"),
-                wchar_decoding.get("encoding", "utf-16le"),
-                char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value),
-            )
+            with perf_phase("py::fetchmany::cpp_call"):
+                ret = ddbc_bindings.DDBCSQLFetchMany(
+                    self.hstmt,
+                    rows_data,
+                    size,
+                    char_decoding.get("encoding", "utf-16le"),
+                    wchar_decoding.get("encoding", "utf-16le"),
+                    char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value),
+                )
 
-            if self.hstmt:
-                self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))
+            with perf_phase("py::fetchmany::diag_records"):
+                if self.hstmt:
+                    self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))
 
             # Update rownumber for the number of rows actually fetched
             if rows_data and self._has_result_set:
@@ -2573,17 +2599,18 @@ def fetchmany(self, size: Optional[int] = None) -> List[Row]:
 
             # Convert raw data to Row objects
             uuid_idx = self._uuid_str_indices
-            return [
-                Row(
-                    row_data,
-                    column_map,
-                    cursor=self,
-                    converter_map=converter_map,
-                    uuid_str_indices=uuid_idx,
-                    column_map_lower=column_map_lower,
-                )
-                for row_data in rows_data
-            ]
+            with perf_phase("py::fetchmany::row_wrap"):
+                return [
+                    Row(
+                        row_data,
+                        column_map,
+                        cursor=self,
+                        converter_map=converter_map,
+                        uuid_str_indices=uuid_idx,
+                        column_map_lower=column_map_lower,
+                    )
+                    for row_data in rows_data
+                ]
         except Exception as e:
             # On error, don't increment rownumber - rethrow the error
             raise e
@@ -2605,19 +2632,21 @@ def fetchall(self) -> List[Row]:
         # Fetch raw data
         rows_data = []
         try:
-            ret = ddbc_bindings.DDBCSQLFetchAll(
-                self.hstmt,
-                rows_data,
-                char_decoding.get("encoding", "utf-16le"),
-                wchar_decoding.get("encoding", "utf-16le"),
-                char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value),
-            )
+            with perf_phase("py::fetchall::cpp_call"):
+                ret = ddbc_bindings.DDBCSQLFetchAll(
+                    self.hstmt,
+                    rows_data,
+                    char_decoding.get("encoding", "utf-16le"),
+                    wchar_decoding.get("encoding", "utf-16le"),
+                    char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value),
+                )
 
             # Check for errors
             check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
 
-            if self.hstmt:
-                self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))
+            with perf_phase("py::fetchall::diag_records"):
+                if self.hstmt:
+                    self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))
 
             # Update rownumber for the number of rows actually fetched
             if rows_data and self._has_result_set:
@@ -2635,17 +2664,18 @@ def fetchall(self) -> List[Row]:
 
             # Convert raw data to Row objects
             uuid_idx = self._uuid_str_indices
-            return [
-                Row(
-                    row_data,
-                    column_map,
-                    cursor=self,
-                    converter_map=converter_map,
-                    uuid_str_indices=uuid_idx,
-                    column_map_lower=column_map_lower,
-                )
-                for row_data in rows_data
-            ]
+            with perf_phase("py::fetchall::row_wrap"):
+                return [
+                    Row(
+                        row_data,
+                        column_map,
+                        cursor=self,
+                        converter_map=converter_map,
+                        uuid_str_indices=uuid_idx,
+                        column_map_lower=column_map_lower,
+                    )
+                    for row_data in rows_data
+                ]
         except Exception as e:
             # On error, don't increment rownumber - rethrow the error
             raise e
diff --git a/mssql_python/perf_timer.py b/mssql_python/perf_timer.py
new file mode 100644
index 000000000..79eafaffc
--- /dev/null
+++ b/mssql_python/perf_timer.py
@@ -0,0 +1,134 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+"""
+Lightweight phase-level profiling for the Python layer.
+
+Usage in cursor.py:
+    from mssql_python.perf_timer import perf_phase
+
+    with perf_phase("py::execute::param_type_detection"):
+        ...
+
+Control from profiler script:
+    from mssql_python.perf_timer import enable, disable, get_stats, reset
+
+Stats dict matches the C++ profiling format so both layers can be
+printed with the same reporter. Entries use a "py::" prefix to
+distinguish from C++ timers.
+"""
+
+import time
+from contextlib import contextmanager
+
+_enabled = False
+_stats: dict[str, dict] = {}
+_timeline: list[dict] = []
+_timeline_enabled = False
+_epoch_ns: int = 0
+
+
+def enable():
+    global _enabled
+    _enabled = True
+
+
+def disable():
+    global _enabled
+    _enabled = False
+
+
+def is_enabled() -> bool:
+    return _enabled
+
+
+def reset():
+    _stats.clear()
+    _timeline.clear()
+
+
+def reset_stats_only():
+    _stats.clear()
+
+
+def enable_timeline():
+    global _timeline_enabled, _epoch_ns
+    _timeline_enabled = True
+    _epoch_ns = time.perf_counter_ns()
+
+
+def disable_timeline():
+    global _timeline_enabled
+    _timeline_enabled = False
+
+
+def get_timeline() -> list[dict]:
+    return [
+        {
+            "name": ev["name"],
+            "start_us": ev["start_ns"] // 1000,
+            "duration_us": ev["duration_ns"] // 1000,
+        }
+        for ev in _timeline
+    ]
+
+
+def get_stats() -> dict:
+    out = {}
+    for name, s in _stats.items():
+        out[name] = {
+            "calls": s["calls"],
+            "total_us": s["total_ns"] // 1000,
+            "min_us": s["min_ns"] // 1000,
+            "max_us": s["max_ns"] // 1000,
+        }
+    return out
+
+
+@contextmanager
+def perf_phase(name: str):
+    if not _enabled:
+        yield
+        return
+    t0 = time.perf_counter_ns()
+    yield
+    elapsed = time.perf_counter_ns() - t0
+    _record(name, elapsed, t0)
+
+
+def perf_start() -> int:
+    if not _enabled:
+        return 0
+    return time.perf_counter_ns()
+
+
+def perf_stop(name: str, t0: int):
+    if not _enabled:
+        return
+    _record(name, time.perf_counter_ns() - t0, t0)
+
+
+def _record(name: str, elapsed: int, start_ns: int = 0):
+    entry = _stats.get(name)
+    if entry is None:
+        _stats[name] = {
+            "calls": 1,
+            "total_ns": elapsed,
+            "min_ns": elapsed,
+            "max_ns": elapsed,
+        }
+    else:
+        entry["calls"] += 1
+        entry["total_ns"] += elapsed
+        if elapsed < entry["min_ns"]:
+            entry["min_ns"] = elapsed
+        if elapsed > entry["max_ns"]:
+            entry["max_ns"] = elapsed
+
+    if _timeline_enabled and start_ns:
+        _timeline.append(
+            {
+                "name": name,
+                "start_ns": start_ns - _epoch_ns,
+                "duration_ns": elapsed,
+            }
+        )
diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp
index 4b366575c..6886a458c 100644
--- a/mssql_python/pybind/connection/connection.cpp
+++ b/mssql_python/pybind/connection/connection.cpp
@@ -17,6 +17,7 @@
 
 // Logging uses LOG() macro for all diagnostic output
 #include "logger_bridge.hpp"
+#include "performance_counter.hpp"
 
 static SqlHandlePtr getEnvHandle() {
     static SqlHandlePtr envHandle = []() -> SqlHandlePtr {
@@ -48,6 +49,7 @@ static SqlHandlePtr getEnvHandle() {
 //-------------------------------------------------------------------------------------------------
 Connection::Connection(const std::u16string& conn_str, bool use_pool)
     : _connStr(conn_str), _autocommit(false), _fromPool(use_pool) {
+    PERF_TIMER("Connection::Connection");
     allocateDbcHandle();
 }
 
@@ -57,6 +59,7 @@ Connection::~Connection() {
 
 // Allocates connection handle
 void Connection::allocateDbcHandle() {
+    PERF_TIMER("Connection::allocateDbcHandle");
     auto _envHandle = getEnvHandle();
     SQLHANDLE dbc = nullptr;
     LOG("Allocating SQL Connection Handle");
@@ -66,6 +69,7 @@ void Connection::allocateDbcHandle() {
 }
 
 void Connection::connect(const py::dict& attrs_before) {
+    PERF_TIMER("Connection::connect");
     LOG("Connecting to database");
     // Apply access token before connect
     if (!attrs_before.is_none() && py::len(attrs_before) > 0) {
@@ -83,6 +87,7 @@ void Connection::connect(const py::dict& attrs_before) {
         // and SQL Server authentication — all pure I/O that doesn't need the GIL.
         // This allows other Python threads to run concurrently.
         py::gil_scoped_release release;
+        PERF_TIMER("Connection::connect::SQLDriverConnect_call");
         ret = SQLDriverConnect_ptr(_dbcHandle->get(), nullptr, connStrPtr, SQL_NTS, nullptr,
                                    0, nullptr, SQL_DRIVER_NOPROMPT);
     }
@@ -91,6 +96,7 @@ void Connection::connect(const py::dict& attrs_before) {
 }
 
 void Connection::disconnect() {
+    PERF_TIMER("Connection::disconnect");
     if (_dbcHandle) {
         LOG("Disconnecting from database");
 
@@ -185,6 +191,7 @@ void Connection::checkError(SQLRETURN ret) const {
 }
 
 void Connection::commit() {
+    PERF_TIMER("Connection::commit");
     if (!_dbcHandle) {
         ThrowStdException("Connection handle not allocated");
     }
@@ -200,6 +207,7 @@ void Connection::commit() {
 }
 
 void Connection::rollback() {
+    PERF_TIMER("Connection::rollback");
     if (!_dbcHandle) {
         ThrowStdException("Connection handle not allocated");
     }
@@ -215,6 +223,7 @@ void Connection::rollback() {
 }
 
 void Connection::setAutocommit(bool enable) {
+    PERF_TIMER("Connection::setAutocommit");
     if (!_dbcHandle) {
         ThrowStdException("Connection handle not allocated");
     }
@@ -253,6 +262,7 @@ bool Connection::getAutocommit() const {
 }
 
 SqlHandlePtr Connection::allocStatementHandle() {
+    PERF_TIMER("Connection::allocStatementHandle");
     if (!_dbcHandle) {
         ThrowStdException("Connection handle not allocated");
     }
@@ -516,6 +526,7 @@ std::chrono::steady_clock::time_point Connection::lastUsed() const {
 ConnectionHandle::ConnectionHandle(const std::u16string& connStr, bool usePool,
                                    const py::dict& attrsBefore)
     : _usePool(usePool), _connStr(connStr) {
+    PERF_TIMER("ConnectionHandle::ConnectionHandle");
     if (_usePool) {
         _conn = ConnectionPoolManager::getInstance().acquireConnection(_connStr, attrsBefore);
     } else {
@@ -531,6 +542,7 @@ ConnectionHandle::~ConnectionHandle() {
 }
 
 void ConnectionHandle::close() {
+    PERF_TIMER("ConnectionHandle::close");
     if (!_conn) {
         ThrowStdException("Connection object is not initialized");
     }
@@ -543,6 +555,7 @@ void ConnectionHandle::close() {
 }
 
 void ConnectionHandle::commit() {
+    PERF_TIMER("ConnectionHandle::commit");
     if (!_conn) {
         ThrowStdException("Connection object is not initialized");
     }
@@ -550,6 +563,7 @@ void ConnectionHandle::commit() {
 }
 
 void ConnectionHandle::rollback() {
+    PERF_TIMER("ConnectionHandle::rollback");
     if (!_conn) {
         ThrowStdException("Connection object is not initialized");
     }
@@ -557,6 +571,7 @@ void ConnectionHandle::rollback() {
 }
 
 void ConnectionHandle::setAutocommit(bool enabled) {
+    PERF_TIMER("ConnectionHandle::setAutocommit");
     if (!_conn) {
         ThrowStdException("Connection object is not initialized");
     }
@@ -571,6 +586,7 @@ bool ConnectionHandle::getAutocommit() const {
 }
 
 SqlHandlePtr ConnectionHandle::allocStatementHandle() {
+    PERF_TIMER("ConnectionHandle::allocStatementHandle");
     if (!_conn) {
         ThrowStdException("Connection object is not initialized");
     }
@@ -635,6 +651,7 @@ py::object Connection::getInfo(SQLUSMALLINT infoType) const {
 }
 
 py::object ConnectionHandle::getInfo(SQLUSMALLINT infoType) const {
+    PERF_TIMER("ConnectionHandle::getInfo");
     if (!_conn) {
         ThrowStdException("Connection object is not initialized");
     }
@@ -642,6 +659,7 @@ py::object ConnectionHandle::getInfo(SQLUSMALLINT infoType) const {
 }
 
 void ConnectionHandle::setAttr(int attribute, py::object value) {
+    PERF_TIMER("ConnectionHandle::setAttr");
     if (!_conn) {
         ThrowStdException("Connection not established");
     }
diff --git a/mssql_python/pybind/connection/connection_pool.cpp b/mssql_python/pybind/connection/connection_pool.cpp
index db891d081..6ba61328b 100644
--- a/mssql_python/pybind/connection/connection_pool.cpp
+++ b/mssql_python/pybind/connection/connection_pool.cpp
@@ -8,12 +8,14 @@
 
 // Logging uses LOG() macro for all diagnostic output
 #include "logger_bridge.hpp"
+#include "performance_counter.hpp"
 
 ConnectionPool::ConnectionPool(size_t max_size, int idle_timeout_secs)
     : _max_size(max_size), _idle_timeout_secs(idle_timeout_secs), _current_size(0) {}
 
 std::shared_ptr<Connection> ConnectionPool::acquire(const std::u16string& connStr,
                                                     const py::dict& attrs_before) {
+    PERF_TIMER("ConnectionPool::acquire");
     std::vector<std::shared_ptr<Connection>> to_disconnect;
     std::shared_ptr<Connection> valid_conn = nullptr;
     bool needs_connect = false;
@@ -120,6 +122,7 @@ std::shared_ptr<Connection> ConnectionPool::acquire(const std::u16string& connSt
 }
 
 void ConnectionPool::release(std::shared_ptr<Connection> conn) {
+    PERF_TIMER("ConnectionPool::release");
     bool should_disconnect = false;
     {
         std::lock_guard<std::mutex> lock(_mutex);
@@ -145,6 +148,7 @@ void ConnectionPool::release(std::shared_ptr<Connection> conn) {
 }
 
 void ConnectionPool::close() {
+    PERF_TIMER("ConnectionPool::close");
     std::vector<std::shared_ptr<Connection>> to_close;
     {
         std::lock_guard<std::mutex> lock(_mutex);
@@ -170,6 +174,7 @@ ConnectionPoolManager& ConnectionPoolManager::getInstance() {
 
 std::shared_ptr<Connection> ConnectionPoolManager::acquireConnection(const std::u16string& connStr,
                                                                      const py::dict& attrs_before) {
+    PERF_TIMER("ConnectionPoolManager::acquireConnection");
     std::shared_ptr<ConnectionPool> pool;
     {
         std::lock_guard<std::mutex> lock(_manager_mutex);
diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp
index 3cb00814c..13f077e88 100644
--- a/mssql_python/pybind/ddbc_bindings.cpp
+++ b/mssql_python/pybind/ddbc_bindings.cpp
@@ -8,6 +8,7 @@
 #include "connection/connection.h"
 #include "connection/connection_pool.h"
 #include "logger_bridge.hpp"
+#include "performance_counter.hpp"
 #include "utf_utils.h"
 
 
@@ -573,6 +574,7 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par
                          std::vector<ParamInfo>& paramInfos,
                          std::vector<std::shared_ptr<void>>& paramBuffers,
                          const std::string& charEncoding = "utf-8") {
+    PERF_TIMER("BindParameters");
     LOG("BindParameters: Starting parameter binding for statement handle %p "
         "with %zu parameters",
         (void*)hStmt, params.size());
@@ -985,12 +987,16 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par
             }
         }
         assert(SQLBindParameter_ptr && SQLGetStmtAttr_ptr && SQLSetDescField_ptr);
-        RETCODE rc = SQLBindParameter_ptr(
-            hStmt, static_cast<SQLUSMALLINT>(paramIndex + 1), /* 1-based indexing */
-            static_cast<SQLUSMALLINT>(paramInfo.inputOutputType),
-            static_cast<SQLSMALLINT>(paramInfo.paramCType),
-            static_cast<SQLSMALLINT>(paramInfo.paramSQLType), paramInfo.columnSize,
-            paramInfo.decimalDigits, dataPtr, bufferLength, strLenOrIndPtr);
+        RETCODE rc;
+        {
+            PERF_TIMER("BindParameters::SQLBindParameter_call");
+            rc = SQLBindParameter_ptr(
+                hStmt, static_cast<SQLUSMALLINT>(paramIndex + 1), /* 1-based indexing */
+                static_cast<SQLUSMALLINT>(paramInfo.inputOutputType),
+                static_cast<SQLSMALLINT>(paramInfo.paramCType),
+                static_cast<SQLSMALLINT>(paramInfo.paramSQLType), paramInfo.columnSize,
+                paramInfo.decimalDigits, dataPtr, bufferLength, strLenOrIndPtr);
+        }
         if (!SQL_SUCCEEDED(rc)) {
             LOG("BindParameters: SQLBindParameter failed for param[%d] - "
                 "SQLRETURN=%d, C_Type=%d, SQL_Type=%d",
@@ -1380,6 +1386,7 @@ DriverLoader& DriverLoader::getInstance() {
 }
 
 void DriverLoader::loadDriver() {
+    PERF_TIMER("DriverLoader::loadDriver");
     std::call_once(m_onceFlag, [this]() {
         LoadDriverOrThrowException();
         m_driverLoaded = true;
@@ -1427,6 +1434,7 @@ void SqlHandle::markImplicitlyFreed() {
  * If you need destruction logs, use explicit close() methods instead.
  */
 void SqlHandle::free() {
+    PERF_TIMER("SqlHandle::free");
     if (_handle && SQLFreeHandle_ptr) {
         // GH-610: Clear describe cache to prevent memory leak.
         describeCache.clear();
@@ -1529,6 +1537,7 @@ SQLRETURN SQLResetStmt_wrap(SqlHandlePtr statementHandle) {
 }
 
 SQLRETURN SQLGetTypeInfo_Wrapper(SqlHandlePtr StatementHandle, SQLSMALLINT DataType) {
+    PERF_TIMER("SQLGetTypeInfo_Wrapper");
     if (!SQLGetTypeInfo_ptr) {
         ThrowStdException("SQLGetTypeInfo function not loaded");
     }
@@ -1540,6 +1549,7 @@ SQLRETURN SQLGetTypeInfo_Wrapper(SqlHandlePtr StatementHandle, SQLSMALLINT DataT
 
 SQLRETURN SQLProcedures_wrap(SqlHandlePtr StatementHandle, const py::object& catalogObj,
                              const py::object& schemaObj, const py::object& procedureObj) {
+    PERF_TIMER("SQLProcedures_wrap");
     if (!SQLProcedures_ptr) {
         ThrowStdException("SQLProcedures function not loaded");
     }
@@ -1563,6 +1573,7 @@ SQLRETURN SQLForeignKeys_wrap(SqlHandlePtr StatementHandle, const py::object& pk
                               const py::object& pkSchemaObj, const py::object& pkTableObj,
                               const py::object& fkCatalogObj, const py::object& fkSchemaObj,
                               const py::object& fkTableObj) {
+    PERF_TIMER("SQLForeignKeys_wrap");
     if (!SQLForeignKeys_ptr) {
         ThrowStdException("SQLForeignKeys function not loaded");
     }
@@ -1594,6 +1605,7 @@ SQLRETURN SQLForeignKeys_wrap(SqlHandlePtr StatementHandle, const py::object& pk
 
 SQLRETURN SQLPrimaryKeys_wrap(SqlHandlePtr StatementHandle, const py::object& catalogObj,
                               const py::object& schemaObj, const std::u16string& table) {
+    PERF_TIMER("SQLPrimaryKeys_wrap");
     if (!SQLPrimaryKeys_ptr) {
         ThrowStdException("SQLPrimaryKeys function not loaded");
     }
@@ -1615,6 +1627,7 @@ SQLRETURN SQLPrimaryKeys_wrap(SqlHandlePtr StatementHandle, const py::object& ca
 SQLRETURN SQLStatistics_wrap(SqlHandlePtr StatementHandle, const py::object& catalogObj,
                              const py::object& schemaObj, const std::u16string& table,
                              SQLUSMALLINT unique, SQLUSMALLINT reserved) {
+    PERF_TIMER("SQLStatistics_wrap");
     if (!SQLStatistics_ptr) {
         ThrowStdException("SQLStatistics function not loaded");
     }
@@ -1636,6 +1649,7 @@ SQLRETURN SQLStatistics_wrap(SqlHandlePtr StatementHandle, const py::object& cat
 SQLRETURN SQLColumns_wrap(SqlHandlePtr StatementHandle, const py::object& catalogObj,
                           const py::object& schemaObj, const py::object& tableObj,
                           const py::object& columnObj) {
+    PERF_TIMER("SQLColumns_wrap");
     if (!SQLColumns_ptr) {
         ThrowStdException("SQLColumns function not loaded");
     }
@@ -1660,6 +1674,7 @@ SQLRETURN SQLColumns_wrap(SqlHandlePtr StatementHandle, const py::object& catalo
 
 // Helper function to check for driver errors
 ErrorInfo SQLCheckError_Wrap(SQLSMALLINT handleType, SqlHandlePtr handle, SQLRETURN retcode) {
+    PERF_TIMER("SQLCheckError_Wrap");
     LOG("SQLCheckError: Checking ODBC errors - handleType=%d, retcode=%d", handleType, retcode);
     ErrorInfo errorInfo;
     if (retcode == SQL_INVALID_HANDLE) {
@@ -1698,6 +1713,7 @@ ErrorInfo SQLCheckError_Wrap(SQLSMALLINT handleType, SqlHandlePtr handle, SQLRET
 }
 
 py::list SQLGetAllDiagRecords(SqlHandlePtr handle) {
+    PERF_TIMER("SQLGetAllDiagRecords");
     LOG("SQLGetAllDiagRecords: Retrieving all diagnostic records for handle "
         "%p, handleType=%d",
         (void*)handle->get(), handle->type());
@@ -1745,6 +1761,7 @@ py::list SQLGetAllDiagRecords(SqlHandlePtr handle) {
 
 // Wrap SQLExecDirect
 SQLRETURN SQLExecDirect_wrap(SqlHandlePtr StatementHandle, const std::u16string& Query) {
+    PERF_TIMER("SQLExecDirect_wrap");
     LOG("SQLExecDirect: Executing query directly - statement_handle=%p, "
         "query_length=%zu chars",
         (void*)StatementHandle->get(), Query.length());
@@ -1780,6 +1797,7 @@ SQLRETURN SQLExecDirect_wrap(SqlHandlePtr StatementHandle, const std::u16string&
 SQLRETURN SQLTables_wrap(SqlHandlePtr StatementHandle, const std::u16string& catalog,
                          const std::u16string& schema, const std::u16string& table,
                          const std::u16string& tableType) {
+    PERF_TIMER("SQLTables_wrap");
     if (!SQLTables_ptr) {
         LOG("SQLTables: Function pointer not initialized, loading driver");
         DriverLoader::getInstance().loadDriver();
@@ -1814,6 +1832,7 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, const std::u16stri
                           const py::list& params, std::vector<ParamInfo>& paramInfos,
                           py::list& isStmtPrepared, const bool usePrepare,
                           const py::dict& encodingSettings) {
+    PERF_TIMER("SQLExecute_wrap");
     LOG("SQLExecute: Executing %s query - statement_handle=%p, "
         "param_count=%zu, query_length=%zu chars",
         (params.size() > 0 ? "parameterized" : "direct"), (void*)statementHandle->get(),
@@ -2051,6 +2070,7 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list&
                              std::vector<ParamInfo>& paramInfos, size_t paramSetSize,
                              std::vector<std::shared_ptr<void>>& paramBuffers,
                              const std::string& charEncoding = "utf-8") {
+    PERF_TIMER("BindParameterArray");
     LOG("BindParameterArray: Starting column-wise array binding - "
         "param_count=%zu, param_set_size=%zu",
         columnwise_params.size(), paramSetSize);
@@ -2665,12 +2685,15 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list&
             LOG("BindParameterArray: Calling SQLBindParameter - "
                 "param_index=%d, buffer_length=%lld",
                 paramIndex, static_cast<long long>(bufferLength));
-            RETCODE rc =
-                SQLBindParameter_ptr(hStmt, static_cast<SQLUSMALLINT>(paramIndex + 1),
-                                     static_cast<SQLUSMALLINT>(info.inputOutputType),
-                                     static_cast<SQLSMALLINT>(info.paramCType),
-                                     static_cast<SQLSMALLINT>(info.paramSQLType), info.columnSize,
-                                     info.decimalDigits, dataPtr, bufferLength, strLenOrIndArray);
+            RETCODE rc;
+            {
+                PERF_TIMER("BindParameterArray::SQLBindParameter_call");
+                rc = SQLBindParameter_ptr(hStmt, static_cast<SQLUSMALLINT>(paramIndex + 1),
+                                         static_cast<SQLUSMALLINT>(info.inputOutputType),
+                                         static_cast<SQLSMALLINT>(info.paramCType),
+                                         static_cast<SQLSMALLINT>(info.paramSQLType), info.columnSize,
+                                         info.decimalDigits, dataPtr, bufferLength, strLenOrIndArray);
+            }
             if (!SQL_SUCCEEDED(rc)) {
                 LOG("BindParameterArray: SQLBindParameter failed - "
                     "param_index=%d, SQLRETURN=%d",
@@ -2694,6 +2717,7 @@ SQLRETURN SQLExecuteMany_wrap(const SqlHandlePtr statementHandle, const std::u16
                               const py::list& columnwise_params,
                               std::vector<ParamInfo>& paramInfos, size_t paramSetSize,
                               const py::dict& encodingSettings) {
+    PERF_TIMER("SQLExecuteMany_wrap");
     LOG("SQLExecuteMany: Starting batch execution - param_count=%zu, "
         "param_set_size=%zu",
         columnwise_params.size(), paramSetSize);
@@ -2857,6 +2881,7 @@ SQLRETURN SQLExecuteMany_wrap(const SqlHandlePtr statementHandle, const std::u16
 
 // Wrap SQLNumResultCols
 SQLSMALLINT SQLNumResultCols_wrap(SqlHandlePtr statementHandle) {
+    PERF_TIMER("SQLNumResultCols_wrap");
     LOG("SQLNumResultCols: Getting number of columns in result set for "
         "statement_handle=%p",
         (void*)statementHandle->get());
@@ -2874,6 +2899,7 @@ SQLSMALLINT SQLNumResultCols_wrap(SqlHandlePtr statementHandle) {
 
 // Wrap SQLDescribeCol
 SQLRETURN SQLDescribeCol_wrap(SqlHandlePtr StatementHandle, py::list& ColumnMetadata) {
+    PERF_TIMER("SQLDescribeCol_wrap");
     LOG("SQLDescribeCol: Getting column descriptions for statement_handle=%p",
         (void*)StatementHandle->get());
     if (!SQLDescribeCol_ptr) {
@@ -2920,6 +2946,7 @@ SQLRETURN SQLSpecialColumns_wrap(SqlHandlePtr StatementHandle, SQLSMALLINT ident
                                  const py::object& catalogObj, const py::object& schemaObj,
                                  const std::u16string& table, SQLSMALLINT scope,
                                  SQLSMALLINT nullable) {
+    PERF_TIMER("SQLSpecialColumns_wrap");
     if (!SQLSpecialColumns_ptr) {
         ThrowStdException("SQLSpecialColumns function not loaded");
     }
@@ -2940,6 +2967,7 @@ SQLRETURN SQLSpecialColumns_wrap(SqlHandlePtr StatementHandle, SQLSMALLINT ident
 
 // Wrap SQLFetch to retrieve rows
 SQLRETURN SQLFetch_wrap(SqlHandlePtr StatementHandle) {
+    PERF_TIMER("SQLFetch_wrap");
     LOG("SQLFetch: Fetching next row for statement_handle=%p", (void*)StatementHandle->get());
     if (!SQLFetch_ptr) {
         LOG("SQLFetch: Function pointer not initialized, loading driver");
@@ -2954,6 +2982,7 @@ SQLRETURN SQLFetch_wrap(SqlHandlePtr StatementHandle) {
 // Non-static so it can be called from inline functions in header
 py::object FetchLobColumnData(SQLHSTMT hStmt, SQLUSMALLINT colIndex, SQLSMALLINT cType,
                               bool isWideChar, bool isBinary, const std::string& charEncoding) {
+    PERF_TIMER("FetchLobColumnData");
     std::vector<char> buffer;
     SQLRETURN ret = SQL_SUCCESS_WITH_INFO;
     int loopCount = 0;
@@ -3135,6 +3164,7 @@ SQLRETURN SQLGetData_wrap(SqlHandlePtr StatementHandle, SQLUSMALLINT colCount, p
                           const std::string& charEncoding = "utf-16le",
                           const std::string& wcharEncoding = "utf-16le",
                           int charCtype = SQL_C_WCHAR) {
+    PERF_TIMER("SQLGetData_wrap");
     // Note: wcharEncoding parameter is reserved for future use
     // Currently WCHAR data always uses UTF-16LE for Windows compatibility
     (void)wcharEncoding;  // Suppress unused parameter warning
@@ -3809,6 +3839,7 @@ SQLRETURN SQLGetData_wrap(SqlHandlePtr StatementHandle, SQLUSMALLINT colCount, p
 
 SQLRETURN SQLFetchScroll_wrap(SqlHandlePtr StatementHandle, SQLSMALLINT FetchOrientation,
                               SQLLEN FetchOffset, py::list& row_data) {
+    PERF_TIMER("SQLFetchScroll_wrap");
     LOG("SQLFetchScroll_wrap: Fetching with scroll orientation=%d, offset=%ld", FetchOrientation,
         (long)FetchOffset);
     if (!SQLFetchScroll_ptr) {
@@ -3845,6 +3876,7 @@ SQLRETURN SQLFetchScroll_wrap(SqlHandlePtr StatementHandle, SQLSMALLINT FetchOri
 // TODO: Move to anonymous namespace, since it is not used outside this file
 SQLRETURN SQLBindColums(SQLHSTMT hStmt, ColumnBuffers& buffers, py::list& columnNames,
                         SQLUSMALLINT numCols, int fetchSize, int charCtype = SQL_C_WCHAR) {
+    PERF_TIMER("SQLBindColums");
     SQLRETURN ret = SQL_SUCCESS;
     const bool useWideChar = (charCtype == SQL_C_WCHAR);
     // Bind columns based on their data types
@@ -4011,11 +4043,13 @@ SQLRETURN FetchBatchData(SQLHSTMT hStmt, ColumnBuffers& buffers, py::list& colum
                          const std::vector<SQLUSMALLINT>& lobColumns,
                          const std::string& charEncoding = "utf-16le",
                          int charCtype = SQL_C_WCHAR) {
+    PERF_TIMER("FetchBatchData");
     LOG("FetchBatchData: Fetching data in batches");
     SQLRETURN ret;
     {
         // Release the GIL during the blocking ODBC fetch
         py::gil_scoped_release release;
+        PERF_TIMER("FetchBatchData::SQLFetchScroll_call");
         ret = SQLFetchScroll_ptr(hStmt, SQL_FETCH_NEXT, 0);
     }
     if (ret == SQL_NO_DATA) {
@@ -4029,6 +4063,7 @@ SQLRETURN FetchBatchData(SQLHSTMT hStmt, ColumnBuffers& buffers, py::list& colum
         return ret;
     }
     // Pre-cache column metadata to avoid repeated dictionary lookups
+    PERF_TIMER("FetchBatchData::cache_column_metadata");
     struct ColumnInfo {
         SQLSMALLINT dataType;
         SQLULEN columnSize;
@@ -4150,6 +4185,7 @@ SQLRETURN FetchBatchData(SQLHSTMT hStmt, ColumnBuffers& buffers, py::list& colum
     // Create each row, fill it completely, then append to results list
     // This prevents data corruption (no partially-filled rows) and simplifies
     // error handling
+    PERF_TIMER("FetchBatchData::construct_rows");
     PyObject* rowsList = rows.ptr();
 
     // RAII wrapper to ensure row cleanup on exception (CRITICAL: prevents
@@ -4474,6 +4510,7 @@ SQLRETURN FetchMany_wrap(SqlHandlePtr StatementHandle, py::list& rows, int fetch
                          const std::string& charEncoding = "utf-16le",
                          const std::string& wcharEncoding = "utf-16le",
                          int charCtype = SQL_C_WCHAR) {
+    PERF_TIMER("FetchMany_wrap");
     // Issue #531: upgrade SQL_C_CHAR + utf-8 to SQL_C_WCHAR on Windows so the
     // driver does lossless UTF-16 conversion instead of returning ACP bytes.
     charCtype = EffectiveCharCtypeForFetch(charCtype, charEncoding);
@@ -4683,6 +4720,7 @@ int32_t days_from_civil(int y, int m, int d) {
 SQLRETURN FetchArrowBatch_wrap(SqlHandlePtr StatementHandle, py::list& capsules,
                                int arrowBatchSize,
                                int charCtype) {
+    PERF_TIMER("FetchArrowBatch_wrap");
     // Fetch narrow char data as SQL_C_CHAR if on Linux/macOS and configured by the user
     charCtype = EffectiveCharCtypeForFetch(charCtype, "utf-8");
 
@@ -5593,6 +5631,7 @@ SQLRETURN FetchAll_wrap(SqlHandlePtr StatementHandle, py::list& rows,
                         const std::string& charEncoding = "utf-16le",
                         const std::string& wcharEncoding = "utf-16le",
                         int charCtype = SQL_C_WCHAR) {
+    PERF_TIMER("FetchAll_wrap");
     // Issue #531: upgrade SQL_C_CHAR + utf-8 to SQL_C_WCHAR on Windows so the
     // driver does lossless UTF-16 conversion instead of returning ACP bytes.
     charCtype = EffectiveCharCtypeForFetch(charCtype, charEncoding);
@@ -5739,6 +5778,7 @@ SQLRETURN FetchOne_wrap(SqlHandlePtr StatementHandle, py::list& row,
                         const std::string& charEncoding = "utf-16le",
                         const std::string& wcharEncoding = "utf-16le",
                         int charCtype = SQL_C_WCHAR) {
+    PERF_TIMER("FetchOne_wrap");
     // Issue #531: upgrade SQL_C_CHAR + utf-8 to SQL_C_WCHAR on Windows so the
     // driver does lossless UTF-16 conversion instead of returning ACP bytes.
     charCtype = EffectiveCharCtypeForFetch(charCtype, charEncoding);
@@ -5773,6 +5813,7 @@ SQLRETURN FetchOne_wrap(SqlHandlePtr StatementHandle, py::list& row,
 
 // Wrap SQLMoreResults
 SQLRETURN SQLMoreResults_wrap(SqlHandlePtr StatementHandle) {
+    PERF_TIMER("SQLMoreResults_wrap");
     LOG("SQLMoreResults_wrap: Check for more results");
     if (!SQLMoreResults_ptr) {
         LOG("SQLMoreResults_wrap: Function pointer not initialized. Loading "
@@ -5787,6 +5828,7 @@ SQLRETURN SQLMoreResults_wrap(SqlHandlePtr StatementHandle) {
 
 // Wrap SQLFreeHandle
 SQLRETURN SQLFreeHandle_wrap(SQLSMALLINT HandleType, SqlHandlePtr Handle) {
+    PERF_TIMER("SQLFreeHandle_wrap");
     LOG("SQLFreeHandle_wrap: Free SQL handle type=%d", HandleType);
     // Guard against a null/None handle being passed from Python - dereferencing
     // Handle->get() on a null shared_ptr would segfault.
@@ -5818,6 +5860,7 @@ SQLRETURN SQLFreeHandle_wrap(SQLSMALLINT HandleType, SqlHandlePtr Handle) {
 
 // Wrap SQLRowCount
 SQLLEN SQLRowCount_wrap(SqlHandlePtr StatementHandle) {
+    PERF_TIMER("SQLRowCount_wrap");
     LOG("SQLRowCount_wrap: Get number of rows affected by last execute");
     if (!SQLRowCount_ptr) {
         LOG("SQLRowCount_wrap: Function pointer not initialized. Loading the "
@@ -6012,6 +6055,27 @@ PYBIND11_MODULE(ddbc_bindings, m) {
               return SQLColumns_wrap(StatementHandle, catalog, schema, table, column);
           });
 
+    // Add profiling submodule
+    auto profiling = m.def_submodule("profiling", "Performance profiling");
+    profiling.def("enable", []() { mssql_profiling::PerformanceCounter::instance().enable(); }, 
+                  "Enable performance profiling");
+    profiling.def("disable", []() { mssql_profiling::PerformanceCounter::instance().disable(); }, 
+                  "Disable performance profiling");
+    profiling.def("get_stats", []() { return mssql_profiling::PerformanceCounter::instance().get_stats(); }, 
+                  "Get profiling statistics");
+    profiling.def("get_timeline", []() { return mssql_profiling::PerformanceCounter::instance().get_timeline(); },
+                  "Get timeline events (list of {name, start_us, duration_us})");
+    profiling.def("reset", []() { mssql_profiling::PerformanceCounter::instance().reset(); }, 
+                  "Reset profiling statistics and timeline");
+    profiling.def("reset_stats_only", []() { mssql_profiling::PerformanceCounter::instance().reset_stats_only(); }, 
+                  "Reset profiling statistics but keep timeline");
+    profiling.def("is_enabled", []() { return mssql_profiling::PerformanceCounter::instance().is_enabled(); }, 
+                  "Check if profiling is enabled");
+    profiling.def("enable_timeline", []() { mssql_profiling::PerformanceCounter::instance().enable_timeline(); },
+                  "Enable timeline recording (resets epoch)");
+    profiling.def("disable_timeline", []() { mssql_profiling::PerformanceCounter::instance().disable_timeline(); },
+                  "Disable timeline recording");
+
     // Add a version attribute
     m.attr("__version__") = "1.0.0";
 
diff --git a/mssql_python/pybind/performance_counter.hpp b/mssql_python/pybind/performance_counter.hpp
new file mode 100644
index 000000000..37f63122b
--- /dev/null
+++ b/mssql_python/pybind/performance_counter.hpp
@@ -0,0 +1,163 @@
+/*
+ * Performance Profiling for mssql-python
+ * Thread-safe performance counter with Python API
+ */
+
+#pragma once
+
+#include <chrono>
+#include <string>
+#include <vector>
+#include <unordered_map>
+#include <mutex>
+#include <pybind11/pybind11.h>
+#include <pybind11/stl.h>
+
+namespace py = pybind11;
+
+namespace mssql_profiling {
+
+// Platform detection
+#if defined(_WIN32) || defined(_WIN64)
+    #define PROFILING_PLATFORM "windows"
+#elif defined(__linux__)
+    #define PROFILING_PLATFORM "linux"
+#elif defined(__APPLE__) || defined(__MACH__)
+    #define PROFILING_PLATFORM "macos"
+#else
+    #define PROFILING_PLATFORM "unknown"
+#endif
+
+struct PerfStats {
+    int64_t total_time_us = 0;
+    int64_t call_count = 0;
+    int64_t min_time_us = INT64_MAX;
+    int64_t max_time_us = 0;
+};
+
+struct TimelineEvent {
+    std::string name;
+    int64_t start_us;   // offset from epoch_
+    int64_t duration_us;
+};
+
+class PerformanceCounter {
+private:
+    std::unordered_map<std::string, PerfStats> counters_;
+    std::vector<TimelineEvent> timeline_;
+    std::mutex mutex_;
+    bool enabled_ = false;
+    bool timeline_enabled_ = false;
+    std::chrono::time_point<std::chrono::high_resolution_clock> epoch_;
+
+public:
+    static PerformanceCounter& instance() {
+        static PerformanceCounter counter;
+        return counter;
+    }
+
+    void enable() { enabled_ = true; }
+    void disable() { enabled_ = false; }
+    bool is_enabled() const { return enabled_; }
+
+    void enable_timeline() {
+        timeline_enabled_ = true;
+        epoch_ = std::chrono::high_resolution_clock::now();
+    }
+    void disable_timeline() { timeline_enabled_ = false; }
+    bool is_timeline_enabled() const { return timeline_enabled_; }
+
+    void record(const std::string& name, int64_t duration_us,
+                std::chrono::time_point<std::chrono::high_resolution_clock> start) {
+        if (!enabled_) return;
+        
+        std::lock_guard<std::mutex> lock(mutex_);
+        auto& stats = counters_[name];
+        stats.total_time_us += duration_us;
+        stats.call_count++;
+        stats.min_time_us = std::min(stats.min_time_us, duration_us);
+        stats.max_time_us = std::max(stats.max_time_us, duration_us);
+
+        if (timeline_enabled_) {
+            auto offset = std::chrono::duration_cast<std::chrono::microseconds>(start - epoch_).count();
+            timeline_.push_back({name, offset, duration_us});
+        }
+    }
+
+    py::dict get_stats() {
+        std::lock_guard<std::mutex> lock(mutex_);
+        py::dict result;
+        
+        for (const auto& [name, stats] : counters_) {
+            py::dict d;
+            d["total_us"] = stats.total_time_us;
+            d["calls"] = stats.call_count;
+            d["avg_us"] = stats.call_count > 0 ? stats.total_time_us / stats.call_count : 0;
+            d["min_us"] = stats.min_time_us == INT64_MAX ? 0 : stats.min_time_us;
+            d["max_us"] = stats.max_time_us;
+            d["platform"] = PROFILING_PLATFORM;
+            result[py::str(name)] = d;
+        }
+        
+        return result;
+    }
+
+    void reset() {
+        std::lock_guard<std::mutex> lock(mutex_);
+        counters_.clear();
+        timeline_.clear();
+    }
+
+    void reset_stats_only() {
+        std::lock_guard<std::mutex> lock(mutex_);
+        counters_.clear();
+    }
+
+    py::list get_timeline() {
+        std::lock_guard<std::mutex> lock(mutex_);
+        py::list result;
+        for (const auto& ev : timeline_) {
+            py::dict d;
+            d["name"] = ev.name;
+            d["start_us"] = ev.start_us;
+            d["duration_us"] = ev.duration_us;
+            result.append(d);
+        }
+        return result;
+    }
+};
+
+// RAII timer - automatically records on destruction
+class ScopedTimer {
+private:
+    const char* name_;
+    std::chrono::time_point<std::chrono::high_resolution_clock> start_;
+    
+public:
+    explicit ScopedTimer(const char* name) : name_(name) {
+        if (PerformanceCounter::instance().is_enabled()) {
+            start_ = std::chrono::high_resolution_clock::now();
+        }
+    }
+    
+    ~ScopedTimer() {
+        if (PerformanceCounter::instance().is_enabled()) {
+            auto end = std::chrono::high_resolution_clock::now();
+            auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start_).count();
+            PerformanceCounter::instance().record(name_, duration, start_);
+        }
+    }
+};
+
+} // namespace mssql_profiling
+
+// Convenience macro - use __COUNTER__ for unique variable names even with nested timers
+// __COUNTER__ is supported by MSVC, GCC, and Clang
+#define PERF_TIMER_CONCAT_IMPL(x, y) x##y
+#define PERF_TIMER_CONCAT(x, y) PERF_TIMER_CONCAT_IMPL(x, y)
+
+// PROFILING ENABLED - Creates actual timers
+#define PERF_TIMER(name) mssql_profiling::ScopedTimer PERF_TIMER_CONCAT(_perf_timer_, __COUNTER__)("ddbc::" name)
+
+// PROFILING DISABLED - Uncomment below and comment above to make PERF_TIMER a no-op
+// #define PERF_TIMER(name) do {} while(0)
diff --git a/profiler/README.md b/profiler/README.md
new file mode 100644
index 000000000..ff2384da0
--- /dev/null
+++ b/profiler/README.md
@@ -0,0 +1,88 @@
+# mssql-python profiler
+
+Unified Python + C++ performance instrumentation for the mssql-python driver.
+
+Timers on both layers are merged into a single sorted view. Python-layer entries use a `py::` prefix, C++ entries have no prefix — so you can immediately see where time is spent across the boundary.
+
+## Quick start
+
+```bash
+# Set connection string
+export DB_CONNECTION_STRING="Server=localhost,1433;Database=master;UID=sa;Pwd=...;Encrypt=no;TrustServerCertificate=yes;"
+
+# Run all scenarios
+python -m profiler
+
+# Run specific scenarios
+python -m profiler --scenarios fetchall insertmanyvalues
+
+# List available scenarios
+python -m profiler --list
+
+# Pass connection string directly
+python -m profiler --conn-str "Server=..."
+```
+
+## Programmatic usage
+
+```python
+from profiler import Profiler
+
+with Profiler("Server=localhost,1433;...") as p:
+    results = p.run("fetchall", "insertmanyvalues")
+    # results is a list of dicts with keys: title, wall_ms, cpp, py, detail
+```
+
+## Available scenarios
+
+| Name | What it measures |
+|---|---|
+| `connect` | Connection establishment |
+| `select` | `cursor.execute()` SELECT + `fetchall()` (100 rows) |
+| `insert` | 100 individual `cursor.execute()` INSERTs (9 params each) |
+| `executemany` | `cursor.executemany()` with 5K rows |
+| `fetchall` | `cursor.fetchall()` on 50K rows |
+| `fetchone` | `cursor.fetchone()` loop over 1K rows |
+| `fetchmany` | `cursor.fetchmany(1000)` loop over 50K rows |
+| `commit_rollback` | 100 commits + 100 rollbacks |
+| `arrow` | `cursor.fetch_arrow()` on 50K rows |
+| `insertmanyvalues` | SQLAlchemy pattern — 100K rows via batched `cursor.execute()` with 2000 params/call |
+
+## Architecture
+
+```
+profiler/
+├── __init__.py     # Public API: Profiler class
+├── __main__.py     # CLI entry point (python -m profiler)
+├── core.py         # Profiler orchestration, connection management
+├── scenarios.py    # Self-contained benchmark functions
+├── reporter.py     # Stats merging and table formatting
+└── README.md
+
+mssql_python/
+└── perf_timer.py   # Python-layer instrumentation (lives in the library)
+```
+
+**`perf_timer.py`** is the in-library instrumentation — `perf_phase()` context managers and `perf_start()`/`perf_stop()` pairs embedded in `cursor.py`. It's compile-time-toggled: when disabled (`_enabled = False`), each timer is a single `if` check (~20ns).
+
+**`profiler/`** is the runner — it enables both layers, executes scenarios, collects stats, and reports.
+
+## Output format
+
+Both layers report `{calls, total_us, min_us, max_us}` per timer. The reporter merges and sorts by `total_us` descending:
+
+```
+====================================================================================
+INSERTMANYVALUES (100,000 rows, 2000 params/call)
+====================================================================================
+  Function                                         Calls    Total(ms)      Avg(us)
+  ---------------------------------------------------------------------------------
+  py::execute::param_type_detection                  100       2009.3      20092.6    <-- Python
+  py::execute::cpp_call                              100       1414.6      14146.1    <-- Python
+  SQLExecute_wrap                                    100       1367.6      13676.3    <-- C++
+  py::execute::diag_records                          100        962.2       9621.5    <-- Python
+  SQLGetAllDiagRecords                               100        961.0       9610.3    <-- C++
+  BindParameters                                     100        189.4       1893.9    <-- C++
+```
+
+The `py::execute::cpp_call` timer wraps the C++ call from the Python side — so `cpp_call - SQLExecute_wrap` = pybind11 boundary crossing overhead.
diff --git a/profiler/__init__.py b/profiler/__init__.py
new file mode 100644
index 000000000..26eea4618
--- /dev/null
+++ b/profiler/__init__.py
@@ -0,0 +1,21 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+"""
+mssql-python profiler — unified Python + C++ performance instrumentation.
+
+Usage:
+    python -m profiler                          # run all scenarios
+    python -m profiler --scenarios fetch insert  # run specific scenarios
+    python -m profiler --conn-str "Server=..."   # pass connection string
+
+Programmatic:
+    from profiler import Profiler
+
+    p = Profiler(conn_str)
+    p.run("fetchall", "insertmanyvalues")
+    p.report()
+"""
+
+from profiler.core import Profiler
+
+__all__ = ["Profiler"]
diff --git a/profiler/__main__.py b/profiler/__main__.py
new file mode 100644
index 000000000..0ef91058e
--- /dev/null
+++ b/profiler/__main__.py
@@ -0,0 +1,67 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+"""
+CLI entry point: python -m profiler [options]
+"""
+
+import argparse
+import sys
+
+from profiler import Profiler
+from profiler.scenarios import SCENARIOS
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        prog="python -m profiler",
+        description="mssql-python performance profiler — Python + C++ instrumentation",
+    )
+    parser.add_argument(
+        "--conn-str",
+        help="Connection string (or set DB_CONNECTION_STRING env var)",
+    )
+    parser.add_argument(
+        "--scenarios",
+        nargs="+",
+        metavar="NAME",
+        choices=list(SCENARIOS.keys()),
+        help=f"Scenarios to run (default: all). Choices: {', '.join(SCENARIOS.keys())}",
+    )
+    parser.add_argument(
+        "--script",
+        metavar="FILE",
+        help="Run a custom .py script. The script gets `conn` and `cursor` injected.",
+    )
+    parser.add_argument(
+        "--timeline",
+        action="store_true",
+        help="Show chronological timeline instead of aggregate stats",
+    )
+    parser.add_argument(
+        "--list",
+        action="store_true",
+        help="List available scenarios and exit",
+    )
+    args = parser.parse_args()
+
+    if args.list:
+        print("Available scenarios:")
+        for name in SCENARIOS:
+            print(f"  {name}")
+        return
+
+    try:
+        with Profiler(args.conn_str, timeline=args.timeline) as p:
+            if args.script:
+                p.run_script(args.script)
+            elif args.scenarios:
+                p.run(*args.scenarios)
+            else:
+                p.run()
+    except ValueError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/profiler/core.py b/profiler/core.py
new file mode 100644
index 000000000..b7899cdf3
--- /dev/null
+++ b/profiler/core.py
@@ -0,0 +1,230 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+"""
+Profiler core — orchestrates scenarios, collects stats from both layers.
+
+    from profiler import Profiler
+
+    p = Profiler("Server=localhost,1433;UID=sa;Pwd=...;Encrypt=no;TrustServerCertificate=yes;")
+    p.run()              # all scenarios
+    p.run("fetchall")    # one scenario
+"""
+
+from __future__ import annotations
+
+import os
+import platform
+import sys
+
+from profiler.reporter import print_stats, print_timeline
+from profiler.scenarios import SCENARIOS, setup_test_data
+
+
+class _ProfilingContext:
+    """Thin wrapper that enables/disables/collects from both C++ and Python profiling."""
+
+    def __init__(self):
+        from mssql_python import ddbc_bindings, perf_timer
+
+        self._cpp = ddbc_bindings.profiling
+        self._py = perf_timer
+        self._timeline_mode = False
+
+    def set_timeline(self, on: bool):
+        self._timeline_mode = on
+
+    def enable(self, timeline: bool = False):
+        self._cpp.enable()
+        self._py.enable()
+        if timeline or self._timeline_mode:
+            self._cpp.enable_timeline()
+            self._py.enable_timeline()
+
+    def collect(self) -> tuple[dict, dict]:
+        cpp = self._cpp.get_stats()
+        py = self._py.get_stats()
+        if self._timeline_mode:
+            self._cpp.reset_stats_only()
+            self._py.reset_stats_only()
+        else:
+            self._cpp.reset()
+            self._py.reset()
+        return cpp, py
+
+    def collect_timeline(self) -> tuple[list, list]:
+        cpp_tl = self._cpp.get_timeline()
+        py_tl = self._py.get_timeline()
+        self._cpp.reset()
+        self._py.reset()
+        return cpp_tl, py_tl
+
+    def disable_timeline(self):
+        self._cpp.disable_timeline()
+        self._py.disable_timeline()
+
+
+class Profiler:
+    def __init__(self, conn_str: str | None = None, timeline: bool = False):
+        self.conn_str = conn_str or os.getenv("DB_CONNECTION_STRING")
+        if not self.conn_str:
+            raise ValueError(
+                "Connection string required. Pass it directly or set DB_CONNECTION_STRING."
+            )
+        self._ctx = _ProfilingContext()
+        self._timeline = timeline
+        self._ctx.set_timeline(timeline)
+        self._conn = None
+        self._table = None
+        self._results: list[dict] = []
+
+    def _ensure_connection(self):
+        if self._conn is None:
+            from mssql_python import connect
+
+            self._conn = connect(self.conn_str)
+            self._conn.autocommit = False
+
+    def _ensure_test_data(self):
+        if self._table is None:
+            self._ensure_connection()
+            print("Setting up test data...", end=" ", flush=True)
+            self._table = setup_test_data(self._conn)
+            # Drain any stats leaked from setup
+            self._ctx.enable()
+            self._ctx.collect()
+            print("Done", flush=True)
+
+    def run(self, *scenario_names: str) -> list[dict]:
+        names = list(scenario_names) if scenario_names else list(SCENARIOS.keys())
+        unknown = set(names) - set(SCENARIOS.keys())
+        if unknown:
+            raise ValueError(f"Unknown scenarios: {unknown}. Available: {list(SCENARIOS.keys())}")
+
+        self._print_header()
+        results = []
+
+        for i, name in enumerate(names, 1):
+            fn, needs_table = SCENARIOS[name]
+            print(f"\n{'#' * 100}")
+            print(f"# {i}. {name.upper()}")
+            print(f"{'#' * 100}")
+
+            # Build args based on what the scenario function needs
+            if name == "connect":
+                result = fn(self.conn_str, self._ctx)
+            elif name == "insertmanyvalues":
+                self._ensure_connection()
+                result = fn(self._conn, self._ctx)
+            elif name == "commit_rollback":
+                self._ensure_connection()
+                result = fn(self._conn, self._ctx)
+            elif needs_table:
+                self._ensure_test_data()
+                result = fn(self._conn, self._table, self._ctx)
+            else:
+                self._ensure_connection()
+                result = fn(self._conn, self._ctx)
+
+            # Collect timeline if enabled
+            if self._timeline:
+                cpp_tl, py_tl = self._ctx.collect_timeline()
+                self._ctx.disable_timeline()
+                result["cpp_timeline"] = cpp_tl
+                result["py_timeline"] = py_tl
+
+            # Print result
+            detail = result.get("detail", "")
+            if detail:
+                print(f"\n  {detail}, Wall clock: {result['wall_ms']:.1f}ms")
+            else:
+                print(f"\n  Wall clock: {result['wall_ms']:.1f}ms")
+
+            if self._timeline:
+                print_timeline(
+                    result.get("cpp_timeline"), result.get("py_timeline"), result["title"]
+                )
+            else:
+                print_stats(result["cpp"], result["py"], result["title"])
+
+            results.append(result)
+
+        self._results = results
+        self._print_footer()
+        return results
+
+    def run_script(self, script_path: str) -> dict:
+        """Run a user-supplied .py script and report whatever timers it hits.
+
+        The script gets `conn` (a live Connection) and `cursor` (a fresh Cursor)
+        injected into its namespace.
+        """
+        import time
+        from pathlib import Path
+
+        path = Path(script_path)
+        if not path.is_file():
+            raise FileNotFoundError(f"Script not found: {script_path}")
+
+        self._ensure_connection()
+        cursor = self._conn.cursor()
+
+        self._print_header()
+        print(f"\n{'#' * 100}")
+        print(f"# CUSTOM: {path.name}")
+        print(f"{'#' * 100}")
+
+        ns = {"conn": self._conn, "cursor": cursor}
+        code = compile(path.read_text(), str(path), "exec")
+
+        self._ctx.enable(timeline=self._timeline)
+        t0 = time.perf_counter()
+        exec(code, ns)  # noqa: S102
+        wall_ms = (time.perf_counter() - t0) * 1000
+        cpp, py = self._ctx.collect()
+
+        result = {
+            "title": f"CUSTOM: {path.name}",
+            "wall_ms": wall_ms,
+            "cpp": cpp,
+            "py": py,
+        }
+
+        if self._timeline:
+            cpp_tl, py_tl = self._ctx.collect_timeline()
+            self._ctx.disable_timeline()
+            result["cpp_timeline"] = cpp_tl
+            result["py_timeline"] = py_tl
+
+        cursor.close()
+
+        print(f"\n  Wall clock: {wall_ms:.1f}ms")
+        if self._timeline:
+            print_timeline(result.get("cpp_timeline"), result.get("py_timeline"), result["title"])
+        else:
+            print_stats(cpp, py, result["title"])
+        self._print_footer()
+        return result
+
+    def close(self):
+        if self._conn:
+            self._conn.close()
+            self._conn = None
+            self._table = None
+
+    def _print_header(self):
+        print("=" * 100)
+        print("mssql-python profiler")
+        print("=" * 100)
+        print(f"Platform: {platform.system()} {platform.release()} ({platform.machine()})")
+        print(f"Python:   {platform.python_version()}")
+
+    def _print_footer(self):
+        print(f"\n{'=' * 100}")
+        print("PROFILING COMPLETE")
+        print("=" * 100)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        self.close()
diff --git a/profiler/reporter.py b/profiler/reporter.py
new file mode 100644
index 000000000..d962718ea
--- /dev/null
+++ b/profiler/reporter.py
@@ -0,0 +1,89 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+"""Stats collection and reporting — merges Python (py::) and C++ timer data."""
+
+from __future__ import annotations
+
+
+def merge_stats(cpp_stats: dict | None, py_stats: dict | None) -> dict:
+    merged = {}
+    for name, s in (cpp_stats or {}).items():
+        merged[name] = s
+    for name, s in (py_stats or {}).items():
+        merged[name] = s
+    return merged
+
+
+def merge_timeline(cpp_timeline: list | None, py_timeline: list | None) -> list[dict]:
+    events = list(cpp_timeline or []) + list(py_timeline or [])
+    events.sort(key=lambda e: e["start_us"])
+    return events
+
+
+def format_stats_table(stats: dict, title: str) -> str:
+    if not stats:
+        return f"\n{title}: No data collected"
+
+    lines = [
+        "",
+        "=" * 100,
+        title,
+        "=" * 100,
+        f"  {'Function':<55} {'Calls':>8} {'Total(ms)':>12} "
+        f"{'Avg(us)':>12} {'Min(us)':>10} {'Max(us)':>10}",
+        f"  {'-' * 107}",
+    ]
+    for name, s in sorted(stats.items(), key=lambda x: x[1]["total_us"], reverse=True):
+        total_ms = s["total_us"] / 1000.0
+        avg_us = s["total_us"] / s["calls"] if s["calls"] > 0 else 0
+        min_us = 0 if s["min_us"] > 1e15 else s["min_us"]
+        lines.append(
+            f"  {name:<55} {s['calls']:>8} {total_ms:>12.3f} "
+            f"{avg_us:>12.1f} {min_us:>10.1f} {s['max_us']:>10.1f}"
+        )
+    return "\n".join(lines)
+
+
+def format_timeline(events: list[dict], title: str) -> str:
+    if not events:
+        return f"\n{title}: No timeline events"
+
+    lines = [
+        "",
+        "=" * 110,
+        f"TIMELINE: {title}",
+        "=" * 110,
+        f"  {'Start(ms)':>10}  {'Dur(ms)':>10}  {'End(ms)':>10}  {'Function'}",
+        f"  {'-' * 104}",
+    ]
+
+    # Build a simple nesting stack based on time overlap
+    stack: list[tuple[int, int]] = []  # (start_us, end_us) of active spans
+    for ev in events:
+        s = ev["start_us"]
+        d = ev["duration_us"]
+        e = s + d
+
+        # Pop spans that don't fully contain this event
+        while stack and stack[-1][1] < e:
+            stack.pop()
+
+        depth = len(stack)
+        indent = "  " * depth
+        start_ms = s / 1000.0
+        dur_ms = d / 1000.0
+        end_ms = e / 1000.0
+
+        lines.append(f"  {start_ms:>10.3f}  {dur_ms:>10.3f}  {end_ms:>10.3f}  {indent}{ev['name']}")
+
+        stack.append((s, e))
+
+    return "\n".join(lines)
+
+
+def print_stats(cpp_stats: dict | None, py_stats: dict | None, title: str):
+    print(format_stats_table(merge_stats(cpp_stats, py_stats), title))
+
+
+def print_timeline(cpp_timeline: list | None, py_timeline: list | None, title: str):
+    print(format_timeline(merge_timeline(cpp_timeline, py_timeline), title))
diff --git a/profiler/scenarios.py b/profiler/scenarios.py
new file mode 100644
index 000000000..0cc64bb7e
--- /dev/null
+++ b/profiler/scenarios.py
@@ -0,0 +1,344 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+"""
+Profiling scenarios — each is a self-contained benchmark that yields
+a (title, wall_ms, cpp_stats, py_stats) result.
+
+Scenarios generate their own test data and clean up after themselves.
+Only requires a connection string and a live SQL Server instance.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from mssql_python.connection import Connection
+
+# ---------------------------------------------------------------------------
+# Defaults
+# ---------------------------------------------------------------------------
+ROW_COUNT = 50_000
+EXECUTEMANY_ROWS = 5_000
+FETCHMANY_SIZE = 1_000
+FETCHONE_ROWS = 1_000
+INSERT_COUNT = 100
+COMMIT_ROLLBACK_COUNT = 100
+IMV_ROWS_PER_BATCH = 1_000  # 2000 params, under 2100 limit
+IMV_TOTAL_ROWS = 100_000
+
+_TEST_TABLE = "#perf_test"
+
+_CREATE_TABLE = f"""
+IF OBJECT_ID('tempdb..{_TEST_TABLE}', 'U') IS NOT NULL DROP TABLE {_TEST_TABLE};
+CREATE TABLE {_TEST_TABLE} (
+    id INT IDENTITY(1,1) PRIMARY KEY,
+    int_col INT, bigint_col BIGINT, float_col FLOAT,
+    varchar_col VARCHAR(100), nvarchar_col NVARCHAR(100),
+    date_col DATE, datetime_col DATETIME2,
+    decimal_col DECIMAL(18,4), bit_col BIT
+);
+"""
+
+_INSERT_COLS = (
+    "int_col, bigint_col, float_col, varchar_col, nvarchar_col, "
+    "date_col, datetime_col, decimal_col, bit_col"
+)
+_INSERT_PLACEHOLDERS = "?, ?, ?, ?, ?, ?, ?, ?, ?"
+
+
+def _make_rows(n: int) -> list[tuple]:
+    return [
+        (
+            i,
+            i * 1_000_000,
+            i * 1.5,
+            f"row_{i}_data",
+            f"unicode_row_{i}",
+            "2024-06-15",
+            "2024-06-15 14:30:00.123456",
+            f"{i}.1234",
+            i % 2,
+        )
+        for i in range(n)
+    ]
+
+
+def setup_test_data(conn: "Connection", row_count: int = ROW_COUNT) -> str:
+    cursor = conn.cursor()
+    cursor.execute(_CREATE_TABLE)
+    conn.commit()
+    cursor.executemany(
+        f"INSERT INTO {_TEST_TABLE} ({_INSERT_COLS}) VALUES ({_INSERT_PLACEHOLDERS})",
+        _make_rows(row_count),
+    )
+    conn.commit()
+    cursor.close()
+    return _TEST_TABLE
+
+
+# ---------------------------------------------------------------------------
+# Individual scenarios
+# ---------------------------------------------------------------------------
+
+
+def connect(conn_str: str, ctx) -> dict:
+    from mssql_python import connect as _connect
+
+    ctx.enable()
+    t0 = time.perf_counter()
+    c = _connect(conn_str)
+    wall_ms = (time.perf_counter() - t0) * 1000
+    cpp, py = ctx.collect()
+    c.close()
+    return {"title": "CONNECT", "wall_ms": wall_ms, "cpp": cpp, "py": py}
+
+
+def execute_select(conn, table, ctx) -> dict:
+    cursor = conn.cursor()
+    ctx.enable()
+    t0 = time.perf_counter()
+    cursor.execute(f"SELECT * FROM {table} WHERE id <= 100")
+    rows = cursor.fetchall()
+    wall_ms = (time.perf_counter() - t0) * 1000
+    cpp, py = ctx.collect()
+    cursor.close()
+    return {
+        "title": f"EXECUTE SELECT ({len(rows)} rows)",
+        "wall_ms": wall_ms,
+        "cpp": cpp,
+        "py": py,
+        "detail": f"Rows: {len(rows)}",
+    }
+
+
+def execute_insert(conn, table, ctx, count: int = INSERT_COUNT) -> dict:
+    cursor = conn.cursor()
+    ctx.enable()
+    t0 = time.perf_counter()
+    for i in range(count):
+        cursor.execute(
+            f"INSERT INTO {table} ({_INSERT_COLS}) VALUES ({_INSERT_PLACEHOLDERS})",
+            (
+                i,
+                i * 1000,
+                1.5,
+                f"insert_{i}",
+                f"ins_{i}",
+                "2025-01-01",
+                "2025-01-01 12:00:00",
+                "99.99",
+                1,
+            ),
+        )
+    wall_ms = (time.perf_counter() - t0) * 1000
+    cpp, py = ctx.collect()
+    conn.commit()
+    cursor.close()
+    return {
+        "title": f"EXECUTE INSERT ({count}x)",
+        "wall_ms": wall_ms,
+        "cpp": cpp,
+        "py": py,
+        "detail": f"{count} individual INSERTs",
+    }
+
+
+def executemany(conn, table, ctx, row_count: int = EXECUTEMANY_ROWS) -> dict:
+    cursor = conn.cursor()
+    params = [
+        (i, i * 1000, 1.5, f"batch_{i}", f"b_{i}", "2025-01-01", "2025-01-01 12:00:00", "99.99", 1)
+        for i in range(row_count)
+    ]
+    ctx.enable()
+    t0 = time.perf_counter()
+    cursor.executemany(
+        f"INSERT INTO {table} ({_INSERT_COLS}) VALUES ({_INSERT_PLACEHOLDERS})",
+        params,
+    )
+    wall_ms = (time.perf_counter() - t0) * 1000
+    cpp, py = ctx.collect()
+    conn.commit()
+    cursor.close()
+    return {
+        "title": f"EXECUTEMANY ({row_count} rows)",
+        "wall_ms": wall_ms,
+        "cpp": cpp,
+        "py": py,
+        "detail": f"{row_count} rows via executemany",
+    }
+
+
+def fetchall(conn, table, ctx) -> dict:
+    cursor = conn.cursor()
+    cursor.execute(f"SELECT * FROM {table}")
+    ctx.enable()
+    t0 = time.perf_counter()
+    rows = cursor.fetchall()
+    wall_ms = (time.perf_counter() - t0) * 1000
+    cpp, py = ctx.collect()
+    cursor.close()
+    return {
+        "title": f"FETCHALL ({len(rows)} rows)",
+        "wall_ms": wall_ms,
+        "cpp": cpp,
+        "py": py,
+        "detail": f"Rows: {len(rows)}",
+    }
+
+
+def fetchone(conn, table, ctx, row_count: int = FETCHONE_ROWS) -> dict:
+    cursor = conn.cursor()
+    cursor.execute(f"SELECT TOP {row_count} * FROM {table}")
+    ctx.enable()
+    t0 = time.perf_counter()
+    count = 0
+    while True:
+        row = cursor.fetchone()
+        if row is None:
+            break
+        count += 1
+    wall_ms = (time.perf_counter() - t0) * 1000
+    cpp, py = ctx.collect()
+    cursor.close()
+    return {
+        "title": f"FETCHONE ({count} rows)",
+        "wall_ms": wall_ms,
+        "cpp": cpp,
+        "py": py,
+        "detail": f"Rows: {count}",
+    }
+
+
+def fetchmany(conn, table, ctx, batch_size: int = FETCHMANY_SIZE) -> dict:
+    cursor = conn.cursor()
+    cursor.execute(f"SELECT * FROM {table}")
+    ctx.enable()
+    t0 = time.perf_counter()
+    total = 0
+    while True:
+        batch = cursor.fetchmany(batch_size)
+        if not batch:
+            break
+        total += len(batch)
+    wall_ms = (time.perf_counter() - t0) * 1000
+    cpp, py = ctx.collect()
+    cursor.close()
+    return {
+        "title": f"FETCHMANY ({total} rows, batch={batch_size})",
+        "wall_ms": wall_ms,
+        "cpp": cpp,
+        "py": py,
+        "detail": f"Rows: {total}, Batch size: {batch_size}",
+    }
+
+
+def commit_rollback(conn, ctx, count: int = COMMIT_ROLLBACK_COUNT) -> dict:
+    conn.autocommit = False
+    cursor = conn.cursor()
+    ctx.enable()
+    t0 = time.perf_counter()
+    for _ in range(count):
+        cursor.execute("SELECT 1")
+        conn.commit()
+    for _ in range(count):
+        cursor.execute("SELECT 1")
+        conn.rollback()
+    wall_ms = (time.perf_counter() - t0) * 1000
+    cpp, py = ctx.collect()
+    cursor.close()
+    return {
+        "title": f"COMMIT/ROLLBACK ({count} each)",
+        "wall_ms": wall_ms,
+        "cpp": cpp,
+        "py": py,
+        "detail": f"{count} commits + {count} rollbacks",
+    }
+
+
+def fetch_arrow(conn, table, ctx) -> dict:
+    cursor = conn.cursor()
+    cursor.execute(f"SELECT * FROM {table}")
+    ctx.enable()
+    t0 = time.perf_counter()
+    try:
+        batch = cursor.fetch_arrow(size=ROW_COUNT)
+        wall_ms = (time.perf_counter() - t0) * 1000
+        cpp, py = ctx.collect()
+        row_count = batch.num_rows if batch else 0
+        cursor.close()
+        return {
+            "title": f"FETCH ARROW ({row_count} rows)",
+            "wall_ms": wall_ms,
+            "cpp": cpp,
+            "py": py,
+            "detail": f"Arrow rows: {row_count}",
+        }
+    except Exception as e:
+        ctx.collect()  # drain counters
+        cursor.close()
+        return {
+            "title": "FETCH ARROW",
+            "wall_ms": 0,
+            "cpp": None,
+            "py": None,
+            "detail": f"Skipped: {e}",
+        }
+
+
+def insertmanyvalues(
+    conn, ctx, rows_per_batch: int = IMV_ROWS_PER_BATCH, total_rows: int = IMV_TOTAL_ROWS
+) -> dict:
+    """SQLAlchemy insertmanyvalues pattern: batched multi-row INSERT via cursor.execute()."""
+    num_batches = total_rows // rows_per_batch
+    params_per_call = rows_per_batch * 2
+
+    cursor = conn.cursor()
+    cursor.execute(
+        "IF OBJECT_ID('tempdb..#imv_bench', 'U') IS NOT NULL DROP TABLE #imv_bench;"
+        "CREATE TABLE #imv_bench (id INT, name VARCHAR(50))"
+    )
+    conn.commit()
+
+    sql = "INSERT INTO #imv_bench (id, name) VALUES " + ",".join(["(?, ?)"] * rows_per_batch)
+    params = []
+    for i in range(rows_per_batch):
+        params.extend([i, f"user_{i:06d}"])
+
+    ctx.enable()
+    t0 = time.perf_counter()
+    for _ in range(num_batches):
+        cursor.execute(sql, params)
+    conn.commit()
+    wall_ms = (time.perf_counter() - t0) * 1000
+    cpp, py = ctx.collect()
+    actual = num_batches * rows_per_batch
+    rps = actual / (wall_ms / 1000) if wall_ms > 0 else 0
+    cursor.close()
+    return {
+        "title": f"INSERTMANYVALUES ({actual:,} rows, {params_per_call} params/call)",
+        "wall_ms": wall_ms,
+        "cpp": cpp,
+        "py": py,
+        "detail": f"{actual:,} rows via {num_batches} execute() calls ({rps:,.0f} rows/s)",
+    }
+
+
+# ---------------------------------------------------------------------------
+# Registry
+# ---------------------------------------------------------------------------
+
+# Maps scenario name -> (function, needs_table)
+SCENARIOS: dict[str, tuple] = {
+    "connect": (connect, False),
+    "select": (execute_select, True),
+    "insert": (execute_insert, True),
+    "executemany": (executemany, True),
+    "fetchall": (fetchall, True),
+    "fetchone": (fetchone, True),
+    "fetchmany": (fetchmany, True),
+    "commit_rollback": (commit_rollback, False),
+    "arrow": (fetch_arrow, True),
+    "insertmanyvalues": (insertmanyvalues, False),
+}