Merge pull request #147 from TidierOrg/join_selectfix

fix join select issue

Author: drizk1

NEWS.md

@@ -1,4 +1,11 @@
 # TidierDB.jl updates
+## v.8.8 - 2025-08-05
+- fixes issue when grouping on a join id 
+- fixes cte construction issue 
+- fixes select before join order execution issue
+- add `~` support for aggregate functions in mutate, similar to TidierData syntax
+- additional AWS/Athena Backend improvements
+
 ## v.8.7 - 2025-07-07
 - AWS Athena backend bug fixes 
 - add `temp` option to `@create_table`, default is `true`

Project.toml

@@ -1,7 +1,7 @@
 name = "TidierDB"
 uuid = "86993f9b-bbba-4084-97c5-ee15961ad48b"
 authors = ["Daniel Rizk <rizk.daniel.12@gmail.com> and contributors"]
-version = "0.8.7"
+version = "0.8.8"
 
 [deps]
 Arrow = "69666777-d1a9-59fb-9406-91d4454c9d45"

docs/examples/UserGuide/agg_window.jl

@@ -18,10 +18,10 @@ end
 # ## Aggregate Functions in `@mutate`
 # By default, `@mutate`/`@transmute` supports (however, you can easily expand this list)
 # - `maximum`, `minimum`, `mean`, `std`, `sum`, `cumsum`
-# To use aggregate sql functions that are built in to any database backend, but exist outside of the TidierDB parser list above, simply wrap the function call in `agg()`
+# To use aggregate sql functions that are built in to any database backend, but exist outside of the TidierDB parser list above, simply prefix the function with `~` or wrap the function call in `agg()`
 @chain mtcars begin 
      @group_by(cyl)
-     @mutate(kurt = agg(kurtosis(mpg)))
+     @mutate(kurt = ~kurtosis(mpg))
      @select cyl mpg kurt
      @head()
      @collect 

docs/examples/UserGuide/ex_joining.jl

@@ -49,7 +49,7 @@ end;
 query2 = @chain mtcars @filter(mpg>20) @mutate(mpg = mpg *4); 
 
 @chain query begin
-    @left_join(t(query2), cyl == cyl)
+    @left_join(query2, cyl == cyl)
     @summarize(avg_mean = mean(mpg), _by = efficiency)
     @mutate(mean = avg_mean / 4 )
     @collect
@@ -63,7 +63,7 @@ end
 # mt2 = dt(db, "ducks_db.mt2")
 # other_db = @chain dt(db, "ducks_db.mt2") @filter(!str_detect(car, "M"))
 # @chain mtcars begin
-#     @left_join(t(other_db), model == car)
+#     @left_join(other_db, model == car)
 #     @select(model, fuel_efficiency)
 #     @head(5)
 #     @collect
@@ -120,7 +120,7 @@ end
 end;
 
 @chain dt(db, "viewer") begin # access the view like any other table
-    @left_join(t(query2), cyl == cyl)
+    @left_join(query2, cyl == cyl)
     @summarize(avg_mean = mean(mpg), _by = efficiency)
     @mutate(mean = avg_mean / 4 )
     @collect
@@ -131,7 +131,7 @@ end
 prices = dt(db, "https://duckdb.org/data/prices.csv", "prices");
 holdings = dt(db, "https://duckdb.org/data/holdings.csv", "holdings");
 @chain holdings begin
-    @inner_join(t(prices), ticker = ticker, closest(when >= when))
+    @inner_join(prices, ticker = ticker, closest(when >= when))
     @select(holdings.ticker, holdings.when) 
     @mutate(value = price * shares)
     @collect

src/TidierDB.jl

@@ -41,7 +41,7 @@ using GZip
  struct databricks <: SQLBackend end
 
  const  _warning_ = Ref(false)
