fix: preserve the rest of a struct when one field has a processing error

[jzaralim]

Description

Fixes #5980. This also applies to items in arrays and keys/values in maps.

Testing done

Added QTT tests

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[stevenpyzhang]

I think this should be considered a breaking change. Users would expect to lose the entire struct when a field has an error to null, but now it's being kept.

If you ran some CSAS/CTAS on a server without this change, process some records that cause a field processing error, stopped server, "upgraded" server to one with this change, and then processed the same records that ran into field errors, would the query still output null, or would we get this new behavior?

@jzaralim could you check what the behavior is here?

[jzaralim]

@stevenpyzhang Tested it out, the old records remained as null, and new records follow the new behavior. It doesn't change the old records, but it's still a change in behaviour so I'm not sure if this is a breaking change or not.

|mmmmmmm              |null                                                                     |
|new record           |{NUM=44, AA=null}                                                        |

[stevenpyzhang]

Sorry, breaking change isn't the right term here (the query didn't break, it ran but with different behavior), I meant this change is backwards incompatible since the behavior changed. Query behavior change is something we should try to avoid as users may have some logic expecting this "bugged behavior". Ideally we keep the old behavior for now and trigger the new behavior with a config, see #5769 for an example of this. We'd eventually completely get rid of the old behavior and config when we reach some major release.

[stevenpyzhang]

lgtm, can you verify one more time in the example above that queries issued on servers before this fix keep the old behavior of dropping the entire record even after upgrading to a server with the fix?

[jzaralim]

Sure, here's an example:

-- on ksqlDB 6.2
SET 'auto.offset.reset'='earliest';
CREATE STREAM S (BDAY STRING) WITH (KAFKA_TOPIC='S', KEY_FORMAT='KAFKA', PARTITIONS=2, REPLICAS=1, VALUE_FORMAT='JSON');
CREATE STREAM OLD AS SELECT  BDAY, STRUCT(NUM:=44, B:=PARSE_TIMESTAMP(BDAY, 'yyyy-MM-dd')) FROM S;
INSERT INTO S VALUES ('abc');

-- current branch
SET 'auto.offset.reset'='earliest';
CREATE STREAM NEW AS SELECT  BDAY, STRUCT(NUM:=44, B:=PARSE_TIMESTAMP(BDAY, 'yyyy-MM-dd')) FROM S S;
INSERT INTO S VALUES ('def');

SELECT * FROM OLD EMIT CHANGES;
+-------------------------------------------------------+----------------------------------------------+
|BDAY                                                   |KSQL_COL_0                                    |
+-------------------------------------------------------+----------------------------------------------+
|abc                                                    |null                                          |
|def                                                    |null                                          |

SELECT * FROM NEW EMIT CHANGES;
+-------------------------------------------------------+----------------------------------------------+
|BDAY                                                   |KSQL_COL_0                                    |
+-------------------------------------------------------+----------------------------------------------+
|abc                                                    |{NUM=44, B=null}                              |
|def                                                    |{NUM=44, B=null}                              |