Address PR review: fix predicate TypeError, placeholder style, connection doc, system test conn kwarg, SSL naming

Author: Amit Kumar

providers/ibm/db2/README.rst

@@ -52,13 +52,14 @@ To use this provider, you need to configure a Db2 connection in Airflow:
    - **Login**: Db2 username
    - **Password**: Db2 password
    - **Port**: Db2 port (default: 50000)
-   - **Extra**: Optional JSON with additional parameters:
+   - **Extra**: Optional JSON with additional parameters (keys are
+     converted to uppercase by the hook):
 
      .. code-block:: json
 
         {
-          "ssl": true,
-          "sslcert": "/path/to/cert.pem"
+          "SECURITY": "SSL",
+          "SSLServerCertificate": "/path/to/cert.crt"
         }
 
 Usage
@@ -72,7 +73,7 @@ Using the Db2Hook
     from airflow.providers.ibm.db2.hooks.db2 import Db2Hook
 
     hook = Db2Hook(db2_conn_id="db2_default")
-    records = hook.get_records("SELECT * FROM employees WHERE dept = %s", parameters=("IT",))
+    records = hook.get_records("SELECT * FROM employees WHERE dept = ?", parameters=("IT",))
 
 Using with SQLExecuteQueryOperator
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -87,8 +88,8 @@ Using with SQLExecuteQueryOperator
         query_task = SQLExecuteQueryOperator(
             task_id="query_db2",
             conn_id="db2_default",
-            sql="SELECT * FROM employees WHERE dept = %(dept)s",
-            parameters={"dept": "IT"},
+            sql="SELECT * FROM employees WHERE dept = ?",
+            parameters=("IT",),
         )
 
 Features

providers/ibm/db2/docs/connections/db2.rst

@@ -28,8 +28,9 @@ Configuring the Connection
 Host (required)
     The host to connect to.
 
-Schema (optional)
-    Specify the schema name to be used in the database.
+Schema (required)
+    The Db2 database name to connect to. Maps to the ``DATABASE`` keyword in
+    the Db2 connection string.
 
 Login (required)
     Specify the user name to connect.
@@ -38,23 +39,31 @@ Password (required)
     Specify the password to connect.
 
 Port (optional)
-    Port of the Db2 database. Default is 50000.
+    Port of the Db2 database. Default is ``50000``.
 
 Extra (optional)
-    Specify the extra parameters (as json dictionary) that can be used in the
-    Db2 connection. The following parameters are supported:
+    Specify extra parameters (as a JSON dictionary) that are appended verbatim
+    to the Db2 connection string. Parameter names are converted to uppercase
+    automatically.
 
-    * ``database`` - The database name to connect to.
-    * ``protocol`` - The protocol to use (default: TCPIP).
-    * ``security`` - Security protocol (e.g., SSL).
+    .. note::
+
+        Do **not** put ``database`` or ``protocol`` here — ``database`` is
+        taken from the **Schema** field above, and ``protocol`` is always set
+        to ``TCPIP`` by the hook. Values placed in extras for these keys will
+        be appended as duplicates and ignored by the driver.
+
+    Common parameters:
+
+    * ``SECURITY`` - Enable SSL (set to ``"SSL"``).
+    * ``SSLServerCertificate`` - Path to the server SSL certificate.
     * Any other parameter supported by the IBM Db2 driver.
 
-    Example "extras" field:
+    Example "extras" field for SSL:
 
     .. code-block:: json
 
        {
-          "database": "sample",
-          "protocol": "TCPIP",
-          "security": "SSL"
+          "SECURITY": "SSL",
+          "SSLServerCertificate": "/path/to/server.crt"
        }

providers/ibm/db2/src/airflow/providers/ibm/db2/dialects/db2.py