- const window_agg_fxns = [:lead, :lag, :dense_rank, :nth_value, :ntile, :rank_dense, :row_number, :first_value, :last_value, :cume_dist, :count]
+ const window_agg_fxns = [:lead, :lag, :dense_rank, :nth_value, :ntile, :rank_dense, :row_number, :first_value, :last_value, :cume_dist, :count, :first, :last]
  current_sql_mode = Ref{SQLBackend}(duckdb())
  const color = Ref{Bool}(true)
  function set_sql_mode(mode::SQLBackend) current_sql_mode[] = mode end

src/TidierDB_macros.jl

@@ -2,9 +2,7 @@
 $docstring_select
 """
 macro select(sqlquery, exprs...)
-
     exprs = parse_blocks(exprs...)
-
     return quote
         exprs_str = map(expr -> isa(expr, Symbol) ? string(expr) : expr, $exprs)
         sq = t($(esc(sqlquery)))
@@ -17,26 +15,22 @@ macro select(sqlquery, exprs...)
                 if occursin(".", col)
                     table_col_split = split(col, ".")
                     table_name, col_name = table_col_split[1], table_col_split[2]
-
-                    # Iterate and update current_selxn based on matches
                     for idx in eachindex(sq.metadata.current_selxn)
-                        if sq.metadata.table_name[idx] == table_name && 
-                           sq.metadata.name[idx] == col_name
+                        if sq.metadata.table_name[idx] == table_name && sq.metadata.name[idx] == col_name
                             sq.metadata.current_selxn[idx] = 2
                         end
                     end
                 else
-                    # Direct matching for columns without 'table.' prefix
                     matching_indices = findall(sq.metadata.name .== col)
                     sq.metadata.current_selxn[matching_indices] .= 1
                 end
             end
         end
-        
         sq
     end
 end
 
+
 """
 $docstring_filter
 """
@@ -181,25 +175,113 @@ end
 
 function groupby_exp(expr, sq)
     if isa(expr, Expr) && expr.head == :(=) && isa(expr.args[1], Symbol)
-        # Extract column alias name as string
         col_name = string(expr.args[1])
         if current_sql_mode[] == snowflake()
             col_name = uppercase(col_name)  # COV_EXCL_LINE
         end
         push!(sq.metadata, Dict("name" => col_name, "type" => "UNKNOWN", "current_selxn" => 1, "table_name" => "table"))
-        # Convert the right-hand side expression to a SQL expression
-        col_expr = expr_to_sql(expr.args[2], sq)
-        col_expr = string(col_expr)
-        # Return the alias and the SQL snippet (wrapped in parentheses to be safe)
+        col_expr = expr_to_sql(expr.args[2], sq) |> string
         return col_name, "(" * col_expr * ") AS " * col_name
     else
         error("Unsupported expression in @group_by: $(expr)")
     end
 end
 
+# Single helper to get the expression behind an alias in the current SELECT
+function _expr_for_alias(select_sql::AbstractString, alias::AbstractString)
+    s = strip(String(select_sql))
+    body = startswith(uppercase(s), "SELECT ") ? s[8:end] : s
+
+    m = findfirst(Regex("(?i)\\bAS\\s+$alias\\b"), body)
+    m === nothing && return nothing
+
+    as_start = first(m) - 1
+    depth = 0
+    i = as_start
+    start_idx = firstindex(body)
+
+    while i >= firstindex(body)
+        c = body[i]
+        if c == ')'
+            depth += 1
+        elseif c == '('
+            depth = max(depth - 1, 0)
+        elseif c == ',' && depth == 0
+            start_idx = nextind(body, i)
+            break
+        end
+        i = prevind(body, i)
+    end
+
+    expr = strip(body[start_idx:as_start])
+    return expr == "" ? nothing : expr
+end
+
 """
 $docstring_group_by
 """
+macro group_by(sqlquery, columns...)
+    columns = parse_blocks(columns...)
+    return quote
+        columns_str = map(col -> isa(col, Symbol) ? string(col) : col, $columns)
+        sq = t($(esc(sqlquery)))
+        if isa(sq, SQLQuery)
+            try
+                # Build GROUP BY items; if a name matches a SELECT alias, use its expression
+                group_items = String[]
+                for c in columns_str
+                    expr = _expr_for_alias(sq.select, c)
+                    if expr !== nothing
+                        push!(group_items, expr)          # e.g., COALESCE(t2.id, t1.id)
+                    else
+                        for nm in parse_tidy_db([c], sq.metadata)
+                            push!(group_items, nm)        # qualified name
+                        end
+                    end
+                end
+                sq.groupBy = "GROUP BY " * join(group_items, ", ")
+
+                # If no projection yet or it's SELECT *, project only the grouping columns
+                sel = strip(String(sq.select))
+                if isempty(sel) || occursin(r"(?i)^SELECT\s+\*$", sel)
+                    sq.select = "SELECT " * join(group_items, ", ")
+                end
+            catch
+                # Handle expression/alias form: group_by(alias = expr, ...)
+                sq.groupBy_exprs = true
+                local group_expressions = String[]  # "(expr) AS alias"
+                local group_aliases     = String[]  # "alias"
+
+                for col in $columns
+                    if isa(col, Expr) && col.head == :(=)
+                        let tup = groupby_exp(col, sq)
+                            push!(group_expressions, tup[2])
+                            push!(group_aliases,    tup[1])
+                        end
+                    else
+                        for nm in parse_tidy_db([col], sq.metadata)
+                            push!(group_expressions, nm)
+                            push!(group_aliases,     nm)
+                        end
+                    end
+                end
+
+                # FIX #1: GROUP BY aliases (not "(expr) AS alias")
+                sq.groupBy = "GROUP BY " * join(group_aliases, ", ")
+
+                # FIX #2: If no projection yet or it's SELECT *, expose the aliased select items
+                sel = strip(String(sq.select))
+                if isempty(sel) || occursin(r"(?i)^SELECT\s+\*$", sel)
+                    sq.select = "SELECT " * join(group_expressions, ", ")
+                end
+            end
+        else
+            error("Expected sqlquery to be an instance of SQLQuery")
+        end
+        sq
+    end
+end
+#=
 macro group_by(sqlquery, columns...)
     columns = parse_blocks(columns...)
     return quote
@@ -248,6 +330,9 @@ macro group_by(sqlquery, columns...)
         sq
     end
 end
+=#
+
+
 
 
 """
