feat(structured-ingestion) DBI-1096: Generic flattening schema projection for structured ingestion #4781
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| package structured | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "fmt" | ||
| "math" | ||
|
|
||
| "github.com/PeerDB-io/peerdb/flow/generated/protos" | ||
| "github.com/PeerDB-io/peerdb/flow/shared/types" | ||
| ) | ||
|
|
||
| const MalformedDataColumn = "malformed_data" | ||
|
|
||
| type MalformedReason int | ||
|
|
||
| const ( | ||
| ReasonUnexpected MalformedReason = iota | ||
| ReasonTypeMismatch | ||
| ReasonNaN | ||
| ReasonDuplicatedFields | ||
| ) | ||
|
|
||
| func (r MalformedReason) String() string { | ||
| switch r { | ||
| case ReasonUnexpected: | ||
| return "unexpected" | ||
|
pfcoperez marked this conversation as resolved.
Outdated
|
||
| case ReasonTypeMismatch: | ||
| return "type_mismatch" | ||
| case ReasonNaN: | ||
| return "not_a_number" | ||
| case ReasonDuplicatedFields: | ||
| return "duplicated_fields" | ||
| default: | ||
| return "" | ||
| } | ||
| } | ||
|
|
||
| // This type represents malformed data state. | ||
| // As structured is applied to a document based on a ClickHouse schema, problematic fields | ||
| // get recorded in this state. | ||
| type MalformedData struct { | ||
| // Map from seen problematic fields to their respective reasons. | ||
| reasons map[string]MalformedReason | ||
| // Optionally, the problematic values | ||
| values map[string]types.QValue | ||
| } | ||
|
|
||
| // NewMalformedData creates a new instance of MalformedData. | ||
| func NewMalformedData() *MalformedData { | ||
| return &MalformedData{ | ||
| reasons: make(map[string]MalformedReason), | ||
| values: make(map[string]types.QValue), | ||
| } | ||
| } | ||
|
|
||
| // AddField adds a problematic field along with its reason and, when value is not nil, its value to the | ||
| // MalformedData. | ||
| func (m *MalformedData) AddField(field string, reason MalformedReason, value types.QValue) { | ||
| m.reasons[field] = reason | ||
| if value != nil { | ||
| m.values[field] = value | ||
| } | ||
| } | ||
|
|
||
| func (m *MalformedData) IsEmpty() bool { | ||
| return len(m.reasons) == 0 | ||
| } | ||
|
|
||
| // MarshalJSON implements json.Marshaler, serializing the malformed fields state in the shape of a JSON object | ||
| // which is query friendly if ingested into ClickHouse: | ||
| // | ||
| // {"<field>": {"<reason>": true, "value": <value>}} | ||
| func (m *MalformedData) MarshalJSON() ([]byte, error) { | ||
| fields := make(map[string]map[string]any, len(m.reasons)) | ||
| for name, reason := range m.reasons { | ||
| reasonName := reason.String() | ||
| if reasonName == "" { | ||
| return nil, fmt.Errorf("malformed field %q has unknown reason %d", name, int(reason)) | ||
| } | ||
| fields[name] = map[string]any{reasonName: true} | ||
| } | ||
| for name, value := range m.values { | ||
| field, ok := fields[name] | ||
| if !ok { | ||
| return nil, fmt.Errorf("malformed field %q has a value but no reason", name) | ||
| } | ||
| // encoded from the Go value: nulls become JSON null, QValueJSON stays a (quoted) string | ||
| var jsonValue any | ||
| if value != nil { | ||
| jsonValue = value.Value() | ||
| } | ||
| if isJSONRepresentable(jsonValue) { | ||
| marshaledValue, err := json.Marshal(jsonValue) | ||
| if err != nil { | ||
| field["value"] = fmt.Sprintf("%v", jsonValue) | ||
| } else { | ||
| field["value"] = json.RawMessage(marshaledValue) | ||
| } | ||
| } else { | ||
| fields[name] = map[string]any{ReasonNaN.String(): true} | ||
|
Contributor
There was a problem hiding this comment. i think claude already mentioned this but here it's overriding the default reason and iiuc is only set because json.Marshal does not support it. could instead convert the NaN/-Inf/+Inf value as string and then the original reason can be used and reasonNaN can be removed (since ClickHouse supports NaN/Inf/-Inf for floats)
Member
Author
There was a problem hiding this comment. Yes Claude flagged it too. I disregarded its comment for this reason: In these cases, the data could be malformed for two different reasons: The out of domain value and the potentially being an unexpected or wrong type column. I want to keep the malformed column simple and avoid the situation were we record or need to detect all the possible ways in which a value is broken. So I decided that it was better to flag the completely out of domain value because:
Contributor
There was a problem hiding this comment. maybe I am misunderstanding something here so want to clarify. my understanding is that I can't think of a valid scenario where NaN is a valid reason on during inferencing itself: if it's a float, clickhouse can handle it; if it's not a float, then it's a typeMismatch error of some sort. Does this match your understanding or did I miss something here
Member
Author
There was a problem hiding this comment. @jgao54 You understood it perfectly! There is a wrong premise in this comment of mine:
I assumed we didn't accept these float values for ingestion but you clarified that they do, thank you! Corrected in #4798 |
||
| } | ||
| } | ||
| return json.Marshal(fields) | ||
| } | ||
|
|
||
| func (m *MalformedData) AsQValue() (types.QValue, error) { | ||
| jsonData, err := m.MarshalJSON() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return types.QValueJSON{Val: string(jsonData)}, nil | ||
| } | ||
|
|
||
| // Describes the column malformed data is recorded in. | ||
| func MalformedDataFieldDescription() *protos.FieldDescription { | ||
| return &protos.FieldDescription{ | ||
| Name: MalformedDataColumn, | ||
| Type: string(types.QValueKindJSON), | ||
| TypeModifier: -1, | ||
| Nullable: true, | ||
| } | ||
| } | ||
|
|
||
| // isJSONRepresentable reports whether v can be encoded by encoding/json, which rejects NaN and ±Inf floats. | ||
| func isJSONRepresentable(v any) bool { | ||
| validateFloat := func(f float64) bool { | ||
| return !math.IsNaN(f) && !math.IsInf(f, 0) | ||
| } | ||
| switch f := v.(type) { | ||
| case float64: | ||
| return validateFloat(f) | ||
| case float32: | ||
| return validateFloat(float64(f)) | ||
| default: | ||
| return true | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| package structured | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "math" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/PeerDB-io/peerdb/flow/shared/types" | ||
| ) | ||
|
|
||
| func TestMalformedDataMarshalJSON(t *testing.T) { | ||
| tests := []struct { | ||
| setup func(m *MalformedData) | ||
| desc string | ||
| expected string | ||
| }{ | ||
| { | ||
| desc: "empty", | ||
| setup: func(*MalformedData) {}, | ||
| expected: `{}`, | ||
| }, | ||
| { | ||
| desc: "reason without value", | ||
| setup: func(m *MalformedData) { m.AddField("a", ReasonUnexpected, nil) }, | ||
| expected: `{"a":{"unexpected":true}}`, | ||
| }, | ||
| { | ||
| desc: "not a number is a reason without value", | ||
| setup: func(m *MalformedData) { m.AddField("a", ReasonNaN, nil) }, | ||
| expected: `{"a":{"not_a_number":true}}`, | ||
| }, | ||
| { | ||
| desc: "string value", | ||
| setup: func(m *MalformedData) { m.AddField("a", ReasonTypeMismatch, types.QValueString{Val: "x"}) }, | ||
| expected: `{"a":{"type_mismatch":true,"value":"x"}}`, | ||
| }, | ||
| { | ||
| desc: "duplicated field with value", | ||
| setup: func(m *MalformedData) { m.AddField("a", ReasonDuplicatedFields, types.QValueString{Val: "x"}) }, | ||
| expected: `{"a":{"duplicated_fields":true,"value":"x"}}`, | ||
| }, | ||
| { | ||
| desc: "JSON value is kept as a quoted string", | ||
| setup: func(m *MalformedData) { m.AddField("a", ReasonTypeMismatch, types.QValueJSON{Val: `{"k":1}`}) }, | ||
| expected: `{"a":{"type_mismatch":true,"value":"{\"k\":1}"}}`, | ||
| }, | ||
| { | ||
| desc: "null value", | ||
| setup: func(m *MalformedData) { | ||
| m.AddField("a", ReasonTypeMismatch, types.QValueNull(types.QValueKindInt64)) | ||
| }, | ||
| expected: `{"a":{"type_mismatch":true,"value":null}}`, | ||
| }, | ||
| { | ||
| desc: "scalar values are encoded natively", | ||
| setup: func(m *MalformedData) { | ||
| m.AddField("i", ReasonTypeMismatch, types.QValueInt64{Val: 1}) | ||
| m.AddField("b", ReasonTypeMismatch, types.QValueBoolean{Val: true}) | ||
| m.AddField("f", ReasonTypeMismatch, types.QValueFloat64{Val: 1.5}) | ||
| }, | ||
| expected: `{"b":{"type_mismatch":true,"value":true},"f":{"type_mismatch":true,"value":1.5},"i":{"type_mismatch":true,"value":1}}`, | ||
| }, | ||
| { | ||
| desc: "value is overwritten by a later AddField for the same field", | ||
| setup: func(m *MalformedData) { | ||
| m.AddField("a", ReasonUnexpected, types.QValueString{Val: "x"}) | ||
| m.AddField("a", ReasonTypeMismatch, types.QValueString{Val: "y"}) | ||
| }, | ||
| expected: `{"a":{"type_mismatch":true,"value":"y"}}`, | ||
| }, | ||
| { | ||
| desc: "non-finite float values are dropped and the reason replaced by not_a_number", | ||
| setup: func(m *MalformedData) { | ||
| m.AddField("nan", ReasonTypeMismatch, types.QValueFloat64{Val: math.NaN()}) | ||
| m.AddField("inf", ReasonUnexpected, types.QValueFloat64{Val: math.Inf(1)}) | ||
| m.AddField("neginf", ReasonTypeMismatch, types.QValueFloat64{Val: math.Inf(-1)}) | ||
| m.AddField("nan32", ReasonTypeMismatch, types.QValueFloat32{Val: float32(math.NaN())}) | ||
| m.AddField("finite", ReasonTypeMismatch, types.QValueFloat64{Val: 1.5}) | ||
| }, | ||
| expected: `{` + | ||
| `"finite":{"type_mismatch":true,"value":1.5},` + | ||
| `"inf":{"not_a_number":true},` + | ||
| `"nan":{"not_a_number":true},` + | ||
| `"nan32":{"not_a_number":true},` + | ||
| `"neginf":{"not_a_number":true}}`, | ||
| }, | ||
| { | ||
| desc: "compound values holding non-finite floats fall back to their string representation", | ||
| setup: func(m *MalformedData) { | ||
| m.AddField("arr64", ReasonTypeMismatch, types.QValueArrayFloat64{Val: []float64{1, math.NaN()}}) | ||
| m.AddField("arr32", ReasonUnexpected, types.QValueArrayFloat32{Val: []float32{float32(math.Inf(1))}}) | ||
| m.AddField("finite", ReasonTypeMismatch, types.QValueArrayFloat64{Val: []float64{1, 2}}) | ||
| }, | ||
| expected: `{` + | ||
| `"arr32":{"unexpected":true,"value":"[+Inf]"},` + | ||
| `"arr64":{"type_mismatch":true,"value":"[1 NaN]"},` + | ||
| `"finite":{"type_mismatch":true,"value":[1,2]}}`, | ||
| }, | ||
| } | ||
|
|
||
| for _, test := range tests { | ||
| t.Run(test.desc, func(t *testing.T) { | ||
| m := NewMalformedData() | ||
| test.setup(m) | ||
| // marshal through encoding/json to also prove the Marshaler is picked up via the pointer | ||
| b, err := json.Marshal(m) | ||
| require.NoError(t, err) | ||
| require.JSONEq(t, test.expected, string(b)) | ||
| require.Equal(t, test.expected, string(b), "output must be deterministic") | ||
| }) | ||
| } | ||
|
|
||
| t.Run("unknown reason is an error", func(t *testing.T) { | ||
| m := NewMalformedData() | ||
| m.AddField("a", MalformedReason(99), nil) | ||
| _, err := json.Marshal(m) | ||
| require.ErrorContains(t, err, `"a"`) | ||
| require.ErrorContains(t, err, "unknown reason 99") | ||
| }) | ||
|
|
||
| t.Run("value without reason is an error", func(t *testing.T) { | ||
| m := NewMalformedData() | ||
| m.AddField("a", ReasonTypeMismatch, nil) | ||
| m.values["b"] = types.QValueString{Val: "x"} // AddField cannot produce this state | ||
| _, err := json.Marshal(m) | ||
| require.ErrorContains(t, err, `"b"`) | ||
| require.ErrorContains(t, err, "no reason") | ||
| }) | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.