DGS-21988 Fix transformation of nullable JSON props (#1469)

Author: rayokota

schemaregistry/serde/jsonschema/json_schema_test.go

@@ -64,6 +64,25 @@ const (
     }
   }
 }
+`
+	demoSchemaWithNullable = `
+{
+  "type": "object",
+  "properties": {
+    "IntField": { "type": "integer" },
+    "DoubleField": { "type": "number" },
+    "StringField": { 
+       "type": ["string", "null"],
+       "confluent:tags": [ "PII" ]
+    },
+    "BoolField": { "type": "boolean" },
+    "BytesField": { 
+       "type": "string",
+       "contentEncoding": "base64",
+       "confluent:tags": [ "PII" ]
+    }
+  }
+}
 `
 	demoSchemaWithUnion = `
 {
@@ -672,6 +691,71 @@ func TestJSONSchemaSerdeWithCELFieldTransform(t *testing.T) {
 	serde.MaybeFail("deserialization", err, serde.Expect(&newobj, &obj2))
 }
 
+func TestJSONSchemaSerdeWithCELFieldTransformWithNullable(t *testing.T) {
+	serde.MaybeFail = serde.InitFailFunc(t)
+	var err error
+
+	conf := schemaregistry.NewConfig("mock://")
+
+	client, err := schemaregistry.NewClient(conf)
+	serde.MaybeFail("Schema Registry configuration", err)
+
+	serConfig := NewSerializerConfig()
+	serConfig.AutoRegisterSchemas = false
+	serConfig.UseLatestVersion = true
+	ser, err := NewSerializer(client, serde.ValueSerde, serConfig)
+	serde.MaybeFail("Serializer configuration", err)
+
+	encRule := schemaregistry.Rule{
+		Name: "test-cel",
+		Kind: "TRANSFORM",
+		Mode: "WRITE",
+		Type: "CEL_FIELD",
+		Expr: "name == 'StringField' ; value + '-suffix'",
+	}
+	ruleSet := schemaregistry.RuleSet{
+		DomainRules: []schemaregistry.Rule{encRule},
+	}
+
+	info := schemaregistry.SchemaInfo{
+		Schema:     demoSchemaWithNullable,
+		SchemaType: "JSON",
+		RuleSet:    &ruleSet,
+	}
+
+	id, err := client.Register("topic1-value", info, false)
+	serde.MaybeFail("Schema registration", err)
+	if id <= 0 {
+		t.Errorf("Expected valid schema id, found %d", id)
+	}
+
+	obj := JSONDemoSchema{}
+	obj.IntField = 123
+	obj.DoubleField = 45.67
+	obj.StringField = "hi"
+	obj.BoolField = true
+	obj.BytesField = base64.StdEncoding.EncodeToString([]byte{1, 2})
+
+	bytes, err := ser.Serialize("topic1", &obj)
+	serde.MaybeFail("serialization", err)
+
+	deserConfig := NewDeserializerConfig()
+	deser, err := NewDeserializer(client, serde.ValueSerde, deserConfig)
+	serde.MaybeFail("Deserializer configuration", err)
+	deser.Client = ser.Client
+
+	obj2 := JSONDemoSchema{}
+	obj2.IntField = 123
+	obj2.DoubleField = 45.67
+	obj2.StringField = "hi-suffix"
+	obj2.BoolField = true
+	obj2.BytesField = base64.StdEncoding.EncodeToString([]byte{1, 2})
+
+	var newobj JSONDemoSchema
+	err = deser.DeserializeInto("topic1", bytes, &newobj)
+	serde.MaybeFail("deserialization", err, serde.Expect(&newobj, &obj2))
+}
+
 func TestJSONSchemaSerdeWithCELFieldTransformWithDef(t *testing.T) {
 	serde.MaybeFail = serde.InitFailFunc(t)
 	var err error

schemaregistry/serde/jsonschema/json_schema_util.go

@@ -33,6 +33,19 @@ func transform(ctx serde.RuleContext, schema *jsonschema2.Schema, path string, m
 	if fieldCtx != nil {
 		fieldCtx.Type = getType(schema)
 	}
+	if len(schema.Types) > 1 {
+		originalTypes := schema.Types
+		subschema, err := validateSubtypes(schema, msg)
+		if err != nil {
+			return nil, err
+		}
+		if subschema != nil {
+			result, err := transform(ctx, subschema, path, msg, fieldTransform)
+			schema.Types = originalTypes // restore original types
+			return result, err
+		}
+		schema.Types = originalTypes // restore original types
+	}
 	if len(schema.AllOf) > 0 {
 		subschema, err := validateSubschemas(schema.AllOf, msg)
 		if err != nil {
@@ -182,6 +195,21 @@ func transformArray(ctx serde.RuleContext, msg *reflect.Value, sch *jsonschema2.
 	return msg, nil
 }
 
+func validateSubtypes(schema *jsonschema2.Schema, msg *reflect.Value) (*jsonschema2.Schema, error) {
+	val := deref(msg)
+	for _, typ := range schema.Types {
+		schema.Types = []string{typ}
+		valid, err := validate(schema, val)
+		if err != nil {
+			return nil, err
+		}
+		if valid {
+			return schema, nil
+		}
+	}
+	return nil, nil
+}
+
 func validateSubschemas(subschemas []*jsonschema2.Schema, msg *reflect.Value) (*jsonschema2.Schema, error) {
 	val := deref(msg)
 	for _, subschema := range subschemas {