feat(structured-ingestion) DBI-1096: Generic flattening schema projection for structured ingestion

flow/connectors/clickhouse/clickhouse.go

@@ -394,81 +394,86 @@ func (c *ClickHouseConnector) GetFlags(ctx context.Context) ([]string, error) {
 	return flags, nil
 }
 
+// QValueKindForType maps a ClickHouse column type to the QValueKind expected for its values.
+func QValueKindForType(columnType string) (types.QValueKind, error) {
+	switch columnType {
+	case "String", "Nullable(String)", "LowCardinality(String)", "LowCardinality(Nullable(String))":
+		return types.QValueKindString, nil
+	case "Bool", "Nullable(Bool)":
+		return types.QValueKindBoolean, nil
+	case "Int8", "Nullable(Int8)":
+		return types.QValueKindInt8, nil
+	case "Int16", "Nullable(Int16)":
+		return types.QValueKindInt16, nil
+	case "Int32", "Nullable(Int32)":
+		return types.QValueKindInt32, nil
+	case "Int64", "Nullable(Int64)":
+		return types.QValueKindInt64, nil
+	case "Int256", "Nullable(Int256)":
+		return types.QValueKindInt256, nil
+	case "UInt8", "Nullable(UInt8)":
+		return types.QValueKindUInt8, nil
+	case "UInt16", "Nullable(UInt16)":
+		return types.QValueKindUInt16, nil
+	case "UInt32", "Nullable(UInt32)":
+		return types.QValueKindUInt32, nil
+	case "UInt64", "Nullable(UInt64)":
+		return types.QValueKindUInt64, nil
+	case "UInt256", "Nullable(UInt256)":
+		return types.QValueKindUInt256, nil
+	case "UUID", "Nullable(UUID)":
+		return types.QValueKindUUID, nil
+	case "DateTime64(6)", "Nullable(DateTime64(6))", "DateTime64(9)", "Nullable(DateTime64(9))":
+		return types.QValueKindTimestamp, nil
+	case "Time64(6)", "Nullable(Time64(6))":
+		return types.QValueKindTime, nil
+	case "Date32", "Nullable(Date32)":
+		return types.QValueKindDate, nil
+	case "Float32", "Nullable(Float32)":
+		return types.QValueKindFloat32, nil
+	case "Float64", "Nullable(Float64)":
+		return types.QValueKindFloat64, nil
+	case "Array(Int32)":
+		return types.QValueKindArrayInt32, nil
+	case "Array(Float32)":
+		return types.QValueKindArrayFloat32, nil
+	case "Array(Float64)":
+		return types.QValueKindArrayFloat64, nil
+	case "Array(String)", "Array(LowCardinality(String))":
+		return types.QValueKindArrayString, nil
+	case "Array(UUID)":
+		return types.QValueKindArrayUUID, nil
+	case "Array(DateTime64(6))":
+		return types.QValueKindArrayTimestamp, nil
+	case "Array(Int64)":
+		return types.QValueKindArrayInt64, nil
+	case "Array(Bool)":
+		return types.QValueKindArrayBoolean, nil
+	case "Array(Date)":
+		return types.QValueKindArrayDate, nil
+	case "JSON", "Nullable(JSON)":
+		return types.QValueKindJSON, nil
+	default:
+		if strings.Contains(columnType, "Decimal") {
+			if strings.HasPrefix(columnType, "Array(") {
+				return types.QValueKindArrayNumeric, nil
+			}
+			return types.QValueKindNumeric, nil
+		}
+		return types.QValueKindInvalid, fmt.Errorf("failed to resolve QValueKind for %s", columnType)
+	}
+}
+
 func GetTableSchemaForTable(tm *protos.TableMapping, columns []driver.ColumnType) (*protos.TableSchema, error) {
 	colFields := make([]*protos.FieldDescription, 0, len(columns))
 	for _, column := range columns {
 		if slices.Contains(tm.Exclude, column.Name()) {
 			continue
 		}
 
-		var qkind types.QValueKind
-		switch column.DatabaseTypeName() {
-		case "String", "Nullable(String)", "LowCardinality(String)", "LowCardinality(Nullable(String))":
-			qkind = types.QValueKindString
-		case "Bool", "Nullable(Bool)":
-			qkind = types.QValueKindBoolean
-		case "Int8", "Nullable(Int8)":
-			qkind = types.QValueKindInt8
-		case "Int16", "Nullable(Int16)":
-			qkind = types.QValueKindInt16
-		case "Int32", "Nullable(Int32)":
-			qkind = types.QValueKindInt32
-		case "Int64", "Nullable(Int64)":
-			qkind = types.QValueKindInt64
-		case "Int256", "Nullable(Int256)":
-			qkind = types.QValueKindInt256
-		case "UInt8", "Nullable(UInt8)":
-			qkind = types.QValueKindUInt8
-		case "UInt16", "Nullable(UInt16)":
-			qkind = types.QValueKindUInt16
-		case "UInt32", "Nullable(UInt32)":
-			qkind = types.QValueKindUInt32
-		case "UInt64", "Nullable(UInt64)":
-			qkind = types.QValueKindUInt64
-		case "UInt256", "Nullable(UInt256)":
-			qkind = types.QValueKindUInt256
-		case "UUID", "Nullable(UUID)":
-			qkind = types.QValueKindUUID
-		case "DateTime64(6)", "Nullable(DateTime64(6))", "DateTime64(9)", "Nullable(DateTime64(9))":
-			qkind = types.QValueKindTimestamp
-		case "Time64(6)", "Nullable(Time64(6))":
-			qkind = types.QValueKindTime
-		case "Date32", "Nullable(Date32)":
-			qkind = types.QValueKindDate
-		case "Float32", "Nullable(Float32)":
-			qkind = types.QValueKindFloat32
-		case "Float64", "Nullable(Float64)":
-			qkind = types.QValueKindFloat64
-		case "Array(Int32)":
-			qkind = types.QValueKindArrayInt32
-		case "Array(Float32)":
-			qkind = types.QValueKindArrayFloat32
-		case "Array(Float64)":
-			qkind = types.QValueKindArrayFloat64
-		case "Array(String)", "Array(LowCardinality(String))":
-			qkind = types.QValueKindArrayString
-		case "Array(UUID)":
-			qkind = types.QValueKindArrayUUID
-		case "Array(DateTime64(6))":
-			qkind = types.QValueKindArrayTimestamp
-		case "Array(Int64)":
-			qkind = types.QValueKindArrayInt64
-		case "Array(Bool)":
-			qkind = types.QValueKindArrayBoolean
-		case "Array(Date)":
-			qkind = types.QValueKindArrayDate
-		case "JSON":
-			qkind = types.QValueKindJSON
-		default:
-			if strings.Contains(column.DatabaseTypeName(), "Decimal") {
-				if strings.HasPrefix(column.DatabaseTypeName(), "Array(") {
-					qkind = types.QValueKindArrayNumeric
-				} else {
-					qkind = types.QValueKindNumeric
-				}
-			} else {
-				return nil, fmt.Errorf("failed to resolve QValueKind for %s", column.DatabaseTypeName())
-			}
+		qkind, err := QValueKindForType(column.DatabaseTypeName())
+		if err != nil {
+			return nil, err
 		}
 
 		colFields = append(colFields, &protos.FieldDescription{

flow/connectors/utils/structured/malformed_data.go

@@ -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 = "_peerdb_malformed_data"
+
+type MalformedReason int
+
+const (
+	ReasonUnexpected MalformedReason = iota
+	ReasonTypeMismatch
+	ReasonNaN
+	ReasonDuplicatedFields
+)
+
+func (r MalformedReason) String() string {
+	switch r {
+	case ReasonUnexpected:
+		return "unexpected_field"
+	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}
+		}
+	}
+	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
+	}
+}