We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[CT-2848] [Bug] `dbt show` adding decimal places to non-decimal values
### Is this a regression in a recent version of dbt-core?

- [X] I believe this is a regression in dbt-core functionality
- [X] I have searched the existing issues, and I could not find an existing issue for this regression

### Current Behavior

`dbt show` is adding decimal places to numeric/integer values incorrectly. This is causing problems for users in the dbt Cloud IDE, as the IDE uses that command to render the preview tab. 

### Expected/Previous Behavior

Previously (we think!) this was not an issue, and decimal places were properly handled by dbt show

### Steps To Reproduce

1. use dbt-snowflake>=1.5.0
2. create a model in your project:

```sql
# in my_model.sql
select
  cast(1 as numeric) as _numeric,
  cast(1 as integer) as _integer,
  cast(1 as decimal) as _decimal
```
3. run `dbt show -s my_model --output JSON`
4. get this!
```zsh
❯ dbt show -s my_model --output json
20:33:15  Running with dbt=1.5.3
20:33:16  Registered adapter: snowflake=1.5.2
20:33:16  Unable to do partial parsing because of a version mismatch
20:33:17  Found 3 models, 4 tests, 0 snapshots, 0 analyses, 436 macros, 0 operations, 0 seed files, 0 sources, 0 exposures, 0 metrics, 0 groups
20:33:17  
20:33:18  Concurrency: 8 threads (target='dev')
20:33:18  
20:33:19  {
  "node": "my_model",
  "show": [
    {
      "_NUMERIC": 1.0,
      "_INTEGER": 1.0,
      "_DECIMAL": 1.0
    }
  ]
}
```

### Relevant log output

_No response_

### Environment

```markdown
- OS: mac
- Python: Python 3.9.16
- dbt (working version): unknown
- dbt (regression version): 1.5.3
```


### Which database adapter are you using with dbt?

snowflake

### Additional Context

_No response_

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `core/dbt/clients/agate_helper.py`
Content:
```
1 from codecs import BOM_UTF8
2 
3 import agate
4 import datetime
5 import isodate
6 import json
7 import dbt.utils
8 from typing import Iterable, List, Dict, Union, Optional, Any
9 
10 from dbt.exceptions import DbtRuntimeError
11 
12 BOM = BOM_UTF8.decode("utf-8")  # '\ufeff'
13 
14 
15 class Number(agate.data_types.Number):
16     # undo the change in https://github.com/wireservice/agate/pull/733
17     # i.e. do not cast True and False to numeric 1 and 0
18     def cast(self, d):
19         if type(d) == bool:
20             raise agate.exceptions.CastError("Do not cast True to 1 or False to 0.")
21         else:
22             return super().cast(d)
23 
24 
25 class ISODateTime(agate.data_types.DateTime):
26     def cast(self, d):
27         # this is agate.data_types.DateTime.cast with the "clever" bits removed
28         # so we only handle ISO8601 stuff
29         if isinstance(d, datetime.datetime) or d is None:
30             return d
31         elif isinstance(d, datetime.date):
32             return datetime.datetime.combine(d, datetime.time(0, 0, 0))
33         elif isinstance(d, str):
34             d = d.strip()
35             if d.lower() in self.null_values:
36                 return None
37         try:
38             return isodate.parse_datetime(d)
39         except:  # noqa
40             pass
41 
42         raise agate.exceptions.CastError('Can not parse value "%s" as datetime.' % d)
43 
44 
45 def build_type_tester(
46     text_columns: Iterable[str], string_null_values: Optional[Iterable[str]] = ("null", "")
47 ) -> agate.TypeTester:
48 
49     types = [
50         Number(null_values=("null", "")),
51         agate.data_types.Date(null_values=("null", ""), date_format="%Y-%m-%d"),
52         agate.data_types.DateTime(null_values=("null", ""), datetime_format="%Y-%m-%d %H:%M:%S"),
53         ISODateTime(null_values=("null", "")),
54         agate.data_types.Boolean(
55             true_values=("true",), false_values=("false",), null_values=("null", "")
56         ),
57         agate.data_types.Text(null_values=string_null_values),
58     ]
59     force = {k: agate.data_types.Text(null_values=string_null_values) for k in text_columns}
60     return agate.TypeTester(force=force, types=types)
61 
62 
63 DEFAULT_TYPE_TESTER = build_type_tester(())
64 
65 
66 def table_from_rows(
67     rows: List[Any],
68     column_names: Iterable[str],
69     text_only_columns: Optional[Iterable[str]] = None,
70 ) -> agate.Table:
71     if text_only_columns is None:
72         column_types = DEFAULT_TYPE_TESTER
73     else:
74         # If text_only_columns are present, prevent coercing empty string or
75         # literal 'null' strings to a None representation.
76         column_types = build_type_tester(text_only_columns, string_null_values=())
77 
78     return agate.Table(rows, column_names, column_types=column_types)
79 
80 
81 def table_from_data(data, column_names: Iterable[str]) -> agate.Table:
82     "Convert a list of dictionaries into an Agate table"
83 
84     # The agate table is generated from a list of dicts, so the column order
85     # from `data` is not preserved. We can use `select` to reorder the columns
86     #
87     # If there is no data, create an empty table with the specified columns
88 
89     if len(data) == 0:
90         return agate.Table([], column_names=column_names)
91     else:
92         table = agate.Table.from_object(data, column_types=DEFAULT_TYPE_TESTER)
93         return table.select(column_names)
94 
95 
96 def table_from_data_flat(data, column_names: Iterable[str]) -> agate.Table:
97     """
98     Convert a list of dictionaries into an Agate table. This method does not
99     coerce string values into more specific types (eg. '005' will not be
100     coerced to '5'). Additionally, this method does not coerce values to
101     None (eg. '' or 'null' will retain their string literal representations).
102     """
103 
104     rows = []
105     text_only_columns = set()
106     for _row in data:
107         row = []
108         for col_name in column_names:
109             value = _row[col_name]
110             if isinstance(value, (dict, list, tuple)):
111                 # Represent container types as json strings
112                 value = json.dumps(value, cls=dbt.utils.JSONEncoder)
113                 text_only_columns.add(col_name)
114             elif isinstance(value, str):
115                 text_only_columns.add(col_name)
116             row.append(value)
117 
118         rows.append(row)
119 
120     return table_from_rows(
121         rows=rows, column_names=column_names, text_only_columns=text_only_columns
122     )
123 
124 
125 def empty_table():
126     "Returns an empty Agate table. To be used in place of None"
127 
128     return agate.Table(rows=[])
129 
130 
131 def as_matrix(table):
132     "Return an agate table as a matrix of data sans columns"
133 
134     return [r.values() for r in table.rows.values()]
135 
136 
137 def from_csv(abspath, text_columns, delimiter=","):
138     type_tester = build_type_tester(text_columns=text_columns)
139     with open(abspath, encoding="utf-8") as fp:
140         if fp.read(1) != BOM:
141             fp.seek(0)
142         return agate.Table.from_csv(fp, column_types=type_tester, delimiter=delimiter)
143 
144 
145 class _NullMarker:
146     pass
147 
148 
149 NullableAgateType = Union[agate.data_types.DataType, _NullMarker]
150 
151 
152 class ColumnTypeBuilder(Dict[str, NullableAgateType]):
153     def __init__(self):
154         super().__init__()
155 
156     def __setitem__(self, key, value):
157         if key not in self:
158             super().__setitem__(key, value)
159             return
160 
161         existing_type = self[key]
162         if isinstance(existing_type, _NullMarker):
163             # overwrite
164             super().__setitem__(key, value)
165         elif isinstance(value, _NullMarker):
166             # use the existing value
167             return
168         elif not isinstance(value, type(existing_type)):
169             # actual type mismatch!
170             raise DbtRuntimeError(
171                 f"Tables contain columns with the same names ({key}), "
172                 f"but different types ({value} vs {existing_type})"
173             )
174 
175     def finalize(self) -> Dict[str, agate.data_types.DataType]:
176         result: Dict[str, agate.data_types.DataType] = {}
177         for key, value in self.items():
178             if isinstance(value, _NullMarker):
179                 # this is what agate would do.
180                 result[key] = agate.data_types.Number()
181             else:
182                 result[key] = value
183         return result
184 
185 
186 def _merged_column_types(tables: List[agate.Table]) -> Dict[str, agate.data_types.DataType]:
187     # this is a lot like agate.Table.merge, but with handling for all-null
188     # rows being "any type".
189     new_columns: ColumnTypeBuilder = ColumnTypeBuilder()
190     for table in tables:
191         for i in range(len(table.columns)):
192             column_name: str = table.column_names[i]
193             column_type: NullableAgateType = table.column_types[i]
194             # avoid over-sensitive type inference
195             if all(x is None for x in table.columns[column_name]):
196                 column_type = _NullMarker()
197             new_columns[column_name] = column_type
198 
199     return new_columns.finalize()
200 
201 
202 def merge_tables(tables: List[agate.Table]) -> agate.Table:
203     """This is similar to agate.Table.merge, but it handles rows of all 'null'
204     values more gracefully during merges.
205     """
206     new_columns = _merged_column_types(tables)
207     column_names = tuple(new_columns.keys())
208     column_types = tuple(new_columns.values())
209 
210     rows: List[agate.Row] = []
211     for table in tables:
212         if table.column_names == column_names and table.column_types == column_types:
213             rows.extend(table.rows)
214         else:
215             for row in table.rows:
216                 data = [row.get(name, None) for name in column_names]
217                 rows.append(agate.Row(data, column_names))
218     # _is_fork to tell agate that we already made things into `Row`s.
219     return agate.Table(rows, column_names, column_types, _is_fork=True)
220
```
--- 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/core/dbt/clients/agate_helper.py b/core/dbt/clients/agate_helper.py
--- a/core/dbt/clients/agate_helper.py
+++ b/core/dbt/clients/agate_helper.py
@@ -12,6 +12,20 @@
 BOM = BOM_UTF8.decode("utf-8")  # '\ufeff'
 
 
