FEAT: mssql-python driver changes to support bulk copy logging. (#430)

### Work Item / Issue Reference  
<!-- 
IMPORTANT: Please follow the PR template guidelines below.
For mssql-python maintainers: Insert your ADO Work Item ID below 
For external contributors: Insert Github Issue number below
Only one reference is required - either GitHub issue OR ADO Work Item.
-->

<!-- mssql-python maintainers: ADO Work Item -->
AB#42113

-------------------------------------------------------------------
### Summary   
This pull request introduces significant improvements to the logging
system in the `mssql_python` package, focusing on performance
optimization, better integration with Rust (`py-core`), and enhanced
test coverage. The changes optimize how logging levels are checked,
ensure that logging from Rust code is handled correctly, and add
thorough tests for these new features.

**Key changes:**

### Logging Performance and Behavior Improvements
- Switched the default logging level from `CRITICAL` to `WARNING`,
ensuring that warnings and errors are visible by default without
explicit configuration. Also, a default stderr handler is now added for
warnings and errors.
[[1]](diffhunk://#diff-a8f9a63854a2b8541d3d9260e97948990dba646ca62b8e0d135055fd7f543f28L91-R91)
[[2]](diffhunk://#diff-a8f9a63854a2b8541d3d9260e97948990dba646ca62b8e0d135055fd7f543f28L107-R122)
- Introduced cached logging level checks (`_cached_level` and
`_is_debug_enabled`) for fast, low-overhead decision-making in logging
calls, with a new `is_debug_enabled` property for quick checks.
[[1]](diffhunk://#diff-a8f9a63854a2b8541d3d9260e97948990dba646ca62b8e0d135055fd7f543f28L107-R122)
[[2]](diffhunk://#diff-a8f9a63854a2b8541d3d9260e97948990dba646ca62b8e0d135055fd7f543f28R505-R515)
[[3]](diffhunk://#diff-a8f9a63854a2b8541d3d9260e97948990dba646ca62b8e0d135055fd7f543f28R621-R625)
- Updated all logging calls in `_bulkcopy` (in `cursor.py`) to use the
new fast check for debug logging, and to avoid unnecessary logger
passing to Rust unless logging is enabled.
[[1]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280R2577-R2591)
[[2]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280L2691-R2706)
[[3]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280R2719-R2727)

### Rust (`py-core`) Logging Integration
- Added a new `py_core_log` method to the logger, allowing Rust-side
code to emit logs with custom source locations and levels, and ensuring
these logs are formatted and filtered consistently with Python logs.
- Enhanced the CSV log formatter to recognize and format logs
originating from `py-core` distinctly.

### Robustness and Error Handling
- Ensured that logging always allows `WARNING` and `ERROR` messages,
even when the logger is otherwise disabled, both in Python and
Rust-originated logs.
[[1]](diffhunk://#diff-a8f9a63854a2b8541d3d9260e97948990dba646ca62b8e0d135055fd7f543f28L355-R416)
[[2]](diffhunk://#diff-a8f9a63854a2b8541d3d9260e97948990dba646ca62b8e0d135055fd7f543f28R349-R386)
- Improved error logging in `_bulkcopy` for various failure cases, such
as missing dependencies or invalid parameters.
[[1]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280R2577-R2591)
[[2]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280R2623)
[[3]](diffhunk://#diff-deceea46ae01082ce8400e14fa02f4b7585afb7b5ed9885338b66494f5f38280R2633-R2637)

### Test Suite Enhancements
- Added comprehensive tests for the new `is_debug_enabled` property and
for `py_core_log`, covering level filtering, custom source locations,
CSV formatting, error fallback, and integration with different Rust
files.
- Updated test cleanup utilities to maintain consistency with the new
cached logging level logic.
[[1]](diffhunk://#diff-39a609c16b2d72fabefaf23a0f99dccae8bd4bf5e481928bdeaf53d6131a0fd4R34-R35)
[[2]](diffhunk://#diff-39a609c16b2d72fabefaf23a0f99dccae8bd4bf5e481928bdeaf53d6131a0fd4R51-R52)

These changes collectively improve logging performance, reliability, and
cross-language integration, while ensuring robust test coverage for the
new features.
**High Level Design**
<img width="426" height="986" alt="image"
src="https://github.com/user-attachments/assets/be4ddc70-7974-484e-9ad0-22e17321e16d"
/>


**Success Test Log**
<img width="1790" height="952" alt="image"
src="https://github.com/user-attachments/assets/073a07e1-3235-4587-a747-274e5aa06113"
/>

**Failure Test Log**
<img width="1887" height="957" alt="image"
src="https://github.com/user-attachments/assets/8d39e902-1f13-4cce-a252-2dc12deb94ac"
/>

Author: subrata-ms

mssql_python/cursor.py

@@ -2575,16 +2575,21 @@ def bulkcopy(
             ValueError: If table_name is empty or parameters are invalid
             RuntimeError: If connection string is not available
         """
+        # Fast check if logging is enabled to avoid overhead
+        is_logging_enabled = logger.is_debug_enabled
+
         try:
             import mssql_py_core
         except ImportError as exc:
+            logger.error("_bulkcopy: Failed to import mssql_py_core module")
             raise ImportError(
                 "Bulk copy requires the mssql_py_core library which is not available. "
                 "This is an unexpected error. "
             ) from exc
 
         # Validate inputs
         if not table_name or not isinstance(table_name, str):
+            logger.error("_bulkcopy: Invalid table_name parameter")
             raise ValueError("table_name must be a non-empty string")
 
         # Validate that data is iterable (but not a string or bytes, which are technically iterable)
@@ -2616,6 +2621,7 @@ def bulkcopy(
 
         # Get and parse connection string
         if not hasattr(self.connection, "connection_str"):
+            logger.error("_bulkcopy: Connection string not available")
             raise RuntimeError("Connection string not available for bulk copy")
 
         # Use the proper connection string parser that handles braced values
@@ -2652,9 +2658,14 @@ def bulkcopy(
         pycore_connection = None
         pycore_cursor = None
         try:
-            pycore_connection = mssql_py_core.PyCoreConnection(pycore_context)
+            # Only pass logger to Rust if logging is enabled (performance optimization)
+            pycore_connection = mssql_py_core.PyCoreConnection(
+                pycore_context, python_logger=logger if is_logging_enabled else None
+            )
             pycore_cursor = pycore_connection.cursor()
 
+            # Call bulkcopy with explicit keyword arguments
+            # The API signature: bulkcopy(table_name, data_source, batch_size=0, timeout=30, ...)
             result = pycore_cursor.bulkcopy(
                 table_name,
                 iter(data),
@@ -2667,6 +2678,14 @@ def bulkcopy(
                 keep_nulls=keep_nulls,
                 fire_triggers=fire_triggers,
                 use_internal_transaction=use_internal_transaction,
+                python_logger=logger if is_logging_enabled else None,  # Only pass logger if enabled
+            )
+
+            logger.info(
+                "_bulkcopy: Bulk copy completed successfully - rows_copied=%s, batch_count=%s, elapsed_time=%s",
+                result.get("rows_copied", "N/A"),
+                result.get("batch_count", "N/A"),
+                result.get("elapsed_time", "N/A"),
             )
 
             return result
@@ -2694,8 +2713,7 @@ def bulkcopy(
                     try:
                         resource.close()
                     except Exception as cleanup_error:
-                        # Log cleanup errors at debug level to aid troubleshooting
-                        # without masking the original exception
+                        # Log cleanup errors only - aids troubleshooting without masking original exception
                         logger.debug(
                             "Failed to close bulk copy resource %s: %s",
                             type(resource).__name__,

mssql_python/logging.py

@@ -104,7 +104,22 @@ def __init__(self):
         self._handler_lock = threading.RLock()  # Reentrant lock for handler operations
         self._cleanup_registered = False  # Track if atexit cleanup is registered
 
-        # Don't setup handlers yet - do it lazily when setLevel is called
+        # Cached level for fast checks (avoid repeated isEnabledFor calls)
+        self._cached_level = logging.WARNING
+        self._is_debug_enabled = False
+
+        # Set up default stderr handler for WARNING and ERROR messages
+        # This ensures warnings are always visible even when logging is not enabled
+        import sys
+
+        default_handler = logging.StreamHandler(sys.stderr)
+        default_handler.setLevel(logging.WARNING)
+        # Simple format for warnings - no CSV formatting needed
+        default_handler.setFormatter(logging.Formatter("[%(name)s] %(levelname)s: %(message)s"))
+        self._logger.addHandler(default_handler)
+        self._default_handler = default_handler  # Keep reference for later removal
+
+        # Don't setup full handlers yet - do it lazily when setLevel is called
         # This prevents creating log files when user changes output mode before enabling logging
 
     def _setup_handlers(self):
@@ -145,15 +160,20 @@ def _setup_handlers(self):
         # Custom formatter to extract source from message and format as CSV
         class CSVFormatter(logging.Formatter):
             def format(self, record):
-                # Extract source from message (e.g., [Python] or [DDBC])
-                msg = record.getMessage()
-                if msg.startswith("[") and "]" in msg:
-                    end_bracket = msg.index("]")
-                    source = msg[1:end_bracket]
-                    message = msg[end_bracket + 2 :].strip()  # Skip '] '
+                # Check if this is from py-core (via py_core_log method)
+                if hasattr(record, "funcName") and record.funcName == "py-core":
+                    source = "py-core"
+                    message = record.getMessage()
                 else:
-                    source = "Unknown"
-                    message = msg
+                    # Extract source from message (e.g., [Python] or [DDBC])
+                    msg = record.getMessage()
+                    if msg.startswith("[") and "]" in msg:
+                        end_bracket = msg.index("]")
+                        source = msg[1:end_bracket]
+                        message = msg[end_bracket + 2 :].strip()  # Skip '] '
+                    else:
+                        source = "Unknown"
+                        message = msg
 
                 # Format timestamp with milliseconds using period separator
                 timestamp = self.formatTime(record, "%Y-%m-%d %H:%M:%S")
@@ -326,6 +346,44 @@ def _write_log_header(self):
                 pass  # Even stderr notification failed
             # Don't crash - logging continues without header
 
+    def py_core_log(self, level: int, msg: str, filename: str = "cursor.rs", lineno: int = 0):
+        """
+        Logging method for py-core (Rust/TDS) code with custom source location.
+
+        Args:
+            level: Log level (DEBUG, INFO, WARNING, ERROR)
+            msg: Message string (already formatted)
+            filename: Source filename (e.g., 'cursor.rs')
+            lineno: Line number in source file
+        """
+        try:
+            # Fast level check using cached level (same optimization as _log method)
+            # Exception: Always allow WARNING and ERROR messages through
+            if level < self._cached_level and level < logging.WARNING:
+                return
+
+            # Create a custom LogRecord with Rust source location
+            import logging as log_module
+
+            record = log_module.LogRecord(
+                name=self._logger.name,
+                level=level,
+                pathname=filename,
+                lineno=lineno,
+                msg=msg,
+                args=(),
+                exc_info=None,
+                func="py-core",
+                sinfo=None,
+            )
+            self._logger.handle(record)
+        except Exception:
+            # Fallback - use regular logging
+            try:
+                self._logger.log(level, msg)
+            except:
+                pass
+
     def _log(self, level: int, msg: str, add_prefix: bool = True, *args, **kwargs):
         """
         Internal logging method with exception safety.
@@ -352,8 +410,10 @@ def _log(self, level: int, msg: str, add_prefix: bool = True, *args, **kwargs):
             All other failures are silently ignored to prevent app crashes.
         """
         try:
-            # Fast level check (zero overhead if disabled)
-            if not self._logger.isEnabledFor(level):
+            # Fast level check using cached level (zero overhead if disabled)
+            # This avoids the overhead of isEnabledFor() method call
+            # Exception: Always allow WARNING and ERROR messages through
+            if level < self._cached_level and level < logging.WARNING:
                 return
 
             # Add prefix if requested (only after level check)
@@ -364,8 +424,9 @@ def _log(self, level: int, msg: str, add_prefix: bool = True, *args, **kwargs):
             if args:
                 msg = msg % args
 
-            # Log the message (no args since already formatted)
-            self._logger.log(level, msg, **kwargs)
+            # Log the message with proper stack level to capture caller's location
+            # stacklevel=3 skips: _log -> debug/info/warning/error -> actual caller
+            self._logger.log(level, msg, stacklevel=3, **kwargs)
         except Exception:
             # Last resort: Try stderr fallback for any logging failure
             # This helps diagnose critical issues (disk full, permission denied, etc.)
@@ -441,6 +502,17 @@ def _setLevel(
         # Set level (atomic operation, no lock needed)
         self._logger.setLevel(level)
 
+        # Cache level for fast checks (avoid repeated isEnabledFor calls)
+        # Note: These updates are not atomic across both variables, creating a brief
+        # window where reads might see inconsistent state (e.g., updated _cached_level
+        # but old _is_debug_enabled). This is an acceptable benign race condition:
+        # - Worst case: one log statement might be incorrectly allowed/blocked
+        # - Duration: nanoseconds (single Python bytecode instruction gap)
+        # - Impact: negligible - next check will see consistent state
+        # - Alternative (locking) would add overhead to every log call
+        self._cached_level = level
+        self._is_debug_enabled = level <= logging.DEBUG
+
         # Notify C++ bridge of level change
         self._notify_cpp_level_change(level)
 
@@ -546,6 +618,11 @@ def level(self) -> int:
         """Get the current logging level"""
         return self._logger.level
 
+    @property
+    def is_debug_enabled(self) -> bool:
+        """Fast check if debug logging is enabled (cached for performance)"""
+        return self._is_debug_enabled
+
 
 # ============================================================================
 # Module-level exports (Primary API)