in solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/FacetStream.java [122:395]
public FacetStream(StreamExpression expression, StreamFactory factory) throws IOException {
// grab all parameters out
String collectionName = factory.getValueOperand(expression, 0);
if (collectionName.indexOf('"') > -1) {
collectionName = collectionName.replace("\"", "").replace(" ", "");
}
List<StreamExpressionNamedParameter> namedParams = factory.getNamedOperands(expression);
StreamExpressionNamedParameter bucketExpression =
factory.getNamedOperand(expression, "buckets");
StreamExpressionNamedParameter bucketSortExpression =
factory.getNamedOperand(expression, "bucketSorts");
List<StreamExpression> metricExpressions =
factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, Metric.class);
StreamExpressionNamedParameter bucketLimitExpression =
factory.getNamedOperand(expression, "bucketSizeLimit");
StreamExpressionNamedParameter zkHostExpression = factory.getNamedOperand(expression, "zkHost");
StreamExpressionNamedParameter rowsExpression = factory.getNamedOperand(expression, "rows");
StreamExpressionNamedParameter offsetExpression = factory.getNamedOperand(expression, "offset");
StreamExpressionNamedParameter overfetchExpression =
factory.getNamedOperand(expression, "overfetch");
StreamExpressionNamedParameter refineExpression = factory.getNamedOperand(expression, "refine");
StreamExpressionNamedParameter methodExpression = factory.getNamedOperand(expression, "method");
// Validate there are no unknown parameters
if (expression.getParameters().size() != 1 + namedParams.size() + metricExpressions.size()) {
throw new IOException(
String.format(Locale.ROOT, "invalid expression %s - unknown operands found", expression));
}
// Collection Name
if (null == collectionName) {
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - collectionName expected as first operand",
expression));
}
// Named parameters - passed directly to solr as SolrParams
if (0 == namedParams.size()) {
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - at least one named parameter expected. eg. 'q=*:*'",
expression));
}
// pull out known named params
ModifiableSolrParams params = new ModifiableSolrParams();
for (StreamExpressionNamedParameter namedParam : namedParams) {
if (!namedParam.getName().equals("zkHost")
&& !namedParam.getName().equals("buckets")
&& !namedParam.getName().equals("bucketSorts")
&& !namedParam.getName().equals("bucketSizeLimit")
&& !namedParam.getName().equals("method")
&& !namedParam.getName().equals("offset")
&& !namedParam.getName().equals("rows")
&& !namedParam.getName().equals("refine")
&& !namedParam.getName().equals("overfetch")) {
params.add(namedParam.getName(), namedParam.getParameter().toString().trim());
}
}
if (params.get("q") == null) {
params.set("q", "*:*");
}
// buckets, required - comma separated
Bucket[] buckets = null;
if (null != bucketExpression) {
if (bucketExpression.getParameter() instanceof StreamExpressionValue) {
String[] keys =
((StreamExpressionValue) bucketExpression.getParameter()).getValue().split(",");
if (0 != keys.length) {
buckets = new Bucket[keys.length];
for (int idx = 0; idx < keys.length; ++idx) {
buckets[idx] = new Bucket(keys[idx].trim());
}
}
}
}
if (null == buckets) {
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - at least one bucket expected. eg. 'buckets=\"name\"'",
expression));
}
// Construct the metrics
Metric[] metrics = new Metric[metricExpressions.size()];
for (int idx = 0; idx < metricExpressions.size(); ++idx) {
metrics[idx] = factory.constructMetric(metricExpressions.get(idx));
}
if (metrics.length == 0) {
metrics = new Metric[1];
metrics[0] = new CountMetric();
}
String bucketSortString = null;
if (bucketSortExpression == null) {
bucketSortString = metrics[0].getIdentifier() + " desc";
} else {
bucketSortString = ((StreamExpressionValue) bucketSortExpression.getParameter()).getValue();
if (bucketSortString.contains("(")
&& metricExpressions.size() == 0
&& (!bucketSortString.equals("count(*) desc")
&& !bucketSortString.equals("count(*) asc"))) {
// Attempting bucket sort on a metric that is not going to be calculated.
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - the bucketSort is being performed on a metric that is not being calculated.",
expression));
}
}
FieldComparator[] bucketSorts = parseBucketSorts(bucketSortString, buckets);
if (null == bucketSorts || 0 == bucketSorts.length) {
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - at least one bucket sort expected. eg. 'bucketSorts=\"name asc\"'",
expression));
}
boolean refine = false;
if (refineExpression != null) {
String refineStr = ((StreamExpressionValue) refineExpression.getParameter()).getValue();
if (refineStr != null) {
refine = Boolean.parseBoolean(refineStr);
}
}
if (bucketLimitExpression != null
&& (rowsExpression != null || offsetExpression != null || overfetchExpression != null)) {
throw new IOException("bucketSizeLimit is incompatible with rows, offset and overfetch.");
}
String methodStr = null;
if (methodExpression != null) {
methodStr = ((StreamExpressionValue) methodExpression.getParameter()).getValue();
}
int overfetchInt = 250;
if (overfetchExpression != null) {
String overfetchStr = ((StreamExpressionValue) overfetchExpression.getParameter()).getValue();
overfetchInt = Integer.parseInt(overfetchStr);
}
int offsetInt = 0;
if (offsetExpression != null) {
String offsetStr = ((StreamExpressionValue) offsetExpression.getParameter()).getValue();
offsetInt = Integer.parseInt(offsetStr);
}
int rowsInt = Integer.MIN_VALUE;
int bucketLimit = Integer.MIN_VALUE;
boolean bucketLimitSet = false;
if (null != rowsExpression) {
String rowsStr = ((StreamExpressionValue) rowsExpression.getParameter()).getValue();
try {
rowsInt = Integer.parseInt(rowsStr);
if (rowsInt <= 0 && rowsInt != -1) {
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - limit '%s' must be greater than 0 or -1.",
expression,
rowsStr));
}
// Rows is set so configure the bucketLimitSize
if (rowsInt == -1) {
bucketLimit = rowsInt = Integer.MAX_VALUE;
} else if (overfetchInt == -1) {
bucketLimit = Integer.MAX_VALUE;
} else {
bucketLimit = offsetInt + overfetchInt + rowsInt;
}
} catch (NumberFormatException e) {
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - limit '%s' is not a valid integer.",
expression,
rowsStr));
}
}
if (bucketLimitExpression != null) {
String bucketLimitStr =
((StreamExpressionValue) bucketLimitExpression.getParameter()).getValue();
try {
bucketLimit = Integer.parseInt(bucketLimitStr);
bucketLimitSet = true;
if (bucketLimit <= 0 && bucketLimit != -1) {
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - bucketSizeLimit '%s' must be greater than 0 or -1.",
expression,
bucketLimitStr));
}
// Bucket limit is set. So set rows.
if (bucketLimit == -1) {
rowsInt = bucketLimit = Integer.MAX_VALUE;
} else {
rowsInt = bucketLimit;
}
} catch (NumberFormatException e) {
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - bucketSizeLimit '%s' is not a valid integer.",
expression,
bucketLimitStr));
}
}
if (rowsExpression == null && bucketLimitExpression == null) {
rowsInt = 10;
if (overfetchInt == -1) {
bucketLimit = Integer.MAX_VALUE;
} else {
bucketLimit = offsetInt + overfetchInt + rowsInt;
}
}
// zkHost, optional - if not provided then will look into factory list to get
String zkHost = null;
if (null == zkHostExpression) {
zkHost = factory.getCollectionZkHost(collectionName);
if (zkHost == null) {
zkHost = factory.getDefaultZkHost();
}
} else if (zkHostExpression.getParameter() instanceof StreamExpressionValue) {
zkHost = ((StreamExpressionValue) zkHostExpression.getParameter()).getValue();
}
if (null == zkHost) {
throw new IOException(
String.format(
Locale.ROOT,
"invalid expression %s - zkHost not found for collection '%s'",
expression,
collectionName));
}
// We've got all the required items
init(
collectionName,
params,
buckets,
bucketSorts,
metrics,
rowsInt,
offsetInt,
bucketLimit,
refine,
methodStr,
bucketLimitSet,
overfetchInt,
zkHost);
}