-module(lightspeed@pipeline@quality).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/pipeline/quality.gleam").
-export([field/3, schema/3, string_value/2, int_value/2, bool_value/2, null_value/1, valid_schema/1, field_type_label/1, check_compatibility/2, validate/2, value_label/1, validation_error_label/1, validation_result_label/1, schema_signature/1]).
-export_type([field_type/0, field/0, schema/0, value/0, boundary_value/0, validation_error/0, validation_result/0]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
?MODULEDOC(" Data-quality and schema-evolution contracts for ETL pipeline boundaries.\n").
-type field_type() :: string_type | int_type | bool_type.
-type field() :: {field, binary(), field_type(), boolean()}.
-type schema() :: {schema, binary(), integer(), list(field())}.
-type value() :: {string_value, binary()} |
{int_value, integer()} |
{bool_value, boolean()} |
null_value.
-type boundary_value() :: {boundary_value, binary(), value()}.
-type validation_error() :: {invalid_schema, binary()} |
{missing_field, binary()} |
{unexpected_field, binary()} |
{type_mismatch, binary(), field_type(), binary()}.
-type validation_result() :: {validation_passed, list(boundary_value())} |
{validation_failed, list(validation_error())}.
-file("src/lightspeed/pipeline/quality.gleam", 52).
?DOC(" Build one field contract.\n").
-spec field(binary(), field_type(), boolean()) -> field().
field(Name, Kind, Required) ->
{field, Name, Kind, Required}.
-file("src/lightspeed/pipeline/quality.gleam", 57).
?DOC(" Build one schema contract.\n").
-spec schema(binary(), integer(), list(field())) -> schema().
schema(Name, Version, Fields) ->
{schema, Name, Version, Fields}.
-file("src/lightspeed/pipeline/quality.gleam", 62).
?DOC(" Build one named string value.\n").
-spec string_value(binary(), binary()) -> boundary_value().
string_value(Name, Value) ->
{boundary_value, Name, {string_value, Value}}.
-file("src/lightspeed/pipeline/quality.gleam", 67).
?DOC(" Build one named integer value.\n").
-spec int_value(binary(), integer()) -> boundary_value().
int_value(Name, Value) ->
{boundary_value, Name, {int_value, Value}}.
-file("src/lightspeed/pipeline/quality.gleam", 72).
?DOC(" Build one named boolean value.\n").
-spec bool_value(binary(), boolean()) -> boundary_value().
bool_value(Name, Value) ->
{boundary_value, Name, {bool_value, Value}}.
-file("src/lightspeed/pipeline/quality.gleam", 77).
?DOC(" Build one named null value.\n").
-spec null_value(binary()) -> boundary_value().
null_value(Name) ->
{boundary_value, Name, null_value}.
-file("src/lightspeed/pipeline/quality.gleam", 376).
-spec contains_name(list(binary()), binary()) -> boolean().
contains_name(Names, Name) ->
case Names of
[] ->
false;
[Entry | Rest] ->
case Entry =:= Name of
true ->
true;
false ->
contains_name(Rest, Name)
end
end.
-file("src/lightspeed/pipeline/quality.gleam", 314).
-spec field_names_unique(list(field()), list(binary())) -> boolean().
field_names_unique(Fields, Seen) ->
case Fields of
[] ->
true;
[Entry | Rest] ->
case contains_name(Seen, erlang:element(2, Entry)) of
true ->
false;
false ->
field_names_unique(Rest, [erlang:element(2, Entry) | Seen])
end
end.
-file("src/lightspeed/pipeline/quality.gleam", 307).
-spec fields_valid(list(field())) -> boolean().
fields_valid(Fields) ->
case Fields of
[] ->
true;
[Entry | Rest] ->
(erlang:element(2, Entry) /= <<""/utf8>>) andalso fields_valid(Rest)
end.
-file("src/lightspeed/pipeline/quality.gleam", 82).
?DOC(" Validate schema invariants.\n").
-spec valid_schema(schema()) -> boolean().
valid_schema(Schema) ->
((((erlang:element(2, Schema) /= <<""/utf8>>) andalso (erlang:element(
3,
Schema
)
> 0))
andalso (erlang:element(4, Schema) /= []))
andalso fields_valid(erlang:element(4, Schema)))
andalso field_names_unique(erlang:element(4, Schema), []).
-file("src/lightspeed/pipeline/quality.gleam", 325).
-spec find_field(list(field()), binary()) -> gleam@option:option(field()).
find_field(Fields, Name) ->
case Fields of
[] ->
none;
[Entry | Rest] ->
case erlang:element(2, Entry) =:= Name of
true ->
{some, Entry};
false ->
find_field(Rest, Name)
end
end.
-file("src/lightspeed/pipeline/quality.gleam", 235).
-spec compatible_new_fields(list(field()), list(field())) -> {ok, nil} |
{error, binary()}.
compatible_new_fields(Previous_fields, Next_fields) ->
case Next_fields of
[] ->
{ok, nil};
[Entry | Rest] ->
case {find_field(Previous_fields, erlang:element(2, Entry)),
erlang:element(4, Entry)} of
{none, true} ->
{error,
<<"new_required_field:"/utf8,
(erlang:element(2, Entry))/binary>>};
{_, _} ->
compatible_new_fields(Previous_fields, Rest)
end
end.
-file("src/lightspeed/pipeline/quality.gleam", 145).
?DOC(" Field-type label.\n").
-spec field_type_label(field_type()) -> binary().
field_type_label(Kind) ->
case Kind of
string_type ->
<<"string"/utf8>>;
int_type ->
<<"int"/utf8>>;
bool_type ->
<<"bool"/utf8>>
end.
-file("src/lightspeed/pipeline/quality.gleam", 205).
-spec compatible_previous_fields(list(field()), list(field())) -> {ok, nil} |
{error, binary()}.
compatible_previous_fields(Previous_fields, Next_fields) ->
case Previous_fields of
[] ->
{ok, nil};
[Previous | Rest] ->
case find_field(Next_fields, erlang:element(2, Previous)) of
none ->
{error,
<<"missing_field:"/utf8,
(erlang:element(2, Previous))/binary>>};
{some, Next} ->
case erlang:element(3, Previous) /= erlang:element(3, Next) of
true ->
{error,
<<<<<<<<<<"type_changed:"/utf8,
(erlang:element(2, Previous))/binary>>/binary,
":"/utf8>>/binary,
(field_type_label(
erlang:element(3, Previous)
))/binary>>/binary,
"->"/utf8>>/binary,
(field_type_label(erlang:element(3, Next)))/binary>>};
false ->
case erlang:element(4, Previous) /= erlang:element(
4,
Next
) of
true ->
{error,
<<"requiredness_changed:"/utf8,
(erlang:element(2, Previous))/binary>>};
false ->
compatible_previous_fields(
Rest,
Next_fields
)
end
end
end
end.
-file("src/lightspeed/pipeline/quality.gleam", 91).
?DOC(" Check whether `next` is compatibility-safe relative to `previous`.\n").
-spec check_compatibility(schema(), schema()) -> {ok, nil} | {error, binary()}.
check_compatibility(Previous, Next) ->
case erlang:element(2, Previous) /= erlang:element(2, Next) of
true ->
{error,
<<<<<<"schema_name_mismatch:"/utf8,
(erlang:element(2, Previous))/binary>>/binary,
":"/utf8>>/binary,
(erlang:element(2, Next))/binary>>};
false ->
case erlang:element(3, Next) < erlang:element(3, Previous) of
true ->
{error,
<<<<<<"schema_version_regression:"/utf8,
(erlang:integer_to_binary(
erlang:element(3, Previous)
))/binary>>/binary,
"->"/utf8>>/binary,
(erlang:integer_to_binary(erlang:element(3, Next)))/binary>>};
false ->
case valid_schema(Previous) andalso valid_schema(Next) of
false ->
{error, <<"invalid_schema"/utf8>>};
true ->
case compatible_previous_fields(
erlang:element(4, Previous),
erlang:element(4, Next)
) of
{error, Reason} ->
{error, Reason};
{ok, _} ->
compatible_new_fields(
erlang:element(4, Previous),
erlang:element(4, Next)
)
end
end
end
end.
-file("src/lightspeed/pipeline/quality.gleam", 288).
-spec validate_unexpected_payload_fields(
list(boundary_value()),
list(field()),
list(validation_error())
) -> list(validation_error()).
validate_unexpected_payload_fields(Payload, Fields, Errors_rev) ->
case Payload of
[] ->
Errors_rev;
[Entry | Rest] ->
case find_field(Fields, erlang:element(2, Entry)) of
none ->
validate_unexpected_payload_fields(
Rest,
Fields,
[{unexpected_field, erlang:element(2, Entry)} |
Errors_rev]
);
{some, _} ->
validate_unexpected_payload_fields(Rest, Fields, Errors_rev)
end
end.
-file("src/lightspeed/pipeline/quality.gleam", 367).
-spec value_kind_label(value()) -> binary().
value_kind_label(Value) ->
case Value of
{string_value, _} ->
<<"string"/utf8>>;
{int_value, _} ->
<<"int"/utf8>>;
{bool_value, _} ->
<<"bool"/utf8>>;
null_value ->
<<"null"/utf8>>
end.
-file("src/lightspeed/pipeline/quality.gleam", 350).
-spec value_matches(field(), value()) -> boolean().
value_matches(Entry, Value) ->
case Value of
null_value ->
not erlang:element(4, Entry);
{string_value, _} ->
erlang:element(3, Entry) =:= string_type;
{int_value, _} ->
erlang:element(3, Entry) =:= int_type;
{bool_value, _} ->
erlang:element(3, Entry) =:= bool_type
end.
-file("src/lightspeed/pipeline/quality.gleam", 336).
-spec find_value(list(boundary_value()), binary()) -> gleam@option:option(boundary_value()).
find_value(Payload, Name) ->
case Payload of
[] ->
none;
[Entry | Rest] ->
case erlang:element(2, Entry) =:= Name of
true ->
{some, Entry};
false ->
find_value(Rest, Name)
end
end.
-file("src/lightspeed/pipeline/quality.gleam", 249).
-spec validate_schema_fields(
list(field()),
list(boundary_value()),
list(boundary_value()),
list(validation_error())
) -> {list(boundary_value()), list(validation_error())}.
validate_schema_fields(Fields, Payload, Normalized_rev, Errors_rev) ->
case Fields of
[] ->
{Normalized_rev, Errors_rev};
[Entry | Rest] ->
{Next_normalized, Next_errors} = case find_value(
Payload,
erlang:element(2, Entry)
) of
none ->
case erlang:element(4, Entry) of
true ->
{Normalized_rev,
[{missing_field, erlang:element(2, Entry)} |
Errors_rev]};
false ->
{[null_value(erlang:element(2, Entry)) |
Normalized_rev],
Errors_rev}
end;
{some, Value} ->
case value_matches(Entry, erlang:element(3, Value)) of
true ->
{[Value | Normalized_rev], Errors_rev};
false ->
{Normalized_rev,
[{type_mismatch,
erlang:element(2, Entry),
erlang:element(3, Entry),
value_kind_label(
erlang:element(3, Value)
)} |
Errors_rev]}
end
end,
validate_schema_fields(Rest, Payload, Next_normalized, Next_errors)
end.
-file("src/lightspeed/pipeline/quality.gleam", 120).
?DOC(" Validate one payload against one schema contract.\n").
-spec validate(schema(), list(boundary_value())) -> validation_result().
validate(Schema, Payload) ->
case valid_schema(Schema) of
false ->
{validation_failed, [{invalid_schema, <<"invalid_schema"/utf8>>}]};
true ->
{Normalized_rev, Schema_errors_rev} = validate_schema_fields(
erlang:element(4, Schema),
Payload,
[],
[]
),
Errors_rev = validate_unexpected_payload_fields(
Payload,
erlang:element(4, Schema),
Schema_errors_rev
),
case Errors_rev of
[] ->
{validation_passed, lists:reverse(Normalized_rev)};
_ ->
{validation_failed, lists:reverse(Errors_rev)}
end
end.
-file("src/lightspeed/pipeline/quality.gleam", 387).
-spec bool_label(boolean()) -> binary().
bool_label(Value) ->
case Value of
true ->
<<"true"/utf8>>;
false ->
<<"false"/utf8>>
end.
-file("src/lightspeed/pipeline/quality.gleam", 154).
?DOC(" Value label.\n").
-spec value_label(value()) -> binary().
value_label(Value) ->
case Value of
{string_value, V} ->
<<"string:"/utf8, V/binary>>;
{int_value, V@1} ->
<<"int:"/utf8, (erlang:integer_to_binary(V@1))/binary>>;
{bool_value, V@2} ->
<<"bool:"/utf8, (bool_label(V@2))/binary>>;
null_value ->
<<"null"/utf8>>
end.
-file("src/lightspeed/pipeline/quality.gleam", 164).
?DOC(" Validation-error label.\n").
-spec validation_error_label(validation_error()) -> binary().
validation_error_label(Error) ->
case Error of
{invalid_schema, Reason} ->
<<"invalid_schema:"/utf8, Reason/binary>>;
{missing_field, Field} ->
<<"missing_field:"/utf8, Field/binary>>;
{unexpected_field, Field@1} ->
<<"unexpected_field:"/utf8, Field@1/binary>>;
{type_mismatch, Field@2, Expected, Actual} ->
<<<<<<<<<<"type_mismatch:"/utf8, Field@2/binary>>/binary,
":expected="/utf8>>/binary,
(field_type_label(Expected))/binary>>/binary,
":actual="/utf8>>/binary,
Actual/binary>>
end.
-file("src/lightspeed/pipeline/quality.gleam", 394).
-spec join_with(binary(), list(binary())) -> binary().
join_with(Separator, Values) ->
case Values of
[] ->
<<""/utf8>>;
[Value] ->
Value;
[Value@1 | Rest] ->
<<<<Value@1/binary, Separator/binary>>/binary,
(join_with(Separator, Rest))/binary>>
end.
-file("src/lightspeed/pipeline/quality.gleam", 180).
?DOC(" Validation-result label.\n").
-spec validation_result_label(validation_result()) -> binary().
validation_result_label(Result) ->
case Result of
{validation_passed, Normalized} ->
<<"valid:"/utf8,
(join_with(
<<","/utf8>>,
gleam@list:map(
Normalized,
fun(Entry) ->
<<<<(erlang:element(2, Entry))/binary, "="/utf8>>/binary,
(value_label(erlang:element(3, Entry)))/binary>>
end
)
))/binary>>;
{validation_failed, Errors} ->
<<"invalid:"/utf8,
(join_with(
<<","/utf8>>,
gleam@list:map(Errors, fun validation_error_label/1)
))/binary>>
end.
-file("src/lightspeed/pipeline/quality.gleam", 359).
-spec field_signature(field()) -> binary().
field_signature(Field) ->
<<<<<<<<(erlang:element(2, Field))/binary, ":"/utf8>>/binary,
(field_type_label(erlang:element(3, Field)))/binary>>/binary,
":required="/utf8>>/binary,
(bool_label(erlang:element(4, Field)))/binary>>.
-file("src/lightspeed/pipeline/quality.gleam", 196).
?DOC(" Stable schema signature.\n").
-spec schema_signature(schema()) -> binary().
schema_signature(Schema) ->
<<<<<<<<<<"name="/utf8, (erlang:element(2, Schema))/binary>>/binary,
"|version="/utf8>>/binary,
(erlang:integer_to_binary(erlang:element(3, Schema)))/binary>>/binary,
"|fields="/utf8>>/binary,
(join_with(
<<","/utf8>>,
gleam@list:map(erlang:element(4, Schema), fun field_signature/1)
))/binary>>.
