From 4545f43b2d424da31662d6a6846dbf0a8bf078c5 Mon Sep 17 00:00:00 2001 From: Helder Gregorio Date: Sun, 20 Sep 2026 14:57:55 +0000 Subject: [PATCH] Bend 2 PoC: IPC stream ordering laws Add dev/bend2/ipc_stream_ordering, a checked Bend 2 model of the message-ordering rules of the Arrow IPC streaming and file formats: schema first and once, dictionaries defined before a record batch uses them, delta batches appending to an existing dictionary, replacement batches accepted in streams and rejected in files, and nothing accepted after end-of-stream. main.bend is the validating reader (a state machine over abstract messages with a dictionary table of id -> segment count) plus a writer model and a runnable trace. LAWS.bend states 22 laws, each quoting the spec sentence or the ArrowReader/ArrowWriter behaviour it formalises, and keeps four more commented out as not expressible or not modelled (8-byte padding, int64 lengths, footer consistency, footer-order delta application). PROOF.bend proves all 22; `bend PROOF.bend` prints "All terms check." in 0.3 s. The README documents the model, every law, how to run, the JS and native builds, and seven mutation tests (undefined dictionary accepted, replacement allowed in files, messages accepted after EOS, delta with no base, delta ignored, schema omitted by the writer, second schema accepted), each of which the checker rejects. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01SGvC3jYdiD3415vL8FrKtx --- dev/bend2/ipc_stream_ordering/LAWS.bend | 323 ++++++++++ dev/bend2/ipc_stream_ordering/PROOF.bend | 717 +++++++++++++++++++++++ dev/bend2/ipc_stream_ordering/README.md | 266 +++++++++ dev/bend2/ipc_stream_ordering/main.bend | 407 +++++++++++++ 4 files changed, 1713 insertions(+) create mode 100644 dev/bend2/ipc_stream_ordering/LAWS.bend create mode 100644 dev/bend2/ipc_stream_ordering/PROOF.bend create mode 100644 dev/bend2/ipc_stream_ordering/README.md create mode 100644 dev/bend2/ipc_stream_ordering/main.bend diff --git a/dev/bend2/ipc_stream_ordering/LAWS.bend b/dev/bend2/ipc_stream_ordering/LAWS.bend new file mode 100644 index 0000000000..7578d4521f --- /dev/null +++ b/dev/bend2/ipc_stream_ordering/LAWS.bend @@ -0,0 +1,323 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# The laws of the IPC ordering model. The human states them; PROOF.bend +# must prove them. Each law quotes the sentence of the Arrow columnar +# format specification (docs/source/format/Columnar.rst, sections "IPC +# Streaming Format", "IPC File Format" and "Dictionary Messages") or the +# behaviour of this repository's ArrowReader / ArrowWriter that it +# formalises. Laws that cannot be proven or expressed are kept at the +# end, commented out, with the reason. + +import Base +import ./main.bend as R + +# the truth of a Bool as a type: Unit when True, Empty when False +def T(b: Bool) -> Data: + match b: + case True{}: + Unit + case False{}: + Empty + +# a reader that has read the schema, in any format, with any table +def body(fmt: R.Fmt, declared: List<&2, Nat>, dicts: List<&2, R.DEntry>) -> R.RState: + R.RState{fmt, R.PBody{}, declared, dicts} + +# 1. Schema first, schema once +# ---------------------------- + +# LAW: "The schema comes first in the stream." A stream whose first +# message is not a schema is rejected at that message. +# (ArrowStreamReader.readSchema: "Expected schema but header was ...") +law schema_first: + for +fmt : R.Fmt + for m : R.Msg + for e : {False{} == R.is_schema(m) : Bool} + {R.verdict(m, R.start(fmt)) == R.VReject{} : R.Verdict} + +# LAW (anti-vacuity): a schema as the first message is accepted, and the +# reader then knows exactly the ids the schema declared. +law schema_accepted: + for +fmt : R.Fmt + for +ids : List<&2, Nat> + {R.step_r(R.MSchema{ids}, R.start(fmt)) == (body(fmt, ids, Nil{}), R.VAccept{}) : R.RState & R.Verdict} + +# LAW: "it is the same for all of the record batches that follow": a +# second schema is rejected. +# (ArrowStreamReader.loadNextBatch: "Expected RecordBatch or +# DictionaryBatch but header was Schema") +law schema_once: + for +fmt : R.Fmt + for +ids : List<&2, Nat> + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + {R.verdict(R.MSchema{ids}, body(fmt, declared, dicts)) == R.VReject{} : R.Verdict} + +# 2. Dictionaries before use +# -------------------------- + +# LAW: "before any dictionary key is used in a RecordBatch it should be +# defined in a DictionaryBatch." A record batch that uses an id with no +# dictionary is rejected. +law batch_needs_dictionary: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for +refs : List<&2, Nat> + for +id : Nat + for w : T(R.has(id, refs)) + for e : {False{} == R.tbl_has(id, dicts) : Bool} + {R.verdict(R.MBatch{refs}, body(fmt, declared, dicts)) == R.VReject{} : R.Verdict} + +# LAW (anti-vacuity): a record batch whose dictionaries are all defined +# is accepted and leaves the reader unchanged. +law batch_accepted: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for +refs : List<&2, Nat> + for w : T(R.all_defined(refs, dicts)) + {R.step_r(R.MBatch{refs}, body(fmt, declared, dicts)) == (body(fmt, declared, dicts), R.VAccept{}) : R.RState & R.Verdict} + +# LAW: "An edge-case for interleaved dictionary and record batches occurs +# when the record batches contain dictionary encoded arrays that are +# completely null. In this case, the dictionary for the encoded column +# might appear after the first record batch." A batch that uses no +# dictionary key is accepted with an empty table. +# (ArrowStreamReader.checkDictionaries: "vector.getNullCount() < +# vector.getValueCount()") +law all_null_batch_needs_no_dictionary: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + {R.verdict(R.MBatch{Nil{}}, body(fmt, declared, Nil{})) == R.VAccept{} : R.Verdict} + +# 3. Delta dictionaries +# --------------------- + +# LAW: "The dictionary isDelta flag allows existing dictionaries to be +# expanded": a delta batch for an id with no dictionary is rejected. +# This is the spec's rule; ArrowReader.loadDictionary is more lenient and +# appends the delta to the empty vector it created from the schema. The +# model follows the spec. +law delta_needs_base: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for +id : Nat + for e : {False{} == R.tbl_has(id, dicts) : Bool} + {R.verdict(R.MDict{id, True{}}, body(fmt, declared, dicts)) == R.VReject{} : R.Verdict} + +# LAW: "A dictionary batch with isDelta set indicates that its vector +# should be concatenated with those of any previous batches with the same +# id." A delta on an existing, declared id is accepted in both formats +# and the dictionary gains one segment. +# (ArrowReader.loadDictionary: VectorBatchAppender.batchAppend) +law delta_appends: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for +id : Nat + for wd : T(R.has(id, declared)) + for wt : T(R.tbl_has(id, dicts)) + {R.verdict(R.MDict{id, True{}}, body(fmt, declared, dicts)) == R.VAccept{} : R.Verdict} + & {R.tbl_segs(id, R.dicts_of(R.next(R.MDict{id, True{}}, body(fmt, declared, dicts)))) == 1n+R.tbl_segs(id, dicts) : Nat} + +# 4. Replacement dictionaries +# --------------------------- + +# LAW: a first, non-delta dictionary batch for a declared id is accepted +# in both formats and defines a one-segment dictionary. +law dictionary_defined: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for +id : Nat + for wd : T(R.has(id, declared)) + for e : {False{} == R.tbl_has(id, dicts) : Bool} + {R.verdict(R.MDict{id, False{}}, body(fmt, declared, dicts)) == R.VAccept{} : R.Verdict} + & {R.tbl_segs(id, R.dicts_of(R.next(R.MDict{id, False{}}, body(fmt, declared, dicts)))) == 1n : Nat} + +# LAW: "if isDelta is set to false, then the dictionary replaces the +# existing dictionary for the same ID." In a stream the replacement is +# accepted and the dictionary is back to one segment. +law replacement_in_stream: + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for +id : Nat + for wd : T(R.has(id, declared)) + for wt : T(R.tbl_has(id, dicts)) + {R.verdict(R.MDict{id, False{}}, body(R.FStream{}, declared, dicts)) == R.VAccept{} : R.Verdict} + & {R.tbl_segs(id, R.dicts_of(R.next(R.MDict{id, False{}}, body(R.FStream{}, declared, dicts)))) == 1n : Nat} + +# LAW: "The IPC File format does not support dictionary replacement, +# i.e. only one non-delta dictionary batch can be emitted for a given +# dictionary ID." In a file the replacement is rejected and the table is +# unchanged. +# (ArrowFileWriter.ensureDictionariesWritten: "Replacement dictionaries +# are not supported in the IPC file format.") +law no_replacement_in_file: + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for +id : Nat + for wt : T(R.tbl_has(id, dicts)) + {R.step_r(R.MDict{id, False{}}, body(R.FFile{}, declared, dicts)) == (body(R.FFile{}, declared, dicts), R.VReject{}) : R.RState & R.Verdict} + +# LAW: "The dictionary types are found in the schema": a dictionary batch +# for an id the schema did not declare is rejected, delta or not. +# (ArrowReader.loadDictionary: "Dictionary ID ... not defined in schema") +law undeclared_dictionary_rejected: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for +id : Nat + for delta : Bool + for e : {False{} == R.has(id, declared) : Bool} + {R.verdict(R.MDict{id, delta}, body(fmt, declared, dicts)) == R.VReject{} : R.Verdict} + +# LAW (invariant): every dictionary in the table was declared by the +# schema. It holds at the start and every message keeps it, so it holds +# in every reachable state. +law declared_start: + for +fmt : R.Fmt + T(R.tbl_declared(R.dicts_of(R.start(fmt)), R.declared_of(R.start(fmt)))) + +law declared_kept: + for +fmt : R.Fmt + for phase : R.Phase + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for m : R.Msg + for w : T(R.tbl_declared(dicts, declared)) + T(R.tbl_declared(R.dicts_of(R.next(m, R.RState{fmt, phase, declared, dicts})), R.declared_of(R.next(m, R.RState{fmt, phase, declared, dicts})))) + +# 5. End of stream +# ---------------- + +# LAW: "The stream writer can signal end-of-stream (EOS)": EOS after the +# schema is accepted and ends the stream. +law eos_accepted: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + {R.step_r(R.MEos{}, body(fmt, declared, dicts)) == (R.RState{fmt, R.PEnd{}, declared, dicts}, R.VAccept{}) : R.RState & R.Verdict} + +# LAW: no message is accepted after EOS, whatever it is. +# (MessageChannelReader.readNext returns null at EOS and +# ArrowStreamReader.loadNextBatch then answers false forever) +law nothing_after_eos: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for m : R.Msg + {R.verdict(m, R.RState{fmt, R.PEnd{}, declared, dicts}) == R.VReject{} : R.Verdict} + +# LAW (trace form): every message of any stream that follows an EOS is +# rejected. +law eos_is_final: + for +fmt : R.Fmt + for +declared : List<&2, Nat> + for +dicts : List<&2, R.DEntry> + for +msgs : List<&2, R.Msg> + T(R.all_rejected(R.verdicts(msgs, R.RState{fmt, R.PEnd{}, declared, dicts}))) + +# 6. Anti-vacuity: the spec's own examples +# ---------------------------------------- + +# LAW: the delta example of "Dictionary Messages" (schema, dictionary 0, +# batch, dictionary 0 delta, batch, EOS) is accepted in full by both +# formats and leaves dictionary 0 with two segments. +law spec_delta_example: + for +fmt : R.Fmt + T(R.all_accepted(R.verdicts(R.delta_example(), R.start(fmt)))) + & {R.tbl_segs(0n, R.dicts_of(R.final(R.delta_example(), R.start(fmt)))) == 2n : Nat} + +# LAW: the replacement example of "Dictionary Messages" is accepted in +# full by a stream reader and leaves dictionary 0 with one segment. +law spec_replacement_example_stream: + T(R.all_accepted(R.verdicts(R.replacement_example(), R.start(R.FStream{})))) + & {R.tbl_segs(0n, R.dicts_of(R.final(R.replacement_example(), R.start(R.FStream{})))) == 1n : Nat} + +# LAW: the same replacement example is not accepted in full by a file +# reader: exactly the fourth message, the replacement, is rejected. +law spec_replacement_example_file: + {R.verdicts(R.replacement_example(), R.start(R.FFile{})) == [R.VAccept{}, R.VAccept{}, R.VAccept{}, R.VReject{}, R.VAccept{}, R.VAccept{}] : List<&2, R.Verdict>} + +# 7. Writer / reader round trip +# ----------------------------- + +# LAW: " ... +# ... ": what ArrowWriter emits for a schema +# with distinct dictionary ids (ArrowWriter.dictionaryIdsUsed is a Set) +# and any number of record batches is accepted in full by the reader of +# either format. +law writer_round_trip: + for +fmt : R.Fmt + for +ids : List<&2, Nat> + for +n : Nat + for w : T(R.nodup(ids)) + T(R.all_accepted(R.verdicts(R.write(ids, n), R.start(fmt)))) + +# LAW: after the round trip every declared dictionary is defined and the +# reader has seen EOS. +law writer_round_trip_state: + for +fmt : R.Fmt + for +ids : List<&2, Nat> + for +n : Nat + for w : T(R.nodup(ids)) + T(R.all_defined(ids, R.dicts_of(R.final(R.write(ids, n), R.start(fmt))))) + & {R.phase_of(R.final(R.write(ids, n), R.start(fmt))) == R.PEnd{} : R.Phase} + +# Not proven or not expressible +# ----------------------------- + +# NOT EXPRESSIBLE: "each message ... is padded to an 8-byte boundary" +# (Encapsulated message format). The model has no bytes: a message is +# an abstract constructor with no length, and Bend's Base ships no +# theory of Nat.mod, so even with lengths modelled as Nat the proof of +# `Nat.mod(padded(n), 8n) == 0n` would need a divisibility library +# written from scratch first. +# law metadata_padded_to_8: +# for +n : Nat +# {Nat.mod(R.padded(n), 8n) == 0n : Nat} + +# NOT EXPRESSIBLE: "bodyLength: long" (Message.fbs) and Block.offset / +# metaDataLength / bodyLength (File.fbs) are int64 / int32. Bend has no +# 64-bit integer type (only Nat, U32 and F32), so overflow and sign rules +# of the wire format cannot be stated; lengths would be unbounded Nats. +# law body_length_fits_int64: +# for m : R.Msg +# {Nat.is_lt(R.body_length(m), 9223372036854775808n) == True{} : Bool} + +# NOT MODELLED: "The metadata version, schema and custom metadata +# serialized in the IPC file footer MUST be identical to ... the embedded +# IPC stream" and "The dictionaries and record batches serialized in the +# IPC file footer SHOULD be listed in the same order as they appear in +# the embedded IPC stream". The footer is not part of this model: the +# reader here is the sequential stream reader that the file format +# embeds, and ArrowFileReader reads the footer instead of the EOS. A +# footer model would add a list of blocks per message kind and a law +# equating it to the filtered stream; it is left for a follow-up. +# law footer_matches_stream: +# for +ids : List<&2, Nat> +# for +n : Nat +# {R.footer_dicts(R.write_file(ids, n)) == ids : List<&2, Nat>} + +# NOT MODELLED: "Delta dictionary batches in an IPC File are applied in +# the order they appear in the file footer." Random access over the +# footer is not modelled; the reader here consumes messages in stream +# order, which by the SHOULD above is the footer order. diff --git a/dev/bend2/ipc_stream_ordering/PROOF.bend b/dev/bend2/ipc_stream_ordering/PROOF.bend new file mode 100644 index 0000000000..32a1d6ad25 --- /dev/null +++ b/dev/bend2/ipc_stream_ordering/PROOF.bend @@ -0,0 +1,717 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# The proofs. Imports the model (as R) and the claims (as Laws) and +# fills every law they state; `bend PROOF.bend` is the whole check. +# +# Two idioms recur. A computed Bool cannot be matched in place, so every +# case split is a `.fin` helper that takes the verdict `c` as a +# parameter, usually next to an equation `e : {c == }` +# or a truth `w : T(c)` that the match refines. And a rewrite `%e : P` +# with `e : {a == b}` replaces `b` by `a` in the goal, so lemmas put the +# simpler side on the left. + +import Base +import ./main.bend as R +import ./LAWS.bend as Laws + +# Proof kit +# --------- + +# T(True{}) from an equation +def true_T(b: Bool, e: {True{} == b : Bool}) -> Laws.T(b): + %e : Laws.T(_) + Unit{} + +# an equation from T(b) +def T_true(b: Bool, w: Laws.T(b)) -> {True{} == b : Bool}: + match b: + case True{}: + {==} + case False{}: + Empty.absurd({True{} == False{} : Bool}, w) + +def disc(b: Bool) -> Type: + match b: + case True{}: + Unit + case False{}: + Empty + +def ndisc(b: Bool) -> Type: + match b: + case True{}: + Empty + case False{}: + Unit + +# True and False clash +def true_ne_false(e: {True{} == False{} : Bool}) -> Empty: + %e : disc(_) + Unit{} + +def false_ne_true(e: {False{} == True{} : Bool}) -> Empty: + %e : ndisc(_) + Unit{} + +# T(c && b) splits into T(c) and T(b); c is a parameter so it can be matched +def split.fin(c: Bool, -b: Bool, w: Laws.T(c && b), -P: Type, k: Laws.T(c) -> Laws.T(b) -> P) -> P: + match c: + case True{}: + k(Unit{}, w) + case False{}: + Empty.absurd(P, w) + +# T(c) and T(b) join into T(c && b) +def join.fin(c: Bool, -b: Bool, wc: Laws.T(c), wb: Laws.T(b)) -> Laws.T(c && b): + match c: + case True{}: + wb + case False{}: + Empty.absurd(Laws.T(False{} && b), wc) + +# T(a) gives T(a || b) +def or_l(a: Bool, -b: Bool, wa: Laws.T(a)) -> Laws.T(a || b): + match a: + case True{}: + Unit{} + case False{}: + Empty.absurd(Laws.T(False{} || b), wa) + +# T(b) gives T(a || b) +def or_r(a: Bool, -b: Bool, wb: Laws.T(b)) -> Laws.T(a || b): + match a: + case True{}: + Unit{} + case False{}: + wb + +# (a || b) || c gives b || (a || c) +def or_rot(a: Bool, b: Bool, -c: Bool, w: Laws.T((a || b) || c)) -> Laws.T(b || (a || c)): + match a b: + case True{} True{}: + Unit{} + case True{} False{}: + Unit{} + case False{} True{}: + Unit{} + case False{} False{}: + w + +# a && False is False +def and_false(a: Bool) -> {False{} == (a && False{}) : Bool}: + match a: + case True{}: + {==} + case False{}: + {==} + +# T(not(a)) is a == False +def not_T(a: Bool, w: Laws.T(Bool.not(a))) -> {False{} == a : Bool}: + match a: + case True{}: + Empty.absurd({False{} == True{} : Bool}, w) + case False{}: + {==} + +# not(a || b) splits into not(a) and not(b) +def not_or_split(a: Bool, -b: Bool, w: Laws.T(Bool.not(a || b)), -P: Type, + k: Laws.T(Bool.not(a)) -> Laws.T(Bool.not(b)) -> P) -> P: + match a: + case True{}: + Empty.absurd(P, w) + case False{}: + k(Unit{}, w) + +# not(a) and not(b) join into not(a || b) +def not_or_join(a: Bool, -b: Bool, wa: Laws.T(Bool.not(a)), wb: Laws.T(Bool.not(b))) -> Laws.T(Bool.not(a || b)): + match a: + case True{}: + Empty.absurd(Laws.T(Bool.not(True{} || b)), wa) + case False{}: + wb + +# Lemmas on Nat +# ------------- + +# a nat equals itself +def eq_refl(n: Nat) -> {True{} == Nat.is_eq(n, n) : Bool}: + match n: + case 0n: + {==} + case 1n+p: + eq_refl(p) + +# a true Nat.is_eq is an equality, oriented to replace x by h +def eq_sound(x: Nat, h: Nat, e: {True{} == Nat.is_eq(x, h) : Bool}) -> {h == x : Nat}: + match x h: + case 0n 0n: + {==} + case 0n 1n+q: + Empty.absurd({1n+q == 0n : Nat}, false_ne_true(Equal.sym(Bool, True{}, False{}, e))) + case 1n+p 0n: + Empty.absurd({0n == 1n+p : Nat}, false_ne_true(Equal.sym(Bool, True{}, False{}, e))) + case 1n+p 1n+q: + %eq_sound(p, q, e) : {1n+q == 1n+_ : Nat} + {==} + +# Nat.is_eq is symmetric +def eq_comm(a: Nat, b: Nat) -> {Nat.is_eq(a, b) == Nat.is_eq(b, a) : Bool}: + match a b: + case 0n 0n: + {==} + case 0n 1n+q: + {==} + case 1n+p 0n: + {==} + case 1n+p 1n+q: + eq_comm(p, q) + +# not(x == id) gives not(id == x) +def not_eq_comm(x: Nat, id: Nat, w: Laws.T(Bool.not(Nat.is_eq(x, id)))) -> Laws.T(Bool.not(Nat.is_eq(id, x))): + %eq_comm(x, id) : Laws.T(Bool.not(_)) + w + +# 1. Schema first, schema once +# ---------------------------- + +def Laws.schema_first(fmt, m, e): + match m: + case R.MSchema{ids}: + Empty.absurd({R.verdict(R.MSchema{ids}, R.start(fmt)) == R.VReject{} : R.Verdict}, false_ne_true(e)) + case R.MDict{id, delta}: + {==} + case R.MBatch{refs}: + {==} + case R.MEos{}: + {==} + +def Laws.schema_accepted(fmt, ids): + {==} + +def Laws.schema_once(fmt, ids, declared, dicts): + {==} + +# 2. Dictionaries before use +# -------------------------- + +# a batch using an undefined id fails all_defined: the step case, over +# the verdict c of Nat.is_eq(x, id) for the head x of refs +def missing.fin(+id: Nat, +x: Nat, -t: List<&2, Nat>, +dicts: List<&2, R.DEntry>, + rec: Laws.T(R.has(id, t)) -> {False{} == R.all_defined(t, dicts) : Bool}, + c: Bool, ec: {c == Nat.is_eq(x, id) : Bool}, w: Laws.T(c || R.has(id, t)), + e: {False{} == R.tbl_has(id, dicts) : Bool}) + -> {False{} == (R.tbl_has(x, dicts) && R.all_defined(t, dicts)) : Bool}: + match c: + case True{}: + %eq_sound(x, id, ec) : {False{} == (R.tbl_has(_, dicts) && R.all_defined(t, dicts)) : Bool} + %e : {False{} == (_ && R.all_defined(t, dicts)) : Bool} + {==} + case False{}: + %rec(w) : {False{} == (R.tbl_has(x, dicts) && _) : Bool} + and_false(R.tbl_has(x, dicts)) + +def missing(+id: Nat, refs: List<&2, Nat>, +dicts: List<&2, R.DEntry>, w: Laws.T(R.has(id, refs)), + +e: {False{} == R.tbl_has(id, dicts) : Bool}) -> {False{} == R.all_defined(refs, dicts) : Bool}: + match refs: + case Nil{}: + Empty.absurd({False{} == R.all_defined(Nil{}, dicts) : Bool}, w) + case Con{+x, t}: + missing.fin(id, x, t, dicts, wt => missing(id, t, dicts, wt, e), Nat.is_eq(x, id), {==}, w, e) + +def Laws.batch_needs_dictionary(fmt, declared, dicts, refs, id, w, e): + %missing(id, refs, dicts, w, e) : {R.resp(R.step_batch(fmt, declared, dicts, _)) == R.VReject{} : R.Verdict} + {==} + +# the batch arm, over the verdict c of all_defined +def batch_acc.fin(-fmt: R.Fmt, -declared: List<&2, Nat>, -dicts: List<&2, R.DEntry>, c: Bool, w: Laws.T(c)) + -> {R.step_batch(fmt, declared, dicts, c) == (R.RState{fmt, R.PBody{}, declared, dicts}, R.VAccept{}) : R.RState & R.Verdict}: + match c: + case True{}: + {==} + case False{}: + Empty.absurd({R.step_batch(fmt, declared, dicts, False{}) == (R.RState{fmt, R.PBody{}, declared, dicts}, R.VAccept{}) : R.RState & R.Verdict}, w) + +def Laws.batch_accepted(fmt, declared, dicts, refs, w): + batch_acc.fin(fmt, declared, dicts, R.all_defined(refs, dicts), w) + +def Laws.all_null_batch_needs_no_dictionary(fmt, declared): + {==} + +# 3. Delta dictionaries +# --------------------- + +# a delta with no base is rejected whether or not the id is declared +def delta_base.fin(-fmt: R.Fmt, -declared: List<&2, Nat>, -dicts: List<&2, R.DEntry>, -id: Nat, c: Bool) + -> {R.resp(R.step_dict(fmt, declared, dicts, id, True{}, c, False{})) == R.VReject{} : R.Verdict}: + match c: + case True{}: + {==} + case False{}: + {==} + +def Laws.delta_needs_base(fmt, declared, dicts, id, e): + %e : {R.resp(R.step_dict(fmt, declared, dicts, id, True{}, R.has(id, declared), _)) == R.VReject{} : R.Verdict} + delta_base.fin(fmt, declared, dicts, id, R.has(id, declared)) + +# a delta adds one segment to the visible dictionary: the step case, +# over the verdict c of Nat.is_eq(eid, id) for the head entry +def segs_bump.fin(+id: Nat, +eid: Nat, +segs: Nat, +rest: List<&2, R.DEntry>, + rec: Laws.T(R.tbl_has(id, rest)) -> {R.tbl_segs(id, R.tbl_bump(id, rest)) == 1n+R.tbl_segs(id, rest) : Nat}, + c: Bool, e: {c == Nat.is_eq(eid, id) : Bool}, w: Laws.T(c || R.tbl_has(id, rest))) + -> {R.tbl_segs(id, R.bump_put(eid, segs, R.tbl_bump(id, rest), c)) == 1n+R.pick_nat(Nat.is_eq(eid, id), segs, R.tbl_segs(id, rest)) : Nat}: + match c: + case True{}: + %e : {R.pick_nat(_, 1n+segs, R.tbl_segs(id, R.tbl_bump(id, rest))) == 1n+R.pick_nat(_, segs, R.tbl_segs(id, rest)) : Nat} + {==} + case False{}: + %e : {R.pick_nat(_, segs, R.tbl_segs(id, R.tbl_bump(id, rest))) == 1n+R.pick_nat(_, segs, R.tbl_segs(id, rest)) : Nat} + rec(w) + +def segs_bump(+id: Nat, dicts: List<&2, R.DEntry>, w: Laws.T(R.tbl_has(id, dicts))) + -> {R.tbl_segs(id, R.tbl_bump(id, dicts)) == 1n+R.tbl_segs(id, dicts) : Nat}: + match dicts: + case Nil{}: + Empty.absurd({R.tbl_segs(id, R.tbl_bump(id, Nil{})) == 1n+R.tbl_segs(id, Nil{}) : Nat}, w) + case Con{R.DEntry{+eid, +segs}, +rest}: + segs_bump.fin(id, eid, segs, rest, wr => segs_bump(id, rest, wr), Nat.is_eq(eid, id), {==}, w) + +# the claim of delta_appends, over the two verdicts +def DeltaGoal(+fmt: R.Fmt, +declared: List<&2, Nat>, +dicts: List<&2, R.DEntry>, +id: Nat, +c1: Bool, +c2: Bool) -> Type: + {R.resp(R.step_dict(fmt, declared, dicts, id, True{}, c1, c2)) == R.VAccept{} : R.Verdict} + & {R.tbl_segs(id, R.dicts_of(R.state(R.step_dict(fmt, declared, dicts, id, True{}, c1, c2)))) == 1n+R.tbl_segs(id, dicts) : Nat} + +def delta_app.fin(-fmt: R.Fmt, -declared: List<&2, Nat>, +dicts: List<&2, R.DEntry>, +id: Nat, + c1: Bool, wd: Laws.T(c1), c2: Bool, e2: {c2 == R.tbl_has(id, dicts) : Bool}, wt: Laws.T(c2)) + -> DeltaGoal(fmt, declared, dicts, id, c1, c2): + match c1 c2: + case True{} True{}: + ({==}, segs_bump(id, dicts, true_T(R.tbl_has(id, dicts), e2))) + case True{} False{}: + Empty.absurd(DeltaGoal(fmt, declared, dicts, id, True{}, False{}), wt) + case False{} True{}: + Empty.absurd(DeltaGoal(fmt, declared, dicts, id, False{}, True{}), wd) + case False{} False{}: + Empty.absurd(DeltaGoal(fmt, declared, dicts, id, False{}, False{}), wd) + +def Laws.delta_appends(fmt, declared, dicts, id, wd, wt): + delta_app.fin(fmt, declared, dicts, id, R.has(id, declared), wd, R.tbl_has(id, dicts), {==}, wt) + +# 4. Replacement dictionaries +# --------------------------- + +# the claim of dictionary_defined and replacement_in_stream, over the +# two verdicts: accepted, and the id has one segment afterwards +def DefGoal(+fmt: R.Fmt, +declared: List<&2, Nat>, +dicts: List<&2, R.DEntry>, +id: Nat, +c1: Bool, +c2: Bool) -> Type: + {R.resp(R.step_dict(fmt, declared, dicts, id, False{}, c1, c2)) == R.VAccept{} : R.Verdict} + & {R.tbl_segs(id, R.dicts_of(R.state(R.step_dict(fmt, declared, dicts, id, False{}, c1, c2)))) == 1n : Nat} + +# a freshly defined id has one segment +def segs_define(+id: Nat, -dicts: List<&2, R.DEntry>) -> {R.tbl_segs(id, R.tbl_define(id, dicts)) == 1n : Nat}: + %eq_refl(id) : {R.pick_nat(_, 1n, R.tbl_segs(id, dicts)) == 1n : Nat} + {==} + +def defined.fin(-fmt: R.Fmt, -declared: List<&2, Nat>, -dicts: List<&2, R.DEntry>, +id: Nat, c1: Bool, wd: Laws.T(c1)) + -> DefGoal(fmt, declared, dicts, id, c1, False{}): + match c1: + case True{}: + ({==}, segs_define(id, dicts)) + case False{}: + Empty.absurd(DefGoal(fmt, declared, dicts, id, False{}, False{}), wd) + +def Laws.dictionary_defined(fmt, declared, dicts, id, wd, e): + %e : DefGoal(fmt, declared, dicts, id, R.has(id, declared), _) + defined.fin(fmt, declared, dicts, id, R.has(id, declared), wd) + +def replace.fin(-declared: List<&2, Nat>, -dicts: List<&2, R.DEntry>, +id: Nat, c1: Bool, wd: Laws.T(c1), c2: Bool, wt: Laws.T(c2)) + -> DefGoal(R.FStream{}, declared, dicts, id, c1, c2): + match c1 c2: + case True{} True{}: + ({==}, segs_define(id, dicts)) + case True{} False{}: + Empty.absurd(DefGoal(R.FStream{}, declared, dicts, id, True{}, False{}), wt) + case False{} True{}: + Empty.absurd(DefGoal(R.FStream{}, declared, dicts, id, False{}, True{}), wd) + case False{} False{}: + Empty.absurd(DefGoal(R.FStream{}, declared, dicts, id, False{}, False{}), wd) + +def Laws.replacement_in_stream(declared, dicts, id, wd, wt): + replace.fin(declared, dicts, id, R.has(id, declared), wd, R.tbl_has(id, dicts), wt) + +# the claim of no_replacement_in_file, over the two verdicts +def FileGoal(+declared: List<&2, Nat>, +dicts: List<&2, R.DEntry>, +id: Nat, +c1: Bool, +c2: Bool) -> Type: + {R.step_dict(R.FFile{}, declared, dicts, id, False{}, c1, c2) == (R.RState{R.FFile{}, R.PBody{}, declared, dicts}, R.VReject{}) : R.RState & R.Verdict} + +def no_replace.fin(-declared: List<&2, Nat>, -dicts: List<&2, R.DEntry>, -id: Nat, c1: Bool, c2: Bool, wt: Laws.T(c2)) + -> FileGoal(declared, dicts, id, c1, c2): + match c1 c2: + case True{} True{}: + {==} + case True{} False{}: + Empty.absurd(FileGoal(declared, dicts, id, True{}, False{}), wt) + case False{} True{}: + {==} + case False{} False{}: + Empty.absurd(FileGoal(declared, dicts, id, False{}, False{}), wt) + +def Laws.no_replacement_in_file(declared, dicts, id, wt): + no_replace.fin(declared, dicts, id, R.has(id, declared), R.tbl_has(id, dicts), wt) + +def Laws.undeclared_dictionary_rejected(fmt, declared, dicts, id, delta, e): + %e : {R.resp(R.step_dict(fmt, declared, dicts, id, delta, _, R.tbl_has(id, dicts))) == R.VReject{} : R.Verdict} + {==} + +# Invariant: the table only holds declared ids +# -------------------------------------------- + +def Laws.declared_start(fmt): + Unit{} + +# a bump keeps every entry's id: the step case, over the verdict c +def bump_decl.fin(+eid: Nat, -segs: Nat, -id: Nat, -rest: List<&2, R.DEntry>, +declared: List<&2, Nat>, c: Bool, + we: Laws.T(R.has(eid, declared)), wr: Laws.T(R.tbl_declared(R.tbl_bump(id, rest), declared))) + -> Laws.T(R.tbl_declared(R.bump_put(eid, segs, R.tbl_bump(id, rest), c), declared)): + match c: + case True{}: + join.fin(R.has(eid, declared), R.tbl_declared(R.tbl_bump(id, rest), declared), we, wr) + case False{}: + join.fin(R.has(eid, declared), R.tbl_declared(R.tbl_bump(id, rest), declared), we, wr) + +def bump_declared(+id: Nat, dicts: List<&2, R.DEntry>, +declared: List<&2, Nat>, w: Laws.T(R.tbl_declared(dicts, declared))) + -> Laws.T(R.tbl_declared(R.tbl_bump(id, dicts), declared)): + match dicts: + case Nil{}: + Unit{} + case Con{R.DEntry{+eid, segs}, +rest}: + split.fin(R.has(eid, declared), R.tbl_declared(rest, declared), w, + Laws.T(R.tbl_declared(R.tbl_bump(id, R.DEntry{eid, segs} <> rest), declared)), + we => wr => bump_decl.fin(eid, segs, id, rest, declared, Nat.is_eq(eid, id), we, bump_declared(id, rest, declared, wr))) + +# the claim of declared_kept after one step of step_dict +def KeptGoal(+fmt: R.Fmt, +declared: List<&2, Nat>, +dicts: List<&2, R.DEntry>, +id: Nat, +delta: Bool, +c1: Bool, +c2: Bool) -> Type: + Laws.T(R.tbl_declared(R.dicts_of(R.state(R.step_dict(fmt, declared, dicts, id, delta, c1, c2))), R.declared_of(R.state(R.step_dict(fmt, declared, dicts, id, delta, c1, c2))))) + +# a replacement: the stream adds a declared id, the file changes nothing +def kept_replace.fin(fmt: R.Fmt, +declared: List<&2, Nat>, +dicts: List<&2, R.DEntry>, +id: Nat, + e1: {True{} == R.has(id, declared) : Bool}, w: Laws.T(R.tbl_declared(dicts, declared))) + -> KeptGoal(fmt, declared, dicts, id, False{}, True{}, True{}): + match fmt: + case R.FStream{}: + join.fin(R.has(id, declared), R.tbl_declared(dicts, declared), true_T(R.has(id, declared), e1), w) + case R.FFile{}: + w + +# the dictionary arm, over the declared verdict c1, isDelta and the exists verdict c2 +def kept_dict.fin(fmt: R.Fmt, +declared: List<&2, Nat>, +dicts: List<&2, R.DEntry>, +id: Nat, + c1: Bool, e1: {c1 == R.has(id, declared) : Bool}, delta: Bool, c2: Bool, w: Laws.T(R.tbl_declared(dicts, declared))) + -> KeptGoal(fmt, declared, dicts, id, delta, c1, c2): + match c1: + case False{}: + w + case True{}: + match delta c2: + case True{} True{}: + bump_declared(id, dicts, declared, w) + case True{} False{}: + w + case False{} False{}: + join.fin(R.has(id, declared), R.tbl_declared(dicts, declared), true_T(R.has(id, declared), e1), w) + case False{} True{}: + kept_replace.fin(fmt, declared, dicts, id, e1, w) + +# the batch arm: the table is unchanged either way +def kept_batch.fin(-fmt: R.Fmt, -declared: List<&2, Nat>, -dicts: List<&2, R.DEntry>, c: Bool, w: Laws.T(R.tbl_declared(dicts, declared))) + -> Laws.T(R.tbl_declared(R.dicts_of(R.state(R.step_batch(fmt, declared, dicts, c))), R.declared_of(R.state(R.step_batch(fmt, declared, dicts, c))))): + match c: + case True{}: + w + case False{}: + w + +def Laws.declared_kept(fmt, phase, declared, dicts, m, w): + match phase: + case R.PStart{}: + match m: + case R.MSchema{ids}: + Unit{} + case R.MDict{id, delta}: + w + case R.MBatch{refs}: + w + case R.MEos{}: + w + case R.PBody{}: + match m: + case R.MSchema{ids}: + w + case R.MDict{+id, delta}: + kept_dict.fin(fmt, declared, dicts, id, R.has(id, declared), {==}, delta, R.tbl_has(id, dicts), w) + case R.MBatch{refs}: + kept_batch.fin(fmt, declared, dicts, R.all_defined(refs, dicts), w) + case R.MEos{}: + w + case R.PEnd{}: + w + +# 5. End of stream +# ---------------- + +def Laws.eos_accepted(fmt, declared, dicts): + {==} + +def Laws.nothing_after_eos(fmt, declared, dicts, m): + {==} + +def Laws.eos_is_final(fmt, declared, dicts, msgs): + match msgs: + case Nil{}: + Unit{} + case Con{m, rest}: + Laws.eos_is_final(fmt, declared, dicts, rest) + +# 6. Anti-vacuity: the spec's own examples +# ---------------------------------------- + +def Laws.spec_delta_example(fmt): + (Unit{}, {==}) + +def Laws.spec_replacement_example_stream(): + (Unit{}, {==}) + +def Laws.spec_replacement_example_file(): + {==} + +# 7. Writer / reader round trip +# ----------------------------- +# The writer emits one dictionary per id in `rem`, and the reader's table +# grows by one entry per message. The induction carries: every remaining +# id is declared (sub), every declared id is remaining or defined +# (covers), the remaining ids are distinct (nodup) and none of them is +# defined yet (disj). The last two make every dictionary a first +# definition, which is what a file reader needs. + +# every declared id is still to be written or already defined +def covers(ids: List<&2, Nat>, +rem: List<&2, Nat>, +dicts: List<&2, R.DEntry>) -> Bool: + match ids: + case Nil{}: + True{} + case Con{+x, t}: + (R.has(x, rem) || R.tbl_has(x, dicts)) && covers(t, rem, dicts) + +# no remaining id is defined +def disj(rem: List<&2, Nat>, +dicts: List<&2, R.DEntry>) -> Bool: + match rem: + case Nil{}: + True{} + case Con{+x, t}: + Bool.not(R.tbl_has(x, dicts)) && disj(t, dicts) + +# Lemmas on sub +def sub_cons(xs: List<&2, Nat>, +y: Nat, +ys: List<&2, Nat>, w: Laws.T(R.sub(xs, ys))) -> Laws.T(R.sub(xs, y <> ys)): + match xs: + case Nil{}: + Unit{} + case Con{+x, t}: + split.fin(R.has(x, ys), R.sub(t, ys), w, Laws.T(R.sub(x <> t, y <> ys)), + wx => wt => join.fin(R.has(x, y <> ys), R.sub(t, y <> ys), or_r(Nat.is_eq(y, x), R.has(x, ys), wx), sub_cons(t, y, ys, wt))) + +def sub_refl(xs: List<&2, Nat>) -> Laws.T(R.sub(xs, xs)): + match xs: + case Nil{}: + Unit{} + case Con{+x, +t}: + join.fin(R.has(x, x <> t), R.sub(t, x <> t), + or_l(Nat.is_eq(x, x), R.has(x, t), true_T(Nat.is_eq(x, x), eq_refl(x))), + sub_cons(t, x, t, sub_refl(t))) + +# Lemmas on covers +def covers_init(ids: List<&2, Nat>, +all: List<&2, Nat>, +dicts: List<&2, R.DEntry>, w: Laws.T(R.sub(ids, all))) -> Laws.T(covers(ids, all, dicts)): + match ids: + case Nil{}: + Unit{} + case Con{+x, t}: + split.fin(R.has(x, all), R.sub(t, all), w, Laws.T(covers(x <> t, all, dicts)), + wx => wt => join.fin(R.has(x, all) || R.tbl_has(x, dicts), covers(t, all, dicts), or_l(R.has(x, all), R.tbl_has(x, dicts), wx), covers_init(t, all, dicts, wt))) + +def covers_step(ids: List<&2, Nat>, +id: Nat, +t: List<&2, Nat>, +dicts: List<&2, R.DEntry>, w: Laws.T(covers(ids, id <> t, dicts))) + -> Laws.T(covers(ids, t, R.DEntry{id, 1n} <> dicts)): + match ids: + case Nil{}: + Unit{} + case Con{+x, rest}: + split.fin((Nat.is_eq(id, x) || R.has(x, t)) || R.tbl_has(x, dicts), covers(rest, id <> t, dicts), w, + Laws.T(covers(x <> rest, t, R.DEntry{id, 1n} <> dicts)), + wx => wr => join.fin(R.has(x, t) || (Nat.is_eq(id, x) || R.tbl_has(x, dicts)), covers(rest, t, R.DEntry{id, 1n} <> dicts), + or_rot(Nat.is_eq(id, x), R.has(x, t), R.tbl_has(x, dicts), wx), covers_step(rest, id, t, dicts, wr))) + +def covers_nil(ids: List<&2, Nat>, +dicts: List<&2, R.DEntry>, w: Laws.T(covers(ids, Nil{}, dicts))) -> Laws.T(R.all_defined(ids, dicts)): + match ids: + case Nil{}: + Unit{} + case Con{+x, t}: + split.fin(R.tbl_has(x, dicts), covers(t, Nil{}, dicts), w, Laws.T(R.all_defined(x <> t, dicts)), + wx => wt => join.fin(R.tbl_has(x, dicts), R.all_defined(t, dicts), wx, covers_nil(t, dicts, wt))) + +# Lemmas on disj +def disj_nil(rem: List<&2, Nat>) -> Laws.T(disj(rem, Nil{})): + match rem: + case Nil{}: + Unit{} + case Con{x, t}: + disj_nil(t) + +def disj_step(t: List<&2, Nat>, +id: Nat, +dicts: List<&2, R.DEntry>, wn: Laws.T(Bool.not(R.has(id, t))), wd: Laws.T(disj(t, dicts))) + -> Laws.T(disj(t, R.DEntry{id, 1n} <> dicts)): + match t: + case Nil{}: + Unit{} + case Con{+x, rest}: + not_or_split(Nat.is_eq(x, id), R.has(id, rest), wn, Laws.T(disj(x <> rest, R.DEntry{id, 1n} <> dicts)), + wnx => wnr => split.fin(Bool.not(R.tbl_has(x, dicts)), disj(rest, dicts), wd, Laws.T(disj(x <> rest, R.DEntry{id, 1n} <> dicts)), + wdx => wdr => join.fin(Bool.not(Nat.is_eq(id, x) || R.tbl_has(x, dicts)), disj(rest, R.DEntry{id, 1n} <> dicts), + not_or_join(Nat.is_eq(id, x), R.tbl_has(x, dicts), not_eq_comm(x, id, wnx), wdx), disj_step(rest, id, dicts, wnr, wdr)))) + +# The invariant and its step +def Inv(+ids: List<&2, Nat>, +rem: List<&2, Nat>, +dicts: List<&2, R.DEntry>) -> Type: + Laws.T(R.sub(rem, ids)) & Laws.T(covers(ids, rem, dicts)) & Laws.T(R.nodup(rem)) & Laws.T(disj(rem, dicts)) + +def InvStep(+ids: List<&2, Nat>, +id: Nat, +t: List<&2, Nat>, +dicts: List<&2, R.DEntry>) -> Type: + Laws.T(R.has(id, ids)) & Laws.T(Bool.not(R.tbl_has(id, dicts))) & Inv(ids, t, R.DEntry{id, 1n} <> dicts) + +def inv_step(+ids: List<&2, Nat>, +id: Nat, +t: List<&2, Nat>, +dicts: List<&2, R.DEntry>, inv: Inv(ids, id <> t, dicts)) -> InvStep(ids, id, t, dicts): + (wsub, wcov, wnd, wdis) = inv + split.fin(R.has(id, ids), R.sub(t, ids), wsub, InvStep(ids, id, t, dicts), + wh => wsub2 => split.fin(Bool.not(R.has(id, t)), R.nodup(t), wnd, InvStep(ids, id, t, dicts), + wnh => wnd2 => split.fin(Bool.not(R.tbl_has(id, dicts)), disj(t, dicts), wdis, InvStep(ids, id, t, dicts), + wnt => wdis2 => (wh, wnt, wsub2, covers_step(ids, id, t, dicts, wcov), wnd2, disj_step(t, id, dicts, wnh, wdis2))))) + +def inv_init(+ids: List<&2, Nat>, w: Laws.T(R.nodup(ids))) -> Inv(ids, ids, Nil{}): + (sub_refl(ids), covers_init(ids, ids, Nil{}, sub_refl(ids)), w, disj_nil(ids)) + +# The batches: accepted while every id is defined, and the state is unchanged +def batches_ok.fin(-fmt: R.Fmt, -ids: List<&2, Nat>, -dicts: List<&2, R.DEntry>, -p: Nat, c: Bool, w: Laws.T(c), + rec: Laws.T(R.all_accepted(R.verdicts(R.emit_batches(p, ids), R.RState{fmt, R.PBody{}, ids, dicts})))) + -> Laws.T(R.accepted(R.resp(R.step_batch(fmt, ids, dicts, c))) && R.all_accepted(R.verdicts(R.emit_batches(p, ids), R.state(R.step_batch(fmt, ids, dicts, c))))): + match c: + case True{}: + rec + case False{}: + Empty.absurd(Laws.T(R.accepted(R.resp(R.step_batch(fmt, ids, dicts, False{}))) && R.all_accepted(R.verdicts(R.emit_batches(p, ids), R.state(R.step_batch(fmt, ids, dicts, False{}))))), w) + +def batches_ok(n: Nat, +ids: List<&2, Nat>, +fmt: R.Fmt, +dicts: List<&2, R.DEntry>, +w: Laws.T(R.all_defined(ids, dicts))) + -> Laws.T(R.all_accepted(R.verdicts(R.emit_batches(n, ids), R.RState{fmt, R.PBody{}, ids, dicts}))): + match n: + case 0n: + Unit{} + case 1n+p: + batches_ok.fin(fmt, ids, dicts, p, R.all_defined(ids, dicts), w, batches_ok(p, ids, fmt, dicts, w)) + +# The dictionaries: each is a first definition of a declared id +def dicts_ok.fin(-fmt: R.Fmt, -ids: List<&2, Nat>, -dicts: List<&2, R.DEntry>, -id: Nat, -t: List<&2, Nat>, -n: Nat, + c1: Bool, wh: Laws.T(c1), c2: Bool, wn: Laws.T(Bool.not(c2)), + rec: Laws.T(R.all_accepted(R.verdicts(R.emit_dicts(t, ids, n), R.RState{fmt, R.PBody{}, ids, R.DEntry{id, 1n} <> dicts})))) + -> Laws.T(R.accepted(R.resp(R.step_dict(fmt, ids, dicts, id, False{}, c1, c2))) && R.all_accepted(R.verdicts(R.emit_dicts(t, ids, n), R.state(R.step_dict(fmt, ids, dicts, id, False{}, c1, c2))))): + match c1: + case False{}: + Empty.absurd(Laws.T(R.accepted(R.resp(R.step_dict(fmt, ids, dicts, id, False{}, False{}, c2))) && R.all_accepted(R.verdicts(R.emit_dicts(t, ids, n), R.state(R.step_dict(fmt, ids, dicts, id, False{}, False{}, c2))))), wh) + case True{}: + match c2: + case True{}: + Empty.absurd(Laws.T(R.accepted(R.resp(R.step_dict(fmt, ids, dicts, id, False{}, True{}, True{}))) && R.all_accepted(R.verdicts(R.emit_dicts(t, ids, n), R.state(R.step_dict(fmt, ids, dicts, id, False{}, True{}, True{}))))), wn) + case False{}: + rec + +# the step, with the stepped invariant as a parameter so it can be destructured +def dicts_ok.go(-fmt: R.Fmt, +ids: List<&2, Nat>, +dicts: List<&2, R.DEntry>, +id: Nat, -t: List<&2, Nat>, -n: Nat, + st: InvStep(ids, id, t, dicts), + rec: Inv(ids, t, R.DEntry{id, 1n} <> dicts) -> Laws.T(R.all_accepted(R.verdicts(R.emit_dicts(t, ids, n), R.RState{fmt, R.PBody{}, ids, R.DEntry{id, 1n} <> dicts})))) + -> Laws.T(R.accepted(R.resp(R.step_dict(fmt, ids, dicts, id, False{}, R.has(id, ids), R.tbl_has(id, dicts)))) && R.all_accepted(R.verdicts(R.emit_dicts(t, ids, n), R.state(R.step_dict(fmt, ids, dicts, id, False{}, R.has(id, ids), R.tbl_has(id, dicts)))))): + (wh, wnt, inv2) = st + dicts_ok.fin(fmt, ids, dicts, id, t, n, R.has(id, ids), wh, R.tbl_has(id, dicts), wnt, rec(inv2)) + +def dicts_ok(rem: List<&2, Nat>, +ids: List<&2, Nat>, +n: Nat, +fmt: R.Fmt, +dicts: List<&2, R.DEntry>, inv: Inv(ids, rem, dicts)) + -> Laws.T(R.all_accepted(R.verdicts(R.emit_dicts(rem, ids, n), R.RState{fmt, R.PBody{}, ids, dicts}))): + match rem: + case Nil{}: + (wsub, wcov, wnd, wdis) = inv + batches_ok(n, ids, fmt, dicts, covers_nil(ids, dicts, wcov)) + case Con{+id, +t}: + dicts_ok.go(fmt, ids, dicts, id, t, n, inv_step(ids, id, t, dicts, inv), inv2 => dicts_ok(t, ids, n, fmt, R.DEntry{id, 1n} <> dicts, inv2)) + +def Laws.writer_round_trip(fmt, ids, n, w): + dicts_ok(ids, ids, n, fmt, Nil{}, inv_init(ids, w)) + +# The final state: the batches leave the table alone and EOS ends the stream +def final_batches.fin(-fmt: R.Fmt, -ids: List<&2, Nat>, -dicts: List<&2, R.DEntry>, -p: Nat, c: Bool, w: Laws.T(c), + rec: {R.RState{fmt, R.PEnd{}, ids, dicts} == R.final(R.emit_batches(p, ids), R.RState{fmt, R.PBody{}, ids, dicts}) : R.RState}) + -> {R.RState{fmt, R.PEnd{}, ids, dicts} == R.final(R.emit_batches(p, ids), R.state(R.step_batch(fmt, ids, dicts, c))) : R.RState}: + match c: + case True{}: + rec + case False{}: + Empty.absurd({R.RState{fmt, R.PEnd{}, ids, dicts} == R.final(R.emit_batches(p, ids), R.state(R.step_batch(fmt, ids, dicts, False{}))) : R.RState}, w) + +def final_batches(n: Nat, +ids: List<&2, Nat>, +fmt: R.Fmt, +dicts: List<&2, R.DEntry>, +w: Laws.T(R.all_defined(ids, dicts))) + -> {R.RState{fmt, R.PEnd{}, ids, dicts} == R.final(R.emit_batches(n, ids), R.RState{fmt, R.PBody{}, ids, dicts}) : R.RState}: + match n: + case 0n: + {==} + case 1n+p: + final_batches.fin(fmt, ids, dicts, p, R.all_defined(ids, dicts), w, final_batches(p, ids, fmt, dicts, w)) + +# the claim of writer_round_trip_state for a reader state r +def FinalGoal(+ids: List<&2, Nat>, +r: R.RState) -> Type: + Laws.T(R.all_defined(ids, R.dicts_of(r))) & {R.phase_of(r) == R.PEnd{} : R.Phase} + +def final_ok.fin(-fmt: R.Fmt, -ids: List<&2, Nat>, -dicts: List<&2, R.DEntry>, -id: Nat, -t: List<&2, Nat>, -n: Nat, + c1: Bool, wh: Laws.T(c1), c2: Bool, wn: Laws.T(Bool.not(c2)), + rec: FinalGoal(ids, R.final(R.emit_dicts(t, ids, n), R.RState{fmt, R.PBody{}, ids, R.DEntry{id, 1n} <> dicts}))) + -> FinalGoal(ids, R.final(R.emit_dicts(t, ids, n), R.state(R.step_dict(fmt, ids, dicts, id, False{}, c1, c2)))): + match c1: + case False{}: + Empty.absurd(FinalGoal(ids, R.final(R.emit_dicts(t, ids, n), R.state(R.step_dict(fmt, ids, dicts, id, False{}, False{}, c2)))), wh) + case True{}: + match c2: + case True{}: + Empty.absurd(FinalGoal(ids, R.final(R.emit_dicts(t, ids, n), R.state(R.step_dict(fmt, ids, dicts, id, False{}, True{}, True{})))), wn) + case False{}: + rec + +def final_ok.go(-fmt: R.Fmt, +ids: List<&2, Nat>, +dicts: List<&2, R.DEntry>, +id: Nat, -t: List<&2, Nat>, -n: Nat, + st: InvStep(ids, id, t, dicts), + rec: Inv(ids, t, R.DEntry{id, 1n} <> dicts) -> FinalGoal(ids, R.final(R.emit_dicts(t, ids, n), R.RState{fmt, R.PBody{}, ids, R.DEntry{id, 1n} <> dicts}))) + -> FinalGoal(ids, R.final(R.emit_dicts(t, ids, n), R.state(R.step_dict(fmt, ids, dicts, id, False{}, R.has(id, ids), R.tbl_has(id, dicts))))): + (wh, wnt, inv2) = st + final_ok.fin(fmt, ids, dicts, id, t, n, R.has(id, ids), wh, R.tbl_has(id, dicts), wnt, rec(inv2)) + +# the Nil case: the batches leave the table, EOS ends the stream +def final_ok.nil(+fmt: R.Fmt, +ids: List<&2, Nat>, +n: Nat, +dicts: List<&2, R.DEntry>, +wall: Laws.T(R.all_defined(ids, dicts))) + -> FinalGoal(ids, R.final(R.emit_batches(n, ids), R.RState{fmt, R.PBody{}, ids, dicts})): + %final_batches(n, ids, fmt, dicts, wall) : FinalGoal(ids, _) + (wall, {==}) + +def final_ok(rem: List<&2, Nat>, +ids: List<&2, Nat>, +n: Nat, +fmt: R.Fmt, +dicts: List<&2, R.DEntry>, inv: Inv(ids, rem, dicts)) + -> FinalGoal(ids, R.final(R.emit_dicts(rem, ids, n), R.RState{fmt, R.PBody{}, ids, dicts})): + match rem: + case Nil{}: + (wsub, wcov, wnd, wdis) = inv + final_ok.nil(fmt, ids, n, dicts, covers_nil(ids, dicts, wcov)) + case Con{+id, +t}: + final_ok.go(fmt, ids, dicts, id, t, n, inv_step(ids, id, t, dicts, inv), inv2 => final_ok(t, ids, n, fmt, R.DEntry{id, 1n} <> dicts, inv2)) + +def Laws.writer_round_trip_state(fmt, ids, n, w): + final_ok(ids, ids, n, fmt, Nil{}, inv_init(ids, w)) diff --git a/dev/bend2/ipc_stream_ordering/README.md b/dev/bend2/ipc_stream_ordering/README.md new file mode 100644 index 0000000000..fd21328535 --- /dev/null +++ b/dev/bend2/ipc_stream_ordering/README.md @@ -0,0 +1,266 @@ + + +# Bend 2 proof of concept: IPC stream ordering laws + +A checked Bend 2 model of the message-ordering rules of the Arrow IPC +streaming and file formats: schema first, dictionaries before use, +delta and replacement dictionaries, end of stream. The rules come from +the columnar format specification (`docs/source/format/Columnar.rst`, +sections "IPC Streaming Format", "IPC File Format" and "Dictionary +Messages") and from how this repository's `ArrowStreamReader`, +`ArrowReader`, `ArrowFileWriter` and `ArrowStreamWriter` enforce them. + +Written against Bend 2.0.21, commit `6018e28` of +[bendlang/bend](https://github.com/bendlang/bend), on 2026-09-20. +The research notes and the first model (the Flight SQL +prepared-statement lifecycle) are on a sibling branch under +`dev/bend2/`; this directory is self-contained and copies the proof +helpers it needs. + +## 1. What is modelled + +`main.bend` models a validating reader as a state machine over a list +of abstract messages. Bytes, flatbuffers, alignment, the file magic and +the file footer are not modelled. + +| Concept | Model | +| --- | --- | +| Format | `Fmt`: `FStream{}` or `FFile{}`. They differ only in dictionary replacement. | +| Message | `Msg`: `MSchema{ids}` (the dictionary ids the schema's fields declare), `MDict{id, delta}` (`DictionaryBatch.id`, `isDelta`), `MBatch{refs}` (the dictionary ids a record batch actually uses), `MEos{}`. | +| Reader | `RState{fmt, phase, declared, dicts}`: the format, a phase (`PStart`, `PBody`, `PEnd`), the ids the schema declared, and the dictionary table. | +| Dictionary table | `List`, an association list. A non-delta batch pushes a fresh one-segment entry in front (so a replacement shadows the old dictionary); a delta bumps the segment count of the visible entry. `tbl_has` and `tbl_segs` read the table. | +| Verdict | `VAccept{}` or `VReject{}` per message. | +| Trace | `verdicts(msgs, r)` gives one verdict per message; `final(msgs, r)` the reader afterwards. | +| Writer | `write(ids, n)`: schema, one non-delta dictionary per id, `n` record batches using every id, EOS. This is the sequence `ArrowWriter.writeBatch` / `end` produces for one `VectorSchemaRoot` in either format. | + +The reader's rules, as `step` implements them: + +- `PStart`: a schema is accepted and moves to `PBody`; anything else is + rejected. +- `PBody`, schema: rejected. Dictionary batch: rejected unless the id + was declared; a delta is accepted only if a dictionary for the id + exists and then adds a segment; a non-delta on a new id defines it; + a non-delta on an existing id is a replacement, accepted in a stream + and rejected in a file. Record batch: accepted iff every id it uses + has a dictionary. EOS: accepted, moves to `PEnd`. +- `PEnd`: everything is rejected. + +Two modelling choices worth stating: + +- **Record batch references.** `MBatch{refs}` carries the ids whose + keys the batch actually uses. A dictionary-encoded column that is + entirely null uses no key, so `refs` omits it. This is the spec's + all-null edge case and matches `ArrowStreamReader.checkDictionaries`, + which only complains when `nullCount < valueCount`. +- **Delta with no base.** The spec says `isDelta` "allows existing + dictionaries to be expanded" and that a delta "should be concatenated + with those of any previous batches with the same id". The model + rejects a delta for an id with no dictionary. `ArrowReader.loadDictionary` + is more lenient: it pre-creates an empty vector per declared id in + `initialize()` and appends the delta to it, so Java accepts this case. + The law records the choice; mutation M4 below shows that switching to + Java's rule is caught. + +One observation about the Java reader fell out of writing the model. +`ArrowStreamReader.checkDictionaries` tests +`!dictionaries.containsKey(encoding.getId())`, but `dictionaries` is +populated for every declared id in `ArrowReader.initialize()` before +any batch is read, so the test is always false and a record batch whose +dictionary was never sent is loaded against an empty dictionary rather +than rejected. The model follows the spec here (law +`batch_needs_dictionary`). No Java code is changed in this PR. + +## 2. The laws + +`LAWS.bend` states 22 laws; `PROOF.bend` proves all of them. Each law +quotes its source. `T(b)` is the truth of a Bool as a type. + +### Schema first, schema once + +| Law | Spec / Java sentence | Claim | +| --- | --- | --- | +| `schema_first` | "The schema comes first in the stream." (`readSchema`: "Expected schema but header was ...") | Any non-schema first message is rejected. | +| `schema_accepted` | anti-vacuity | A first schema is accepted and the reader records exactly its ids. | +| `schema_once` | "it is the same for all of the record batches that follow" | A second schema is rejected. | + +### Dictionaries before use + +| Law | Spec / Java sentence | Claim | +| --- | --- | --- | +| `batch_needs_dictionary` | "before any dictionary key is used in a RecordBatch it should be defined in a DictionaryBatch" | A batch using an id with no dictionary is rejected. | +| `batch_accepted` | anti-vacuity | A batch whose ids are all defined is accepted and leaves the reader unchanged. | +| `all_null_batch_needs_no_dictionary` | the all-null edge-case note; `checkDictionaries` | A batch that uses no key is accepted with an empty table. | + +### Delta dictionaries + +| Law | Spec / Java sentence | Claim | +| --- | --- | --- | +| `delta_needs_base` | "allows existing dictionaries to be expanded" | A delta for an id with no dictionary is rejected (spec's rule, see above). | +| `delta_appends` | "its vector should be concatenated with those of any previous batches with the same id" (`VectorBatchAppender.batchAppend`) | A delta on a declared, existing id is accepted in both formats and the segment count grows by one. | + +### Replacement dictionaries + +| Law | Spec / Java sentence | Claim | +| --- | --- | --- | +| `dictionary_defined` | first `` | A first non-delta batch for a declared id is accepted in both formats and yields one segment. | +| `replacement_in_stream` | "if isDelta is set to false, then the dictionary replaces the existing dictionary for the same ID" | In a stream the replacement is accepted and the id is back to one segment. | +| `no_replacement_in_file` | "The IPC File format does not support dictionary replacement" (`ArrowFileWriter`: "Replacement dictionaries are not supported") | In a file the replacement is rejected and the reader is unchanged. | +| `undeclared_dictionary_rejected` | "The dictionary types are found in the schema" (`loadDictionary`: "Dictionary ID ... not defined in schema") | A dictionary batch for an undeclared id is rejected, delta or not. | +| `declared_start`, `declared_kept` | invariant | Every dictionary in the table was declared by the schema, at the start and after every message. | + +### End of stream + +| Law | Spec / Java sentence | Claim | +| --- | --- | --- | +| `eos_accepted` | "The stream writer can signal end-of-stream (EOS)" | EOS after the schema is accepted and ends the stream. | +| `nothing_after_eos` | `MessageChannelReader.readNext` returns null at EOS | No message is accepted after EOS. | +| `eos_is_final` | trace form of the above | Every message of any list that follows EOS is rejected. | + +### Anti-vacuity: the spec's own examples + +| Law | Claim | +| --- | --- | +| `spec_delta_example` | The delta example (schema, dict 0, batch, dict 0 delta, batch, EOS) is accepted in full by both formats and leaves dictionary 0 with two segments. | +| `spec_replacement_example_stream` | The replacement example is accepted in full by a stream reader, one segment at the end. | +| `spec_replacement_example_file` | A file reader answers exactly `accept accept accept reject accept accept` on it: only the replacement is refused. | + +### Writer / reader round trip + +| Law | Claim | +| --- | --- | +| `writer_round_trip` | For any format, any list of distinct dictionary ids (`ArrowWriter.dictionaryIdsUsed` is a `Set`) and any number of batches, every message the writer emits is accepted. | +| `writer_round_trip_state` | Afterwards every declared dictionary is defined and the reader is in `PEnd`. | + +The round trip is the one real induction. The invariant carried over +the writer's remaining ids is: every remaining id is declared, every +declared id is either remaining or already defined, the remaining ids +are distinct, and none of them is defined yet. The last two make every +dictionary a first definition, which is what the file reader needs. + +### Not proven or not expressible + +These stay in `LAWS.bend`, commented out, with the reason above each: + +| Law | Status | Why | +| --- | --- | --- | +| `metadata_padded_to_8` | NOT EXPRESSIBLE | The model has no bytes or lengths, and Bend's Base has no `Nat.mod` theory; the divisibility library would have to be written first. | +| `body_length_fits_int64` | NOT EXPRESSIBLE | Bend has only `Nat`, `U32` and `F32`; `bodyLength: long` and the footer's `int64` offsets cannot be typed, so overflow and sign rules cannot be stated. | +| `footer_matches_stream` | NOT MODELLED | The footer is outside the model; the reader is the sequential stream reader that a file embeds. A footer model (a block list per message kind, equated with the filtered stream) is a natural follow-up. | +| footer-order application of deltas | NOT MODELLED | Random access over the footer is not modelled; the sequential reader applies deltas in stream order, which the spec's SHOULD makes the footer order. | + +## 3. How to run + +The installer host was blocked from this environment, so the checker +runs from a clone with bun: + +```sh +git clone --depth 1 https://github.com/bendlang/bend.git /tmp/bend +BEND="bun /tmp/bend/bend2/main.ts" +cd dev/bend2/ipc_stream_ordering +$BEND PROOF.bend # All terms check. +$BEND main.bend # runs the six sample traces +$BEND main.bend -o out.js # JavaScript; node out.js +$BEND main.bend -o out # native, via clang +``` + +`main` prints one line per trace: the format, the verdict per message, +and `id:segments` for every declared id in the final table: + +``` +delta stream | accept accept accept accept accept accept | table 0:2 +delta file | accept accept accept accept accept accept | table 0:2 +replacement stream | accept accept accept accept accept accept | table 0:1 +replacement file | accept accept accept reject accept accept | table 0:1 +bad stream | accept accept reject accept accept accept reject | table 0:1 1:1 +writer file | accept accept accept accept accept accept | table 3:1 7:1 +``` + +The `bad` trace uses dictionary 1 before defining it (third message +rejected) and sends a batch after EOS (last message rejected). + +## 4. Results + +| Item | Lines | +| --- | ---: | +| Model (`main.bend`) | 407 | +| Laws (`LAWS.bend`, 22 laws + 4 commented) | 323 | +| Proofs (`PROOF.bend`) | 717 | + +`bend PROOF.bend` prints `All terms check.` in 0.29 s wall time. The JS +build is 26 KB and the native binary 1.1 MB; both print the six traces +above. A deliberately false law added to a copy (`MEos` accepted as the +first message) was rejected with the expected/observed verdicts, so the +gate is live. + +### Mutation tests + +Each bug was introduced into a copy of `main.bend` alone, `bend +PROOF.bend` was run, and the copy discarded. The committed model is +unchanged. + +| # | Mutation in `main.bend` | Real-world bug | Checker result | +| --- | --- | --- | --- | +| M1 | `step_batch` accepts when `all_defined` is false | reader loads a batch whose dictionary was never sent | rejected at `Laws.batch_needs_dictionary`: expected `VAccept`, observed `VReject` | +| M2 | `step_replace` accepts in `FFile` | file reader allows dictionary replacement | rejected at `no_replace.fin`: state `(RState{FFile, PBody, declared, DEntry{id,1n} <> dicts}, VAccept)` vs `VReject` | +| M3 | `step` in `PEnd` delegates to `step_body` | reader keeps accepting after EOS | rejected at `Laws.declared_kept` (first failure reported), `nothing_after_eos` and `eos_is_final` also fail | +| M4 | delta with no base defines a dictionary | Java's lenient `loadDictionary` rule | rejected at `delta_base.fin`: expected `VAccept`, observed `VReject` | +| M5 | `bump_put` does not increment `segs` | delta batch ignored | rejected at `segs_bump.fin`: `pick_nat(.., segs, ..)` vs `pick_nat(.., 1n+segs, ..)` | +| M6 | `write` omits the schema | writer forgets the schema message | rejected at `Laws.writer_round_trip`: reader still in `PStart` with `[]` declared | +| M7 | `step_body` accepts a second schema and adopts its ids | reader lets the schema change mid-stream | rejected at `Laws.schema_once` | + +M3 shows the brittleness noted in the research notes: the mutation +breaks three laws, but the first error the checker prints is at the +invariant proof, whose `PEnd` arm relied on the state being unchanged. +Either way the build is blocked. + +### What it cost + +Roughly 1.8 lines of proof per line of model. About a third of +`PROOF.bend` is the generic kit (splitting and joining `T(a && b)`, +`T(a || b)` introductions, `Nat.is_eq` reflexivity, soundness and +symmetry) and the `.fin` helpers that let a computed Bool be matched. +The single-step laws (sections 1 to 5) are each a few lines once the kit +exists; the round trip took the invariant above and about 150 lines. + +Two Bend restrictions shaped the proofs: a computed value cannot be +matched or destructured in place (hence every `.fin` and `.go` helper +takes the verdict or the tuple as a parameter), and a `match` on a +parameter must respect binder order (the replacement arm of +`declared_kept` had to move into its own def to match on `fmt` after +`c1`). Type-returning goal defs also count usages, so their parameters +are marked `+`. + +## 5. Sources + +- Arrow columnar format, IPC and dictionary sections: + https://github.com/apache/arrow/blob/main/docs/source/format/Columnar.rst +- `arrow-format/Message.fbs` (`DictionaryBatch.isDelta`), + `arrow-format/Schema.fbs` (`DictionaryEncoding.id`), + `arrow-format/File.fbs` (`Footer.dictionaries`, `recordBatches`) +- `vector/src/main/java/org/apache/arrow/vector/ipc/ArrowStreamReader.java` + (`readSchema`, `loadNextBatch`, `checkDictionaries`), + `ArrowReader.java` (`initialize`, `loadDictionary`), + `ArrowWriter.java`, `ArrowStreamWriter.java` + (`ensureDictionariesWritten` rewrites changed dictionaries), + `ArrowFileWriter.java` (dictionaries written once, no replacement), + `ArrowFileReader.java`, `message/MessageChannelReader.java` +- Bend 2 repository, `guide/GUIDE.md`, `bend2/base.bend`, demos + `proof_insertion_sort` and `app_win_is_bug_2d`: + https://github.com/bendlang/bend diff --git a/dev/bend2/ipc_stream_ordering/main.bend b/dev/bend2/ipc_stream_ordering/main.bend new file mode 100644 index 0000000000..2da1dbbab9 --- /dev/null +++ b/dev/bend2/ipc_stream_ordering/main.bend @@ -0,0 +1,407 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# A model of the message-ordering rules of the Arrow IPC streaming and +# file formats, written in Bend 2. Messages are abstract: a schema that +# declares which dictionary ids exist, dictionary batches (id, isDelta), +# record batches (the dictionary ids they actually reference) and +# end-of-stream. Bytes, flatbuffers, alignment and the file footer are +# not modelled. +# +# The reader is a state machine: one message in, one verdict out, and a +# dictionary table (id -> number of segments) that accumulates. The +# table is an association list where a new definition is pushed in front +# of the old one, so a replacement shadows what it replaces and a delta +# bumps the segment count of the visible entry. +# +# bend main.bend # checks the model and runs the sample traces in main +# bend PROOF.bend # checks LAWS.bend against this model + +import Base + +# Protocol +# -------- + +# the two IPC containers; they differ only in dictionary replacement +type Fmt is Data: + FStream{} + FFile{} + +# one encapsulated message, reduced to what the ordering rules look at +type Msg is Data: + MSchema{ids: List<&2, Nat>} # Schema: the dictionary ids its fields declare + MDict{id: Nat, delta: Bool} # DictionaryBatch{id, isDelta} + MBatch{refs: List<&2, Nat>} # RecordBatch: the dictionary ids it uses + MEos{} # 0xFFFFFFFF 0x00000000 + +# the reader's answer to one message +type Verdict is Data: + VAccept{} + VReject{} + +# where the reader is in the stream +type Phase is Data: + PStart{} # nothing read yet: only a schema is legal + PBody{} # schema read: dictionaries, batches and EOS are legal + PEnd{} # EOS read: nothing is legal + +# one visible dictionary: its id and how many batches make it up +type DEntry is Data: + DEntry{id: Nat, segs: Nat} + +# reader state: format, phase, ids declared by the schema, dictionary table +type RState is Data: + RState{fmt: Fmt, phase: Phase, declared: List<&2, Nat>, dicts: List<&2, DEntry>} + +# Lists of ids +# ------------ + +# whether h occurs in xs +def has(+h: Nat, xs: List<&2, Nat>) -> Bool: + match xs: + case Nil{}: + False{} + case Con{+x, t}: + Nat.is_eq(x, h) || has(h, t) + +# whether every x in xs occurs in ys +def sub(xs: List<&2, Nat>, +ys: List<&2, Nat>) -> Bool: + match xs: + case Nil{}: + True{} + case Con{+x, t}: + has(x, ys) && sub(t, ys) + +# whether no id occurs twice +def nodup(xs: List<&2, Nat>) -> Bool: + match xs: + case Nil{}: + True{} + case Con{+x, +t}: + Bool.not(has(x, t)) && nodup(t) + +# Dictionary table +# ---------------- + +# whether id has a dictionary +def tbl_has(+id: Nat, t: List<&2, DEntry>) -> Bool: + match t: + case Nil{}: + False{} + case Con{DEntry{+eid, segs}, rest}: + Nat.is_eq(eid, id) || tbl_has(id, rest) + +def pick_nat(c: Bool, a: Nat, b: Nat) -> Nat: + match c: + case True{}: + a + case False{}: + b + +# how many segments the visible dictionary for id has; 0 when none +def tbl_segs(+id: Nat, t: List<&2, DEntry>) -> Nat: + match t: + case Nil{}: + 0n + case Con{DEntry{+eid, segs}, rest}: + pick_nat(Nat.is_eq(eid, id), segs, tbl_segs(id, rest)) + +# the entry, with one more segment when hit +def bump_put(eid: Nat, segs: Nat, rest: List<&2, DEntry>, hit: Bool) -> List<&2, DEntry>: + match hit: + case True{}: + DEntry{eid, 1n+segs} <> rest + case False{}: + DEntry{eid, segs} <> rest + +# a delta batch for id: one more segment on every entry for id +def tbl_bump(+id: Nat, t: List<&2, DEntry>) -> List<&2, DEntry>: + match t: + case Nil{}: + Nil{} + case Con{DEntry{+eid, segs}, rest}: + bump_put(eid, segs, tbl_bump(id, rest), Nat.is_eq(eid, id)) + +# a non-delta batch for id: a fresh one-segment dictionary, shadowing any older one +def tbl_define(id: Nat, t: List<&2, DEntry>) -> List<&2, DEntry>: + DEntry{id, 1n} <> t + +# whether every referenced id has a dictionary +def all_defined(refs: List<&2, Nat>, +t: List<&2, DEntry>) -> Bool: + match refs: + case Nil{}: + True{} + case Con{+x, rest}: + tbl_has(x, t) && all_defined(rest, t) + +# whether every dictionary in the table was declared by the schema +def tbl_declared(t: List<&2, DEntry>, +declared: List<&2, Nat>) -> Bool: + match t: + case Nil{}: + True{} + case Con{DEntry{+eid, segs}, rest}: + has(eid, declared) && tbl_declared(rest, declared) + +# Transitions +# ----------- + +# before the schema: only a schema is accepted +def step_start(fmt: Fmt, declared: List<&2, Nat>, dicts: List<&2, DEntry>, m: Msg) -> RState & Verdict: + match m: + case MSchema{ids}: + (RState{fmt, PBody{}, ids, Nil{}}, VAccept{}) + case MDict{id, delta}: + (RState{fmt, PStart{}, declared, dicts}, VReject{}) + case MBatch{refs}: + (RState{fmt, PStart{}, declared, dicts}, VReject{}) + case MEos{}: + (RState{fmt, PStart{}, declared, dicts}, VReject{}) + +# a non-delta batch for an id that already has a dictionary: a +# replacement, legal in a stream and illegal in a file +def step_replace(fmt: Fmt, declared: List<&2, Nat>, dicts: List<&2, DEntry>, id: Nat) -> RState & Verdict: + match fmt: + case FStream{}: + (RState{FStream{}, PBody{}, declared, tbl_define(id, dicts)}, VAccept{}) + case FFile{}: + (RState{FFile{}, PBody{}, declared, dicts}, VReject{}) + +# a dictionary batch for a declared id, by isDelta and by whether a +# dictionary for the id exists +def step_dict_ok(fmt: Fmt, declared: List<&2, Nat>, dicts: List<&2, DEntry>, id: Nat, delta: Bool, exists: Bool) -> RState & Verdict: + match delta exists: + case True{} True{}: + (RState{fmt, PBody{}, declared, tbl_bump(id, dicts)}, VAccept{}) + case True{} False{}: + (RState{fmt, PBody{}, declared, dicts}, VReject{}) + case False{} False{}: + (RState{fmt, PBody{}, declared, tbl_define(id, dicts)}, VAccept{}) + case False{} True{}: + step_replace(fmt, declared, dicts, id) + +# a dictionary batch: rejected unless the schema declared the id +def step_dict(fmt: Fmt, declared: List<&2, Nat>, dicts: List<&2, DEntry>, id: Nat, delta: Bool, in_schema: Bool, exists: Bool) -> RState & Verdict: + match in_schema: + case True{}: + step_dict_ok(fmt, declared, dicts, id, delta, exists) + case False{}: + (RState{fmt, PBody{}, declared, dicts}, VReject{}) + +# a record batch: accepted when every dictionary it uses is defined +def step_batch(fmt: Fmt, declared: List<&2, Nat>, dicts: List<&2, DEntry>, ok: Bool) -> RState & Verdict: + match ok: + case True{}: + (RState{fmt, PBody{}, declared, dicts}, VAccept{}) + case False{}: + (RState{fmt, PBody{}, declared, dicts}, VReject{}) + +# after the schema +def step_body(fmt: Fmt, +declared: List<&2, Nat>, +dicts: List<&2, DEntry>, m: Msg) -> RState & Verdict: + match m: + case MSchema{ids}: + (RState{fmt, PBody{}, declared, dicts}, VReject{}) + case MDict{+id, delta}: + step_dict(fmt, declared, dicts, id, delta, has(id, declared), tbl_has(id, dicts)) + case MBatch{refs}: + step_batch(fmt, declared, dicts, all_defined(refs, dicts)) + case MEos{}: + (RState{fmt, PEnd{}, declared, dicts}, VAccept{}) + +# one message against one state, by phase +def step(fmt: Fmt, phase: Phase, +declared: List<&2, Nat>, +dicts: List<&2, DEntry>, m: Msg) -> RState & Verdict: + match phase: + case PStart{}: + step_start(fmt, declared, dicts, m) + case PBody{}: + step_body(fmt, declared, dicts, m) + case PEnd{}: + (RState{fmt, PEnd{}, declared, dicts}, VReject{}) + +def step_r(m: Msg, r: RState) -> RState & Verdict: + RState{fmt, phase, declared, dicts} = r + step(fmt, phase, declared, dicts, m) + +# projections of a step +def state(sv: RState & Verdict) -> RState: + (s, v) = sv + s + +def resp(sv: RState & Verdict) -> Verdict: + (s, v) = sv + v + +def next(m: Msg, r: RState) -> RState: + state(step_r(m, r)) + +def verdict(m: Msg, r: RState) -> Verdict: + resp(step_r(m, r)) + +def start(fmt: Fmt) -> RState: + RState{fmt, PStart{}, Nil{}, Nil{}} + +# Traces +# ------ + +# the verdict on every message of a stream, in order +def verdicts(msgs: List<&2, Msg>, +r: RState) -> List<&2, Verdict>: + match msgs: + case Nil{}: + Nil{} + case Con{+m, rest}: + verdict(m, r) <> verdicts(rest, next(m, r)) + +# the reader after a whole stream +def final(msgs: List<&2, Msg>, r: RState) -> RState: + match msgs: + case Nil{}: + r + case Con{m, rest}: + final(rest, next(m, r)) + +def accepted(v: Verdict) -> Bool: + match v: + case VAccept{}: + True{} + case VReject{}: + False{} + +def all_accepted(vs: List<&2, Verdict>) -> Bool: + match vs: + case Nil{}: + True{} + case Con{v, rest}: + accepted(v) && all_accepted(rest) + +def all_rejected(vs: List<&2, Verdict>) -> Bool: + match vs: + case Nil{}: + True{} + case Con{v, rest}: + Bool.not(accepted(v)) && all_rejected(rest) + +# Observers +# --------- + +def is_schema(m: Msg) -> Bool: + match m: + case MSchema{ids}: + True{} + case MDict{id, delta}: + False{} + case MBatch{refs}: + False{} + case MEos{}: + False{} + +def phase_of(r: RState) -> Phase: + RState{fmt, phase, declared, dicts} = r + phase + +def dicts_of(r: RState) -> List<&2, DEntry>: + RState{fmt, phase, declared, dicts} = r + dicts + +def declared_of(r: RState) -> List<&2, Nat>: + RState{fmt, phase, declared, dicts} = r + declared + +# Writer +# ------ + +# What ArrowWriter does for one VectorSchemaRoot: the schema, then one +# non-delta dictionary batch per dictionary id in use, then the record +# batches, then EOS. The Java stream and file writers emit the same +# sequence; the file writer adds a footer that is not modelled. + +# n record batches that use every id, then EOS +def emit_batches(n: Nat, +ids: List<&2, Nat>) -> List<&2, Msg>: + match n: + case 0n: + [MEos{}] + case 1n+p: + MBatch{ids} <> emit_batches(p, ids) + +# one dictionary batch per remaining id, then the batches +def emit_dicts(rem: List<&2, Nat>, ids: List<&2, Nat>, n: Nat) -> List<&2, Msg>: + match rem: + case Nil{}: + emit_batches(n, ids) + case Con{id, t}: + MDict{id, False{}} <> emit_dicts(t, ids, n) + +def write(+ids: List<&2, Nat>, n: Nat) -> List<&2, Msg>: + MSchema{ids} <> emit_dicts(ids, ids, n) + +# Show +# ---- + +def Verdict.show(v: Verdict) -> String: + match v: + case VAccept{}: + "accept" + case VReject{}: + "reject" + +def verdicts_show(vs: List<&2, Verdict>) -> String: + match vs: + case Nil{}: + "" + case Con{v, rest}: + " " ++ Verdict.show(v) ++ verdicts_show(rest) + +def Fmt.show(fmt: Fmt) -> String: + match fmt: + case FStream{}: + "stream" + case FFile{}: + "file " + +# "id:segments" for every declared id, in the reader's final table +def table_show(ids: List<&2, Nat>, +dicts: List<&2, DEntry>) -> String: + match ids: + case Nil{}: + "" + case Con{+id, rest}: + " " ++ Nat.show(id) ++ ":" ++ Nat.show(tbl_segs(id, dicts)) ++ table_show(rest, dicts) + +def final_show(r: RState) -> String: + RState{fmt, phase, declared, dicts} = r + table_show(declared, dicts) + +def trace_show(+fmt: Fmt, +msgs: List<&2, Msg>) -> String: + Fmt.show(fmt) ++ " |" ++ verdicts_show(verdicts(msgs, start(fmt))) ++ " | table" ++ final_show(final(msgs, start(fmt))) + +# the spec's delta example: schema, dict 0, batch, dict 0 delta, batch, EOS +def delta_example() -> List<&2, Msg>: + [MSchema{[0n]}, MDict{0n, False{}}, MBatch{[0n]}, MDict{0n, True{}}, MBatch{[0n]}, MEos{}] + +# the spec's replacement example: legal in a stream, illegal in a file +def replacement_example() -> List<&2, Msg>: + [MSchema{[0n]}, MDict{0n, False{}}, MBatch{[0n]}, MDict{0n, False{}}, MBatch{[0n]}, MEos{}] + +# a batch that uses dictionary 1 before it is defined, and a message after EOS +def bad_example() -> List<&2, Msg>: + [MSchema{[0n, 1n]}, MDict{0n, False{}}, MBatch{[0n, 1n]}, MDict{1n, False{}}, MBatch{[0n, 1n]}, MEos{}, MBatch{[0n]}] + +def main() -> IO(Unit): + do IO: + IO.print("delta " ++ trace_show(FStream{}, delta_example())) + IO.print("delta " ++ trace_show(FFile{}, delta_example())) + IO.print("replacement " ++ trace_show(FStream{}, replacement_example())) + IO.print("replacement " ++ trace_show(FFile{}, replacement_example())) + IO.print("bad " ++ trace_show(FStream{}, bad_example())) + IO.print("writer " ++ trace_show(FFile{}, write([3n, 7n], 2n)))