| /* |
| * 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 java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Locale; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.stream.Collectors; |
| |
| 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.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.eval.AndEvaluator; |
| import org.apache.solr.client.solrj.io.eval.EqualToEvaluator; |
| import org.apache.solr.client.solrj.io.eval.GreaterThanEqualToEvaluator; |
| import org.apache.solr.client.solrj.io.eval.GreaterThanEvaluator; |
| import org.apache.solr.client.solrj.io.eval.LessThanEqualToEvaluator; |
| import org.apache.solr.client.solrj.io.eval.LessThanEvaluator; |
| import org.apache.solr.client.solrj.io.eval.NotEvaluator; |
| import org.apache.solr.client.solrj.io.eval.OrEvaluator; |
| import org.apache.solr.client.solrj.io.eval.RawValueEvaluator; |
| import org.apache.solr.client.solrj.io.eval.RecursiveBooleanEvaluator; |
| import org.apache.solr.client.solrj.io.stream.CloudSolrStream; |
| import org.apache.solr.client.solrj.io.stream.FacetStream; |
| import org.apache.solr.client.solrj.io.stream.HavingStream; |
| import org.apache.solr.client.solrj.io.stream.ParallelStream; |
| import org.apache.solr.client.solrj.io.stream.RankStream; |
| import org.apache.solr.client.solrj.io.stream.RollupStream; |
| import org.apache.solr.client.solrj.io.stream.SortStream; |
| import org.apache.solr.client.solrj.io.stream.StatsStream; |
| import org.apache.solr.client.solrj.io.stream.StreamContext; |
| import org.apache.solr.client.solrj.io.stream.TupleStream; |
| import org.apache.solr.client.solrj.io.stream.UniqueStream; |
| 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.Bucket; |
| import org.apache.solr.client.solrj.io.stream.metrics.CountDistinctMetric; |
| import org.apache.solr.client.solrj.io.stream.metrics.CountMetric; |
| import org.apache.solr.client.solrj.io.stream.metrics.MaxMetric; |
| import org.apache.solr.client.solrj.io.stream.metrics.MeanMetric; |
| import org.apache.solr.client.solrj.io.stream.metrics.Metric; |
| import org.apache.solr.client.solrj.io.stream.metrics.MinMetric; |
| import org.apache.solr.client.solrj.io.stream.metrics.SumMetric; |
| import org.apache.solr.common.params.CommonParams; |
| import org.apache.solr.common.params.ModifiableSolrParams; |
| |
| import static org.apache.solr.client.solrj.io.stream.metrics.CountDistinctMetric.APPROX_COUNT_DISTINCT; |
| import static org.apache.solr.client.solrj.io.stream.metrics.CountDistinctMetric.COUNT_DISTINCT; |
| import static org.apache.solr.common.params.CommonParams.SORT; |
| |
| /** |
| * Table based on a Solr collection |
| */ |
| class SolrTable extends AbstractQueryableTable implements TranslatableTable { |
| private static final String DEFAULT_QUERY = "*:*"; |
| |
| 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; |
| } |
| |
| 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 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 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; |
| } |
| } |
| |
| 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, 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, |
| @SuppressWarnings("rawtypes") 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, |
| final String offset) { |
| // 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, offset); |
| } else { |
| if (buckets.isEmpty()) { |
| tupleStream = handleStats(zk, collection, q, metricPairs, fields); |
| } 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); |
| } |
| |
| StreamContext streamContext = new StreamContext(); |
| streamContext.setSolrClientCache(schema.getSolrClientCache()); |
| tupleStream.setStreamContext(streamContext); |
| |
| 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 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_DISTINCT: |
| return new CountDistinctMetric(metricPair.getValue()); |
| case APPROX_COUNT_DISTINCT: |
| return new CountDistinctMetric(metricPair.getValue(), true); |
| 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()); |
| } |
| } |
| |
| @SuppressWarnings({"rawtypes"}) |
| private TupleStream handleSelect(String zk, |
| String collection, |
| String query, |
| List<Map.Entry<String, Class>> fields, |
| List<Pair<String, String>> orders, |
| String limit, |
| String offset) 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.isEmpty()) { |
| params.add(SORT, getSort(orders)); |
| } else { |
| if (limit == null) { |
| params.add(SORT, "_version_ desc"); |
| fl = fl + ",_version_"; |
| } else { |
| params.add(SORT, "score desc"); |
| if (!fl.contains("score")) { |
| fl = fl + ",score"; |
| } |
| } |
| } |
| |
| params.add(CommonParams.FL, fl); |
| |
| if (offset != null && limit == null) { |
| throw new IOException("OFFSET without LIMIT not supported by Solr! Specify desired limit using 'FETCH NEXT <LIMIT> ROWS ONLY'"); |
| } |
| |
| if (limit != null) { |
| int limitInt = Integer.parseInt(limit); |
| // if there's an offset, then we need to fetch offset + limit rows from each shard and then sort accordingly |
| LimitStream limitStream; |
| if (offset != null) { |
| int offsetInt = Integer.parseInt(offset); |
| int rows = limitInt + offsetInt; |
| params.add(CommonParams.START, "0"); // tricky ... we need all rows up to limit + offset |
| params.add(CommonParams.ROWS, String.valueOf(rows)); |
| // re-sort all the streams back from the shards |
| StreamComparator streamSorter = new MultipleFieldComparator(getComps(orders)); |
| limitStream = new LimitStream(new SortStream(new CloudSolrStream(zk, collection, params), streamSorter), limitInt, offsetInt); |
| } else { |
| params.add(CommonParams.ROWS, limit); |
| limitStream = new LimitStream(new CloudSolrStream(zk, collection, params), limitInt); |
| } |
| return limitStream; |
| } 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(); |
| } |
| |
| @SuppressWarnings({"rawtypes"}) |
| 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(); |
| for (String field : fieldSet) { |
| |
| if (buf.length() > 0) { |
| buf.append(","); |
| } |
| |
| |
| buf.append(field); |
| } |
| |
| return buf.toString(); |
| } |
| |
| @SuppressWarnings({"unchecked", "rawtypes"}) |
| 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; |
| } |
| |
| @SuppressWarnings({"rawtypes"}) |
| 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 { |
| |
| Map<String, Class> fmap = new HashMap<>(); |
| for (Map.Entry<String, Class> entry : fields) { |
| fmap.put(entry.getKey(), entry.getValue()); |
| } |
| |
| 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); |
| } else { |
| for (Metric metric : metrics) { |
| Class c = fmap.get(metric.getIdentifier()); |
| if (Long.class.equals(c)) { |
| metric.outputLong = true; |
| } |
| } |
| } |
| |
| 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); |
| params.set(CommonParams.WT, CommonParams.JAVABIN); |
| //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", AndEvaluator.class) |
| .withFunctionName("or", OrEvaluator.class) |
| .withFunctionName("not", NotEvaluator.class) |
| .withFunctionName("eq", EqualToEvaluator.class) |
| .withFunctionName("gt", GreaterThanEvaluator.class) |
| .withFunctionName("lt", LessThanEvaluator.class) |
| .withFunctionName("val", RawValueEvaluator.class) |
| .withFunctionName("lteq", LessThanEqualToEvaluator.class) |
| .withFunctionName("having", HavingStream.class) |
| .withFunctionName("gteq", GreaterThanEqualToEvaluator.class); |
| |
| if (havingPredicate != null) { |
| RecursiveBooleanEvaluator booleanOperation = (RecursiveBooleanEvaluator) factory.constructEvaluator(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); |
| @SuppressWarnings("resource") final 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; |
| } |
| |
| @SuppressWarnings({"rawtypes"}) |
| 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; |
| } |
| |
| @SuppressWarnings({"rawtypes"}) |
| 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 { |
| |
| |
| Map<String, Class> fmap = new HashMap<>(); |
| for (Map.Entry<String, Class> f : fields) { |
| fmap.put(f.getKey(), f.getValue()); |
| } |
| |
| 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(); |
| } else { |
| for (Metric metric : metrics) { |
| Class c = fmap.get(metric.getIdentifier()); |
| if (Long.class.equals(c)) { |
| metric.outputLong = true; |
| } |
| } |
| } |
| |
| 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", AndEvaluator.class) |
| .withFunctionName("or", OrEvaluator.class) |
| .withFunctionName("not", NotEvaluator.class) |
| .withFunctionName("eq", EqualToEvaluator.class) |
| .withFunctionName("val", RawValueEvaluator.class) |
| .withFunctionName("gt", GreaterThanEvaluator.class) |
| .withFunctionName("lt", LessThanEvaluator.class) |
| .withFunctionName("lteq", LessThanEqualToEvaluator.class) |
| .withFunctionName("gteq", GreaterThanEqualToEvaluator.class); |
| |
| if (havingPredicate != null) { |
| RecursiveBooleanEvaluator booleanOperation = (RecursiveBooleanEvaluator) factory.constructEvaluator(StreamExpressionParser.parse(havingPredicate)); |
| tupleStream = new HavingStream(tupleStream, booleanOperation); |
| } |
| |
| if (lim != null) { |
| tupleStream = new LimitStream(tupleStream, limit); |
| } |
| |
| return tupleStream; |
| } |
| |
| @SuppressWarnings({"rawtypes"}) |
| 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); |
| params.set(CommonParams.WT, CommonParams.JAVABIN); |
| //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. |
| @SuppressWarnings("resource") final 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()]); |
| } |
| |
| @SuppressWarnings({"rawtypes"}) |
| private TupleStream handleStats(String zk, |
| String collection, |
| String query, |
| List<Pair<String, String>> metricPairs, |
| List<Map.Entry<String, Class>> fields) throws IOException { |
| |
| |
| Map<String, Class> fmap = new HashMap<>(); |
| for (Map.Entry<String, Class> entry : fields) { |
| fmap.put(entry.getKey(), entry.getValue()); |
| } |
| |
| ModifiableSolrParams solrParams = new ModifiableSolrParams(); |
| solrParams.add(CommonParams.Q, query); |
| Metric[] metrics = buildMetrics(metricPairs, false).toArray(new Metric[0]); |
| |
| for (Metric metric : metrics) { |
| Class c = fmap.get(metric.getIdentifier()); |
| if (Long.class.equals(c)) { |
| metric.outputLong = true; |
| } |
| } |
| |
| 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({"rawtypes", "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, String offset) { |
| return getTable().query(getProperties(), fields, query, order, buckets, metricPairs, limit, negativeQuery, havingPredicate, offset); |
| } |
| } |
| } |
