🥮 mooncakes.io
    Sign in

    moondelta-core

    MoonBit-native Delta Lake table-format kernel: transaction log replay, snapshots, time travel, scan planning and append writes.

    delta-lake
    table-format
    lakehouse
    parquet
    data
    Download zip
    Version
    0.1.0
    License
    Apache-2.0
    Repository
    Last updated
    9 hours ago
    Downloads
    4

    Dependencies

    #moondelta-core

    CI

    MoonBit 原生的 Delta Lake 表格式内核：回放 _delta_log 事务日志、构建快照、时间旅行、分区/统计剪枝的扫描规划，读取数据文件，并在受支持的列类型上 append 提交新版本。

    生态里已经有 Parquet 编解码、DataFrame、对 Parquet 跑 SQL 的查询引擎、Iceberg 只读——缺的是"带事务日志的表"这一层。本项目补这一层。

    参赛申报书：申报书.md ｜ 第三方声明：THIRD_PARTY_NOTICES.md

    #状态

    MVP（0.1.0）：

    • 日志解析与回放：protocol / metaData / add / remove / commitInfo，_last_checkpoint 识别
    • 快照、版本历史、时间旅行（snapshot_at）
    • 扫描规划：分区等值过滤 + 文件级 min/max 统计剪枝
    • 读取：flat schema + UNCOMPRESSED/SNAPPY 的 Delta 数据文件
    • 写入：create_table + append_rows（Int32/Int64/String/Binary，Required/Optional）
    • CLI：inspect / history / scan / read / create / version
    • 10 个单元测试 + 真实 Delta 表 fixture（delta-rs、apache/parquet-testing）

    已用真实数据验证：delta-rs table_a（13 个版本回放 → 120 个文件）、Databricks 写出的 COVID-19_NYT 表（157,865 行 6 列，snappy）。

    #快速开始

    要求：MoonBit 工具链（moon / moonc，建议 2026-10 之后的版本）。

    # 从源码构建
moon check --target native
moon test --target native

# CLI（源码根目录为 src/，因此包路径带 src/ 前缀）
moon run --target native src/cmd/delta -- inspect fixtures/delta/table_a
moon run --target native src/cmd/delta -- history fixtures/delta/table_a
moon run --target native src/cmd/delta -- scan fixtures/delta/table_a --where "id>=60"
moon run --target native src/cmd/delta -- read fixtures/delta/covid --limit 5

    inspect fixtures/delta/table_a 输出示例：

    table            : fixtures/delta/table_a
current version  : 12  (log versions: 13)
files            : 120  rows: 120  bytes: 54232
protocol         : reader=1 writer=1
format           : parquet
partition columns: (none)
schema:
  id: int
last checkpoint  : version=10
note             : checkpoint files exist; checkpoint-only tables are not readable by this kernel

    scan ... --where "id>=60" 会利用日志里的 min/max 统计把 120 个文件剪到 60 个（skipped stats: 60）。

    #作为库使用

    let table = @delta.DeltaTable::open("path/to/table")
    let snapshot = table.snapshot()              // 最新版本
    let old = table.snapshot_at(1)               // 时间旅行
    let report = @delta.scan(snapshot, {
      partitions: [("dt", "2026-10-18")],
      predicate: Some({ column: "id", op: GeV, value: 60.0 }),
    })
    let rows = @delta.read_snapshot("path/to/table", snapshot, Some(100))
    for c in rows.columns {
      println(c.name + " : " + @delta.delta_type_name(c.type_))
    }

    写入：

    @delta.create_table(root, schema_json, [])
    let columns : Array[@delta.ReadColumn] = [
      { name: "id", type_: Int64Type, cells: [CellI64(1L), CellI64(2L)] },
      { name: "name", type_: StringType, cells: [CellStr("a"), CellStr("b")] },
    ]
    @delta.append_rows(root, columns, 1750000000000L) // 写 parquet + 提交新版本

    #能力边界（遇到即报错，不静默降级）

    维度支持不支持
    schemaflat 原始类型嵌套 struct / list / map
    压缩UNCOMPRESSED、SNAPPYZSTD、GZIP、BROTLI、LZ4
    逻辑类型string、int、long、float、double、boolean、date、timestamp(INT96)decimal、uuid 等（会丢语义）
    表日志完整、可从头回放的本地表checkpoint-only / 日志被截断的表
    写入类型Int32 / Int64 / String / Binary（Required/Optional）Boolean / Float / Double / Timestamp / Repeated
    写入表非分区表分区表 append（MVP 明确拒绝）
    存储本地文件系统对象存储（S3/R2）、删除向量、列映射

    这些边界来自底层 Parquet 实现的真实能力（见 src/vendor/parquet/NOTICE.md），已在测试里固化为明确的错误信息（不是 panic，也不是静默 0 行）。

    #依赖说明：vendored parquet

    mizchi/parquet@0.2.2 声明了 mizchi/duckdb@0.6.4，后者仍使用旧版 moon.mod.json，当前 MoonBit 工具链会拒绝解析。该依赖只被上游的 DuckDB 演示子包使用，因此本项目把 parquet 的 src/ 以 Apache-2.0 许可证 vendored 到 src/vendor/parquet，删除了 cmd/duckdb_e2e 与 playground_bridge 两个子包，reader/writer 实现未改动。详见 src/vendor/parquet/NOTICE.md。

    #Fixtures 与数据来源

    fixtures/ 中仅包含测试数据（不含上游实现代码），来源与许可证见 fixtures/README.md：

    • fixtures/delta/table_a：delta-rs 测试表（13 个日志版本 + checkpoint），Apache-2.0
    • fixtures/delta/covid：Databricks 写出的 COVID-19_NYT 表数据文件，Apache-2.0
    • fixtures/parquet/list_columns.parquet：apache/parquet-testing 的嵌套类型样例（负向用例），Apache-2.0

    #路线图

    • 写入扩展：Double / Boolean / Timestamp 列
    • 读取扩展：ZSTD/GZIP/BROTLI 解压、嵌套 schema、checkpoint parquet 解析
    • 对象存储（S3/R2）抽象
    • 与 Arrow IPC / DataFrame / moonice（Iceberg）的互操作示例
    • 增量扫描与变更数据（CDF）读取

    #开发

    moon check --target native
