🥮 mooncakes.io

    external-sort

    Stable bounded-memory external sorting and merge planning for MoonBit

    external-sort
    bounded-memory
    k-way-merge
    jsonl
    Download zip
    Author
    Version
    0.1.0
    License
    Apache-2.0
    Repository
    Last updated
    yesterday
    Downloads
    1

    Dependencies

    #yyyt0807/external-sort

    Stable bounded-memory external sorting primitives for MoonBit.

    ///|
    test {
      let config = SortConfig::new(memory_budget_bytes=1024, max_record_bytes=256)
      let builder = RunBuilder::new(config)
      ignore(builder.push("second", "b"))
      ignore(builder.push("first", "a"))
      guard builder.finish() is Some(run) else { fail("expected run") }
      assert_eq(run.map(record => record.payload), ["first", "second"])
    }

    The root package is portable. Native filesystem execution and recovery live in yyyt0807/external-sort/adapter/native; the executable is cmd/moon-external-sort.

    CsvError

    source
    pub(all) suberror CsvError {
      InvalidDialect(String)
      MalformedCsv(offset~ : Int, message~ : String)
      CsvFieldMissing(index~ : Int, available~ : Int)
    } derive(Eq, 
    Debug    )

    CSV parsing and rendering failures. This codec operates on one logical record, so physical line framing remains the caller's responsibility.

    SortError

    source
    pub(all) suberror SortError {
      InvalidConfig(String)
      InvalidKey(String)
      RecordTooLarge(actual~ : Int, limit~ : Int)
      SelectionBudgetExceeded(required~ : Int, limit~ : Int)
      SequenceExhausted
      MalformedRun(String)
      InvalidManifest(String)
    } derive(Eq, 
    Debug    )

    Structured failures that callers can distinguish from host I/O failures.

    CsvDialect

    source
    pub(all) struct CsvDialect {
      delimiter : Char
      quote : Char
    } derive(Eq, 
    Debug    )

    Syntax characters for the single-record CSV codec.

    CsvDialect::new

    source
    fn CsvDialect::new(delimiter? : Char, quote? : Char) -> CsvDialect raise CsvError

    Construct a validated dialect.

    DecimalKeyValue

    source
    pub(all) struct DecimalKeyValue {
      negative : Bool
      digits : String
      scale : Int
    } derive(Eq, 
    Debug    )

    Canonical arbitrary-precision decimal used only for ordering. digits has no leading zero, scale counts fractional digits, and zero is never negative.

    DecimalKeyValue::to_canonical_string

    source
    fn DecimalKeyValue::to_canonical_string(self : DecimalKeyValue) -> String

    Render a unique plain-decimal representation suitable for Run files.

    GroupStatistics

    source
    pub(all) struct GroupStatistics {
      records : Int64
      distinct_keys : Int64
      singleton_groups : Int64
      repeated_groups : Int64
      largest_group_records : Int64
      first_key : SortValue?
      last_key : SortValue?
      verification : VerificationReport
    } derive(Eq, 
    Debug    )

    Constant-space statistics over adjacent equal-key groups.

    GroupStatistics::average_group_size

    source
    fn GroupStatistics::average_group_size(self : GroupStatistics) -> Double

    GroupStatistics::duplicate_records

    source
    fn GroupStatistics::duplicate_records(self : GroupStatistics) -> Int64

    GroupStatistics::is_sorted

    source
    fn GroupStatistics::is_sorted(self : GroupStatistics) -> Bool

    JobEstimate

    source
    pub(all) struct JobEstimate {
      input_records : Int64
      input_payload_bytes : Int64
      average_payload_bytes : Int64
      records_per_run : Int64
      initial_runs : Int64
      merge_passes : Int
      minimum_payload_io_bytes : Int64
      peak_payload_generation_bytes : Int64
      schedule : Array[MergePassEstimate]
      saturated : Bool
    } derive(Eq, 
    Debug    )

    Static planning result. Byte estimates describe original payload bytes; internal JSONL escaping and host filesystem allocation are intentionally not presented as exact values.

    JobManifest

    source
    pub(all) struct JobManifest {
      phase : JobPhase
      input_path : String
      output_path : String
      work_directory : String
      config : SortConfig
      selector : KeySelector
      input_records : Int64
      initial_run_count : Int
      merge_pass : Int
      runs : Array[RunDescriptor]
    } derive(Eq, 
    Debug    )

    Portable recovery state. A Native adapter commits this document atomically only after every Run referenced by it has been fully written.

    JobPhase

    source
    pub(all) enum JobPhase {
      RunsReady
      Merging
      Published
    } derive(Eq, 
    Debug    )

    Durable phases in which a Sort Job may be recovered.

    KeyKind

    source
    pub(all) enum KeyKind {
      TextKey
      IntegerKey
      DecimalKey
    } derive(Eq, 
    Debug    )

    The supported v0.1 Sort Key domains.

    KeySelector

    source
    pub(all) enum KeySelector {
      WholeRecord
      DelimitedField(index~ : Int, delimiter~ : String)
      CsvField(index~ : Int, delimiter~ : Char)
      JsonField(String)
    } derive(Eq, 
    Debug    )

    A reproducible way to derive the Sort Key from one textual Input Record.

    LineFramer

    source
    pub struct LineFramer {
      max_record_bytes : Int
      pending : Array[Byte]
      finished : Bool
    }

    Bounded byte-oriented newline framing. It avoids an unbounded host read_until buffer before max_record_bytes can be checked.

    LineFramer::finish

    source
    fn LineFramer::finish(self : LineFramer) -> Bytes? raise SortError

    Return the final unterminated line, if any. A trailing LF does not create an additional record.

    LineFramer::new

    source
    fn LineFramer::new(max_record_bytes : Int) -> LineFramer raise SortError

    LineFramer::pending_bytes

    source
    fn LineFramer::pending_bytes(self : LineFramer) -> Int

    LineFramer::push

    source
    fn LineFramer::push(self : LineFramer, chunk : Bytes) -> Array[Bytes] raise SortError

    Consume an arbitrary byte chunk and return every complete line without its LF delimiter. Empty lines are retained.

    MergeGroup

    source
    pub(all) struct MergeGroup {
      start : Int
      end : Int
    } derive(Eq, 
    Debug    )

    One bounded Merge Group expressed as half-open Run indexes.

    MergePassEstimate

    source
    pub(all) struct MergePassEstimate {
      pass : Int
      input_runs : Int64
      output_runs : Int64
      largest_group : Int
    } derive(Eq, 
    Debug    )

    One row in a multi-pass merge schedule.

    OrderViolation

    source
    pub(all) struct OrderViolation {
      previous_index : Int64
      current_index : Int64
      previous_key : SortValue
      current_key : SortValue
    } derive(Eq, 
    Debug    )

    The first adjacent pair that violates the configured order.

    RunBuilder

    source
    pub struct RunBuilder {
      config : SortConfig
      records : Array[SortRecord]
      retained_bytes : Int
      next_position : Int64
    }

    Incrementally creates bounded, internally ordered Runs.

    RunBuilder::finish

    source
    fn RunBuilder::finish(self : RunBuilder) -> Array[SortRecord]?

    Complete the final Run, or return None when no records remain.

    RunBuilder::new

    source
    fn RunBuilder::new(config : SortConfig, start_position? : Int64) -> RunBuilder raise SortError

    RunBuilder::next_position

    source
    fn RunBuilder::next_position(self : RunBuilder) -> Int64

    RunBuilder::push

    source
    fn RunBuilder::push(self : RunBuilder, payload : String, key_text : String) -> Array[SortRecord]? raise SortError

    Accept one record and return a completed Run when the next record would exceed the Resource Budget. The new record remains in the next Run.

    RunBuilder::retained_bytes

    source
    fn RunBuilder::retained_bytes(self : RunBuilder) -> Int

    RunCapacityEstimate

    source
    pub(all) struct RunCapacityEstimate {
      worst_case_record_bytes : Int64
      records_per_run : Int64
      conservative_merge_head_bytes : Int64
      merge_heads_fit_run_budget : Bool
    } derive(Eq, 
    Debug    )

    Conservative record and merge-head capacity derived from a SortConfig.

    RunDescriptor

    source
    pub(all) struct RunDescriptor {
      path : String
      record_count : Int64
    } derive(Eq, 
    Debug    )

    One committed Run referenced by a Job Manifest.

    SortConfig

    source
    pub(all) struct SortConfig {
      order : SortOrder
      key_kind : KeyKind
      memory_budget_bytes : Int
      max_open_runs : Int
      max_record_bytes : Int
    } derive(Eq, 
    Debug    )

    Hard algorithmic limits for a Sort Job.

    SortConfig::new

    source
    fn SortConfig::new(order? : SortOrder, key_kind? : KeyKind, memory_budget_bytes? : Int, max_open_runs? : Int, max_record_bytes? : Int) -> SortConfig raise SortError

    Construct and validate a Resource Budget.

    SortOrder

    source
    pub(all) enum SortOrder {
      Ascending
      Descending
    } derive(Eq, 
    Debug    )

    Whether Sort Keys are compared from least to greatest or greatest to least.

    SortRecord

    source
    pub(all) struct SortRecord {
      key : SortValue
      input_position : Int64
      payload : String
    } derive(Eq, 
    Debug    )

    One Input Record. input_position is always an ascending tie-breaker, even for descending Sort Jobs, which makes the ordering stable.

    SortValue

    source
    pub(all) enum SortValue {
      TextValue(String)
      IntegerValue(Int64)
      DecimalValue(DecimalKeyValue)
    } derive(Eq, 
    Debug    )

    A parsed Sort Key. Keeping the value typed prevents lexical ordering from silently being used for signed integers.

    SortedGroupCounter

    source
    pub struct SortedGroupCounter {
      config : SortConfig
      verifier : SortednessVerifier
      previous_key : SortValue?
      first_key : SortValue?
      last_key : SortValue?
      records : Int64
      distinct_keys : Int64
      singleton_groups : Int64
      repeated_groups : Int64
      current_group_records : Int64
      largest_group_records : Int64
      next_position : Int64
      finished : Bool
    }

    Streaming aggregation intended for sorted output. Statistics remain useful when input is disordered, while verification preserves the first violation.

    SortedGroupCounter::finish

    source

    Complete the last group and return immutable statistics.

    SortedGroupCounter::new

    source
    fn SortedGroupCounter::new(config : SortConfig, start_position? : Int64) -> SortedGroupCounter raise SortError

    SortedGroupCounter::push

    source
    fn SortedGroupCounter::push(self : SortedGroupCounter, payload : String, key_text : String) -> Unit raise SortError

    Parse and aggregate one external key.

    SortedGroupCounter::push_record

    source
    fn SortedGroupCounter::push_record(self : SortedGroupCounter, record : SortRecord) -> Unit raise SortError

    Aggregate a record carrying an original position, enabling stability checks.

    SortednessVerifier

    source
    pub struct SortednessVerifier {
      config : SortConfig
      previous : SortRecord?
      records : Int64
      equal_adjacent_keys : Int64
      violation : OrderViolation?
    }

    Stateful constant-space verifier for an already ordered record stream.

    SortednessVerifier::new

    source

    SortednessVerifier::push

    source
    fn SortednessVerifier::push(self : SortednessVerifier, payload : String, key_text : String) -> Unit raise SortError

    Verify one payload/key pair. Once a violation is found, later records are counted but do not replace the first diagnostic.

    SortednessVerifier::push_record

    source
    fn SortednessVerifier::push_record(self : SortednessVerifier, current : SortRecord) -> Unit

    Verify a record that already carries its original position. This overload can also detect unstable ordering among equal keys.

    SortednessVerifier::report

    source

    TopKResult

    source
    pub(all) struct TopKResult {
      records : Array[SortRecord]
      input_records : Int64
      discarded_records : Int64
      retained_bytes : Int
    } derive(Eq, 
    Debug    )

    A completed stable Top-K selection and its resource accounting.

    TopKSelector

    source
    pub struct TopKSelector {
      config : SortConfig
      limit : Int
      heap : Array[SortRecord]
      retained_bytes : Int
      input_records : Int64
      discarded_records : Int64
      next_position : Int64
      finished : Bool
    }

    Streaming stable Top-K selection. The heap root is the worst retained record, so an incoming better record can replace it in logarithmic time.

    TopKSelector::finish

    source
    fn TopKSelector::finish(self : TopKSelector) -> TopKResult raise SortError

    Return retained records in final output order. This method is single-use so accounting cannot silently diverge after its array is handed to the caller.

    TopKSelector::new

    source
    fn TopKSelector::new(config : SortConfig, limit : Int, start_position? : Int64) -> TopKSelector raise SortError

    TopKSelector::observed_count

    source
    fn TopKSelector::observed_count(self : TopKSelector) -> Int64

    TopKSelector::push

    source
    fn TopKSelector::push(self : TopKSelector, payload : String, key_text : String) -> Unit raise SortError

    Consider one record. Records equal on key remain ordered by their original input positions because compare_records supplies the stable tie-breaker.

    TopKSelector::retained_bytes

    source
    fn TopKSelector::retained_bytes(self : TopKSelector) -> Int

    TopKSelector::retained_count

    source
    fn TopKSelector::retained_count(self : TopKSelector) -> Int

    VerificationReport

    source
    pub(all) struct VerificationReport {
      records : Int64
      equal_adjacent_keys : Int64
      violation : OrderViolation?
    } derive(Eq, 
    Debug    )

    Summary produced by a streaming sortedness check.

    VerificationReport::is_sorted

    source
    fn VerificationReport::is_sorted(self : VerificationReport) -> Bool

    build_merge_schedule

    source
    fn build_merge_schedule(initial_runs : Int64, max_open_runs : Int) -> Array[MergePassEstimate] raise SortError

    Build every pass needed to reduce initial_runs to one. The empty and singleton cases need no merge passes.

    compare_decimal_keys

    source
    fn compare_decimal_keys(left : DecimalKeyValue, right : DecimalKeyValue) -> Int

    Compare exact decimal values without allocation proportional to their aligned scale and without integer overflow.

    compare_records

    source
    fn compare_records(left : SortRecord, right : SortRecord, order : SortOrder) -> Int

    Compare two records according to Sort Key and stable Input Position.

    compare_sort_values

    source
    fn compare_sort_values(left : SortValue, right : SortValue) -> Int

    Compare keys without applying direction or stable position tie-breaking.

    decode_manifest

    source
    fn decode_manifest(text : String) -> JobManifest raise SortError

    decode_run_record

    source
    fn decode_run_record(line : String) -> SortRecord raise SortError

    Decode and validate one Run record.

    encode_csv_field

    source
    fn encode_csv_field(value : String, dialect? : CsvDialect) -> String

    Render one field with the minimum quoting necessary for the dialect.

    encode_csv_record

    source
    fn encode_csv_record(fields : Array[String], dialect? : CsvDialect) -> String

    Encode a complete record using the supplied dialect.

    encode_manifest

    source
    fn encode_manifest(manifest : JobManifest) -> String

    encode_run_record

    source
    fn encode_run_record(record : SortRecord) -> String

    Encode one Run record as a self-contained JSON line. JSON escaping preserves tabs, control characters, and arbitrary UTF-8 payloads without a custom delimiter ambiguity.

    estimate_job

    source
    fn estimate_job(input_records : Int64, input_payload_bytes : Int64, config : SortConfig) -> JobEstimate raise SortError

    Estimate a job from record count and original payload bytes. The model is deterministic and saturates instead of wrapping Int64 counters.

    estimate_run_capacity

    source
    fn estimate_run_capacity(config : SortConfig) -> RunCapacityEstimate

    Compute conservative record retention from configured maximum input size.

    estimated_record_bytes

    source
    fn estimated_record_bytes(record : SortRecord) -> Int

    Conservative retained-size estimate used to enforce the declared budget.

    job_estimate_json

    source
    fn job_estimate_json(estimate : JobEstimate) -> String

    Stable JSON form suitable for CLI wrappers and benchmark records.

    merge_pass_count

    source
    fn merge_pass_count(run_count : Int, max_open_runs : Int) -> Int raise SortError

    Number of Merge Passes required to reduce run_count Runs to one.

    merge_sorted_runs

    source
    fn merge_sorted_runs(runs : Array[Array[SortRecord]], order : SortOrder) -> Array[SortRecord]

    Reference k-way merge over materialized Runs. Native storage adapters use the same ordering contract while loading only one head per open Run.

    parse_csv_record

    source
    fn parse_csv_record(record : String, dialect? : CsvDialect) -> Array[String] raise CsvError

    Parse one CSV record. Quotes are allowed only at the beginning of a field, and a doubled quote inside a quoted field decodes to one quote.

    parse_decimal_key

    source
    fn parse_decimal_key(text : String) -> DecimalKeyValue raise SortError

    Parse a plain decimal without converting through floating point. Exponents, NaN and infinity are deliberately rejected to keep the accepted language reproducible across MoonBit backends.

    parse_key

    source
    fn parse_key(text : String, kind : KeyKind) -> SortValue raise SortError

    Parse a typed Sort Key from its external text representation.

    plan_merge_pass

    source
    fn plan_merge_pass(run_count : Int, max_open_runs : Int) -> Array[MergeGroup] raise SortError

    Plan one Merge Pass without opening files or allocating record buffers.

    select_csv_field

    source
    fn select_csv_field(record : String, index : Int, dialect? : CsvDialect) -> String raise CsvError

    Select one decoded field and report the available field count on failure.

    validate_selector

    source
    fn validate_selector(selector : KeySelector) -> Unit raise SortError

    Validate a Key Selector before reading input.

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io