From 2b99b5d3542f8695e419e851c2f07758ef46373d Mon Sep 17 00:00:00 2001 From: blerer Date: Thu, 16 Jul 2015 22:14:45 +0200 Subject: [PATCH] Allow the selection of columns together with aggregates Patch by Benjamin Lerer; reviewed by Robert Stupp for CASSANDRA-9767 --- NEWS.txt | 4 + doc/cql3/CQL.textile | 5 +- .../selection/AbstractFunctionSelector.java | 9 +- .../cassandra/cql3/selection/Selection.java | 7 +- .../cql3/selection/SelectorFactories.java | 12 -- .../cql3/selection/SimpleSelector.java | 8 +- .../selection/WritetimeOrTTLSelector.java | 8 +- .../operations/AggregationTest.java | 135 ++++++++++++++++-- 8 files changed, 150 insertions(+), 38 deletions(-) diff --git a/NEWS.txt b/NEWS.txt index e114944939..e9c6ef83d8 100644 --- a/NEWS.txt +++ b/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' diff --git a/doc/cql3/CQL.textile b/doc/cql3/CQL.textile index b49d37b4d7..2749d35f76 100644 --- a/doc/cql3/CQL.textile +++ b/doc/cql3/CQL.textile @@ -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 diff --git a/src/java/org/apache/cassandra/cql3/selection/AbstractFunctionSelector.java b/src/java/org/apache/cassandra/cql3/selection/AbstractFunctionSelector.java index 956efcaabf..abf52e154d 100644 --- a/src/java/org/apache/cassandra/cql3/selection/AbstractFunctionSelector.java +++ b/src/java/org/apache/cassandra/cql3/selection/AbstractFunctionSelector.java @@ -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 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 extends Selector public boolean isAggregateSelectorFactory() { - return fun.isAggregate() || factories.containsOnlyAggregateFunctions(); + return fun.isAggregate() || factories.doesAggregation(); } }; } diff --git a/src/java/org/apache/cassandra/cql3/selection/Selection.java b/src/java/org/apache/cassandra/cql3/selection/Selection.java index ea66fcc892..13e030ff53 100644 --- a/src/java/org/apache/cassandra/cql3/selection/Selection.java +++ b/src/java/org/apache/cassandra/cql3/selection/Selection.java @@ -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 getOutputRow(int protocolVersion) throws InvalidRequestException diff --git a/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java b/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java index 81905e69d5..5ea29577d6 100644 --- a/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java +++ b/src/java/org/apache/cassandra/cql3/selection/SelectorFactories.java @@ -108,18 +108,6 @@ final class SelectorFactories implements Iterable factories.add(SimpleSelector.newFactory(def, index)); } - /** - * Checks if this SelectorFactories contains only factories for aggregates. - * - * @return true if this SelectorFactories contains only factories for aggregates, - * false 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). * diff --git a/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java b/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java index 6c4dc049ed..2e0514a6d6 100644 --- a/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java +++ b/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java @@ -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; } diff --git a/src/java/org/apache/cassandra/cql3/selection/WritetimeOrTTLSelector.java b/src/java/org/apache/cassandra/cql3/selection/WritetimeOrTTLSelector.java index b3607f396b..131827fdf7 100644 --- a/src/java/org/apache/cassandra/cql3/selection/WritetimeOrTTLSelector.java +++ b/src/java/org/apache/cassandra/cql3/selection/WritetimeOrTTLSelector.java @@ -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; } diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java index 62461b8c8a..51e4a280ae 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/operations/AggregationTest.java @@ -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 {