+class Integer(agate.data_types.DataType):
+    def cast(self, d):
+        # by default agate will cast none as a Number
+        # but we need to cast it as an Integer to preserve
+        # the type when merging and unioning tables
+        if type(d) == int or d is None:
+            return d
+        else:
+            raise agate.exceptions.CastError('Can not parse value "%s" as Integer.' % d)
+
+    def jsonify(self, d):
+        return d
+
+
 class Number(agate.data_types.Number):
     # undo the change in https://github.com/wireservice/agate/pull/733
     # i.e. do not cast True and False to numeric 1 and 0
@@ -47,6 +61,7 @@
 ) -> agate.TypeTester:
 
     types = [
+        Integer(null_values=("null", "")),
         Number(null_values=("null", "")),
         agate.data_types.Date(null_values=("null", ""), date_format="%Y-%m-%d"),
         agate.data_types.DateTime(null_values=("null", ""), datetime_format="%Y-%m-%d %H:%M:%S"),
@@ -165,6 +180,13 @@
         elif isinstance(value, _NullMarker):
             # use the existing value
             return
+        # when one table column is Number while another is Integer, force the column to Number on merge
+        elif isinstance(value, Integer) and isinstance(existing_type, agate.data_types.Number):
+            # use the existing value
+            return
+        elif isinstance(existing_type, Integer) and isinstance(value, agate.data_types.Number):
+            # overwrite
+            super().__setitem__(key, value)
         elif not isinstance(value, type(existing_type)):
             # actual type mismatch!
             raise DbtRuntimeError(
@@ -176,8 +198,9 @@
         result: Dict[str, agate.data_types.DataType] = {}
         for key, value in self.items():
             if isinstance(value, _NullMarker):
-                # this is what agate would do.
-                result[key] = agate.data_types.Number()
+                # agate would make it a Number but we'll make it Integer so that if this column
+                # gets merged with another Integer column, it won't get forced to a Number
+                result[key] = Integer()
             else:
                 result[key] = value
         return result

{"golden_diff": "diff --git a/core/dbt/clients/agate_helper.py b/core/dbt/clients/agate_helper.py\n--- a/core/dbt/clients/agate_helper.py\n+++ b/core/dbt/clients/agate_helper.py\n@@ -12,6 +12,20 @@\n BOM = BOM_UTF8.decode(\"utf-8\")  # '\\ufeff'\n \n \n+class Integer(agate.data_types.DataType):\n+    def cast(self, d):\n+        # by default agate will cast none as a Number\n+        # but we need to cast it as an Integer to preserve\n+        # the type when merging and unioning tables\n+        if type(d) == int or d is None:\n+            return d\n+        else:\n+            raise agate.exceptions.CastError('Can not parse value \"%s\" as Integer.' % d)\n+\n+    def jsonify(self, d):\n+        return d\n+\n+\n class Number(agate.data_types.Number):\n     # undo the change in https://github.com/wireservice/agate/pull/733\n     # i.e. do not cast True and False to numeric 1 and 0\n@@ -47,6 +61,7 @@\n ) -> agate.TypeTester:\n \n     types = [\n+        Integer(null_values=(\"null\", \"\")),\n         Number(null_values=(\"null\", \"\")),\n         agate.data_types.Date(null_values=(\"null\", \"\"), date_format=\"%Y-%m-%d\"),\n         agate.data_types.DateTime(null_values=(\"null\", \"\"), datetime_format=\"%Y-%m-%d %H:%M:%S\"),\n@@ -165,6 +180,13 @@\n         elif isinstance(value, _NullMarker):\n             # use the existing value\n             return\n+        # when one table column is Number while another is Integer, force the column to Number on merge\n+        elif isinstance(value, Integer) and isinstance(existing_type, agate.data_types.Number):\n+            # use the existing value\n+            return\n+        elif isinstance(existing_type, Integer) and isinstance(value, agate.data_types.Number):\n+            # overwrite\n+            super().__setitem__(key, value)\n         elif not isinstance(value, type(existing_type)):\n             # actual type mismatch!\n             raise DbtRuntimeError(\n@@ -176,8 +198,9 @@\n         result: Dict[str, agate.data_types.DataType] = {}\n         for key, value in self.items():\n             if isinstance(value, _NullMarker):\n-                # this is what agate would do.\n-                result[key] = agate.data_types.Number()\n+                # agate would make it a Number but we'll make it Integer so that if this column\n+                # gets merged with another Integer column, it won't get forced to a Number\n+                result[key] = Integer()\n             else:\n                 result[key] = value\n         return result\n", "issue": "[CT-2848] [Bug] `dbt show` adding decimal places to non-decimal values\n### Is this a regression in a recent version of dbt-core?\r\n\r\n- [X] I believe this is a regression in dbt-core functionality\r\n- [X] I have searched the existing issues, and I could not find an existing issue for this regression\r\n\r\n### Current Behavior\r\n\r\n`dbt show` is adding decimal places to numeric/integer values incorrectly. This is causing problems for users in the dbt Cloud IDE, as the IDE uses that command to render the preview tab. \r\n\r\n### Expected/Previous Behavior\r\n\r\nPreviously (we think!) this was not an issue, and decimal places were properly handled by dbt show\r\n\r\n### Steps To Reproduce\r\n\r\n1. use dbt-snowflake>=1.5.0\r\n2. create a model in your project:\r\n\r\n```sql\r\n# in my_model.sql\r\nselect\r\n  cast(1 as numeric) as _numeric,\r\n  cast(1 as integer) as _integer,\r\n  cast(1 as decimal) as _decimal\r\n```\r\n3. run `dbt show -s my_model --output JSON`\r\n4. get this!\r\n```zsh\r\n\u276f dbt show -s my_model --output json\r\n20:33:15  Running with dbt=1.5.3\r\n20:33:16  Registered adapter: snowflake=1.5.2\r\n20:33:16  Unable to do partial parsing because of a version mismatch\r\n20:33:17  Found 3 models, 4 tests, 0 snapshots, 0 analyses, 436 macros, 0 operations, 0 seed files, 0 sources, 0 exposures, 0 metrics, 0 groups\r\n20:33:17  \r\n20:33:18  Concurrency: 8 threads (target='dev')\r\n20:33:18  \r\n20:33:19  {\r\n  \"node\": \"my_model\",\r\n  \"show\": [\r\n    {\r\n      \"_NUMERIC\": 1.0,\r\n      \"_INTEGER\": 1.0,\r\n      \"_DECIMAL\": 1.0\r\n    }\r\n  ]\r\n}\r\n```\r\n\r\n### Relevant log output\r\n\r\n_No response_\r\n\r\n### Environment\r\n\r\n```markdown\r\n- OS: mac\r\n- Python: Python 3.9.16\r\n- dbt (working version): unknown\r\n- dbt (regression version): 1.5.3\r\n```\r\n\r\n\r\n### Which database adapter are you using with dbt?\r\n\r\nsnowflake\r\n\r\n### Additional Context\r\n\r\n_No response_\n", "before_files": [{"content": "from codecs import BOM_UTF8\n\nimport agate\nimport datetime\nimport isodate\nimport json\nimport dbt.utils\nfrom typing import Iterable, List, Dict, Union, Optional, Any\n\nfrom dbt.exceptions import DbtRuntimeError\n\nBOM = BOM_UTF8.decode(\"utf-8\")  # '\\ufeff'\n\n\nclass Number(agate.data_types.Number):\n    # undo the change in https://github.com/wireservice/agate/pull/733\n    # i.e. do not cast True and False to numeric 1 and 0\n    def cast(self, d):\n        if type(d) == bool:\n            raise agate.exceptions.CastError(\"Do not cast True to 1 or False to 0.\")\n        else:\n            return super().cast(d)\n\n\nclass ISODateTime(agate.data_types.DateTime):\n    def cast(self, d):\n        # this is agate.data_types.DateTime.cast with the \"clever\" bits removed\n        # so we only handle ISO8601 stuff\n        if isinstance(d, datetime.datetime) or d is None:\n            return d\n        elif isinstance(d, datetime.date):\n            return datetime.datetime.combine(d, datetime.time(0, 0, 0))\n        elif isinstance(d, str):\n            d = d.strip()\n            if d.lower() in self.null_values:\n                return None\n        try:\n            return isodate.parse_datetime(d)\n        except:  # noqa\n            pass\n\n        raise agate.exceptions.CastError('Can not parse value \"%s\" as datetime.' % d)\n\n\ndef build_type_tester(\n    text_columns: Iterable[str], string_null_values: Optional[Iterable[str]] = (\"null\", \"\")\n) -> agate.TypeTester:\n\n    types = [\n        Number(null_values=(\"null\", \"\")),\n        agate.data_types.Date(null_values=(\"null\", \"\"), date_format=\"%Y-%m-%d\"),\n        agate.data_types.DateTime(null_values=(\"null\", \"\"), datetime_format=\"%Y-%m-%d %H:%M:%S\"),\n        ISODateTime(null_values=(\"null\", \"\")),\n        agate.data_types.Boolean(\n            true_values=(\"true\",), false_values=(\"false\",), null_values=(\"null\", \"\")\n        ),\n        agate.data_types.Text(null_values=string_null_values),\n    ]\n    force = {k: agate.data_types.Text(null_values=string_null_values) for k in text_columns}\n    return agate.TypeTester(force=force, types=types)\n\n\nDEFAULT_TYPE_TESTER = build_type_tester(())\n\n\ndef table_from_rows(\n    rows: List[Any],\n    column_names: Iterable[str],\n    text_only_columns: Optional[Iterable[str]] = None,\n) -> agate.Table:\n    if text_only_columns is None:\n        column_types = DEFAULT_TYPE_TESTER\n    else:\n        # If text_only_columns are present, prevent coercing empty string or\n        # literal 'null' strings to a None representation.\n        column_types = build_type_tester(text_only_columns, string_null_values=())\n\n    return agate.Table(rows, column_names, column_types=column_types)\n\n\ndef table_from_data(data, column_names: Iterable[str]) -> agate.Table:\n    \"Convert a list of dictionaries into an Agate table\"\n\n    # The agate table is generated from a list of dicts, so the column order\n    # from `data` is not preserved. We can use `select` to reorder the columns\n    #\n    # If there is no data, create an empty table with the specified columns\n\n    if len(data) == 0:\n        return agate.Table([], column_names=column_names)\n    else:\n        table = agate.Table.from_object(data, column_types=DEFAULT_TYPE_TESTER)\n        return table.select(column_names)\n\n\ndef table_from_data_flat(data, column_names: Iterable[str]) -> agate.Table:\n    \"\"\"\n    Convert a list of dictionaries into an Agate table. This method does not\n    coerce string values into more specific types (eg. '005' will not be\n    coerced to '5'). Additionally, this method does not coerce values to\n    None (eg. '' or 'null' will retain their string literal representations).\n    \"\"\"\n\n    rows = []\n    text_only_columns = set()\n    for _row in data:\n        row = []\n        for col_name in column_names:\n            value = _row[col_name]\n            if isinstance(value, (dict, list, tuple)):\n                # Represent container types as json strings\n                value = json.dumps(value, cls=dbt.utils.JSONEncoder)\n                text_only_columns.add(col_name)\n            elif isinstance(value, str):\n                text_only_columns.add(col_name)\n            row.append(value)\n\n        rows.append(row)\n\n    return table_from_rows(\n        rows=rows, column_names=column_names, text_only_columns=text_only_columns\n    )\n\n\ndef empty_table():\n    \"Returns an empty Agate table. To be used in place of None\"\n\n    return agate.Table(rows=[])\n\n\ndef as_matrix(table):\n    \"Return an agate table as a matrix of data sans columns\"\n\n    return [r.values() for r in table.rows.values()]\n\n\ndef from_csv(abspath, text_columns, delimiter=\",\"):\n    type_tester = build_type_tester(text_columns=text_columns)\n    with open(abspath, encoding=\"utf-8\") as fp:\n        if fp.read(1) != BOM:\n            fp.seek(0)\n        return agate.Table.from_csv(fp, column_types=type_tester, delimiter=delimiter)\n\n\nclass _NullMarker:\n    pass\n\n\nNullableAgateType = Union[agate.data_types.DataType, _NullMarker]\n\n\nclass ColumnTypeBuilder(Dict[str, NullableAgateType]):\n    def __init__(self):\n        super().__init__()\n\n    def __setitem__(self, key, value):\n        if key not in self:\n            super().__setitem__(key, value)\n            return\n\n        existing_type = self[key]\n        if isinstance(existing_type, _NullMarker):\n            # overwrite\n            super().__setitem__(key, value)\n        elif isinstance(value, _NullMarker):\n            # use the existing value\n            return\n        elif not isinstance(value, type(existing_type)):\n            # actual type mismatch!\n            raise DbtRuntimeError(\n                f\"Tables contain columns with the same names ({key}), \"\n                f\"but different types ({value} vs {existing_type})\"\n            )\n\n    def finalize(self) -> Dict[str, agate.data_types.DataType]:\n        result: Dict[str, agate.data_types.DataType] = {}\n        for key, value in self.items():\n            if isinstance(value, _NullMarker):\n                # this is what agate would do.\n                result[key] = agate.data_types.Number()\n            else:\n                result[key] = value\n        return result\n\n\ndef _merged_column_types(tables: List[agate.Table]) -> Dict[str, agate.data_types.DataType]:\n    # this is a lot like agate.Table.merge, but with handling for all-null\n    # rows being \"any type\".\n    new_columns: ColumnTypeBuilder = ColumnTypeBuilder()\n    for table in tables:\n        for i in range(len(table.columns)):\n            column_name: str = table.column_names[i]\n            column_type: NullableAgateType = table.column_types[i]\n            # avoid over-sensitive type inference\n            if all(x is None for x in table.columns[column_name]):\n                column_type = _NullMarker()\n            new_columns[column_name] = column_type\n\n    return new_columns.finalize()\n\n\ndef merge_tables(tables: List[agate.Table]) -> agate.Table:\n    \"\"\"This is similar to agate.Table.merge, but it handles rows of all 'null'\n    values more gracefully during merges.\n    \"\"\"\n    new_columns = _merged_column_types(tables)\n    column_names = tuple(new_columns.keys())\n    column_types = tuple(new_columns.values())\n\n    rows: List[agate.Row] = []\n    for table in tables:\n        if table.column_names == column_names and table.column_types == column_types:\n            rows.extend(table.rows)\n        else:\n            for row in table.rows:\n                data = [row.get(name, None) for name in column_names]\n                rows.append(agate.Row(data, column_names))\n    # _is_fork to tell agate that we already made things into `Row`s.\n    return agate.Table(rows, column_names, column_types, _is_fork=True)\n", "path": "core/dbt/clients/agate_helper.py"}], "after_files": [{"content": "from codecs import BOM_UTF8\n\nimport agate\nimport datetime\nimport isodate\nimport json\nimport dbt.utils\nfrom typing import Iterable, List, Dict, Union, Optional, Any\n\nfrom dbt.exceptions import DbtRuntimeError\n\nBOM = BOM_UTF8.decode(\"utf-8\")  # '\\ufeff'\n\n\nclass Integer(agate.data_types.DataType):\n    def cast(self, d):\n        # by default agate will cast none as a Number\n        # but we need to cast it as an Integer to preserve\n        # the type when merging and unioning tables\n        if type(d) == int or d is None:\n            return d\n        else:\n            raise agate.exceptions.CastError('Can not parse value \"%s\" as Integer.' % d)\n\n    def jsonify(self, d):\n        return d\n\n\nclass Number(agate.data_types.Number):\n    # undo the change in https://github.com/wireservice/agate/pull/733\n    # i.e. do not cast True and False to numeric 1 and 0\n    def cast(self, d):\n        if type(d) == bool:\n            raise agate.exceptions.CastError(\"Do not cast True to 1 or False to 0.\")\n        else:\n            return super().cast(d)\n\n\nclass ISODateTime(agate.data_types.DateTime):\n    def cast(self, d):\n        # this is agate.data_types.DateTime.cast with the \"clever\" bits removed\n        # so we only handle ISO8601 stuff\n        if isinstance(d, datetime.datetime) or d is None:\n            return d\n        elif isinstance(d, datetime.date):\n            return datetime.datetime.combine(d, datetime.time(0, 0, 0))\n        elif isinstance(d, str):\n            d = d.strip()\n            if d.lower() in self.null_values:\n                return None\n        try:\n            return isodate.parse_datetime(d)\n        except:  # noqa\n            pass\n\n        raise agate.exceptions.CastError('Can not parse value \"%s\" as datetime.' % d)\n\n\ndef build_type_tester(\n    text_columns: Iterable[str], string_null_values: Optional[Iterable[str]] = (\"null\", \"\")\n) -> agate.TypeTester:\n\n    types = [\n        Integer(null_values=(\"null\", \"\")),\n        Number(null_values=(\"null\", \"\")),\n        agate.data_types.Date(null_values=(\"null\", \"\"), date_format=\"%Y-%m-%d\"),\n        agate.data_types.DateTime(null_values=(\"null\", \"\"), datetime_format=\"%Y-%m-%d %H:%M:%S\"),\n        ISODateTime(null_values=(\"null\", \"\")),\n        agate.data_types.Boolean(\n            true_values=(\"true\",), false_values=(\"false\",), null_values=(\"null\", \"\")\n        ),\n        agate.data_types.Text(null_values=string_null_values),\n    ]\n    force = {k: agate.data_types.Text(null_values=string_null_values) for k in text_columns}\n    return agate.TypeTester(force=force, types=types)\n\n\nDEFAULT_TYPE_TESTER = build_type_tester(())\n\n\ndef table_from_rows(\n    rows: List[Any],\n    column_names: Iterable[str],\n    text_only_columns: Optional[Iterable[str]] = None,\n) -> agate.Table:\n    if text_only_columns is None:\n        column_types = DEFAULT_TYPE_TESTER\n    else:\n        # If text_only_columns are present, prevent coercing empty string or\n        # literal 'null' strings to a None representation.\n        column_types = build_type_tester(text_only_columns, string_null_values=())\n\n    return agate.Table(rows, column_names, column_types=column_types)\n\n\ndef table_from_data(data, column_names: Iterable[str]) -> agate.Table:\n    \"Convert a list of dictionaries into an Agate table\"\n\n    # The agate table is generated from a list of dicts, so the column order\n    # from `data` is not preserved. We can use `select` to reorder the columns\n    #\n    # If there is no data, create an empty table with the specified columns\n\n    if len(data) == 0:\n        return agate.Table([], column_names=column_names)\n    else:\n        table = agate.Table.from_object(data, column_types=DEFAULT_TYPE_TESTER)\n        return table.select(column_names)\n\n\ndef table_from_data_flat(data, column_names: Iterable[str]) -> agate.Table:\n    \"\"\"\n    Convert a list of dictionaries into an Agate table. This method does not\n    coerce string values into more specific types (eg. '005' will not be\n    coerced to '5'). Additionally, this method does not coerce values to\n    None (eg. '' or 'null' will retain their string literal representations).\n    \"\"\"\n\n    rows = []\n    text_only_columns = set()\n    for _row in data:\n        row = []\n        for col_name in column_names:\n            value = _row[col_name]\n            if isinstance(value, (dict, list, tuple)):\n                # Represent container types as json strings\n                value = json.dumps(value, cls=dbt.utils.JSONEncoder)\n                text_only_columns.add(col_name)\n            elif isinstance(value, str):\n                text_only_columns.add(col_name)\n            row.append(value)\n\n        rows.append(row)\n\n    return table_from_rows(\n        rows=rows, column_names=column_names, text_only_columns=text_only_columns\n    )\n\n\ndef empty_table():\n    \"Returns an empty Agate table. To be used in place of None\"\n\n    return agate.Table(rows=[])\n\n\ndef as_matrix(table):\n    \"Return an agate table as a matrix of data sans columns\"\n\n    return [r.values() for r in table.rows.values()]\n\n\ndef from_csv(abspath, text_columns, delimiter=\",\"):\n    type_tester = build_type_tester(text_columns=text_columns)\n    with open(abspath, encoding=\"utf-8\") as fp:\n        if fp.read(1) != BOM:\n            fp.seek(0)\n        return agate.Table.from_csv(fp, column_types=type_tester, delimiter=delimiter)\n\n\nclass _NullMarker:\n    pass\n\n\nNullableAgateType = Union[agate.data_types.DataType, _NullMarker]\n\n\nclass ColumnTypeBuilder(Dict[str, NullableAgateType]):\n    def __init__(self):\n        super().__init__()\n\n    def __setitem__(self, key, value):\n        if key not in self:\n            super().__setitem__(key, value)\n            return\n\n        existing_type = self[key]\n        if isinstance(existing_type, _NullMarker):\n            # overwrite\n            super().__setitem__(key, value)\n        elif isinstance(value, _NullMarker):\n            # use the existing value\n            return\n        # when one table column is Number while another is Integer, force the column to Number on merge\n        elif isinstance(value, Integer) and isinstance(existing_type, agate.data_types.Number):\n            # use the existing value\n            return\n        elif isinstance(existing_type, Integer) and isinstance(value, agate.data_types.Number):\n            # overwrite\n            super().__setitem__(key, value)\n        elif not isinstance(value, type(existing_type)):\n            # actual type mismatch!\n            raise DbtRuntimeError(\n                f\"Tables contain columns with the same names ({key}), \"\n                f\"but different types ({value} vs {existing_type})\"\n            )\n\n    def finalize(self) -> Dict[str, agate.data_types.DataType]:\n        result: Dict[str, agate.data_types.DataType] = {}\n        for key, value in self.items():\n            if isinstance(value, _NullMarker):\n                # agate would make it a Number but we'll make it Integer so that if this column\n                # gets merged with another Integer column, it won't get forced to a Number\n                result[key] = Integer()\n            else:\n                result[key] = value\n        return result\n\n\ndef _merged_column_types(tables: List[agate.Table]) -> Dict[str, agate.data_types.DataType]:\n    # this is a lot like agate.Table.merge, but with handling for all-null\n    # rows being \"any type\".\n    new_columns: ColumnTypeBuilder = ColumnTypeBuilder()\n    for table in tables:\n        for i in range(len(table.columns)):\n            column_name: str = table.column_names[i]\n            column_type: NullableAgateType = table.column_types[i]\n            # avoid over-sensitive type inference\n            if all(x is None for x in table.columns[column_name]):\n                column_type = _NullMarker()\n            new_columns[column_name] = column_type\n\n    return new_columns.finalize()\n\n\ndef merge_tables(tables: List[agate.Table]) -> agate.Table:\n    \"\"\"This is similar to agate.Table.merge, but it handles rows of all 'null'\n    values more gracefully during merges.\n    \"\"\"\n    new_columns = _merged_column_types(tables)\n    column_names = tuple(new_columns.keys())\n    column_types = tuple(new_columns.values())\n\n    rows: List[agate.Row] = []\n    for table in tables:\n        if table.column_names == column_names and table.column_types == column_types:\n            rows.extend(table.rows)\n        else:\n            for row in table.rows:\n                data = [row.get(name, None) for name in column_names]\n                rows.append(agate.Row(data, column_names))\n    # _is_fork to tell agate that we already made things into `Row`s.\n    return agate.Table(rows, column_names, column_types, _is_fork=True)\n", "path": "core/dbt/clients/agate_helper.py"}]}