SNOW-3485471: cte dedup for self-join case

[sfc-gh-aling]

1. Which Jira issue is this PR addressing? Make sure that there is an accompanying issue to your PR.

In SCOS, spark.table() translates into SnowflakePlan nodes with non-empty expr_to_alias. The TPCDS_Q39 query aliases the same inventory DataFrame as inv1 and inv2 and self-joins them. The join path
deep-copies the plan for each branch and calls add_aliases separately, resulting in both branches carrying the same alias values under different UUID keys. The old key-sort hashing produced
different hashes for the two branches, so the CTE optimizer did not detect them as duplicates and inlined the full aggregation pipeline twice. The value-sort fix correctly identifies them as
the same computation and emits a single CTE.

2. Fill out the following pre-review checklist:

   • I am adding a new automated test(s) to verify correctness of my new code
     • If this test skips Local Testing mode, I'm requesting review from @snowflakedb/local-testing
   • I am adding new logging messages
   • I am adding a new telemetry message
   • I am adding new credentials
   • I am adding a new dependency
   • If this is a new feature/behavior, I'm adding the Local Testing parity changes.
   • I acknowledge that I have ensured my changes to be thread-safe. Follow the link for more information: Thread-safe Developer Guidelines
   • If adding any arguments to public Snowpark APIs or creating new public Snowpark APIs, I acknowledge that I have ensured my changes include AST support. Follow the link for more information: AST Support Guidelines

3. Please describe how your code solves the related issue.

• hash node by expr_to_alias value instead expr_to_alias items
• conflicting alias detection to avoid over optimization

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 95.39%. Comparing base (495acce) to head (c29f910).

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #4245   +/-   ##
=======================================
  Coverage   95.38%   95.39%           
=======================================
  Files         171      171           
  Lines       44243    44252    +9     
  Branches     7557     7559    +2     
=======================================
+ Hits        42203    42215   +12     
+ Misses       1252     1250    -2     
+ Partials      788      787    -1     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[sfc-gh-aalam]

what kind of cases will this produce same hash-keys. When it is wrong, what kind of risk are we dealing with

[sfc-gh-aling]

1. I assume you are asking the hash-keys for a node -- same hash-keys are produced by nodes with same sql, sql params, df aliases, and expr_to_alias values (before the change it's by expr_to_alias.items().

(hash-keys for expr_to_alias are generated from uuid, it can rarely happen that two uuid collide)

2. the only risk is parent column resolution as we relax the standard of repeated node identification. I introduced extra logic to detect conflicting expr_to_alias (same expr maps to two different alias) in nodes to prevent CTE in this case. thank you for bringing it up

[sfc-gh-joshi]

Is there any chance of duplicate value entries here that would silently get swallowed by the sorted(set(...)) call?

[sfc-gh-aling]

Yes, duplicate alias can happen but consolidating them is harmless in our case, it's not reflected in the newly-generated CTE optimized node, and the generated sql in the hash is enough for dup node detection in the self-join case.

the expr_to_alias is mostly for internal column name resolution in the downstream compilation stage.

theoretically I think this info shall be excluded from the hash computation of a node, but right now I keep it as a defensive layer to distinguish nodes.

[sfc-gh-aling]

for clarity, more CTE is better for this case, see the below comparison between the generated sql:

Before (1 CTE, filter appears 3 times):

  WITH CTE_values AS (
    SELECT $1 AS "A", $2 AS "B" FROM VALUES (1,2),(3,4)                                                                                                                             
  )                                                                                                                                                                                 
  SELECT * FROM (                                                                                                                                                                   
    (                                                                                                                                                                               
      SELECT "A_XXX", "B_XXX", "A" AS "A_YYY", "B" AS "B_YYY"                                                                                                                       
      FROM (                                                                                                                                                                        
        (                                                                                                                                                                           
          SELECT "A" AS "A_XXX", "B" AS "B_XXX"                                                                                                                                     
          FROM CTE_values WHERE ("A" = 1)       -- filter #1 (df1, inline)                                                                                                          
        ) AS SNOWPARK_LEFT                                                                                                                                                          
        INNER JOIN (                                                                                                                                                                
          SELECT "A", "B"                                                                                                                                                           
          FROM CTE_values WHERE ("A" = 1)       -- filter #2 (df2, inline)                                                                                                          
        ) AS SNOWPARK_RIGHT                                                                                                                                                         
      )                                                           
    ) AS SNOWPARK_LEFT
    INNER JOIN (      
      SELECT "A", "B"
      FROM CTE_values WHERE ("A" = 1)           -- filter #3 (df3, inline)
    ) AS SNOWPARK_RIGHT                                                   
  )                                                                                                                                                                                 

After (2 CTEs, filter appears 2 times):

  WITH CTE_values AS (                                                                                                                                                              
    SELECT $1 AS "A", $2 AS "B" FROM VALUES (1,2),(3,4)                                                                                                                             
  ),                                                                                                                                                                                
  CTE_filtered AS (                              -- df2 and df3 deduplicated here                                                                                                   
    SELECT "A", "B"                                                                                                                                                                 
    FROM CTE_values WHERE ("A" = 1)             -- filter #1 (in CTE body)                                                                                                          
  )                                                                                                                                                                                 
  SELECT * FROM (                                                                                                                                                                   
    (                                                             
      SELECT "A_XXX", "B_XXX", "A" AS "A_YYY", "B" AS "B_YYY"
      FROM (                                                 
        (   
          SELECT "A" AS "A_XXX", "B" AS "B_XXX"
          FROM CTE_values WHERE ("A" = 1)       -- filter #2 (df1, still inline)
        ) AS SNOWPARK_LEFT                                                      
        INNER JOIN (SELECT * FROM CTE_filtered) AS SNOWPARK_RIGHT                                                                                                                   
      )                                                                                                                                                                             
    ) AS SNOWPARK_LEFT                                                                                                                                                              
    INNER JOIN (SELECT * FROM CTE_filtered) AS SNOWPARK_RIGHT                                                                                                                       
  )   

[sfc-gh-joshi]

Is there any chance of duplicate value entries here that would silently get swallowed by the sorted(set(...)) call?