Add row based schema validation code to detect schema mismatch

[jackjlli]

Description

This PR adds row based schema validation code to detect schema mismatch.

This validation will capture the schema mismatch on row basis, so that we don't miss any cases when converting the raw data to pinot data.

Sample mismatch detailed information:

Found data type mismatch from the following Pinot columns: [column2]
Found single-value multi-value field mismatch from the following Pinot columns: [column3]
Found multi-value structure mismatch from the following Pinot columns: [column3, column16]

[Jackie-Jiang]

High level question, do we really need row level validation? What is the overhead of this validation?
Seems like the validation only applies to the Avro reader. In that case, validating Pinot schema and Avro schema should be enough. IMO we are paying a lot of overhead for very little gain.

[mayankshriv]

@jackjlli Could you describe the case which is not caught by schema level check?

[mayankshriv]

If this PR is just adding a check, why does it need to modify the functionality?

[jackjlli]

This change is to get the single value type for single value type. This method should return the same single value type for single value type itself.

[jackjlli]

And this method would be called in https://github.com/apache/incubator-pinot/blob/2cfaed37cf581362b87a36e924cdd5744d430e03/pinot-core/src/main/java/org/apache/pinot/core/data/recordtransformer/DataTypeTransformer.java#L112
If the single value type between source and dest are the same, the data type are the same. E.g. if source is string_array and dest is string, the data type is the same, even though we should mark the flag of single-value multi-value mismatch.

[jackjlli]

Most of the cases can be covered by validating pinot schema and avro schema. One tricky thing is that when all the fields are required to be fetched, we convert the avro generic record to string first, then parse it as a json:
https://github.com/apache/incubator-pinot/blob/d54b04a2562f86dfb3adaa02ff400951d8108738/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java#L49

  /**
   * Converts from a GenericRecord to a json map
   */
  public static Map<String, Object> genericRecordToJson(GenericRecord genericRecord) {
    try {
      String jsonString = genericRecord.toString();
      return DEFAULT_MAPPER.readValue(jsonString, new TypeReference<Map<String, Object>>() {
      });
    } catch (IOException e) {
      throw new IllegalStateException("Caught exception when converting generic record " + genericRecord + " to JSON");
    }
  }

The data type of the value from the k-v pair might get changed.

[Jackie-Jiang]

Most of the cases can be covered by validating pinot schema and avro schema. One tricky thing is that when all the fields are required to be fetched, we convert the avro generic record to string first, then parse it as a json:
https://github.com/apache/incubator-pinot/blob/d54b04a2562f86dfb3adaa02ff400951d8108738/pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroRecordExtractor.java#L49

  /**
   * Converts from a GenericRecord to a json map
   */
  public static Map<String, Object> genericRecordToJson(GenericRecord genericRecord) {
    try {
      String jsonString = genericRecord.toString();
      return DEFAULT_MAPPER.readValue(jsonString, new TypeReference<Map<String, Object>>() {
      });
    } catch (IOException e) {
      throw new IllegalStateException("Caught exception when converting generic record " + genericRecord + " to JSON");
    }
  }

The data type of the value from the k-v pair might get changed.

I don't think we ever init record extractor without the fields. Also, converting to string then serializing as json doesn't seem correct. We should fix that instead of adding the row based validation.

[jackjlli]

This PR can be closed as we've already had the schema validation in another PR(#5873).