natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Problem upgrading to 3.0 with UDA 

patch by Robert Stupp; reviewed by Aleksey Yeschenko for CASSANDRA-10434
This commit is contained in:
Robert Stupp 2015-10-08 20:45:29 +02:00
parent d76d3a5d59
commit e1eed527f8
4 changed files with 188 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/java/org/apache/cassandra/cql3/functions/UDAggregate.java
View File

@ -25,7 +25,6 @@ import com.google.common.collect.ImmutableSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.schema.Functions;
@ -57,7 +56,8 @@ public class UDAggregate extends AbstractFunction implements AggregateFunction
this.initcond = initcond;
}
public static UDAggregate create(FunctionName name,
public static UDAggregate create(Functions functions,
FunctionName name,
List<AbstractType<?>> argTypes,
AbstractType<?> returnType,
FunctionName stateFunc,
@ -73,8 +73,8 @@ public class UDAggregate extends AbstractFunction implements AggregateFunction
return new UDAggregate(name,
argTypes,
returnType,
resolveScalar(name, stateFunc, stateTypes),
finalFunc != null ? resolveScalar(name, finalFunc, finalTypes) : null,
resolveScalar(functions, name, stateFunc, stateTypes),
finalFunc != null ? resolveScalar(functions, name, finalFunc, finalTypes) : null,
initcond);
}
@ -194,9 +194,9 @@ public class UDAggregate extends AbstractFunction implements AggregateFunction
};
}
private static ScalarFunction resolveScalar(FunctionName aName, FunctionName fName, List<AbstractType<?>> argTypes) throws InvalidRequestException
private static ScalarFunction resolveScalar(Functions functions, FunctionName aName, FunctionName fName, List<AbstractType<?>> argTypes) throws InvalidRequestException
{
Optional<Function> fun = Schema.instance.findFunction(fName, argTypes);
Optional<Function> fun = functions.find(fName, argTypes);
if (!fun.isPresent())
throw new InvalidRequestException(String.format("Referenced state function '%s %s' for aggregate '%s' does not exist",
fName,

20
src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java
View File

@ -163,7 +163,9 @@ public final class LegacySchemaMigrator
Collection<Table> tables = readTables(keyspaceName);
Collection<Type> types = readTypes(keyspaceName);
Collection<Function> functions = readFunctions(keyspaceName);
Collection<Aggregate> aggregates = readAggregates(keyspaceName);
Functions.Builder functionsBuilder = Functions.builder();
functions.forEach(udf -> functionsBuilder.add(udf.metadata));
Collection<Aggregate> aggregates = readAggregates(functionsBuilder.build(), keyspaceName);
return new Keyspace(timestamp, keyspaceName, params, tables, types, functions, aggregates);
}
@ -811,7 +813,7 @@ public final class LegacySchemaMigrator
* Reading UDAs
*/
private static Collection<Aggregate> readAggregates(String keyspaceName)
private static Collection<Aggregate> readAggregates(Functions functions, String keyspaceName)
{
String query = format("SELECT aggregate_name, signature FROM %s.%s WHERE keyspace_name = ?",
SystemKeyspace.NAME,
@ -820,14 +822,14 @@ public final class LegacySchemaMigrator
query(query, keyspaceName).forEach(row -> aggregateSignatures.put(row.getString("aggregate_name"), row.getList("signature", UTF8Type.instance)));
Collection<Aggregate> aggregates = new ArrayList<>();
aggregateSignatures.entries().forEach(pair -> aggregates.add(readAggregate(keyspaceName, pair.getKey(), pair.getValue())));
aggregateSignatures.entries().forEach(pair -> aggregates.add(readAggregate(functions, keyspaceName, pair.getKey(), pair.getValue())));
return aggregates;
}
private static Aggregate readAggregate(String keyspaceName, String aggregateName, List<String> signature)
private static Aggregate readAggregate(Functions functions, String keyspaceName, String aggregateName, List<String> signature)
{
long timestamp = readAggregateTimestamp(keyspaceName, aggregateName, signature);
UDAggregate metadata = readAggregateMetadata(keyspaceName, aggregateName, signature);
UDAggregate metadata = readAggregateMetadata(functions, keyspaceName, aggregateName, signature);
return new Aggregate(timestamp, metadata);
}
@ -841,9 +843,9 @@ public final class LegacySchemaMigrator
return query(query, keyspaceName, aggregateName, signature).one().getLong("timestamp");
}
private static UDAggregate readAggregateMetadata(String keyspaceName, String functionName, List<String> signature)
private static UDAggregate readAggregateMetadata(Functions functions, String keyspaceName, String functionName, List<String> signature)
{
String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND function_name = ? AND signature = ?",
String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND aggregate_name = ? AND signature = ?",
SystemKeyspace.NAME,
SystemKeyspace.LEGACY_AGGREGATES);
UntypedResultSet.Row row = query(query, keyspaceName, functionName, signature).one();
@ -863,13 +865,13 @@ public final class LegacySchemaMigrator
AbstractType<?> returnType = parseType(row.getString("return_type"));
FunctionName stateFunc = new FunctionName(keyspaceName, row.getString("state_func"));
AbstractType<?> stateType = parseType(row.getString("state_type"));
FunctionName finalFunc = row.has("final_func") ? new FunctionName(keyspaceName, row.getString("final_func")) : null;
AbstractType<?> stateType = row.has("state_type") ? parseType(row.getString("state_type")) : null;
ByteBuffer initcond = row.has("initcond") ? row.getBytes("initcond") : null;
try
{
return UDAggregate.create(name, argTypes, returnType, stateFunc, finalFunc, stateType, initcond);
return UDAggregate.create(functions, name, argTypes, returnType, stateFunc, finalFunc, stateType, initcond);
}
catch (InvalidRequestException reason)
{

18
src/java/org/apache/cassandra/schema/SchemaKeyspace.java
View File

@ -776,8 +776,8 @@ public final class SchemaKeyspace
Types types = createTypesFromPartition(serializedTypes);
Collection<UDFunction> udfs = createFunctionsFromFunctionsPartition(serializedFunctions);
Collection<UDAggregate> udas = createAggregatesFromAggregatesPartition(serializedAggregates);
Functions functions = org.apache.cassandra.schema.Functions.builder().add(udfs).add(udas).build();
Functions functions = org.apache.cassandra.schema.Functions.builder().add(udfs).build();
functions = createAggregatesFromAggregatesPartition(functions, serializedAggregates);
return KeyspaceMetadata.create(name, params, tables, views, types, functions);
}
@ -1635,16 +1635,20 @@ public final class SchemaKeyspace
.build();
}
private static Collection<UDAggregate> createAggregatesFromAggregatesPartition(RowIterator partition)
private static Functions createAggregatesFromAggregatesPartition(Functions functions, RowIterator partition)
{
List<UDAggregate> aggregates = new ArrayList<>();
String query = String.format("SELECT * FROM %s.%s", NAME, AGGREGATES);
for (UntypedResultSet.Row row : QueryProcessor.resultify(query, partition))
aggregates.add(createAggregateFromAggregateRow(row));
return aggregates;
functions = functions.with(createAggregateFromAggregateRow(functions, row));
return functions;
}
private static UDAggregate createAggregateFromAggregateRow(UntypedResultSet.Row row)
{
return createAggregateFromAggregateRow(Schema.instance.getKSMetaData(row.getString("keyspace_name")).functions, row);
}
private static UDAggregate createAggregateFromAggregateRow(Functions functions, UntypedResultSet.Row row)
{
String ksName = row.getString("keyspace_name");
String functionName = row.getString("aggregate_name");
@ -1673,7 +1677,7 @@ public final class SchemaKeyspace
try
{
return UDAggregate.create(name, argTypes, returnType, stateFunc, finalFunc, stateType, initcond);
return UDAggregate.create(functions, name, argTypes, returnType, stateFunc, finalFunc, stateType, initcond);
}
catch (InvalidRequestException reason)
{

163
test/unit/org/apache/cassandra/schema/LegacySchemaMigratorTest.java
View File

@ -251,7 +251,9 @@ public class LegacySchemaMigratorTest
keyspaces.add(keyspaceWithTriggers());
keyspaces.add(keyspaceWithUDTs());
keyspaces.add(keyspaceWithUDFs());
keyspaces.add(keyspaceWithUDFsAndUDTs());
keyspaces.add(keyspaceWithUDAs());
keyspaces.add(keyspaceWithUDAsAndUDTs());
return keyspaces;
}
@ -326,7 +328,6 @@ public class LegacySchemaMigratorTest
{
String keyspace = KEYSPACE_PREFIX + "UDFs";
UDFunction udf1 = UDFunction.create(new FunctionName(keyspace, "udf"),
ImmutableList.of(new ColumnIdentifier("col1", false), new ColumnIdentifier("col2", false)),
ImmutableList.of(BytesType.instance, Int32Type.instance),
@ -360,17 +361,173 @@ public class LegacySchemaMigratorTest
Functions.of(udf1, udf2, udf3));
}
// TODO: add representative UDAs set
private static KeyspaceMetadata keyspaceWithUDAs()
{
String keyspace = KEYSPACE_PREFIX + "UDAs";
UDFunction udf1 = UDFunction.create(new FunctionName(keyspace, "udf1"),
ImmutableList.of(new ColumnIdentifier("col1", false), new ColumnIdentifier("col2", false)),
ImmutableList.of(Int32Type.instance, Int32Type.instance),
Int32Type.instance,
false,
"java",
"return 42;");
UDFunction udf2 = UDFunction.create(new FunctionName(keyspace, "udf2"),
ImmutableList.of(new ColumnIdentifier("col1", false), new ColumnIdentifier("col2", false)),
ImmutableList.of(LongType.instance, Int32Type.instance),
LongType.instance,
false,
"java",
"return 42L;");
UDFunction udf3 = UDFunction.create(new FunctionName(keyspace, "udf3"),
ImmutableList.of(new ColumnIdentifier("col1", false)),
ImmutableList.of(LongType.instance),
DoubleType.instance,
false,
"java",
"return 42d;");
Functions udfs = Functions.builder().add(udf1).add(udf2).add(udf3).build();
UDAggregate uda1 = UDAggregate.create(udfs, new FunctionName(keyspace, "uda1"),
ImmutableList.of(udf1.argTypes().get(1)),
udf1.returnType(),
udf1.name(),
null,
udf1.argTypes().get(0),
null
);
UDAggregate uda2 = UDAggregate.create(udfs, new FunctionName(keyspace, "uda2"),
ImmutableList.of(udf2.argTypes().get(1)),
udf3.returnType(),
udf2.name(),
udf3.name(),
udf2.argTypes().get(0),
LongType.instance.decompose(0L)
);
return KeyspaceMetadata.create(keyspace,
KeyspaceParams.simple(1),
Tables.none(),
Views.none(),
Types.none(),
Functions.of());
Functions.of(udf1, udf2, udf3, uda1, uda2));
}
private static KeyspaceMetadata keyspaceWithUDFsAndUDTs()
{
String keyspace = KEYSPACE_PREFIX + "UDFUDTs";
UserType udt1 = new UserType(keyspace,
bytes("udt1"),
new ArrayList<ByteBuffer>() {{ add(bytes("col1")); add(bytes("col2")); }},
new ArrayList<AbstractType<?>>() {{ add(UTF8Type.instance); add(Int32Type.instance); }});
UserType udt2 = new UserType(keyspace,
bytes("udt2"),
new ArrayList<ByteBuffer>() {{ add(bytes("col1")); add(bytes("col2")); }},
new ArrayList<AbstractType<?>>() {{ add(ListType.getInstance(udt1, false)); add(Int32Type.instance); }});
UDFunction udf1 = UDFunction.create(new FunctionName(keyspace, "udf"),
ImmutableList.of(new ColumnIdentifier("col1", false), new ColumnIdentifier("col2", false)),
ImmutableList.of(udt1, udt2),
LongType.instance,
false,
"java",
"return 42L;");
// an overload with the same name, not a typo
UDFunction udf2 = UDFunction.create(new FunctionName(keyspace, "udf"),
ImmutableList.of(new ColumnIdentifier("col3", false), new ColumnIdentifier("col4", false)),
ImmutableList.of(AsciiType.instance, LongType.instance),
Int32Type.instance,
true,
"java",
"return 42;");
UDFunction udf3 = UDFunction.create(new FunctionName(keyspace, "udf3"),
ImmutableList.of(new ColumnIdentifier("col4", false)),
ImmutableList.of(new TupleType(Arrays.asList(udt1, udt2))),
BooleanType.instance,
false,
"java",
"return true;");
return KeyspaceMetadata.create(keyspace,
KeyspaceParams.simple(1),
Tables.none(),
Views.none(),
Types.of(udt1, udt2),
Functions.of(udf1, udf2, udf3));
}
private static KeyspaceMetadata keyspaceWithUDAsAndUDTs()
{
String keyspace = KEYSPACE_PREFIX + "UDAUDTs";
UserType udt1 = new UserType(keyspace,
bytes("udt1"),
new ArrayList<ByteBuffer>() {{ add(bytes("col1")); add(bytes("col2")); }},
new ArrayList<AbstractType<?>>() {{ add(UTF8Type.instance); add(Int32Type.instance); }});
UserType udt2 = new UserType(keyspace,
bytes("udt2"),
new ArrayList<ByteBuffer>() {{ add(bytes("col1")); add(bytes("col2")); }},
new ArrayList<AbstractType<?>>() {{ add(ListType.getInstance(udt1, false)); add(Int32Type.instance); }});
UDFunction udf1 = UDFunction.create(new FunctionName(keyspace, "udf1"),
ImmutableList.of(new ColumnIdentifier("col1", false), new ColumnIdentifier("col2", false)),
ImmutableList.of(udt1, udt2),
udt1,
false,
"java",
"return null;");
UDFunction udf2 = UDFunction.create(new FunctionName(keyspace, "udf2"),
ImmutableList.of(new ColumnIdentifier("col1", false), new ColumnIdentifier("col2", false)),
ImmutableList.of(udt2, udt1),
udt2,
false,
"java",
"return null;");
UDFunction udf3 = UDFunction.create(new FunctionName(keyspace, "udf3"),
ImmutableList.of(new ColumnIdentifier("col1", false)),
ImmutableList.of(udt2),
DoubleType.instance,
false,
"java",
"return 42d;");
Functions udfs = Functions.builder().add(udf1).add(udf2).add(udf3).build();
UDAggregate uda1 = UDAggregate.create(udfs, new FunctionName(keyspace, "uda1"),
ImmutableList.of(udf1.argTypes().get(1)),
udf1.returnType(),
udf1.name(),
null,
udf1.argTypes().get(0),
null
);
UDAggregate uda2 = UDAggregate.create(udfs, new FunctionName(keyspace, "uda2"),
ImmutableList.of(udf2.argTypes().get(1)),
udf3.returnType(),
udf2.name(),
udf3.name(),
udf2.argTypes().get(0),
LongType.instance.decompose(0L)
);
return KeyspaceMetadata.create(keyspace,
KeyspaceParams.simple(1),
Tables.none(),
Views.none(),
Types.of(udt1, udt2),
Functions.of(udf1, udf2, udf3, uda1, uda2));
}
/*