diff --git a/solr/core/src/java/org/apache/solr/handler/StreamHandler.java b/solr/core/src/java/org/apache/solr/handler/StreamHandler.java index 63e521c87cb..a4ac55934c3 100644 --- a/solr/core/src/java/org/apache/solr/handler/StreamHandler.java +++ b/solr/core/src/java/org/apache/solr/handler/StreamHandler.java @@ -171,6 +171,8 @@ public class StreamHandler extends RequestHandlerBase implements SolrCoreAware, .withFunctionName("predict", PredictEvaluator.class) .withFunctionName("regress", RegressionEvaluator.class) .withFunctionName("cov", CovarianceEvaluator.class) + .withFunctionName("conv", ConvolutionEvaluator.class) + // metrics .withFunctionName("min", MinMetric.class) .withFunctionName("max", MaxMetric.class) diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ConvolutionEvaluator.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ConvolutionEvaluator.java new file mode 100644 index 00000000000..6d6e3e33836 --- /dev/null +++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ConvolutionEvaluator.java @@ -0,0 +1,82 @@ +/* + * 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.client.solrj.io.stream; + +import java.io.IOException; +import java.util.List; +import java.util.ArrayList; + +import org.apache.commons.math3.util.MathArrays; +import org.apache.solr.client.solrj.io.Tuple; +import org.apache.solr.client.solrj.io.eval.ComplexEvaluator; +import org.apache.solr.client.solrj.io.eval.StreamEvaluator; +import org.apache.solr.client.solrj.io.stream.expr.Explanation; +import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType; +import org.apache.solr.client.solrj.io.stream.expr.Expressible; +import org.apache.solr.client.solrj.io.stream.expr.StreamExpression; +import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter; +import org.apache.solr.client.solrj.io.stream.expr.StreamFactory; + +public class ConvolutionEvaluator extends ComplexEvaluator implements Expressible { + + private static final long serialVersionUID = 1; + + public ConvolutionEvaluator(StreamExpression expression, StreamFactory factory) throws IOException { + super(expression, factory); + } + + public List evaluate(Tuple tuple) throws IOException { + StreamEvaluator colEval1 = subEvaluators.get(0); + StreamEvaluator colEval2 = subEvaluators.get(1); + + List numbers1 = (List)colEval1.evaluate(tuple); + List numbers2 = (List)colEval2.evaluate(tuple); + double[] column1 = new double[numbers1.size()]; + double[] column2 = new double[numbers2.size()]; + + for(int i=0; i conList = new ArrayList(); + for(double d :conArray) { + conList.add(d); + } + + return conList; + } + + @Override + public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException { + StreamExpression expression = new StreamExpression(factory.getFunctionName(getClass())); + return expression; + } + + @Override + public Explanation toExplanation(StreamFactory factory) throws IOException { + return new Explanation(nodeId.toString()) + .withExpressionType(ExpressionType.EVALUATOR) + .withFunctionName(factory.getFunctionName(getClass())) + .withImplementingClass(getClass().getName()) + .withExpression(toExpression(factory).toString()); + } +} \ No newline at end of file diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java index 31d3171b171..f674236a400 100644 --- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java +++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java @@ -5299,6 +5299,58 @@ public class StreamExpressionTest extends SolrCloudTestCase { assertTrue(tuples.get(0).getDouble("cov").equals(-625.0D)); } + @Test + public void testConvolution() throws Exception { + UpdateRequest updateRequest = new UpdateRequest(); + + int i=0; + while(i<50) { + updateRequest.add(id, "id_"+(++i),"test_dt", getDateString("2016", "5", "1"), "price_f", "400.00"); + } + + while(i<100) { + updateRequest.add(id, "id_"+(++i),"test_dt", getDateString("2015", "5", "1"), "price_f", "300.0"); + } + + while(i<150) { + updateRequest.add(id, "id_"+(++i),"test_dt", getDateString("2014", "5", "1"), "price_f", "500.0"); + } + + while(i<250) { + updateRequest.add(id, "id_"+(++i),"test_dt", getDateString("2013", "5", "1"), "price_f", "100.00"); + } + + updateRequest.commit(cluster.getSolrClient(), COLLECTIONORALIAS); + + String expr = "timeseries("+COLLECTIONORALIAS+", q=\"*:*\", start=\"2013-01-01T01:00:00.000Z\", " + + "end=\"2016-12-01T01:00:00.000Z\", " + + "gap=\"+1YEAR\", " + + "field=\"test_dt\", " + + "count(*), sum(price_f), max(price_f), min(price_f))"; + + String cexpr = "let(a="+expr+", b=select("+expr+",mult(2, count(*)) as nvalue), c=col(a, count(*)), d=col(b, nvalue), tuple(colc=c, cold=d, conv=conv(c,d)))"; + + ModifiableSolrParams paramsLoc = new ModifiableSolrParams(); + paramsLoc.set("expr", cexpr); + paramsLoc.set("qt", "/stream"); + + String url = cluster.getJettySolrRunners().get(0).getBaseUrl().toString()+"/"+COLLECTIONORALIAS; + TupleStream solrStream = new SolrStream(url, paramsLoc); + + StreamContext context = new StreamContext(); + solrStream.setStreamContext(context); + List tuples = getTuples(solrStream); + assertTrue(tuples.size() == 1); + List convolution = (List)(tuples.get(0)).get("conv"); + assertTrue(convolution.size() == 7); + assertTrue(convolution.get(0).equals(20000D)); + assertTrue(convolution.get(1).equals(20000D)); + assertTrue(convolution.get(2).equals(25000D)); + assertTrue(convolution.get(3).equals(30000D)); + assertTrue(convolution.get(4).equals(15000D)); + assertTrue(convolution.get(5).equals(10000D)); + assertTrue(convolution.get(6).equals(5000D)); + } @Test