SNOW-2306184: config-ng refactor

.github/workflows/test.yaml

@@ -12,7 +12,7 @@ on:
       - features/*
 
 env:
-  TERM: unknown  # Disables colors in rich
+  TERM: unknown # Disables colors in rich
 
 permissions:
   contents: read
@@ -43,3 +43,29 @@ jobs:
       - name: Test with hatch
         run: hatch run test-cov
       - uses: codecov/codecov-action@18283e04ce6e62d37312384ff67231eb8fd56d24
+
+  tests-config-ng:
+    needs: define-matrix
+    strategy:
+      fail-fast: true
+      matrix:
+        os: ${{ fromJSON(needs.define-matrix.outputs.os) }}
+        python-version: ${{ fromJSON(needs.define-matrix.outputs.python) }}
+    runs-on: ${{ matrix.os }}
+    env:
+      SNOWFLAKE_CLI_CONFIG_V2_ENABLED: 1
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install hatch
+        run: |
+          pip install -U click==8.2.1 hatch
+          hatch env create default
+      - name: Test with hatch
+        run: hatch run test-cov
+      - uses: codecov/codecov-action@18283e04ce6e62d37312384ff67231eb8fd56d24

.github/workflows/test_trusted.yaml

@@ -46,3 +46,31 @@ jobs:
           SNOWFLAKE_CONNECTIONS_INTEGRATION_DATABASE: ${{ secrets.SNOWFLAKE_DATABASE }}
           SNOWFLAKE_CONNECTIONS_INTEGRATION_PRIVATE_KEY_RAW: ${{ secrets.SNOWFLAKE_PRIVATE_KEY_RAW }}
         run: python -m hatch run ${{ inputs.hatch-run }}
+
+  tests-trusted-ng:
+    runs-on: ${{ inputs.runs-on }}
+    env:
+      SNOWFLAKE_CLI_CONFIG_V2_ENABLED: 1
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ inputs.python-version }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip click==8.2.1 hatch
+          python -m hatch env create ${{ inputs.python-env }}
+      - name: Run integration tests
+        env:
+          GH_TOKEN: ${{ secrets.SNOWFLAKE_GITHUB_TOKEN }}
+          TERM: unknown
+          SNOWFLAKE_CONNECTIONS_INTEGRATION_AUTHENTICATOR: SNOWFLAKE_JWT
+          SNOWFLAKE_CONNECTIONS_INTEGRATION_HOST: ${{ secrets.SNOWFLAKE_HOST }}
+          SNOWFLAKE_CONNECTIONS_INTEGRATION_USER: ${{ secrets.SNOWFLAKE_USER }}
+          SNOWFLAKE_CONNECTIONS_INTEGRATION_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
+          SNOWFLAKE_CONNECTIONS_INTEGRATION_DATABASE: ${{ secrets.SNOWFLAKE_DATABASE }}
+          SNOWFLAKE_CONNECTIONS_INTEGRATION_PRIVATE_KEY_RAW: ${{ secrets.SNOWFLAKE_PRIVATE_KEY_RAW }}
+        run: python -m hatch run ${{ inputs.hatch-run }}

RELEASE-NOTES.md

@@ -15,6 +15,7 @@
  -->
 # Unreleased version
 ## Backward incompatibility
+* **Configuration System (NG)**: File-based configuration sources (`snowsql_config`, `cli_config_toml`, `connections_toml`) now use **connection-level replacement** instead of field-level merging. When a later file source defines a connection, it completely replaces the entire connection from earlier file sources - fields are NOT inherited. Environment variables and CLI arguments continue to overlay per-field on top of the file-derived connection. This provides more predictable configuration behavior where file-defined connections are atomic units.
 
 ## Deprecations
 

src/snowflake/cli/_app/telemetry.py

@@ -61,6 +61,11 @@ class CLITelemetryField(Enum):
     COMMAND_CI_ENVIRONMENT = "command_ci_environment"
     # Configuration
     CONFIG_FEATURE_FLAGS = "config_feature_flags"
+    CONFIG_PROVIDER_TYPE = "config_provider_type"
+    CONFIG_SOURCES_USED = "config_sources_used"
+    CONFIG_SOURCE_WINS = "config_source_wins"
+    CONFIG_TOTAL_KEYS_RESOLVED = "config_total_keys_resolved"
+    CONFIG_KEYS_WITH_OVERRIDES = "config_keys_with_overrides"
     # Metrics
     COUNTERS = "counters"
     SPANS = "spans"
@@ -219,6 +224,55 @@ def python_version() -> str:
     return f"{py_ver.major}.{py_ver.minor}.{py_ver.micro}"
 
 
+def _get_config_telemetry() -> TelemetryDict:
+    """Get configuration resolution telemetry data."""
+    try:
+        from snowflake.cli.api.config_ng.telemetry_integration import (
+            get_config_telemetry_payload,
+        )
+        from snowflake.cli.api.config_provider import (
+            AlternativeConfigProvider,
+            get_config_provider_singleton,
+        )
+
+        provider = get_config_provider_singleton()
+
+        # Identify which config provider is being used
+        provider_type = (
+            "ng" if isinstance(provider, AlternativeConfigProvider) else "legacy"
+        )
+
+        result: TelemetryDict = {CLITelemetryField.CONFIG_PROVIDER_TYPE: provider_type}
+
+        # Get detailed telemetry if using ng config
+        if isinstance(provider, AlternativeConfigProvider):
+            provider._ensure_initialized()  # noqa: SLF001
+            payload = get_config_telemetry_payload(provider._resolver)  # noqa: SLF001
+
+            # Map payload keys to telemetry fields
+            if payload:
+                if "config_sources_used" in payload:
+                    result[CLITelemetryField.CONFIG_SOURCES_USED] = payload[
+                        "config_sources_used"
+                    ]
+                if "config_source_wins" in payload:
+                    result[CLITelemetryField.CONFIG_SOURCE_WINS] = payload[
+                        "config_source_wins"
+                    ]
+                if "config_total_keys_resolved" in payload:
+                    result[CLITelemetryField.CONFIG_TOTAL_KEYS_RESOLVED] = payload[
+                        "config_total_keys_resolved"
+                    ]
+                if "config_keys_with_overrides" in payload:
+                    result[CLITelemetryField.CONFIG_KEYS_WITH_OVERRIDES] = payload[
+                        "config_keys_with_overrides"
+                    ]
+
+        return result
+    except Exception:
+        return {}
+
+
 class CLITelemetryClient:
     @property
     def _ctx(self) -> _CliGlobalContextAccess:
@@ -239,6 +293,7 @@ def generate_telemetry_data_dict(
                 k: str(v) for k, v in get_feature_flags_section().items()
             },
             **_find_command_info(),
+            **_get_config_telemetry(),
             **telemetry_payload,
         }
         # To map Enum to string, so we don't have to use .value every time

src/snowflake/cli/_plugins/connection/commands.py

@@ -94,11 +94,32 @@ def _mask_sensitive_parameters(connection_params: dict):
 
 
 @app.command(name="list")
-def list_connections(**options) -> CommandResult:
+def list_connections(
+    all_sources: bool = typer.Option(
+        False,
+        "--all",
+        "-a",
+        help="Include connections from all sources (environment variables, SnowSQL config). "
+        "By default, only shows connections from configuration files.",
+    ),
+    **options,
+) -> CommandResult:
     """
     Lists configured connections.
     """
-    connections = get_all_connections()
+    from snowflake.cli.api.config_provider import (
+        get_config_provider_singleton,
+        is_alternative_config_enabled,
+    )
+
+    # Use provider directly for config_ng to pass the flag
+    if is_alternative_config_enabled():
+        provider = get_config_provider_singleton()
+        connections = provider.get_all_connections(include_env_connections=all_sources)
+    else:
+        # Legacy provider ignores the flag
+        connections = get_all_connections()
+
     default_connection = get_default_connection_name()
     result = (
         {

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

@@ -18,7 +18,6 @@
 import yaml
 from snowflake.cli._plugins.stage.manager import StageManager
 from snowflake.cli.api.artifacts.upload import sync_artifacts_with_stage
-from snowflake.cli.api.commands.utils import parse_key_value_variables
 from snowflake.cli.api.console.console import cli_console
 from snowflake.cli.api.constants import (
     DEFAULT_SIZE_LIMIT_MB,
@@ -102,8 +101,15 @@ def execute(
             if configuration:
                 query += f" CONFIGURATION {configuration}"
             if variables:
+                from snowflake.cli.api.commands.common import Variable
+                from snowflake.cli.api.config_ng import get_merged_variables
+
+                # Get merged variables from SnowSQL config and CLI -D parameters
+                merged_vars_dict = get_merged_variables(variables)
+                # Convert dict to List[Variable] for compatibility with parse_execute_variables
+                parsed_variables = [Variable(k, v) for k, v in merged_vars_dict.items()]
                 query += StageManager.parse_execute_variables(
-                    parse_key_value_variables(variables)
+                    parsed_variables
                 ).removeprefix(" using")
             stage_path = StagePath.from_stage_str(from_stage)
             query += f" FROM {stage_path.absolute_path()}"

src/snowflake/cli/_plugins/helpers/commands.py

@@ -30,6 +30,7 @@
     get_all_connections,
     set_config_value,
 )
+from snowflake.cli.api.config_provider import ALTERNATIVE_CONFIG_ENV_VAR
 from snowflake.cli.api.console import cli_console
 from snowflake.cli.api.output.types import (
     CollectionResult,
@@ -317,3 +318,89 @@ def check_snowsql_env_vars(**options):
 
     results.append(MessageResult(summary))
     return MultipleResults(results)
+
+
+@app.command(
+    name="show-config-sources",
+    requires_connection=False,
+    hidden=os.environ.get(ALTERNATIVE_CONFIG_ENV_VAR, "").lower()
+    not in ("1", "true", "yes", "on"),
+)
+def show_config_sources(
+    key: Optional[str] = typer.Argument(
+        None,
+        help="Specific configuration key to show resolution for (e.g., 'account', 'user'). If not provided, shows summary for all keys.",
+    ),
+    show_details: bool = typer.Option(
+        False,
+        "--show-details",
+        "-d",
+        help="Show detailed resolution chains for all sources consulted.",
+    ),
+    export_file: Optional[Path] = typer.Option(
+        None,
+        "--export",
+        "-e",
+        help="Export complete resolution history to JSON file for support or debugging.",
+        file_okay=True,
+        dir_okay=False,
+    ),
+    **options,
+) -> CommandResult:
+    """
+    Show where configuration values come from.
+
+    This command displays the configuration resolution process, showing which
+    source (CLI arguments, environment variables, or config files) provided
+    each configuration value. Useful for debugging configuration issues.
+
+    Examples:
+
+        # Show summary of all configuration resolution
+        snow helpers show-config-sources
+
+        # Show detailed resolution for all keys
+        snow helpers show-config-sources --show-details
+
+        # Show resolution for a specific key
+        snow helpers show-config-sources account
+
+        # Show detailed resolution for a specific key
+        snow helpers show-config-sources account --show-details
+
+        # Export complete resolution history to file
+        snow helpers show-config-sources --export config_debug.json
+
+    Note: This command requires the enhanced configuration system to be enabled.
+    Set SNOWFLAKE_CLI_CONFIG_V2_ENABLED=true to enable it.
+    """
+    from snowflake.cli.api.config_ng import (
+        export_resolution_history,
+        is_resolution_logging_available,
+    )
+    from snowflake.cli.api.config_ng.resolution_logger import (
+        get_configuration_explanation_results,
+    )
+
+    if not is_resolution_logging_available():
+        return MessageResult(
+            f"⚠️  Configuration resolution logging is not available.\n\n"
+            f"To enable it, set the environment variable:\n"
+            f"    export {ALTERNATIVE_CONFIG_ENV_VAR}=true\n\n"
+            f"Then run this command again to see where configuration values come from."
+        )
+
+    # Export if requested
+    if export_file:
+        success = export_resolution_history(export_file)
+        if not success:
+            return MessageResult(
+                f"❌ Failed to export resolution history to {export_file}"
+            )
+        return MessageResult(
+            f"✅ Resolution history exported to: {export_file}\n\n"
+            f"This file contains complete details about configuration resolution "
+            f"and can be attached to support tickets."
+        )
+
+    return get_configuration_explanation_results(key=key, verbose=show_details)

src/snowflake/cli/_plugins/sql/commands.py

@@ -29,7 +29,6 @@
 )
 from snowflake.cli.api.commands.overrideable_parameter import OverrideableOption
 from snowflake.cli.api.commands.snow_typer import SnowTyperFactory
-from snowflake.cli.api.commands.utils import parse_key_value_variables
 from snowflake.cli.api.exceptions import CliArgumentError
 from snowflake.cli.api.output.types import (
     CommandResult,
@@ -136,9 +135,9 @@ def execute_sql(
     The command supports variable substitution that happens on client-side.
     """
 
-    data = {}
-    if data_override:
-        data = {v.key: v.value for v in parse_key_value_variables(data_override)}
+    from snowflake.cli.api.config_ng import get_merged_variables
+
+    data = get_merged_variables(data_override)
 
     template_syntax_config = _parse_template_syntax_config(enabled_templating)
 

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

@@ -38,7 +38,6 @@
     OnErrorType,
     Variable,
 )
-from snowflake.cli.api.commands.utils import parse_key_value_variables
 from snowflake.cli.api.console import cli_console
 from snowflake.cli.api.constants import PYTHON_3_12
 from snowflake.cli.api.exceptions import CliError
@@ -608,7 +607,12 @@ def execute(
             filtered_file_list, key=lambda f: (path.dirname(f), path.basename(f))
         )
 
-        parsed_variables = parse_key_value_variables(variables)
+        from snowflake.cli.api.config_ng import get_merged_variables
+
+        # Get merged variables from SnowSQL config and CLI -D parameters
+        merged_vars_dict = get_merged_variables(variables)
+        # Convert dict back to List[Variable] for compatibility with existing methods
+        parsed_variables = [Variable(k, v) for k, v in merged_vars_dict.items()]
         sql_variables = self.parse_execute_variables(parsed_variables)
         python_variables = self._parse_python_variables(parsed_variables)
         results = []