Fix postgres text handling

Fix incorrect maximum text field length on redshift
On postgres, pass through text without size information to fix archival of long text fields
Add a test that makes sure postgres archives work with very long text entries

Author: Jacob Beck

core/dbt/adapters/base/relation.py

@@ -361,7 +361,7 @@ def string_size(self):
 
         if self.dtype == 'text' or self.char_size is None:
             # char_size should never be None. Handle it reasonably just in case
-            return 255
+            return 256
         else:
             return int(self.char_size)
 

plugins/postgres/dbt/adapters/postgres/__init__.py

@@ -1,6 +1,7 @@
 # these are mostly just exports, #noqa them so flake8 will be happy
 from dbt.adapters.postgres.connections import PostgresConnectionManager  # noqa
 from dbt.adapters.postgres.connections import PostgresCredentials
+from dbt.adapters.postgres.relation import PostgresColumn  # noqa
 from dbt.adapters.postgres.impl import PostgresAdapter
 
 from dbt.adapters.base import AdapterPlugin

plugins/postgres/dbt/adapters/postgres/impl.py

@@ -1,6 +1,7 @@
 from dbt.adapters.base.meta import available
 from dbt.adapters.sql import SQLAdapter
 from dbt.adapters.postgres import PostgresConnectionManager
+from dbt.adapters.postgres import PostgresColumn
 import dbt.compat
 import dbt.exceptions
 
@@ -11,6 +12,7 @@
 
 class PostgresAdapter(SQLAdapter):
     ConnectionManager = PostgresConnectionManager
+    Column = PostgresColumn
 
     @classmethod
     def date_function(cls):

plugins/postgres/dbt/adapters/postgres/relation.py

@@ -0,0 +1,10 @@
+from dbt.adapters.base import Column
+
+
+class PostgresColumn(Column):
+    @property
+    def data_type(self):
+        # on postgres, do not convert 'text' to 'varchar()'
+        if self.dtype.lower() == 'text':
+            return self.dtype
+        return super(PostgresColumn, self).data_type

plugins/redshift/dbt/adapters/redshift/impl.py

@@ -1,10 +1,13 @@
 from dbt.adapters.postgres import PostgresAdapter
 from dbt.adapters.redshift import RedshiftConnectionManager
+from dbt.adapters.base import Column
 from dbt.logger import GLOBAL_LOGGER as logger  # noqa
 
 
 class RedshiftAdapter(PostgresAdapter):
     ConnectionManager = RedshiftConnectionManager
+    # postgres overrides this, so override the override.
+    Column = Column
 
     AdapterSpecificConfigs = frozenset({"sort_type", "dist", "sort", "bind"})
 

test/integration/004_simple_archive_test/seed_longtext.sql

@@ -0,0 +1,9 @@
+create table {database}.{schema}.super_long (
+    id INTEGER,
+    longstring TEXT,
+    updated_at TIMESTAMP WITHOUT TIME ZONE
+);
+
+insert into {database}.{schema}.super_long (id, longstring, updated_at) VALUES
+(1, 'short', current_timestamp),
+(2, repeat('a', 500), current_timestamp);

test/integration/004_simple_archive_test/test-archives-longtext/longtext.sql

@@ -0,0 +1,12 @@
+{% archive archive_actual %}
+    {{
+        config(
+            target_database=var('target_database', database),
+            target_schema=schema,
+            unique_key='id',
+            strategy='timestamp',
+            updated_at='updated_at',
+        )
+    }}
+    select * from {{database}}.{{schema}}.super_long
+{% endarchive %}

test/integration/004_simple_archive_test/test_simple_archive.py

@@ -498,3 +498,38 @@ def test__bigquery__archive_with_new_field(self):
 
             self.assertEqual(expected_name, actual_name, "names are different")
             self.assertEqual(expected_type, actual_type, "data types are different")
+
+
+class TestLongText(DBTIntegrationTest):
+
+    @property
+    def schema(self):
+        return "simple_archive_004"
+
+    @property
+    def models(self):
+        return "test/integration/004_simple_archive_test/models"
+
+    def run_archive(self):
+        return self.run_dbt(['archive'])
+
+    @property
+    def project_config(self):
+        return {
+            "archive-paths": ['test/integration/004_simple_archive_test/test-archives-longtext'],
+        }
+
+    @use_profile('postgres')
+    def test__postgres__long_text(self):
+        self.run_sql_file('test/integration/004_simple_archive_test/seed_longtext.sql')
+        results = self.run_dbt(['archive'])
+        self.assertEqual(len(results), 1)
+
+        with self.adapter.connection_named('test'):
+            status, results = self.adapter.execute(
+                'select * from {}.{}.archive_actual'.format(self.default_database, self.unique_schema()),
+                fetch=True
+            )
+        self.assertEqual(len(results), 2)
+        got_names = set(r.get('longstring') for r in results)
+        self.assertEqual(got_names, {'a' * 500, 'short'})