feat: add Connection interface

[stephaniewang526]

Background: go/bq-sql-client-java

This PR provides a completely new Connection interface which defines separate APIs for different types of queries. This allows us to provide an industry standard way for database applications to build against the Java client library. We provide 3 new JDBC-esque API methods:

[In-scope for this PR] executeSelect - Only supports read only SELECT queries.
[In-scope for this PR] dryRun - returns som query processing statistics including schema and query parameters.
[Not in this PR] executeUpdate - Only supports DML and DDL.
[Not in this PR] execute - Any SQL - scripts, DML, DDL, SELECT etc statements.

We also integrate with the BigQueryStorage client library and use the high throughput Read API when applicable to parse query results using Arrow format. Arrow has shown better performance over Avro as the row serialization format in this experiment.

[o-shevchenko]

Hi
I use executeSelect to read data from BQ. The read is extremely slow, but I don't see much confs to tune reading.

Read of 100_000 rows takes 23930 ms.
Is it the expected speed? Is it possible to improve it?

Mono.fromCallable { bigQueryOptionsBuilder.build().service }
            .flatMap { context ->
                val connectionSettings = ConnectionSettings.newBuilder()
                    .setRequestTimeout(10L)
                    .setUseReadAPI(true)
                    .setMaxResults(1000)
                    .setNumBufferedRows(1000)
                    .setUseQueryCache(true)
                    .build();
                val connection = context.createConnection(connectionSettings)
                val bqResult = connection.executeSelect(sql)
                val result = Flux.usingWhen(
                    Mono.just(bqResult.resultSet),
                    { resultSet -> resultSet.toFlux(bqResult.schema) },
                    { _ -> Mono.fromRunnable<Unit> { connection.close() } }
                )
                Mono.just(Data(result, bqResult.schema.toSchema()))
            }
            ...
            
fun ResultSet.toFlux(schema:Schema): Flux<DataRecord> {
    return Flux.generate<DataRecord> { sink ->
        if (next()) {
            sink.next(toDataRecord(schema))
        } else {
            sink.complete()
        }
    }
}