*! version 1.2.3 20261030 github.com/mdroste/stata-ctools

program define csort, rclass
    version 14.1

    * Check observation limit (Stata plugin API limitation)
    if _N > 2147483647 {
        di as error "ctools does not support datasets exceeding 2^31 (2.147 billion) observations"
        di as error "This is a limitation of Stata's plugin API"
        exit 920
    }

    syntax varlist, [Verbose ALGorithm(string) STReam(integer 0) THReads(integer 0) NOSORTedby NOSTReam]

    * Start wall-clock timer as early as possible to match rmsg
    local __do_timing = ("`verbose'" != "")
    if `__do_timing' {
        local __ct_elapsed90 = 0
        if "`__ct_elapsed90'" == "" local __ct_elapsed90 = 0
        quietly _ctools_clock __ct_start90
    }

    * =========================================================================
    * UPFRONT VALIDATION - check all options before any data manipulation
    * =========================================================================

    * Count variables for the sort
    local nvars : word count `varlist'
    if (`nvars' == 0) {
        di as error "csort: no sort variables specified"
        exit 198
    }

    * Validate algorithm option early (before any data operations)
    * Default is "auto" which intelligently selects based on data characteristics
    local alg_code ""
    local alg_name "auto"
    if "`algorithm'" != "" {
        local algorithm = lower("`algorithm'")
        if "`algorithm'" == "lsd" | "`algorithm'" == "0" {
            local alg_code "alg=lsd"
            local alg_name "lsd"
        }
        else if "`algorithm'" == "msd" | "`algorithm'" == "1" {
            local alg_code "alg=msd"
            local alg_name "msd"
        }
        else if "`algorithm'" == "timsort" | "`algorithm'" == "tim" | "`algorithm'" == "2" {
            local alg_code "alg=timsort"
            local alg_name "timsort"
        }
        else if "`algorithm'" == "sample" | "`algorithm'" == "3" {
            local alg_code "alg=sample"
            local alg_name "sample"
        }
        else if "`algorithm'" == "counting" | "`algorithm'" == "count" | "`algorithm'" == "4" {
            local alg_code "alg=counting"
            local alg_name "counting"
        }
        else if "`algorithm'" == "merge" | "`algorithm'" == "5" {
            local alg_code "alg=merge"
            local alg_name "merge"
        }
        else if "`algorithm'" == "ips4o" | "`algorithm'" == "6" {
            local alg_code "alg=ips4o"
            local alg_name "ips4o"
        }
        else if "`algorithm'" == "auto" | "`algorithm'" == "7" {
            local alg_code "alg=auto"
            local alg_name "auto"
        }
        else {
            di as error "csort: invalid algorithm '`algorithm''"
            di as error "Valid options: auto (default), lsd, msd, timsort, sample, counting, merge, ips4o"
            exit 198
        }
    }

    * Every column may need to move, including non-key strL payloads. Reject
    * unsupported storage before the sortedby shortcut as well as plugin I/O.
    unab allvars : *
    foreach v of local allvars {
        local storage : type `v'
        if "`storage'" == "strL" {
            di as error "csort: strL I/O is unsupported; recast to str2045 when lossless"
            exit 109
        }
    }

    * =========================================================================
    * EARLY EXIT: check if data is already sorted by the requested variables
    * =========================================================================

    * Skip check if if/in specified (sortedby applies to full dataset only)
    if `"`if'"' == "" & `"`in'"' == "" {
        local __sortedby : sortedby
        if "`__sortedby'" == "`varlist'" {
            exit 0
        }
    }

    * =========================================================================
    * END UPFRONT VALIDATION
    * =========================================================================

    * Start pre-plugin timer
    if `__do_timing' {
        local __ct_elapsed91 = 0
        local __ct_elapsed92 = 0
        local __ct_elapsed93 = 0
        if "`__ct_elapsed91'" == "" local __ct_elapsed91 = 0
        quietly _ctools_clock __ct_start91
    }

    * Load the platform-appropriate ctools plugin if not already loaded
    _ctools_load
    * Stata scopes plugin registrations to the calling ado program.
    capture program ctools_plugin, plugin using("`__ctools_plugin'")
    if _rc != 0 & _rc != 110 exit 601
    capture confirm number 0

    * Find indices of sort variables within all variables (1-based). A
    * repeated key adds nothing to the order (sorting by x x is sorting by x),
    * so the plugin gets each key once; the final sort keeps varlist as typed.
    local sortkeys : list uniq varlist
    local var_indices ""
    foreach sortvar of local sortkeys {
        local idx : list posof "`sortvar'" in allvars
        local var_indices "`var_indices' `idx'"
    }

    * Set string width metadata for flat buffer optimization
    * _ctools_strw sets local __ctools_strw (read automatically by C plugin)
    _ctools_strw `allvars'
    local strw_code "strw:`__ctools_strw'"

    * Build stream option string
    * stream(#) enables streaming mode with # variables loaded at a time (1-16)
    local stream_code ""
    if `stream' > 0 {
        * Validate and constrain batch size
        local batch_size = `stream'
        if `batch_size' > 16 {
            local batch_size = 16
        }
        local stream_code "stream(`batch_size')"
    }

    * Build nostream option string (disables auto-streaming)
    local nostream_code ""
    if "`nostream'" != "" {
        local nostream_code "nostream"
    }

    * Build threads option string
    local threads_code ""
    if `threads' > 0 {
        local threads_code "threads(`threads')"
    }

    * End pre-plugin timer, start plugin timer
    if `__do_timing' {
        quietly _ctools_clock __ct_now
        local __ct_elapsed91 = `__ct_elapsed91' + (`__ct_now' - `__ct_start91')
        if "`__ct_elapsed92'" == "" local __ct_elapsed92 = 0
        quietly _ctools_clock __ct_start92
    }

    * Call the C plugin with ALL variables (so it can sort the entire dataset)
    plugin call ctools_plugin `allvars' `if' `in', "csort `threads_code' `var_indices' `alg_code' `stream_code' `nostream_code' `strw_code'"

    * End plugin timer, start post-plugin timer
    if `__do_timing' {
        quietly _ctools_clock __ct_now
        local __ct_elapsed92 = `__ct_elapsed92' + (`__ct_now' - `__ct_start92')
        if "`__ct_elapsed93'" == "" local __ct_elapsed93 = 0
        quietly _ctools_clock __ct_start93
    }

    * Set sort order in Stata (use stable to match csort's stable radix sort)
    * Skip if nosortedby option is specified
    if "`nosortedby'" == "" {
        sort `varlist', stable
    }
    else {
        * The plugin reordered the rows, but Stata still holds the previous
        * sort marker, which by, merge and csort's own early exit trust.
        _csort_clear_sortedby
    }

    * End post-plugin timer (wall-clock timer 90 stays running)
    if `__do_timing' {
        quietly _ctools_clock __ct_now
        local __ct_elapsed93 = `__ct_elapsed93' + (`__ct_now' - `__ct_start93')

        * Extract timer values (timer 90 still running — captures up to this point)
        * Elapsed seconds are held in caller-local timing state.
        local __time_preplugin = `__ct_elapsed91'
        * Elapsed seconds are held in caller-local timing state.
        local __time_plugin = `__ct_elapsed92'
        * Elapsed seconds are held in caller-local timing state.
        local __time_postplugin = `__ct_elapsed93'

        * Calculate plugin call overhead (time in plugin call but outside C code)
        local __plugin_call_overhead = `__time_plugin' - _csort_time_total

        local __ado_overhead = `__time_preplugin' + `__time_postplugin'
        local __ado_total = `__ado_overhead' + `__plugin_call_overhead'

        * Check if stream mode was used
        capture local __is_stream = _csort_stream
        if _rc != 0 local __is_stream = 0

        * Check if permutation time is available
        capture local __time_permute = _csort_time_permute
        if _rc != 0 local __time_permute = 0

        * Check if streaming time is available
        capture local __time_stream = _csort_time_stream
        if _rc != 0 local __time_stream = 0

        di as text ""
        di as text "{hline 55}"
        if `__is_stream' {
            di as text "csort timing breakdown (streaming mode):"
        }
        else {
            di as text "csort timing breakdown:"
        }
        di as text "{hline 55}"
        di as text "  C plugin internals:"

        if `__is_stream' {
            * Memory-efficient mode timing display
            di as text "    Load key vars:          " as result %8.4f _csort_time_load " sec"
            di as text "    Sort (compute order):   " as result %8.4f _csort_time_sort " sec"
            di as text "    Permute keys in C:      " as result %8.4f `__time_permute' " sec"
            di as text "    Store sorted keys:      " as result %8.4f _csort_time_store " sec"
            di as text "    Stream non-key vars:    " as result %8.4f `__time_stream' " sec"

            * Detailed breakdown of stream non-key vars phase
            capture local __stream_build_inv = _csort_stream_time_build_inv
            if _rc == 0 {
                capture local __stream_scatter = _csort_stream_time_scatter
                capture local __stream_writeback = _csort_stream_time_writeback
                capture local __stream_strings = _csort_stream_time_strings
                capture local __n_numeric = _csort_stream_n_numeric
                capture local __n_string = _csort_stream_n_string
                if _rc == 0 {
                    di as text "      ├─ Build inv perm:    " as result %8.4f `__stream_build_inv' " sec"
                    di as text "      ├─ Scatter (read+buf):" as result %8.4f `__stream_scatter' " sec" as text " (" as result `__n_numeric' as text " numeric vars)"
                    di as text "      ├─ Writeback to Stata:" as result %8.4f `__stream_writeback' " sec"
                    if `__n_string' > 0 {
                        di as text "      └─ String vars:       " as result %8.4f `__stream_strings' " sec" as text " (" as result `__n_string' as text " string vars)"
                    }
                }
            }
        }
        else {
            * Standard mode timing display
            di as text "    Data load:              " as result %8.4f _csort_time_load " sec"
            if `__time_permute' > 0 {
                di as text "    Sort (compute order):   " as result %8.4f _csort_time_sort " sec"
                di as text "    Sort (apply permute):   " as result %8.4f `__time_permute' " sec"
            }
            else {
                di as text "    Sort:                   " as result %8.4f _csort_time_sort " sec"
            }
            di as text "    Data store:             " as result %8.4f _csort_time_store " sec"
            di as text "    Memory cleanup:         " as result %8.4f _csort_time_cleanup " sec"
        }

        di as text "  {hline 53}"
        di as text "    C plugin total:         " as result %8.4f _csort_time_total " sec"
        di as text "  {hline 53}"
        di as text "  Stata overhead:"
        di as text "    Pre-plugin parsing:     " as result %8.4f `__time_preplugin' " sec"
        di as text "    Plugin call overhead:   " as result %8.4f `__plugin_call_overhead' " sec"
        di as text "    Post-plugin sort:       " as result %8.4f `__time_postplugin' " sec"
        di as text "  {hline 53}"
        di as text "    Stata overhead total:   " as result %8.4f `__ado_total' " sec"
        di as text "{hline 55}"
        * Stop wall-clock timer right before reading
        quietly _ctools_clock __ct_now
        local __ct_elapsed90 = `__ct_elapsed90' + (`__ct_now' - `__ct_start90')
        * Elapsed seconds are held in caller-local timing state.
        local __time_total = `__ct_elapsed90'
        di as text "    Wall clock total:       " as result %8.4f `__time_total' " sec"
        di as text "{hline 55}"

        * Display thread diagnostics
        capture local __threads_max = _csort_threads_max
        if _rc == 0 {
            capture local __openmp_enabled = _csort_openmp_enabled
            if _rc != 0 local __openmp_enabled = 0
            di as text ""
            di as text "  Thread diagnostics:"
            di as text "    OpenMP enabled:         " as result %8.0f `__openmp_enabled'
            di as text "    Max threads available:  " as result %8.0f `__threads_max'
            di as text "{hline 55}"
        }
    }

    * =========================================================================
    * STORE RESULTS IN r()
    * =========================================================================

    * Get values from scalars set by C plugin
    capture local __time_load = _csort_time_load
    if _rc != 0 local __time_load = .
    capture local __time_sort = _csort_time_sort
    if _rc != 0 local __time_sort = .
    capture local __time_permute = _csort_time_permute
    if _rc != 0 local __time_permute = .
    capture local __time_store = _csort_time_store
    if _rc != 0 local __time_store = .
    capture local __time_stream = _csort_time_stream
    if _rc != 0 local __time_stream = .
    capture local __time_cleanup = _csort_time_cleanup
    if _rc != 0 local __time_cleanup = .
    capture local __time_total_c = _csort_time_total
    if _rc != 0 local __time_total_c = .
    capture local __is_stream = _csort_stream
    if _rc != 0 local __is_stream = 0
    capture local __threads_max = _csort_threads_max
    if _rc != 0 local __threads_max = .
    capture local __openmp_enabled = _csort_openmp_enabled
    if _rc != 0 local __openmp_enabled = .

    * Post r() results
    return scalar N = _N
    return scalar time_load = `__time_load'
    return scalar time_sort = `__time_sort'
    return scalar time_permute = `__time_permute'
    return scalar time_store = `__time_store'
    return scalar time_stream = `__time_stream'
    return scalar time_cleanup = `__time_cleanup'
    return scalar time_total = `__time_total_c'
    return scalar stream = `__is_stream'
    return scalar threads_max = `__threads_max'
    return scalar openmp_enabled = `__openmp_enabled'

    return local algorithm "`alg_name'"
    return local sortvars "`varlist'"
    return local cmd "csort"

end

* Clear Stata's sort marker after the plugin permuted the rows without a
* final sort: a real change to a marked variable drops the marker, so change
* observation 1 of the first marked variable and restore it.
program define _csort_clear_sortedby
    local marked : sortedby
    if "`marked'" == "" | _N == 0 exit
    local v : word 1 of `marked'
    * Stata drops the sort marker when a sort variable changes: change one
    * value with replace, then put the original back through Mata (no copy
    * of the variable is needed)
    capture confirm string variable `v'
    if _rc {
        mata: __ctools_sb_hold = st_data(1, "`v'")
        quietly replace `v' = cond(`v' == 0, 1, 0) in 1
        mata: st_store(1, "`v'", __ctools_sb_hold)
    }
    else {
        mata: __ctools_sb_hold = st_sdata(1, "`v'")
        quietly replace `v' = cond(`v' == "", "x", "") in 1
        mata: st_sstore(1, "`v'", __ctools_sb_hold)
    }
    mata: mata drop __ctools_sb_hold
end