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

Change serialization format for UDT 

patch by slebresne; reviewed by iamaleksey for CASSANDRA-7209
This commit is contained in:
Sylvain Lebresne 2014-05-13 14:03:54 +02:00
parent 86cd423489
commit db9ef0b6db
11 changed files with 231 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -16,6 +16,7 @@
* remove unused classes (CASSANDRA-7197)
* Limit user types to the keyspace they are defined in (CASSANDRA-6643)
* Add validate method to CollectionType (CASSANDRA-7208)
* New serialization format for UDT values (CASSANDRA-7209)
Merged from 2.0:
* Always reallocate buffers in HSHA (CASSANDRA-6285)
* (Hadoop) support authentication in CqlRecordReader (CASSANDRA-7221)

9
doc/native_protocol_v3.spec
View File

@ -758,12 +758,13 @@ Table of Contents
7. User defined types
This section describe the serialization format for User defined types (UDT) values.
This section describes the serialization format for User defined types (UDT) values.
UDT values are the values of the User Defined Types as defined in section 4.2.5.2.
A UDT value is a [short] n indicating the number of values (field) of UDT values
followed by n elements. Each element is a [short bytes] representing the serialized
field.
A UDT value is composed of successive [bytes] values, one for each field of the UDT
value (in the order defined by the type). A UDT value will generally have one value
for each field of the type it represents, but it is allowed to have less values than
the type has fields.
8. Result paging

4
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -216,8 +216,8 @@ public final class CFMetaData
public static final CFMetaData SchemaUserTypesCf = compile("CREATE TABLE " + SystemKeyspace.SCHEMA_USER_TYPES_CF + " ("
+ "keyspace_name text,"
+ "type_name text,"
+ "column_names list<text>,"
+ "column_types list<text>,"
+ "field_names list<text>,"
+ "field_types list<text>,"
+ "PRIMARY KEY (keyspace_name, type_name)"
+ ") WITH COMMENT='Defined user types' AND gc_grace_seconds=8640");

16
src/java/org/apache/cassandra/config/UTMetaData.java
View File

@ -53,8 +53,8 @@ public final class UTMetaData
{
String keyspace = row.getString("keyspace_name");
ByteBuffer name = ByteBufferUtil.bytes(row.getString("type_name"));
List<String> rawColumns = row.getList("column_names", UTF8Type.instance);
List<String> rawTypes = row.getList("column_types", UTF8Type.instance);
List<String> rawColumns = row.getList("field_names", UTF8Type.instance);
List<String> rawTypes = row.getList("field_types", UTF8Type.instance);
List<ByteBuffer> columns = new ArrayList<>(rawColumns.size());
for (String rawColumn : rawColumns)
@ -97,13 +97,13 @@ public final class UTMetaData
Composite prefix = CFMetaData.SchemaUserTypesCf.comparator.make(newType.name);
CFRowAdder adder = new CFRowAdder(cf, prefix, timestamp);
adder.resetCollection("column_names");
adder.resetCollection("column_types");
adder.resetCollection("field_names");
adder.resetCollection("field_types");
for (ByteBuffer name : newType.columnNames)
adder.addListEntry("column_names", name);
for (AbstractType<?> type : newType.types)
adder.addListEntry("column_types", type.toString());
for (ByteBuffer name : newType.fieldNames)
adder.addListEntry("field_names", name);
for (AbstractType<?> type : newType.fieldTypes)
adder.addListEntry("field_types", type.toString());
return mutation;
}

21
src/java/org/apache/cassandra/cql3/UserTypes.java
View File

