Do not create already-existing schemas (#2186)

CHANGELOG.md

@@ -1,3 +1,8 @@
+## dbt next (release TBD)
+
+### Fixes
+- If database quoting is enabled, do not attempt to create schemas that already exist ([#2186](https://github.com/fishtown-analytics/dbt/issues/2186), [#2187](https://github.com/fishtown-analytics/dbt/pull/2187))
+
 ## dbt 0.16.0rc2 (March 4, 2020)
 
 ### Under the hood

core/dbt/task/runnable.py

@@ -7,6 +7,7 @@
 
 from dbt.task.base import ConfiguredTask
 from dbt.adapters.base import SchemaSearchMap
+from dbt.adapters.base.relation import InformationSchema
 from dbt.adapters.factory import get_adapter
 from dbt.logger import (
     GLOBAL_LOGGER as logger,
@@ -41,6 +42,12 @@
 RUNNING_STATE = DbtProcessState('running')
 
 
+def _lower(value: Optional[str]) -> Optional[str]:
+    if value is None:
+        return value
+    return value.lower()
+
+
 def write_manifest(config, manifest):
     if dbt.flags.WRITE_JSON:
         manifest.write(os.path.join(config.target_path, MANIFEST_FILE_NAME))
@@ -406,15 +413,26 @@ def create_schemas(self, adapter, selected_uids: Iterable[str]):
                 include_policy=include_policy,
                 information_schema_view=None,
             )
-            required_databases.append(str(db_only))
+            required_databases.append(db_only)
 
         existing_schemas_lowered: Set[Tuple[str, Optional[str]]] = set()
 
-        def list_schemas(db: str) -> List[Tuple[str, str]]:
-            with adapter.connection_named(f'list_{db}'):
+        def list_schemas(info: InformationSchema) -> List[Tuple[str, str]]:
+            # the database name should never be None here (or where are we
+            # listing schemas from?)
+            if info.database is None:
+                raise InternalException(
+                    f'Got an invalid information schema of {info} (database '
+                    f'was None)'
+                )
+            database_name = info.database
+            database_quoted = str(info)
+            with adapter.connection_named(f'list_{database_name}'):
+                # we should never create a null schema, so just filter them out
                 return [
-                    (db.lower(), s.lower())
-                    for s in adapter.list_schemas(db)
+                    (database_name.lower(), s.lower())
+                    for s in adapter.list_schemas(database_quoted)
+                    if s is not None
                 ]
 
         def create_schema(db: str, schema: str) -> None:

test/integration/001_simple_copy_test/test_simple_copy.py

@@ -1,7 +1,11 @@
+import io
+import json
 import os
+from unittest import mock
 from pytest import mark
 
 from test.integration.base import DBTIntegrationTest, use_profile
+from dbt.logger import log_manager
 
 
 class BaseTestSimpleCopy(DBTIntegrationTest):
@@ -21,7 +25,6 @@ def models(self):
     def project_config(self):
         return self.seed_quote_cfg_with({})
 
-
     def seed_quote_cfg_with(self, extra):
         cfg = {
             'seeds': {
@@ -392,3 +395,54 @@ def project_config(self):
     def test_postgres_run_mixed_case(self):
         self.run_dbt()
         self.run_dbt()
+
+
+class TestQuotedDatabase(BaseTestSimpleCopy):
+
+    @property
+    def project_config(self):
+        return self.seed_quote_cfg_with({
+            'quoting': {
+                'database': True,
+            },
+            "data-paths": [self.dir("seed-initial")],
+        })
+
+    def setUp(self):
+        super().setUp()
+        self.initial_stdout = log_manager.stdout
+        self.initial_stderr = log_manager.stderr
+        self.stringbuf = io.StringIO()
+        log_manager.set_output_stream(self.stringbuf)
+
+    def tearDown(self):
+        log_manager.set_output_stream(self.initial_stdout, self.initial_stderr)
+        super().tearDown()
+
+    def seed_get_json(self, expect_pass=True):
+        self.run_dbt(
+            ['--debug', '--log-format=json', '--single-threaded', 'seed'],
+            expect_pass=expect_pass
+        )
+        logs = []
+        for line in self.stringbuf.getvalue().split('\n'):
+            try:
+                log = json.loads(line)
+            except ValueError:
+                continue
+
+            if log['extra'].get('run_state') != 'internal':
+                continue
+            logs.append(log)
+        self.assertGreater(len(logs), 0)
+        return logs
+
+    @use_profile('postgres')
+    def test_postgres_no_create_schemas(self):
+        logs = self.seed_get_json()
+        for log in logs:
+            msg = log['message']
+            self.assertFalse(
+                'create schema if not exists' in msg,
+                f'did not expect schema creation: {msg}'
+            )