http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Table.java ---------------------------------------------------------------------- diff --git a/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Table.java b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Table.java new file mode 100644 index 0000000..636aa5f --- /dev/null +++ b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Table.java @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.elasticsearch2; + +import org.apache.calcite.adapter.elasticsearch.AbstractElasticsearchTable; +import org.apache.calcite.linq4j.AbstractEnumerable; +import org.apache.calcite.linq4j.Enumerable; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.linq4j.function.Function1; + +import org.apache.calcite.util.Util; + +import org.elasticsearch.client.Client; +import org.elasticsearch.search.SearchHit; + +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +/** + * Table based on an Elasticsearch2 type. + */ +public class Elasticsearch2Table extends AbstractElasticsearchTable { + private final Client client; + + /** + * Creates an Elasticsearch2Table. + */ + public Elasticsearch2Table(Client client, String indexName, String typeName) { + super(indexName, typeName); + this.client = client; + } + + @Override protected Enumerable<Object> find(String index, List<String> ops, + List<Map.Entry<String, Class>> fields) { + final String dbName = index; + + final String queryString = "{" + Util.toString(ops, "", ", ", "") + "}"; + + final Function1<SearchHit, Object> getter = Elasticsearch2Enumerator.getter(fields); + + return new AbstractEnumerable<Object>() { + public Enumerator<Object> enumerator() { + final Iterator<SearchHit> cursor = client.prepareSearch(dbName).setTypes(typeName) + .setSource(queryString).execute().actionGet().getHits().iterator(); + return new Elasticsearch2Enumerator(cursor, getter); + } + }; + } +} + +// End Elasticsearch2Table.java
http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/package-info.java ---------------------------------------------------------------------- diff --git a/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/package-info.java b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/package-info.java new file mode 100644 index 0000000..a1c10c9 --- /dev/null +++ b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/package-info.java @@ -0,0 +1,26 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * Query provider based on an Elasticsearch2 DB. + */ +@PackageMarker +package org.apache.calcite.adapter.elasticsearch2; + +import org.apache.calcite.avatica.util.PackageMarker; + +// End package-info.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch2/src/test/java/org/apache/calcite/test/Elasticsearch2AdapterIT.java ---------------------------------------------------------------------- diff --git a/elasticsearch2/src/test/java/org/apache/calcite/test/Elasticsearch2AdapterIT.java b/elasticsearch2/src/test/java/org/apache/calcite/test/Elasticsearch2AdapterIT.java new file mode 100644 index 0000000..4e0c2b6 --- /dev/null +++ b/elasticsearch2/src/test/java/org/apache/calcite/test/Elasticsearch2AdapterIT.java @@ -0,0 +1,270 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.test; + +import org.apache.calcite.util.Util; + +import com.google.common.base.Function; +import com.google.common.collect.ImmutableMap; + +import org.junit.Test; + +import java.util.List; +import javax.annotation.Nullable; + +/** + * Tests for the {@code org.apache.calcite.adapter.elasticsearch2} package. + * + * <p>Before calling this test, you need to populate Elasticsearch, as follows: + * + * <blockquote><code> + * git clone https://github.com/vlsi/calcite-test-dataset<br> + * cd calcite-test-dataset<br> + * mvn install + * </code></blockquote> + * + * <p>This will create a virtual machine with Elasticsearch and the "zips" test + * dataset. + */ +public class Elasticsearch2AdapterIT { + /** + * Whether to run Elasticsearch tests. Enabled by default, however test is only + * included if "it" profile is activated ({@code -Pit}). To disable, + * specify {@code -Dcalcite.test.elasticsearch=false} on the Java command line. + */ + private static final boolean ENABLED = Util.getBooleanProperty("calcite.test.elasticsearch", + true); + + /** Connection factory based on the "zips-es" model. */ + private static final ImmutableMap<String, String> ZIPS = ImmutableMap.of("model", + Elasticsearch2AdapterIT.class.getResource("/elasticsearch-zips-model.json").getPath()); + + /** Whether to run this test. */ + private boolean enabled() { + return ENABLED; + } + + /** Returns a function that checks that a particular Elasticsearch pipeline is + * generated to implement a query. */ + private static Function<List, Void> elasticsearchChecker(final String... strings) { + return new Function<List, Void>() { + @Nullable + @Override public Void apply(@Nullable List actual) { + Object[] actualArray = actual == null || actual.isEmpty() ? null + : ((List) actual.get(0)).toArray(); + CalciteAssert.assertArrayEqual("expected Elasticsearch query not found", strings, + actualArray); + return null; + } + }; + } + + @Test public void testSort() { + final String explain = "PLAN=ElasticsearchToEnumerableConverter\n" + + " ElasticsearchSort(sort0=[$4], dir0=[ASC])\n" + + " ElasticsearchProject(city=[CAST(ITEM($0, 'city')):VARCHAR(20) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], longitude=[CAST(ITEM(ITEM($0, 'loc'), 0)):FLOAT], latitude=[CAST(ITEM(ITEM($0, 'loc'), 1)):FLOAT], pop=[CAST(ITEM($0, 'pop')):INTEGER], state=[CAST(ITEM($0, 'state')):VARCHAR(2) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], id=[CAST(ITEM($0, 'id')):VARCHAR(5) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"])\n" + + " ElasticsearchTableScan(table=[[elasticsearch_raw, zips]])"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select * from zips order by \"state\"") + .returnsCount(10) + .explainContains(explain); + } + + @Test public void testSortLimit() { + final String sql = "select \"state\", \"id\" from zips\n" + + "order by \"state\", \"id\" offset 2 rows fetch next 3 rows only"; + CalciteAssert.that() + .with(ZIPS) + .query(sql) + .returnsUnordered("state=AK; id=99503", + "state=AK; id=99504", + "state=AK; id=99505") + .queryContains( + elasticsearchChecker( + "\"fields\" : [\"state\", \"id\"], \"script_fields\": {}", + "\"sort\": [ {\"state\": \"asc\"}, {\"id\": \"asc\"}]", + "\"from\": 2", + "\"size\": 3")); + } + + @Test public void testOffsetLimit() { + final String sql = "select \"state\", \"id\" from zips\n" + + "offset 2 fetch next 3 rows only"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .runs() + .queryContains( + elasticsearchChecker( + "\"from\": 2", + "\"size\": 3", + "\"fields\" : [\"state\", \"id\"], \"script_fields\": {}")); + } + + @Test public void testLimit() { + final String sql = "select \"state\", \"id\" from zips\n" + + "fetch next 3 rows only"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .runs() + .queryContains( + elasticsearchChecker( + "\"size\": 3", + "\"fields\" : [\"state\", \"id\"], \"script_fields\": {}")); + } + + @Test public void testFilterSort() { + final String sql = "select * from zips\n" + + "where \"city\" = 'SPRINGFIELD' and \"id\" >= '70000'\n" + + "order by \"state\", \"id\""; + final String explain = "PLAN=ElasticsearchToEnumerableConverter\n" + + " ElasticsearchSort(sort0=[$4], sort1=[$5], dir0=[ASC], dir1=[ASC])\n" + + " ElasticsearchProject(city=[CAST(ITEM($0, 'city')):VARCHAR(20) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], longitude=[CAST(ITEM(ITEM($0, 'loc'), 0)):FLOAT], latitude=[CAST(ITEM(ITEM($0, 'loc'), 1)):FLOAT], pop=[CAST(ITEM($0, 'pop')):INTEGER], state=[CAST(ITEM($0, 'state')):VARCHAR(2) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], id=[CAST(ITEM($0, 'id')):VARCHAR(5) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"])\n" + + " ElasticsearchFilter(condition=[AND(=(CAST(ITEM($0, 'city')):VARCHAR(20) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\", 'SPRINGFIELD'), >=(CAST(ITEM($0, 'id')):VARCHAR(5) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\", '70000'))])\n" + + " ElasticsearchTableScan(table=[[elasticsearch_raw, zips]])"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .returnsOrdered( + "city=SPRINGFIELD; longitude=-92.54567; latitude=35.274879; pop=752; state=AR; id=72157", + "city=SPRINGFIELD; longitude=-102.617322; latitude=37.406727; pop=1992; state=CO; id=81073", + "city=SPRINGFIELD; longitude=-90.577479; latitude=30.415738; pop=5597; state=LA; id=70462", + "city=SPRINGFIELD; longitude=-123.015259; latitude=44.06106; pop=32384; state=OR; id=97477", + "city=SPRINGFIELD; longitude=-122.917108; latitude=44.056056; pop=27521; state=OR; id=97478") + .queryContains( + elasticsearchChecker("\"query\" : {\"constant_score\":{\"filter\":{\"bool\":" + + "{\"must\":[{\"term\":{\"city\":\"springfield\"}},{\"range\":{\"id\":{\"gte\":\"70000\"}}}]}}}}", + "\"fields\" : [\"city\", \"pop\", \"state\", \"id\"], \"script_fields\": {\"longitude\":{\"script\":\"_source.loc[0]\"}, \"latitude\":{\"script\":\"_source.loc[1]\"}}", + "\"sort\": [ {\"state\": \"asc\"}, {\"id\": \"asc\"}]")) + .explainContains(explain); + } + + @Test public void testFilterSortDesc() { + final String sql = "select * from zips\n" + + "where \"pop\" BETWEEN 20000 AND 20100\n" + + "order by \"state\" desc, \"pop\""; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .limit(4) + .returnsOrdered( + "city=SHERIDAN; longitude=-106.964795; latitude=44.78486; pop=20025; state=WY; id=82801", + "city=MOUNTLAKE TERRAC; longitude=-122.304036; latitude=47.793061; pop=20059; state=WA; id=98043", + "city=FALMOUTH; longitude=-77.404537; latitude=38.314557; pop=20039; state=VA; id=22405", + "city=FORT WORTH; longitude=-97.318409; latitude=32.725551; pop=20012; state=TX; id=76104"); + } + + @Test public void testFilterRedundant() { + final String sql = "select * from zips\n" + + "where \"state\" > 'CA' and \"state\" < 'AZ' and \"state\" = 'OK'"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .runs() + .queryContains( + elasticsearchChecker("" + + "\"query\" : {\"constant_score\":{\"filter\":{\"bool\":" + + "{\"must\":[{\"term\":{\"state\":\"ok\"}}]}}}}", + "\"fields\" : [\"city\", \"pop\", \"state\", \"id\"], \"script_fields\": {\"longitude\":{\"script\":\"_source.loc[0]\"}, \"latitude\":{\"script\":\"_source.loc[1]\"}}")); + } + + @Test public void testInPlan() { + final String[] searches = { + "\"query\" : {\"constant_score\":{\"filter\":{\"bool\":{\"should\":" + + "[{\"bool\":{\"must\":[{\"term\":{\"pop\":20012}}]}},{\"bool\":{\"must\":[{\"term\":" + + "{\"pop\":15590}}]}}]}}}}", + "\"fields\" : [\"city\", \"pop\", \"state\", \"id\"], \"script_fields\": {\"longitude\":{\"script\":\"_source.loc[0]\"}, \"latitude\":{\"script\":\"_source.loc[1]\"}}" + }; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select * from zips where \"pop\" in (20012, 15590)") + .returnsUnordered( + "city=COVINA; longitude=-117.884285; latitude=34.08596; pop=15590; state=CA; id=91723", + "city=ARLINGTON; longitude=-97.091987; latitude=32.654752; pop=15590; state=TX; id=76018", + "city=CROFTON; longitude=-76.680166; latitude=39.011163; pop=15590; state=MD; id=21114", + "city=FORT WORTH; longitude=-97.318409; latitude=32.725551; pop=20012; state=TX; id=76104", + "city=DINUBA; longitude=-119.39087; latitude=36.534931; pop=20012; state=CA; id=93618") + .queryContains(elasticsearchChecker(searches)); + } + + @Test public void testZips() { + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select \"state\", \"city\" from zips") + .returnsCount(10); + } + + @Test public void testProject() { + final String sql = "select \"state\", \"city\", 0 as \"zero\"\n" + + "from zips\n" + + "order by \"state\", \"city\""; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .limit(2) + .returnsUnordered("state=AK; city=ELMENDORF AFB; zero=0", + "state=AK; city=EIELSON AFB; zero=0") + .queryContains( + elasticsearchChecker("\"sort\": [ {\"state\": \"asc\"}, {\"city\": \"asc\"}]", + "\"fields\" : [\"state\", \"city\"], \"script_fields\": {\"zero\":{\"script\": \"0\"}}")); + } + + @Test public void testFilter() { + final String explain = "PLAN=ElasticsearchToEnumerableConverter\n" + + " ElasticsearchProject(state=[CAST(ITEM($0, 'state')):VARCHAR(2) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], city=[CAST(ITEM($0, 'city')):VARCHAR(20) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"])\n" + + " ElasticsearchFilter(condition=[=(CAST(ITEM($0, 'state')):VARCHAR(2) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\", 'CA')])\n" + + " ElasticsearchTableScan(table=[[elasticsearch_raw, zips]])"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select \"state\", \"city\" from zips where \"state\" = 'CA'") + .limit(2) + .returnsUnordered("state=CA; city=LOS ANGELES", + "state=CA; city=LOS ANGELES") + .explainContains(explain); + } + + @Test public void testFilterReversed() { + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select \"state\", \"city\" from zips where 'WI' < \"state\"") + .limit(2) + .returnsUnordered("state=WV; city=WELCH", + "state=WV; city=HANOVER"); + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select \"state\", \"city\" from zips where \"state\" > 'WI'") + .limit(2) + .returnsUnordered("state=WV; city=WELCH", + "state=WV; city=HANOVER"); + } +} + +// End Elasticsearch2AdapterIT.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch2/src/test/resources/elasticsearch-zips-model.json ---------------------------------------------------------------------- diff --git a/elasticsearch2/src/test/resources/elasticsearch-zips-model.json b/elasticsearch2/src/test/resources/elasticsearch-zips-model.json new file mode 100644 index 0000000..6c00a9c --- /dev/null +++ b/elasticsearch2/src/test/resources/elasticsearch-zips-model.json @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the License); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an AS IS BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +{ + "version": "1.0", + "defaultSchema": "elasticsearch", + "schemas": [ + { + "type": "custom", + "name": "elasticsearch_raw", + "factory": "org.apache.calcite.adapter.elasticsearch2.Elasticsearch2SchemaFactory", + "operand": { + "coordinates": "{'127.0.0.1': 9300}", + "userConfig": "{'bulk.flush.max.actions': 10, 'bulk.flush.max.size.mb': 1}", + "index": "usa" + } + }, + { + "name": "elasticsearch", + "tables": [ + { + "name": "ZIPS", + "type": "view", + "sql": [ + "select cast(_MAP['city'] AS varchar(20)) AS \"city\",\n", + " cast(_MAP['loc'][0] AS float) AS \"longitude\",\n", + " cast(_MAP['loc'][1] AS float) AS \"latitude\",\n", + " cast(_MAP['pop'] AS integer) AS \"pop\",\n", + " cast(_MAP['state'] AS varchar(2)) AS \"state\",\n", + " cast(_MAP['id'] AS varchar(5)) AS \"id\"\n", + "from \"elasticsearch_raw\".\"zips\"" + ] + } + ] + } + ] +} http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch2/src/test/resources/log4j.properties ---------------------------------------------------------------------- diff --git a/elasticsearch2/src/test/resources/log4j.properties b/elasticsearch2/src/test/resources/log4j.properties new file mode 100644 index 0000000..834e2db --- /dev/null +++ b/elasticsearch2/src/test/resources/log4j.properties @@ -0,0 +1,24 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to you under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Root logger is configured at INFO and is sent to A1 +log4j.rootLogger=INFO, A1 + +# A1 goes to the console +log4j.appender.A1=org.apache.log4j.ConsoleAppender + +# Set the pattern for each log message +log4j.appender.A1.layout=org.apache.log4j.PatternLayout +log4j.appender.A1.layout.ConversionPattern=%d [%t] %-5p - %m%n http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch5/pom.xml ---------------------------------------------------------------------- diff --git a/elasticsearch5/pom.xml b/elasticsearch5/pom.xml new file mode 100644 index 0000000..a599bac --- /dev/null +++ b/elasticsearch5/pom.xml @@ -0,0 +1,153 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with +this work for additional information regarding copyright ownership. +The ASF licenses this file to you under the Apache License, Version 2.0 +(the "License"); you may not use this file except in compliance with +the License. You may obtain a copy of the License at + +http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.calcite</groupId> + <artifactId>calcite</artifactId> + <version>1.14.0-SNAPSHOT</version> + </parent> + + <artifactId>calcite-elasticsearch5</artifactId> + <packaging>jar</packaging> + <version>1.14.0-SNAPSHOT</version> + <name>Calcite Elasticsearch5</name> + <description>Elasticsearch5 adapter for Calcite</description> + + <properties> + <top.dir>${project.basedir}/..</top.dir> + </properties> + + <dependencies> + <dependency> + <groupId>com.google.guava</groupId> + <artifactId>guava</artifactId> + </dependency> + <dependency> + <groupId>org.apache.calcite</groupId> + <artifactId>calcite-core</artifactId> + <type>jar</type> + </dependency> + <dependency> + <groupId>org.apache.calcite</groupId> + <artifactId>calcite-core</artifactId> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </dependency> + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.calcite</groupId> + <artifactId>calcite-linq4j</artifactId> + </dependency> + <dependency> + <groupId>org.elasticsearch.client</groupId> + <artifactId>transport</artifactId> + <version>${elasticsearch5-java-driver.version}</version> + </dependency> + <dependency> + <groupId>org.elasticsearch</groupId> + <artifactId>elasticsearch</artifactId> + <version>${elasticsearch5-java-driver.version}</version> + </dependency> + <dependency> + <groupId>com.carrotsearch</groupId> + <artifactId>hppc</artifactId> + <version>${hppc.version}</version> + </dependency> + <dependency> + <groupId>com.google.code.findbugs</groupId> + <artifactId>jsr305</artifactId> + </dependency> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-api</artifactId> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <artifactId>maven-dependency-plugin</artifactId> + <version>${maven-dependency-plugin.version}</version> + <executions> + <execution> + <id>analyze</id> + <goals> + <goal>analyze-only</goal> + </goals> + <configuration> + <failOnWarning>true</failOnWarning> + <!-- ignore "unused but declared" warnings --> + <ignoredUnusedDeclaredDependencies> + <ignoredUnusedDeclaredDependency>org.apache.calcite.avatica:avatica</ignoredUnusedDeclaredDependency> + <ignoredUnusedDeclaredDependency>org.slf4j:slf4j-api</ignoredUnusedDeclaredDependency> + <ignoredUnusedDeclaredDependency>org.slf4j:slf4j-log4j12</ignoredUnusedDeclaredDependency> + </ignoredUnusedDeclaredDependencies> + </configuration> + </execution> + </executions> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-jar-plugin</artifactId> + <executions> + <execution> + <goals> + <goal>test-jar</goal> + </goals> + </execution> + </executions> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-release-plugin</artifactId> + </plugin> + <!-- Parent module has the same plugin and does the work of + generating -sources.jar for each project. But without the + plugin declared here, IDEs don't know the sources are + available. --> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-source-plugin</artifactId> + <executions> + <execution> + <id>attach-sources</id> + <phase>verify</phase> + <goals> + <goal>jar-no-fork</goal> + <goal>test-jar-no-fork</goal> + </goals> + </execution> + </executions> + </plugin> + </plugins> + </build> +</project> http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Enumerator.java ---------------------------------------------------------------------- diff --git a/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Enumerator.java b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Enumerator.java new file mode 100644 index 0000000..262058a --- /dev/null +++ b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Enumerator.java @@ -0,0 +1,154 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.elasticsearch5; + +import org.apache.calcite.avatica.util.DateTimeUtils; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.linq4j.function.Function1; +import org.apache.calcite.linq4j.tree.Primitive; + +import org.elasticsearch.search.SearchHit; + +import java.util.Date; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +/** + * Enumerator that reads from an Elasticsearch type. + */ +public class Elasticsearch5Enumerator implements Enumerator<Object> { + private final Iterator<SearchHit> cursor; + private final Function1<SearchHit, Object> getter; + private Object current; + + /** + * Creates an Elasticsearch5Enumerator. + * + * @param cursor Iterator over Elasticsearch {@link SearchHit} objects + * @param getter Converts an object into a list of fields + */ + public Elasticsearch5Enumerator(Iterator<SearchHit> cursor, + Function1<SearchHit, Object> getter) { + this.cursor = cursor; + this.getter = getter; + } + + public Object current() { + return current; + } + + public boolean moveNext() { + if (cursor.hasNext()) { + SearchHit map = cursor.next(); + current = getter.apply(map); + return true; + } else { + current = null; + return false; + } + } + + public void reset() { + throw new UnsupportedOperationException(); + } + + public void close() { + // nothing to do + } + + private static Function1<SearchHit, Map> mapGetter() { + return new Function1<SearchHit, Map>() { + public Map apply(SearchHit searchHitFields) { + return (Map) searchHitFields.getFields(); + } + }; + } + + private static Function1<SearchHit, Object> singletonGetter(final String fieldName, + final Class fieldClass) { + return new Function1<SearchHit, Object>() { + public Object apply(SearchHit searchHitFields) { + if (searchHitFields.getFields().isEmpty()) { + return convert(searchHitFields.getSource(), fieldClass); + } else { + return convert(searchHitFields.getFields(), fieldClass); + } + } + }; + } + + /** + * Function that extracts a given set of fields from {@link SearchHit} + * objects. + * + * @param fields List of fields to project + */ + private static Function1<SearchHit, Object[]> listGetter( + final List<Map.Entry<String, Class>> fields) { + return new Function1<SearchHit, Object[]>() { + public Object[] apply(SearchHit searchHitFields) { + Object[] objects = new Object[fields.size()]; + for (int i = 0; i < fields.size(); i++) { + final Map.Entry<String, Class> field = fields.get(i); + final String name = field.getKey(); + if (searchHitFields.getFields().isEmpty()) { + objects[i] = convert(searchHitFields.getSource().get(name), + field.getValue()); + } else { + objects[i] = convert(searchHitFields.getField(name).getValue(), + field.getValue()); + } + } + return objects; + } + }; + } + + static Function1<SearchHit, Object> getter(List<Map.Entry<String, Class>> fields) { + //noinspection unchecked + return fields == null + ? (Function1) mapGetter() + : fields.size() == 1 + ? singletonGetter(fields.get(0).getKey(), fields.get(0).getValue()) + : (Function1) listGetter(fields); + } + + private static Object convert(Object o, Class clazz) { + if (o == null) { + return null; + } + Primitive primitive = Primitive.of(clazz); + if (primitive != null) { + clazz = primitive.boxClass; + } else { + primitive = Primitive.ofBox(clazz); + } + if (clazz.isInstance(o)) { + return o; + } + if (o instanceof Date && primitive != null) { + o = ((Date) o).getTime() / DateTimeUtils.MILLIS_PER_DAY; + } + if (o instanceof Number && primitive != null) { + return primitive.number((Number) o); + } + return o; + } +} + +// End Elasticsearch5Enumerator.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Schema.java ---------------------------------------------------------------------- diff --git a/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Schema.java b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Schema.java new file mode 100644 index 0000000..6e16f2f --- /dev/null +++ b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Schema.java @@ -0,0 +1,138 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.elasticsearch5; + +import org.apache.calcite.adapter.elasticsearch.ElasticsearchSchema; +import org.apache.calcite.schema.Table; +import org.apache.calcite.schema.impl.AbstractSchema; + +import com.carrotsearch.hppc.cursors.ObjectObjectCursor; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; + +import org.elasticsearch.action.admin.indices.get.GetIndexRequest; +import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest; +import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse; +import org.elasticsearch.client.Client; +import org.elasticsearch.client.transport.TransportClient; +import org.elasticsearch.cluster.metadata.MappingMetaData; +import org.elasticsearch.cluster.node.DiscoveryNode; +import org.elasticsearch.common.collect.ImmutableOpenMap; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.transport.InetSocketTransportAddress; +import org.elasticsearch.common.transport.TransportAddress; +import org.elasticsearch.transport.client.PreBuiltTransportClient; + +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * Schema mapped onto an index of ELASTICSEARCH types. + * + * <p>Each table in the schema is an ELASTICSEARCH type in that index. + */ +public class Elasticsearch5Schema extends AbstractSchema + implements ElasticsearchSchema { + final String index; + + private transient Client client; + + /** + * Creates an Elasticsearch5 schema. + * + * @param coordinates Map of Elasticsearch node locations (host, port) + * @param userConfig Map of user-specified configurations + * @param indexName Elasticsearch database name, e.g. "usa". + */ + Elasticsearch5Schema(Map<String, Integer> coordinates, + Map<String, String> userConfig, String indexName) { + super(); + + final List<InetSocketAddress> transportAddresses = new ArrayList<>(); + for (Map.Entry<String, Integer> coordinate: coordinates.entrySet()) { + transportAddresses.add( + new InetSocketAddress(coordinate.getKey(), coordinate.getValue())); + } + + open(transportAddresses, userConfig); + + if (client != null) { + final String[] indices = client.admin().indices() + .getIndex(new GetIndexRequest().indices(indexName)) + .actionGet().getIndices(); + if (indices.length == 1) { + index = indices[0]; + } else { + index = null; + } + } else { + index = null; + } + } + + @Override protected Map<String, Table> getTableMap() { + final ImmutableMap.Builder<String, Table> builder = ImmutableMap.builder(); + + try { + final GetMappingsResponse response = client.admin().indices() + .getMappings(new GetMappingsRequest().indices(index)) + .get(); + ImmutableOpenMap<String, MappingMetaData> mapping = + response.getMappings().get(index); + for (ObjectObjectCursor<String, MappingMetaData> c : mapping) { + builder.put(c.key, new Elasticsearch5Table(client, index, c.key)); + } + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException(e); + } + return builder.build(); + } + + private void open(List<InetSocketAddress> transportAddresses, + Map<String, String> userConfig) { + final List<TransportAddress> transportNodes = new ArrayList<>(transportAddresses.size()); + for (InetSocketAddress address : transportAddresses) { + transportNodes.add(new InetSocketTransportAddress(address)); + } + + Settings settings = Settings.builder().put(userConfig).build(); + + final TransportClient transportClient = new PreBuiltTransportClient(settings); + for (TransportAddress transport : transportNodes) { + transportClient.addTransportAddress(transport); + } + + final List<DiscoveryNode> nodes = + ImmutableList.copyOf(transportClient.connectedNodes()); + if (nodes.isEmpty()) { + throw new RuntimeException("Cannot connect to any elasticsearch nodes"); + } + + client = transportClient; + } + + @Override public String getIndex() { + return index; + } +} + +// End Elasticsearch5Schema.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5SchemaFactory.java ---------------------------------------------------------------------- diff --git a/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5SchemaFactory.java b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5SchemaFactory.java new file mode 100644 index 0000000..b573d37 --- /dev/null +++ b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5SchemaFactory.java @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.elasticsearch5; + +import org.apache.calcite.schema.Schema; +import org.apache.calcite.schema.SchemaFactory; +import org.apache.calcite.schema.SchemaPlus; + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; + +import java.io.IOException; +import java.util.Map; + +/** + * Factory that creates an {@link Elasticsearch5Schema}. + * + * <p>Allows a custom schema to be included in a model.json file. + */ +@SuppressWarnings("UnusedDeclaration") +public class Elasticsearch5SchemaFactory implements SchemaFactory { + + public Elasticsearch5SchemaFactory() { + } + + @Override public Schema create(SchemaPlus parentSchema, String name, + Map<String, Object> operand) { + final Map map = (Map) operand; + + final ObjectMapper mapper = new ObjectMapper(); + mapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true); + + try { + final Map<String, Integer> coordinates = + mapper.readValue((String) map.get("coordinates"), + new TypeReference<Map<String, Integer>>() { }); + final Map<String, String> userConfig = + mapper.readValue((String) map.get("userConfig"), + new TypeReference<Map<String, String>>() { }); + final String index = (String) map.get("index"); + return new Elasticsearch5Schema(coordinates, userConfig, index); + } catch (IOException e) { + throw new RuntimeException("Cannot parse values from json", e); + } + } +} + +// End Elasticsearch5SchemaFactory.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Table.java ---------------------------------------------------------------------- diff --git a/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Table.java b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Table.java new file mode 100644 index 0000000..175e3cc --- /dev/null +++ b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/Elasticsearch5Table.java @@ -0,0 +1,86 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.elasticsearch5; + +import org.apache.calcite.adapter.elasticsearch.AbstractElasticsearchTable; +import org.apache.calcite.linq4j.AbstractEnumerable; +import org.apache.calcite.linq4j.Enumerable; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.linq4j.function.Function1; + +import org.apache.calcite.util.Util; + +import org.elasticsearch.client.Client; +import org.elasticsearch.common.xcontent.NamedXContentRegistry; +import org.elasticsearch.common.xcontent.XContent; +import org.elasticsearch.common.xcontent.XContentParser; +import org.elasticsearch.common.xcontent.json.JsonXContent; +import org.elasticsearch.index.query.QueryParseContext; +import org.elasticsearch.search.SearchHit; +import org.elasticsearch.search.builder.SearchSourceBuilder; + +import java.io.IOException; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +/** + * Table based on an Elasticsearch5 type. + */ +public class Elasticsearch5Table extends AbstractElasticsearchTable { + private final Client client; + + /** + * Creates an Elasticsearch5Table. + */ + public Elasticsearch5Table(Client client, String indexName, String typeName) { + super(indexName, typeName); + this.client = client; + } + + @Override protected Enumerable<Object> find(String index, List<String> ops, + List<Map.Entry<String, Class>> fields) { + final String dbName = index; + + final SearchSourceBuilder searchSourceBuilder; + if (ops.isEmpty()) { + searchSourceBuilder = new SearchSourceBuilder(); + } else { + String queryString = "{" + Util.toString(ops, "", ", ", "") + "}"; + NamedXContentRegistry xContentRegistry = NamedXContentRegistry.EMPTY; + XContent xContent = JsonXContent.jsonXContent; + try (XContentParser parser = xContent.createParser(xContentRegistry, queryString)) { + final QueryParseContext queryParseContext = new QueryParseContext(parser); + searchSourceBuilder = SearchSourceBuilder.fromXContent(queryParseContext); + } catch (IOException ex) { + throw new RuntimeException(ex); + } + } + final Function1<SearchHit, Object> getter = Elasticsearch5Enumerator.getter(fields); + + return new AbstractEnumerable<Object>() { + public Enumerator<Object> enumerator() { + final Iterator<SearchHit> cursor = client.prepareSearch(dbName).setTypes(typeName) + .setSource(searchSourceBuilder) + .execute().actionGet().getHits().iterator(); + return new Elasticsearch5Enumerator(cursor, getter); + } + }; + } +} + +// End Elasticsearch5Table.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/package-info.java ---------------------------------------------------------------------- diff --git a/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/package-info.java b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/package-info.java new file mode 100644 index 0000000..70c4695 --- /dev/null +++ b/elasticsearch5/src/main/java/org/apache/calcite/adapter/elasticsearch5/package-info.java @@ -0,0 +1,26 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * Query provider based on an Elasticsearch5 DB. + */ +@PackageMarker +package org.apache.calcite.adapter.elasticsearch5; + +import org.apache.calcite.avatica.util.PackageMarker; + +// End package-info.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch5/src/test/java/org/apache/calcite/test/Elasticsearch5AdapterIT.java ---------------------------------------------------------------------- diff --git a/elasticsearch5/src/test/java/org/apache/calcite/test/Elasticsearch5AdapterIT.java b/elasticsearch5/src/test/java/org/apache/calcite/test/Elasticsearch5AdapterIT.java new file mode 100644 index 0000000..1cf52fe --- /dev/null +++ b/elasticsearch5/src/test/java/org/apache/calcite/test/Elasticsearch5AdapterIT.java @@ -0,0 +1,270 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.test; + +import org.apache.calcite.util.Util; + +import com.google.common.base.Function; +import com.google.common.collect.ImmutableMap; + +import org.junit.Test; + +import java.util.List; +import javax.annotation.Nullable; + +/** + * Tests for the {@code org.apache.calcite.adapter.elasticsearch} package. + * + * <p>Before calling this test, you need to populate Elasticsearch, as follows: + * + * <blockquote><code> + * git clone https://github.com/vlsi/calcite-test-dataset<br> + * cd calcite-test-dataset<br> + * mvn install + * </code></blockquote> + * + * <p>This will create a virtual machine with Elasticsearch and the "zips" test + * dataset. + */ +public class Elasticsearch5AdapterIT { + /** + * Whether to run Elasticsearch tests. Enabled by default, however test is only + * included if "it" profile is activated ({@code -Pit}). To disable, + * specify {@code -Dcalcite.test.elasticsearch=false} on the Java command line. + */ + private static final boolean ENABLED = Util.getBooleanProperty("calcite.test.elasticsearch", + true); + + /** Connection factory based on the "zips-es" model. */ + private static final ImmutableMap<String, String> ZIPS = ImmutableMap.of("model", + Elasticsearch5AdapterIT.class.getResource("/elasticsearch-zips-model.json").getPath()); + + /** Whether to run this test. */ + private boolean enabled() { + return ENABLED; + } + + /** Returns a function that checks that a particular Elasticsearch pipeline is + * generated to implement a query. */ + private static Function<List, Void> elasticsearchChecker(final String... strings) { + return new Function<List, Void>() { + @Nullable + @Override public Void apply(@Nullable List actual) { + Object[] actualArray = actual == null || actual.isEmpty() ? null + : ((List) actual.get(0)).toArray(); + CalciteAssert.assertArrayEqual("expected Elasticsearch query not found", strings, + actualArray); + return null; + } + }; + } + + @Test public void testSort() { + final String explain = "PLAN=ElasticsearchToEnumerableConverter\n" + + " ElasticsearchSort(sort0=[$4], dir0=[ASC])\n" + + " ElasticsearchProject(city=[CAST(ITEM($0, 'city')):VARCHAR(20) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], longitude=[CAST(ITEM(ITEM($0, 'loc'), 0)):FLOAT], latitude=[CAST(ITEM(ITEM($0, 'loc'), 1)):FLOAT], pop=[CAST(ITEM($0, 'pop')):INTEGER], state=[CAST(ITEM($0, 'state')):VARCHAR(2) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], id=[CAST(ITEM($0, 'id')):VARCHAR(5) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"])\n" + + " ElasticsearchTableScan(table=[[elasticsearch_raw, zips]])"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select * from zips order by \"state\"") + .returnsCount(10) + .explainContains(explain); + } + + @Test public void testSortLimit() { + final String sql = "select \"state\", \"id\" from zips\n" + + "order by \"state\", \"id\" offset 2 rows fetch next 3 rows only"; + CalciteAssert.that() + .with(ZIPS) + .query(sql) + .returnsUnordered("state=AK; id=99503", + "state=AK; id=99504", + "state=AK; id=99505") + .queryContains( + elasticsearchChecker( + "\"fields\" : [\"state\", \"id\"], \"script_fields\": {}", + "\"sort\": [ {\"state\": \"asc\"}, {\"id\": \"asc\"}]", + "\"from\": 2", + "\"size\": 3")); + } + + @Test public void testOffsetLimit() { + final String sql = "select \"state\", \"id\" from zips\n" + + "offset 2 fetch next 3 rows only"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .runs() + .queryContains( + elasticsearchChecker( + "\"from\": 2", + "\"size\": 3", + "\"fields\" : [\"state\", \"id\"], \"script_fields\": {}")); + } + + @Test public void testLimit() { + final String sql = "select \"state\", \"id\" from zips\n" + + "fetch next 3 rows only"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .runs() + .queryContains( + elasticsearchChecker( + "\"size\": 3", + "\"fields\" : [\"state\", \"id\"], \"script_fields\": {}")); + } + + @Test public void testFilterSort() { + final String sql = "select * from zips\n" + + "where \"city\" = 'SPRINGFIELD' and \"id\" >= '70000'\n" + + "order by \"state\", \"id\""; + final String explain = "PLAN=ElasticsearchToEnumerableConverter\n" + + " ElasticsearchSort(sort0=[$4], sort1=[$5], dir0=[ASC], dir1=[ASC])\n" + + " ElasticsearchProject(city=[CAST(ITEM($0, 'city')):VARCHAR(20) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], longitude=[CAST(ITEM(ITEM($0, 'loc'), 0)):FLOAT], latitude=[CAST(ITEM(ITEM($0, 'loc'), 1)):FLOAT], pop=[CAST(ITEM($0, 'pop')):INTEGER], state=[CAST(ITEM($0, 'state')):VARCHAR(2) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], id=[CAST(ITEM($0, 'id')):VARCHAR(5) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"])\n" + + " ElasticsearchFilter(condition=[AND(=(CAST(ITEM($0, 'city')):VARCHAR(20) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\", 'SPRINGFIELD'), >=(CAST(ITEM($0, 'id')):VARCHAR(5) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\", '70000'))])\n" + + " ElasticsearchTableScan(table=[[elasticsearch_raw, zips]])"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .returnsOrdered( + "city=SPRINGFIELD; longitude=-92.54567; latitude=35.274879; pop=752; state=AR; id=72157", + "city=SPRINGFIELD; longitude=-102.617322; latitude=37.406727; pop=1992; state=CO; id=81073", + "city=SPRINGFIELD; longitude=-90.577479; latitude=30.415738; pop=5597; state=LA; id=70462", + "city=SPRINGFIELD; longitude=-123.015259; latitude=44.06106; pop=32384; state=OR; id=97477", + "city=SPRINGFIELD; longitude=-122.917108; latitude=44.056056; pop=27521; state=OR; id=97478") + .queryContains( + elasticsearchChecker("\"query\" : {\"constant_score\":{\"filter\":{\"bool\":" + + "{\"must\":[{\"term\":{\"city\":\"springfield\"}},{\"range\":{\"id\":{\"gte\":\"70000\"}}}]}}}}", + "\"fields\" : [\"city\", \"pop\", \"state\", \"id\"], \"script_fields\": {\"longitude\":{\"script\":\"_source.loc[0]\"}, \"latitude\":{\"script\":\"_source.loc[1]\"}}", + "\"sort\": [ {\"state\": \"asc\"}, {\"id\": \"asc\"}]")) + .explainContains(explain); + } + + @Test public void testFilterSortDesc() { + final String sql = "select * from zips\n" + + "where \"pop\" BETWEEN 20000 AND 20100\n" + + "order by \"state\" desc, \"pop\""; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .limit(4) + .returnsOrdered( + "city=SHERIDAN; longitude=-106.964795; latitude=44.78486; pop=20025; state=WY; id=82801", + "city=MOUNTLAKE TERRAC; longitude=-122.304036; latitude=47.793061; pop=20059; state=WA; id=98043", + "city=FALMOUTH; longitude=-77.404537; latitude=38.314557; pop=20039; state=VA; id=22405", + "city=FORT WORTH; longitude=-97.318409; latitude=32.725551; pop=20012; state=TX; id=76104"); + } + + @Test public void testFilterRedundant() { + final String sql = "select * from zips\n" + + "where \"state\" > 'CA' and \"state\" < 'AZ' and \"state\" = 'OK'"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .runs() + .queryContains( + elasticsearchChecker("" + + "\"query\" : {\"constant_score\":{\"filter\":{\"bool\":" + + "{\"must\":[{\"term\":{\"state\":\"ok\"}}]}}}}", + "\"fields\" : [\"city\", \"pop\", \"state\", \"id\"], \"script_fields\": {\"longitude\":{\"script\":\"_source.loc[0]\"}, \"latitude\":{\"script\":\"_source.loc[1]\"}}")); + } + + @Test public void testInPlan() { + final String[] searches = { + "\"query\" : {\"constant_score\":{\"filter\":{\"bool\":{\"should\":" + + "[{\"bool\":{\"must\":[{\"term\":{\"pop\":20012}}]}},{\"bool\":{\"must\":[{\"term\":" + + "{\"pop\":15590}}]}}]}}}}", + "\"fields\" : [\"city\", \"pop\", \"state\", \"id\"], \"script_fields\": {\"longitude\":{\"script\":\"_source.loc[0]\"}, \"latitude\":{\"script\":\"_source.loc[1]\"}}" + }; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select * from zips where \"pop\" in (20012, 15590)") + .returnsUnordered( + "city=COVINA; longitude=-117.884285; latitude=34.08596; pop=15590; state=CA; id=91723", + "city=ARLINGTON; longitude=-97.091987; latitude=32.654752; pop=15590; state=TX; id=76018", + "city=CROFTON; longitude=-76.680166; latitude=39.011163; pop=15590; state=MD; id=21114", + "city=FORT WORTH; longitude=-97.318409; latitude=32.725551; pop=20012; state=TX; id=76104", + "city=DINUBA; longitude=-119.39087; latitude=36.534931; pop=20012; state=CA; id=93618") + .queryContains(elasticsearchChecker(searches)); + } + + @Test public void testZips() { + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select \"state\", \"city\" from zips") + .returnsCount(10); + } + + @Test public void testProject() { + final String sql = "select \"state\", \"city\", 0 as \"zero\"\n" + + "from zips\n" + + "order by \"state\", \"city\""; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query(sql) + .limit(2) + .returnsUnordered("state=AK; city=ELMENDORF AFB; zero=0", + "state=AK; city=EIELSON AFB; zero=0") + .queryContains( + elasticsearchChecker("\"sort\": [ {\"state\": \"asc\"}, {\"city\": \"asc\"}]", + "\"fields\" : [\"state\", \"city\"], \"script_fields\": {\"zero\":{\"script\": \"0\"}}")); + } + + @Test public void testFilter() { + final String explain = "PLAN=ElasticsearchToEnumerableConverter\n" + + " ElasticsearchProject(state=[CAST(ITEM($0, 'state')):VARCHAR(2) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"], city=[CAST(ITEM($0, 'city')):VARCHAR(20) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\"])\n" + + " ElasticsearchFilter(condition=[=(CAST(ITEM($0, 'state')):VARCHAR(2) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\", 'CA')])\n" + + " ElasticsearchTableScan(table=[[elasticsearch_raw, zips]])"; + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select \"state\", \"city\" from zips where \"state\" = 'CA'") + .limit(2) + .returnsUnordered("state=CA; city=LOS ANGELES", + "state=CA; city=LOS ANGELES") + .explainContains(explain); + } + + @Test public void testFilterReversed() { + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select \"state\", \"city\" from zips where 'WI' < \"state\"") + .limit(2) + .returnsUnordered("state=WV; city=WELCH", + "state=WV; city=HANOVER"); + CalciteAssert.that() + .enable(enabled()) + .with(ZIPS) + .query("select \"state\", \"city\" from zips where \"state\" > 'WI'") + .limit(2) + .returnsUnordered("state=WV; city=WELCH", + "state=WV; city=HANOVER"); + } +} + +// End Elasticsearch5AdapterIT.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch5/src/test/resources/elasticsearch-zips-model.json ---------------------------------------------------------------------- diff --git a/elasticsearch5/src/test/resources/elasticsearch-zips-model.json b/elasticsearch5/src/test/resources/elasticsearch-zips-model.json new file mode 100644 index 0000000..626923e --- /dev/null +++ b/elasticsearch5/src/test/resources/elasticsearch-zips-model.json @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the License); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an AS IS BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +{ + "version": "1.0", + "defaultSchema": "elasticsearch", + "schemas": [ + { + "type": "custom", + "name": "elasticsearch_raw", + "factory": "org.apache.calcite.adapter.elasticsearch5.Elasticsearch5SchemaFactory", + "operand": { + "coordinates": "{'127.0.0.1': 9300}", + "userConfig": "{'bulk.flush.max.actions': 10, 'bulk.flush.max.size.mb': 1}", + "index": "usa" + } + }, + { + "name": "elasticsearch", + "tables": [ + { + "name": "ZIPS", + "type": "view", + "sql": [ + "select cast(_MAP['city'] AS varchar(20)) AS \"city\",\n", + " cast(_MAP['loc'][0] AS float) AS \"longitude\",\n", + " cast(_MAP['loc'][1] AS float) AS \"latitude\",\n", + " cast(_MAP['pop'] AS integer) AS \"pop\",\n", + " cast(_MAP['state'] AS varchar(2)) AS \"state\",\n", + " cast(_MAP['id'] AS varchar(5)) AS \"id\"\n", + "from \"elasticsearch_raw\".\"zips\"" + ] + } + ] + } + ] +} http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch5/src/test/resources/log4j.properties ---------------------------------------------------------------------- diff --git a/elasticsearch5/src/test/resources/log4j.properties b/elasticsearch5/src/test/resources/log4j.properties new file mode 100644 index 0000000..834e2db --- /dev/null +++ b/elasticsearch5/src/test/resources/log4j.properties @@ -0,0 +1,24 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to you under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Root logger is configured at INFO and is sent to A1 +log4j.rootLogger=INFO, A1 + +# A1 goes to the console +log4j.appender.A1=org.apache.log4j.ConsoleAppender + +# Set the pattern for each log message +log4j.appender.A1.layout=org.apache.log4j.PatternLayout +log4j.appender.A1.layout.ConversionPattern=%d [%t] %-5p - %m%n http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/pom.xml ---------------------------------------------------------------------- diff --git a/pom.xml b/pom.xml index 3372f99..a8f4df3 100644 --- a/pom.xml +++ b/pom.xml @@ -61,6 +61,7 @@ limitations under the License. <commons-lang3.version>3.2</commons-lang3.version> <commons-logging.version>1.1.3</commons-logging.version> <elasticsearch-java-driver.version>2.3.2</elasticsearch-java-driver.version> + <elasticsearch5-java-driver.version>5.5.2</elasticsearch5-java-driver.version> <findbugs.version>3.0.1</findbugs.version> <fmpp-maven-plugin.version>1.0</fmpp-maven-plugin.version> <foodmart-data-hsqldb.version>0.3</foodmart-data-hsqldb.version> @@ -139,7 +140,8 @@ limitations under the License. <module>cassandra</module> <module>core</module> <module>druid</module> - <module>elasticsearch</module> + <module>elasticsearch2</module> + <module>elasticsearch5</module> <module>example</module> <module>file</module> <module>linq4j</module> http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/site/_docs/adapter.md ---------------------------------------------------------------------- diff --git a/site/_docs/adapter.md b/site/_docs/adapter.md index ef8e761..3987804 100644 --- a/site/_docs/adapter.md +++ b/site/_docs/adapter.md @@ -30,7 +30,10 @@ presenting the data as tables within a schema. * [Cassandra adapter](cassandra_adapter.html) (<a href="{{ site.apiRoot }}/org/apache/calcite/adapter/cassandra/package-summary.html">calcite-cassandra</a>) * CSV adapter (<a href="{{ site.apiRoot }}/org/apache/calcite/adapter/csv/package-summary.html">example/csv</a>) * [Druid adapter](druid_adapter.html) (<a href="{{ site.apiRoot }}/org/apache/calcite/adapter/druid/package-summary.html">calcite-druid</a>) -* [Elasticsearch adapter](elasticsearch_adapter.html) (<a href="{{ site.apiRoot }}/org/apache/calcite/adapter/elasticsearch/package-summary.html">calcite-elasticsearch</a>) +* [Elasticsearch adapter](elasticsearch_adapter.html) + (<a href="{{ site.apiRoot }}/org/apache/calcite/adapter/elasticsearch2/package-summary.html">calcite-elasticsearch2</a> + and + <a href="{{ site.apiRoot }}/org/apache/calcite/adapter/elasticsearch5/package-summary.html">calcite-elasticsearch5</a>) * [File adapter](file_adapter.html) (<a href="{{ site.apiRoot }}/org/apache/calcite/adapter/file/package-summary.html">calcite-file</a>) * JDBC adapter (part of <a href="{{ site.apiRoot }}/org/apache/calcite/adapter/jdbc/package-summary.html">calcite-core</a>) * MongoDB adapter (<a href="{{ site.apiRoot }}/org/apache/calcite/adapter/mongodb/package-summary.html">calcite-mongodb</a>) http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/site/_docs/elasticsearch_adapter.md ---------------------------------------------------------------------- diff --git a/site/_docs/elasticsearch_adapter.md b/site/_docs/elasticsearch_adapter.md index d87d9e3..21f2459 100644 --- a/site/_docs/elasticsearch_adapter.md +++ b/site/_docs/elasticsearch_adapter.md @@ -47,7 +47,7 @@ A basic example of a model file is given below: { "type": "custom", "name": "elasticsearch", - "factory": "org.apache.calcite.adapter.elasticsearch.ElasticsearchSchemaFactory", + "factory": "org.apache.calcite.adapter.elasticsearch2.Elasticsearch2SchemaFactory", "operand": { "coordinates": "{'127.0.0.1': 9300}", "userConfig": "{'bulk.flush.max.actions': 10, 'bulk.flush.max.size.mb': 1}", @@ -58,6 +58,9 @@ A basic example of a model file is given below: } {% endhighlight %} +This adapter is targeted for Elasticsearch 2.x. To use Calcite with Elasticsearch 5.x+ you can use the factory +of the adapter targeted for Elasticsearch 5.x: `org.apache.calcite.adapter.elasticsearch5.Elasticsearch5SchemaFactory` + Assuming this file is stored as `model.json`, you can connect to Elasticsearch via [`sqlline`](https://github.com/julianhyde/sqlline) as follows: http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/sqlline ---------------------------------------------------------------------- diff --git a/sqlline b/sqlline index 6e4b4e9..b0fc489 100755 --- a/sqlline +++ b/sqlline @@ -37,7 +37,7 @@ if [ ! -f target/fullclasspath.txt ]; then fi CP= -for module in core cassandra druid elasticsearch file mongodb spark splunk example/csv example/function; do +for module in core cassandra druid elasticsearch2 elasticsearch5 file mongodb spark splunk example/csv example/function; do CP=${CP}${module}/target/classes: CP=${CP}${module}/target/test-classes: done http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/sqlline.bat ---------------------------------------------------------------------- diff --git a/sqlline.bat b/sqlline.bat index 99548de..53f7159 100644 --- a/sqlline.bat +++ b/sqlline.bat @@ -23,6 +23,6 @@ :: Copy dependency jars on first call. (To force jar refresh, remove target\dependencies) if not exist target\dependencies (call mvn -B dependency:copy-dependencies -DoverWriteReleases=false -DoverWriteSnapshots=false -DoverWriteIfNewer=true -DoutputDirectory=target\dependencies) -java -Xmx1G -cp ".\target\dependencies\*;core\target\dependencies\*;cassandra\target\dependencies\*;druid\target\dependencies\*;elasticsearch\target\dependencies\*;file\target\dependencies\*;mongodb\target\dependencies\*;spark\target\dependencies\*;splunk\target\dependencies\*" sqlline.SqlLine --verbose=true %* +java -Xmx1G -cp ".\target\dependencies\*;core\target\dependencies\*;cassandra\target\dependencies\*;druid\target\dependencies\*;elasticsearch2\target\dependencies\*;elasticsearch5\target\dependencies\*;file\target\dependencies\*;mongodb\target\dependencies\*;spark\target\dependencies\*;splunk\target\dependencies\*" sqlline.SqlLine --verbose=true %* :: End sqlline.bat
