diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index dcb5077..00d8edf 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -7,9 +7,6 @@ on:
     branches: [ main ]
   workflow_call:
 
-env:
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true
-
 jobs:
   test:
     runs-on: ubuntu-latest
@@ -46,7 +43,7 @@ jobs:
       uses: actions/cache@v5
       with:
         path: ~/.cache/pip
-        key: ${{ runner.os }}-py${{ matrix.python-version }}-mongo${{ matrix.mongodb-version }}-sqlalchemy-${{ matrix.sqlalchemy-version }}-pip-${{ hashFiles('**/requirements-test.txt', 'pyproject.toml') }}
+        key: ${{ runner.os }}-py${{ matrix.python-version }}-mongo${{ matrix.mongodb-version }}-sqlalchemy-${{ matrix.sqlalchemy-version }}-pip-${{ hashFiles('**/requirements-test.txt', '**/requirements-test-sqlalchemy14.txt', '**/requirements-test-sqlalchemy2.txt', 'pyproject.toml') }}
         restore-keys: |
           ${{ runner.os }}-py${{ matrix.python-version }}-mongo${{ matrix.mongodb-version }}-sqlalchemy-${{ matrix.sqlalchemy-version }}-pip-
 
@@ -60,10 +57,12 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        # Install the target SQLAlchemy version for this matrix entry first to ensure
-        # tests run against both 1.x and 2.x series.
-        pip install "SQLAlchemy==${{ matrix.sqlalchemy-version }}"
         pip install -r requirements-test.txt
+        if [ "${{ matrix.sqlalchemy-version }}" = "1.4.*" ]; then
+          pip install -r requirements-test-sqlalchemy14.txt
+        else
+          pip install -r requirements-test-sqlalchemy2.txt
+        fi
         pip install black isort
 
     - name: Wait for MongoDB to be ready
diff --git a/pymongosql/sqlalchemy_mongodb/sqlalchemy_dialect.py b/pymongosql/sqlalchemy_mongodb/sqlalchemy_dialect.py
index 8c89cbf..16f2c06 100644
--- a/pymongosql/sqlalchemy_mongodb/sqlalchemy_dialect.py
+++ b/pymongosql/sqlalchemy_mongodb/sqlalchemy_dialect.py
@@ -214,6 +214,37 @@ def __getattribute__(self, name):
             return pymongosql
         return super().__getattribute__(name)
 