@ -35,7 +35,10 @@ public abstract class UserTypes
public static ColumnSpecification fieldSpecOf(ColumnSpecification column, int field)
{
return new ColumnSpecification(column.ksName, column.cfName, new ColumnIdentifier(column.name + "." + field, true), ((UserType)column.type).types.get(field));
return new ColumnSpecification(column.ksName,
column.cfName,
new ColumnIdentifier(column.name + "." + field, true),
((UserType)column.type).fieldTypes.get(field));
}
public static class Literal implements Term.Raw
@ -54,9 +57,9 @@ public abstract class UserTypes
UserType ut = (UserType)receiver.type;
boolean allTerminal = true;
List<Term> values = new ArrayList<>(entries.size());
for (int i = 0; i < ut.types.size(); i++)
for (int i = 0; i < ut.fieldTypes.size(); i++)
{
ColumnIdentifier field = new ColumnIdentifier(ut.columnNames.get(i), UTF8Type.instance);
ColumnIdentifier field = new ColumnIdentifier(ut.fieldNames.get(i), UTF8Type.instance);
Term value = entries.get(field).prepare(keyspace, fieldSpecOf(receiver, i));
if (value instanceof Term.NonTerminal)
@ -74,9 +77,9 @@ public abstract class UserTypes
throw new InvalidRequestException(String.format("Invalid user type literal for %s of type %s", receiver, receiver.type.asCQL3Type()));
UserType ut = (UserType)receiver.type;
for (int i = 0; i < ut.types.size(); i++)
for (int i = 0; i < ut.fieldTypes.size(); i++)
{
ColumnIdentifier field = new ColumnIdentifier(ut.columnNames.get(i), UTF8Type.instance);
ColumnIdentifier field = new ColumnIdentifier(ut.fieldNames.get(i), UTF8Type.instance);
Term.Raw value = entries.get(field);
if (value == null)
throw new InvalidRequestException(String.format("Invalid user type literal for %s: missing field %s", receiver, field));
@ -140,7 +143,7 @@ public abstract class UserTypes
public void collectMarkerSpecification(VariableSpecifications boundNames)
{
for (int i = 0; i < type.types.size(); i++)
for (int i = 0; i < type.fieldTypes.size(); i++)
values.get(i).collectMarkerSpecification(boundNames);
}
@ -151,14 +154,14 @@ public abstract class UserTypes
options = options.withProtocolVersion(3);
ByteBuffer[] buffers = new ByteBuffer[values.size()];
for (int i = 0; i < type.types.size(); i++)
for (int i = 0; i < type.fieldTypes.size(); i++)
{
ByteBuffer buffer = values.get(i).bindAndGet(options);
if (buffer == null)
throw new InvalidRequestException("null is not supported inside user type literals");
if (buffer.remaining() > FBUtilities.MAX_UNSIGNED_SHORT)
throw new InvalidRequestException(String.format("Value for field %s is too long. User type fields are limited to %d bytes but %d bytes provided",
UTF8Type.instance.getString(type.columnNames.get(i)),
UTF8Type.instance.getString(type.fieldNames.get(i)),
FBUtilities.MAX_UNSIGNED_SHORT,
buffer.remaining()));
@ -175,7 +178,7 @@ public abstract class UserTypes
@Override
public ByteBuffer bindAndGet(QueryOptions options) throws InvalidRequestException
{
return CompositeType.build(bindInternal(options));
return UserType.buildValue(bindInternal(options));
}
}
}

26
src/java/org/apache/cassandra/cql3/statements/AlterTypeStatement.java
View File

@ -138,8 +138,8 @@ public abstract class AlterTypeStatement extends SchemaAlteringStatement
private static int getIdxOfField(UserType type, ColumnIdentifier field)
{
for (int i = 0; i < type.types.size(); i++)
if (field.bytes.equals(type.columnNames.get(i)))
for (int i = 0; i < type.fieldTypes.size(); i++)
if (field.bytes.equals(type.fieldNames.get(i)))
return i;
return -1;
}
@ -183,8 +183,8 @@ public abstract class AlterTypeStatement extends SchemaAlteringStatement
return updated;
// Otherwise, check for nesting
List<AbstractType<?>> updatedTypes = updateTypes(ut.types, keyspace, toReplace, updated);
return updatedTypes == null ? null : new UserType(ut.keyspace, ut.name, new ArrayList<>(ut.columnNames), updatedTypes);
List<AbstractType<?>> updatedTypes = updateTypes(ut.fieldTypes, keyspace, toReplace, updated);
return updatedTypes == null ? null : new UserType(ut.keyspace, ut.name, new ArrayList<>(ut.fieldNames), updatedTypes);
}
else if (type instanceof CompositeType)
{
@ -275,12 +275,12 @@ public abstract class AlterTypeStatement extends SchemaAlteringStatement
if (getIdxOfField(toUpdate, fieldName) >= 0)
throw new InvalidRequestException(String.format("Cannot add new field %s to type %s: a field of the same name already exists", fieldName, name));
List<ByteBuffer> newNames = new ArrayList<>(toUpdate.columnNames.size() + 1);
newNames.addAll(toUpdate.columnNames);
List<ByteBuffer> newNames = new ArrayList<>(toUpdate.fieldNames.size() + 1);
newNames.addAll(toUpdate.fieldNames);
newNames.add(fieldName.bytes);
List<AbstractType<?>> newTypes = new ArrayList<>(toUpdate.types.size() + 1);
newTypes.addAll(toUpdate.types);
List<AbstractType<?>> newTypes = new ArrayList<>(toUpdate.fieldTypes.size() + 1);
newTypes.addAll(toUpdate.fieldTypes);
newTypes.add(type.prepare(keyspace()).getType());
return new UserType(toUpdate.keyspace, toUpdate.name, newNames, newTypes);
@ -292,12 +292,12 @@ public abstract class AlterTypeStatement extends SchemaAlteringStatement
if (idx < 0)
throw new InvalidRequestException(String.format("Unknown field %s in type %s", fieldName, name));
AbstractType<?> previous = toUpdate.types.get(idx);
AbstractType<?> previous = toUpdate.fieldTypes.get(idx);
if (!type.prepare(keyspace()).getType().isCompatibleWith(previous))
throw new InvalidRequestException(String.format("Type %s is incompatible with previous type %s of field %s in user type %s", type, previous.asCQL3Type(), fieldName, name));
List<ByteBuffer> newNames = new ArrayList<>(toUpdate.columnNames);
List<AbstractType<?>> newTypes = new ArrayList<>(toUpdate.types);
List<ByteBuffer> newNames = new ArrayList<>(toUpdate.fieldNames);
List<AbstractType<?>> newTypes = new ArrayList<>(toUpdate.fieldTypes);
newTypes.set(idx, type.prepare(keyspace()).getType());
return new UserType(toUpdate.keyspace, toUpdate.name, newNames, newTypes);
@ -321,8 +321,8 @@ public abstract class AlterTypeStatement extends SchemaAlteringStatement
protected UserType makeUpdatedType(UserType toUpdate) throws InvalidRequestException
{
List<ByteBuffer> newNames = new ArrayList<>(toUpdate.columnNames);
List<AbstractType<?>> newTypes = new ArrayList<>(toUpdate.types);
List<ByteBuffer> newNames = new ArrayList<>(toUpdate.fieldNames);
List<AbstractType<?>> newTypes = new ArrayList<>(toUpdate.fieldTypes);
for (Map.Entry<ColumnIdentifier, ColumnIdentifier> entry : renames.entrySet())
{

8
src/java/org/apache/cassandra/cql3/statements/CreateTypeStatement.java
View File

@ -74,12 +74,12 @@ public class CreateTypeStatement extends SchemaAlteringStatement
public static void checkForDuplicateNames(UserType type) throws InvalidRequestException
{
for (int i = 0; i < type.types.size() - 1; i++)
for (int i = 0; i < type.fieldTypes.size() - 1; i++)
{
ByteBuffer fieldName = type.columnNames.get(i);
for (int j = i+1; j < type.types.size(); j++)
ByteBuffer fieldName = type.fieldNames.get(i);
for (int j = i+1; j < type.fieldTypes.size(); j++)
{
if (fieldName.equals(type.columnNames.get(j)))
if (fieldName.equals(type.fieldNames.get(j)))
throw new InvalidRequestException(String.format("Duplicate field name %s in type %s",
UTF8Type.instance.getString(fieldName),
UTF8Type.instance.getString(type.name)));

21
src/java/org/apache/cassandra/cql3/statements/DropTypeStatement.java
View File

@ -91,18 +91,19 @@ public class DropTypeStatement extends SchemaAlteringStatement
private boolean isUsedBy(AbstractType<?> toCheck) throws RequestValidationException
{
if (toCheck instanceof CompositeType)
if (toCheck instanceof UserType)
{
UserType ut = (UserType)toCheck;
if (name.getKeyspace().equals(ut.keyspace) && name.getUserTypeName().equals(ut.name))
return true;
for (AbstractType<?> subtype : ut.fieldTypes)
if (isUsedBy(subtype))
return true;
}
else if (toCheck instanceof CompositeType)
{
CompositeType ct = (CompositeType)toCheck;
if ((ct instanceof UserType))
{
UserType ut = (UserType)ct;
if (name.getKeyspace().equals(ut.keyspace) && name.getUserTypeName().equals(ut.name))
return true;
}
// Also reach into subtypes
for (AbstractType<?> subtype : ct.types)
if (isUsedBy(subtype))
return true;

10
src/java/org/apache/cassandra/cql3/statements/Selection.java
View File

@ -142,13 +142,13 @@ public abstract class Selection
throw new InvalidRequestException(String.format("Invalid field selection: %s of type %s is not a user type", withField.selected, type.asCQL3Type()));
UserType ut = (UserType)type;
for (int i = 0; i < ut.types.size(); i++)
for (int i = 0; i < ut.fieldTypes.size(); i++)
{
if (!ut.columnNames.get(i).equals(withField.field.bytes))
if (!ut.fieldNames.get(i).equals(withField.field.bytes))
continue;
if (metadata != null)
metadata.add(makeFieldSelectSpec(cfm, withField, ut.types.get(i), raw.alias));
metadata.add(makeFieldSelectSpec(cfm, withField, ut.fieldTypes.get(i), raw.alias));
return new FieldSelector(ut, i, selected);
}
throw new InvalidRequestException(String.format("%s of type %s has no field %s", withField.selected, type.asCQL3Type(), withField.field));
@ -472,13 +472,13 @@ public abstract class Selection
public AbstractType<?> getType()
{
return type.types.get(field);
return type.fieldTypes.get(field);
}
@Override
public String toString()
{
return String.format("%s.%s", selected, UTF8Type.instance.getString(type.columnNames.get(field)));
return String.format("%s.%s", selected, UTF8Type.instance.getString(type.fieldNames.get(field)));
}
}

175
src/java/org/apache/cassandra/db/marshal/UserType.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import com.google.common.base.Objects;
@ -26,27 +27,27 @@ import com.google.common.base.Objects;
import org.apache.cassandra.cql3.CQL3Type;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.SyntaxException;
import org.apache.cassandra.serializers.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Pair;
/**
* A user defined type.
*
* The serialized format and sorting is exactly the one of CompositeType, but
* we keep additional metadata (the name of the type and the names
* of the columns).
*/
public class UserType extends CompositeType
public class UserType extends AbstractType<ByteBuffer>
{
public final String keyspace;
public final ByteBuffer name;
public final List<ByteBuffer> columnNames;
public final List<ByteBuffer> fieldNames;
public final List<AbstractType<?>> fieldTypes;
public UserType(String keyspace, ByteBuffer name, List<ByteBuffer> columnNames, List<AbstractType<?>> types)
public UserType(String keyspace, ByteBuffer name, List<ByteBuffer> fieldNames, List<AbstractType<?>> fieldTypes)
{
super(types);
assert fieldNames.size() == fieldTypes.size();
this.keyspace = keyspace;
this.name = name;
this.columnNames = columnNames;
this.fieldNames = fieldNames;
this.fieldTypes = fieldTypes;
}
public static UserType getInstance(TypeParser parser) throws ConfigurationException, SyntaxException
@ -69,10 +70,160 @@ public class UserType extends CompositeType
return UTF8Type.instance.compose(name);
}
public int compare(ByteBuffer o1, ByteBuffer o2)
{
if (!o1.hasRemaining() || !o2.hasRemaining())
return o1.hasRemaining() ? 1 : o2.hasRemaining() ? -1 : 0;
ByteBuffer bb1 = o1.duplicate();
ByteBuffer bb2 = o2.duplicate();
int i = 0;
while (bb1.remaining() > 0 && bb2.remaining() > 0)
{
AbstractType<?> comparator = fieldTypes.get(i);
int size1 = bb1.getInt();
int size2 = bb2.getInt();
// Handle nulls
if (size1 < 0)
{
if (size2 < 0)
continue;
return -1;
}
if (size2 < 0)
return 1;
ByteBuffer value1 = ByteBufferUtil.readBytes(bb1, size1);
ByteBuffer value2 = ByteBufferUtil.readBytes(bb2, size2);
int cmp = comparator.compare(value1, value2);
if (cmp != 0)
return cmp;
++i;
}
if (bb1.remaining() == 0)
return bb2.remaining() == 0 ? 0 : -1;
// bb1.remaining() > 0 && bb2.remaining() == 0
return 1;
}
@Override
public void validate(ByteBuffer bytes) throws MarshalException
{
ByteBuffer input = bytes.duplicate();
for (int i = 0; i < fieldTypes.size(); i++)
{
// we allow the input to have less fields than declared so as to support field addition.
if (!input.hasRemaining())
return;
if (input.remaining() < 4)
throw new MarshalException(String.format("Not enough bytes to read size of %dth field %s", i, fieldNames.get(i)));
int size = input.getInt();
// We don't handle null just yet, but we should fix that soon (CASSANDRA-7206)
if (size < 0)
throw new MarshalException("Nulls are not yet supported inside UDT values");
if (input.remaining() < size)
throw new MarshalException(String.format("Not enough bytes to read %dth field %s", i, fieldNames.get(i)));
ByteBuffer field = ByteBufferUtil.readBytes(input, size);
fieldTypes.get(i).validate(field);
}
// We're allowed to get less fields than declared, but not more
if (input.hasRemaining())
throw new MarshalException("Invalid remaining data after end of UDT value");
}
/**
* Split a UDT value into its fields values.
*/
public ByteBuffer[] split(ByteBuffer value)
{
ByteBuffer[] fields = new ByteBuffer[fieldTypes.size()];
ByteBuffer input = value.duplicate();
for (int i = 0; i < fieldTypes.size(); i++)
{
if (!input.hasRemaining())
return Arrays.copyOfRange(fields, 0, i);
int size = input.getInt();
fields[i] = size < 0 ? null : ByteBufferUtil.readBytes(input, size);
}
return fields;
}
public static ByteBuffer buildValue(ByteBuffer[] fields)
{
int totalLength = 0;
for (ByteBuffer field : fields)
totalLength += 4 + field.remaining();
ByteBuffer result = ByteBuffer.allocate(totalLength);
for (ByteBuffer field : fields)
{
result.putInt(field.remaining());
result.put(field.duplicate());
}
result.rewind();
return result;
}
@Override
public String getString(ByteBuffer value)
{
StringBuilder sb = new StringBuilder();
ByteBuffer input = value.duplicate();
for (int i = 0; i < fieldTypes.size(); i++)
{
if (!input.hasRemaining())
return sb.toString();
if (i > 0)
sb.append(":");
AbstractType<?> type = fieldTypes.get(i);
int size = input.getInt();
assert size >= 0; // We don't support nulls yet, but we will likely do with #7206 and we'll need
// a way to represent it as a string (without it conflicting with a user value)
ByteBuffer field = ByteBufferUtil.readBytes(input, size);
// We use ':' as delimiter so escape it if it's in the generated string
sb.append(field == null ? "null" : type.getString(value).replaceAll(":", "\\\\:"));
}
return sb.toString();
}
public ByteBuffer fromString(String source)
{
// Split the input on non-escaped ':' characters
List<String> fieldStrings = AbstractCompositeType.split(source);
ByteBuffer[] fields = new ByteBuffer[fieldStrings.size()];
for (int i = 0; i < fieldStrings.size(); i++)
{
AbstractType<?> type = fieldTypes.get(i);
// TODO: we'll need to handle null somehow here once we support them
String fieldString = fieldStrings.get(i).replaceAll("\\\\:", ":");
fields[i] = type.fromString(fieldString);
}
return buildValue(fields);
}
public TypeSerializer<ByteBuffer> getSerializer()
{
return BytesSerializer.instance;
}
@Override
public final int hashCode()
{
return Objects.hashCode(keyspace, name, columnNames, types);
return Objects.hashCode(keyspace, name, fieldNames, fieldTypes);
}
@Override
@ -82,7 +233,7 @@ public class UserType extends CompositeType
return false;
UserType that = (UserType)o;
return keyspace.equals(that.keyspace) && name.equals(that.name) && columnNames.equals(that.columnNames) && types.equals(that.types);
return keyspace.equals(that.keyspace) && name.equals(that.name) && fieldNames.equals(that.fieldNames) && fieldTypes.equals(that.fieldTypes);
}
@Override
@ -94,6 +245,6 @@ public class UserType extends CompositeType
@Override
public String toString()
{
return getClass().getName() + TypeParser.stringifyUserTypeParameters(keyspace, name, columnNames, types);
return getClass().getName() + TypeParser.stringifyUserTypeParameters(keyspace, name, fieldNames, fieldTypes);
}
}

14
src/java/org/apache/cassandra/transport/DataType.java
View File

@ -135,11 +135,11 @@ public enum DataType implements OptionCodec.Codecable<DataType>
UserType udt = (UserType)value;
CBUtil.writeString(udt.keyspace, cb);
CBUtil.writeString(UTF8Type.instance.compose(udt.name), cb);
cb.writeShort(udt.columnNames.size());
for (int i = 0; i < udt.columnNames.size(); i++)
cb.writeShort(udt.fieldNames.size());
for (int i = 0; i < udt.fieldNames.size(); i++)
{
CBUtil.writeString(UTF8Type.instance.compose(udt.columnNames.get(i)), cb);
codec.writeOne(DataType.fromType(udt.types.get(i), version), cb, version);
CBUtil.writeString(UTF8Type.instance.compose(udt.fieldNames.get(i)), cb);
codec.writeOne(DataType.fromType(udt.fieldTypes.get(i), version), cb, version);
}
break;
}
@ -166,10 +166,10 @@ public enum DataType implements OptionCodec.Codecable<DataType>
size += CBUtil.sizeOfString(udt.keyspace);
size += CBUtil.sizeOfString(UTF8Type.instance.compose(udt.name));
size += 2;
for (int i = 0; i < udt.columnNames.size(); i++)
for (int i = 0; i < udt.fieldNames.size(); i++)
{
size += CBUtil.sizeOfString(UTF8Type.instance.compose(udt.columnNames.get(i)));
size += codec.oneSerializedSize(DataType.fromType(udt.types.get(i), version), version);
size += CBUtil.sizeOfString(UTF8Type.instance.compose(udt.fieldNames.get(i)));
size += codec.oneSerializedSize(DataType.fromType(udt.fieldTypes.get(i), version), version);
}
return size;
default: