presto-elasticsearch/src/main/java/io/prestosql/elasticsearch/ElasticsearchPageSource.java/ElasticsearchPageSource/ElasticsearchPageSource
class ElasticsearchPageSource:    public ElasticsearchPageSource(
            ElasticsearchClient client,
            ConnectorSession session,
            ElasticsearchTableHandle table,
            ElasticsearchSplit split,
            List<ElasticsearchColumnHandle> columns)
    {
        requireNonNull(client, "client is null");
        requireNonNull(columns, "columns is null");

        this.columns = ImmutableList.copyOf(columns);

        decoders = createDecoders(session, columns);

        // When the _source field is requested, we need to bypass column pruning when fetching the document
        boolean needAllFields = columns.stream()
                .map(ElasticsearchColumnHandle::getName)
                .anyMatch(isEqual(SOURCE.getName()));

        // Columns to fetch as doc_fields instead of pulling them out of the JSON source
        // This is convenient for types such as DATE, TIMESTAMP, etc, which have multiple possible
        // representations in JSON, but a single normalized representation as doc_field.
        List<String> documentFields = flattenFields(columns).entrySet().stream()
                .filter(entry -> entry.getValue().equals(TIMESTAMP))
                .map(Map.Entry::getKey)
                .collect(toImmutableList());

        columnBuilders = columns.stream()
                .map(ElasticsearchColumnHandle::getType)
                .map(type -> type.createBlockBuilder(null, 1))
                .toArray(BlockBuilder[]::new);

        List<String> requiredFields = columns.stream()
                .map(ElasticsearchColumnHandle::getName)
                .filter(name -> !BuiltinColumns.NAMES.contains(name))
                .collect(toList());

        long start = System.nanoTime();
        SearchResponse searchResponse = client.beginSearch(
                table.getIndex(),
                split.getShard(),
                buildSearchQuery(table.getConstraint(), columns, table.getQuery()),
                needAllFields ? Optional.empty() : Optional.of(requiredFields),
                documentFields);
        readTimeNanos += System.nanoTime() - start;
        this.iterator = new SearchHitIterator(client, () -> searchResponse);
    }