+    @staticmethod
+    def _normalize_collection_name(statement: str) -> str:
+        """Extract a collection name from the compiler's DROP COLLECTION placeholder."""
+        collection_name = statement[len("-- DROP COLLECTION ") :].strip()
+        if collection_name.startswith('"') and collection_name.endswith('"'):
+            return collection_name[1:-1]
+        return collection_name
+
+    def _handle_ddl_placeholder(self, cursor, statement: str) -> bool:
+        """Handle SQLAlchemy DDL placeholders without routing them through the SQL parser."""
+        if statement == "-- Collection will be created on first insert":
+            return True
+
+        if statement.startswith("-- DROP COLLECTION "):
+            cursor.connection.database.drop_collection(self._normalize_collection_name(statement))
+            return True
+
+        return False
+
+    def do_execute(self, cursor, statement, parameters, context=None):
+        """Execute statements, handling MongoDB DDL placeholders directly."""
+        if self._handle_ddl_placeholder(cursor, statement):
+            return None
+        return super().do_execute(cursor, statement, parameters, context=context)
+
+    def do_execute_no_params(self, cursor, statement, context=None):
+        """Execute parameterless statements, handling MongoDB DDL placeholders directly."""
+        if self._handle_ddl_placeholder(cursor, statement):
+            return None
+        return super().do_execute_no_params(cursor, statement, context=context)
+
     def create_connect_args(self, url: url.URL) -> Tuple[List[Any], Dict[str, Any]]:
         """Create connection arguments from SQLAlchemy URL.
 
diff --git a/pyproject.toml b/pyproject.toml
index d78614b..8e5f8ce 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -41,6 +41,15 @@ dependencies = [
 [project.optional-dependencies]
 retry = ["tenacity>=9.0.0"]
 sqlalchemy = ["sqlalchemy>=1.4.0"]
+pandas_sqlalchemy14 = [
+    "sqlalchemy>=1.4.0,<2.0.0",
+    "pandas>=2.1.0,<2.2.0; python_version < '3.13'",
+]
+pandas_sqlalchemy2 = [
+    "sqlalchemy>=2.0.0,<3.0.0",
+    "pandas>=3.0.0,<4.0.0; python_version >= '3.11'",
+    "pandas>=2.2.0,<3.0.0; python_version < '3.11'",
+]
 dev = [
     "pytest>=7.0.0",
     "pytest-cov>=4.0.0", 
diff --git a/requirements-test-sqlalchemy14.txt b/requirements-test-sqlalchemy14.txt
new file mode 100644
index 0000000..d43e243
--- /dev/null
+++ b/requirements-test-sqlalchemy14.txt
@@ -0,0 +1,7 @@
+-r requirements-test.txt
+
+# Supported compatibility pair for the SQLAlchemy 1.4 test lane.
+SQLAlchemy>=1.4.0,<2.0.0
+# pandas>=2.2 has SQLAlchemy 1.4 engine compatibility issues in read_sql,
+# while pandas 2.1.x does not provide wheels for Python >=3.13.
+pandas>=2.1.0,<2.2.0; python_version < "3.13"
\ No newline at end of file
diff --git a/requirements-test-sqlalchemy2.txt b/requirements-test-sqlalchemy2.txt
new file mode 100644
index 0000000..0e3d23a
--- /dev/null
+++ b/requirements-test-sqlalchemy2.txt
@@ -0,0 +1,8 @@
+-r requirements-test.txt
+
+# Supported compatibility for the SQLAlchemy 2.x test lane.
+SQLAlchemy>=2.0.0,<3.0.0
+# pandas 3 is not available on Python < 3.11,
+# so keep a resolver-safe fallback for those interpreters.
+pandas>=3.0.0,<4.0.0; python_version >= "3.11"
+pandas>=2.2.0,<3.0.0; python_version < "3.11"
diff --git a/requirements-test.txt b/requirements-test.txt
index 7b57803..b353f4a 100644
--- a/requirements-test.txt
+++ b/requirements-test.txt
@@ -1,6 +1,10 @@
 -r requirements.txt
 -r requirements-optional.txt
 
+# Shared test tooling only. Version-coupled pandas/SQLAlchemy combinations live in
+# requirements-test-sqlalchemy14.txt and requirements-test-sqlalchemy2.txt.
+# This keeps local and CI installs explicit about which compatibility pair they use.
+
 # Test dependencies
 pytest>=7.0.0
 pytest-cov>=4.0.0
diff --git a/tests/conftest.py b/tests/conftest.py
index dc50af4..e6a8e5f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -115,6 +115,14 @@ class Base(DeclarativeBase):
     Base = None
     Session = None
 
+try:
+    import pandas as pd
+
+    HAS_PANDAS = True
+except ImportError:
+    pd = None
+    HAS_PANDAS = False
+
 # SQLAlchemy fixtures for dialect testing
 if HAS_SQLALCHEMY:
 
diff --git a/tests/test_pandas_sqlalchemy.py b/tests/test_pandas_sqlalchemy.py
new file mode 100644
index 0000000..1f52351
--- /dev/null
+++ b/tests/test_pandas_sqlalchemy.py
@@ -0,0 +1,103 @@
+# -*- coding: utf-8 -*-
+import pytest
+
+from tests.conftest import HAS_PANDAS, HAS_SQLALCHEMY
+
+if HAS_PANDAS:
+    import pandas as pd
+
+
+pytestmark = pytest.mark.skipif(
+    not (HAS_SQLALCHEMY and HAS_PANDAS),
+    reason="Pandas and SQLAlchemy are required for pandas SQLAlchemy integration tests",
+)
+
+
+class TestPandasSQLAlchemy:
+    """Compatibility tests for pandas through the SQLAlchemy engine."""
+
+    TEST_COLLECTION = "test_pandas_sqlalchemy"
+
+    @pytest.fixture(autouse=True)
+    def setup_teardown(self, conn):
+        """Keep pandas write tests isolated from shared seeded collections."""
+        db = conn.database
+        if self.TEST_COLLECTION in db.list_collection_names():
+            db.drop_collection(self.TEST_COLLECTION)
+        yield
+        if self.TEST_COLLECTION in db.list_collection_names():
+            db.drop_collection(self.TEST_COLLECTION)
+
+    def test_read_sql_returns_dataframe(self, sqlalchemy_engine):
+        """pandas.read_sql should load a projected query into a DataFrame."""
+        query = "SELECT _id, name, age FROM users LIMIT 5"
+
+        dataframe = pd.read_sql(query, sqlalchemy_engine)
+
+        assert isinstance(dataframe, pd.DataFrame)
+        assert not dataframe.empty
+        assert list(dataframe.columns) == ["_id", "name", "age"]
+        assert len(dataframe) <= 5
+        assert dataframe["_id"].notna().all()
+        assert dataframe["name"].notna().all()
+
+    def test_to_sql_append_writes_rows(self, sqlalchemy_engine, conn):
+        """pandas should support read_sql, iloc selection, and to_sql append."""
+        db = conn.database
+        source = pd.read_sql(
+            "SELECT _id, name, age, city, active FROM users LIMIT 5",
+            sqlalchemy_engine,
+        )
+        selected = source.iloc[[1, 3]].copy()
+        selected["source_collection"] = "users"
+
+        selected.to_sql(self.TEST_COLLECTION, sqlalchemy_engine, if_exists="append", index=False)
+
+        docs = list(
+            db[self.TEST_COLLECTION]
+            .find({}, {"_id": 1, "name": 1, "age": 1, "active": 1, "source_collection": 1})
+            .sort("name", 1)
+        )
+
+        expected = (
+            selected[["_id", "name", "age", "active", "source_collection"]].sort_values("name").to_dict("records")
+        )
+
+        assert len(docs) == len(expected)
+        assert docs == expected
+
+        round_trip = pd.read_sql(
+            f"SELECT _id, name, age, active, source_collection FROM {self.TEST_COLLECTION}",
+            sqlalchemy_engine,
+        )
+        round_trip = round_trip.sort_values("name").reset_index(drop=True)
+        expected_frame = (
+            selected[["_id", "name", "age", "active", "source_collection"]].sort_values("name").reset_index(drop=True)
+        )
+
+        assert isinstance(round_trip, pd.DataFrame)
+        assert list(round_trip.columns) == ["_id", "name", "age", "active", "source_collection"]
+        assert round_trip.to_dict("records") == expected_frame.to_dict("records")
+
+    def test_to_sql_replace_recreates_collection(self, sqlalchemy_engine, conn):
+        """pandas.to_sql with replace should drop and recreate the collection transparently."""
+        db = conn.database
+        db[self.TEST_COLLECTION].insert_one({"_id": "stale", "name": "stale"})
+
+        replacement = pd.read_sql(
+            "SELECT _id, name, age, active FROM users LIMIT 2",
+            sqlalchemy_engine,
+        ).copy()
+        replacement["source_collection"] = "users"
+
+        replacement.to_sql(self.TEST_COLLECTION, sqlalchemy_engine, if_exists="replace", index=False)
+
+        docs = list(
+            db[self.TEST_COLLECTION]
+            .find({}, {"_id": 1, "name": 1, "age": 1, "active": 1, "source_collection": 1})
+            .sort("_id", 1)
+        )
+        expected = replacement.sort_values("_id").reset_index(drop=True)
+
+        assert len(docs) == len(expected)
+        assert docs == expected.to_dict("records")
diff --git a/tests/test_sqlalchemy_dialect.py b/tests/test_sqlalchemy_dialect.py
index a89ddbf..e2fe7b2 100644
--- a/tests/test_sqlalchemy_dialect.py
+++ b/tests/test_sqlalchemy_dialect.py
@@ -392,6 +392,20 @@ def test_ddl_compiler(self):
         drop_result = PyMongoSQLDDLCompiler.visit_drop_table(mock_compiler, drop_mock)
         self.assertIn("DROP COLLECTION", drop_result)
 
+    def test_ddl_placeholder_handlers(self):
+        """Test that SQLAlchemy DDL placeholders are handled without SQL parsing."""
+        mock_cursor = Mock()
+        mock_cursor.connection.database.drop_collection = Mock()
+
+        self.assertTrue(
+            self.dialect._handle_ddl_placeholder(mock_cursor, "-- Collection will be created on first insert")
+        )
+        mock_cursor.connection.database.drop_collection.assert_not_called()
+
+        handled = self.dialect._handle_ddl_placeholder(mock_cursor, "-- DROP COLLECTION test_table")
+        self.assertTrue(handled)
+        mock_cursor.connection.database.drop_collection.assert_called_once_with("test_table")
+
 
 class TestSQLAlchemyIntegration(unittest.TestCase):
     """Integration tests for SQLAlchemy functionality."""