mirror of https://github.com/apache/cassandra
Allow the selection of columns together with aggregates
Patch by Benjamin Lerer; reviewed by Robert Stupp for CASSANDRA-9767
This commit is contained in:
blerer
parent
d97fc9b77f
commit
2b99b5d354
4
NEWS.txt
4
NEWS.txt
|
|
@ -18,6 +18,10 @@ using the provided 'sstableupgrade' tool.
|
|||
|
||||
New features
|
||||
------------
|
||||
- Selecting columns,scalar functions, UDT fields, writetime or ttl together
|
||||
with aggregated is now possible. The value returned for the columns,
|
||||
scalar functions, UDT fields, writetime and ttl will be the ones for
|
||||
the first row matching the query.
|
||||
- Windows is now a supported platform. Powershell execution for startup scripts
|
||||
is highly recommended and can be enabled via an administrator command-prompt
|
||||
with: 'powershell set-executionpolicy unrestricted'
|
||||
|
|
|
|||
|
|
@ -1843,7 +1843,10 @@ A number of functions are provided to "convert" the native types into binary dat
|
|||
|
||||
h2(#aggregates). Aggregates
|
||||
|
||||
CQL3 distinguishes between built-in aggregates (so called 'native aggregates') and "user-defined aggregates":#udas. CQL3 includes several native aggregates, described below:
|
||||
Aggregate functions work on a set of rows. They receive values for each row and returns one value for the whole set.
|
||||
If @normal@ columns, @scalar functions@, @UDT@ fields, @writetime@ or @ttl@ are selected together with aggregate functions, the values returned for them will be the ones of the first row matching the query.
|
||||
|
||||
CQL3 distinguishes between built-in aggregates (so called 'native aggregates') and "user-defined aggregates":#udas. CQL3 includes several native aggregates, described below:
|
||||
|
||||
h3(#countFct). Count
|
||||
|
||||
|
|
|
|||
|
|
@ -19,7 +19,6 @@ package org.apache.cassandra.cql3.selection;
|
|||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
import com.google.common.collect.Iterables;
|
||||
|
|
@ -49,12 +48,6 @@ abstract class AbstractFunctionSelector<T extends Function> extends Selector
|
|||
if (factories.doesAggregation())
|
||||
throw new InvalidRequestException("aggregate functions cannot be used as arguments of aggregate functions");
|
||||
}
|
||||
else
|
||||
{
|
||||
if (factories.doesAggregation() && !factories.containsOnlyAggregateFunctions())
|
||||
throw new InvalidRequestException(String.format("arguments of function %s must be either all aggregates or no aggregates",
|
||||
fun.name()));
|
||||
}
|
||||
|
||||
return new Factory()
|
||||
{
|
||||
|
|
@ -109,7 +102,7 @@ abstract class AbstractFunctionSelector<T extends Function> extends Selector
|
|||
|
||||
public boolean isAggregateSelectorFactory()
|
||||
{
|
||||
return fun.isAggregate() || factories.containsOnlyAggregateFunctions();
|
||||
return fun.isAggregate() || factories.doesAggregation();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -485,9 +485,6 @@ public abstract class Selection
|
|||
factories.containsTTLSelectorFactory());
|
||||
|
||||
this.factories = factories;
|
||||
|
||||
if (factories.doesAggregation() && !factories.containsOnlyAggregateFunctions())
|
||||
throw new InvalidRequestException("the select clause must either contain only aggregates or no aggregate");
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -506,7 +503,7 @@ public abstract class Selection
|
|||
|
||||
public boolean isAggregate()
|
||||
{
|
||||
return factories.containsOnlyAggregateFunctions();
|
||||
return factories.doesAggregation();
|
||||
}
|
||||
|
||||
protected Selectors newSelectors() throws InvalidRequestException
|
||||
|
|
@ -523,7 +520,7 @@ public abstract class Selection
|
|||
|
||||
public boolean isAggregate()
|
||||
{
|
||||
return factories.containsOnlyAggregateFunctions();
|
||||
return factories.doesAggregation();
|
||||
}
|
||||
|
||||
public List<ByteBuffer> getOutputRow(int protocolVersion) throws InvalidRequestException
|
||||
|
|
|
|||
|
|
@ -108,18 +108,6 @@ final class SelectorFactories implements Iterable<Selector.Factory>
|
|||
factories.add(SimpleSelector.newFactory(def, index));
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if this <code>SelectorFactories</code> contains only factories for aggregates.
|
||||
*
|
||||
* @return <code>true</code> if this <code>SelectorFactories</code> contains only factories for aggregates,
|
||||
* <code>false</code> otherwise.
|
||||
*/
|
||||
public boolean containsOnlyAggregateFunctions()
|
||||
{
|
||||
int size = factories.size();
|
||||
return size != 0 && numberOfAggregateFactories == size;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether the selector built by this factory does aggregation or not (either directly or in a sub-selector).
|
||||
*
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ public final class SimpleSelector extends Selector
|
|||
private final int idx;
|
||||
private final AbstractType<?> type;
|
||||
private ByteBuffer current;
|
||||
private boolean isSet;
|
||||
|
||||
public static Factory newFactory(final ColumnDefinition def, final int idx)
|
||||
{
|
||||
|
|
@ -64,7 +65,11 @@ public final class SimpleSelector extends Selector
|
|||
@Override
|
||||
public void addInput(int protocolVersion, ResultSetBuilder rs) throws InvalidRequestException
|
||||
{
|
||||
current = rs.current.get(idx);
|
||||
if (!isSet)
|
||||
{
|
||||
isSet = true;
|
||||
current = rs.current.get(idx);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -76,6 +81,7 @@ public final class SimpleSelector extends Selector
|
|||
@Override
|
||||
public void reset()
|
||||
{
|
||||
isSet = false;
|
||||
current = null;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@
|
|||
package org.apache.cassandra.cql3.selection;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.Collections;
|
||||
|
||||
import org.apache.cassandra.config.ColumnDefinition;
|
||||
import org.apache.cassandra.cql3.ColumnSpecification;
|
||||
|
|
@ -34,6 +33,7 @@ final class WritetimeOrTTLSelector extends Selector
|
|||
private final int idx;
|
||||
private final boolean isWritetime;
|
||||
private ByteBuffer current;
|
||||
private boolean isSet;
|
||||
|
||||
public static Factory newFactory(final ColumnDefinition def, final int idx, final boolean isWritetime)
|
||||
{
|
||||
|
|
@ -73,6 +73,11 @@ final class WritetimeOrTTLSelector extends Selector
|
|||
|
||||
public void addInput(int protocolVersion, ResultSetBuilder rs)
|
||||
{
|
||||
if (isSet)
|
||||
return;
|
||||
|
||||
isSet = true;
|
||||
|
||||
if (isWritetime)
|
||||
{
|
||||
long ts = rs.timestamps[idx];
|
||||
|
|
@ -92,6 +97,7 @@ final class WritetimeOrTTLSelector extends Selector
|
|||
|
||||
public void reset()
|
||||
{
|
||||
isSet = false;
|
||||
current = null;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -24,19 +24,26 @@ import java.util.Date;
|
|||
import java.util.TimeZone;
|
||||
|
||||
import org.apache.commons.lang3.time.DateUtils;
|
||||
import org.junit.Assert;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.apache.cassandra.cql3.CQLTester;
|
||||
import org.apache.cassandra.cql3.QueryProcessor;
|
||||
import org.apache.cassandra.cql3.UntypedResultSet;
|
||||
import org.apache.cassandra.cql3.UntypedResultSet.Row;
|
||||
import org.apache.cassandra.cql3.functions.Functions;
|
||||
import org.apache.cassandra.cql3.functions.UDAggregate;
|
||||
import org.apache.cassandra.cql3.CQLTester;
|
||||
import org.apache.cassandra.exceptions.FunctionExecutionException;
|
||||
import org.apache.cassandra.exceptions.InvalidRequestException;
|
||||
import org.apache.cassandra.service.ClientState;
|
||||
import org.apache.cassandra.transport.Event;
|
||||
import org.apache.cassandra.transport.messages.ResultMessage;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
public class AggregationTest extends CQLTester
|
||||
{
|
||||
@Test
|
||||
|
|
@ -84,6 +91,116 @@ public class AggregationTest extends CQLTester
|
|||
assertRows(execute("SELECT COUNT(b), count(c), count(e), count(f) FROM %s"), row(4L, 3L, 3L, 3L));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAggregateWithColumns() throws Throwable
|
||||
{
|
||||
createTable("CREATE TABLE %s (a int, b int, c int, primary key (a, b))");
|
||||
|
||||
// Test with empty table
|
||||
assertColumnNames(execute("SELECT count(b), max(b) as max, b, c as first FROM %s"),
|
||||
"system.count(b)", "max", "b", "first");
|
||||
assertRows(execute("SELECT count(b), max(b) as max, b, c as first FROM %s"),
|
||||
row(0L, null, null, null));
|
||||
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (1, 2, null)");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (2, 4, 6)");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (4, 8, 12)");
|
||||
|
||||
assertRows(execute("SELECT count(b), max(b) as max, b, c as first FROM %s"),
|
||||
row(3L, 8, 2, null));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAggregateWithUdtFields() throws Throwable
|
||||
{
|
||||
String myType = createType("CREATE TYPE %s (x int)");
|
||||
createTable("CREATE TABLE %s (a int primary key, b frozen<" + myType + ">, c frozen<" + myType + ">)");
|
||||
|
||||
// Test with empty table
|
||||
assertColumnNames(execute("SELECT count(b.x), max(b.x) as max, b.x, c.x as first FROM %s"),
|
||||
"system.count(b.x)", "max", "b.x", "first");
|
||||
assertRows(execute("SELECT count(b.x), max(b.x) as max, b.x, c.x as first FROM %s"),
|
||||
row(0L, null, null, null));
|
||||
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (1, {x:2}, null)");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (2, {x:4}, {x:6})");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (4, {x:8}, {x:12})");
|
||||
|
||||
assertRows(execute("SELECT count(b.x), max(b.x) as max, b.x, c.x as first FROM %s"),
|
||||
row(3L, 8, 2, null));
|
||||
|
||||
assertInvalidMessage("Invalid field selection: max(b) of type blob is not a user type",
|
||||
"SELECT max(b).x as max FROM %s");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAggregateWithFunctions() throws Throwable
|
||||
{
|
||||
createTable("CREATE TABLE %s (a int, b double, c double, primary key(a, b))");
|
||||
|
||||
String copySign = createFunction(KEYSPACE,
|
||||
"double, double",
|
||||
"CREATE OR REPLACE FUNCTION %s(magnitude double, sign double) " +
|
||||
"RETURNS NULL ON NULL INPUT " +
|
||||
"RETURNS double " +
|
||||
"LANGUAGE JAVA " +
|
||||
"AS 'return Double.valueOf(Math.copySign(magnitude, sign));';");
|
||||
|
||||
// Test with empty table
|
||||
assertColumnNames(execute("SELECT count(b), max(b) as max, " + copySign + "(b, c), " + copySign + "(c, b) as first FROM %s"),
|
||||
"system.count(b)", "max", copySign + "(b, c)", "first");
|
||||
assertRows(execute("SELECT count(b), max(b) as max, " + copySign + "(b, c), " + copySign + "(c, b) as first FROM %s"),
|
||||
row(0L, null, null, null));
|
||||
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (0, -1.2, 2.1)");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (0, 1.3, -3.4)");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (0, 1.4, 1.2)");
|
||||
|
||||
assertRows(execute("SELECT count(b), max(b) as max, " + copySign + "(b, c), " + copySign + "(c, b) as first FROM %s"),
|
||||
row(3L, 1.4, 1.2, -2.1));
|
||||
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (1, -1.2, null)");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (1, 1.3, -3.4)");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (1, 1.4, 1.2)");
|
||||
assertRows(execute("SELECT count(b), max(b) as max, " + copySign + "(b, c), " + copySign + "(c, b) as first FROM %s WHERE a = 1"),
|
||||
row(3L, 1.4, null, null));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAggregateWithWithWriteTimeOrTTL() throws Throwable
|
||||
{
|
||||
createTable("CREATE TABLE %s (a int primary key, b int, c int)");
|
||||
|
||||
// Test with empty table
|
||||
assertColumnNames(execute("SELECT count(writetime(b)), min(ttl(b)) as min, writetime(b), ttl(c) as first FROM %s"),
|
||||
"system.count(writetime(b))", "min", "writetime(b)", "first");
|
||||
assertRows(execute("SELECT count(writetime(b)), min(ttl(b)) as min, writetime(b), ttl(c) as first FROM %s"),
|
||||
row(0L, null, null, null));
|
||||
|
||||
long today = System.currentTimeMillis() * 1000;
|
||||
long yesterday = today - (DateUtils.MILLIS_PER_DAY * 1000);
|
||||
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (1, 2, null) USING TTL 5;");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (2, 4, 6) USING TTL 2;");
|
||||
execute("INSERT INTO %s (a, b, c) VALUES (4, 8, 12) USING TIMESTAMP " + yesterday );
|
||||
|
||||
assertRows(execute("SELECT count(writetime(b)), count(ttl(b)) FROM %s"),
|
||||
row(3L, 2L));
|
||||
|
||||
UntypedResultSet resultSet = execute("SELECT min(ttl(b)), ttl(b) FROM %s");
|
||||
assertEquals(1, resultSet.size());
|
||||
Row row = resultSet.one();
|
||||
assertTrue(row.getInt("ttl(b)") > 4);
|
||||
assertTrue(row.getInt("system.min(ttl(b))") <= 2);
|
||||
|
||||
resultSet = execute("SELECT min(writetime(b)), writetime(b) FROM %s");
|
||||
assertEquals(1, resultSet.size());
|
||||
row = resultSet.one();
|
||||
|
||||
assertTrue(row.getLong("writetime(b)") >= today);
|
||||
assertTrue(row.getLong("system.min(writetime(b))") == yesterday);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFunctionsWithCompactStorage() throws Throwable
|
||||
{
|
||||
|
|
@ -113,8 +230,6 @@ public class AggregationTest extends CQLTester
|
|||
execute("INSERT INTO %s (a, b, c) VALUES (1, 3, 8)");
|
||||
|
||||
assertInvalidSyntax("SELECT max(b), max(c) FROM %s WHERE max(a) = 1");
|
||||
assertInvalidMessage("only aggregates or no aggregate", "SELECT max(b), c FROM %s");
|
||||
assertInvalidMessage("only aggregates or no aggregate", "SELECT b, max(c) FROM %s");
|
||||
assertInvalidMessage("aggregate functions cannot be used as arguments of aggregate functions", "SELECT max(sum(c)) FROM %s");
|
||||
assertInvalidSyntax("SELECT COUNT(2) FROM %s");
|
||||
}
|
||||
|
|
@ -155,8 +270,8 @@ public class AggregationTest extends CQLTester
|
|||
assertRows(execute("SELECT " + copySign + "(max(c), min(c)) FROM %s"), row(-1.4));
|
||||
assertRows(execute("SELECT " + copySign + "(c, d) FROM %s"), row(1.2), row(-1.3), row(1.4));
|
||||
assertRows(execute("SELECT max(" + copySign + "(c, d)) FROM %s"), row(1.4));
|
||||
assertInvalidMessage("must be either all aggregates or no aggregates", "SELECT " + copySign + "(c, max(c)) FROM %s");
|
||||
assertInvalidMessage("must be either all aggregates or no aggregates", "SELECT " + copySign + "(max(c), c) FROM %s");
|
||||
assertRows(execute("SELECT " + copySign + "(c, max(c)) FROM %s"), row(1.2));
|
||||
assertRows(execute("SELECT " + copySign + "(max(c), c) FROM %s"), row(-1.4));;
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -824,10 +939,10 @@ public class AggregationTest extends CQLTester
|
|||
"STYPE int");
|
||||
|
||||
ResultMessage.Prepared prepared = QueryProcessor.prepare("SELECT " + a + "(b) FROM " + otherKS + ".jsdp", ClientState.forInternalCalls(), false);
|
||||
Assert.assertNotNull(QueryProcessor.instance.getPrepared(prepared.statementId));
|
||||
assertNotNull(QueryProcessor.instance.getPrepared(prepared.statementId));
|
||||
|
||||
execute("DROP AGGREGATE " + a + "(int)");
|
||||
Assert.assertNull(QueryProcessor.instance.getPrepared(prepared.statementId));
|
||||
assertNull(QueryProcessor.instance.getPrepared(prepared.statementId));
|
||||
|
||||
//
|
||||
|
||||
|
|
@ -836,11 +951,11 @@ public class AggregationTest extends CQLTester
|
|||
"STYPE int");
|
||||
|
||||
prepared = QueryProcessor.prepare("SELECT " + a + "(b) FROM " + otherKS + ".jsdp", ClientState.forInternalCalls(), false);
|
||||
Assert.assertNotNull(QueryProcessor.instance.getPrepared(prepared.statementId));
|
||||
assertNotNull(QueryProcessor.instance.getPrepared(prepared.statementId));
|
||||
|
||||
execute("DROP KEYSPACE " + otherKS + ";");
|
||||
|
||||
Assert.assertNull(QueryProcessor.instance.getPrepared(prepared.statementId));
|
||||
assertNull(QueryProcessor.instance.getPrepared(prepared.statementId));
|
||||
}
|
||||
finally
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in New Issue