Pulsar SQL Support Avro Schema ByteBuffer Type

[gaoran10]

Fixes #6749

Motivation

Currently, the Pulsar SQL couldn't support AvroSchema use the ByteBuffer as the field type. For example, use the POJO class as below.

@Data
public static class LogFile {
    int id;
    String name;
    ByteBuffer data;
}

Producer<LogFile> producer = pulsarClient.newProducer(Schema.AVRO(LogFile.class)).topic(topic).create();

Error Log

2020-05-08T23:34:47.079+0800	ERROR	SplitRunner-5-101	com.facebook.presto.execution.executor.TaskExecutor	Error processing Split 20200508_153445_00006_nxngm.1.0-1 PulsarSplit{splitId=1, connectorId='pulsar', originSchemaName='bytes-sql-test4', schemaName='public/default', tableName='bytes-sql-test4', splitSize=4, schema='{"type":"record","name":"LogFile","namespace":"com.ran.schema.KeyValueSchemaTest$","fields":[{"name":"id","type":"int"},{"name":"name","type":["null","string"],"default":null},{"name":"data","type":["null","bytes"],"default":null}]}', schemaType=AVRO, startPositionEntryId=5, endPositionEntryId=9, startPositionLedgerId=3359, endPositionLedgerId=3359, schemaInfoProperties={"__alwaysAllowNull":"true"}} (start = 4.34095226272178E8, wall = 546 ms, cpu = 0 ms, wait = 0 ms, calls = 1)
java.lang.ClassCastException: java.nio.HeapByteBuffer cannot be cast to [B
	at org.apache.pulsar.sql.presto.PulsarRecordCursor.getSlice(PulsarRecordCursor.java:516)
	at com.facebook.presto.spi.RecordPageSource.getNextPage(RecordPageSource.java:117)
	at com.facebook.presto.operator.TableScanOperator.getOutput(TableScanOperator.java:242)
	at com.facebook.presto.operator.Driver.processInternal(Driver.java:373)
	at com.facebook.presto.operator.Driver.lambda$processFor$8(Driver.java:282)
	at com.facebook.presto.operator.Driver.tryWithLock(Driver.java:672)
	at com.facebook.presto.operator.Driver.processFor(Driver.java:276)
	at com.facebook.presto.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:973)
	at com.facebook.presto.execution.executor.PrioritizedSplitRunner.process(PrioritizedSplitRunner.java:162)
	at com.facebook.presto.execution.executor.TaskExecutor$TaskRunner.run(TaskExecutor.java:477)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Modifications

When the presto field record type is VarbinaryType.VARBINARY, check the record type is ByteBuffer, byte[], ByteBuf or others, and to process the field record by the type.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

Added unit test for getting bytes from VarbinaryType.VARBINARY type field record.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

• Does this pull request introduce a new feature? (no)

[gaoran10]

/pulsarbot run-failure-checks

[codelipenghui]

/pulsarbot run-failure-checks

[jiazhai]

/pulsarbot run-failure-checks

[jiazhai]

/pulsarbot run-failure-checks

[gaoran10]

/pulsarbot run-failure-checks

[jiazhai]

/pulsarbot run-failure-checks