@@ -339,67 +424,77 @@ macro rename(sqlquery, renamings...)
     renamings = parse_blocks(renamings...)
 
     return quote
-        # Prepare the renaming rules from the macro arguments
+        # Build old->new map from "new = old" pairs
         renamings_dict = Dict{String, String}()
-        for renaming in $(esc(renamings))
-            if isa(renaming, Expr) && renaming.head == :(=) && isa(renaming.args[1], Symbol)
-                # Map original column names to new names for renaming
-                renamings_dict[string(renaming.args[2])] = string(renaming.args[1])
+        for r in $(esc(renamings))
+            if isa(r, Expr) && r.head == :(=) && isa(r.args[1], Symbol)
+                # value11 = value1  =>  "value1" => "value11"
+                renamings_dict[string(r.args[2])] = string(r.args[1])
             else
-                throw("Unsupported renaming format in @rename: $(renaming)")
+                throw("Unsupported renaming format in @rename: $(r)")
             end
         end
 
         sq = t($(esc(sqlquery)))
-        
         if isa(sq, SQLQuery)
-            # Generate a new CTE name
+            # New CTE wrapper
             new_cte_name = "cte_" * string(sq.cte_count + 1)
             sq.cte_count += 1
-     
-            # Determine the select clause for the new CTE
-            select_clause = if isempty(sq.select) || sq.select == "SELECT *"
-                # If select is *, list all columns with renaming applied
-                all_columns = sq.metadata[!, :name]
-                join([haskey(renamings_dict, col) ? col * " AS " * renamings_dict[col] : col for col in all_columns], ", ")
+
+            # Build projection with renaming
+            select_clause = if isempty(strip(sq.select)) || strip(uppercase(sq.select)) == "SELECT *" || strip(sq.select) == "*"
+                # Only include columns with current_selxn != 0
+                mask = sq.metadata.current_selxn .!= 0
+                cols = Vector{String}(sq.metadata.name[mask])
+                join([haskey(renamings_dict, c) ? string(c, " AS ", renamings_dict[c]) : c for c in cols], ", ")
             else
