lucene-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbern...@apache.org
Subject [4/7] lucene-solr:branch_6x: Calcite changes
Date Fri, 03 Mar 2017 16:34:54 GMT
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrFilter.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrFilter.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrFilter.java
new file mode 100644
index 0000000..ce12aec
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrFilter.java
@@ -0,0 +1,382 @@
+/*
+ * 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.solr.handler.sql;
+
+import org.apache.calcite.plan.*;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+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.sql.SqlKind;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Implementation of a {@link org.apache.calcite.rel.core.Filter} relational expression in Solr.
+ */
+class SolrFilter extends Filter implements SolrRel {
+  SolrFilter(
+      RelOptCluster cluster,
+      RelTraitSet traitSet,
+      RelNode child,
+      RexNode condition) {
+    super(cluster, traitSet, child, condition);
+    assert getConvention() == SolrRel.CONVENTION;
+    assert getConvention() == child.getConvention();
+  }
+
+  @Override public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+    return super.computeSelfCost(planner, mq).multiplyBy(0.1);
+  }
+
+  public SolrFilter copy(RelTraitSet traitSet, RelNode input, RexNode condition) {
+    return new SolrFilter(getCluster(), traitSet, input, condition);
+  }
+
+  public void implement(Implementor implementor) {
+    implementor.visitChild(0, getInput());
+    if(getInput() instanceof SolrAggregate) {
+      HavingTranslator translator = new HavingTranslator(SolrRules.solrFieldNames(getRowType()), implementor.reverseAggMappings);
+      String havingPredicate = translator.translateMatch(condition);
+      implementor.setHavingPredicate(havingPredicate);
+    } else {
+      Translator translator = new Translator(SolrRules.solrFieldNames(getRowType()));
+      String query = translator.translateMatch(condition);
+      implementor.addQuery(query);
+      implementor.setNegativeQuery(translator.negativeQuery);
+    }
+  }
+
+  private static class Translator {
+
+    private final List<String> fieldNames;
+    public boolean negativeQuery = true;
+
+    Translator(List<String> fieldNames) {
+      this.fieldNames = fieldNames;
+    }
+
+    private String translateMatch(RexNode condition) {
+      if (condition.getKind().belongsTo(SqlKind.COMPARISON)) {
+        return translateComparison(condition);
+      } else if (condition.isA(SqlKind.AND)) {
+        return "(" + translateAnd(condition) + ")";
+      } else if (condition.isA(SqlKind.OR)) {
+        return "(" + translateOr(condition) + ")";
+      } else {
+        return null;
+      }
+    }
+
+    private String translateOr(RexNode condition) {
+      List<String> ors = new ArrayList<>();
+      for (RexNode node : RelOptUtil.disjunctions(condition)) {
+        ors.add(translateMatch(node));
+      }
+      return String.join(" OR ", ors);
+    }
+
+    private String translateAnd(RexNode node0) {
+      List<String> andStrings = new ArrayList();
+      List<String> notStrings = new ArrayList();
+
+      List<RexNode> ands = new ArrayList();
+      List<RexNode> nots = new ArrayList();
+      RelOptUtil.decomposeConjunction(node0, ands, nots);
+
+
+      for (RexNode node : ands) {
+        andStrings.add(translateMatch(node));
+      }
+
+      String andString = String.join(" AND ", andStrings);
+
+      if (nots.size() > 0) {
+        for (RexNode node : nots) {
+          notStrings.add(translateMatch(node));
+        }
+        String notString = String.join(" NOT ", notStrings);
+        return "(" + andString + ") NOT (" + notString + ")";
+      } else {
+        return andString;
+      }
+    }
+
+    private String translateComparison(RexNode node) {
+      Pair<String, RexLiteral> binaryTranslated = null;
+      if (((RexCall) node).getOperands().size() == 2) {
+        binaryTranslated = translateBinary((RexCall) node);
+      }
+
+      switch (node.getKind()) {
+        case NOT:
+          return "-" + translateComparison(((RexCall) node).getOperands().get(0));
+        case EQUALS:
+          String terms = binaryTranslated.getValue().toString().trim();
+          terms = terms.replace("'","");
+          if (!terms.startsWith("(") && !terms.startsWith("[") && !terms.startsWith("{")) {
+            terms = "\"" + terms + "\"";
+          }
+
+          String clause = binaryTranslated.getKey() + ":" + terms;
+          this.negativeQuery = false;
+          return clause;
+        case NOT_EQUALS:
+          return "-(" + binaryTranslated.getKey() + ":" + binaryTranslated.getValue() + ")";
+        case LESS_THAN:
+          this.negativeQuery = false;
+          return "(" + binaryTranslated.getKey() + ": [ * TO " + binaryTranslated.getValue() + " })";
+        case LESS_THAN_OR_EQUAL:
+          this.negativeQuery = false;
+          return "(" + binaryTranslated.getKey() + ": [ * TO " + binaryTranslated.getValue() + " ])";
+        case GREATER_THAN:
+          this.negativeQuery = false;
+          return "(" + binaryTranslated.getKey() + ": { " + binaryTranslated.getValue() + " TO * ])";
+        case GREATER_THAN_OR_EQUAL:
+          this.negativeQuery = false;
+          return "(" + binaryTranslated.getKey() + ": [ " + binaryTranslated.getValue() + " TO * ])";
+        default:
+          throw new AssertionError("cannot translate " + node);
+      }
+    }
+
+    /**
+     * Translates a call to a binary operator, reversing arguments if necessary.
+     */
+    private Pair<String, RexLiteral> translateBinary(RexCall call) {
+      List<RexNode> operands = call.getOperands();
+      if (operands.size() != 2) {
+        throw new AssertionError("Invalid number of arguments - " + operands.size());
+      }
+      final RexNode left = operands.get(0);
+      final RexNode right = operands.get(1);
+      final Pair<String, RexLiteral> a = translateBinary2(left, right);
+      if (a != null) {
+        return a;
+      }
+      final Pair<String, RexLiteral> b = translateBinary2(right, left);
+      if (b != null) {
+        return b;
+      }
+      throw new AssertionError("cannot translate call " + call);
+    }
+
+    /**
+     * Translates a call to a binary operator. Returns whether successful.
+     */
+    private Pair<String, RexLiteral> translateBinary2(RexNode left, RexNode right) {
+      switch (right.getKind()) {
+        case LITERAL:
+          break;
+        default:
+          return null;
+      }
+      final RexLiteral rightLiteral = (RexLiteral) right;
+      switch (left.getKind()) {
+        case INPUT_REF:
+          final RexInputRef left1 = (RexInputRef) left;
+          String name = fieldNames.get(left1.getIndex());
+          return new Pair<>(name, rightLiteral);
+        case CAST:
+          return translateBinary2(((RexCall) left).operands.get(0), right);
+//        case OTHER_FUNCTION:
+//          String itemName = SolrRules.isItem((RexCall) left);
+//          if (itemName != null) {
+//            return translateOp2(op, itemName, rightLiteral);
+//          }
+        default:
+          return null;
+      }
+    }
+  }
+
+  private static class HavingTranslator {
+
+    private final List<String> fieldNames;
+    private Map<String,String> reverseAggMappings;
+
+    HavingTranslator(List<String> fieldNames, Map<String, String> reverseAggMappings) {
+      this.fieldNames = fieldNames;
+      this.reverseAggMappings = reverseAggMappings;
+    }
+
+    private String translateMatch(RexNode condition) {
+      if (condition.getKind().belongsTo(SqlKind.COMPARISON)) {
+        return translateComparison(condition);
+      } else if (condition.isA(SqlKind.AND)) {
+        return translateAnd(condition);
+      } else if (condition.isA(SqlKind.OR)) {
+        return translateOr(condition);
+      } else {
+        return null;
+      }
+    }
+
+    private String translateOr(RexNode condition) {
+      List<String> ors = new ArrayList<>();
+      for (RexNode node : RelOptUtil.disjunctions(condition)) {
+        ors.add(translateMatch(node));
+      }
+      StringBuilder builder = new StringBuilder();
+
+      builder.append("or(");
+      int i = 0;
+      for (i = 0; i < ors.size(); i++) {
+        if (i > 0) {
+          builder.append(",");
+        }
+
+        builder.append(ors.get(i));
+      }
+      builder.append(")");
+      return builder.toString();
+    }
+
+    private String translateAnd(RexNode node0) {
+      List<String> andStrings = new ArrayList();
+      List<String> notStrings = new ArrayList();
+
+      List<RexNode> ands = new ArrayList();
+      List<RexNode> nots = new ArrayList();
+
+      RelOptUtil.decomposeConjunction(node0, ands, nots);
+
+      for (RexNode node : ands) {
+        andStrings.add(translateMatch(node));
+      }
+
+      StringBuilder builder = new StringBuilder();
+
+      builder.append("and(");
+      for (int i = 0; i < andStrings.size(); i++) {
+        if (i > 0) {
+          builder.append(",");
+        }
+
+        builder.append(andStrings.get(i));
+      }
+      builder.append(")");
+
+
+      if (nots.size() > 0) {
+        for (RexNode node : nots) {
+          notStrings.add(translateMatch(node));
+        }
+
+        StringBuilder notBuilder = new StringBuilder();
+        for(int i=0; i< notStrings.size(); i++) {
+          if(i > 0) {
+            notBuilder.append(",");
+          }
+          notBuilder.append("not(");
+          notBuilder.append(notStrings.get(i));
+          notBuilder.append(")");
+        }
+
+        return "and(" + builder.toString() + ","+ notBuilder.toString()+")";
+      } else {
+        return builder.toString();
+      }
+    }
+
+    private String translateComparison(RexNode node) {
+      Pair<String, RexLiteral> binaryTranslated = null;
+      if (((RexCall) node).getOperands().size() == 2) {
+        binaryTranslated = translateBinary((RexCall) node);
+      }
+
+      switch (node.getKind()) {
+        case EQUALS:
+          String terms = binaryTranslated.getValue().toString().trim();
+          String clause = "eq(" + binaryTranslated.getKey() + "," + terms + ")";
+          return clause;
+        case NOT_EQUALS:
+          return "not(eq(" + binaryTranslated.getKey() + "," + binaryTranslated.getValue() + "))";
+        case LESS_THAN:
+          return "lt(" + binaryTranslated.getKey() + "," + binaryTranslated.getValue() + ")";
+        case LESS_THAN_OR_EQUAL:
+          return "lteq(" + binaryTranslated.getKey() + "," + binaryTranslated.getValue() + ")";
+        case GREATER_THAN:
+          return "gt(" + binaryTranslated.getKey() + "," + binaryTranslated.getValue() + ")";
+        case GREATER_THAN_OR_EQUAL:
+          return "gteq(" + binaryTranslated.getKey() + "," + binaryTranslated.getValue() + ")";
+        default:
+          throw new AssertionError("cannot translate " + node);
+      }
+    }
+
+    /**
+     * Translates a call to a binary operator, reversing arguments if necessary.
+     */
+    private Pair<String, RexLiteral> translateBinary(RexCall call) {
+      List<RexNode> operands = call.getOperands();
+      if (operands.size() != 2) {
+        throw new AssertionError("Invalid number of arguments - " + operands.size());
+      }
+      final RexNode left = operands.get(0);
+      final RexNode right = operands.get(1);
+      final Pair<String, RexLiteral> a = translateBinary2(left, right);
+
+      if (a != null) {
+        if(reverseAggMappings.containsKey(a.getKey())) {
+          return new Pair<String, RexLiteral>(reverseAggMappings.get(a.getKey()),a.getValue());
+        }
+        return a;
+      }
+      final Pair<String, RexLiteral> b = translateBinary2(right, left);
+      if (b != null) {
+        return b;
+      }
+      throw new AssertionError("cannot translate call " + call);
+    }
+
+    /**
+     * Translates a call to a binary operator. Returns whether successful.
+     */
+    private Pair<String, RexLiteral> translateBinary2(RexNode left, RexNode right) {
+      switch (right.getKind()) {
+        case LITERAL:
+          break;
+        default:
+          return null;
+      }
+
+      final RexLiteral rightLiteral = (RexLiteral) right;
+      switch (left.getKind()) {
+        case INPUT_REF:
+          final RexInputRef left1 = (RexInputRef) left;
+          String name = fieldNames.get(left1.getIndex());
+          return new Pair<>(name, rightLiteral);
+        case CAST:
+          return translateBinary2(((RexCall) left).operands.get(0), right);
+//        case OTHER_FUNCTION:
+//          String itemName = SolrRules.isItem((RexCall) left);
+//          if (itemName != null) {
+//            return translateOp2(op, itemName, rightLiteral);
+//          }
+        default:
+          return null;
+      }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrMethod.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrMethod.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrMethod.java
new file mode 100644
index 0000000..b0bf801
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrMethod.java
@@ -0,0 +1,44 @@
+/*
+ * 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.solr.handler.sql;
+
+import org.apache.calcite.linq4j.tree.Types;
+
+import java.lang.reflect.Method;
+import java.util.List;
+
+/**
+ * Builtin methods in the Solr adapter.
+ */
+enum SolrMethod {
+  SOLR_QUERYABLE_QUERY(SolrTable.SolrQueryable.class,
+                       "query",
+                       List.class,
+                       String.class,
+                       List.class,
+                       List.class,
+                       List.class,
+                       String.class,
+                       String.class,
+                       String.class);
+
+  public final Method method;
+
+  SolrMethod(Class clazz, String methodName, Class... argumentTypes) {
+    this.method = Types.lookupMethod(clazz, methodName, argumentTypes);
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrProject.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrProject.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrProject.java
new file mode 100644
index 0000000..c4217f2
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrProject.java
@@ -0,0 +1,64 @@
+/*
+ * 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.solr.handler.sql;
+
+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 java.util.List;
+
+/**
+ * Implementation of {@link org.apache.calcite.rel.core.Project} relational expression in Solr.
+ */
+class SolrProject extends Project implements SolrRel {
+  SolrProject(RelOptCluster cluster, RelTraitSet traitSet,
+              RelNode input, List<? extends RexNode> projects, RelDataType rowType) {
+    super(cluster, traitSet, input, projects, rowType);
+    assert getConvention() == SolrRel.CONVENTION;
+    assert getConvention() == input.getConvention();
+  }
+
+  @Override
+  public Project copy(RelTraitSet traitSet, RelNode input, List<RexNode> projects, RelDataType rowType) {
+    return new SolrProject(getCluster(), traitSet, input, projects, rowType);
+  }
+
+  @Override
+  public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+    return super.computeSelfCost(planner, mq).multiplyBy(0.1);
+  }
+
+  public void implement(Implementor implementor) {
+    implementor.visitChild(0, getInput());
+    final SolrRules.RexToSolrTranslator translator = new SolrRules.RexToSolrTranslator(
+        (JavaTypeFactory) getCluster().getTypeFactory(), SolrRules.solrFieldNames(getInput().getRowType()));
+    for (Pair<RexNode, String> pair : getNamedProjects()) {
+      final String name = pair.right;
+      final String expr = pair.left.accept(translator);
+      implementor.addFieldMapping(name, expr);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrRel.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrRel.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrRel.java
new file mode 100644
index 0000000..557cfe0
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrRel.java
@@ -0,0 +1,105 @@
+/*
+ * 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.solr.handler.sql;
+
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.util.Pair;
+import org.apache.solr.client.solrj.io.ops.BooleanOperation;
+
+import java.util.*;
+
+/**
+ * Relational expression that uses Solr calling convention.
+ */
+interface SolrRel extends RelNode {
+  void implement(Implementor implementor);
+
+  /** Calling convention for relational operations that occur in Solr. */
+  Convention CONVENTION = new Convention.Impl("Solr", SolrRel.class);
+
+  /** Callback for the implementation process that converts a tree of {@link SolrRel} nodes into a Solr query. */
+  class Implementor {
+    final Map<String, String> fieldMappings = new HashMap<>();
+    final Map<String, String> reverseAggMappings = new HashMap<>();
+    String query = null;
+    String havingPredicate;
+    boolean negativeQuery;
+    String limitValue = null;
+    final List<Pair<String, String>> orders = new ArrayList<>();
+    final List<String> buckets = new ArrayList<>();
+    final List<Pair<String, String>> metricPairs = new ArrayList<>();
+
+    RelOptTable table;
+    SolrTable solrTable;
+
+    void addFieldMapping(String key, String val) {
+      if(key != null && !fieldMappings.containsKey(key)) {
+        this.fieldMappings.put(key, val);
+      }
+    }
+
+    void addReverseAggMapping(String key, String val) {
+      if(key != null && !reverseAggMappings.containsKey(key)) {
+        this.reverseAggMappings.put(key, val);
+      }
+    }
+
+    void addQuery(String query) {
+      this.query = query;
+    }
+
+    void setNegativeQuery(boolean negativeQuery) {
+      this.negativeQuery = negativeQuery;
+    }
+
+    void addOrder(String column, String direction) {
+      column = this.fieldMappings.getOrDefault(column, column);
+      this.orders.add(new Pair<>(column, direction));
+    }
+
+    void addBucket(String bucket) {
+      bucket = this.fieldMappings.getOrDefault(bucket, bucket);
+      this.buckets.add(bucket);
+    }
+
+    void addMetricPair(String outName, String metric, String column) {
+      column = this.fieldMappings.getOrDefault(column, column);
+      this.metricPairs.add(new Pair<>(metric, column));
+
+      String metricIdentifier = metric.toLowerCase(Locale.ROOT) + "(" + column + ")";
+      if(outName != null) {
+        this.addFieldMapping(outName, metricIdentifier);
+      }
+    }
+
+    void setHavingPredicate(String havingPredicate) {
+      this.havingPredicate = havingPredicate;
+    }
+
+
+    void setLimit(String limit) {
+      limitValue = limit;
+    }
+
+    void visitChild(int ordinal, RelNode input) {
+      assert ordinal == 0;
+      ((SolrRel) input).implement(this);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrRules.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrRules.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrRules.java
new file mode 100644
index 0000000..4cbadda
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrRules.java
@@ -0,0 +1,234 @@
+/*
+ * 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.solr.handler.sql;
+
+import org.apache.calcite.adapter.java.JavaTypeFactory;
+import org.apache.calcite.plan.*;
+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.LogicalAggregate;
+import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rel.logical.LogicalProject;
+import org.apache.calcite.rel.logical.LogicalSort;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+
+import java.util.AbstractList;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Predicate;
+
+/**
+ * Rules and relational operators for
+ * {@link SolrRel#CONVENTION}
+ * calling convention.
+ */
+class SolrRules {
+  static final RelOptRule[] RULES = {
+      SolrSortRule.SORT_RULE,
+      SolrFilterRule.FILTER_RULE,
+      SolrProjectRule.PROJECT_RULE,
+      SolrAggregateRule.AGGREGATE_RULE,
+  };
+
+  static List<String> solrFieldNames(final RelDataType rowType) {
+    return SqlValidatorUtil.uniquify(
+        new AbstractList<String>() {
+          @Override
+          public String get(int index) {
+            return rowType.getFieldList().get(index).getName();
+          }
+
+          @Override
+          public int size() {
+            return rowType.getFieldCount();
+          }
+        }, true);
+  }
+
+  /** Translator from {@link RexNode} to strings in Solr's expression language. */
+  static class RexToSolrTranslator extends RexVisitorImpl<String> {
+    private final JavaTypeFactory typeFactory;
+    private final List<String> inFields;
+
+    RexToSolrTranslator(JavaTypeFactory typeFactory, List<String> inFields) {
+      super(true);
+      this.typeFactory = typeFactory;
+      this.inFields = inFields;
+    }
+
+    @Override
+    public String visitInputRef(RexInputRef inputRef) {
+      return inFields.get(inputRef.getIndex());
+    }
+
+    @Override
+    public String visitCall(RexCall call) {
+      final List<String> strings = visitList(call.operands);
+      if (call.getKind() == SqlKind.CAST) {
+        return strings.get(0);
+      }
+
+      return super.visitCall(call);
+    }
+
+    private 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 Solr calling convention. */
+  abstract static class SolrConverterRule extends ConverterRule {
+    final Convention out = SolrRel.CONVENTION;
+
+    SolrConverterRule(Class<? extends RelNode> clazz, String description) {
+      this(clazz, relNode -> true, description);
+    }
+
+    <R extends RelNode> SolrConverterRule(Class<R> clazz, Predicate<RelNode> predicate, String description) {
+      super(clazz, Convention.NONE, SolrRel.CONVENTION, description);
+    }
+  }
+
+  /**
+   * Rule to convert a {@link LogicalFilter} to a {@link SolrFilter}.
+   */
+  private static class SolrFilterRule extends SolrConverterRule {
+    private static boolean isNotFilterByExpr(List<RexNode> rexNodes, List<String> fieldNames) {
+
+      // We dont have a way to filter by result of aggregator now
+      boolean result = true;
+
+      for (RexNode rexNode : rexNodes) {
+        if (rexNode instanceof RexCall) {
+          result = result && isNotFilterByExpr(((RexCall) rexNode).getOperands(), fieldNames);
+        } else if (rexNode instanceof RexInputRef) {
+          result = result && !fieldNames.get(((RexInputRef) rexNode).getIndex()).startsWith("EXPR$");
+        }
+      }
+      return result;
+    }
+
+    private static final Predicate<RelNode> FILTER_PREDICATE = relNode -> {
+      List<RexNode> filterOperands = ((RexCall) ((LogicalFilter) relNode).getCondition()).getOperands();
+      return isNotFilterByExpr(filterOperands, SolrRules.solrFieldNames(relNode.getRowType()));
+    };
+
+    private static final SolrFilterRule FILTER_RULE = new SolrFilterRule();
+
+    private SolrFilterRule() {
+      super(LogicalFilter.class, FILTER_PREDICATE, "SolrFilterRule");
+    }
+
+    public RelNode convert(RelNode rel) {
+      final LogicalFilter filter = (LogicalFilter) rel;
+      final RelTraitSet traitSet = filter.getTraitSet().replace(out);
+      return new SolrFilter(
+          rel.getCluster(),
+          traitSet,
+          convert(filter.getInput(), out),
+          filter.getCondition());
+    }
+  }
+
+  /**
+   * Rule to convert a {@link LogicalProject} to a {@link SolrProject}.
+   */
+  private static class SolrProjectRule extends SolrConverterRule {
+    private static final SolrProjectRule PROJECT_RULE = new SolrProjectRule();
+
+    private SolrProjectRule() {
+      super(LogicalProject.class, "SolrProjectRule");
+    }
+
+    public RelNode convert(RelNode rel) {
+      final LogicalProject project = (LogicalProject) rel;
+      final RelNode converted = convert(project.getInput(), out);
+      final RelTraitSet traitSet = project.getTraitSet().replace(out);
+      return new SolrProject(
+          rel.getCluster(),
+          traitSet,
+          converted,
+          project.getProjects(),
+          project.getRowType());
+    }
+  }
+
+  /**
+   * Rule to convert a {@link LogicalSort} to a {@link SolrSort}.
+   */
+  private static class SolrSortRule extends SolrConverterRule {
+    static final SolrSortRule SORT_RULE = new SolrSortRule(LogicalSort.class, "SolrSortRule");
+
+    SolrSortRule(Class<? extends RelNode> clazz, String description) {
+      super(clazz, description);
+    }
+
+    @Override
+    public RelNode convert(RelNode rel) {
+      final Sort sort = (Sort) rel;
+      final RelTraitSet traitSet = sort.getTraitSet().replace(out).replace(sort.getCollation());
+      return new SolrSort(
+          rel.getCluster(),
+          traitSet,
+          convert(sort.getInput(), traitSet.replace(RelCollations.EMPTY)),
+          sort.getCollation(),
+          sort.offset,
+          sort.fetch);
+    }
+  }
+
+  /**
+   * Rule to convert an {@link LogicalAggregate} to an {@link SolrAggregate}.
+   */
+  private static class SolrAggregateRule extends SolrConverterRule {
+//    private static final Predicate<RelNode> AGGREGATE_PREDICTE = relNode ->
+//        Aggregate.IS_SIMPLE.apply(((LogicalAggregate)relNode));// &&
+//        !((LogicalAggregate)relNode).containsDistinctCall();
+
+    private static final RelOptRule AGGREGATE_RULE = new SolrAggregateRule();
+
+    private SolrAggregateRule() {
+      super(LogicalAggregate.class, "SolrAggregateRule");
+    }
+
+    @Override
+    public RelNode convert(RelNode rel) {
+      final LogicalAggregate agg = (LogicalAggregate) rel;
+      final RelTraitSet traitSet = agg.getTraitSet().replace(out);
+      return new SolrAggregate(
+          rel.getCluster(),
+          traitSet,
+          convert(agg.getInput(), traitSet.simplify()),
+          agg.indicator,
+          agg.getGroupSet(),
+          agg.getGroupSets(),
+          agg.getAggCallList());
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrSchema.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrSchema.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrSchema.java
new file mode 100644
index 0000000..83fa537
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrSchema.java
@@ -0,0 +1,128 @@
+/*
+ * 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.solr.handler.sql;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.calcite.rel.type.*;
+import org.apache.calcite.schema.Table;
+import org.apache.calcite.schema.impl.AbstractSchema;
+import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.request.LukeRequest;
+import org.apache.solr.client.solrj.response.LukeResponse;
+import org.apache.solr.common.cloud.Aliases;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.luke.FieldFlag;
+
+import java.io.IOException;
+import java.util.EnumSet;
+import java.util.Map;
+import java.util.Properties;
+
+class SolrSchema extends AbstractSchema {
+  final Properties properties;
+
+  SolrSchema(Properties properties) {
+    super();
+    this.properties = properties;
+  }
+
+  @Override
+  protected Map<String, Table> getTableMap() {
+    String zk = this.properties.getProperty("zk");
+    try(CloudSolrClient cloudSolrClient = new CloudSolrClient.Builder().withZkHost(zk).build()) {
+      cloudSolrClient.connect();
+      ZkStateReader zkStateReader = cloudSolrClient.getZkStateReader();
+      ClusterState clusterState = zkStateReader.getClusterState();
+
+      final ImmutableMap.Builder<String, Table> builder = ImmutableMap.builder();
+
+      for (String collection : clusterState.getCollectionsMap().keySet()) {
+        builder.put(collection, new SolrTable(this, collection));
+      }
+
+      Aliases aliases = zkStateReader.getAliases();
+      if(aliases.collectionAliasSize() > 0) {
+        for (Map.Entry<String, String> alias : aliases.getCollectionAliasMap().entrySet()) {
+          builder.put(alias.getKey(), new SolrTable(this, alias.getValue()));
+        }
+      }
+
+      return builder.build();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private Map<String, LukeResponse.FieldInfo> getFieldInfo(String collection) {
+    String zk = this.properties.getProperty("zk");
+    try(CloudSolrClient cloudSolrClient = new CloudSolrClient.Builder().withZkHost(zk).build()) {
+      cloudSolrClient.connect();
+      LukeRequest lukeRequest = new LukeRequest();
+      lukeRequest.setNumTerms(0);
+      LukeResponse lukeResponse = lukeRequest.process(cloudSolrClient, collection);
+      return lukeResponse.getFieldInfo();
+    } catch (SolrServerException | IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  RelProtoDataType getRelDataType(String collection) {
+    // Temporary type factory, just for the duration of this method. Allowable
+    // because we're creating a proto-type, not a type; before being used, the
+    // proto-type will be copied into a real type factory.
+    final RelDataTypeFactory typeFactory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+    final RelDataTypeFactory.FieldInfoBuilder fieldInfo = typeFactory.builder();
+    Map<String, LukeResponse.FieldInfo> luceneFieldInfoMap = getFieldInfo(collection);
+
+    for(Map.Entry<String, LukeResponse.FieldInfo> entry : luceneFieldInfoMap.entrySet()) {
+      LukeResponse.FieldInfo luceneFieldInfo = entry.getValue();
+
+      RelDataType type;
+      switch (luceneFieldInfo.getType()) {
+        case "string":
+          type = typeFactory.createJavaType(String.class);
+          break;
+        case "int":
+        case "long":
+          type = typeFactory.createJavaType(Long.class);
+          break;
+        case "float":
+        case "double":
+          type = typeFactory.createJavaType(Double.class);
+          break;
+        default:
+          type = typeFactory.createJavaType(String.class);
+      }
+
+      EnumSet<FieldFlag> flags = luceneFieldInfo.parseFlags(luceneFieldInfo.getSchema());
+      /*
+      if(flags != null && flags.contains(FieldFlag.MULTI_VALUED)) {
+        type = typeFactory.createArrayType(type, -1);
+      }
+      */
+
+      fieldInfo.add(entry.getKey(), type).nullable(true);
+    }
+    fieldInfo.add("_query_",typeFactory.createJavaType(String.class));
+    fieldInfo.add("score",typeFactory.createJavaType(Double.class));
+
+    return RelDataTypeImpl.proto(fieldInfo.build());
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrSort.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrSort.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrSort.java
new file mode 100644
index 0000000..1c5274a
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrSort.java
@@ -0,0 +1,79 @@
+/*
+ * 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.solr.handler.sql;
+
+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 java.util.List;
+
+/**
+ * Implementation of {@link org.apache.calcite.rel.core.Sort} relational expression in Solr.
+ */
+class SolrSort extends Sort implements SolrRel {
+
+  SolrSort(RelOptCluster cluster, RelTraitSet traitSet, RelNode child, RelCollation collation, RexNode offset,
+           RexNode fetch) {
+    super(cluster, traitSet, child, collation, offset, fetch);
+
+    assert getConvention() == SolrRel.CONVENTION;
+    assert getConvention() == child.getConvention();
+  }
+
+  @Override
+  public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+    return planner.getCostFactory().makeZeroCost();
+  }
+
+  @Override
+  public Sort copy(RelTraitSet traitSet, RelNode input, RelCollation newCollation, RexNode offset, RexNode fetch) {
+    return new SolrSort(getCluster(), traitSet, input, collation, offset, fetch);
+  }
+
+  public void implement(Implementor implementor) {
+    implementor.visitChild(0, getInput());
+
+    List<RelFieldCollation> sortCollations = collation.getFieldCollations();
+    if (!sortCollations.isEmpty()) {
+      // Construct a series of order clauses from the desired collation
+      final List<RelDataTypeField> fields = getRowType().getFieldList();
+      for (RelFieldCollation fieldCollation : sortCollations) {
+        final String name = fields.get(fieldCollation.getFieldIndex()).getName();
+        String direction = "asc";
+        if (fieldCollation.getDirection().equals(RelFieldCollation.Direction.DESCENDING)) {
+          direction = "desc";
+        }
+        implementor.addOrder(name, direction);
+      }
+    }
+
+
+    if(fetch != null) {
+      implementor.setLimit(((RexLiteral) fetch).getValue().toString());
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrTable.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrTable.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrTable.java
new file mode 100644
index 0000000..6784323
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrTable.java
@@ -0,0 +1,842 @@
+/*
+ * 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.solr.handler.sql;
+
+import org.apache.calcite.adapter.java.AbstractQueryableTable;
+import org.apache.calcite.linq4j.*;
+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.rel.type.RelProtoDataType;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.TranslatableTable;
+import org.apache.calcite.schema.impl.AbstractTableQueryable;
+import org.apache.calcite.util.Pair;
+import org.apache.solr.client.solrj.io.comp.ComparatorOrder;
+import org.apache.solr.client.solrj.io.comp.FieldComparator;
+import org.apache.solr.client.solrj.io.comp.MultipleFieldComparator;
+import org.apache.solr.client.solrj.io.comp.StreamComparator;
+import org.apache.solr.client.solrj.io.eq.FieldEqualitor;
+import org.apache.solr.client.solrj.io.eq.MultipleFieldEqualitor;
+import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
+import org.apache.solr.client.solrj.io.ops.AndOperation;
+import org.apache.solr.client.solrj.io.ops.BooleanOperation;
+import org.apache.solr.client.solrj.io.ops.EqualsOperation;
+import org.apache.solr.client.solrj.io.ops.GreaterThanEqualToOperation;
+import org.apache.solr.client.solrj.io.ops.GreaterThanOperation;
+import org.apache.solr.client.solrj.io.ops.LessThanEqualToOperation;
+import org.apache.solr.client.solrj.io.ops.LessThanOperation;
+import org.apache.solr.client.solrj.io.ops.NotOperation;
+import org.apache.solr.client.solrj.io.ops.OrOperation;
+import org.apache.solr.client.solrj.io.stream.*;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParser;
+import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
+import org.apache.solr.client.solrj.io.stream.metrics.*;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * Table based on a Solr collection
+ */
+class SolrTable extends AbstractQueryableTable implements TranslatableTable {
+  private static final String DEFAULT_QUERY = "*:*";
+  private static final String DEFAULT_VERSION_FIELD = "_version_";
+
+  private final String collection;
+  private final SolrSchema schema;
+  private RelProtoDataType protoRowType;
+
+  SolrTable(SolrSchema schema, String collection) {
+    super(Object[].class);
+    this.schema = schema;
+    this.collection = collection;
+  }
+
+  public String toString() {
+    return "SolrTable {" + collection + "}";
+  }
+
+  public RelDataType getRowType(RelDataTypeFactory typeFactory) {
+    if (protoRowType == null) {
+      protoRowType = schema.getRelDataType(collection);
+    }
+    return protoRowType.apply(typeFactory);
+  }
+  
+  private Enumerable<Object> query(final Properties properties) {
+    return query(properties, Collections.emptyList(), null, Collections.emptyList(), Collections.emptyList(),
+        Collections.emptyList(), null, null, null);
+  }
+
+  /** Executes a Solr query on the underlying table.
+   *
+   * @param properties Connections properties
+   * @param fields List of fields to project
+   * @param query A string for the query
+   * @return Enumerator of results
+   */
+  private Enumerable<Object> query(final Properties properties,
+                                   final List<Map.Entry<String, Class>> fields,
+                                   final String query,
+                                   final List<Pair<String, String>> orders,
+                                   final List<String> buckets,
+                                   final List<Pair<String, String>> metricPairs,
+                                   final String limit,
+                                   final String negativeQuery,
+                                   final String havingPredicate) {
+    // SolrParams should be a ModifiableParams instead of a map
+    boolean mapReduce = "map_reduce".equals(properties.getProperty("aggregationMode"));
+    boolean negative = Boolean.parseBoolean(negativeQuery);
+
+    String q = null;
+
+    if (query == null) {
+      q = DEFAULT_QUERY;
+    } else {
+      if(negative) {
+        q = DEFAULT_QUERY + " AND " + query;
+      } else {
+        q = query;
+      }
+    }
+
+    TupleStream tupleStream;
+    String zk = properties.getProperty("zk");
+    try {
+      if (metricPairs.isEmpty() && buckets.isEmpty()) {
+        tupleStream = handleSelect(zk, collection, q, fields, orders, limit);
+      } else {
+        if(buckets.isEmpty()) {
+          tupleStream = handleStats(zk, collection, q, metricPairs);
+        } else {
+          if(mapReduce) {
+            tupleStream = handleGroupByMapReduce(zk,
+                                                 collection,
+                                                 properties,
+                                                 fields,
+                                                 q,
+                                                 orders,
+                                                 buckets,
+                                                 metricPairs,
+                                                 limit,
+                                                 havingPredicate);
+          } else {
+            tupleStream = handleGroupByFacet(zk,
+                                             collection,
+                                             fields,
+                                             q,
+                                             orders,
+                                             buckets,
+                                             metricPairs,
+                                             limit,
+                                             havingPredicate);
+          }
+        }
+      }
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+
+    final TupleStream finalStream = tupleStream;
+
+    return new AbstractEnumerable<Object>() {
+      // Use original fields list to make sure only the fields specified are enumerated
+      public Enumerator<Object> enumerator() {
+        return new SolrEnumerator(finalStream, fields);
+      }
+    };
+  }
+
+  private static StreamComparator bucketSortComp(List<Bucket> buckets, Map<String,String> dirs) {
+    FieldComparator[] comps = new FieldComparator[buckets.size()];
+    for(int i=0; i<buckets.size(); i++) {
+      ComparatorOrder comparatorOrder = ComparatorOrder.fromString(dirs.get(buckets.get(i).toString()));
+      String sortKey = buckets.get(i).toString();
+      comps[i] = new FieldComparator(sortKey, comparatorOrder);
+    }
+
+    if(comps.length == 1) {
+      return comps[0];
+    } else {
+      return new MultipleFieldComparator(comps);
+    }
+  }
+
+  private static StreamComparator bucketSortComp(Bucket[] buckets, String dir) {
+    FieldComparator[] comps = new FieldComparator[buckets.length];
+    for(int i=0; i<buckets.length; i++) {
+      ComparatorOrder comparatorOrder = ascDescComp(dir);
+      String sortKey = buckets[i].toString();
+      comps[i] = new FieldComparator(sortKey, comparatorOrder);
+    }
+
+    if(comps.length == 1) {
+      return comps[0];
+    } else {
+      return new MultipleFieldComparator(comps);
+    }
+  }
+
+  private String getSortDirection(Map.Entry<String, String> order) {
+    String direction = order.getValue();
+    return direction == null ? "asc" : direction;
+  }
+
+  private StreamComparator getComp(List<? extends Map.Entry<String, String>> orders) {
+    FieldComparator[] comps = new FieldComparator[orders.size()];
+    for(int i = 0; i < orders.size(); i++) {
+      Map.Entry<String, String> order = orders.get(i);
+      String direction = getSortDirection(order);
+      ComparatorOrder comparatorOrder = ComparatorOrder.fromString(direction);
+      String sortKey = order.getKey();
+      comps[i] = new FieldComparator(sortKey, comparatorOrder);
+    }
+
+    if(comps.length == 1) {
+      return comps[0];
+    } else {
+      return new MultipleFieldComparator(comps);
+    }
+  }
+
+  private List<Metric> buildMetrics(List<Pair<String, String>> metricPairs, boolean ifEmptyCount) {
+    List<Metric> metrics = new ArrayList<>(metricPairs.size());
+    metrics.addAll(metricPairs.stream().map(this::getMetric).collect(Collectors.toList()));
+    if(metrics.size() == 0 && ifEmptyCount) {
+      metrics.add(new CountMetric());
+    }
+    return metrics;
+  }
+
+  private Metric getMetric(Pair<String, String> metricPair) {
+    switch (metricPair.getKey()) {
+      case "COUNT":
+        return new CountMetric(metricPair.getValue());
+      case "SUM":
+      case "$SUM0":
+        return new SumMetric(metricPair.getValue());
+      case "MIN":
+        return new MinMetric(metricPair.getValue());
+      case "MAX":
+        return new MaxMetric(metricPair.getValue());
+      case "AVG":
+        return new MeanMetric(metricPair.getValue());
+      default:
+        throw new IllegalArgumentException(metricPair.getKey());
+    }
+  }
+
+  private TupleStream handleSelect(String zk,
+                                   String collection,
+                                   String query,
+                                   List<Map.Entry<String, Class>> fields,
+                                   List<Pair<String, String>> orders,
+                                   String limit) throws IOException {
+
+    ModifiableSolrParams params = new ModifiableSolrParams();
+    params.add(CommonParams.Q, query);
+
+    //Validate the fields
+    for(Map.Entry<String, Class> entry : fields) {
+      String fname = entry.getKey();
+      if(limit == null && "score".equals(fname)) {
+        throw new IOException("score is not a valid field for unlimited queries.");
+      }
+
+      if(fname.contains("*")) {
+        throw new IOException("* is not supported for column selection.");
+      }
+    }
+
+    String fl = getFields(fields);
+
+    if(orders.size() > 0) {
+      params.add(CommonParams.SORT, getSort(orders));
+    } else {
+      if(limit == null) {
+        params.add(CommonParams.SORT, "_version_ desc");
+        fl = fl+",_version_";
+      } else {
+        params.add(CommonParams.SORT, "score desc");
+        if(fl.indexOf("score") == -1) {
+          fl = fl + ",score";
+        }
+      }
+    }
+
+    params.add(CommonParams.FL, fl);
+
+    if (limit != null) {
+      params.add(CommonParams.ROWS, limit);
+      return new LimitStream(new CloudSolrStream(zk, collection, params), Integer.parseInt(limit));
+    } else {
+      params.add(CommonParams.QT, "/export");
+      return new CloudSolrStream(zk, collection, params);
+    }
+  }
+
+  private String getSort(List<Pair<String, String>> orders) {
+    StringBuilder buf = new StringBuilder();
+    for(Pair<String, String> pair : orders) {
+      if(buf.length() > 0) {
+        buf.append(",");
+      }
+      buf.append(pair.getKey()).append(" ").append(pair.getValue());
+    }
+
+    return buf.toString();
+  }
+
+  private String getSingleSort(Pair<String, String> order) {
+    StringBuilder buf = new StringBuilder();
+    buf.append(order.getKey()).append(" ").append(order.getValue());
+    return buf.toString();
+  }
+
+  private String getFields(List<Map.Entry<String, Class>> fields) {
+    StringBuilder buf = new StringBuilder();
+    for(Map.Entry<String, Class> field : fields) {
+
+      if(buf.length() > 0) {
+        buf.append(",");
+      }
+
+      buf.append(field.getKey());
+    }
+
+    return buf.toString();
+  }
+
+  private String getFields(Set<String> fieldSet) {
+    StringBuilder buf = new StringBuilder();
+    boolean appendVersion = true;
+    for(String field : fieldSet) {
+
+      if(buf.length() > 0) {
+        buf.append(",");
+      }
+
+      if(field.equals("_version_")) {
+        appendVersion = false;
+      }
+
+      buf.append(field);
+    }
+
+    if(appendVersion){
+      buf.append(",_version_");
+    }
+
+    return buf.toString();
+  }
+
+
+  private Set<String> getFieldSet(Metric[] metrics, List<Map.Entry<String, Class>> fields) {
+    HashSet set = new HashSet();
+    for(Metric metric : metrics) {
+      for(String column : metric.getColumns()) {
+        set.add(column);
+      }
+    }
+
+    for(Map.Entry<String, Class> field : fields) {
+      if(field.getKey().indexOf('(') == -1) {
+        set.add(field.getKey());
+      }
+    }
+
+    return set;
+  }
+
+  private static String getSortDirection(List<Pair<String, String>> orders) {
+    if(orders != null && orders.size() > 0) {
+      for(Pair<String,String> item : orders) {
+        return item.getValue();
+      }
+    }
+
+    return "asc";
+  }
+
+  private static String bucketSort(Bucket[] buckets, String dir) {
+    StringBuilder buf = new StringBuilder();
+    boolean comma = false;
+    for(Bucket bucket : buckets) {
+      if(comma) {
+        buf.append(",");
+      }
+      buf.append(bucket.toString()).append(" ").append(dir);
+      comma = true;
+    }
+
+    return buf.toString();
+  }
+
+  private static String getPartitionKeys(Bucket[] buckets) {
+    StringBuilder buf = new StringBuilder();
+    boolean comma = false;
+    for(Bucket bucket : buckets) {
+      if(comma) {
+        buf.append(",");
+      }
+      buf.append(bucket.toString());
+      comma = true;
+    }
+    return buf.toString();
+  }
+
+  private static boolean sortsEqual(Bucket[] buckets, String direction, List<Pair<String, String>> orders) {
+
+    if(buckets.length != orders.size()) {
+      return false;
+    }
+
+    for(int i=0; i< buckets.length; i++) {
+      Bucket bucket = buckets[i];
+      Pair<String, String> order = orders.get(i);
+      if(!bucket.toString().equals(order.getKey())) {
+        return false;
+      }
+
+      if(!order.getValue().toLowerCase(Locale.ROOT).contains(direction.toLowerCase(Locale.ROOT))) {
+        return false;
+      }
+    }
+
+    return true;
+  }
+
+  private TupleStream handleGroupByMapReduce(String zk,
+                                             String collection,
+                                             Properties properties,
+                                             final List<Map.Entry<String, Class>> fields,
+                                             final String query,
+                                             final List<Pair<String, String>> orders,
+                                             final List<String> _buckets,
+                                             final List<Pair<String, String>> metricPairs,
+                                             final String limit,
+                                             final String havingPredicate) throws IOException {
+
+    int numWorkers = Integer.parseInt(properties.getProperty("numWorkers", "1"));
+
+    Bucket[] buckets = buildBuckets(_buckets, fields);
+    Metric[] metrics = buildMetrics(metricPairs, false).toArray(new Metric[0]);
+
+    if(metrics.length == 0) {
+      return handleSelectDistinctMapReduce(zk, collection, properties, fields, query, orders, buckets, limit);
+    }
+
+    Set<String> fieldSet = getFieldSet(metrics, fields);
+
+    if(metrics.length == 0) {
+      throw new IOException("Group by queries must include atleast one aggregate function.");
+    }
+
+    String fl = getFields(fieldSet);
+    String sortDirection = getSortDirection(orders);
+    String sort = bucketSort(buckets, sortDirection);
+
+    ModifiableSolrParams params = new ModifiableSolrParams();
+
+    params.set(CommonParams.FL, fl);
+    params.set(CommonParams.Q, query);
+    //Always use the /export handler for Group By Queries because it requires exporting full result sets.
+    params.set(CommonParams.QT, "/export");
+
+    if(numWorkers > 1) {
+      params.set("partitionKeys", getPartitionKeys(buckets));
+    }
+
+    params.set("sort", sort);
+
+    TupleStream tupleStream = null;
+
+    CloudSolrStream cstream = new CloudSolrStream(zk, collection, params);
+    tupleStream = new RollupStream(cstream, buckets, metrics);
+
+    StreamFactory factory = new StreamFactory()
+        .withFunctionName("search", CloudSolrStream.class)
+        .withFunctionName("parallel", ParallelStream.class)
+        .withFunctionName("rollup", RollupStream.class)
+        .withFunctionName("sum", SumMetric.class)
+        .withFunctionName("min", MinMetric.class)
+        .withFunctionName("max", MaxMetric.class)
+        .withFunctionName("avg", MeanMetric.class)
+        .withFunctionName("count", CountMetric.class)
+        .withFunctionName("and", AndOperation.class)
+        .withFunctionName("or", OrOperation.class)
+        .withFunctionName("not", NotOperation.class)
+        .withFunctionName("eq", EqualsOperation.class)
+        .withFunctionName("gt", GreaterThanOperation.class)
+        .withFunctionName("lt", LessThanOperation.class)
+        .withFunctionName("lteq", LessThanEqualToOperation.class)
+        .withFunctionName("having", HavingStream.class)
+        .withFunctionName("gteq", GreaterThanEqualToOperation.class);
+
+    if(havingPredicate != null) {
+      BooleanOperation booleanOperation = (BooleanOperation)factory.constructOperation(StreamExpressionParser.parse(havingPredicate));
+      tupleStream = new HavingStream(tupleStream, booleanOperation);
+    }
+
+    if(numWorkers > 1) {
+      // Do the rollups in parallel
+      // Maintain the sort of the Tuples coming from the workers.
+      StreamComparator comp = bucketSortComp(buckets, sortDirection);
+      ParallelStream parallelStream = new ParallelStream(zk, collection, tupleStream, numWorkers, comp);
+
+
+      parallelStream.setStreamFactory(factory);
+      tupleStream = parallelStream;
+    }
+
+    //TODO: Currently we are not pushing down the having clause.
+    //      We need to push down the having clause to ensure that LIMIT does not cut off records prior to the having filter.
+
+    if(orders != null && orders.size() > 0) {
+      if(!sortsEqual(buckets, sortDirection, orders)) {
+        int lim = (limit == null) ? 100 : Integer.parseInt(limit);
+        StreamComparator comp = getComp(orders);
+        //Rank the Tuples
+        //If parallel stream is used ALL the Rolled up tuples from the workers will be ranked
+        //Providing a true Top or Bottom.
+        tupleStream = new RankStream(tupleStream, lim, comp);
+      } else {
+        // Sort is the same as the same as the underlying stream
+        // Only need to limit the result, not Rank the result
+        if(limit != null) {
+          tupleStream = new LimitStream(tupleStream, Integer.parseInt(limit));
+        }
+      }
+    } else {
+      //No order by, check for limit
+      if(limit != null) {
+        tupleStream = new LimitStream(tupleStream, Integer.parseInt(limit));
+      }
+    }
+
+    return tupleStream;
+  }
+
+  private Bucket[] buildBuckets(List<String> buckets, List<Map.Entry<String, Class>> fields) {
+    Bucket[] bucketsArray = new Bucket[buckets.size()];
+
+    int i=0;
+    for(Map.Entry<String,Class> field : fields) {
+      String fieldName = field.getKey();
+      if(buckets.contains(fieldName)) {
+        bucketsArray[i++] = new Bucket(fieldName);
+      }
+    }
+
+    return bucketsArray;
+  }
+
+  private TupleStream handleGroupByFacet(String zkHost,
+                                         String collection,
+                                         final List<Map.Entry<String, Class>> fields,
+                                         final String query,
+                                         final List<Pair<String, String>> orders,
+                                         final List<String> bucketFields,
+                                         final List<Pair<String, String>> metricPairs,
+                                         final String lim,
+                                         final String havingPredicate) throws IOException {
+
+    ModifiableSolrParams solrParams = new ModifiableSolrParams();
+    solrParams.add(CommonParams.Q, query);
+
+    Bucket[] buckets = buildBuckets(bucketFields, fields);
+    Metric[] metrics = buildMetrics(metricPairs, true).toArray(new Metric[0]);
+    if(metrics.length == 0) {
+      metrics = new Metric[1];
+      metrics[0] = new CountMetric();
+    }
+
+    int limit = lim != null ? Integer.parseInt(lim) : 1000;
+
+    FieldComparator[] sorts = null;
+
+    if(orders == null || orders.size() == 0) {
+      sorts = new FieldComparator[buckets.length];
+      for(int i=0; i<sorts.length; i++) {
+        sorts[i] = new FieldComparator("index", ComparatorOrder.ASCENDING);
+      }
+    } else {
+      sorts = getComps(orders);
+    }
+
+    int overfetch = (int)(limit * 1.25);
+
+    TupleStream tupleStream = new FacetStream(zkHost,
+                                              collection,
+                                              solrParams,
+                                              buckets,
+                                              metrics,
+                                              sorts,
+                                              overfetch);
+
+
+
+    StreamFactory factory = new StreamFactory()
+        .withFunctionName("search", CloudSolrStream.class)
+        .withFunctionName("parallel", ParallelStream.class)
+        .withFunctionName("rollup", RollupStream.class)
+        .withFunctionName("sum", SumMetric.class)
+        .withFunctionName("min", MinMetric.class)
+        .withFunctionName("max", MaxMetric.class)
+        .withFunctionName("avg", MeanMetric.class)
+        .withFunctionName("count", CountMetric.class)
+        .withFunctionName("and", AndOperation.class)
+        .withFunctionName("or", OrOperation.class)
+        .withFunctionName("not", NotOperation.class)
+        .withFunctionName("eq", EqualsOperation.class)
+        .withFunctionName("gt", GreaterThanOperation.class)
+        .withFunctionName("lt", LessThanOperation.class)
+        .withFunctionName("lteq", LessThanEqualToOperation.class)
+        .withFunctionName("gteq", GreaterThanEqualToOperation.class);
+
+    if(havingPredicate != null) {
+      BooleanOperation booleanOperation = (BooleanOperation)factory.constructOperation(StreamExpressionParser.parse(havingPredicate));
+      tupleStream = new HavingStream(tupleStream, booleanOperation);
+    }
+
+    if(lim != null)
+    {
+      tupleStream = new LimitStream(tupleStream, limit);
+    }
+
+    return tupleStream;
+  }
+
+  private TupleStream handleSelectDistinctMapReduce(final String zkHost,
+                                                    final String collection,
+                                                    final Properties properties,
+                                                    final List<Map.Entry<String, Class>> fields,
+                                                    final String query,
+                                                    final List<Pair<String, String>> orders,
+                                                    final Bucket[] buckets,
+                                                    final String limit) throws IOException{
+
+    int numWorkers = Integer.parseInt(properties.getProperty("numWorkers", "1"));
+
+    String fl = getFields(fields);
+
+    String sort = null;
+    StreamEqualitor ecomp = null;
+    StreamComparator comp = null;
+
+    if(orders != null && orders.size() > 0) {
+      StreamComparator[] adjustedSorts = adjustSorts(orders, buckets);
+      // Because of the way adjustSorts works we know that each FieldComparator has a single
+      // field name. For this reason we can just look at the leftFieldName
+      FieldEqualitor[] fieldEqualitors = new FieldEqualitor[adjustedSorts.length];
+      StringBuilder buf = new StringBuilder();
+      for(int i=0; i<adjustedSorts.length; i++) {
+        FieldComparator fieldComparator = (FieldComparator)adjustedSorts[i];
+        fieldEqualitors[i] = new FieldEqualitor(fieldComparator.getLeftFieldName());
+        if(i>0) {
+          buf.append(",");
+        }
+        buf.append(fieldComparator.getLeftFieldName()).append(" ").append(fieldComparator.getOrder().toString());
+      }
+
+      sort = buf.toString();
+
+      if(adjustedSorts.length == 1) {
+        ecomp = fieldEqualitors[0];
+        comp = adjustedSorts[0];
+      } else {
+        ecomp = new MultipleFieldEqualitor(fieldEqualitors);
+        comp = new MultipleFieldComparator(adjustedSorts);
+      }
+    } else {
+      StringBuilder sortBuf = new StringBuilder();
+      FieldEqualitor[] equalitors = new FieldEqualitor[buckets.length];
+      StreamComparator[] streamComparators = new StreamComparator[buckets.length];
+      for(int i=0; i<buckets.length; i++) {
+        equalitors[i] = new FieldEqualitor(buckets[i].toString());
+        streamComparators[i] = new FieldComparator(buckets[i].toString(), ComparatorOrder.ASCENDING);
+        if(i>0) {
+          sortBuf.append(',');
+        }
+        sortBuf.append(buckets[i].toString()).append(" asc");
+      }
+
+      sort = sortBuf.toString();
+
+      if(equalitors.length == 1) {
+        ecomp = equalitors[0];
+        comp = streamComparators[0];
+      } else {
+        ecomp = new MultipleFieldEqualitor(equalitors);
+        comp = new MultipleFieldComparator(streamComparators);
+      }
+    }
+
+    ModifiableSolrParams params = new ModifiableSolrParams();
+
+    params.set(CommonParams.FL, fl);
+    params.set(CommonParams.Q, query);
+    //Always use the /export handler for Distinct Queries because it requires exporting full result sets.
+    params.set(CommonParams.QT, "/export");
+
+    if(numWorkers > 1) {
+      params.set("partitionKeys", getPartitionKeys(buckets));
+    }
+
+    params.set("sort", sort);
+
+    TupleStream tupleStream = null;
+
+    CloudSolrStream cstream = new CloudSolrStream(zkHost, collection, params);
+    tupleStream = new UniqueStream(cstream, ecomp);
+
+    if(numWorkers > 1) {
+      // Do the unique in parallel
+      // Maintain the sort of the Tuples coming from the workers.
+      ParallelStream parallelStream = new ParallelStream(zkHost, collection, tupleStream, numWorkers, comp);
+
+      StreamFactory factory = new StreamFactory()
+          .withFunctionName("search", CloudSolrStream.class)
+          .withFunctionName("parallel", ParallelStream.class)
+          .withFunctionName("unique", UniqueStream.class);
+
+      parallelStream.setStreamFactory(factory);
+      tupleStream = parallelStream;
+    }
+
+    if(limit != null) {
+      tupleStream = new LimitStream(tupleStream, Integer.parseInt(limit));
+    }
+
+    return tupleStream;
+  }
+
+
+  private StreamComparator[] adjustSorts(List<Pair<String, String>> orders, Bucket[] buckets) throws IOException {
+    List<FieldComparator> adjustedSorts = new ArrayList();
+    Set<String> bucketFields = new HashSet();
+    Set<String> sortFields = new HashSet();
+
+    ComparatorOrder comparatorOrder = ComparatorOrder.ASCENDING;
+    for(Pair<String, String> order : orders) {
+      sortFields.add(order.getKey());
+      adjustedSorts.add(new FieldComparator(order.getKey(), ascDescComp(order.getValue())));
+      comparatorOrder = ascDescComp(order.getValue());
+    }
+
+    for(Bucket bucket : buckets) {
+      bucketFields.add(bucket.toString());
+    }
+
+    for(String sf : sortFields) {
+      if(!bucketFields.contains(sf)) {
+        throw new IOException("All sort fields must be in the field list.");
+      }
+    }
+
+    //Add sort fields if needed
+    if(sortFields.size() < buckets.length) {
+      for(Bucket bucket : buckets) {
+        String b = bucket.toString();
+        if(!sortFields.contains(b)) {
+          adjustedSorts.add(new FieldComparator(bucket.toString(), comparatorOrder));
+        }
+      }
+    }
+
+    return adjustedSorts.toArray(new FieldComparator[adjustedSorts.size()]);
+  }
+
+  private TupleStream handleStats(String zk,
+                                  String collection,
+                                  String query,
+                                  List<Pair<String, String>> metricPairs) {
+
+
+    ModifiableSolrParams solrParams = new ModifiableSolrParams();
+    solrParams.add(CommonParams.Q, query);
+    Metric[] metrics = buildMetrics(metricPairs, false).toArray(new Metric[0]);
+    return new StatsStream(zk, collection, solrParams, metrics);
+  }
+
+  public <T> Queryable<T> asQueryable(QueryProvider queryProvider, SchemaPlus schema, String tableName) {
+    return new SolrQueryable<>(queryProvider, schema, this, tableName);
+  }
+
+  public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable relOptTable) {
+    final RelOptCluster cluster = context.getCluster();
+    return new SolrTableScan(cluster, cluster.traitSetOf(SolrRel.CONVENTION), relOptTable, this, null);
+  }
+
+  @SuppressWarnings("WeakerAccess")
+  public static class SolrQueryable<T> extends AbstractTableQueryable<T> {
+    SolrQueryable(QueryProvider queryProvider, SchemaPlus schema, SolrTable table, String tableName) {
+      super(queryProvider, schema, table, tableName);
+    }
+
+    public Enumerator<T> enumerator() {
+      @SuppressWarnings("unchecked")
+      final Enumerable<T> enumerable = (Enumerable<T>) getTable().query(getProperties());
+      return enumerable.enumerator();
+    }
+
+    private SolrTable getTable() {
+      return (SolrTable) table;
+    }
+
+    private Properties getProperties() {
+      return schema.unwrap(SolrSchema.class).properties;
+    }
+
+    /** Called via code-generation.
+     *
+     * @see SolrMethod#SOLR_QUERYABLE_QUERY
+     */
+    @SuppressWarnings("UnusedDeclaration")
+    public Enumerable<Object> query(List<Map.Entry<String, Class>> fields, String query, List<Pair<String, String>> order,
+                                    List<String> buckets, List<Pair<String, String>> metricPairs, String limit, String negativeQuery, String havingPredicate) {
+      return getTable().query(getProperties(), fields, query, order, buckets, metricPairs, limit, negativeQuery, havingPredicate);
+    }
+  }
+
+  private static FieldComparator[] getComps(List<Pair<String, String>> orders) {
+    FieldComparator[] comps = new FieldComparator[orders.size()];
+    for(int i=0; i<orders.size(); i++) {
+      Pair<String,String> sortItem = orders.get(i);
+      String ordering = sortItem.getValue();
+      ComparatorOrder comparatorOrder = ascDescComp(ordering);
+      String sortKey = sortItem.getKey();
+      comps[i] = new FieldComparator(sortKey, comparatorOrder);
+    }
+
+    return comps;
+  }
+
+  private static ComparatorOrder ascDescComp(String s) {
+    if(s.toLowerCase(Locale.ROOT).contains("desc")) {
+      return ComparatorOrder.DESCENDING;
+    } else {
+      return ComparatorOrder.ASCENDING;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrTableScan.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrTableScan.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrTableScan.java
new file mode 100644
index 0000000..88c53ac
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrTableScan.java
@@ -0,0 +1,81 @@
+/*
+ * 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.solr.handler.sql;
+
+import org.apache.calcite.plan.*;
+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 a Solr collection.
+ */
+class SolrTableScan extends TableScan implements SolrRel {
+  private final SolrTable solrTable;
+  private final RelDataType projectRowType;
+
+  /**
+   * Creates a SolrTableScan.
+   *
+   * @param cluster        Cluster
+   * @param traitSet       Traits
+   * @param table          Table
+   * @param solrTable      Solr table
+   * @param projectRowType Fields and types to project; null to project raw row
+   */
+  SolrTableScan(RelOptCluster cluster, RelTraitSet traitSet, RelOptTable table, SolrTable solrTable,
+                RelDataType projectRowType) {
+    super(cluster, traitSet, table);
+    this.solrTable = solrTable;
+    this.projectRowType = projectRowType;
+
+    assert solrTable != null;
+    assert getConvention() == SolrRel.CONVENTION;
+  }
+
+  @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 RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+    assert inputs.isEmpty();
+    return this;
+  }
+
+  @Override
+  public RelDataType deriveRowType() {
+    return projectRowType != null ? projectRowType : super.deriveRowType();
+  }
+
+  @Override
+  public void register(RelOptPlanner planner) {
+    planner.addRule(SolrToEnumerableConverterRule.INSTANCE);
+    for (RelOptRule rule : SolrRules.RULES) {
+      planner.addRule(rule);
+    }
+  }
+
+  public void implement(Implementor implementor) {
+    implementor.solrTable = solrTable;
+    implementor.table = table;
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrToEnumerableConverter.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrToEnumerableConverter.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrToEnumerableConverter.java
new file mode 100644
index 0000000..10d4d4c
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrToEnumerableConverter.java
@@ -0,0 +1,135 @@
+/*
+ * 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.solr.handler.sql;
+
+import com.google.common.collect.Lists;
+import org.apache.calcite.adapter.enumerable.*;
+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.*;
+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 java.util.AbstractList;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Relational expression representing a scan of a table in Solr
+ */
+class SolrToEnumerableConverter extends ConverterImpl implements EnumerableRel {
+  SolrToEnumerableConverter(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
+    super(cluster, ConventionTraitDef.INSTANCE, traits, input);
+  }
+
+  @Override
+  public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+    return new SolrToEnumerableConverter(getCluster(), traitSet, sole(inputs));
+  }
+
+  @Override
+  public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+    return super.computeSelfCost(planner, mq).multiplyBy(.1);
+  }
+
+  public Result implement(EnumerableRelImplementor implementor, Prefer pref) {
+    // Generates a call to "query" with the appropriate fields
+    final BlockBuilder list = new BlockBuilder();
+    final SolrRel.Implementor solrImplementor = new SolrRel.Implementor();
+    solrImplementor.visitChild(0, getInput());
+    final RelDataType rowType = getRowType();
+    final PhysType physType = PhysTypeImpl.of(implementor.getTypeFactory(), rowType, pref.prefer(JavaRowFormat.ARRAY));
+    final Expression table = list.append("table", solrImplementor.table.getExpression(SolrTable.SolrQueryable.class));
+    final Expression fields =
+        list.append("fields",
+            constantArrayList(
+                Pair.zip(generateFields(SolrRules.solrFieldNames(rowType), solrImplementor.fieldMappings),
+                    new AbstractList<Class>() {
+                      @Override
+                      public Class get(int index) {
+                        return physType.fieldClass(index);
+                      }
+
+                      @Override
+                      public int size() {
+                        return rowType.getFieldCount();
+                      }
+                    }),
+                Pair.class));
+    final Expression query = list.append("query", Expressions.constant(solrImplementor.query, String.class));
+    final Expression orders = list.append("orders", constantArrayList(solrImplementor.orders, Pair.class));
+    final Expression buckets = list.append("buckets", constantArrayList(solrImplementor.buckets, String.class));
+    final Expression metricPairs = list.append("metricPairs", constantArrayList(solrImplementor.metricPairs, Pair.class));
+    final Expression limit = list.append("limit", Expressions.constant(solrImplementor.limitValue));
+    final Expression negativeQuery = list.append("negativeQuery", Expressions.constant(Boolean.toString(solrImplementor.negativeQuery), String.class));
+    final Expression havingPredicate = list.append("havingTest", Expressions.constant(solrImplementor.havingPredicate, String.class));
+    Expression enumerable = list.append("enumerable", Expressions.call(table, SolrMethod.SOLR_QUERYABLE_QUERY.method,
+        fields, query, orders, buckets, metricPairs, limit, negativeQuery, havingPredicate));
+    Hook.QUERY_PLAN.run(query);
+    list.add(Expressions.return_(null, enumerable));
+    return implementor.result(physType, list.toBlock());
+  }
+
+  private List<String> generateFields(List<String> queryFields, Map<String, String> fieldMappings) {
+    if(fieldMappings.isEmpty()) {
+      return queryFields;
+    } else {
+      List<String> fields = new ArrayList<>();
+      for(String field : queryFields) {
+        fields.add(getField(fieldMappings, field));
+      }
+      return fields;
+    }
+  }
+
+  private String getField(Map<String, String> fieldMappings, String field) {
+    String retField = field;
+    while(fieldMappings.containsKey(field)) {
+      field = fieldMappings.getOrDefault(field, retField);
+      if(retField.equals(field)) {
+        break;
+      } else {
+        retField = field;
+      }
+    }
+    return retField;
+  }
+
+  /**
+   * 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 "{ConstantExpression("x"), ConstantExpression("y")}".
+   */
+  private static <T> List<Expression> constantList(List<T> values) {
+    return Lists.transform(values, Expressions::constant);
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/SolrToEnumerableConverterRule.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/SolrToEnumerableConverterRule.java b/solr/core/src/java/org/apache/solr/handler/sql/SolrToEnumerableConverterRule.java
new file mode 100644
index 0000000..80365ca
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/SolrToEnumerableConverterRule.java
@@ -0,0 +1,39 @@
+/*
+ * 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.solr.handler.sql;
+
+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 SolrRel#CONVENTION} to {@link EnumerableConvention}.
+ */
+class SolrToEnumerableConverterRule extends ConverterRule {
+  static final ConverterRule INSTANCE = new SolrToEnumerableConverterRule();
+
+  private SolrToEnumerableConverterRule() {
+    super(RelNode.class, SolrRel.CONVENTION, EnumerableConvention.INSTANCE, "SolrToEnumerableConverterRule");
+  }
+
+  @Override
+  public RelNode convert(RelNode rel) {
+    RelTraitSet newTraitSet = rel.getTraitSet().replace(getOutConvention());
+    return new SolrToEnumerableConverter(rel.getCluster(), newTraitSet, rel);
+  }
+}

http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3370dbed/solr/core/src/java/org/apache/solr/handler/sql/package-info.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/sql/package-info.java b/solr/core/src/java/org/apache/solr/handler/sql/package-info.java
new file mode 100644
index 0000000..5aef90d
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/handler/sql/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+
+/**
+ * Classes related to Apache Calcite implementation in {@link org.apache.solr.handler.SQLHandler}
+ */
+package org.apache.solr.handler.sql;
\ No newline at end of file


Mime
View raw message