SOLR-8530: Add HavingStream to Streaming API and StreamingExpressions

This commit is contained in:
Joel Bernstein 2017-01-03 13:09:49 -05:00
parent 5b1f6b2ba4
commit 1da283ef2c
11 changed files with 882 additions and 0 deletions

View File

@ -0,0 +1,101 @@
/*
* 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.ops;
import java.io.IOException;
import java.util.List;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
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 AndOperation implements BooleanOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
protected BooleanOperation leftOperand;
protected BooleanOperation rightOperand;
public void operate(Tuple tuple) {
leftOperand.operate(tuple);
rightOperand.operate(tuple);
}
public AndOperation(BooleanOperation leftOperand, BooleanOperation rightOperand) {
this.leftOperand = leftOperand;
this.rightOperand = rightOperand;
}
public AndOperation(StreamExpression expression, StreamFactory factory) throws IOException {
List<StreamExpression> operationExpressions = factory.getExpressionOperandsRepresentingTypes(expression, BooleanOperation.class);
if(operationExpressions != null && operationExpressions.size() == 2) {
StreamExpression left = operationExpressions.get(0);
StreamOperation leftOp = factory.constructOperation(left);
if(leftOp instanceof BooleanOperation) {
leftOperand = (BooleanOperation) leftOp;
} else {
throw new IOException("The And/Or Operation requires a BooleanOperation.");
}
StreamExpression right = operationExpressions.get(1);
StreamOperation rightOp = factory.constructOperation(right);
if(rightOp instanceof BooleanOperation) {
rightOperand = (BooleanOperation) rightOp;
} else {
throw new IOException("The And/Or Operation requires a BooleanOperation.");
}
} else {
throw new IOException("The And/Or Operation requires a BooleanOperations.");
}
}
public boolean evaluate() {
return leftOperand.evaluate() && rightOperand.evaluate();
}
@Override
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(leftOperand instanceof Expressible) {
expression.addParameter(leftOperand.toExpression(factory));
} else {
throw new IOException("This left operand of the AndOperation contains a non-expressible operation - it cannot be converted to an expression");
}
if(rightOperand instanceof Expressible) {
expression.addParameter(rightOperand.toExpression(factory));
} else {
throw new IOException("This the right operand of the AndOperation contains a non-expressible operation - it cannot be converted to an expression");
}
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

View File

@ -0,0 +1,24 @@
/*
* 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.ops;
import org.apache.solr.client.solrj.io.Tuple;
public interface BooleanOperation extends StreamOperation {
public abstract boolean evaluate();
}

View File

@ -0,0 +1,70 @@
/*
* 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.ops;
import java.io.IOException;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
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.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class EqualsOperation extends LeafOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
public void operate(Tuple tuple) {
this.tuple = tuple;
}
public EqualsOperation(String field, double val) {
super(field, val);
}
public EqualsOperation(StreamExpression expression, StreamFactory factory) throws IOException {
super(expression, factory);
}
public boolean evaluate() {
Double d = tuple.getDouble(field);
if(d == null) {
return false;
}
return d.doubleValue() == val;
}
public StreamExpression toExpression(StreamFactory factory) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
expression.addParameter(field);
expression.addParameter(Double.toString(val));
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

View File

@ -0,0 +1,70 @@
/*
* 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.ops;
import java.io.IOException;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
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.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class GreaterThanEqualToOperation extends LeafOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
public void operate(Tuple tuple) {
this.tuple = tuple;
}
public GreaterThanEqualToOperation(String field, double val) {
super(field, val);
}
public GreaterThanEqualToOperation(StreamExpression expression, StreamFactory factory) throws IOException {
super(expression, factory);
}
public boolean evaluate() {
Double d = tuple.getDouble(field);
if(d == null) {
return false;
}
return d.doubleValue() >= val;
}
public StreamExpression toExpression(StreamFactory factory) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
expression.addParameter(field);
expression.addParameter(Double.toString(val));
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

View File

@ -0,0 +1,70 @@
/*
* 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.ops;
import java.io.IOException;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
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.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class GreaterThanOperation extends LeafOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
public void operate(Tuple tuple) {
this.tuple = tuple;
}
public GreaterThanOperation(String field, double val) {
super(field, val);
}
public GreaterThanOperation(StreamExpression expression, StreamFactory factory) throws IOException {
super(expression, factory);
}
public boolean evaluate() {
Double d = tuple.getDouble(field);
if(d == null) {
return false;
}
return d.doubleValue() > val;
}
public StreamExpression toExpression(StreamFactory factory) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
expression.addParameter(field);
expression.addParameter(Double.toString(val));
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

View File

@ -0,0 +1,59 @@
/*
* 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.ops;
import java.io.IOException;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
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.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public abstract class LeafOperation implements BooleanOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
protected String field;
protected Double val;
protected Tuple tuple;
public void operate(Tuple tuple) {
this.tuple = tuple;
}
public LeafOperation(String field, double val) {
this.field = field;
this.val = val;
}
public LeafOperation(StreamExpression expression, StreamFactory factory) throws IOException {
this.field = factory.getValueOperand(expression, 0);
this.val = Double.parseDouble(factory.getValueOperand(expression, 1));
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

View File

@ -0,0 +1,70 @@
/*
* 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.ops;
import java.io.IOException;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
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.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class LessThanEqualToOperation extends LeafOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
public void operate(Tuple tuple) {
this.tuple = tuple;
}
public LessThanEqualToOperation(String field, double val) {
super(field, val);
}
public LessThanEqualToOperation(StreamExpression expression, StreamFactory factory) throws IOException {
super(expression, factory);
}
public boolean evaluate() {
Double d = tuple.getDouble(field);
if(d == null) {
return true;
}
return d.doubleValue() <= val;
}
public StreamExpression toExpression(StreamFactory factory) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
expression.addParameter(field);
expression.addParameter(Double.toString(val));
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

View File

@ -0,0 +1,70 @@
/*
* 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.ops;
import java.io.IOException;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
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.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class LessThanOperation extends LeafOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
public void operate(Tuple tuple) {
this.tuple = tuple;
}
public LessThanOperation(String field, double val) {
super(field, val);
}
public LessThanOperation(StreamExpression expression, StreamFactory factory) throws IOException {
super(expression, factory);
}
public boolean evaluate() {
Double d = tuple.getDouble(field);
if(d == null) {
return true;
}
return d.doubleValue() < val;
}
public StreamExpression toExpression(StreamFactory factory) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
expression.addParameter(field);
expression.addParameter(Double.toString(val));
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

View File

@ -0,0 +1,87 @@
/*
* 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.ops;
import java.io.IOException;
import java.util.List;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
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 NotOperation implements BooleanOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
protected BooleanOperation operand;
public void operate(Tuple tuple) {
operand.operate(tuple);
}
public NotOperation(BooleanOperation operand) {
this.operand = operand;
}
public NotOperation(StreamExpression expression, StreamFactory factory) throws IOException {
List<StreamExpression> operationExpressions = factory.getExpressionOperandsRepresentingTypes(expression, BooleanOperation.class);
if(operationExpressions != null && operationExpressions.size() == 1) {
StreamExpression op = operationExpressions.get(0);
StreamOperation streamOp = factory.constructOperation(op);
if(streamOp instanceof BooleanOperation) {
operand = (BooleanOperation) streamOp;
} else {
throw new IOException("The NotOperation requires a BooleanOperation.");
}
} else {
throw new IOException("The NotOperation requires a BooleanOperations.");
}
}
public boolean evaluate() {
return !operand.evaluate();
}
@Override
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(operand instanceof Expressible) {
expression.addParameter(operand.toExpression(factory));
} else {
throw new IOException("The operand of the NotOperation contains a non-expressible operation - it cannot be converted to an expression");
}
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

View File

@ -0,0 +1,71 @@
/*
* 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.ops;
import java.io.IOException;
import java.util.UUID;
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 OrOperation extends AndOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
public OrOperation(BooleanOperation leftOperand, BooleanOperation rightOperand) {
super(leftOperand, rightOperand);
}
public OrOperation(StreamExpression expression, StreamFactory factory) throws IOException {
super(expression, factory);
}
public boolean evaluate() {
return leftOperand.evaluate() || rightOperand.evaluate();
}
@Override
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(leftOperand instanceof Expressible) {
expression.addParameter(leftOperand.toExpression(factory));
} else {
throw new IOException("This left operand of the OrOperation contains a non-expressible operation - it cannot be converted to an expression");
}
if(rightOperand instanceof Expressible) {
expression.addParameter(rightOperand.toExpression(factory));
} else {
throw new IOException("This the right operand of the OrOperation contains a non-expressible operation - it cannot be converted to an expression");
}
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

View File

@ -0,0 +1,190 @@
/*
* 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.ArrayList;
import java.util.List;
import java.util.Locale;
import org.apache.solr.client.solrj.io.Tuple;
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.BooleanOperation;
import org.apache.solr.client.solrj.io.ops.StreamOperation;
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.StreamExplanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
/**
* Iterates over a TupleStream and buffers Tuples that are equal based on a comparator.
* This allows tuples to be grouped by common field(s).
*
* The read() method emits one tuple per group. The fields of the emitted Tuple reflect the first tuple
* encountered in the group.
*
* Use the Tuple.getMaps() method to return all the Tuples in the group. This method returns
* a list of maps (including the group head), which hold the data for each Tuple in the group.
*
* Note: The ReducerStream requires that it's underlying stream be sorted and partitioned by the same
* fields as it's comparator.
*
**/
public class HavingStream extends TupleStream implements Expressible {
private static final long serialVersionUID = 1;
private TupleStream stream;
private BooleanOperation op;
private transient Tuple currentGroupHead;
public HavingStream(TupleStream stream, BooleanOperation op) throws IOException {
init(stream, op);
}
public HavingStream(StreamExpression expression, StreamFactory factory) throws IOException{
// grab all parameters out
List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
List<StreamExpression> operationExpressions = factory.getExpressionOperandsRepresentingTypes(expression, BooleanOperation.class);
// validate expression contains only what we want.
if(expression.getParameters().size() != streamExpressions.size() + 1){
throw new IOException(String.format(Locale.ROOT,"Invalid expression %s - unknown operands found", expression));
}
if(1 != streamExpressions.size()){
throw new IOException(String.format(Locale.ROOT,"Invalid expression %s - expecting a single stream but found %d",expression, streamExpressions.size()));
}
BooleanOperation booleanOperation = null;
if(operationExpressions != null && operationExpressions.size() == 1) {
StreamExpression ex = operationExpressions.get(0);
StreamOperation operation = factory.constructOperation(ex);
if(operation instanceof BooleanOperation) {
booleanOperation = (BooleanOperation) operation;
} else {
throw new IOException("The HavingStream requires a BooleanOperation. A StreamOperation was provided.");
}
} else {
throw new IOException("The HavingStream requires a BooleanOperation.");
}
init(factory.constructStream(streamExpressions.get(0)), booleanOperation);
}
private void init(TupleStream stream, BooleanOperation op) throws IOException{
this.stream = stream;
this.op = op;
}
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
// stream
if(includeStreams){
expression.addParameter(((Expressible) stream).toExpression(factory));
}
else{
expression.addParameter("<stream>");
}
if(op instanceof Expressible) {
expression.addParameter(op.toExpression(factory));
} else {
throw new IOException("This ReducerStream contains a non-expressible operation - it cannot be converted to an expression");
}
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
stream.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString())
.withHelpers(new Explanation[]{
op.toExplanation(factory)
});
}
public void setStreamContext(StreamContext context) {
this.stream.setStreamContext(context);
}
public List<TupleStream> children() {
List<TupleStream> l = new ArrayList<TupleStream>();
l.add(stream);
return l;
}
public void open() throws IOException {
stream.open();
}
public void close() throws IOException {
stream.close();
}
public Tuple read() throws IOException {
while(true) {
Tuple tuple = stream.read();
if(tuple.EOF) {
return tuple;
}
op.operate(tuple);
if(op.evaluate()) {
return tuple;
}
}
}
/** Return the stream sort - ie, the order in which records are returned */
public StreamComparator getStreamSort(){
return stream.getStreamSort();
}
public int getCost() {
return 0;
}
}