-                
-                select_parts = split(sq.select[8:end], ", ")
-                updated_parts = map(select_parts) do part
-                    # Identify the base column name for potential renaming
-                    col = strip(split(part, " AS ")[1])
-                    if haskey(renamings_dict, col)
-                        # Apply renaming to the base column name
-                        string(renamings_dict[col]) * " AS " * col
+                # Edit existing SELECT list (adjust aliases, don't add new columns)
+                s = String(sq.select)
+                body = startswith(uppercase(s), "SELECT ") ? s[8:end] : s
+                parts = split(body, ", ")
+                updated = map(parts) do part
+                    if occursin(r"(?i)\bAS\b", part)
+                        bits = split(part, r"(?i)\bAS\b")
+                        expr  = strip(bits[1])
+                        alias = strip(bits[end])
+                        new_alias = get(renamings_dict, alias, alias)
+                        string(expr, " AS ", new_alias)
                     else
-                        # No renaming needed; keep the original part
-                        part
+                        base = strip(split(part, ".")[end])
+                        if haskey(renamings_dict, base)
+                            string(part, " AS ", renamings_dict[base])
+                        else
+                            part
+                        end
                     end
                 end
-                sq.select = " " * join(updated_parts, ", ")
-
-            end
-            for (old_name, new_name) in renamings_dict
-                sq.metadata[!, :name] = replace.(sq.metadata[!, :name], old_name => new_name)
+                replace(join(updated, ", "), r"(?i)\bAS\s+AS\b" => " AS ")
             end
 
-            if isempty(sq.select) 
-                 sq.select == "SELECT *" 
+            # Update metadata names only for selected cols (current_selxn != 0)
+            for (old_name, new_name) in renamings_dict
+                for i in eachindex(sq.metadata.name)
+                    if sq.metadata.current_selxn[i] != 0 && sq.metadata[i, :name] == old_name
+                        sq.metadata[i, :name] = new_name
+                    end
+                end
             end
 
-            # Create the new CTE with the select clause
+            # Emit CTE and re-point FROM
             new_cte = CTE(name=new_cte_name, select=select_clause, from=sq.from)
             push!(sq.ctes, new_cte)
-
-            # Update the from clause of the SQLQuery to the new CTE
             sq.from = new_cte_name
+
+            # Clear sq.select so subsequent steps project from the new CTE
+            sq.select = ""
         else
             error("Expected sqlquery to be an instance of SQLQuery")
         end
         sq
     end
 end
 
+
 mutable struct DBQuery
     val::String
 end

src/db_parsing.jl

@@ -159,8 +159,19 @@ function parse_tidy_db(exprs, metadata::DataFrame)
 
     # If no columns are explicitly included, default to all columns (with current_selxn == 1) minus any exclusions
     if isempty(included_columns)
-        included_columns = metadata.name[metadata.current_selxn .== 1]
-        included_columns = setdiff(included_columns, excluded_columns)
+        idx = findall(metadata.current_selxn .>= 1)
+        included_columns = String[]
+        for i in idx
+            nm = metadata.name[i]
+            if nm in excluded_columns
+                continue
+            end
+            if metadata.current_selxn[i] == 2
+                push!(included_columns, string(metadata.table_name[i], ".", nm))
+            else
+                push!(included_columns, nm)
+            end
+        end
     else
         included_columns = setdiff(included_columns, excluded_columns)
     end