moon test --target native
moon fmt

    CI：.github/workflows/ci.yml（安装最新 MoonBit 工具链 → moon update → check/test/build）。

    #许可证

    Apache-2.0。Delta Lake 协议参考 https://github.com/delta-io/delta（Apache-2.0）；本项目为原创实现，未复制上游代码。

    DeltaError

    source
    pub(all) suberror DeltaError {
      IoError(String)
      ParseError(String)
      Unsupported(String)
      NotFound(String)
    } derive(Eq, 
    Debug    )

    impl Show for DeltaError

    Action

    source
    pub(all) enum Action {
      ProtocolAction(Protocol)
      MetaDataAction(TableInfo)
      AddAction(AddFile)
      RemoveAction(RemoveFile)
      CommitInfoAction(CommitInfo)
      UnknownAction(String)
    } derive(Eq, 
    Debug    )

    AddFile

    source
    pub(all) struct AddFile {
      path : String
      partition_values : Array[(String, String)]
      size : Int64
      modification_time : Int64
      data_change : Bool
      stats : String?
    } derive(Eq, 
    Debug    )

    Cell

    source
    pub(all) enum Cell {
      CellI32(Int)
      CellI64(Int64)
      CellF32(Float)
      CellF64(Double)
      CellBool(Bool)
      CellStr(String)
      CellBin(Bytes)
      CellTs(Int64)
      CellNull
    } derive(
    Debug    )

    读取路径的统一单元格表示（将 parquet 的值收敛成 Delta 语义类型）

    CheckpointInfo

    source
    pub(all) struct CheckpointInfo {
      version : Int
      size : Int?
      parts : Int?
    } derive(Eq, 
    Debug    )

    ColumnPredicate

    source
    pub(all) struct ColumnPredicate {
      column : String
      op : CompareOp
      value : Double
    } derive(Eq, 
    Debug    )

    CommitInfo

    source
    pub(all) struct CommitInfo {
      version : Int
      timestamp : Int64?
      operation : String?
      operation_parameters : Array[(String, String)]
    } derive(Eq, 
    Debug    )

    CompareOp

    source
    pub(all) enum CompareOp {
      EqV
      GtV
      GeV
      LtV
      LeV
    } derive(Eq, 
    Debug    )

    比较运算符（用于文件级 min/max 统计剪枝）

    DeltaField

    source
    pub(all) struct DeltaField {
      name : String
      type_ : DeltaType
      nullable : Bool
    } derive(Eq, 
    Debug    )

    DeltaSchema

    source
    pub(all) struct DeltaSchema {
      fields : Array[DeltaField]
      raw_json : String
    } derive(Eq, 
    Debug    )

    DeltaTable

    source
    pub(all) struct DeltaTable {
      root : String
    }

    DeltaTable::current_version

    source
    fn DeltaTable::current_version(self : DeltaTable) -> Int raise DeltaError

    DeltaTable::history

    source
    fn DeltaTable::history(self : DeltaTable) -> Array[HistoryEntry] raise DeltaError

    DeltaTable::inspect

    source
    fn DeltaTable::inspect(self : DeltaTable) -> (Snapshot, CheckpointInfo?, Bool) raise DeltaError

    读取表的最新版本号 + snapshot + 是否检测到 checkpoint（用于 CLI 提示）

    DeltaTable::open

    source
    fn DeltaTable::open(root : String) -> DeltaTable raise DeltaError

    DeltaTable::snapshot

    source
    fn DeltaTable::snapshot(self : DeltaTable) -> Snapshot raise DeltaError

    DeltaTable::snapshot_at

    source
    fn DeltaTable::snapshot_at(self : DeltaTable, version : Int) -> Snapshot raise DeltaError

    DeltaTable::versions

    source
    fn DeltaTable::versions(self : DeltaTable) -> Array[Int] raise DeltaError

    DeltaType

    source
    pub(all) enum DeltaType {
      Int32Type
      Int64Type
      FloatType
      DoubleType
      BooleanType
      StringType
      BinaryType
      DateType
      TimestampType
      DecimalType(precision~ : Int, scale~ : Int)
      NestedType(kind~ : String)
      UnknownType(raw~ : String)
    } derive(Eq, 
    Debug    )

    Delta Lake 状态类型（本实现支持的类型子集 + 明确的"不支持"标记）

    说明：Delta 的 schema 可以任意嵌套（struct/array/map）。本内核的读取依赖 mizchi/parquet，而该库当前只支持 flat 列；因此嵌套类型在 schema 解析阶段 被标记为 Nested，读取/写入时给出明确错误，而不是静默返回错误数据。

    HistoryEntry

    source
    pub(all) struct HistoryEntry {
      version : Int
      timestamp : Int64?
      operation : String?
    } derive(Eq, 
    Debug    )

    Protocol

    source
    pub(all) struct Protocol {
      min_reader_version : Int
      min_writer_version : Int
    } derive(Eq, 
    Debug    )

    ReadColumn

    source
    pub(all) struct ReadColumn {
      name : String
      type_ : DeltaType
      cells : Array[Cell]
    } derive(
    Debug    )

    ReadResult

    source
    pub(all) struct ReadResult {
      columns : Array[ReadColumn]
      row_count : Int
      files_read : Int
    } derive(
    Debug    )

    RemoveFile

    source
    pub(all) struct RemoveFile {
      path : String
      deletion_timestamp : Int64?
      data_change : Bool
    } derive(Eq, 
    Debug    )

    ScanFilter

    source
    pub(all) struct ScanFilter {
      partitions : Array[(String, String)]
      predicate : ColumnPredicate?
    } derive(Eq, 
    Debug    )

    扫描过滤条件：分区等值过滤 + 单列统计剪枝

    ScanReport

    source
    pub(all) struct ScanReport {
      files : Array[AddFile]
      skipped_partition : Int
      skipped_stats : Int
      matched_rows : Int64
      matched_bytes : Int64
    } derive(Eq, 
    Debug    )

    扫描结果（不读数据文件，只做文件级规划）

    Snapshot

    source
    pub(all) struct Snapshot {
      version : Int
      protocol : Protocol
      metadata : TableInfo
      files : Array[AddFile]
      total_rows : Int64
      total_bytes : Int64
    } derive(Eq, 
    Debug    )

    TableInfo

    source
    pub(all) struct TableInfo {
      id : String
      name : String?
      schema : DeltaSchema
      partition_columns : Array[String]
      format_provider : String
      configuration : Array[(String, String)]
    } derive(Eq, 
    Debug    )

    append_rows

    source
    fn append_rows(root : String, columns : Array[ReadColumn], timestamp : Int64) -> Int raise DeltaError

    追加一批行（列式输入），写 parquet 数据文件并提交新版本。 只支持非分区表 + int/long/string/binary 列。

    cell_to_display

    source
    fn cell_to_display(cell : Cell) -> String

    单元格的展示形式（CLI / 示例用）

    checkpoint_info

    source
    fn checkpoint_info(root : String) -> CheckpointInfo? raise DeltaError

    读取 _last_checkpoint（不存在返回 None）

    create_table

    source
    fn create_table(root : String, schema_json : String, partition_columns : Array[String]) -> Int raise DeltaError

    创建一张新的 Delta 表（写入 version 0：commitInfo + protocol + metaData）。

    delta_type_name

    source
    fn delta_type_name(ty : DeltaType) -> String

    describe_error

    source
    fn describe_error(e : DeltaError) -> String

    ensure_dir

    source
    fn ensure_dir(path : String) -> Unit raise DeltaError

    逐级创建目录（等价于 mkdir -p）

    ensure_readable_schema

    source
    fn ensure_readable_schema(schema : DeltaSchema) -> Unit raise DeltaError

    判断 schema 是否能在当前读取路径上安全读取；不能则给出明确原因。

    has_checkpoint_file

    source
    fn has_checkpoint_file(root : String) -> Bool

    是否存在 checkpoint 文件（.checkpoint.parquet）

    is_writable_type

    source
    fn is_writable_type(ty : DeltaType) -> Bool

    判断某个类型是否能由写入路径产出

    json_as_array

    source
    fn json_as_array(j : Json) -> Array[Json]?

    json_as_bool

    source
    fn json_as_bool(j : Json) -> Bool?

    json_as_double

    source
    fn json_as_double(j : Json) -> Double?

    json_as_int

    source
    fn json_as_int(j : Json) -> Int?

    json_as_int64

    source
    fn json_as_int64(j : Json) -> Int64?

    json_as_object

    source
    fn json_as_object(j : Json) -> Map[String, Json]?

    JSON 访问小工具。核心 json 包以 prelude 形式导出类型与构造子， 这里统一用模式匹配读取，避免 deprecated accessor。

    json_as_string

    source
    fn json_as_string(j : Json) -> String?

    json_field

    source
    fn json_field(j : Json, key : String) -> Json?

    读取对象字段

    json_field_array

    source
    fn json_field_array(j : Json, key : String) -> Array[Json]?

    读取对象字段为数组

    json_field_bool

    source
    fn json_field_bool(j : Json, key : String) -> Bool?

    读取对象字段为 Bool

    json_field_int

    source
    fn json_field_int(j : Json, key : String) -> Int?

    读取对象字段为 Int

    json_field_int64

    source
    fn json_field_int64(j : Json, key : String) -> Int64?

    读取对象字段为 Int64

    json_field_string

    source
    fn json_field_string(j : Json, key : String) -> String?

    读取对象字段为字符串

    json_object_pairs_string

    source
    fn json_object_pairs_string(j : Json) -> Array[(String, String)]

    JSON 对象 → (key, value-string) 列表（用于 partitionValues / configuration）

    list_versions

    source
    fn list_versions(root : String) -> Array[Int] raise DeltaError

    列出所有日志版本（升序、去重）

    log_dir

    source
    fn log_dir(root : String) -> String

    _delta_log 目录路径

    parse_action

    source
    fn parse_action(line : StringView, version : Int) -> Action raise DeltaError

    解析一行 Delta 日志（一行一个 action）

    parse_add

    source
    fn parse_add(j : Json) -> AddFile raise DeltaError

    parse_commit_info

    source
    fn parse_commit_info(j : Json, version : Int) -> CommitInfo

    parse_metadata

    source
    fn parse_metadata(j : Json) -> TableInfo raise DeltaError

    parse_remove

    source
    fn parse_remove(j : Json) -> RemoveFile raise DeltaError

    parse_schema

    source
    fn parse_schema(raw : String) -> DeltaSchema raise DeltaError

    解析 Delta schemaString（形如 {"type":"struct","fields":[...]}）

    parse_schema_fields

    source
    fn parse_schema_fields(j : Json) -> Array[DeltaField] raise DeltaError

    解析 fields 数组

    read_actions

    source
    fn read_actions(root : String, version : Int) -> Array[Action] raise DeltaError

    读取某个版本的日志 actions

    read_snapshot

    source
    fn read_snapshot(root : String, snapshot : Snapshot, limit : Int?) -> ReadResult raise DeltaError

    读取 snapshot 中的数据文件。

    • root 是表目录（parquet 路径相对它）
    • limit 限制读入行数（None = 全部） 分区列不会出现在数据文件里，这里用 partitionValues 补齐常量列。

    read_text

    source
    fn read_text(path : String) -> String raise DeltaError

    读取文本文件

    scan

    source
    fn scan(snapshot : Snapshot, filter : ScanFilter) -> ScanReport

    文件级扫描规划：分区过滤 + min/max 统计剪枝。 不读取数据文件，因此可离线测试、也可用于"改动了多少行"这类审计场景。

    version_file_name

    source
    fn version_file_name(version : Int) -> String

    版本号 → 日志文件名（20 位零填充，Delta 规范）

    write_text

    source
    fn write_text(path : String, content : String) -> Unit raise DeltaError

    写入文本文件

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io