Source Table in error message

[mbroecheler]

When the FlexibleSchema produces an error, the table name is set to "schema" which makes it hard to know which source was causing the error:

Caused by: java.lang.IllegalArgumentException: Field [type] has non-null constraint but record contains null value [table=schema]
        at com.datasqrl.error.ErrorMessage.asException(ErrorMessage.java:39) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.error.ErrorCollector.exception(ErrorCollector.java:187) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.error.ErrorCollector.fatal(ErrorCollector.java:174) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.error.ErrorCollector.fatal(ErrorCollector.java:170) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.schema.input.FlexibleSchemaValidator.handleNull(FlexibleSchemaValidator.java:114) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.schema.input.FlexibleSchemaValidator.verifyAndAdjust(FlexibleSchemaValidator.java:91) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.schema.input.FlexibleSchemaValidator.convertDataToMatchedType(FlexibleSchemaValidator.java:177) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.schema.input.FlexibleSchemaValidator.verifyAndAdjust(FlexibleSchemaValidator.java:242) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.schema.input.FlexibleSchemaValidator.verifyAndAdjust(FlexibleSchemaValidator.java:148) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.schema.input.FlexibleSchemaValidator.verifyAndAdjust(FlexibleSchemaValidator.java:76) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.schema.input.FlexibleSchemaValidator.verifyAndAdjust(FlexibleSchemaValidator.java:55) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.format.FlexibleSchemaDelegate.deserialize(FlexibleSchemaDelegate.java:62) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at com.datasqrl.format.FlexibleSchemaDelegate.deserialize(FlexibleSchemaDelegate.java:23) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:82) ~[sqrl-run.jar:0.5.9-SNAPSHOT]
        at org.apache.flink.connector.file.table.DeserializationSchemaAdapter$LineBytesInputFormat.readRecord(DeserializationSchemaAdapter.java:197) ~[sqrl-run.jar:0.5.9-SNAPSHOT]

Let's pass in an identifier for the source table so it's easier to troubleshoot.