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)))