public function read(Session $session, $table, KeySet $keySet, array $columns, array $options = [])
    {
        $options += [
            'index' => null,
            'limit' => null,
            'offset' => null,
            'transactionContext' => null
        ];

        $context = $this->pluck('transactionContext', $options);

        $call = function ($resumeToken = null) use ($table, $session, $columns, $keySet, $options) {
            if ($resumeToken) {
                $options['resumeToken'] = $resumeToken;
            }

            return $this->connection->streamingRead([
                'table' => $table,
                'session' => $session->name(),
                'columns' => $columns,
                'keySet' => $this->flattenKeySet($keySet),
                'database' => $session->info()['database']
            ] + $options);
        };

        return new Result($this, $session, $call, $context, $this->mapper);
    }