Support new agate Integer data_type in adapter code (#9004)

Add test for empty seed file with only headers

.changes/unreleased/Fixes-20231031-144837.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix compilation exception running empty seed file and support new Integer agate data_type
+time: 2023-10-31T14:48:37.774871-04:00
+custom:
+  Author: gshank
+  Issue: "8895"

core/dbt/adapters/base/impl.py

@@ -51,6 +51,7 @@
     get_column_value_uncased,
     merge_tables,
     table_from_rows,
+    Integer,
 )
 from dbt.clients.jinja import MacroGenerator
 from dbt.contracts.graph.manifest import Manifest, MacroManifest
@@ -962,6 +963,17 @@ def convert_number_type(cls, agate_table: agate.Table, col_idx: int) -> str:
         """
         raise NotImplementedError("`convert_number_type` is not implemented for this adapter!")
 
+    @classmethod
+    def convert_integer_type(cls, agate_table: agate.Table, col_idx: int) -> str:
+        """Return the type in the database that best maps to the agate.Number
+        type for the given agate table and column index.
+
+        :param agate_table: The table
+        :param col_idx: The index into the agate table for the column.
+        :return: The name of the type in the database
+        """
+        return "integer"
+
     @classmethod
     @abc.abstractmethod
     def convert_boolean_type(cls, agate_table: agate.Table, col_idx: int) -> str:
@@ -1019,6 +1031,7 @@ def convert_type(cls, agate_table: agate.Table, col_idx: int) -> Optional[str]:
     def convert_agate_type(cls, agate_table: agate.Table, col_idx: int) -> Optional[str]:
         agate_type: Type = agate_table.column_types[col_idx]
         conversions: List[Tuple[Type, Callable[..., str]]] = [
+            (Integer, cls.convert_integer_type),
             (agate.Text, cls.convert_text_type),
             (agate.Number, cls.convert_number_type),
             (agate.Boolean, cls.convert_boolean_type),

core/dbt/adapters/sql/impl.py

@@ -75,6 +75,10 @@ def convert_number_type(cls, agate_table: agate.Table, col_idx: int) -> str:
         decimals = agate_table.aggregate(agate.MaxPrecision(col_idx))  # type: ignore[attr-defined]
         return "float8" if decimals else "integer"
 
+    @classmethod
+    def convert_integer_type(cls, agate_table: agate.Table, col_idx: int) -> str:
+        return "integer"
+
     @classmethod
     def convert_boolean_type(cls, agate_table: agate.Table, col_idx: int) -> str:
         return "boolean"

tests/adapter/dbt/tests/adapter/simple_seed/test_seed.py

@@ -365,3 +365,26 @@ def seeds(self, test_data_dir):
     def test_simple_seed(self, project):
         results = run_dbt(["seed"])
         assert len(results) == 3
+
+
+class BaseTestEmptySeed:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "seeds": {
+                "quote_columns": False,
+            },
+        }
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {"empty_with_header.csv": "a,b,c"}
+
+    def test_empty_seeds(self, project):
+        # Should create an empty table and not fail
+        results = run_dbt(["seed"])
+        assert len(results) == 1
+
+
+class TestEmptySeed(BaseTestEmptySeed):
+    pass

tests/unit/mock_adapter.py

@@ -55,6 +55,9 @@ def convert_text_type(self, *args, **kwargs):
         def convert_number_type(self, *args, **kwargs):
             return self.responder.convert_number_type(*args, **kwargs)
 
+        def convert_integer_type(self, *args, **kwargs):
+            return self.responder.convert_integer_type(*args, **kwargs)
+
         def convert_boolean_type(self, *args, **kwargs):
             return self.responder.convert_boolean_type(*args, **kwargs)