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

* DGS-21988 Fix transformation of nullable JSON props

* Minor cleanup

* Minor cleanup

Author: rayokota

schemaregistry/serde/json.ts

@@ -332,6 +332,17 @@ async function transform(ctx: RuleContext, schema: DereferencedJSONSchema, path:
   if (fieldCtx != null) {
     fieldCtx.type = getType(schema)
   }
+  if (schema.type != null && Array.isArray(schema.type) && schema.type.length > 0) {
+    let originalType = schema.type
+    let subschema = validateSubtypes(schema, msg)
+    try {
+      if (subschema != null) {
+        return await transform(ctx, subschema, path, msg, fieldTransform)
+      }
+    } finally {
+      schema.type = originalType
+    }
+  }
   if (schema.allOf != null && schema.allOf.length > 0) {
     let subschema = validateSubschemas(schema.allOf, msg)
     if (subschema != null) {
@@ -406,6 +417,25 @@ async function transformField(ctx: RuleContext, path: string, propName: string,
   }
 }
 
+function validateSubtypes(schema: DereferencedJSONSchema, msg: any): DereferencedJSONSchema | null {
+  if (typeof schema === 'boolean') {
+    return null
+  }
+  if (schema.type == null || !Array.isArray(schema.type) || schema.type.length === 0) {
+    return null
+  }
+  for (let typ of schema.type) {
+    schema.type = typ
+    try {
+      validateJSON(msg, schema)
+      return schema
+    } catch (error) {
+      // ignore
+    }
+  }
+  return null
+}
+
 function validateSubschemas(subschemas: DereferencedJSONSchema[], msg: any): DereferencedJSONSchema | null {
   for (let subschema of subschemas) {
     try {

schemaregistry/test/serde/json.spec.ts

@@ -25,9 +25,13 @@ import {RuleRegistry} from "@confluentinc/schemaregistry/serde/rule-registry";
 import stringify from "json-stringify-deterministic";
 import {JsonataExecutor} from "@confluentinc/schemaregistry/rules/jsonata/jsonata-executor";
 import {clearKmsClients} from "@confluentinc/schemaregistry/rules/encryption/kms-registry";
+import {CelExecutor} from "../../rules/cel/cel-executor";
+import {CelFieldExecutor} from "../../rules/cel/cel-field-executor";
 
 const encryptionExecutor = EncryptionExecutor.register()
 const fieldEncryptionExecutor = FieldEncryptionExecutor.register()
+CelExecutor.register()
+CelFieldExecutor.register()
 JsonataExecutor.register()
 LocalKmsDriver.register()
 
@@ -73,6 +77,25 @@ const demoSchema = `
   }
 }
 `
+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" ]
+    }
+  }
+}
+`
 const demoSchemaWithUnion = `
 {
   "type": "object",
@@ -424,6 +447,102 @@ describe('JsonSerializer', () => {
 
     await expect(() => ser.serialize(topic, diffObj)).rejects.toThrow(SerializationError)
   })
+  it('cel field transform', async () => {
+    let conf: ClientConfig = {
+      baseURLs: [baseURL],
+      cacheCapacity: 1000
+    }
+    let client = SchemaRegistryClient.newClient(conf)
+    let serConfig: JsonSerializerConfig = {
+      useLatestVersion: true,
+    }
+    let ser = new JsonSerializer(client, SerdeType.VALUE, serConfig)
+
+    let encRule: Rule = {
+      name: 'test-cel',
+      kind: 'TRANSFORM',
+      mode: RuleMode.WRITE,
+      type: 'CEL_FIELD',
+      expr: "name == 'stringField' ; value + '-suffix'"
+    }
+    let ruleSet: RuleSet = {
+      domainRules: [encRule]
+    }
+
+    let info: SchemaInfo = {
+      schemaType: 'JSON',
+      schema: demoSchema,
+      ruleSet
+    }
+
+    await client.register(subject, info, false)
+
+    let obj = {
+      intField: 123,
+      doubleField: 45.67,
+      stringField: 'hi',
+      boolField: true,
+      bytesField: Buffer.from([0, 0, 0, 1]).toString('base64')
+    }
+    let bytes = await ser.serialize(topic, obj)
+
+    let deserConfig: JsonDeserializerConfig = {}
+    let deser = new JsonDeserializer(client, SerdeType.VALUE, deserConfig)
+    let obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2.intField).toEqual(obj.intField);
+    expect(obj2.doubleField).toBeCloseTo(obj.doubleField, 0.001);
+    expect(obj2.stringField).toEqual('hi-suffix');
+    expect(obj2.boolField).toEqual(obj.boolField);
+    expect(obj2.bytesField).toEqual(obj.bytesField);
+  })
+  it('cel field transform with nullable', async () => {
+    let conf: ClientConfig = {
+      baseURLs: [baseURL],
+      cacheCapacity: 1000
+    }
+    let client = SchemaRegistryClient.newClient(conf)
+    let serConfig: JsonSerializerConfig = {
+      useLatestVersion: true,
+    }
+    let ser = new JsonSerializer(client, SerdeType.VALUE, serConfig)
+
+    let encRule: Rule = {
+      name: 'test-cel',
+      kind: 'TRANSFORM',
+      mode: RuleMode.WRITE,
+      type: 'CEL_FIELD',
+      expr: "name == 'stringField' ; value + '-suffix'"
+    }
+    let ruleSet: RuleSet = {
+      domainRules: [encRule]
+    }
+
+    let info: SchemaInfo = {
+      schemaType: 'JSON',
+      schema: demoSchemaWithNullable,
+      ruleSet
+    }
+
+    await client.register(subject, info, false)
+
+    let obj = {
+      intField: 123,
+      doubleField: 45.67,
+      stringField: 'hi',
+      boolField: true,
+      bytesField: Buffer.from([0, 0, 0, 1]).toString('base64')
+    }
+    let bytes = await ser.serialize(topic, obj)
+
+    let deserConfig: JsonDeserializerConfig = {}
+    let deser = new JsonDeserializer(client, SerdeType.VALUE, deserConfig)
+    let obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2.intField).toEqual(obj.intField);
+    expect(obj2.doubleField).toBeCloseTo(obj.doubleField, 0.001);
+    expect(obj2.stringField).toEqual('hi-suffix');
+    expect(obj2.boolField).toEqual(obj.boolField);
+    expect(obj2.bytesField).toEqual(obj.bytesField);
+  })
   it('basic encryption', async () => {
     let conf: ClientConfig = {
       baseURLs: [baseURL],