@@ -30,6 +30,7 @@ def get_column_names(
         self,
         table: str,
         schema: str | None = None,
+        predicate=None,
         **kwargs,
     ) -> list[str] | None:
         """

providers/ibm/db2/tests/system/ibm/db2/example_db2.py

@@ -52,7 +52,7 @@
     @task
     def cleanup_tables() -> None:
         """Drop tables if they exist to ensure clean state."""
-        hook = Db2Hook(conn_id="db2_default")
+        hook = Db2Hook(db2_conn_id="db2_default")
         for table in ["EMPLOYEES_BACKUP", "EMPLOYEES"]:
             try:
                 hook.run(f"DROP TABLE {table}")
@@ -96,7 +96,7 @@ def cleanup_tables() -> None:
     @task
     def query_employees() -> int:
         """Query employees and return count using Db2Hook."""
-        hook = Db2Hook(conn_id="db2_default")
+        hook = Db2Hook(db2_conn_id="db2_default")
         sql = "SELECT employee_id, first_name, last_name, department, salary FROM employees ORDER BY employee_id"
         records = hook.get_records(sql)
         return len(records)
@@ -134,7 +134,7 @@ def query_employees() -> int:
     @task
     def display_statistics() -> None:
         """Display department statistics using Db2Hook."""
-        hook = Db2Hook(conn_id="db2_default")
+        hook = Db2Hook(db2_conn_id="db2_default")
         sql = """
             SELECT
                 department,
@@ -163,7 +163,7 @@ def display_statistics() -> None:
     @task
     def verify_backup() -> None:
         """Verify backup table was created successfully."""
-        hook = Db2Hook(conn_id="db2_default")
+        hook = Db2Hook(db2_conn_id="db2_default")
         original_count = hook.get_first("SELECT COUNT(*) FROM employees")[0]
         backup_count = hook.get_first("SELECT COUNT(*) FROM employees_backup")[0]
         if original_count != backup_count:

providers/ibm/db2/tests/unit/ibm/db2/dialects/test_db2.py

@@ -90,6 +90,18 @@ def test_get_column_names_with_schema(self):
             table_name="TEST_TABLE", schema="MY_SCHEMA"
         )
 
+    def test_get_target_fields_passes_predicate_positionally(self):
+        """get_target_fields() passes predicate as a positional arg — must not raise TypeError."""
+        self.test_db_hook.inspector.get_columns.return_value = [
+            {"name": "ID", "autoincrement": True, "identity": {"always": True}},
+            {"name": "NAME", "autoincrement": False},
+        ]
+
+        # Would raise TypeError before the fix (positional predicate hit **kwargs)
+        fields = Db2Dialect(self.test_db_hook).get_target_fields("TEST_TABLE")
+
+        assert fields == ["NAME"]
+
     def test_get_column_names_schema_prefix_in_table(self):
         """get_column_names() splits a 'schema.table' string correctly."""
         self.test_db_hook.inspector.get_columns.return_value = [

providers/ibm/db2/tests/unit/ibm/db2/hooks/test_db2.py

@@ -55,7 +55,7 @@ def mock_connection_with_extras(self):
             password="db2pass",
             schema="testdb",
             port=50000,
-            extra='{"ssl": true, "sslcert": "/path/to/cert.pem"}',
+            extra='{"SECURITY": "SSL", "SSLServerCertificate": "/path/to/cert.crt"}',
         )
         return conn
 
@@ -96,8 +96,8 @@ def test_get_conn_with_ssl(self, mock_get_connection, mock_connection_with_extra
         assert conn == mock_db_conn
         mock_ibm_db_dbi.connect.assert_called_once()
         call_args = mock_ibm_db_dbi.connect.call_args[0][0]
-        assert "SSL=true" in call_args
-        assert "SSLCERT=/path/to/cert.pem" in call_args
+        assert "SECURITY=SSL" in call_args
+        assert "SSLSERVERCERTIFICATE=/path/to/cert.crt" in call_args
 
     @patch("airflow.providers.ibm.db2.hooks.db2.Db2Hook.get_connection")
     def test_get_uri(self, mock_get_connection, mock_connection):