class DirectBigQueryRecordReader extends RecordReader<NullWritable, GenericRecord> {

  @Override
  public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException {
    DirectBigQueryInputSplit split = (DirectBigQueryInputSplit) genericSplit;
    schema = parser.parse(checkNotNull(split.getSchema(), "schema"));

    stream = Stream.newBuilder().setName(checkNotNull(split.getName(), "name")).build();
    ReadRowsRequest request =
        ReadRowsRequest.newBuilder()
            .setReadPosition(StreamPosition.newBuilder().setStream(stream).build())
            .build();

    client = getClient(context.getConfiguration());
    responseIterator = client.readRowsCallable().call(request).iterator();
    recordIterator = Collections.emptyIterator();

    limit = split.getLimit();
    idx = 0;
    finalized = false;
  }

}