#include "tql.h"
#include "lua_engine.h"

extern "C" {
#include "lua.h"
#include "lualib.h"
#include "lauxlib.h"
}

#include <algorithm>
#include <cstdio>
#include <cstring>
#include <unordered_map>

namespace tql {

using namespace data_table;

namespace {

int find_orig_col(const std::vector<std::string>& headers, const std::string& name) {
    for (size_t i = 0; i < headers.size(); ++i) if (headers[i] == name) return (int)i;
    return -1;
}

int find_derived_idx(const std::vector<DerivedColumn>& d, const std::string& name) {
    for (size_t i = 0; i < d.size(); ++i) if (d[i].name == name) return (int)i;
    return -1;
}

Op parse_op(const std::string& s) {
    if (s == "=")          return Op::Eq;
    if (s == "!=")         return Op::Neq;
    if (s == ">")          return Op::Gt;
    if (s == ">=")         return Op::Gte;
    if (s == "<")          return Op::Lt;
    if (s == "<=")         return Op::Lte;
    if (s == "contains")   return Op::Contains;
    if (s == "!contains")  return Op::NotContains;
    if (s == "starts")     return Op::StartsWith;
    if (s == "ends")       return Op::EndsWith;
    return Op::Eq;
}

std::string lua_to_string(lua_State* L, int idx) {
    if (lua_isnil(L, idx)) return "";
    if (lua_isboolean(L, idx)) return lua_toboolean(L, idx) ? "true" : "false";
    size_t n = 0;
    const char* s = luaL_tolstring(L, idx, &n);
    std::string out(s, n);
    lua_pop(L, 1);
    return out;
}

} // anon

std::string lua_string_literal(const std::string& s) {
    std::string out;
    out.reserve(s.size() + 4);
    out += '"';
    for (char c : s) {
        switch (c) {
            case '\\': out += "\\\\"; break;
            case '"':  out += "\\\""; break;
            case '\n': out += "\\n";  break;
            case '\r': out += "\\r";  break;
            case '\t': out += "\\t";  break;
            default:
                if ((unsigned char)c < 0x20) {
                    char b[8]; std::snprintf(b, sizeof(b), "\\%d", (unsigned char)c);
                    out += b;
                } else out += c;
        }
    }
    out += '"';
    return out;
}

std::string color_to_hex(unsigned int c) {
    unsigned int r = c & 0xFF;
    unsigned int g = (c >> 8) & 0xFF;
    unsigned int b = (c >> 16) & 0xFF;
    unsigned int a = (c >> 24) & 0xFF;
    char buf[16];
    if (a == 0xFF) std::snprintf(buf, sizeof(buf), "#%02x%02x%02x", r, g, b);
    else           std::snprintf(buf, sizeof(buf), "#%02x%02x%02x%02x", r, g, b, a);
    return buf;
}

unsigned int hex_to_color(const std::string& s) {
    if (s.size() < 7 || s[0] != '#') return 0xFFFFFFFF;
    auto hex2 = [&](size_t i) -> unsigned int {
        unsigned int v = 0;
        if (i + 1 < s.size()) std::sscanf(s.c_str() + i, "%2x", &v);
        return v;
    };
    unsigned int r = hex2(1), g = hex2(3), b = hex2(5);
    unsigned int a = (s.size() >= 9) ? hex2(7) : 0xFF;
    return r | (g << 8) | (b << 16) | (a << 24);
}

ColumnType column_type_from_string(const std::string& s) {
    if (s == "string") return ColumnType::String;
    if (s == "int")    return ColumnType::Int;
    if (s == "float")  return ColumnType::Float;
    if (s == "bool")   return ColumnType::Bool;
    if (s == "date")   return ColumnType::Date;
    if (s == "json")   return ColumnType::Json;
    return ColumnType::Auto;
}

// Helper: header del Stage 0 dado un col idx eff. Para stages 1+ no aplica
// (los stage outputs tienen sus propios headers).
namespace {
const char* agg_fn_token(AggFn f) {
    switch (f) {
        case AggFn::Count:      return "count";
        case AggFn::Sum:        return "sum";
        case AggFn::Avg:        return "avg";
        case AggFn::Min:        return "min";
        case AggFn::Max:        return "max";
        case AggFn::Distinct:   return "distinct";
        case AggFn::Stddev:     return "stddev";
        case AggFn::Median:     return "median";
        case AggFn::P25:        return "p25";
        case AggFn::P75:        return "p75";
        case AggFn::P90:        return "p90";
        case AggFn::P99:        return "p99";
        case AggFn::Percentile: return "percentile";
    }
    return "?";
}

AggFn agg_fn_from_string(const std::string& s) {
    if (s == "count")      return AggFn::Count;
    if (s == "sum")        return AggFn::Sum;
    if (s == "avg")        return AggFn::Avg;
    if (s == "min")        return AggFn::Min;
    if (s == "max")        return AggFn::Max;
    if (s == "distinct")   return AggFn::Distinct;
    if (s == "stddev")     return AggFn::Stddev;
    if (s == "median")     return AggFn::Median;
    if (s == "p25")        return AggFn::P25;
    if (s == "p75")        return AggFn::P75;
    if (s == "p90")        return AggFn::P90;
    if (s == "p99")        return AggFn::P99;
    if (s == "percentile") return AggFn::Percentile;
    return AggFn::Count;
}
} // anon

std::string emit(const State& state,
                 const std::vector<std::string>& headers,
                 const std::vector<ColumnType>& types)
{
    int orig_cols = (int)headers.size();
    const Stage& raw = state.raw();
    int eff_cols  = orig_cols + (int)raw.derived.size();

    // Build effective headers + types (same indexing as col_visible/order)
    std::vector<std::string> eff_headers(eff_cols);
    std::vector<ColumnType>  eff_types(eff_cols);
    for (int c = 0; c < orig_cols; ++c) {
        eff_headers[c] = headers[c];
        eff_types[c]   = (c < (int)types.size()) ? types[c] : ColumnType::Auto;
    }
    for (int k = 0; k < (int)raw.derived.size(); ++k) {
        eff_headers[orig_cols + k] = raw.derived[k].name;
        eff_types[orig_cols + k]   = raw.derived[k].type;
    }

    // Build order positions: col_idx -> visual order (1-based)
    std::unordered_map<int, int> order_pos;
    for (size_t i = 0; i < state.col_order.size(); ++i) {
        order_pos[state.col_order[i]] = (int)i + 1;
    }

    auto emit_filter_block = [&](const std::vector<Filter>& filters,
                                  const std::vector<std::string>& stage_headers,
                                  const char* indent) -> std::string {
        if (filters.empty()) return {};
        std::string s;
        s += indent; s += "filter = {\n";
        for (const auto& f : filters) {
            std::string col_name = (f.col >= 0 && f.col < (int)stage_headers.size())
                                   ? stage_headers[f.col] : "";
            s += indent; s += "  {";
            s += lua_string_literal(op_label(f.op));
            s += ", ";
            s += lua_string_literal(col_name);
            s += ", ";
            s += lua_string_literal(f.value);
            s += "},\n";
        }
        s += indent; s += "},\n";
        return s;
    };

    auto emit_sort_block = [&](const std::vector<SortClause>& sorts,
                               const char* indent) -> std::string {
        if (sorts.empty()) return {};
        std::string s;
        s += indent; s += "sort = {\n";
        for (const auto& sc : sorts) {
            s += indent; s += "  {";
            s += lua_string_literal(sc.desc ? "desc" : "asc");
            s += ", ";
            s += lua_string_literal(sc.col);
            s += "},\n";
        }
        s += indent; s += "},\n";
        return s;
    };

    std::string out;
    out += "-- TQL v1 (Table Query Language). Round-trip de State <-> Lua.\n";
    out += "-- Schema:\n";
    out += "--   version = 1                       -- bump si breaking change\n";
    out += "--   display = \"table\"                 -- table|bar|line|pie (futuro)\n";
    out += "--   stages = { stage0, stage1, ... }  -- pipeline; stage 0 = Raw\n";
    out += "--   columns = { {name,type,visible,order,color_rules}, ... }\n";
    out += "--\n";
    out += "-- Stage 0 (Raw):     filter + expressions + sort\n";
    out += "-- Stage N (Grouped): filter + breakout + aggregation + sort\n";
    out += "--\n";
    out += "-- filter: {{op, col, val}, ...}        op in =,!=,>,>=,<,<=,contains,!contains,starts,ends\n";
    out += "-- expressions: {[name] = \"lua_body\"}  ej: [\"total\"] = \"return [a] + [b]\"\n";
    out += "-- breakout: {\"col1\", \"col2\"}          group by\n";
    out += "-- aggregation: {{fn, col, arg?}, ...}  fn in count,sum,avg,min,max,distinct,stddev,median,p25,p75,p90,p99,percentile\n";
    out += "-- sort: {{dir, col}, ...}              dir in asc,desc\n";
    out += "return {\n";
    out += "  version = 1,\n";
    out += "  display = ";
    out += lua_string_literal(view_mode_token(state.display));
    out += ",\n";
    if (!state.main_source.empty()) {
        out += "  main_source = ";
        out += lua_string_literal(state.main_source);
        out += ",\n";
    }

    // joins (antes de stages, materializa input)
    if (!state.joins.empty()) {
        out += "  joins = {\n";
        for (const auto& jn : state.joins) {
            out += "    {alias = "    + lua_string_literal(jn.alias);
            out += ", source = "  + lua_string_literal(jn.source);
            out += ", strategy = " + lua_string_literal(join_strategy_token(jn.strategy));
            out += ", on = {";
            for (size_t i = 0; i < jn.on.size(); ++i) {
                if (i) out += ", ";
                out += "{" + lua_string_literal(jn.on[i].first) + ", "
                           + lua_string_literal(jn.on[i].second) + "}";
            }
            out += "}";
            if (!jn.fields.empty()) {
                out += ", fields = {";
                for (size_t i = 0; i < jn.fields.size(); ++i) {
                    if (i) out += ", ";
                    out += lua_string_literal(jn.fields[i]);
                }
                out += "}";
            }
            out += "},\n";
        }
        out += "  },\n";
    }

    out += "  stages = {\n";

    // Recorre todos los stages; stage 0 tiene formato Raw (filter+expr+sort),
    // stages 1+ tienen formato Grouped (filter+breakout+aggregation+sort).
    // Headers para resolver col indices de filters/sorts se computan stage por
    // stage simulando la cadena.
    std::vector<std::string> cur_headers = headers;       // stage input headers
    // Para stage 0 raw, los headers incluyen orig + derived.
    // Construye cur_headers iniciales (= orig); derived se anaden al pasar stage 0.

    for (int si = 0; si < (int)state.stages.size(); ++si) {
        const Stage& stg = state.stages[si];
        out += "    {\n";

        if (si == 0) {
            // Stage 0: orig headers + derived seran disponibles tras expressions.
            // Para los filter col indices, asumimos van con cur_headers = orig.
            // (data_table.cpp solo aplica filters a orig cols al guardar; si en
            //  futuro stage 0 admite filter sobre derived, se traduce a name.)
            std::vector<std::string> s0_headers = headers;
            // Filters
            out += emit_filter_block(stg.filters, s0_headers, "      ");

            // Expressions
            if (!stg.derived.empty()) {
                bool any = false;
                for (const auto& d : stg.derived) if (!d.formula.empty()) { any = true; break; }
                if (any) {
                    out += "      expressions = {\n";
                    for (const auto& d : stg.derived) {
                        if (d.formula.empty()) continue;
                        out += "        [";
                        out += lua_string_literal(d.name);
                        out += "] = ";
                        out += lua_string_literal(d.formula);
                        out += ",\n";
                    }
                    out += "      },\n";
                }
            }

            // Sort (sort.col es string en nuevo modelo).
            out += emit_sort_block(stg.sorts, "      ");

            // Avanza cur_headers para siguiente stage: orig + derived.
            for (const auto& d : stg.derived) cur_headers.push_back(d.name);
        } else {
            // Stage 1+: filter (sobre output del previo, cur_headers).
            out += emit_filter_block(stg.filters, cur_headers, "      ");

            // breakout
            if (!stg.breakouts.empty()) {
                out += "      breakout = {";
                for (size_t i = 0; i < stg.breakouts.size(); ++i) {
                    if (i > 0) out += ", ";
                    out += lua_string_literal(stg.breakouts[i]);
                }
                out += "},\n";
            }

            // aggregation
            if (!stg.aggregations.empty()) {
                out += "      aggregation = {\n";
                for (const auto& a : stg.aggregations) {
                    out += "        {";
                    out += lua_string_literal(agg_fn_token(a.fn));
                    if (a.fn != AggFn::Count) {
                        out += ", ";
                        out += lua_string_literal(a.col);
                    }
                    if (a.fn == AggFn::Percentile) {
                        char buf[32]; std::snprintf(buf, sizeof(buf), "%g", a.arg);
                        out += ", "; out += buf;
                    }
                    out += "},\n";
                }
                out += "      },\n";
            }

            // sort
            out += emit_sort_block(stg.sorts, "      ");

            // Avanza cur_headers para siguiente stage: breakouts + agg aliases.
            std::vector<std::string> next;
            for (const auto& b : stg.breakouts)    next.push_back(b);
            for (const auto& a : stg.aggregations) next.push_back(aggregation_alias(a));
            cur_headers = std::move(next);
        }

        out += "    },\n";
    }
    out += "  },\n";

    // columns (per-col render config) — siempre referidas a los effective cols
    // del STAGE 0 (asumimos viz state para stage 0 / raw). Renderizar columns
    // por cada stage no aporta v1.
    out += "  columns = {\n";
    for (int c = 0; c < eff_cols; ++c) {
        out += "    {";
        out += "name = " + lua_string_literal(eff_headers[c]);
        out += ", type = " + lua_string_literal(column_type_name(eff_types[c]));
        bool vis = (c < (int)state.col_visible.size()) ? state.col_visible[c] : true;
        out += std::string(", visible = ") + (vis ? "true" : "false");
        int order = order_pos.count(c) ? order_pos[c] : c + 1;
        out += ", order = " + std::to_string(order);
        // color rules for this col
        bool first = true;
        for (const auto& cr : state.color_rules) {
            if (cr.col != c) continue;
            if (first) { out += ", color_rules = {"; first = false; }
            else       { out += ", "; }
            out += "{equals = " + lua_string_literal(cr.equals);
            out += ", color = " + lua_string_literal(color_to_hex(cr.color)) + "}";
        }
        if (!first) out += "}";
        out += "},\n";
    }
    out += "  },\n";

    // views (extra viz panels — viz adicional sobre mismos stages)
    auto emit_view = [&](const VizPanel& p) -> std::string {
        std::string s = "    {";
        s += "display = " + lua_string_literal(view_mode_token(p.display));
        if (!p.config.x_col.empty())    s += ", x_col = "   + lua_string_literal(p.config.x_col);
        if (!p.config.cat_col.empty())  s += ", cat_col = " + lua_string_literal(p.config.cat_col);
        if (!p.config.size_col.empty()) s += ", size_col = "+ lua_string_literal(p.config.size_col);
        if (!p.config.y_cols.empty()) {
            s += ", y_cols = {";
            for (size_t i = 0; i < p.config.y_cols.size(); ++i) {
                if (i) s += ", ";
                s += lua_string_literal(p.config.y_cols[i]);
            }
            s += "}";
        }
        if (p.config.primary_color != 0)
            s += ", color = " + lua_string_literal(color_to_hex(p.config.primary_color));
        if (p.config.hist_bins > 0)
            s += ", hist_bins = " + std::to_string(p.config.hist_bins);
        if (p.config.pie_radius > 0)
            s += ", pie_radius = " + std::to_string(p.config.pie_radius);
        if (!p.config.show_legend)  s += ", show_legend = false";
        if (p.config.show_markers)  s += ", show_markers = true";
        if (p.config.locked)        s += ", locked = true";
        s += "},\n";
        return s;
    };

    out += "  views = {\n";
    // Panel 0 = main viz
    VizPanel main_p;
    main_p.display = state.display;
    main_p.config  = state.viz_config;
    out += emit_view(main_p);
    for (const auto& p : state.extra_panels) out += emit_view(p);
    out += "  },\n";

    out += "  visualization_settings = {},\n";
    out += "}\n";
    return out;
}

bool apply(const std::string& lua_text, State& state,
           const std::vector<std::string>& headers,
           const std::vector<ColumnType>& /*types*/,
           const char* const* cells, int rows, int orig_cols,
           std::string* err)
{
    std::vector<std::string> warns;
    auto warn = [&](const std::string& m) { warns.push_back(m); };
    auto finish_with_warns = [&](bool ok) -> bool {
        if (err && !warns.empty()) {
            std::string j;
            for (size_t i = 0; i < warns.size(); ++i) {
                if (i) j += "; ";
                j += warns[i];
            }
            *err = j;
        }
        return ok;
    };

    lua_State* L = lua_engine::raw_state();
    if (!L) { if (err) *err = "lua engine null"; return false; }

    if (luaL_loadbufferx(L, lua_text.data(), lua_text.size(), "tql", "t") != LUA_OK) {
        if (err) *err = lua_tostring(L, -1) ? lua_tostring(L, -1) : "load error";
        lua_pop(L, 1);
        return false;
    }
    if (lua_pcall(L, 0, 1, 0) != LUA_OK) {
        if (err) *err = lua_tostring(L, -1) ? lua_tostring(L, -1) : "exec error";
        lua_pop(L, 1);
        return false;
    }
    if (!lua_istable(L, -1)) {
        if (err) *err = "TQL root must be a table";
        lua_pop(L, 1);
        return false;
    }

    // main_source
    lua_getfield(L, -1, "main_source");
    if (lua_isstring(L, -1)) state.main_source = lua_tostring(L, -1);
    else                     state.main_source.clear();
    lua_pop(L, 1);

    // display
    lua_getfield(L, -1, "display");
    if (lua_isstring(L, -1)) {
        std::string d = lua_tostring(L, -1);
        ViewMode m = view_mode_from_token(d.c_str());
        state.display = m;
        if (d != "table" && std::strcmp(view_mode_token(m), d.c_str()) != 0) {
            warn("unknown display \"" + d + "\" (defaulting to table)");
        }
    }
    lua_pop(L, 1);

    // Validar version.
    lua_getfield(L, -1, "version");
    if (lua_isnil(L, -1)) {
        warn("version missing (assuming 1)");
    } else if (!lua_isnumber(L, -1)) {
        if (err) *err = "version must be a number";
        lua_pop(L, 2);
        return false;
    } else {
        int v = (int)lua_tointeger(L, -1);
        if (v != 1) {
            char buf[64]; std::snprintf(buf, sizeof(buf), "unsupported TQL version %d (expected 1)", v);
            if (err) *err = buf;
            lua_pop(L, 2);
            return false;
        }
    }
    lua_pop(L, 1);

    // Reset partes mutables. Liberar lua_ids antes.
    for (auto& s : state.stages) {
        for (auto& d : s.derived) {
            if (d.lua_id >= 0) lua_engine::release(lua_engine::get(), d.lua_id);
        }
    }
    state.stages.clear();
    state.active_stage = 0;
    state.color_rules.clear();

    // ---- Walk joins[] ----
    state.joins.clear();
    lua_getfield(L, -1, "joins");
    if (lua_istable(L, -1)) {
        int nj = (int)lua_rawlen(L, -1);
        for (int i = 1; i <= nj; ++i) {
            lua_rawgeti(L, -1, i);
            if (!lua_istable(L, -1)) { lua_pop(L, 1); continue; }
            Join jn;
            lua_getfield(L, -1, "alias");
            if (lua_isstring(L, -1)) jn.alias = lua_tostring(L, -1);
            lua_pop(L, 1);
            lua_getfield(L, -1, "source");
            if (lua_isstring(L, -1)) jn.source = lua_tostring(L, -1);
            lua_pop(L, 1);
            lua_getfield(L, -1, "strategy");
            if (lua_isstring(L, -1)) jn.strategy = join_strategy_from_token(lua_tostring(L, -1));
            lua_pop(L, 1);
            lua_getfield(L, -1, "on");
            if (lua_istable(L, -1)) {
                int on_n = (int)lua_rawlen(L, -1);
                for (int k = 1; k <= on_n; ++k) {
                    lua_rawgeti(L, -1, k);
                    if (lua_istable(L, -1) && lua_rawlen(L, -1) >= 2) {
                        lua_rawgeti(L, -1, 1); std::string a = lua_to_string(L, -1); lua_pop(L, 1);
                        lua_rawgeti(L, -1, 2); std::string b = lua_to_string(L, -1); lua_pop(L, 1);
                        jn.on.push_back({a, b});
                    }
                    lua_pop(L, 1);
                }
            }
            lua_pop(L, 1);
            lua_getfield(L, -1, "fields");
            if (lua_istable(L, -1)) {
                int fn_n = (int)lua_rawlen(L, -1);
                for (int k = 1; k <= fn_n; ++k) {
                    lua_rawgeti(L, -1, k);
                    if (lua_isstring(L, -1)) jn.fields.emplace_back(lua_tostring(L, -1));
                    lua_pop(L, 1);
                }
            }
            lua_pop(L, 1);
            state.joins.push_back(jn);
            lua_pop(L, 1);
        }
    }
    lua_pop(L, 1);

    // ---- Walk stages[] ----
    lua_getfield(L, -1, "stages");
    if (lua_istable(L, -1)) {
        int n_stages = (int)lua_rawlen(L, -1);
        // Headers efectivos por stage para resolver filter/sort col indices.
        std::vector<std::string> cur_headers = headers;

        for (int si = 1; si <= n_stages; ++si) {
            lua_rawgeti(L, -1, si);
            if (!lua_istable(L, -1)) { lua_pop(L, 1); continue; }

            Stage stg;

            // Stage 0 expressions (solo aplica a si == 1, pero permitimos en
            // cualquier stage por simetria — el UI no las expone en stages 1+).
            lua_getfield(L, -1, "expressions");
            if (lua_istable(L, -1)) {
                lua_pushnil(L);
                while (lua_next(L, -2) != 0) {
                    if (lua_isstring(L, -2) && lua_isstring(L, -1)) {
                        std::string name    = lua_tostring(L, -2);
                        std::string formula = lua_tostring(L, -1);
                        std::string cerr;
                        int id = lua_engine::compile(lua_engine::get(), formula, &cerr);
                        DerivedColumn d;
                        d.source_col = -1;
                        d.name       = name;
                        d.formula    = formula;
                        d.lua_id     = id;
                        d.compile_error = (id < 0) ? cerr : "";
                        if (id >= 0 && si == 1) {
                            // auto-detect tipo via sample (solo para stage 0).
                            int sample = std::min(64, rows);
                            std::vector<std::string> samples_str;
                            std::vector<const char*> samples_ptr;
                            std::vector<std::string> hn_storage = headers;
                            std::unordered_map<std::string, int> n2c;
                            for (int c = 0; c < orig_cols && c < (int)hn_storage.size(); ++c) {
                                n2c[hn_storage[c]] = c;
                            }
                            for (int r = 0; r < sample; ++r) {
                                lua_engine::RowCtx ctx;
                                ctx.cells        = cells;
                                ctx.orig_cols    = orig_cols;
                                ctx.row          = r;
                                ctx.header_names = &hn_storage;
                                ctx.name_to_col  = &n2c;
                                std::string e;
                                samples_str.emplace_back(
                                    lua_engine::eval(lua_engine::get(), id, ctx, &e));
                            }
                            for (auto& s : samples_str) samples_ptr.push_back(s.c_str());
                            d.type = auto_detect_type(samples_ptr.data(),
                                                       (int)samples_ptr.size(), 1, 0);
                        } else {
                            d.type = ColumnType::String;
                        }
                        stg.derived.push_back(d);
                    }
                    lua_pop(L, 1);
                }
            }
            lua_pop(L, 1);

            // filter
            lua_getfield(L, -1, "filter");
            if (lua_istable(L, -1)) {
                int n = (int)lua_rawlen(L, -1);
                for (int i = 1; i <= n; ++i) {
                    lua_rawgeti(L, -1, i);
                    if (lua_istable(L, -1) && lua_rawlen(L, -1) >= 3) {
                        lua_rawgeti(L, -1, 1); std::string op = lua_to_string(L, -1); lua_pop(L, 1);
                        lua_rawgeti(L, -1, 2); std::string col_name = lua_to_string(L, -1); lua_pop(L, 1);
                        lua_rawgeti(L, -1, 3); std::string val = lua_to_string(L, -1); lua_pop(L, 1);
                        int ci = find_orig_col(cur_headers, col_name);
                        if (ci >= 0) {
                            stg.filters.push_back({ci, parse_op(op), val});
                        } else {
                            warn("stage " + std::to_string(si - 1) + ": filter col \"" + col_name + "\" not found");
                        }
                        if (op != "=" && op != "!=" && op != ">" && op != ">=" &&
                            op != "<" && op != "<=" && op != "contains" &&
                            op != "!contains" && op != "starts" && op != "ends") {
                            warn("stage " + std::to_string(si - 1) + ": unknown filter op \"" + op + "\" (defaulting to =)");
                        }
                    }
                    lua_pop(L, 1);
                }
            }
            lua_pop(L, 1);

            // breakout (solo aplica stages >= 1, no-op silencioso si stage 0).
            // Acepta sufijo ":granularity" para cols Date (fase 10).
            lua_getfield(L, -1, "breakout");
            if (lua_istable(L, -1)) {
                int n = (int)lua_rawlen(L, -1);
                for (int i = 1; i <= n; ++i) {
                    lua_rawgeti(L, -1, i);
                    if (lua_isstring(L, -1)) {
                        std::string bn = lua_tostring(L, -1);
                        std::string clean;
                        parse_breakout_granularity(bn, clean);
                        if (find_orig_col(cur_headers, clean) < 0) {
                            warn("stage " + std::to_string(si - 1) + ": breakout col \"" + clean + "\" not in input headers");
                        }
                        stg.breakouts.emplace_back(bn);
                    }
                    lua_pop(L, 1);
                }
            }
            lua_pop(L, 1);

            // aggregation
            lua_getfield(L, -1, "aggregation");
            if (lua_istable(L, -1)) {
                int n = (int)lua_rawlen(L, -1);
                for (int i = 1; i <= n; ++i) {
                    lua_rawgeti(L, -1, i);
                    if (lua_istable(L, -1) && lua_rawlen(L, -1) >= 1) {
                        Aggregation a;
                        lua_rawgeti(L, -1, 1);
                        std::string fn_name = lua_to_string(L, -1);
                        lua_pop(L, 1);
                        bool known = (fn_name == "count" || fn_name == "sum" || fn_name == "avg" ||
                                      fn_name == "min"   || fn_name == "max" || fn_name == "distinct" ||
                                      fn_name == "stddev"|| fn_name == "median" ||
                                      fn_name == "p25" || fn_name == "p75" || fn_name == "p90" ||
                                      fn_name == "p99" || fn_name == "percentile");
                        if (!known) {
                            warn("stage " + std::to_string(si - 1) + ": unknown aggregation fn \"" + fn_name + "\" (defaulting to count)");
                        }
                        a.fn = agg_fn_from_string(fn_name);
                        if (lua_rawlen(L, -1) >= 2) {
                            lua_rawgeti(L, -1, 2);
                            a.col = lua_to_string(L, -1);
                            lua_pop(L, 1);
                            if (a.fn != AggFn::Count && find_orig_col(cur_headers, a.col) < 0) {
                                warn("stage " + std::to_string(si - 1) + ": aggregation col \"" + a.col + "\" not in input headers");
                            }
                        } else if (a.fn != AggFn::Count) {
                            warn("stage " + std::to_string(si - 1) + ": aggregation \"" + fn_name + "\" requires a column");
                        }
                        if (lua_rawlen(L, -1) >= 3) {
                            lua_rawgeti(L, -1, 3);
                            if (lua_isnumber(L, -1)) a.arg = lua_tonumber(L, -1);
                            lua_pop(L, 1);
                        }
                        stg.aggregations.push_back(a);
                    }
                    lua_pop(L, 1);
                }
            }
            lua_pop(L, 1);

            // sort
            lua_getfield(L, -1, "sort");
            if (lua_istable(L, -1)) {
                int n = (int)lua_rawlen(L, -1);
                for (int i = 1; i <= n; ++i) {
                    lua_rawgeti(L, -1, i);
                    if (lua_istable(L, -1) && lua_rawlen(L, -1) >= 2) {
                        lua_rawgeti(L, -1, 1); std::string dir = lua_to_string(L, -1); lua_pop(L, 1);
                        lua_rawgeti(L, -1, 2); std::string col = lua_to_string(L, -1); lua_pop(L, 1);
                        SortClause sc;
                        sc.col  = col;
                        sc.desc = (dir == "desc");
                        if (dir != "asc" && dir != "desc") {
                            warn("stage " + std::to_string(si - 1) + ": unknown sort dir \"" + dir + "\" (defaulting to asc)");
                        }
                        stg.sorts.push_back(sc);
                    }
                    lua_pop(L, 1);
                }
            }
            lua_pop(L, 1);

            state.stages.push_back(std::move(stg));

            // Advance cur_headers para resolver filter/sort col del siguiente stage.
            const Stage& last = state.stages.back();
            if (si == 1) {
                // Stage 0: cur_headers = orig + derived (sin breakouts/agg).
                for (const auto& d : last.derived) cur_headers.push_back(d.name);
            } else {
                if (!last.breakouts.empty() || !last.aggregations.empty()) {
                    std::vector<std::string> next;
                    for (const auto& b : last.breakouts) next.push_back(b);
                    for (const auto& a : last.aggregations) next.push_back(aggregation_alias(a));
                    cur_headers = std::move(next);
                }
            }

            lua_pop(L, 1); // pop stage entry
        }
    }
    lua_pop(L, 1); // stages

    state.ensure_stage0();

    // ---- Walk columns (per-col render config) ----
    int eff_cols = orig_cols + (int)state.raw().derived.size();
    lua_getfield(L, -1, "columns");
    if (lua_istable(L, -1)) {
        state.col_visible.assign(eff_cols, true);
        std::vector<std::pair<int,int>> order_pairs;
        std::vector<bool> seen(eff_cols, false);

        int n = (int)lua_rawlen(L, -1);
        for (int i = 1; i <= n; ++i) {
            lua_rawgeti(L, -1, i);
            if (!lua_istable(L, -1)) { lua_pop(L, 1); continue; }

            lua_getfield(L, -1, "name");
            std::string nm = lua_to_string(L, -1);
            lua_pop(L, 1);

            int col_idx = find_orig_col(headers, nm);
            if (col_idx < 0) {
                int di = find_derived_idx(state.raw().derived, nm);
                if (di >= 0) col_idx = orig_cols + di;
            }
            if (col_idx < 0 || col_idx >= eff_cols) { lua_pop(L, 1); continue; }
            seen[col_idx] = true;

            // visible
            lua_getfield(L, -1, "visible");
            if (lua_isboolean(L, -1)) state.col_visible[col_idx] = lua_toboolean(L, -1);
            lua_pop(L, 1);

            // order
            lua_getfield(L, -1, "order");
            int order_val = lua_isnumber(L, -1) ? (int)lua_tointeger(L, -1) : (col_idx + 1);
            lua_pop(L, 1);
            order_pairs.emplace_back(order_val, col_idx);

            // type (mutable solo para derived)
            lua_getfield(L, -1, "type");
            if (lua_isstring(L, -1)) {
                std::string tn = lua_tostring(L, -1);
                ColumnType t = column_type_from_string(tn);
                if (col_idx >= orig_cols) {
                    state.raw().derived[col_idx - orig_cols].type = t;
                }
            }
            lua_pop(L, 1);

            // color_rules
            lua_getfield(L, -1, "color_rules");
            if (lua_istable(L, -1)) {
                int rn = (int)lua_rawlen(L, -1);
                for (int j = 1; j <= rn; ++j) {
                    lua_rawgeti(L, -1, j);
                    if (lua_istable(L, -1)) {
                        lua_getfield(L, -1, "equals");
                        std::string eq = lua_to_string(L, -1);
                        lua_pop(L, 1);
                        lua_getfield(L, -1, "color");
                        std::string hx = lua_to_string(L, -1);
                        lua_pop(L, 1);
                        state.color_rules.push_back({col_idx, eq, hex_to_color(hx)});
                    }
                    lua_pop(L, 1);
                }
            }
            lua_pop(L, 1);

            lua_pop(L, 1); // pop entry
        }

        std::sort(order_pairs.begin(), order_pairs.end());
        state.col_order.clear();
        for (auto& p : order_pairs) state.col_order.push_back(p.second);
        for (int c = 0; c < eff_cols; ++c) if (!seen[c]) state.col_order.push_back(c);
    }
    lua_pop(L, 1); // columns

    // ---- Walk views[] (extra viz panels) ----
    state.extra_panels.clear();
    lua_getfield(L, -1, "views");
    if (lua_istable(L, -1)) {
        int n = (int)lua_rawlen(L, -1);
        for (int i = 1; i <= n; ++i) {
            lua_rawgeti(L, -1, i);
            if (!lua_istable(L, -1)) { lua_pop(L, 1); continue; }
            VizPanel p;
            lua_getfield(L, -1, "display");
            if (lua_isstring(L, -1)) p.display = view_mode_from_token(lua_tostring(L, -1));
            lua_pop(L, 1);

            auto read_str = [&](const char* key, std::string& out_s) {
                lua_getfield(L, -1, key);
                if (lua_isstring(L, -1)) out_s = lua_tostring(L, -1);
                lua_pop(L, 1);
            };
            read_str("x_col",    p.config.x_col);
            read_str("cat_col",  p.config.cat_col);
            read_str("size_col", p.config.size_col);

            lua_getfield(L, -1, "y_cols");
            if (lua_istable(L, -1)) {
                int yn = (int)lua_rawlen(L, -1);
                for (int j = 1; j <= yn; ++j) {
                    lua_rawgeti(L, -1, j);
                    if (lua_isstring(L, -1)) p.config.y_cols.emplace_back(lua_tostring(L, -1));
                    lua_pop(L, 1);
                }
            }
            lua_pop(L, 1);

            lua_getfield(L, -1, "color");
            if (lua_isstring(L, -1)) p.config.primary_color = hex_to_color(lua_tostring(L, -1));
            lua_pop(L, 1);

            lua_getfield(L, -1, "hist_bins");
            if (lua_isnumber(L, -1)) p.config.hist_bins = (int)lua_tointeger(L, -1);
            lua_pop(L, 1);

            lua_getfield(L, -1, "pie_radius");
            if (lua_isnumber(L, -1)) p.config.pie_radius = (float)lua_tonumber(L, -1);
            lua_pop(L, 1);

            lua_getfield(L, -1, "show_legend");
            if (lua_isboolean(L, -1)) p.config.show_legend = lua_toboolean(L, -1);
            lua_pop(L, 1);

            lua_getfield(L, -1, "show_markers");
            if (lua_isboolean(L, -1)) p.config.show_markers = lua_toboolean(L, -1);
            lua_pop(L, 1);

            lua_getfield(L, -1, "locked");
            if (lua_isboolean(L, -1)) p.config.locked = lua_toboolean(L, -1);
            lua_pop(L, 1);

            // Panel 0 = main viz (state.display + state.viz_config).
            if (i == 1) {
                state.display    = p.display;
                state.viz_config = p.config;
            } else {
                state.extra_panels.push_back(p);
            }
            lua_pop(L, 1);
        }
    }
    lua_pop(L, 1); // views

    lua_pop(L, 1); // pop root
    return finish_with_warns(true);
}

} // namespace tql