概述

简单看一下 ES 的查询是如何实现的

ElasticsearchSchema

基本属性

public class ElasticsearchSchema extends AbstractSchema {
	//es client
    private final RestClient client;
    private final ObjectMapper mapper;
    private final Map<String, Table> tableMap;
    private final int fetchSize;

构造器

基本的创建

	public ElasticsearchSchema(RestClient client, ObjectMapper mapper, String index) {
        this(client, mapper, index, 5196);
    }

    @VisibleForTesting
    ElasticsearchSchema(RestClient client, ObjectMapper mapper, String index, int fetchSize) {
        this.client = (RestClient)Objects.requireNonNull(client, "client");
        this.mapper = (ObjectMapper)Objects.requireNonNull(mapper, "mapper");
        Preconditions.checkArgument(fetchSize > 0, "invalid fetch size. Expected %s > 0", new Object[]{fetchSize});
        this.fetchSize = fetchSize;
        if (index == null) {
            try {
                this.tableMap = this.createTables(this.indicesFromElastic());
            } catch (IOException var6) {
                throw new UncheckedIOException("Couldn't get indices", var6);
            }
        } else {
            this.tableMap = this.createTables(Collections.singleton(index));
        }

    }

方法

	protected Map<String, Table> getTableMap() {
        return this.tableMap;
    }

    private Map<String, Table> createTables(Iterable<String> indices) {
        ImmutableMap.Builder<String, Table> builder = ImmutableMap.builder();
        Iterator var3 = indices.iterator();

        while(var3.hasNext()) {
            String index = (String)var3.next();
            ElasticsearchTransport transport = new ElasticsearchTransport(this.client, this.mapper, index, this.fetchSize);
			// 创建所有的 ES index => table
            builder.put(index, new ElasticsearchTable(transport));
        }

        return builder.build();
    }

	//向ES服务器,发送GET请求,获得所有的索引库名称
    private Set<String> indicesFromElastic() throws IOException {
        String endpoint = "/_alias";
        Response response = this.client.performRequest(new Request("GET", "/_alias"));
        InputStream is = response.getEntity().getContent();
        Throwable var4 = null;

        HashSet var7;
        try {
            JsonNode root = this.mapper.readTree(is);
            if (!root.isObject() || root.size() <= 0) {
                String message = String.format(Locale.ROOT, "Invalid response for %s/%s Expected object of at least size 1 got %s (of size %d)", response.getHost(), response.getRequestLine(), root.getNodeType(), root.size());
                throw new IllegalStateException(message);
            }

			//获得所有的索引库名称
            Set<String> indices = Sets.newHashSet(root.fieldNames());
            var7 = indices;
        } catch (Throwable var16) {
            var4 = var16;
            throw var16;
        } finally {
            if (is != null) {
                if (var4 != null) {
                    try {
                        is.close();
                    } catch (Throwable var15) {
                        var4.addSuppressed(var15);
                    }
                } else {
                    is.close();
                }
            }

        }

        return var7;
    }

代码写了一堆，实际事情非常简单。

发送请求到 ES，然后根据 index 构建对应的 table 信息。

创建表的核心逻辑是下面：

ElasticsearchTransport transport = new ElasticsearchTransport(this.client, this.mapper, index, this.fetchSize);
// 创建所有的 ES index => table
builder.put(index, new ElasticsearchTable(transport));

ElasticsearchTransport

基本属性

final class ElasticsearchTransport {
    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchTable.class);

	// 默认 fetch size
    static final int DEFAULT_FETCH_SIZE = 5196;
    private final ObjectMapper mapper;
    private final RestClient restClient;
    final String indexName;
    final ElasticsearchVersion version;
    final ElasticsearchMapping mapping;
    final int fetchSize;

构造器

	ElasticsearchTransport(RestClient restClient, ObjectMapper mapper, String indexName, int fetchSize) {
        this.mapper = (ObjectMapper)Objects.requireNonNull(mapper, "mapper");
        this.restClient = (RestClient)Objects.requireNonNull(restClient, "restClient");
        this.indexName = (String)Objects.requireNonNull(indexName, "indexName");
        this.fetchSize = fetchSize;
        this.version = this.version();
        this.mapping = this.fetchAndCreateMapping();
    }

search 查询

	Function<ObjectNode, ElasticsearchJson.Result> search() {
        return this.search(Collections.emptyMap());
    }

    Function<ObjectNode, ElasticsearchJson.Result> search(Map<String, String> httpParams) {
        Objects.requireNonNull(httpParams, "httpParams");
        return (query) -> {
            Hook.QUERY_PLAN.run(query);
            String path = String.format(Locale.ROOT, "/%s/_search", this.indexName);

            HttpPost post;
            try {
                URIBuilder builder = new URIBuilder(path);
                httpParams.forEach(builder::addParameter);
				// 这里实际上是封装了 HTTP 请求
                post = new HttpPost(builder.build());

                String json = this.mapper.writeValueAsString(query);
                LOGGER.debug("Elasticsearch Query: {}", json);
                post.setEntity(new StringEntity(json, ContentType.APPLICATION_JSON));
            } catch (URISyntaxException var7) {
                throw new RuntimeException(var7);
            } catch (JsonProcessingException var8) {
                throw new UncheckedIOException(var8);
            }

            return (ElasticsearchJson.Result)this.rawHttp(ElasticsearchJson.Result.class).apply(post);
        };
    }

http 封装那一步，就是帮 SQL 转换为 ES 的查询语句：

// 把所有的请求参数,转换成:json字符串
// SELECT _MAP['id'],_MAP['title'],_MAP['price'] FROM es.books WHERE _MAP['price'] > 60 LIMIT 2
// {"query":{"constant_score":{"filter":{"range":{"price":{"gt":60}}}}},"_source":["id","title","price"],"size":2}

// SELECT * FROM es.books WHERE _MAP['price'] > 10 offset 0 fetch next 10 rows only
// {"query":{"constant_score":{"filter":{"range":{"price":{"gt":10}}}}},"from":0,"size":10}

// 统计SQL
// SELECT count(*) FROM es.books WHERE _MAP['price'] > 50
// {"query":{"constant_score":{"filter":{"range":{"price":{"gt":50}}}}},"_source":false,"size":0,"stored_fields":"_none_","track_total_hits":true}

小结

这里个人只有 2 步：

1） SQL 转换为标准的 ES

2) ES 通过 http 请求，获得响应

参考资料

https://www.lixin.help/2021/04/11/Calcite-SQL-ES-Source.html