fix:[SNOW-3444080] Preserve customer's original key order inside profiles.yml and env.yml; Refactor some code

RELEASE-NOTES.md

@@ -22,6 +22,7 @@
 
 ## Fixes and improvements
 * `snow app setup` no longer writes `build_compute_pool` / `service_compute_pool` to the generated `snowflake.yml`, and no longer reads the `DEFAULT_SNOWFLAKE_APPS_BUILD_COMPUTE_POOL` / `DEFAULT_SNOWFLAKE_APPS_SERVICE_COMPUTE_POOL` account parameters. Snowflake App Runtime services now always run on server-managed compute pools. Existing projects that set these fields in `snowflake.yml` continue to be honored by `snow app deploy`.
+* `snow dbt deploy` now preserves the key order of `profiles.yml` when staging the file. Previously `yaml.safe_dump` sorted keys alphabetically, silently reordering the customer's input.
 
 
 # v3.21.0

src/snowflake/cli/_plugins/dbt/manager.py

@@ -50,6 +50,10 @@
 _JINJA_EXPR = re.compile(r"\{\{.*?\}\}", re.DOTALL)
 
 
+def _plural(count: int, word: str) -> str:
+    return word if count == 1 else f"{word}s"
+
+
 def _reject_control_chars(value: Optional[str], flag_name: str) -> Optional[str]:
     if value is not None and _CONTROL_CHAR_RE.search(value):
         raise CliError(
@@ -523,37 +527,42 @@ def _validate_role(self, role_name: str) -> bool:
 
     @staticmethod
     def _prepare_profiles_file(profiles_path: Path, tmp_path: Path):
-        # We need to copy profiles.yml file (not symlink) in order to redact
-        # any comments without changing original file. This can be achieved
-        # with pyyaml, which looses comments while reading a yaml file
         source_profiles_file = SecurePath(profiles_path / PROFILES_FILENAME)
         target_profiles_file = SecurePath(tmp_path / PROFILES_FILENAME)
         if target_profiles_file.exists():
             target_profiles_file.unlink()
-        with source_profiles_file.open(
-            read_file_limit_mb=DEFAULT_SIZE_LIMIT_MB
-        ) as sfd, target_profiles_file.open(mode="w") as tfd:
-            yaml.safe_dump(yaml.safe_load(sfd), tfd)
+        with source_profiles_file.open(read_file_limit_mb=DEFAULT_SIZE_LIMIT_MB) as sfd:
+            raw = sfd.read()
+        try:
+            yaml.load(raw, Loader=_NoDuplicatesSafeLoader)
+        except yaml.constructor.ConstructorError as e:
+            raise CliError(f"Failed to parse {PROFILES_FILENAME}: {e.problem}")
+        except yaml.YAMLError as e:
+            raise CliError(f"{PROFILES_FILENAME} is not valid YAML: {e}")
+        with target_profiles_file.open(mode="w") as tfd:
+            tfd.write(raw)
 
     @staticmethod
-    def _validate_and_parse_env_file(env_file: SecurePath) -> Optional[dict]:
-        """Parse and validate env.yml, rejecting duplicate keys and invalid YAML."""
+    def _validate_and_parse_env_file(env_file: SecurePath) -> str:
+        """Validate env.yml and return its raw contents for verbatim staging."""
         with env_file.open(read_file_limit_mb=DEFAULT_SIZE_LIMIT_MB) as sfd:
-            try:
-                return yaml.load(sfd, Loader=_NoDuplicatesSafeLoader)
-            except yaml.constructor.ConstructorError as e:
-                raise CliError(f"Failed to parse {ENV_FILENAME}: {e.problem}")
-            except yaml.YAMLError as e:
-                raise CliError(f"{ENV_FILENAME} is not valid YAML: {e}")
+            raw = sfd.read()
+        try:
+            yaml.load(raw, Loader=_NoDuplicatesSafeLoader)
+        except yaml.constructor.ConstructorError as e:
+            raise CliError(f"Failed to parse {ENV_FILENAME}: {e.problem}")
+        except yaml.YAMLError as e:
+            raise CliError(f"{ENV_FILENAME} is not valid YAML: {e}")
+        return raw
 
     @staticmethod
-    def _write_env_file(content: dict, tmp_path: Path):
-        """Write the parsed env.yml into the staging dir (comments already dropped)."""
+    def _write_env_file(content: str, tmp_path: Path):
+        """Write the validated env.yml into the staging dir verbatim."""
         target_env_file = SecurePath(tmp_path / ENV_FILENAME)
         if target_env_file.exists():
             target_env_file.unlink()
         with target_env_file.open(mode="w") as tfd:
-            yaml.safe_dump(content, tfd)
+            tfd.write(content)
 
     def execute(
         self,
@@ -579,28 +588,25 @@ def execute(
         if use_shell_env_vars:
             shell_vars, dropped_secret_count, skipped_count = _collect_shell_env_vars()
             if dropped_secret_count:
-                _var = "variable" if dropped_secret_count == 1 else "variables"
                 cli_console.message(
                     f"--use-shell-env-vars: dropped {dropped_secret_count} "
-                    f"{DBT_ENV_SECRET_PREFIX}* environment {_var} from "
+                    f"{DBT_ENV_SECRET_PREFIX}* environment {_plural(dropped_secret_count, 'variable')} from "
                     "shell. To forward secrets, use the secrets: block in "
                     "env.yml referencing a Snowflake SECRET object, or if "
                     "those are not sensitive, pass them explicitly via "
                     "--env-vars."
                 )
             if skipped_count:
-                _var = "variable" if skipped_count == 1 else "variables"
                 cli_console.message(
                     f"--use-shell-env-vars: skipped {skipped_count} DBT_* shell "
-                    f"environment {_var} that can't be forwarded; keys "
+                    f"environment {_plural(skipped_count, 'variable')} that can't be forwarded; keys "
                     "must be uppercase and contain only letters, digits, and "
                     "underscores (e.g. export DBT_FOO, not DBT_Foo)."
                 )
             if shell_vars:
-                _var = "variable" if len(shell_vars) == 1 else "variables"
                 cli_console.warning(
                     f"--use-shell-env-vars: forwarded {len(shell_vars)} shell "
-                    f"environment {_var} starting with DBT_* into query text. "
+                    f"environment {_plural(len(shell_vars), 'variable')} starting with DBT_* into query text. "
                     "Never put credentials, tokens, passwords, or other "
                     "confidential data in shell environment variables with "
                     "the DBT_ prefix."

tests/dbt/test_manager.py

@@ -615,36 +615,52 @@ def test_prepare_profiles_file_replaces_existing_symlink_with_file(
 
         assert tmp_profiles_file.is_symlink() is False
 
-    def test_prepare_profiles_file_removes_all_comments(
-        self, tmp_path_factory, profile
-    ):
+    def test_prepare_profiles_file_preserves_comments(self, tmp_path_factory, profile):
         profiles_path = tmp_path_factory.mktemp("profiles")
         dbt_profiles_file = profiles_path / PROFILES_FILENAME
-        # not a comment - valid ones start exactly with `# `
-        profile["#key"] = "#value"
         dbt_profiles_file.write_text(yaml.dump(profile))
         with open(dbt_profiles_file, "a") as fp:
-            fp.write("# full line comment\n")
-            fp.write("key: with # comment\n")
-            fp.write("another_key: with # comment # and one more # and more\n")
-            fp.write("#         password: 123\n")
+            fp.write("# developer hint\n")
+            fp.write("key: value # inline comment\n")
 
         tmp_dbt_path = tmp_path_factory.mktemp("dbt")
 
         DBTManager._prepare_profiles_file(profiles_path, tmp_dbt_path)  # noqa: SLF001
 
-        assert tmp_dbt_path.is_symlink() is False
-        with open(tmp_dbt_path / PROFILES_FILENAME, "r") as fp:
-            actual = yaml.safe_load(fp)
-        expected = profile | {"key": "with", "another_key": "with"}
-        assert actual == expected
+        staged = (tmp_dbt_path / PROFILES_FILENAME).read_text()
+        assert "# developer hint" in staged
+        assert "# inline comment" in staged
+
+    def test_prepare_profiles_file_preserves_key_order(self, tmp_path_factory):
+        # Write keys in a deliberately non-alphabetical order.
+        # Alphabetical would be: account < database < password < role < schema < threads < type < user < warehouse
+        # We write:              type < threads < account < user < password < role < warehouse < database < schema
+        raw_yaml = (
+            "dbt_project:\n"
+            "  target: dev\n"
+            "  outputs:\n"
+            "    dev:\n"
+            "      type: snowflake\n"
+            "      threads: 2\n"
+            "      account: acct\n"
+            "      user: usr\n"
+            "      password: pw\n"
+            "      role: ACCOUNTADMIN\n"
+            "      warehouse: XSMALL\n"
+            "      database: MYDB\n"
+            "      schema: PUBLIC\n"
+        )
+        profiles_path = tmp_path_factory.mktemp("profiles")
+        (profiles_path / PROFILES_FILENAME).write_text(raw_yaml)
+        tmp_dbt_path = tmp_path_factory.mktemp("dbt")
+
+        DBTManager._prepare_profiles_file(profiles_path, tmp_dbt_path)  # noqa: SLF001
 
-        # pyyaml ignores comments, so as safety net we need to check that comments were removed on lower level
-        with open(tmp_dbt_path / PROFILES_FILENAME, "r") as fp:
-            for line in fp:
-                assert "comment" not in line
-                assert "password" not in line
-                assert "# " not in line
+        staged = (tmp_dbt_path / PROFILES_FILENAME).read_text()
+        # type before account (non-alphabetical), threads before database
+        assert staged.index("type:") < staged.index("account:")
+        assert staged.index("threads:") < staged.index("database:")
+        assert staged.index("warehouse:") < staged.index("database:")
 
     def test_validate_profiles_with_valid_default_target(
         self, mock_validate_role, project_path, profile
@@ -1259,13 +1275,15 @@ def test_write_env_file_replaces_existing_file(self, tmp_path_factory, env_yml):
             actual = yaml.safe_load(fp)
         assert actual == env_yml
 
-    def test_load_and_write_env_file_removes_comments(self, tmp_path_factory, env_yml):
+    def test_load_and_write_env_file_preserves_comments(
+        self, tmp_path_factory, env_yml
+    ):
         env_path = tmp_path_factory.mktemp("envs")
         env_file = env_path / ENV_FILENAME
         env_file.write_text(yaml.dump(env_yml))
         with open(env_file, "a") as fp:
-            fp.write("# secret hint left by developer\n")
-            fp.write("extra_key: with # trailing comment\n")
+            fp.write("# developer note\n")
+            fp.write("extra_key: value # inline remark\n")
 
         tmp_dbt_path = tmp_path_factory.mktemp("dbt")
 
@@ -1274,10 +1292,38 @@ def test_load_and_write_env_file_removes_comments(self, tmp_path_factory, env_ym
         )
         DBTManager._write_env_file(content, tmp_dbt_path)  # noqa: SLF001
 
-        with open(tmp_dbt_path / ENV_FILENAME) as fp:
-            for line in fp:
-                assert "# " not in line
-                assert "secret hint" not in line
+        staged = (tmp_dbt_path / ENV_FILENAME).read_text()
+        assert "# developer note" in staged
+        assert "# inline remark" in staged
+
+    def test_write_env_file_preserves_key_order(self, tmp_path_factory):
+        # Write DBT_* keys in reverse-alphabetical order.
+        # Alphabetical would be: ALPHA < BETA < GAMMA < ZETA
+        # We write:              ZETA < GAMMA < ALPHA < BETA
+        raw_yaml = (
+            "env_config:\n"
+            "  environments:\n"
+            "  - name: dev\n"
+            "    env:\n"
+            "      DBT_ZETA: z\n"
+            "      DBT_GAMMA: g\n"
+            "      DBT_ALPHA: a\n"
+            "      DBT_BETA: b\n"
+        )
+        env_path = tmp_path_factory.mktemp("envs")
+        (env_path / ENV_FILENAME).write_text(raw_yaml)
+        tmp_dbt_path = tmp_path_factory.mktemp("dbt")
+
+        content = DBTManager._validate_and_parse_env_file(  # noqa: SLF001
+            SecurePath(env_path / ENV_FILENAME)
+        )
+        DBTManager._write_env_file(content, tmp_dbt_path)  # noqa: SLF001
+
+        staged = (tmp_dbt_path / ENV_FILENAME).read_text()
+        # ZETA before GAMMA before ALPHA before BETA (reverse-alphabetical preserved)
+        assert staged.index("DBT_ZETA") < staged.index("DBT_GAMMA")
+        assert staged.index("DBT_GAMMA") < staged.index("DBT_ALPHA")
+        assert staged.index("DBT_ALPHA") < staged.index("DBT_BETA")
 
     def test_validate_and_parse_env_file_rejects_duplicate_keys(self, tmp_path_factory):
         env_path = tmp_path_factory.mktemp("envs")

tests_integration/test_dbt.py

@@ -477,6 +477,85 @@ def test_deploy_with_env_file_dir(
         _assert_default_environment(name, runner, "dev")
 
 
+@pytest.mark.integration
+@pytest.mark.qa_only
+def test_deploy_preserves_yaml_key_order(
+    runner,
+    snowflake_session,
+    test_database,
+    project_directory,
+    tmp_path,
+):
+    """profiles.yml and env.yml must keep the customer's key order after staging.
+
+    yaml.safe_dump sorts keys alphabetically unless sort_keys=False is passed.
+    This test verifies that the staged copies preserve the original order.
+    """
+    with project_directory("dbt_project") as root_dir:
+        ts = int(datetime.datetime.now().timestamp())
+        name = f"dbt_yaml_order_{ts}"
+
+        # Write profiles.yml with type/threads first — non-alphabetical
+        # (alphabetical: account < database < password < role < schema < threads < type < user < warehouse)
+        raw_profiles = (
+            f"dbt_integration_project:\n"
+            f"  target: dev\n"
+            f"  outputs:\n"
+            f"    dev:\n"
+            f"      type: snowflake\n"
+            f"      threads: 2\n"
+            f"      account: ''\n"
+            f"      user: ''\n"
+            f"      password: ''\n"
+            f"      role: {snowflake_session.role}\n"
+            f"      warehouse: {snowflake_session.warehouse}\n"
+            f"      database: {snowflake_session.database}\n"
+            f"      schema: {snowflake_session.schema}\n"
+        )
+        (root_dir / PROFILES_FILENAME).write_text(raw_profiles)
+
+        # Write env.yml with DBT_* keys in reverse-alphabetical order
+        # (alphabetical: ALPHA < BETA < GAMMA < ZETA)
+        raw_env = (
+            "env_config:\n"
+            "  default_environment: dev\n"
+            "  environments:\n"
+            "  - name: dev\n"
+            "    env:\n"
+            "      DBT_ZETA: z\n"
+            "      DBT_GAMMA: g\n"
+            "      DBT_ALPHA: a\n"
+            "      DBT_BETA: b\n"
+        )
+        (root_dir / ENV_FILENAME).write_text(raw_env)
+
+        result = runner.invoke_with_connection_json(["dbt", "deploy", name])
+        assert result.exit_code == 0, result.output
+
+        db = snowflake_session.database
+        schema = snowflake_session.schema
+        stage_base = f"snow://dbt/{db}.{schema}.{name}/versions/version$1"
+
+        # Download staged profiles.yml and verify key order
+        result = runner.invoke_with_connection(
+            ["sql", "-q", f"GET {stage_base}/profiles.yml file://{tmp_path}/"]
+        )
+        assert result.exit_code == 0, result.output
+        profiles_text = (tmp_path / "profiles.yml").read_text()
+        assert profiles_text.index("type:") < profiles_text.index("account:")
+        assert profiles_text.index("threads:") < profiles_text.index("database:")
+
+        # Download staged env.yml and verify key order
+        result = runner.invoke_with_connection(
+            ["sql", "-q", f"GET {stage_base}/env.yml file://{tmp_path}/"]
+        )
+        assert result.exit_code == 0, result.output
+        env_text = (tmp_path / "env.yml").read_text()
+        assert env_text.index("DBT_ZETA") < env_text.index("DBT_GAMMA")
+        assert env_text.index("DBT_GAMMA") < env_text.index("DBT_ALPHA")
+        assert env_text.index("DBT_ALPHA") < env_text.index("DBT_BETA")
+
+
 @pytest.mark.integration
 def test_execute_with_target(
     runner,