diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index d3fbcafea..a2ead4022 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -72,6 +72,11 @@ class RedshiftSSLMode(StrEnum):
 }
 
 
+@dataclass
+class RedshiftAdapterResponse(AdapterResponse):
+    query_id: int = -1
+
+
 @dataclass
 class RedshiftSSLConfig(dbtClassMixin, Replaceable):  # type: ignore
     ssl: bool = True
@@ -314,13 +319,24 @@ def _get_backend_pid(cls, connection):
             res = c.execute(sql).fetchone()
         return res[0]
 
+    @classmethod
+    def _get_last_query_id(cls, cursor):
+        sql = "select pg_last_query_id();"
+        res = cursor.execute(sql).fetchone()
+        return res[0]
+
     @classmethod
     def get_response(cls, cursor: redshift_connector.Cursor) -> AdapterResponse:
         # redshift_connector.Cursor doesn't have a status message attribute but
         # this function is only used for successful run, so we can just return a dummy
         rows = cursor.rowcount
         message = "SUCCESS"
-        return AdapterResponse(_message=message, rows_affected=rows)
+        query_id = cls._get_last_query_id(cursor)
+        return RedshiftAdapterResponse(
+            _message=message,
+            rows_affected=rows,
+            query_id=query_id,
+        )
 
     @contextmanager
     def exception_handler(self, sql):