We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Bug Report - Postegres / Redshift Extractors have inconsistent behavior
<!--- Provide a general summary of the issue in the Title above -->
<!--- Look through existing open and closed issues to see if someone has reported the issue before -->

## Expected Behavior

```python
# For RedshiftMetadataExtractor
redshift_metadata_extractor_config = ConfigFactory.from_dict({
    'extractor.redshift_metadata.where_clause_suffix':  'schema <> "foo"'
})
```

## Current Behavior

```python
# For RedshiftMetadataExtractor
redshift_metadata_extractor_config = ConfigFactory.from_dict({
    'extractor.redshift_metadata.where_clause_suffix':  'where schema <> "foo"'
})
```

## Possible Solution

Update redshift extractor to properly build suffix

## Screenshots (if appropriate)

## Context

Both the `PostgresMetadataExtractor` and the `RedshiftMetadataExtractor` take the config `extractor.redshift_metadata.where_clause_suffix` as they are both based off of `BasePostgresMetadataExtractor`.

However they require slightly different different sql partials, which is unexpected: 

The `where_clause_suffix` for `RedshiftMetadataExtractor` takes a partial sql statement **without** a `WHERE`.
The `where_clause_suffix` for `PostgresMetadataExtractor` take a partial sql statement **with** a `WHERE`.

The name `where_clause_suffix` implies that its a statement appended to a where clause and should  _not_ (IMHO) require a `WHERE` statement.


<!--- How has this issue affected you? -->
<!--- Providing context helps us come up with a solution that is most useful in the real world -->

