diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py
index 4c03b4df..b209cff7 100644
--- a/mssql_python/cursor.py
+++ b/mssql_python/cursor.py
@@ -2574,9 +2574,13 @@ 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 = 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 installed. "
                 "To install, run: pip install mssql_py_core "
@@ -2584,6 +2588,7 @@ def _bulkcopy(
 
         # 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)
@@ -2615,6 +2620,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
@@ -2624,9 +2630,11 @@ def _bulkcopy(
         params = parser._parse(self.connection.connection_str)
 
         if not params.get("server"):
+            logger.error("_bulkcopy: SERVER parameter missing in connection string")
             raise ValueError("SERVER parameter is required in connection string")
 
         if not params.get("database"):
+            logger.error("_bulkcopy: DATABASE parameter missing in connection string")
             raise ValueError(
                 "DATABASE parameter is required in connection string for bulk copy. "
                 "Specify the target database explicitly to avoid accidentally writing to system databases."
@@ -2688,9 +2696,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 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),
@@ -2703,8 +2716,17 @@ def _bulkcopy(
                 keep_nulls=keep_nulls,
                 fire_triggers=fire_triggers,
                 use_internal_transaction=use_internal_transaction,
+                python_logger=logger if is_logging else None,  # Only pass logger if enabled
             )
 
+            if is_logging:
+                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
 
         except Exception as e:
@@ -2731,8 +2753,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__,
diff --git a/mssql_python/logging.py b/mssql_python/logging.py
index 2cb9361f..8579d2cd 100644
--- a/mssql_python/logging.py
+++ b/mssql_python/logging.py
@@ -88,7 +88,7 @@ def __init__(self):
 
         # Create the underlying Python logger
         self._logger = logging.getLogger("mssql_python")
-        self._logger.setLevel(logging.CRITICAL)  # Disabled by default
+        self._logger.setLevel(logging.WARNING)  # Allow WARNING and ERROR by default
         self._logger.propagate = False  # Don't propagate to root logger
 
         # Add trace ID filter (injects thread_id into every log record)
@@ -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)
diff --git a/tests/test_007_logging.py b/tests/test_007_logging.py
index 6bc0f552..daafc6fc 100644
--- a/tests/test_007_logging.py
+++ b/tests/test_007_logging.py
@@ -31,6 +31,8 @@ def cleanup_logger():
 
     # Disable logging and clear handlers
     logger._logger.setLevel(logging.CRITICAL)
+    logger._cached_level = logging.CRITICAL  # Update cached level
+    logger._is_debug_enabled = False  # Update cached debug flag
     for handler in logger._logger.handlers[:]:
         handler.close()
         logger._logger.removeHandler(handler)
@@ -46,6 +48,8 @@ def cleanup_logger():
 
     # Restore state and cleanup
     logger._logger.setLevel(logging.CRITICAL)
+    logger._cached_level = logging.CRITICAL  # Update cached level
+    logger._is_debug_enabled = False  # Update cached debug flag
     for handler in logger._logger.handlers[:]:
         handler.close()
         logger._logger.removeHandler(handler)
@@ -843,3 +847,190 @@ def __str__(self):
 
         # Application should still be running
         assert True
+
+
+class TestDebugEnabledProperty:
+    """Test is_debug_enabled property for performance-optimized level checks"""
+
+    def test_is_debug_enabled_false_by_default(self, cleanup_logger):
+        """is_debug_enabled should be False when logger is at CRITICAL level (cleanup default)"""
+        # cleanup_logger sets level to CRITICAL
+        assert logger.getLevel() == logging.CRITICAL
+        assert not logger.is_debug_enabled
+
+    def test_is_debug_enabled_true_when_debug_set(self, cleanup_logger):
+        """is_debug_enabled should be True when DEBUG level is set"""
+        setup_logging()
+        assert logger.getLevel() == logging.DEBUG
+        assert logger.is_debug_enabled
+
+    def test_is_debug_enabled_false_when_info_set(self, cleanup_logger):
+        """is_debug_enabled should be False when INFO level is set"""
+        # Set to INFO level (above DEBUG)
+        logger._setLevel(logging.INFO)
+        assert logger.getLevel() == logging.INFO
+        assert not logger.is_debug_enabled
+
+    def test_is_debug_enabled_updates_with_level_change(self, cleanup_logger):
+        """is_debug_enabled should update when logging level changes"""
+        # Start disabled
+        assert not logger.is_debug_enabled
+
+        # Enable debug
+        setup_logging()
+        assert logger.is_debug_enabled
+
+        # Disable again
+        logger._setLevel(logging.WARNING)
+        assert not logger.is_debug_enabled
+
+        # Re-enable
+        logger._setLevel(logging.DEBUG)
+        assert logger.is_debug_enabled
+
+    def test_is_debug_enabled_matches_cached_level(self, cleanup_logger):
+        """is_debug_enabled should match the cached level check"""
+        # Test at various levels
+        test_levels = [
+            logging.DEBUG,
+            logging.INFO,
+            logging.WARNING,
+            logging.ERROR,
+            logging.CRITICAL,
+        ]
+
+        for level in test_levels:
+            logger._setLevel(level)
+            expected = level <= logging.DEBUG
+            assert logger.is_debug_enabled == expected, f"Failed at level {level}"
+
+
+class TestPyCoreLog:
+    """Test py_core_log method for Rust/py-core integration"""
+
+    def test_py_core_log_with_custom_location(self, cleanup_logger, temp_log_dir):
+        """py_core_log should create log record with custom source location"""
+        log_file = os.path.join(temp_log_dir, "pycore_test.log")
+        setup_logging(output=FILE, log_file_path=log_file)
+
+        # Log from "Rust" code
+        logger.py_core_log(
+            logging.INFO, "Bulkcopy operation started", filename="cursor.rs", lineno=270
+        )
+
+        # Verify log file contains custom source location
+        with open(log_file, "r") as f:
+            content = f.read()
+            assert "cursor.rs:270" in content
+            assert "py-core" in content
+            assert "Bulkcopy operation started" in content
+
+    def test_py_core_log_respects_level(self, cleanup_logger, temp_log_dir):
+        """py_core_log should respect cached level check"""
+        log_file = os.path.join(temp_log_dir, "pycore_level_test.log")
+        setup_logging(output=FILE, log_file_path=log_file)
+
+        # Log at DEBUG level (should appear)
+        logger.py_core_log(logging.DEBUG, "Debug message", "test.rs", 1)
+
+        # Disable debug logging
+        logger._setLevel(logging.WARNING)
+
+        # Log at DEBUG level (should NOT appear)
+        logger.py_core_log(logging.DEBUG, "Hidden debug message", "test.rs", 2)
+
+        # Log at WARNING level (should appear)
+        logger.py_core_log(logging.WARNING, "Warning message", "test.rs", 3)
+
+        # Verify file content
+        with open(log_file, "r") as f:
+            content = f.read()
+            assert "Debug message" in content
+            assert "Hidden debug message" not in content
+            assert "Warning message" in content
+
+    def test_py_core_log_allows_warnings_when_disabled(self, cleanup_logger, temp_log_dir):
+        """py_core_log should allow WARNING/ERROR even when logging disabled"""
+        log_file = os.path.join(temp_log_dir, "pycore_warning_test.log")
+        setup_logging(output=FILE, log_file_path=log_file)
+
+        # Disable debug logging (but WARNING should still work)
+        logger._setLevel(logging.CRITICAL)
+
+        # These should NOT appear
+        logger.py_core_log(logging.DEBUG, "Debug blocked", "test.rs", 1)
+        logger.py_core_log(logging.INFO, "Info blocked", "test.rs", 2)
+
+        # These SHOULD appear (WARNING and ERROR always allowed)
+        logger.py_core_log(logging.WARNING, "Warning allowed", "test.rs", 3)
+        logger.py_core_log(logging.ERROR, "Error allowed", "test.rs", 4)
+
+        # Verify file content
+        with open(log_file, "r") as f:
+            content = f.read()
+            assert "Debug blocked" not in content
+            assert "Info blocked" not in content
+            assert "Warning allowed" in content
+            assert "Error allowed" in content
+
+    def test_py_core_log_fallback_on_error(self, cleanup_logger, temp_log_dir):
+        """py_core_log should fallback gracefully on errors"""
+        log_file = os.path.join(temp_log_dir, "pycore_fallback_test.log")
+        setup_logging(output=FILE, log_file_path=log_file)
+
+        # Should not crash even with invalid parameters
+        logger.py_core_log(logging.INFO, "Test message", None, None)  # None filename/lineno
+        logger.py_core_log(logging.INFO, "Test 2", "", -1)  # Empty filename, negative line
+
+        # Application should still be running
+        assert True
+
+        # Verify something was logged (fallback worked)
+        with open(log_file, "r") as f:
+            content = f.read()
+            # At least one message should have been logged
+            assert len(content) > 0
+
+    def test_py_core_log_formats_as_csv(self, cleanup_logger, temp_log_dir):
+        """py_core_log output should follow CSV format"""
+        log_file = os.path.join(temp_log_dir, "pycore_csv_test.log")
+        setup_logging(output=FILE, log_file_path=log_file)
+
+        logger.py_core_log(logging.INFO, "Test message with commas, quotes", "connection.rs", 42)
+
+        # Read and verify CSV format
+        with open(log_file, "r") as f:
+            lines = f.readlines()
+            # Skip header lines (starts with #)
+            data_lines = [line for line in lines if not line.startswith("#")]
+
+            # Should have at least 2 lines: CSV header + data
+            assert len(data_lines) >= 2
+
+            # Check data line format: timestamp, thread_id, level, location, source, message
+            data_line = data_lines[-1]  # Get last logged line
+            parts = [p.strip() for p in data_line.split(",")]
+
+            # Should have 6 parts: timestamp, thread_id, level, location, source, message
+            assert len(parts) >= 6
+            assert "connection.rs:42" in data_line
+            assert "py-core" in data_line
+            assert "INFO" in data_line
+
+    def test_py_core_log_with_different_rust_files(self, cleanup_logger, temp_log_dir):
+        """py_core_log should handle different Rust source files"""
+        log_file = os.path.join(temp_log_dir, "pycore_files_test.log")
+        setup_logging(output=FILE, log_file_path=log_file)
+
+        # Log from different Rust files
+        logger.py_core_log(logging.INFO, "Connection msg", "connection.rs", 100)
+        logger.py_core_log(logging.INFO, "Cursor msg", "cursor.rs", 200)
+        logger.py_core_log(logging.INFO, "Auth msg", "auth.rs", 300)
+
+        # Verify all source files appear in log
+        with open(log_file, "r") as f:
+            content = f.read()
+            assert "connection.rs:100" in content
+            assert "cursor.rs:200" in content
+            assert "auth.rs:300" in content
+            assert content.count("py-core") == 3  # Three entries from py-core