http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchProject.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchProject.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchProject.java deleted file mode 100644 index 660e268..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchProject.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * 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.elasticsearch; - -import org.apache.calcite.adapter.java.JavaTypeFactory; -import org.apache.calcite.plan.RelOptCluster; -import org.apache.calcite.plan.RelOptCost; -import org.apache.calcite.plan.RelOptPlanner; -import org.apache.calcite.plan.RelTraitSet; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.core.Project; -import org.apache.calcite.rel.metadata.RelMetadataQuery; -import org.apache.calcite.rel.type.RelDataType; -import org.apache.calcite.rex.RexNode; -import org.apache.calcite.util.Pair; -import org.apache.calcite.util.Util; - -import java.util.ArrayList; -import java.util.List; - -/** - * Implementation of {@link org.apache.calcite.rel.core.Project} - * relational expression in Elasticsearch. - */ -public class ElasticsearchProject extends Project implements ElasticsearchRel { - public ElasticsearchProject(RelOptCluster cluster, RelTraitSet traitSet, RelNode input, - List<? extends RexNode> projects, RelDataType rowType) { - super(cluster, traitSet, input, projects, rowType); - assert getConvention() == ElasticsearchRel.CONVENTION; - assert getConvention() == input.getConvention(); - } - - @Override public Project copy(RelTraitSet relTraitSet, RelNode input, List<RexNode> projects, - RelDataType relDataType) { - return new ElasticsearchProject(getCluster(), traitSet, input, projects, relDataType); - } - - @Override public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) { - return super.computeSelfCost(planner, mq).multiplyBy(0.1); - } - - @Override public void implement(Implementor implementor) { - implementor.visitChild(0, getInput()); - - final ElasticsearchRules.RexToElasticsearchTranslator translator = - new ElasticsearchRules.RexToElasticsearchTranslator( - (JavaTypeFactory) getCluster().getTypeFactory(), - ElasticsearchRules.elasticsearchFieldNames(getInput().getRowType())); - - final List<String> findItems = new ArrayList<>(); - final List<String> scriptFieldItems = new ArrayList<>(); - for (Pair<RexNode, String> pair: getNamedProjects()) { - final String name = pair.right; - final String expr = pair.left.accept(translator); - - if (expr.equals("\"" + name + "\"")) { - findItems.add(ElasticsearchRules.quote(name)); - } else if (expr.matches("\"literal\":.+")) { - scriptFieldItems.add(ElasticsearchRules.quote(name) + ":{\"script\": " - + expr.split(":")[1] + "}"); - } else { - scriptFieldItems.add(ElasticsearchRules.quote(name) + ":{\"script\":\"_source." - + expr.replaceAll("\"", "") + "\"}"); - } - } - final String findString = Util.toString(findItems, "", ", ", ""); - final String scriptFieldString = "\"script_fields\": {" - + Util.toString(scriptFieldItems, "", ", ", "") + "}"; - final String fieldString = "\"fields\" : [" + findString + "]" - + ", " + scriptFieldString; - - for (String opfield : implementor.list) { - if (opfield.startsWith("\"fields\"")) { - implementor.list.remove(opfield); - } - } - implementor.add(fieldString); - } -} - -// End ElasticsearchProject.java
http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchRel.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchRel.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchRel.java deleted file mode 100644 index e24cb0d..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchRel.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * 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.elasticsearch; - -import org.apache.calcite.plan.Convention; -import org.apache.calcite.plan.RelOptTable; -import org.apache.calcite.rel.RelNode; - -import java.util.ArrayList; -import java.util.List; - -/** - * Relational expression that uses Elasticsearch calling convention. - */ -public interface ElasticsearchRel extends RelNode { - void implement(Implementor implementor); - - /** - * Calling convention for relational operations that occur in Elasticsearch. - */ - Convention CONVENTION = new Convention.Impl("ELASTICSEARCH", ElasticsearchRel.class); - - /** - * Callback for the implementation process that converts a tree of - * {@link ElasticsearchRel} nodes into an Elasticsearch query. - */ - class Implementor { - final List<String> list = new ArrayList<>(); - - RelOptTable table; - ElasticsearchTable elasticsearchTable; - - public void add(String findOp) { - list.add(findOp); - } - - public void visitChild(int ordinal, RelNode input) { - assert ordinal == 0; - ((ElasticsearchRel) input).implement(this); - } - } -} - -// End ElasticsearchRel.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchRules.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchRules.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchRules.java deleted file mode 100644 index ba2978d..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchRules.java +++ /dev/null @@ -1,236 +0,0 @@ -/* - * 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.elasticsearch; - -import org.apache.calcite.adapter.enumerable.RexImpTable; -import org.apache.calcite.adapter.enumerable.RexToLixTranslator; -import org.apache.calcite.adapter.java.JavaTypeFactory; -import org.apache.calcite.plan.Convention; -import org.apache.calcite.plan.RelOptRule; -import org.apache.calcite.plan.RelTrait; -import org.apache.calcite.plan.RelTraitSet; -import org.apache.calcite.rel.RelCollations; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.convert.ConverterRule; -import org.apache.calcite.rel.core.Sort; -import org.apache.calcite.rel.logical.LogicalFilter; -import org.apache.calcite.rel.logical.LogicalProject; -import org.apache.calcite.rel.type.RelDataType; -import org.apache.calcite.rex.RexCall; -import org.apache.calcite.rex.RexInputRef; -import org.apache.calcite.rex.RexLiteral; -import org.apache.calcite.rex.RexNode; -import org.apache.calcite.rex.RexVisitorImpl; -import org.apache.calcite.sql.SqlKind; -import org.apache.calcite.sql.fun.SqlStdOperatorTable; -import org.apache.calcite.sql.type.SqlTypeName; -import org.apache.calcite.sql.validate.SqlValidatorUtil; - -import java.util.AbstractList; -import java.util.ArrayList; -import java.util.List; - -/** - * Rules and relational operators for - * {@link ElasticsearchRel#CONVENTION ELASTICSEARCH} - * calling convention. - */ -class ElasticsearchRules { - static final RelOptRule[] RULES = { - ElasticsearchSortRule.INSTANCE, - ElasticsearchFilterRule.INSTANCE, - ElasticsearchProjectRule.INSTANCE - }; - - private ElasticsearchRules() {} - - /** - * Returns 'string' if it is a call to item['string'], null otherwise. - */ - static String isItem(RexCall call) { - if (call.getOperator() != SqlStdOperatorTable.ITEM) { - return null; - } - final RexNode op0 = call.getOperands().get(0); - final RexNode op1 = call.getOperands().get(1); - - if (op0 instanceof RexInputRef - && ((RexInputRef) op0).getIndex() == 0 - && op1 instanceof RexLiteral - && ((RexLiteral) op1).getValue2() instanceof String) { - return (String) ((RexLiteral) op1).getValue2(); - } - return null; - } - - static List<String> elasticsearchFieldNames(final RelDataType rowType) { - return SqlValidatorUtil.uniquify( - new AbstractList<String>() { - @Override public String get(int index) { - final String name = rowType.getFieldList().get(index).getName(); - return name.startsWith("$") ? "_" + name.substring(2) : name; - } - - @Override public int size() { - return rowType.getFieldCount(); - } - }, - SqlValidatorUtil.EXPR_SUGGESTER, true); - } - - static String quote(String s) { - return "\"" + s + "\""; - } - - /** - * Translator from {@link RexNode} to strings in Elasticsearch's expression - * language. - */ - static class RexToElasticsearchTranslator extends RexVisitorImpl<String> { - private final JavaTypeFactory typeFactory; - private final List<String> inFields; - - RexToElasticsearchTranslator(JavaTypeFactory typeFactory, List<String> inFields) { - super(true); - this.typeFactory = typeFactory; - this.inFields = inFields; - } - - @Override public String visitLiteral(RexLiteral literal) { - if (literal.getValue() == null) { - return "null"; - } - return "\"literal\":\"" - + RexToLixTranslator.translateLiteral(literal, literal.getType(), - typeFactory, RexImpTable.NullAs.NOT_POSSIBLE) - + "\""; - } - - @Override public String visitInputRef(RexInputRef inputRef) { - return quote(inFields.get(inputRef.getIndex())); - } - - @Override public String visitCall(RexCall call) { - final String name = isItem(call); - if (name != null) { - return "\"" + name + "\""; - } - - final List<String> strings = visitList(call.operands); - if (call.getKind() == SqlKind.CAST) { - return strings.get(0).startsWith("$") ? strings.get(0).substring(1) : strings.get(0); - } - if (call.getOperator() == SqlStdOperatorTable.ITEM) { - final RexNode op1 = call.getOperands().get(1); - if (op1 instanceof RexLiteral && op1.getType().getSqlTypeName() == SqlTypeName.INTEGER) { - return stripQuotes(strings.get(0)) + "[" + ((RexLiteral) op1).getValue2() + "]"; - } - } - throw new IllegalArgumentException("Translation of " + call.toString() - + "is not supported by ElasticsearchProject"); - } - - private String stripQuotes(String s) { - return s.startsWith("'") && s.endsWith("'") ? s.substring(1, s.length() - 1) : s; - } - - List<String> visitList(List<RexNode> list) { - final List<String> strings = new ArrayList<>(); - for (RexNode node: list) { - strings.add(node.accept(this)); - } - return strings; - } - } - - /** - * Base class for planner rules that convert a relational expression to - * Elasticsearch calling convention. - */ - abstract static class ElasticsearchConverterRule extends ConverterRule { - final Convention out; - - ElasticsearchConverterRule(Class<? extends RelNode> clazz, RelTrait in, Convention out, - String description) { - super(clazz, in, out, description); - this.out = out; - } - } - - /** - * Rule to convert a {@link org.apache.calcite.rel.core.Sort} to an - * {@link ElasticsearchSort}. - */ - private static class ElasticsearchSortRule extends ElasticsearchConverterRule { - private static final ElasticsearchSortRule INSTANCE = new ElasticsearchSortRule(); - - private ElasticsearchSortRule() { - super(Sort.class, Convention.NONE, ElasticsearchRel.CONVENTION, "ElasticsearchSortRule"); - } - - @Override public RelNode convert(RelNode relNode) { - final Sort sort = (Sort) relNode; - final RelTraitSet traitSet = sort.getTraitSet().replace(out).replace(sort.getCollation()); - return new ElasticsearchSort(relNode.getCluster(), traitSet, - convert(sort.getInput(), traitSet.replace(RelCollations.EMPTY)), sort.getCollation(), - sort.offset, sort.fetch); - } - } - - /** - * Rule to convert a {@link org.apache.calcite.rel.logical.LogicalFilter} to an - * {@link ElasticsearchFilter}. - */ - private static class ElasticsearchFilterRule extends ElasticsearchConverterRule { - private static final ElasticsearchFilterRule INSTANCE = new ElasticsearchFilterRule(); - - private ElasticsearchFilterRule() { - super(LogicalFilter.class, Convention.NONE, ElasticsearchRel.CONVENTION, - "ElasticsearchFilterRule"); - } - - @Override public RelNode convert(RelNode relNode) { - final LogicalFilter filter = (LogicalFilter) relNode; - final RelTraitSet traitSet = filter.getTraitSet().replace(out); - return new ElasticsearchFilter(relNode.getCluster(), traitSet, - convert(filter.getInput(), out), - filter.getCondition()); - } - } - - /** - * Rule to convert a {@link org.apache.calcite.rel.logical.LogicalProject} - * to an {@link ElasticsearchProject}. - */ - private static class ElasticsearchProjectRule extends ElasticsearchConverterRule { - private static final ElasticsearchProjectRule INSTANCE = new ElasticsearchProjectRule(); - - private ElasticsearchProjectRule() { - super(LogicalProject.class, Convention.NONE, ElasticsearchRel.CONVENTION, - "ElasticsearchProjectRule"); - } - - @Override public RelNode convert(RelNode relNode) { - final LogicalProject project = (LogicalProject) relNode; - final RelTraitSet traitSet = project.getTraitSet().replace(out); - return new ElasticsearchProject(project.getCluster(), traitSet, - convert(project.getInput(), out), project.getProjects(), project.getRowType()); - } - } -} - -// End ElasticsearchRules.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchema.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchema.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchema.java deleted file mode 100644 index 960b387..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchema.java +++ /dev/null @@ -1,127 +0,0 @@ -/* - * 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.elasticsearch; - -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 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 ElasticsearchSchema extends AbstractSchema { - final String index; - - private transient Client client; - - /** - * Creates an Elasticsearch 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". - */ - ElasticsearchSchema(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 { - 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 ElasticsearchTable(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.settingsBuilder().put(userConfig).build(); - - final TransportClient transportClient = TransportClient.builder().settings(settings).build(); - 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; - } -} - -// End ElasticsearchSchema.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchemaFactory.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchemaFactory.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchemaFactory.java deleted file mode 100644 index 41ffc10..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSchemaFactory.java +++ /dev/null @@ -1,63 +0,0 @@ -/* - * 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.elasticsearch; - -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 a {@link ElasticsearchSchema}. - * - * <p>Allows a custom schema to be included in a model.json file. - */ -@SuppressWarnings("UnusedDeclaration") -public class ElasticsearchSchemaFactory implements SchemaFactory { - - public ElasticsearchSchemaFactory() { - } - - @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 ElasticsearchSchema(coordinates, userConfig, index); - } catch (IOException e) { - throw new RuntimeException("Cannot parse values from json", e); - } - } -} - -// End ElasticsearchSchemaFactory.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSort.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSort.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSort.java deleted file mode 100644 index 5f5dfe8..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchSort.java +++ /dev/null @@ -1,93 +0,0 @@ -/* - * 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.elasticsearch; - -import org.apache.calcite.plan.RelOptCluster; -import org.apache.calcite.plan.RelOptCost; -import org.apache.calcite.plan.RelOptPlanner; -import org.apache.calcite.plan.RelTraitSet; -import org.apache.calcite.rel.RelCollation; -import org.apache.calcite.rel.RelFieldCollation; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.core.Sort; -import org.apache.calcite.rel.metadata.RelMetadataQuery; -import org.apache.calcite.rel.type.RelDataTypeField; -import org.apache.calcite.rex.RexLiteral; -import org.apache.calcite.rex.RexNode; -import org.apache.calcite.util.Util; - -import java.util.ArrayList; -import java.util.List; - -/** - * Implementation of {@link org.apache.calcite.rel.core.Sort} - * relational expression in Elasticsearch. - */ -public class ElasticsearchSort extends Sort implements ElasticsearchRel { - public ElasticsearchSort(RelOptCluster cluster, RelTraitSet traitSet, RelNode child, - RelCollation collation, RexNode offset, RexNode fetch) { - super(cluster, traitSet, child, collation, offset, fetch); - assert getConvention() == ElasticsearchRel.CONVENTION; - assert getConvention() == child.getConvention(); - } - - @Override public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) { - return super.computeSelfCost(planner, mq).multiplyBy(0.05); - } - - @Override public Sort copy(RelTraitSet traitSet, RelNode relNode, RelCollation relCollation, - RexNode offset, RexNode fetch) { - return new ElasticsearchSort(getCluster(), traitSet, relNode, collation, offset, fetch); - } - - @Override public void implement(Implementor implementor) { - implementor.visitChild(0, getInput()); - if (!collation.getFieldCollations().isEmpty()) { - final List<String> keys = new ArrayList<>(); - final List<RelDataTypeField> fields = getRowType().getFieldList(); - - for (RelFieldCollation fieldCollation: collation.getFieldCollations()) { - final String name = fields.get(fieldCollation.getFieldIndex()).getName(); - keys.add(ElasticsearchRules.quote(name) + ": " + direction(fieldCollation)); - } - - implementor.add("\"sort\": [ " + Util.toString(keys, "{", "}, {", "}") + "]"); - } - - if (offset != null) { - implementor.add("\"from\": " + ((RexLiteral) offset).getValue()); - } - - if (fetch != null) { - implementor.add("\"size\": " + ((RexLiteral) fetch).getValue()); - } - } - - private String direction(RelFieldCollation fieldCollation) { - switch (fieldCollation.getDirection()) { - case DESCENDING: - case STRICTLY_DESCENDING: - return "\"desc\""; - case ASCENDING: - case STRICTLY_ASCENDING: - default: - return "\"asc\""; - } - } -} - -// End ElasticsearchSort.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchTable.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchTable.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchTable.java deleted file mode 100644 index 5e9043c..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchTable.java +++ /dev/null @@ -1,152 +0,0 @@ -/* - * 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.elasticsearch; - -import org.apache.calcite.adapter.java.AbstractQueryableTable; -import org.apache.calcite.linq4j.AbstractEnumerable; -import org.apache.calcite.linq4j.Enumerable; -import org.apache.calcite.linq4j.Enumerator; -import org.apache.calcite.linq4j.QueryProvider; -import org.apache.calcite.linq4j.Queryable; -import org.apache.calcite.linq4j.function.Function1; -import org.apache.calcite.plan.RelOptCluster; -import org.apache.calcite.plan.RelOptTable; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.type.RelDataType; -import org.apache.calcite.rel.type.RelDataTypeFactory; -import org.apache.calcite.schema.SchemaPlus; -import org.apache.calcite.schema.TranslatableTable; -import org.apache.calcite.schema.impl.AbstractTableQueryable; -import org.apache.calcite.sql.type.SqlTypeName; - -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 Elasticsearch type. - */ -public class ElasticsearchTable extends AbstractQueryableTable implements TranslatableTable { - private final Client client; - private final String indexName; - private final String typeName; - - /** - * Creates an ElasticsearchTable. - */ - public ElasticsearchTable(Client client, String indexName, - String typeName) { - super(Object[].class); - this.client = client; - this.indexName = indexName; - this.typeName = typeName; - } - - @Override public String toString() { - return "ElasticsearchTable{" + typeName + "}"; - } - - public RelDataType getRowType(RelDataTypeFactory relDataTypeFactory) { - final RelDataType mapType = relDataTypeFactory.createMapType( - relDataTypeFactory.createSqlType(SqlTypeName.VARCHAR), - relDataTypeFactory.createTypeWithNullability( - relDataTypeFactory.createSqlType(SqlTypeName.ANY), - true)); - return relDataTypeFactory.builder().add("_MAP", mapType).build(); - } - - public <T> Queryable<T> asQueryable(QueryProvider queryProvider, SchemaPlus schema, - String tableName) { - return new ElasticsearchQueryable<>(queryProvider, schema, this, tableName); - } - - public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable relOptTable) { - final RelOptCluster cluster = context.getCluster(); - return new ElasticsearchTableScan(cluster, cluster.traitSetOf(ElasticsearchRel.CONVENTION), - relOptTable, this, null); - } - - /** Executes a "find" operation on the underlying type. - * - * <p>For example, - * <code>client.prepareSearch(index).setTypes(type) - * .setSource("{\"fields\" : [\"state\"]}")</code></p> - * - * @param index Elasticsearch index - * @param ops List of operations represented as Json strings. - * @param fields List of fields to project; or null to return map - * @return Enumerator of results - */ - private 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 = ElasticsearchEnumerator.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 ElasticsearchEnumerator(cursor, getter); - } - }; - } - - /** - * Implementation of {@link org.apache.calcite.linq4j.Queryable} based on - * a {@link org.apache.calcite.adapter.elasticsearch.ElasticsearchTable}. - * - * @param <T> element type - */ - public static class ElasticsearchQueryable<T> extends AbstractTableQueryable<T> { - public ElasticsearchQueryable(QueryProvider queryProvider, SchemaPlus schema, - ElasticsearchTable table, String tableName) { - super(queryProvider, schema, table, tableName); - } - - public Enumerator<T> enumerator() { - return null; - } - - private String getIndex() { - return schema.unwrap(ElasticsearchSchema.class).index; - } - - private ElasticsearchTable getTable() { - return (ElasticsearchTable) table; - } - - /** Called via code-generation. - * - * @see org.apache.calcite.adapter.elasticsearch.ElasticsearchMethod#ELASTICSEARCH_QUERYABLE_FIND - */ - @SuppressWarnings("UnusedDeclaration") - public Enumerable<Object> find(List<String> ops, - List<Map.Entry<String, Class>> fields) { - return getTable().find(getIndex(), ops, fields); - } - } -} - -// End ElasticsearchTable.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchTableScan.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchTableScan.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchTableScan.java deleted file mode 100644 index 636a629..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchTableScan.java +++ /dev/null @@ -1,88 +0,0 @@ -/* - * 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.elasticsearch; - -import org.apache.calcite.plan.RelOptCluster; -import org.apache.calcite.plan.RelOptCost; -import org.apache.calcite.plan.RelOptPlanner; -import org.apache.calcite.plan.RelOptRule; -import org.apache.calcite.plan.RelOptTable; -import org.apache.calcite.plan.RelTraitSet; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.core.TableScan; -import org.apache.calcite.rel.metadata.RelMetadataQuery; -import org.apache.calcite.rel.type.RelDataType; - -import java.util.List; - -/** - * Relational expression representing a scan of an Elasticsearch type. - * - * <p> Additional operations might be applied, - * using the "find" method.</p> - */ -public class ElasticsearchTableScan extends TableScan implements ElasticsearchRel { - private final ElasticsearchTable elasticsearchTable; - private final RelDataType projectRowType; - - /** - * Creates an ElasticsearchTableScan. - * - * @param cluster Cluster - * @param traitSet Trait set - * @param table Table - * @param elasticsearchTable Elasticsearch table - * @param projectRowType Fields and types to project; null to project raw row - */ - protected ElasticsearchTableScan(RelOptCluster cluster, RelTraitSet traitSet, RelOptTable table, - ElasticsearchTable elasticsearchTable, RelDataType projectRowType) { - super(cluster, traitSet, table); - this.elasticsearchTable = elasticsearchTable; - this.projectRowType = projectRowType; - - assert elasticsearchTable != null; - assert getConvention() == ElasticsearchRel.CONVENTION; - } - - @Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) { - assert inputs.isEmpty(); - return this; - } - - @Override public RelDataType deriveRowType() { - return projectRowType != null ? projectRowType : super.deriveRowType(); - } - - @Override public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) { - final float f = projectRowType == null ? 1f : (float) projectRowType.getFieldCount() / 100f; - return super.computeSelfCost(planner, mq).multiplyBy(.1 * f); - } - - @Override public void register(RelOptPlanner planner) { - planner.addRule(ElasticsearchToEnumerableConverterRule.INSTANCE); - for (RelOptRule rule: ElasticsearchRules.RULES) { - planner.addRule(rule); - } - } - - @Override public void implement(Implementor implementor) { - implementor.elasticsearchTable = elasticsearchTable; - implementor.table = table; - } -} - -// End ElasticsearchTableScan.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchToEnumerableConverter.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchToEnumerableConverter.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchToEnumerableConverter.java deleted file mode 100644 index adb88f7..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchToEnumerableConverter.java +++ /dev/null @@ -1,124 +0,0 @@ -/* - * 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.elasticsearch; - -import org.apache.calcite.adapter.enumerable.EnumerableRel; -import org.apache.calcite.adapter.enumerable.EnumerableRelImplementor; -import org.apache.calcite.adapter.enumerable.JavaRowFormat; -import org.apache.calcite.adapter.enumerable.PhysType; -import org.apache.calcite.adapter.enumerable.PhysTypeImpl; - -import org.apache.calcite.linq4j.tree.BlockBuilder; -import org.apache.calcite.linq4j.tree.Expression; -import org.apache.calcite.linq4j.tree.Expressions; -import org.apache.calcite.linq4j.tree.MethodCallExpression; -import org.apache.calcite.plan.ConventionTraitDef; -import org.apache.calcite.plan.RelOptCluster; -import org.apache.calcite.plan.RelOptCost; -import org.apache.calcite.plan.RelOptPlanner; -import org.apache.calcite.plan.RelTraitSet; -import org.apache.calcite.prepare.CalcitePrepareImpl; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.convert.ConverterImpl; -import org.apache.calcite.rel.metadata.RelMetadataQuery; -import org.apache.calcite.rel.type.RelDataType; -import org.apache.calcite.runtime.Hook; -import org.apache.calcite.util.BuiltInMethod; -import org.apache.calcite.util.Pair; - -import com.google.common.base.Function; -import com.google.common.collect.Lists; - -import java.util.AbstractList; -import java.util.List; -import javax.annotation.Nullable; - -/** - * Relational expression representing a scan of a table in an Elasticsearch data source. - */ -public class ElasticsearchToEnumerableConverter extends ConverterImpl implements EnumerableRel { - protected ElasticsearchToEnumerableConverter(RelOptCluster cluster, RelTraitSet traits, - RelNode input) { - super(cluster, ConventionTraitDef.INSTANCE, traits, input); - } - - @Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) { - return new ElasticsearchToEnumerableConverter(getCluster(), traitSet, sole(inputs)); - } - - @Override public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) { - return super.computeSelfCost(planner, mq).multiplyBy(.1); - } - - @Override public Result implement(EnumerableRelImplementor implementor, Prefer prefer) { - final BlockBuilder list = new BlockBuilder(); - final ElasticsearchRel.Implementor elasticsearchImplementor = - new ElasticsearchRel.Implementor(); - elasticsearchImplementor.visitChild(0, getInput()); - final RelDataType rowType = getRowType(); - final PhysType physType = PhysTypeImpl.of(implementor.getTypeFactory(), rowType, - prefer.prefer(JavaRowFormat.ARRAY)); - final Expression fields = list.append("fields", - constantArrayList( - Pair.zip(ElasticsearchRules.elasticsearchFieldNames(rowType), - new AbstractList<Class>() { - @Override public Class get(int index) { - return physType.fieldClass(index); - } - - @Override public int size() { - return rowType.getFieldCount(); - } - }), - Pair.class)); - final Expression table = list.append("table", - elasticsearchImplementor.table - .getExpression(ElasticsearchTable.ElasticsearchQueryable.class)); - List<String> opList = elasticsearchImplementor.list; - final Expression ops = list.append("ops", constantArrayList(opList, String.class)); - Expression enumerable = list.append("enumerable", - Expressions.call(table, ElasticsearchMethod.ELASTICSEARCH_QUERYABLE_FIND.method, ops, - fields)); - if (CalcitePrepareImpl.DEBUG) { - System.out.println("Elasticsearch: " + opList); - } - Hook.QUERY_PLAN.run(opList); - list.add(Expressions.return_(null, enumerable)); - return implementor.result(physType, list.toBlock()); - } - - /** E.g. {@code constantArrayList("x", "y")} returns - * "Arrays.asList('x', 'y')". */ - private static <T> MethodCallExpression constantArrayList(List<T> values, Class clazz) { - return Expressions.call(BuiltInMethod.ARRAYS_AS_LIST.method, - Expressions.newArrayInit(clazz, constantList(values))); - } - - /** E.g. {@code constantList("x", "y")} returns - * {@code {ConstantExpression("x"), ConstantExpression("y")}}. */ - private static <T> List<Expression> constantList(List<T> values) { - return Lists.transform(values, - new Function<T, Expression>() { - @Nullable - @Override public Expression apply(@Nullable T t) { - return Expressions.constant(t); - } - }); - } -} - -// End ElasticsearchToEnumerableConverter.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchToEnumerableConverterRule.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchToEnumerableConverterRule.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchToEnumerableConverterRule.java deleted file mode 100644 index 1047757..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/ElasticsearchToEnumerableConverterRule.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * 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.elasticsearch; - -import org.apache.calcite.adapter.enumerable.EnumerableConvention; -import org.apache.calcite.plan.RelTraitSet; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.convert.ConverterRule; - -/** - * Rule to convert a relational expression from - * {@link ElasticsearchRel#CONVENTION} to {@link EnumerableConvention}. - */ -public class ElasticsearchToEnumerableConverterRule extends ConverterRule { - public static final ConverterRule INSTANCE = new ElasticsearchToEnumerableConverterRule(); - - private ElasticsearchToEnumerableConverterRule() { - super(RelNode.class, ElasticsearchRel.CONVENTION, EnumerableConvention.INSTANCE, - "ElasticsearchToEnumerableConverterRule"); - } - - @Override public RelNode convert(RelNode relNode) { - RelTraitSet newTraitSet = relNode.getTraitSet().replace(getOutConvention()); - return new ElasticsearchToEnumerableConverter(relNode.getCluster(), newTraitSet, relNode); - } -} - -// End ElasticsearchToEnumerableConverterRule.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/package-info.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/package-info.java b/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/package-info.java deleted file mode 100644 index dad800a..0000000 --- a/elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/package-info.java +++ /dev/null @@ -1,26 +0,0 @@ -/* - * 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 Elasticsearch DB. - */ -@PackageMarker -package org.apache.calcite.adapter.elasticsearch; - -import org.apache.calcite.avatica.util.PackageMarker; - -// End package-info.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/test/java/org/apache/calcite/test/ElasticsearchAdapterIT.java ---------------------------------------------------------------------- diff --git a/elasticsearch/src/test/java/org/apache/calcite/test/ElasticsearchAdapterIT.java b/elasticsearch/src/test/java/org/apache/calcite/test/ElasticsearchAdapterIT.java deleted file mode 100644 index 89660a5..0000000 --- a/elasticsearch/src/test/java/org/apache/calcite/test/ElasticsearchAdapterIT.java +++ /dev/null @@ -1,270 +0,0 @@ -/* - * 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 ElasticsearchAdapterIT { - /** - * 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", - ElasticsearchAdapterIT.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 ElasticsearchAdapterIT.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch/src/test/resources/elasticsearch-zips-model.json ---------------------------------------------------------------------- diff --git a/elasticsearch/src/test/resources/elasticsearch-zips-model.json b/elasticsearch/src/test/resources/elasticsearch-zips-model.json deleted file mode 100644 index dcbf2a4..0000000 --- a/elasticsearch/src/test/resources/elasticsearch-zips-model.json +++ /dev/null @@ -1,50 +0,0 @@ -/* - * 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.elasticsearch.ElasticsearchSchemaFactory", - "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/elasticsearch/src/test/resources/log4j.properties ---------------------------------------------------------------------- diff --git a/elasticsearch/src/test/resources/log4j.properties b/elasticsearch/src/test/resources/log4j.properties deleted file mode 100644 index 834e2db..0000000 --- a/elasticsearch/src/test/resources/log4j.properties +++ /dev/null @@ -1,24 +0,0 @@ -# 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/elasticsearch2/pom.xml ---------------------------------------------------------------------- diff --git a/elasticsearch2/pom.xml b/elasticsearch2/pom.xml new file mode 100644 index 0000000..6a645ac --- /dev/null +++ b/elasticsearch2/pom.xml @@ -0,0 +1,148 @@ +<?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-elasticsearch2</artifactId> + <packaging>jar</packaging> + <version>1.14.0-SNAPSHOT</version> + <name>Calcite Elasticsearch</name> + <description>Elasticsearch 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</groupId> + <artifactId>elasticsearch</artifactId> + <version>${elasticsearch-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/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Enumerator.java ---------------------------------------------------------------------- diff --git a/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Enumerator.java b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Enumerator.java new file mode 100644 index 0000000..84370ab --- /dev/null +++ b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Enumerator.java @@ -0,0 +1,152 @@ +/* + * 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.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 Elasticsearch2Enumerator implements Enumerator<Object> { + private final Iterator<SearchHit> cursor; + private final Function1<SearchHit, Object> getter; + private Object current; + + /** + * Creates an ElasticsearchEnumerator. + * + * @param cursor Iterator over Elasticsearch {@link SearchHit} objects + * @param getter Converts an object into a list of fields + */ + public Elasticsearch2Enumerator(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.fields(); + } + }; + } + + private static Function1<SearchHit, Object> singletonGetter(final String fieldName, + final Class fieldClass) { + return new Function1<SearchHit, Object>() { + public Object apply(SearchHit searchHitFields) { + if (searchHitFields.fields().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.fields().isEmpty()) { + objects[i] = convert(searchHitFields.getSource().get(name), field.getValue()); + } else { + objects[i] = convert(searchHitFields.field(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 Elasticsearch2Enumerator.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Schema.java ---------------------------------------------------------------------- diff --git a/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Schema.java b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Schema.java new file mode 100644 index 0000000..668402b --- /dev/null +++ b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2Schema.java @@ -0,0 +1,134 @@ +/* + * 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.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 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 Elasticsearch2Schema extends AbstractSchema + implements ElasticsearchSchema { + final String index; + + private transient Client client; + + /** + * Creates an Elasticsearch2 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". + */ + Elasticsearch2Schema(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 { + 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 Elasticsearch2Table(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.settingsBuilder().put(userConfig).build(); + + final TransportClient transportClient = TransportClient.builder().settings(settings).build(); + 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 Elasticsearch2Schema.java http://git-wip-us.apache.org/repos/asf/calcite/blob/e1525926/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2SchemaFactory.java ---------------------------------------------------------------------- diff --git a/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2SchemaFactory.java b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2SchemaFactory.java new file mode 100644 index 0000000..19378a0 --- /dev/null +++ b/elasticsearch2/src/main/java/org/apache/calcite/adapter/elasticsearch2/Elasticsearch2SchemaFactory.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.elasticsearch2; + +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 Elasticsearch2Schema}. + * + * <p>Allows a custom schema to be included in a model.json file. + */ +@SuppressWarnings("UnusedDeclaration") +public class Elasticsearch2SchemaFactory implements SchemaFactory { + + public Elasticsearch2SchemaFactory() { + } + + @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 Elasticsearch2Schema(coordinates, userConfig, index); + } catch (IOException e) { + throw new RuntimeException("Cannot parse values from json", e); + } + } +} + +// End Elasticsearch2SchemaFactory.java
