public static Object getNestedFieldVal(GenericRecord record, String fieldName, boolean returnNullIfNotFound, boolean consistentLogicalTimestampEnabled) {
    String[] parts = fieldName.split("\\.");
    GenericRecord valueNode = record;

    for (int i = 0; i < parts.length; i++) {
      String part = parts[i];
      Object val;
      try {
        val = HoodieAvroUtils.getFieldVal(valueNode, part, returnNullIfNotFound);
      } catch (AvroRuntimeException e) {
        if (returnNullIfNotFound) {
          return null;
        } else {
          throw new HoodieException(
              fieldName + "(Part -" + parts[i] + ") field not found in record. Acceptable fields were :"
                  + valueNode.getSchema().getFields().stream().map(Field::name).collect(Collectors.toList()));
        }
      }

      if (i == parts.length - 1) {
        // return, if last part of name
        if (val == null) {
          return null;
        } else {
          Schema fieldSchema = valueNode.getSchema().getField(part).schema();
          return convertValueForSpecificDataTypes(fieldSchema, val, consistentLogicalTimestampEnabled);
        }
      } else {
        if (!(val instanceof GenericRecord)) {
          if (returnNullIfNotFound) {
            return null;
          } else {
            throw new HoodieException("Cannot find a record at part value :" + part);
          }
        } else {
          valueNode = (GenericRecord) val;
        }
      }
    }

    // This can only be reached if the length of parts is 0
    if (returnNullIfNotFound) {
      return null;
    } else {
      throw new HoodieException(
          fieldName + " field not found in record. Acceptable fields were :"
              + valueNode.getSchema().getFields().stream().map(Field::name).collect(Collectors.toList()));
    }
  }
@Test
  public void testGetNestedFieldValWithDecimalField() {
    GenericRecord rec = new GenericData.Record(new Schema.Parser().parse(SCHEMA_WITH_DECIMAL_FIELD));
    rec.put("key_col", "key");
    BigDecimal bigDecimal = new BigDecimal("1234.5678");
    ByteBuffer byteBuffer = ByteBuffer.wrap(bigDecimal.unscaledValue().toByteArray());
    rec.put("decimal_col", byteBuffer);

    Object decimalCol = HoodieAvroUtils.getNestedFieldVal(rec, "decimal_col", true, false);
    assertEquals(bigDecimal, decimalCol);

    Object obj = rec.get(1);
    assertTrue(obj instanceof ByteBuffer);
    ByteBuffer buffer = (ByteBuffer) obj;
    assertEquals(0, buffer.position());
  }