Support longs

[ryannedolan]

Summary

Add support for BIGINT aka LONG.

Details

When an input topic has a Long in the Avro schema, the resulting pipeline was trying to cast from Long down to Int, causing a ClassCastException within Flink (see below). This is because AvroConverter was essentially equating Ints and Longs, which works one way but not the other!

Testing Done

I added a BIGINT field to the built-in PERSON table. The existing integration tests should validate this now works.

Appendix

Relevant stack trace:

    Caused by: java.lang.ClassCastException: java.lang.Long cannot be cast to java.lang.Integer
	at org.apache.flink.table.data.GenericRowData.getInt(GenericRowData.java:149)
	at org.apache.flink.table.data.RowData.lambda$createFieldGetter$245ca7d1$6(RowData.java:245)
	at org.apache.flink.table.data.RowData.lambda$createFieldGetter$25774257$1(RowData.java:296)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copyRowData(RowDataSerializer.java:170)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:131)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:48)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copyRowData(RowDataSerializer.java:170)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:131)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:48)
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:80)