## Your Environment
<!--- Include as many relevant details about the environment you experienced the bug in -->
* Amunsen version used: amundsen-databuilder 6.3.1
* Data warehouse stores: Neo4j
* Deployment (k8s or native): terraform
* Link to your fork or repository:

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `databuilder/databuilder/extractor/redshift_metadata_extractor.py`
Content:
```
1 # Copyright Contributors to the Amundsen project.
2 # SPDX-License-Identifier: Apache-2.0
3 
4 from typing import (  # noqa: F401
5     Any, Dict, Iterator, Union,
6 )
7 
8 from pyhocon import ConfigFactory, ConfigTree  # noqa: F401
9 
10 from databuilder.extractor.base_postgres_metadata_extractor import BasePostgresMetadataExtractor
11 
12 
13 class RedshiftMetadataExtractor(BasePostgresMetadataExtractor):
14     """
15     Extracts Redshift table and column metadata from underlying meta store database using SQLAlchemyExtractor
16 
17     This differs from the PostgresMetadataExtractor because in order to support Redshift's late binding views,
18     we need to join the INFORMATION_SCHEMA data against the function PG_GET_LATE_BINDING_VIEW_COLS().
19     """
20 
21     def get_sql_statement(self, use_catalog_as_cluster_name: bool, where_clause_suffix: str) -> str:
22         if use_catalog_as_cluster_name:
23             cluster_source = "CURRENT_DATABASE()"
24         else:
25             cluster_source = f"'{self._cluster}'"
26 
27         return """
28         SELECT
29             *
30         FROM (
31             SELECT
32               {cluster_source} as cluster,
33               c.table_schema as schema,
34               c.table_name as name,
35               pgtd.description as description,
36               c.column_name as col_name,
37               c.data_type as col_type,
38               pgcd.description as col_description,
39               ordinal_position as col_sort_order
40             FROM INFORMATION_SCHEMA.COLUMNS c
41             INNER JOIN
42               pg_catalog.pg_statio_all_tables as st on c.table_schema=st.schemaname and c.table_name=st.relname
43             LEFT JOIN
44               pg_catalog.pg_description pgcd on pgcd.objoid=st.relid and pgcd.objsubid=c.ordinal_position
45             LEFT JOIN
46               pg_catalog.pg_description pgtd on pgtd.objoid=st.relid and pgtd.objsubid=0
47 
48             UNION
49 
50             SELECT
51               {cluster_source} as cluster,
52               view_schema as schema,
53               view_name as name,
54               NULL as description,
55               column_name as col_name,
56               data_type as col_type,
57               NULL as col_description,
58               ordinal_position as col_sort_order
59             FROM
60                 PG_GET_LATE_BINDING_VIEW_COLS()
61                     COLS(view_schema NAME, view_name NAME, column_name NAME, data_type VARCHAR, ordinal_position INT)
62 
63             UNION
64 
65             SELECT
66               {cluster_source} AS cluster,
67               schemaname AS schema,
68               tablename AS name,
69               NULL AS description,
70               columnname AS col_name,
71               external_type AS col_type,
72               NULL AS col_description,
73               columnnum AS col_sort_order
74             FROM svv_external_columns
75         )
76 
77         {where_clause_suffix}
78         ORDER by cluster, schema, name, col_sort_order ;
79         """.format(
80             cluster_source=cluster_source,
81             where_clause_suffix=where_clause_suffix,
82         )
83 
84     def get_scope(self) -> str:
85         return 'extractor.redshift_metadata'
86
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/databuilder/databuilder/extractor/redshift_metadata_extractor.py b/databuilder/databuilder/extractor/redshift_metadata_extractor.py
--- a/databuilder/databuilder/extractor/redshift_metadata_extractor.py
+++ b/databuilder/databuilder/extractor/redshift_metadata_extractor.py
@@ -1,6 +1,7 @@
 # Copyright Contributors to the Amundsen project.
 # SPDX-License-Identifier: Apache-2.0
 
+import logging
 from typing import (  # noqa: F401
     Any, Dict, Iterator, Union,
 )
@@ -9,11 +10,14 @@
 
 from databuilder.extractor.base_postgres_metadata_extractor import BasePostgresMetadataExtractor
 
+LOGGER = logging.getLogger(__name__)
+
 
 class RedshiftMetadataExtractor(BasePostgresMetadataExtractor):
     """
     Extracts Redshift table and column metadata from underlying meta store database using SQLAlchemyExtractor
 
+
     This differs from the PostgresMetadataExtractor because in order to support Redshift's late binding views,
     we need to join the INFORMATION_SCHEMA data against the function PG_GET_LATE_BINDING_VIEW_COLS().
     """
@@ -24,6 +28,15 @@
         else:
             cluster_source = f"'{self._cluster}'"
 
+        if where_clause_suffix:
+            if where_clause_suffix.lower().startswith("where"):
+                LOGGER.warning("you no longer need to begin with 'where' in your suffix")
+                where_clause = where_clause_suffix
+            else:
+                where_clause = f"where {where_clause_suffix}"
+        else:
+            where_clause = ""
+
         return """
         SELECT
             *
@@ -74,11 +87,11 @@
             FROM svv_external_columns
         )
 
-        {where_clause_suffix}
+        {where_clause}
         ORDER by cluster, schema, name, col_sort_order ;
         """.format(
             cluster_source=cluster_source,
-            where_clause_suffix=where_clause_suffix,
+            where_clause=where_clause,
         )
 
     def get_scope(self) -> str:

{"golden_diff": "diff --git a/databuilder/databuilder/extractor/redshift_metadata_extractor.py b/databuilder/databuilder/extractor/redshift_metadata_extractor.py\n--- a/databuilder/databuilder/extractor/redshift_metadata_extractor.py\n+++ b/databuilder/databuilder/extractor/redshift_metadata_extractor.py\n@@ -1,6 +1,7 @@\n # Copyright Contributors to the Amundsen project.\n # SPDX-License-Identifier: Apache-2.0\n \n+import logging\n from typing import (  # noqa: F401\n     Any, Dict, Iterator, Union,\n )\n@@ -9,11 +10,14 @@\n \n from databuilder.extractor.base_postgres_metadata_extractor import BasePostgresMetadataExtractor\n \n+LOGGER = logging.getLogger(__name__)\n+\n \n class RedshiftMetadataExtractor(BasePostgresMetadataExtractor):\n     \"\"\"\n     Extracts Redshift table and column metadata from underlying meta store database using SQLAlchemyExtractor\n \n+\n     This differs from the PostgresMetadataExtractor because in order to support Redshift's late binding views,\n     we need to join the INFORMATION_SCHEMA data against the function PG_GET_LATE_BINDING_VIEW_COLS().\n     \"\"\"\n@@ -24,6 +28,15 @@\n         else:\n             cluster_source = f\"'{self._cluster}'\"\n \n+        if where_clause_suffix:\n+            if where_clause_suffix.lower().startswith(\"where\"):\n+                LOGGER.warning(\"you no longer need to begin with 'where' in your suffix\")\n+                where_clause = where_clause_suffix\n+            else:\n+                where_clause = f\"where {where_clause_suffix}\"\n+        else:\n+            where_clause = \"\"\n+\n         return \"\"\"\n         SELECT\n             *\n@@ -74,11 +87,11 @@\n             FROM svv_external_columns\n         )\n \n-        {where_clause_suffix}\n+        {where_clause}\n         ORDER by cluster, schema, name, col_sort_order ;\n         \"\"\".format(\n             cluster_source=cluster_source,\n-            where_clause_suffix=where_clause_suffix,\n+            where_clause=where_clause,\n         )\n \n     def get_scope(self) -> str:\n", "issue": "Bug Report - Postegres / Redshift Extractors have inconsistent behavior\n<!--- Provide a general summary of the issue in the Title above -->\r\n<!--- Look through existing open and closed issues to see if someone has reported the issue before -->\r\n\r\n## Expected Behavior\r\n\r\n```python\r\n# For RedshiftMetadataExtractor\r\nredshift_metadata_extractor_config = ConfigFactory.from_dict({\r\n    'extractor.redshift_metadata.where_clause_suffix':  'schema <> \"foo\"'\r\n})\r\n```\r\n\r\n## Current Behavior\r\n\r\n```python\r\n# For RedshiftMetadataExtractor\r\nredshift_metadata_extractor_config = ConfigFactory.from_dict({\r\n    'extractor.redshift_metadata.where_clause_suffix':  'where schema <> \"foo\"'\r\n})\r\n```\r\n\r\n## Possible Solution\r\n\r\nUpdate redshift extractor to properly build suffix\r\n\r\n## Screenshots (if appropriate)\r\n\r\n## Context\r\n\r\nBoth the `PostgresMetadataExtractor` and the `RedshiftMetadataExtractor` take the config `extractor.redshift_metadata.where_clause_suffix` as they are both based off of `BasePostgresMetadataExtractor`.\r\n\r\nHowever they require slightly different different sql partials, which is unexpected: \r\n\r\nThe `where_clause_suffix` for `RedshiftMetadataExtractor` takes a partial sql statement **without** a `WHERE`.\r\nThe `where_clause_suffix` for `PostgresMetadataExtractor` take a partial sql statement **with** a `WHERE`.\r\n\r\nThe name `where_clause_suffix` implies that its a statement appended to a where clause and should  _not_ (IMHO) require a `WHERE` statement.\r\n\r\n\r\n<!--- How has this issue affected you? -->\r\n<!--- Providing context helps us come up with a solution that is most useful in the real world -->\r\n\r\n## Your Environment\r\n<!--- Include as many relevant details about the environment you experienced the bug in -->\r\n* Amunsen version used: amundsen-databuilder 6.3.1\r\n* Data warehouse stores: Neo4j\r\n* Deployment (k8s or native): terraform\r\n* Link to your fork or repository:\n", "before_files": [{"content": "# Copyright Contributors to the Amundsen project.\n# SPDX-License-Identifier: Apache-2.0\n\nfrom typing import (  # noqa: F401\n    Any, Dict, Iterator, Union,\n)\n\nfrom pyhocon import ConfigFactory, ConfigTree  # noqa: F401\n\nfrom databuilder.extractor.base_postgres_metadata_extractor import BasePostgresMetadataExtractor\n\n\nclass RedshiftMetadataExtractor(BasePostgresMetadataExtractor):\n    \"\"\"\n    Extracts Redshift table and column metadata from underlying meta store database using SQLAlchemyExtractor\n\n    This differs from the PostgresMetadataExtractor because in order to support Redshift's late binding views,\n    we need to join the INFORMATION_SCHEMA data against the function PG_GET_LATE_BINDING_VIEW_COLS().\n    \"\"\"\n\n    def get_sql_statement(self, use_catalog_as_cluster_name: bool, where_clause_suffix: str) -> str:\n        if use_catalog_as_cluster_name:\n            cluster_source = \"CURRENT_DATABASE()\"\n        else:\n            cluster_source = f\"'{self._cluster}'\"\n\n        return \"\"\"\n        SELECT\n            *\n        FROM (\n            SELECT\n              {cluster_source} as cluster,\n              c.table_schema as schema,\n              c.table_name as name,\n              pgtd.description as description,\n              c.column_name as col_name,\n              c.data_type as col_type,\n              pgcd.description as col_description,\n              ordinal_position as col_sort_order\n            FROM INFORMATION_SCHEMA.COLUMNS c\n            INNER JOIN\n              pg_catalog.pg_statio_all_tables as st on c.table_schema=st.schemaname and c.table_name=st.relname\n            LEFT JOIN\n              pg_catalog.pg_description pgcd on pgcd.objoid=st.relid and pgcd.objsubid=c.ordinal_position\n            LEFT JOIN\n              pg_catalog.pg_description pgtd on pgtd.objoid=st.relid and pgtd.objsubid=0\n\n            UNION\n\n            SELECT\n              {cluster_source} as cluster,\n              view_schema as schema,\n              view_name as name,\n              NULL as description,\n              column_name as col_name,\n              data_type as col_type,\n              NULL as col_description,\n              ordinal_position as col_sort_order\n            FROM\n                PG_GET_LATE_BINDING_VIEW_COLS()\n                    COLS(view_schema NAME, view_name NAME, column_name NAME, data_type VARCHAR, ordinal_position INT)\n\n            UNION\n\n            SELECT\n              {cluster_source} AS cluster,\n              schemaname AS schema,\n              tablename AS name,\n              NULL AS description,\n              columnname AS col_name,\n              external_type AS col_type,\n              NULL AS col_description,\n              columnnum AS col_sort_order\n            FROM svv_external_columns\n        )\n\n        {where_clause_suffix}\n        ORDER by cluster, schema, name, col_sort_order ;\n        \"\"\".format(\n            cluster_source=cluster_source,\n            where_clause_suffix=where_clause_suffix,\n        )\n\n    def get_scope(self) -> str:\n        return 'extractor.redshift_metadata'\n", "path": "databuilder/databuilder/extractor/redshift_metadata_extractor.py"}], "after_files": [{"content": "# Copyright Contributors to the Amundsen project.\n# SPDX-License-Identifier: Apache-2.0\n\nimport logging\nfrom typing import (  # noqa: F401\n    Any, Dict, Iterator, Union,\n)\n\nfrom pyhocon import ConfigFactory, ConfigTree  # noqa: F401\n\nfrom databuilder.extractor.base_postgres_metadata_extractor import BasePostgresMetadataExtractor\n\nLOGGER = logging.getLogger(__name__)\n\n\nclass RedshiftMetadataExtractor(BasePostgresMetadataExtractor):\n    \"\"\"\n    Extracts Redshift table and column metadata from underlying meta store database using SQLAlchemyExtractor\n\n\n    This differs from the PostgresMetadataExtractor because in order to support Redshift's late binding views,\n    we need to join the INFORMATION_SCHEMA data against the function PG_GET_LATE_BINDING_VIEW_COLS().\n    \"\"\"\n\n    def get_sql_statement(self, use_catalog_as_cluster_name: bool, where_clause_suffix: str) -> str:\n        if use_catalog_as_cluster_name:\n            cluster_source = \"CURRENT_DATABASE()\"\n        else:\n            cluster_source = f\"'{self._cluster}'\"\n\n        if where_clause_suffix:\n            if where_clause_suffix.lower().startswith(\"where\"):\n                LOGGER.warning(\"you no longer need to begin with 'where' in your suffix\")\n                where_clause = where_clause_suffix\n            else:\n                where_clause = f\"where {where_clause_suffix}\"\n        else:\n            where_clause = \"\"\n\n        return \"\"\"\n        SELECT\n            *\n        FROM (\n            SELECT\n              {cluster_source} as cluster,\n              c.table_schema as schema,\n              c.table_name as name,\n              pgtd.description as description,\n              c.column_name as col_name,\n              c.data_type as col_type,\n              pgcd.description as col_description,\n              ordinal_position as col_sort_order\n            FROM INFORMATION_SCHEMA.COLUMNS c\n            INNER JOIN\n              pg_catalog.pg_statio_all_tables as st on c.table_schema=st.schemaname and c.table_name=st.relname\n            LEFT JOIN\n              pg_catalog.pg_description pgcd on pgcd.objoid=st.relid and pgcd.objsubid=c.ordinal_position\n            LEFT JOIN\n              pg_catalog.pg_description pgtd on pgtd.objoid=st.relid and pgtd.objsubid=0\n\n            UNION\n\n            SELECT\n              {cluster_source} as cluster,\n              view_schema as schema,\n              view_name as name,\n              NULL as description,\n              column_name as col_name,\n              data_type as col_type,\n              NULL as col_description,\n              ordinal_position as col_sort_order\n            FROM\n                PG_GET_LATE_BINDING_VIEW_COLS()\n                    COLS(view_schema NAME, view_name NAME, column_name NAME, data_type VARCHAR, ordinal_position INT)\n\n            UNION\n\n            SELECT\n              {cluster_source} AS cluster,\n              schemaname AS schema,\n              tablename AS name,\n              NULL AS description,\n              columnname AS col_name,\n              external_type AS col_type,\n              NULL AS col_description,\n              columnnum AS col_sort_order\n            FROM svv_external_columns\n        )\n\n        {where_clause}\n        ORDER by cluster, schema, name, col_sort_order ;\n        \"\"\".format(\n            cluster_source=cluster_source,\n            where_clause=where_clause,\n        )\n\n    def get_scope(self) -> str:\n        return 'extractor.redshift_metadata'\n", "path": "databuilder/databuilder/extractor/redshift_metadata_extractor.py"}]}