From 81a4986316e1282eba82bb9879b89512346b9647 Mon Sep 17 00:00:00 2001
From: Steven Meltser <31104631+SMeltser@users.noreply.github.com>
Date: Thu, 3 Mar 2022 04:08:11 -0500
Subject: [PATCH] Fix: Create table statement parameter ordering when
 configured with backup & dist/sort  (#63)

* set BACKUP parameter before dist/sort keys
* updated changelog
* added test v1
* Finalize backup tests
* update tests
---
 CHANGELOG.md                                  |  1 +
 dbt/include/redshift/macros/adapters.sql      |  2 +-
 .../model_backup_param_before_distkey.sql     |  7 +++
 .../model_backup_param_before_sortkey.sql     |  7 +++
 .../test_backup_table_option.py               | 53 +++++++++++++++++--
 5 files changed, 64 insertions(+), 6 deletions(-)
 create mode 100644 tests/integration/backup_table_tests/models/model_backup_param_before_distkey.sql
 create mode 100644 tests/integration/backup_table_tests/models/model_backup_param_before_sortkey.sql

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 46293bf31..88b0c53e7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 ### Fixes
 - Fix test related to preventing coercion of boolean values (True,False) to numeric values (0,1) in query results ([#58](https://github.com/dbt-labs/dbt-redshift/blob/1.0.latest/CHANGELOG.md))
+- Fix table creation statement ordering when including both the BACKUP parameter as well as the dist/sort keys ([#23](https://github.com/dbt-labs/dbt-redshift/issues/60))
 - Add unique\_id field to docs generation test catalogs; a follow-on PR to core PR ([#4168](https://github.com/dbt-labs/dbt-core/pull/4618)) and core PR ([#4701](https://github.com/dbt-labs/dbt-core/pull/4701))
 
 ## dbt-redshift 1.0.0 (December 3, 2021)
diff --git a/dbt/include/redshift/macros/adapters.sql b/dbt/include/redshift/macros/adapters.sql
index e81e823df..e82931857 100644
--- a/dbt/include/redshift/macros/adapters.sql
+++ b/dbt/include/redshift/macros/adapters.sql
@@ -45,9 +45,9 @@
 
   create {% if temporary -%}temporary{%- endif %} table
     {{ relation.include(database=(not temporary), schema=(not temporary)) }}
+    {% if backup == false -%}backup no{%- endif %}
     {{ dist(_dist) }}
     {{ sort(_sort_type, _sort) }}
-    {% if backup == false -%}backup no{%- endif %}
   as (
     {{ sql }}
   );
diff --git a/tests/integration/backup_table_tests/models/model_backup_param_before_distkey.sql b/tests/integration/backup_table_tests/models/model_backup_param_before_distkey.sql
new file mode 100644
index 000000000..87c586265
--- /dev/null
+++ b/tests/integration/backup_table_tests/models/model_backup_param_before_distkey.sql
@@ -0,0 +1,7 @@
+{{
+    config(
+        materialized='table', backup=False, dist='distkey'
+    )
+}}
+
+select 1 as distkey
\ No newline at end of file
diff --git a/tests/integration/backup_table_tests/models/model_backup_param_before_sortkey.sql b/tests/integration/backup_table_tests/models/model_backup_param_before_sortkey.sql
new file mode 100644
index 000000000..380aacb5c
--- /dev/null
+++ b/tests/integration/backup_table_tests/models/model_backup_param_before_sortkey.sql
@@ -0,0 +1,7 @@
+{{
+    config(
+        materialized='table', backup=False, sort='sortkey'
+    )
+}}
+
+select 1 as sortkey
\ No newline at end of file
diff --git a/tests/integration/backup_table_tests/test_backup_table_option.py b/tests/integration/backup_table_tests/test_backup_table_option.py
index b31da3adf..e32bca803 100644
--- a/tests/integration/backup_table_tests/test_backup_table_option.py
+++ b/tests/integration/backup_table_tests/test_backup_table_option.py
@@ -26,11 +26,12 @@ def check_backup_param_template(self, test_table_name, backup_is_expected):
         # Use raw DDL statement to confirm backup is set correctly on new table
         with open('target/run/test/models/{}.sql'.format(test_table_name), 'r') as ddl_file:
             ddl_statement = ddl_file.readlines()
-            self.assertEqual('backup no' not in ' '.join(ddl_statement).lower(), backup_is_expected)
+            lowercase_statement = ' '.join(ddl_statement).lower()
+            self.assertEqual('backup no' not in lowercase_statement, backup_is_expected)
 
     @use_profile('redshift')
     def test__redshift_backup_table_option(self):
-        self.assertEqual(len(self.run_dbt()), 4)
+        self.assertEqual(len(self.run_dbt()), 6)
 
         # model_backup_undefined should not contain a BACKUP NO parameter in the table DDL
         self.check_backup_param_template('model_backup_undefined', True)
@@ -44,7 +45,6 @@ def test__redshift_backup_table_option(self):
         # Any view should not contain a BACKUP NO parameter, regardless of the specified config (create will fail)
         self.check_backup_param_template('model_backup_true_view', True)
 
-
 class TestBackupTableOptionProjectFalse(DBTIntegrationTest):
     @property
     def schema(self):
@@ -71,11 +71,12 @@ def check_backup_param_template(self, test_table_name, backup_is_expected):
         # Use raw DDL statement to confirm backup is set correctly on new table
         with open('target/run/test/models/{}.sql'.format(test_table_name), 'r') as ddl_file:
             ddl_statement = ddl_file.readlines()
-            self.assertEqual('backup no' not in ' '.join(ddl_statement).lower(), backup_is_expected)
+            lowercase_statement = ' '.join(ddl_statement).lower()
+            self.assertEqual('backup no' not in lowercase_statement, backup_is_expected)
 
     @use_profile('redshift')
     def test__redshift_backup_table_option_project_config_false(self):
-        self.assertEqual(len(self.run_dbt()), 4)
+        self.assertEqual(len(self.run_dbt()), 6)
 
         # model_backup_undefined should contain a BACKUP NO parameter in the table DDL
         self.check_backup_param_template('model_backup_undefined', False)
@@ -88,3 +89,45 @@ def test__redshift_backup_table_option_project_config_false(self):
 
         # Any view should not contain a BACKUP NO parameter, regardless of the specified config (create will fail)
         self.check_backup_param_template('model_backup_true_view', True)
+
+class TestBackupTableOptionOrder(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return 'backup_table_tests'
+
+    @staticmethod
+    def dir(path):
+        return os.path.normpath(path)
+
+    @property
+    def models(self):
+        return self.dir("models")
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2
+        }
+
+    def check_backup_param_template(self, test_table_name, backup_flag_is_expected):
+        # Use raw DDL statement to confirm backup is set correctly on new table
+        with open('target/run/test/models/{}.sql'.format(test_table_name), 'r') as ddl_file:
+            ddl_statement = ddl_file.readlines()
+            lowercase_statement = ' '.join(ddl_statement).lower()
+            self.assertEqual('backup no' not in lowercase_statement, backup_flag_is_expected)
+            if backup_flag_is_expected:
+                distkey_index = lowercase_statement.find('distkey')
+                sortkey_index = lowercase_statement.find('sortkey')
+                backup_index = lowercase_statement.find('backup no')
+                self.assertEqual((backup_index < distkey_index) or distkey_index == -1, backup_flag_is_expected)
+                self.assertEqual((backup_index < sortkey_index) or sortkey_index == -1, backup_flag_is_expected)
+
+    @use_profile('redshift')
+    def test__redshift_backup_table_option_project_config_false(self):
+        self.assertEqual(len(self.run_dbt()), 6)
+
+        # model_backup_param_before_distkey should contain a BACKUP NO parameter which precedes a DISTKEY in the table ddl
+        self.check_backup_param_template('model_backup_param_before_distkey', False)
+        
+        # model_backup_param_before_sortkey should contain a BACKUP NO parameter which precedes a SORTKEY in the table ddl
+        self.check_backup_param_template('model_backup_param_before_sortkey', False)
\ No newline at end of file