dsmiley commented on a change in pull request #1506: URL: https://github.com/apache/lucene-solr/pull/1506#discussion_r428932332
########## File path: solr/core/src/java/org/apache/solr/handler/sql/FilterCalciteConnection.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 java.lang.reflect.Type; +import java.sql.Array; +import java.sql.Blob; +import java.sql.CallableStatement; +import java.sql.Clob; +import java.sql.DatabaseMetaData; +import java.sql.NClob; +import java.sql.PreparedStatement; +import java.sql.SQLClientInfoException; +import java.sql.SQLException; +import java.sql.SQLWarning; +import java.sql.SQLXML; +import java.sql.Savepoint; +import java.sql.Statement; +import java.sql.Struct; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.Executor; + +import org.apache.calcite.adapter.java.JavaTypeFactory; +import org.apache.calcite.config.CalciteConnectionConfig; +import org.apache.calcite.jdbc.CalciteConnection; +import org.apache.calcite.jdbc.CalcitePrepare; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.linq4j.Queryable; +import org.apache.calcite.linq4j.tree.Expression; +import org.apache.calcite.schema.SchemaPlus; + +/** + * A filter that contains another {@link CalciteConnection} and + * allows adding pre- post-method behaviors. + */ +class FilterCalciteConnection implements CalciteConnection { Review comment: What a class... it reminds me of one of the benefits of Kotlin lang ########## File path: solr/core/src/java/org/apache/solr/handler/export/ExportWriter.java ########## @@ -216,14 +376,53 @@ public void write(OutputStream os) throws IOException { return; } + String expr = params.get(StreamParams.EXPR); + if (expr != null) { + StreamFactory streamFactory = initialStreamContext.getStreamFactory(); + try { + StreamExpression expression = StreamExpressionParser.parse(expr); + if (streamFactory.isEvaluator(expression)) { + streamExpression = new StreamExpression(StreamParams.TUPLE); + streamExpression.addParameter(new StreamExpressionNamedParameter(StreamParams.RETURN_VALUE, expression)); + } else { + streamExpression = expression; + } + } catch (Exception e) { + writeException(e, writer, true); + return; + } + streamContext = new StreamContext(); + streamContext.setRequestParams(params); + // nocommit enforce this? + streamContext.setLocal(true); + + streamContext.workerID = 0; + streamContext.numWorkers = 1; + streamContext.setSolrClientCache(initialStreamContext.getSolrClientCache()); + streamContext.setModelCache(initialStreamContext.getModelCache()); + streamContext.setObjectCache(initialStreamContext.getObjectCache()); + streamContext.put("core", req.getCore().getName()); Review comment: naming here is confusing -- name of the thing vs the thing itself. Do you have to adhere to these specific names? Perhaps "core-object" might be more clear than ambiguously solr-core. ########## File path: solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/StatsStream.java ########## @@ -266,91 +252,106 @@ public void close() throws IOException { } public Tuple read() throws IOException { - if(index == 0) { - ++index; + if(!done) { + done = true; return tuple; } else { - Map fields = new HashMap(); - fields.put("EOF", true); - Tuple tuple = new Tuple(fields); - return tuple; + return Tuple.EOF(); } } - private String getJsonFacetString(Metric[] _metrics) { - StringBuilder buf = new StringBuilder(); - appendJson(buf, _metrics); - return "{"+buf.toString()+"}"; + public StreamComparator getStreamSort() { + return null; } - private void appendJson(StringBuilder buf, - Metric[] _metrics) { - - int metricCount = 0; + private void addStats(ModifiableSolrParams params, Metric[] _metrics) { + Map<String, List<String>> m = new HashMap<>(); for(Metric metric : _metrics) { - String identifier = metric.getIdentifier(); - if(!identifier.startsWith("count(")) { - if(metricCount>0) { - buf.append(","); + String metricId = metric.getIdentifier(); + if(metricId.contains("(")) { + metricId = metricId.substring(0, metricId.length()-1); + String[] parts = metricId.split("\\("); + String function = parts[0]; + String column = parts[1]; + List<String> stats = m.get(column); + + if(stats == null) { + stats = new ArrayList<>(); } - if(identifier.startsWith("per(")) { - buf.append("\"facet_").append(metricCount).append("\":\"").append(identifier.replaceFirst("per", "percentile")).append('"'); - } else if(identifier.startsWith("std(")) { - buf.append("\"facet_").append(metricCount).append("\":\"").append(identifier.replaceFirst("std", "stddev")).append('"'); - } else { - buf.append("\"facet_").append(metricCount).append("\":\"").append(identifier).append('"'); + + if(!column.equals("*")) { + m.put(column, stats); + } + + if(function.equals("min")) { + stats.add("min"); + } else if(function.equals("max")) { + stats.add("max"); + } else if(function.equals("sum")) { + stats.add("sum"); + } else if(function.equals("avg")) { + stats.add("mean"); + } else if(function.equals("count")) { + this.doCount = true; } - ++metricCount; } } - } - private void getTuples(NamedList response, - Metric[] metrics) { + for(Entry<String, List<String>> entry : m.entrySet()) { + StringBuilder buf = new StringBuilder(); + List<String> stats = entry.getValue(); + buf.append("{!"); + + for(String stat : stats) { + buf.append(stat).append("=").append("true "); + } - this.tuple = new Tuple(new HashMap()); - NamedList facets = (NamedList)response.get("facets"); - fillTuple(tuple, facets, metrics); + buf.append("}").append(entry.getKey()); + params.add("stats.field", buf.toString()); + } } - private void fillTuple(Tuple t, - NamedList nl, - Metric[] _metrics) { + private Tuple getTuple(NamedList response) { + Tuple tuple = new Tuple(); + SolrDocumentList solrDocumentList = (SolrDocumentList) response.get("response"); + + long count = solrDocumentList.getNumFound(); - if(nl == null) { - return; + if(doCount) { + tuple.put("count(*)", count); } - int m = 0; - for(Metric metric : _metrics) { - String identifier = metric.getIdentifier(); - if(!identifier.startsWith("count(")) { - if(nl.get("facet_"+m) != null) { - Object d = nl.get("facet_" + m); - if(d instanceof Number) { - if (metric.outputLong) { - t.put(identifier, Math.round(((Number)d).doubleValue())); - } else { - t.put(identifier, ((Number)d).doubleValue()); - } - } else { - t.put(identifier, d); - } + if(count != 0) { + NamedList stats = (NamedList)response.get("stats"); + NamedList statsFields = (NamedList)stats.get("stats_fields"); + + for(int i=0; i<statsFields.size(); i++) { + String field = statsFields.getName(i); + NamedList theStats = (NamedList)statsFields.getVal(i); + for(int s=0; s<theStats.size(); s++) { + addStat(tuple, field, theStats.getName(s), theStats.getVal(s)); } - ++m; - } else { - long l = ((Number)nl.get("count")).longValue(); - t.put("count(*)", l); } } + return tuple; } public int getCost() { return 0; } - @Override - public StreamComparator getStreamSort() { - return null; + private void addStat(Tuple tuple, String field, String stat, Object val) { + if(stat.equals("mean")) { Review comment: Nitpick: I see this a lot in your changes here -- lack of a space following the "if". Sometimes you remember, sometimes you don't. ########## File path: solr/core/src/java/org/apache/solr/handler/export/ExportWriter.java ########## @@ -285,22 +484,48 @@ protected void addDocsToItemWriter(List<LeafReaderContext> leaves, IteratorWrite protected void writeDocs(SolrQueryRequest req, IteratorWriter.ItemWriter writer, Sort sort) throws IOException { List<LeafReaderContext> leaves = req.getSearcher().getTopReaderContext().leaves(); SortDoc sortDoc = getSortDoc(req.getSearcher(), sort.getSort()); - int count = 0; final int queueSize = Math.min(DOCUMENT_BATCH_SIZE, totalHits); SortQueue queue = new SortQueue(queueSize, sortDoc); SortDoc[] outDocs = new SortDoc[queueSize]; - while (count < totalHits) { - identifyLowestSortingUnexportedDocs(leaves, sortDoc, queue); - int outDocsIndex = transferBatchToArrayForOutput(queue, outDocs); - - count += (outDocsIndex + 1); - addDocsToItemWriter(leaves, writer, outDocs, outDocsIndex); + if (streamExpression != null) { + streamContext.put(SORT_DOCS_KEY, outDocs); + streamContext.put(SORT_QUEUE_KEY, queue); + streamContext.put(SORT_DOC_KEY, sortDoc); + streamContext.put(TOTAL_HITS_KEY, totalHits); + streamContext.put(EXPORT_WRITER_KEY, this); + streamContext.put(LEAF_READERS_KEY, leaves); + TupleStream tupleStream = createTupleStream(); + tupleStream.open(); + for (;;) { + final Tuple t = tupleStream.read(); + if (t == null) { + break; + } + if (t.EOF) { + break; + } + writer.add((MapWriter) ew -> t.writeMap(ew)); + } + tupleStream.close(); + } else { + int count = 0; + while (count < totalHits) { Review comment: you didn't want to use a for loop here for some reason? (count declaration could be brought in) ########## File path: solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/StatsStream.java ########## @@ -129,51 +121,46 @@ public StatsStream(StreamExpression expression, StreamFactory factory) throws IO zkHost = ((StreamExpressionValue)zkHostExpression.getParameter()).getValue(); } - // We've got all the required items - init(collectionName, params, metrics, zkHost); - } + /* + if(null == zkHost){ + throw new IOException(String.format(Locale.ROOT,"invalid expression %s - zkHost not found for collection '%s'",expression,collectionName)); + } + */ - public String getCollection() { - return this.collection; - } + // metrics, optional - if not provided then why are you using this? + Metric[] metrics = new Metric[metricExpressions.size()]; + for(int idx = 0; idx < metricExpressions.size(); ++idx){ + metrics[idx] = factory.constructMetric(metricExpressions.get(idx)); + } - private void init(String collection, - SolrParams params, - Metric[] metrics, - String zkHost) throws IOException { - this.zkHost = zkHost; - this.collection = collection; - this.metrics = metrics; - this.params = params; + // We've got all the required items + init(zkHost, collectionName, params, metrics); } @Override public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException { + // functionName(collectionName, param1, param2, ..., paramN, sort="comp", sum(fieldA), avg(fieldB)) + // function name StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass())); + // collection - if(collection.indexOf(',') > -1) { - expression.addParameter("\""+collection+"\""); - } else { - expression.addParameter(collection); - } + expression.addParameter(collection); // parameters - ModifiableSolrParams tmpParams = new ModifiableSolrParams(params); - - for (Entry<String, String[]> param : tmpParams.getMap().entrySet()) { - expression.addParameter(new StreamExpressionNamedParameter(param.getKey(), - String.join(",", param.getValue()))); + ModifiableSolrParams mParams = new ModifiableSolrParams(params); + for (Entry<String, String[]> param : mParams.getMap().entrySet()) { Review comment: The ".getMap().entrySet()" portion here is needless; a SolrParams is Iterable. I'll file a PR to make this change comprehensively as I see it commonly in streaming expressions code. ########## File path: solr/core/src/test/org/apache/solr/core/HelloStream.java ########## @@ -67,14 +65,10 @@ public void close() throws IOException { @Override public Tuple read() throws IOException { if (isSentHelloWorld) { - Map m = new HashMap(); - m.put("EOF", true); - return new Tuple(m); + return Tuple.EOF(); } else { isSentHelloWorld = true; - Map m = new HashMap<>(); - m.put("msg", "Hello World!"); - return new Tuple(m); + return new Tuple("msg", "Hello World!"); Review comment: Thanks for cleaning up needless verbosity! ########## File path: solr/solrj/src/java/org/apache/solr/common/params/StreamParams.java ########## @@ -0,0 +1,41 @@ +/* + * 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.common.params; + +/** + * Stream Parameters and Properties. Review comment: Lets say "Streaming Expressions" to be more clear :-) ########## File path: solr/solrj/src/java/org/apache/solr/client/solrj/io/Tuple.java ########## @@ -44,24 +45,35 @@ public boolean EOF; public boolean EXCEPTION; - public Map fields = new HashMap(); + public Map<Object, Object> fields = new HashMap<>(2); public List<String> fieldNames; public Map<String, String> fieldLabels; - public Tuple(){ + public Tuple() { // just an empty tuple } - public Tuple(Map fields) { - if(fields.containsKey("EOF")) { - EOF = true; + public Tuple(Map<?, ?> fields) { Review comment: It's worth a javadoc comment to say this is a copy-constructor ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@lucene.apache.org For additional commands, e-mail: issues-h...@lucene.apache.org
