| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.avro; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.StringReader; |
| import java.io.StringWriter; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Iterator; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.IdentityHashMap; |
| import java.util.LinkedHashMap; |
| import java.util.LinkedHashSet; |
| import java.util.List; |
| import java.util.Locale; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import org.apache.avro.util.internal.JacksonUtils; |
| import org.codehaus.jackson.JsonFactory; |
| import org.codehaus.jackson.JsonNode; |
| import org.codehaus.jackson.JsonParseException; |
| import org.codehaus.jackson.JsonParser; |
| import org.codehaus.jackson.JsonGenerator; |
| import org.codehaus.jackson.map.ObjectMapper; |
| import org.codehaus.jackson.node.DoubleNode; |
| |
| /** An abstract data type. |
| * <p>A schema may be one of: |
| * <ul> |
| * <li>A <i>record</i>, mapping field names to field value data; |
| * <li>An <i>enum</i>, containing one of a small set of symbols; |
| * <li>An <i>array</i> of values, all of the same schema; |
| * <li>A <i>map</i>, containing string/value pairs, of a declared schema; |
| * <li>A <i>union</i> of other schemas; |
| * <li>A <i>fixed</i> sized binary object; |
| * <li>A unicode <i>string</i>; |
| * <li>A sequence of <i>bytes</i>; |
| * <li>A 32-bit signed <i>int</i>; |
| * <li>A 64-bit signed <i>long</i>; |
| * <li>A 32-bit IEEE single-<i>float</i>; or |
| * <li>A 64-bit IEEE <i>double</i>-float; or |
| * <li>A <i>boolean</i>; or |
| * <li><i>null</i>. |
| * </ul> |
| * |
| * A schema can be constructed using one of its static <tt>createXXX</tt> |
| * methods, or more conveniently using {@link SchemaBuilder}. The schema objects are |
| * <i>logically</i> immutable. |
| * There are only two mutating methods - {@link #setFields(List)} and |
| * {@link #addProp(String, String)}. The following restrictions apply on these |
| * two methods. |
| * <ul> |
| * <li> {@link #setFields(List)}, can be called at most once. This method exists |
| * in order to enable clients to build recursive schemas. |
| * <li> {@link #addProp(String, String)} can be called with property names |
| * that are not present already. It is not possible to change or delete an |
| * existing property. |
| * </ul> |
| */ |
| public abstract class Schema extends JsonProperties { |
| static final JsonFactory FACTORY = new JsonFactory(); |
| static final ObjectMapper MAPPER = new ObjectMapper(FACTORY); |
| |
| private static final int NO_HASHCODE = Integer.MIN_VALUE; |
| |
| static { |
| FACTORY.enable(JsonParser.Feature.ALLOW_COMMENTS); |
| FACTORY.setCodec(MAPPER); |
| } |
| |
| /** The type of a schema. */ |
| public enum Type { |
| RECORD, ENUM, ARRAY, MAP, UNION, FIXED, STRING, BYTES, |
| INT, LONG, FLOAT, DOUBLE, BOOLEAN, NULL; |
| private String name; |
| private Type() { this.name = this.name().toLowerCase(Locale.ENGLISH); } |
| public String getName() { return name; } |
| }; |
| |
| private final Type type; |
| private LogicalType logicalType = null; |
| |
| Schema(Type type) { |
| super(SCHEMA_RESERVED); |
| this.type = type; |
| } |
| |
| /** Create a schema for a primitive type. */ |
| public static Schema create(Type type) { |
| switch (type) { |
| case STRING: return new StringSchema(); |
| case BYTES: return new BytesSchema(); |
| case INT: return new IntSchema(); |
| case LONG: return new LongSchema(); |
| case FLOAT: return new FloatSchema(); |
| case DOUBLE: return new DoubleSchema(); |
| case BOOLEAN: return new BooleanSchema(); |
| case NULL: return new NullSchema(); |
| default: throw new AvroRuntimeException("Can't create a: "+type); |
| } |
| } |
| |
| private static final Set<String> SCHEMA_RESERVED = new HashSet<String>(); |
| static { |
| Collections.addAll(SCHEMA_RESERVED, |
| "doc", "fields", "items", "name", "namespace", |
| "size", "symbols", "values", "type", "aliases"); |
| } |
| |
| int hashCode = NO_HASHCODE; |
| |
| @Override public void addProp(String name, JsonNode value) { |
| super.addProp(name, value); |
| hashCode = NO_HASHCODE; |
| } |
| |
| @Override public void addProp(String name, Object value) { |
| super.addProp(name, value); |
| hashCode = NO_HASHCODE; |
| } |
| |
| public LogicalType getLogicalType() { |
| return logicalType; |
| } |
| |
| void setLogicalType(LogicalType logicalType) { |
| this.logicalType = logicalType; |
| } |
| |
| /** Create an anonymous record schema. */ |
| public static Schema createRecord(List<Field> fields) { |
| Schema result = createRecord(null, null, null, false); |
| result.setFields(fields); |
| return result; |
| } |
| |
| /** Create a named record schema. */ |
| public static Schema createRecord(String name, String doc, String namespace, |
| boolean isError) { |
| return new RecordSchema(new Name(name, namespace), doc, isError); |
| } |
| |
| /** Create a named record schema with fields already set. */ |
| public static Schema createRecord(String name, String doc, String namespace, |
| boolean isError, List<Field> fields) { |
| return new RecordSchema(new Name(name, namespace), doc, isError, fields); |
| } |
| |
| /** Create an enum schema. */ |
| public static Schema createEnum(String name, String doc, String namespace, |
| List<String> values) { |
| return new EnumSchema(new Name(name, namespace), doc, |
| new LockableArrayList<String>(values)); |
| } |
| |
| /** Create an array schema. */ |
| public static Schema createArray(Schema elementType) { |
| return new ArraySchema(elementType); |
| } |
| |
| /** Create a map schema. */ |
| public static Schema createMap(Schema valueType) { |
| return new MapSchema(valueType); |
| } |
| |
| /** Create a union schema. */ |
| public static Schema createUnion(List<Schema> types) { |
| return new UnionSchema(new LockableArrayList<Schema>(types)); |
| } |
| |
| /** Create a union schema. */ |
| public static Schema createUnion(Schema... types) { |
| return createUnion(new LockableArrayList<Schema>(types)); |
| } |
| |
| /** Create a union schema. */ |
| public static Schema createFixed(String name, String doc, String space, |
| int size) { |
| return new FixedSchema(new Name(name, space), doc, size); |
| } |
| |
| /** Return the type of this schema. */ |
| public Type getType() { return type; } |
| |
| /** |
| * If this is a record, returns the Field with the |
| * given name <tt>fieldName</tt>. If there is no field by that name, a |
| * <tt>null</tt> is returned. |
| */ |
| public Field getField(String fieldname) { |
| throw new AvroRuntimeException("Not a record: "+this); |
| } |
| |
| /** |
| * If this is a record, returns the fields in it. The returned |
| * list is in the order of their positions. |
| */ |
| public List<Field> getFields() { |
| throw new AvroRuntimeException("Not a record: "+this); |
| } |
| |
| /** |
| * If this is a record, set its fields. The fields can be set |
| * only once in a schema. |
| */ |
| public void setFields(List<Field> fields) { |
| throw new AvroRuntimeException("Not a record: "+this); |
| } |
| |
| /** If this is an enum, return its symbols. */ |
| public List<String> getEnumSymbols() { |
| throw new AvroRuntimeException("Not an enum: "+this); |
| } |
| |
| /** If this is an enum, return a symbol's ordinal value. */ |
| public int getEnumOrdinal(String symbol) { |
| throw new AvroRuntimeException("Not an enum: "+this); |
| } |
| |
| /** If this is an enum, returns true if it contains given symbol. */ |
| public boolean hasEnumSymbol(String symbol) { |
| throw new AvroRuntimeException("Not an enum: "+this); |
| } |
| |
| /** If this is a record, enum or fixed, returns its name, otherwise the name |
| * of the primitive type. */ |
| public String getName() { return type.name; } |
| |
| /** If this is a record, enum, or fixed, returns its docstring, |
| * if available. Otherwise, returns null. */ |
| public String getDoc() { |
| return null; |
| } |
| |
| /** If this is a record, enum or fixed, returns its namespace, if any. */ |
| public String getNamespace() { |
| throw new AvroRuntimeException("Not a named type: "+this); |
| } |
| |
| /** If this is a record, enum or fixed, returns its namespace-qualified name, |
| * otherwise returns the name of the primitive type. */ |
| public String getFullName() { |
| return getName(); |
| } |
| |
| /** If this is a record, enum or fixed, add an alias. */ |
| public void addAlias(String alias) { |
| throw new AvroRuntimeException("Not a named type: "+this); |
| } |
| |
| /** If this is a record, enum or fixed, add an alias. */ |
| public void addAlias(String alias, String space) { |
| throw new AvroRuntimeException("Not a named type: "+this); |
| } |
| |
| /** If this is a record, enum or fixed, return its aliases, if any. */ |
| public Set<String> getAliases() { |
| throw new AvroRuntimeException("Not a named type: "+this); |
| } |
| |
| /** Returns true if this record is an error type. */ |
| public boolean isError() { |
| throw new AvroRuntimeException("Not a record: "+this); |
| } |
| |
| /** If this is an array, returns its element type. */ |
| public Schema getElementType() { |
| throw new AvroRuntimeException("Not an array: "+this); |
| } |
| |
| /** If this is a map, returns its value type. */ |
| public Schema getValueType() { |
| throw new AvroRuntimeException("Not a map: "+this); |
| } |
| |
| /** If this is a union, returns its types. */ |
| public List<Schema> getTypes() { |
| throw new AvroRuntimeException("Not a union: "+this); |
| } |
| |
| /** If this is a union, return the branch with the provided full name. */ |
| public Integer getIndexNamed(String name) { |
| throw new AvroRuntimeException("Not a union: "+this); |
| } |
| |
| /** If this is fixed, returns its size. */ |
| public int getFixedSize() { |
| throw new AvroRuntimeException("Not fixed: "+this); |
| } |
| |
| /** Render this as <a href="http://json.org/">JSON</a>.*/ |
| @Override |
| public String toString() { return toString(false); } |
| |
| /** Render this as <a href="http://json.org/">JSON</a>. |
| * @param pretty if true, pretty-print JSON. |
| */ |
| public String toString(boolean pretty) { |
| try { |
| StringWriter writer = new StringWriter(); |
| JsonGenerator gen = FACTORY.createJsonGenerator(writer); |
| if (pretty) gen.useDefaultPrettyPrinter(); |
| toJson(new Names(), gen); |
| gen.flush(); |
| return writer.toString(); |
| } catch (IOException e) { |
| throw new AvroRuntimeException(e); |
| } |
| } |
| |
| void toJson(Names names, JsonGenerator gen) throws IOException { |
| if (props.size() == 0) { // no props defined |
| gen.writeString(getName()); // just write name |
| } else { |
| gen.writeStartObject(); |
| gen.writeStringField("type", getName()); |
| writeProps(gen); |
| gen.writeEndObject(); |
| } |
| } |
| |
| void fieldsToJson(Names names, JsonGenerator gen) throws IOException { |
| throw new AvroRuntimeException("Not a record: "+this); |
| } |
| |
| public boolean equals(Object o) { |
| if (o == this) return true; |
| if (!(o instanceof Schema)) return false; |
| Schema that = (Schema)o; |
| if (!(this.type == that.type)) return false; |
| return equalCachedHash(that) && props.equals(that.props); |
| } |
| public final int hashCode() { |
| if (hashCode == NO_HASHCODE) |
| hashCode = computeHash(); |
| return hashCode; |
| } |
| |
| int computeHash() { return getType().hashCode() + props.hashCode(); } |
| |
| final boolean equalCachedHash(Schema other) { |
| return (hashCode == other.hashCode) |
| || (hashCode == NO_HASHCODE) |
| || (other.hashCode == NO_HASHCODE); |
| } |
| |
| private static final Set<String> FIELD_RESERVED = new HashSet<String>(); |
| static { |
| Collections.addAll(FIELD_RESERVED, |
| "default","doc","name","order","type","aliases"); |
| } |
| |
| /** A field within a record. */ |
| public static class Field extends JsonProperties { |
| |
| /** How values of this field should be ordered when sorting records. */ |
| public enum Order { |
| ASCENDING, DESCENDING, IGNORE; |
| private String name; |
| private Order() { this.name = this.name().toLowerCase(Locale.ENGLISH); } |
| }; |
| |
| private final String name; // name of the field. |
| private int position = -1; |
| private final Schema schema; |
| private final String doc; |
| private final JsonNode defaultValue; |
| private final Order order; |
| private Set<String> aliases; |
| |
| /** @deprecated use {@link #Field(String, Schema, String, Object)} */ |
| @Deprecated |
| public Field(String name, Schema schema, String doc, |
| JsonNode defaultValue) { |
| this(name, schema, doc, defaultValue, Order.ASCENDING); |
| } |
| /** @deprecated use {@link #Field(String, Schema, String, Object, Order)} */ |
| @Deprecated |
| public Field(String name, Schema schema, String doc, |
| JsonNode defaultValue, Order order) { |
| super(FIELD_RESERVED); |
| this.name = validateName(name); |
| this.schema = schema; |
| this.doc = doc; |
| this.defaultValue = validateDefault(name, schema, defaultValue); |
| this.order = order; |
| } |
| /** |
| * @param defaultValue the default value for this field specified using the mapping |
| * in {@link JsonProperties} |
| */ |
| public Field(String name, Schema schema, String doc, |
| Object defaultValue) { |
| this(name, schema, doc, defaultValue, Order.ASCENDING); |
| } |
| /** |
| * @param defaultValue the default value for this field specified using the mapping |
| * in {@link JsonProperties} |
| */ |
| public Field(String name, Schema schema, String doc, |
| Object defaultValue, Order order) { |
| this(name, schema, doc, JacksonUtils.toJsonNode(defaultValue), order); |
| } |
| public String name() { return name; }; |
| /** The position of this field within the record. */ |
| public int pos() { return position; } |
| /** This field's {@link Schema}. */ |
| public Schema schema() { return schema; } |
| /** Field's documentation within the record, if set. May return null. */ |
| public String doc() { return doc; } |
| /** @deprecated use {@link #defaultVal() } */ |
| @Deprecated public JsonNode defaultValue() { return defaultValue; } |
| /** |
| * @return the default value for this field specified using the mapping |
| * in {@link JsonProperties} |
| */ |
| public Object defaultVal() { return JacksonUtils.toObject(defaultValue); } |
| public Order order() { return order; } |
| @Deprecated public Map<String,String> props() { return getProps(); } |
| public void addAlias(String alias) { |
| if (aliases == null) |
| this.aliases = new LinkedHashSet<String>(); |
| aliases.add(alias); |
| } |
| /** Return the defined aliases as an unmodifieable Set. */ |
| public Set<String> aliases() { |
| if (aliases == null) |
| return Collections.emptySet(); |
| return Collections.unmodifiableSet(aliases); |
| } |
| public boolean equals(Object other) { |
| if (other == this) return true; |
| if (!(other instanceof Field)) return false; |
| Field that = (Field) other; |
| return (name.equals(that.name)) && |
| (schema.equals(that.schema)) && |
| defaultValueEquals(that.defaultValue) && |
| (order == that.order) && |
| props.equals(that.props); |
| } |
| public int hashCode() { return name.hashCode() + schema.computeHash(); } |
| |
| private boolean defaultValueEquals(JsonNode thatDefaultValue) { |
| if (defaultValue == null) |
| return thatDefaultValue == null; |
| if (Double.isNaN(defaultValue.getDoubleValue())) |
| return Double.isNaN(thatDefaultValue.getDoubleValue()); |
| return defaultValue.equals(thatDefaultValue); |
| } |
| |
| @Override |
| public String toString() { |
| return name + " type:" + schema.type + " pos:" + position; |
| } |
| } |
| |
| static class Name { |
| private final String name; |
| private final String space; |
| private final String full; |
| public Name(String name, String space) { |
| if (name == null) { // anonymous |
| this.name = this.space = this.full = null; |
| return; |
| } |
| int lastDot = name.lastIndexOf('.'); |
| if (lastDot < 0) { // unqualified name |
| this.name = validateName(name); |
| } else { // qualified name |
| space = name.substring(0, lastDot); // get space from name |
| this.name = validateName(name.substring(lastDot+1, name.length())); |
| } |
| if ("".equals(space)) |
| space = null; |
| this.space = space; |
| this.full = (this.space == null) ? this.name : this.space+"."+this.name; |
| } |
| public boolean equals(Object o) { |
| if (o == this) return true; |
| if (!(o instanceof Name)) return false; |
| Name that = (Name)o; |
| return full==null ? that.full==null : full.equals(that.full); |
| } |
| public int hashCode() { |
| return full==null ? 0 : full.hashCode(); |
| } |
| public String toString() { return full; } |
| public void writeName(Names names, JsonGenerator gen) throws IOException { |
| if (name != null) gen.writeStringField("name", name); |
| if (space != null) { |
| if (!space.equals(names.space())) |
| gen.writeStringField("namespace", space); |
| } else if (names.space() != null) { // null within non-null |
| gen.writeStringField("namespace", ""); |
| } |
| } |
| public String getQualified(String defaultSpace) { |
| return (space == null || space.equals(defaultSpace)) ? name : full; |
| } |
| } |
| |
| private static abstract class NamedSchema extends Schema { |
| final Name name; |
| final String doc; |
| Set<Name> aliases; |
| public NamedSchema(Type type, Name name, String doc) { |
| super(type); |
| this.name = name; |
| this.doc = doc; |
| if (PRIMITIVES.containsKey(name.full)) { |
| throw new AvroTypeException("Schemas may not be named after primitives: " + name.full); |
| } |
| } |
| public String getName() { return name.name; } |
| public String getDoc() { return doc; } |
| public String getNamespace() { return name.space; } |
| public String getFullName() { return name.full; } |
| public void addAlias(String alias) { |
| addAlias(alias, null); |
| } |
| public void addAlias(String name, String space) { |
| if (aliases == null) |
| this.aliases = new LinkedHashSet<Name>(); |
| if (space == null) |
| space = this.name.space; |
| aliases.add(new Name(name, space)); |
| } |
| public Set<String> getAliases() { |
| Set<String> result = new LinkedHashSet<String>(); |
| if (aliases != null) |
| for (Name alias : aliases) |
| result.add(alias.full); |
| return result; |
| } |
| public boolean writeNameRef(Names names, JsonGenerator gen) |
| throws IOException { |
| if (this.equals(names.get(name))) { |
| gen.writeString(name.getQualified(names.space())); |
| return true; |
| } else if (name.name != null) { |
| names.put(name, this); |
| } |
| return false; |
| } |
| public void writeName(Names names, JsonGenerator gen) throws IOException { |
| name.writeName(names, gen); |
| } |
| public boolean equalNames(NamedSchema that) { |
| return this.name.equals(that.name); |
| } |
| @Override int computeHash() { |
| return super.computeHash() + name.hashCode(); |
| } |
| public void aliasesToJson(JsonGenerator gen) throws IOException { |
| if (aliases == null || aliases.size() == 0) return; |
| gen.writeFieldName("aliases"); |
| gen.writeStartArray(); |
| for (Name alias : aliases) |
| gen.writeString(alias.getQualified(name.space)); |
| gen.writeEndArray(); |
| } |
| |
| } |
| |
| private static class SeenPair { |
| private Object s1; private Object s2; |
| private SeenPair(Object s1, Object s2) { this.s1 = s1; this.s2 = s2; } |
| public boolean equals(Object o) { |
| return this.s1 == ((SeenPair)o).s1 && this.s2 == ((SeenPair)o).s2; |
| } |
| public int hashCode() { |
| return System.identityHashCode(s1) + System.identityHashCode(s2); |
| } |
| } |
| |
| private static final ThreadLocal<Set> SEEN_EQUALS = new ThreadLocal<Set>() { |
| protected Set initialValue() { return new HashSet(); } |
| }; |
| private static final ThreadLocal<Map> SEEN_HASHCODE = new ThreadLocal<Map>() { |
| protected Map initialValue() { return new IdentityHashMap(); } |
| }; |
| |
| @SuppressWarnings(value="unchecked") |
| private static class RecordSchema extends NamedSchema { |
| private List<Field> fields; |
| private Map<String, Field> fieldMap; |
| private final boolean isError; |
| public RecordSchema(Name name, String doc, boolean isError) { |
| super(Type.RECORD, name, doc); |
| this.isError = isError; |
| } |
| |
| public RecordSchema(Name name, String doc, boolean isError, |
| List<Field> fields) { |
| super(Type.RECORD, name, doc); |
| this.isError = isError; |
| setFields(fields); |
| } |
| |
| public boolean isError() { return isError; } |
| |
| @Override |
| public Field getField(String fieldname) { |
| if (fieldMap == null) |
| throw new AvroRuntimeException("Schema fields not set yet"); |
| return fieldMap.get(fieldname); |
| } |
| |
| @Override |
| public List<Field> getFields() { |
| if (fields == null) |
| throw new AvroRuntimeException("Schema fields not set yet"); |
| return fields; |
| } |
| |
| @Override |
| public void setFields(List<Field> fields) { |
| if (this.fields != null) { |
| throw new AvroRuntimeException("Fields are already set"); |
| } |
| int i = 0; |
| fieldMap = new HashMap<String, Field>(); |
| LockableArrayList ff = new LockableArrayList(); |
| for (Field f : fields) { |
| if (f.position != -1) |
| throw new AvroRuntimeException("Field already used: " + f); |
| f.position = i++; |
| final Field existingField = fieldMap.put(f.name(), f); |
| if (existingField != null) { |
| throw new AvroRuntimeException(String.format( |
| "Duplicate field %s in record %s: %s and %s.", |
| f.name(), name, f, existingField)); |
| } |
| ff.add(f); |
| } |
| this.fields = ff.lock(); |
| this.hashCode = NO_HASHCODE; |
| } |
| public boolean equals(Object o) { |
| if (o == this) return true; |
| if (!(o instanceof RecordSchema)) return false; |
| RecordSchema that = (RecordSchema)o; |
| if (!equalCachedHash(that)) return false; |
| if (!equalNames(that)) return false; |
| if (!props.equals(that.props)) return false; |
| Set seen = SEEN_EQUALS.get(); |
| SeenPair here = new SeenPair(this, o); |
| if (seen.contains(here)) return true; // prevent stack overflow |
| boolean first = seen.isEmpty(); |
| try { |
| seen.add(here); |
| return fields.equals(((RecordSchema)o).fields); |
| } finally { |
| if (first) seen.clear(); |
| } |
| } |
| @Override int computeHash() { |
| Map seen = SEEN_HASHCODE.get(); |
| if (seen.containsKey(this)) return 0; // prevent stack overflow |
| boolean first = seen.isEmpty(); |
| try { |
| seen.put(this, this); |
| return super.computeHash() + fields.hashCode(); |
| } finally { |
| if (first) seen.clear(); |
| } |
| } |
| void toJson(Names names, JsonGenerator gen) throws IOException { |
| if (writeNameRef(names, gen)) return; |
| String savedSpace = names.space; // save namespace |
| gen.writeStartObject(); |
| gen.writeStringField("type", isError?"error":"record"); |
| writeName(names, gen); |
| names.space = name.space; // set default namespace |
| if (getDoc() != null) |
| gen.writeStringField("doc", getDoc()); |
| |
| if (fields != null) { |
| gen.writeFieldName("fields"); |
| fieldsToJson(names, gen); |
| } |
| |
| writeProps(gen); |
| aliasesToJson(gen); |
| gen.writeEndObject(); |
| names.space = savedSpace; // restore namespace |
| } |
| |
| void fieldsToJson(Names names, JsonGenerator gen) throws IOException { |
| gen.writeStartArray(); |
| for (Field f : fields) { |
| gen.writeStartObject(); |
| gen.writeStringField("name", f.name()); |
| gen.writeFieldName("type"); |
| f.schema().toJson(names, gen); |
| if (f.doc() != null) |
| gen.writeStringField("doc", f.doc()); |
| if (f.defaultValue() != null) { |
| gen.writeFieldName("default"); |
| gen.writeTree(f.defaultValue()); |
| } |
| if (f.order() != Field.Order.ASCENDING) |
| gen.writeStringField("order", f.order().name); |
| if (f.aliases != null && f.aliases.size() != 0) { |
| gen.writeFieldName("aliases"); |
| gen.writeStartArray(); |
| for (String alias : f.aliases) |
| gen.writeString(alias); |
| gen.writeEndArray(); |
| } |
| f.writeProps(gen); |
| gen.writeEndObject(); |
| } |
| gen.writeEndArray(); |
| } |
| } |
| |
| private static class EnumSchema extends NamedSchema { |
| private final List<String> symbols; |
| private final Map<String,Integer> ordinals; |
| public EnumSchema(Name name, String doc, |
| LockableArrayList<String> symbols) { |
| super(Type.ENUM, name, doc); |
| this.symbols = symbols.lock(); |
| this.ordinals = new HashMap<String,Integer>(); |
| int i = 0; |
| for (String symbol : symbols) |
| if (ordinals.put(validateName(symbol), i++) != null) |
| throw new SchemaParseException("Duplicate enum symbol: "+symbol); |
| } |
| public List<String> getEnumSymbols() { return symbols; } |
| public boolean hasEnumSymbol(String symbol) { |
| return ordinals.containsKey(symbol); } |
| public int getEnumOrdinal(String symbol) { return ordinals.get(symbol); } |
| public boolean equals(Object o) { |
| if (o == this) return true; |
| if (!(o instanceof EnumSchema)) return false; |
| EnumSchema that = (EnumSchema)o; |
| return equalCachedHash(that) |
| && equalNames(that) |
| && symbols.equals(that.symbols) |
| && props.equals(that.props); |
| } |
| @Override int computeHash() { return super.computeHash() + symbols.hashCode(); } |
| void toJson(Names names, JsonGenerator gen) throws IOException { |
| if (writeNameRef(names, gen)) return; |
| gen.writeStartObject(); |
| gen.writeStringField("type", "enum"); |
| writeName(names, gen); |
| if (getDoc() != null) |
| gen.writeStringField("doc", getDoc()); |
| gen.writeArrayFieldStart("symbols"); |
| for (String symbol : symbols) |
| gen.writeString(symbol); |
| gen.writeEndArray(); |
| writeProps(gen); |
| aliasesToJson(gen); |
| gen.writeEndObject(); |
| } |
| } |
| |
| private static class ArraySchema extends Schema { |
| private final Schema elementType; |
| public ArraySchema(Schema elementType) { |
| super(Type.ARRAY); |
| this.elementType = elementType; |
| } |
| public Schema getElementType() { return elementType; } |
| public boolean equals(Object o) { |
| if (o == this) return true; |
| if (!(o instanceof ArraySchema)) return false; |
| ArraySchema that = (ArraySchema)o; |
| return equalCachedHash(that) |
| && elementType.equals(that.elementType) |
| && props.equals(that.props); |
| } |
| @Override int computeHash() { |
| return super.computeHash() + elementType.computeHash(); |
| } |
| void toJson(Names names, JsonGenerator gen) throws IOException { |
| gen.writeStartObject(); |
| gen.writeStringField("type", "array"); |
| gen.writeFieldName("items"); |
| elementType.toJson(names, gen); |
| writeProps(gen); |
| gen.writeEndObject(); |
| } |
| } |
| |
| private static class MapSchema extends Schema { |
| private final Schema valueType; |
| public MapSchema(Schema valueType) { |
| super(Type.MAP); |
| this.valueType = valueType; |
| } |
| public Schema getValueType() { return valueType; } |
| public boolean equals(Object o) { |
| if (o == this) return true; |
| if (!(o instanceof MapSchema)) return false; |
| MapSchema that = (MapSchema)o; |
| return equalCachedHash(that) |
| && valueType.equals(that.valueType) |
| && props.equals(that.props); |
| } |
| @Override int computeHash() { |
| return super.computeHash() + valueType.computeHash(); |
| } |
| void toJson(Names names, JsonGenerator gen) throws IOException { |
| gen.writeStartObject(); |
| gen.writeStringField("type", "map"); |
| gen.writeFieldName("values"); |
| valueType.toJson(names, gen); |
| writeProps(gen); |
| gen.writeEndObject(); |
| } |
| } |
| |
| private static class UnionSchema extends Schema { |
| private final List<Schema> types; |
| private final Map<String,Integer> indexByName |
| = new HashMap<String,Integer>(); |
| public UnionSchema(LockableArrayList<Schema> types) { |
| super(Type.UNION); |
| this.types = types.lock(); |
| int index = 0; |
| for (Schema type : types) { |
| if (type.getType() == Type.UNION) |
| throw new AvroRuntimeException("Nested union: "+this); |
| String name = type.getFullName(); |
| if (name == null) |
| throw new AvroRuntimeException("Nameless in union:"+this); |
| if (indexByName.put(name, index++) != null) |
| throw new AvroRuntimeException("Duplicate in union:" + name); |
| } |
| } |
| public List<Schema> getTypes() { return types; } |
| public Integer getIndexNamed(String name) { return indexByName.get(name); } |
| public boolean equals(Object o) { |
| if (o == this) return true; |
| if (!(o instanceof UnionSchema)) return false; |
| UnionSchema that = (UnionSchema)o; |
| return equalCachedHash(that) |
| && types.equals(that.types) |
| && props.equals(that.props); |
| } |
| @Override int computeHash() { |
| int hash = super.computeHash(); |
| for (Schema type : types) |
| hash += type.computeHash(); |
| return hash; |
| } |
| |
| @Override |
| public void addProp(String name, String value) { |
| throw new AvroRuntimeException("Can't set properties on a union: "+this); |
| } |
| |
| void toJson(Names names, JsonGenerator gen) throws IOException { |
| gen.writeStartArray(); |
| for (Schema type : types) |
| type.toJson(names, gen); |
| gen.writeEndArray(); |
| } |
| } |
| |
| private static class FixedSchema extends NamedSchema { |
| private final int size; |
| public FixedSchema(Name name, String doc, int size) { |
| super(Type.FIXED, name, doc); |
| if (size < 0) |
| throw new IllegalArgumentException("Invalid fixed size: "+size); |
| this.size = size; |
| } |
| public int getFixedSize() { return size; } |
| public boolean equals(Object o) { |
| if (o == this) return true; |
| if (!(o instanceof FixedSchema)) return false; |
| FixedSchema that = (FixedSchema)o; |
| return equalCachedHash(that) |
| && equalNames(that) |
| && size == that.size |
| && props.equals(that.props); |
| } |
| @Override int computeHash() { return super.computeHash() + size; } |
| void toJson(Names names, JsonGenerator gen) throws IOException { |
| if (writeNameRef(names, gen)) return; |
| gen.writeStartObject(); |
| gen.writeStringField("type", "fixed"); |
| writeName(names, gen); |
| if (getDoc() != null) |
| gen.writeStringField("doc", getDoc()); |
| gen.writeNumberField("size", size); |
| writeProps(gen); |
| aliasesToJson(gen); |
| gen.writeEndObject(); |
| } |
| } |
| |
| private static class StringSchema extends Schema { |
| public StringSchema() { super(Type.STRING); } |
| } |
| |
| private static class BytesSchema extends Schema { |
| public BytesSchema() { super(Type.BYTES); } |
| } |
| |
| private static class IntSchema extends Schema { |
| public IntSchema() { super(Type.INT); } |
| } |
| |
| private static class LongSchema extends Schema { |
| public LongSchema() { super(Type.LONG); } |
| } |
| |
| private static class FloatSchema extends Schema { |
| public FloatSchema() { super(Type.FLOAT); } |
| } |
| |
| private static class DoubleSchema extends Schema { |
| public DoubleSchema() { super(Type.DOUBLE); } |
| } |
| |
| private static class BooleanSchema extends Schema { |
| public BooleanSchema() { super(Type.BOOLEAN); } |
| } |
| |
| private static class NullSchema extends Schema { |
| public NullSchema() { super(Type.NULL); } |
| } |
| |
| /** A parser for JSON-format schemas. Each named schema parsed with a parser |
| * is added to the names known to the parser so that subsequently parsed |
| * schemas may refer to it by name. */ |
| public static class Parser { |
| private Names names = new Names(); |
| private boolean validate = true; |
| private boolean validateDefaults = false; |
| |
| /** Adds the provided types to the set of defined, named types known to |
| * this parser. */ |
| public Parser addTypes(Map<String,Schema> types) { |
| for (Schema s : types.values()) |
| names.add(s); |
| return this; |
| } |
| |
| /** Returns the set of defined, named types known to this parser. */ |
| public Map<String,Schema> getTypes() { |
| Map<String,Schema> result = new LinkedHashMap<String,Schema>(); |
| for (Schema s : names.values()) |
| result.put(s.getFullName(), s); |
| return result; |
| } |
| |
| /** Enable or disable name validation. */ |
| public Parser setValidate(boolean validate) { |
| this.validate = validate; |
| return this; |
| } |
| |
| /** True iff names are validated. True by default. */ |
| public boolean getValidate() { return this.validate; } |
| |
| /** Enable or disable default value validation. */ |
| public Parser setValidateDefaults(boolean validateDefaults) { |
| this.validateDefaults = validateDefaults; |
| return this; |
| } |
| |
| /** True iff default values are validated. False by default. */ |
| public boolean getValidateDefaults() { return this.validateDefaults; } |
| |
| /** Parse a schema from the provided file. |
| * If named, the schema is added to the names known to this parser. */ |
| public Schema parse(File file) throws IOException { |
| return parse(FACTORY.createJsonParser(file)); |
| } |
| |
| /** Parse a schema from the provided stream. |
| * If named, the schema is added to the names known to this parser. |
| * The input stream stays open after the parsing. */ |
| public Schema parse(InputStream in) throws IOException { |
| return parse(FACTORY.createJsonParser(in).disable( |
| JsonParser.Feature.AUTO_CLOSE_SOURCE)); |
| } |
| |
| /** Read a schema from one or more json strings */ |
| public Schema parse(String s, String... more) { |
| StringBuilder b = new StringBuilder(s); |
| for (String part : more) |
| b.append(part); |
| return parse(b.toString()); |
| } |
| |
| /** Parse a schema from the provided string. |
| * If named, the schema is added to the names known to this parser. */ |
| public Schema parse(String s) { |
| try { |
| return parse(FACTORY.createJsonParser(new StringReader(s))); |
| } catch (IOException e) { |
| throw new SchemaParseException(e); |
| } |
| } |
| |
| private Schema parse(JsonParser parser) throws IOException { |
| boolean saved = validateNames.get(); |
| boolean savedValidateDefaults = VALIDATE_DEFAULTS.get(); |
| try { |
| validateNames.set(validate); |
| VALIDATE_DEFAULTS.set(validateDefaults); |
| return Schema.parse(MAPPER.readTree(parser), names); |
| } catch (JsonParseException e) { |
| throw new SchemaParseException(e); |
| } finally { |
| parser.close(); |
| validateNames.set(saved); |
| VALIDATE_DEFAULTS.set(savedValidateDefaults); |
| } |
| } |
| } |
| |
| /** |
| * Constructs a Schema object from JSON schema file <tt>file</tt>. |
| * The contents of <tt>file</tt> is expected to be in UTF-8 format. |
| * @param file The file to read the schema from. |
| * @return The freshly built Schema. |
| * @throws IOException if there was trouble reading the contents |
| * @throws JsonParseException if the contents are invalid |
| * @deprecated use {@link Schema.Parser} instead. |
| */ |
| public static Schema parse(File file) throws IOException { |
| return new Parser().parse(file); |
| } |
| |
| /** |
| * Constructs a Schema object from JSON schema stream <tt>in</tt>. |
| * The contents of <tt>in</tt> is expected to be in UTF-8 format. |
| * @param in The input stream to read the schema from. |
| * @return The freshly built Schema. |
| * @throws IOException if there was trouble reading the contents |
| * @throws JsonParseException if the contents are invalid |
| * @deprecated use {@link Schema.Parser} instead. |
| */ |
| public static Schema parse(InputStream in) throws IOException { |
| return new Parser().parse(in); |
| } |
| |
| /** Construct a schema from <a href="http://json.org/">JSON</a> text. |
| * @deprecated use {@link Schema.Parser} instead. |
| */ |
| public static Schema parse(String jsonSchema) { |
| return new Parser().parse(jsonSchema); |
| } |
| |
| /** Construct a schema from <a href="http://json.org/">JSON</a> text. |
| * @param validate true if names should be validated, false if not. |
| * @deprecated use {@link Schema.Parser} instead. |
| */ |
| public static Schema parse(String jsonSchema, boolean validate) { |
| return new Parser().setValidate(validate).parse(jsonSchema); |
| } |
| |
| static final Map<String,Type> PRIMITIVES = new HashMap<String,Type>(); |
| static { |
| PRIMITIVES.put("string", Type.STRING); |
| PRIMITIVES.put("bytes", Type.BYTES); |
| PRIMITIVES.put("int", Type.INT); |
| PRIMITIVES.put("long", Type.LONG); |
| PRIMITIVES.put("float", Type.FLOAT); |
| PRIMITIVES.put("double", Type.DOUBLE); |
| PRIMITIVES.put("boolean", Type.BOOLEAN); |
| PRIMITIVES.put("null", Type.NULL); |
| } |
| |
| static class Names extends LinkedHashMap<Name, Schema> { |
| private String space; // default namespace |
| |
| public Names() {} |
| public Names(String space) { this.space = space; } |
| |
| public String space() { return space; } |
| public void space(String space) { this.space = space; } |
| |
| @Override |
| public Schema get(Object o) { |
| Name name; |
| if (o instanceof String) { |
| Type primitive = PRIMITIVES.get((String)o); |
| if (primitive != null) return Schema.create(primitive); |
| name = new Name((String)o, space); |
| if (!containsKey(name)) // if not in default |
| name = new Name((String)o, ""); // try anonymous |
| } else { |
| name = (Name)o; |
| } |
| return super.get(name); |
| } |
| public boolean contains(Schema schema) { |
| return get(((NamedSchema)schema).name) != null; |
| } |
| public void add(Schema schema) { |
| put(((NamedSchema)schema).name, schema); |
| } |
| @Override |
| public Schema put(Name name, Schema schema) { |
| if (containsKey(name)) |
| throw new SchemaParseException("Can't redefine: "+name); |
| return super.put(name, schema); |
| } |
| } |
| |
| private static ThreadLocal<Boolean> validateNames |
| = new ThreadLocal<Boolean>() { |
| @Override protected Boolean initialValue() { |
| return true; |
| } |
| }; |
| |
| private static String validateName(String name) { |
| if (!validateNames.get()) return name; // not validating names |
| int length = name.length(); |
| if (length == 0) |
| throw new SchemaParseException("Empty name"); |
| char first = name.charAt(0); |
| if (!(Character.isLetter(first) || first == '_')) |
| throw new SchemaParseException("Illegal initial character: "+name); |
| for (int i = 1; i < length; i++) { |
| char c = name.charAt(i); |
| if (!(Character.isLetterOrDigit(c) || c == '_')) |
| throw new SchemaParseException("Illegal character in: "+name); |
| } |
| return name; |
| } |
| |
| private static final ThreadLocal<Boolean> VALIDATE_DEFAULTS |
| = new ThreadLocal<Boolean>() { |
| @Override protected Boolean initialValue() { |
| return false; |
| } |
| }; |
| |
| private static JsonNode validateDefault(String fieldName, Schema schema, |
| JsonNode defaultValue) { |
| if (VALIDATE_DEFAULTS.get() && (defaultValue != null) |
| && !isValidDefault(schema, defaultValue)) { // invalid default |
| String message = "Invalid default for field "+fieldName |
| +": "+defaultValue+" not a "+schema; |
| throw new AvroTypeException(message); // throw exception |
| } |
| return defaultValue; |
| } |
| |
| private static boolean isValidDefault(Schema schema, JsonNode defaultValue) { |
| if (defaultValue == null) |
| return false; |
| switch (schema.getType()) { |
| case STRING: |
| case BYTES: |
| case ENUM: |
| case FIXED: |
| return defaultValue.isTextual(); |
| case INT: |
| case LONG: |
| case FLOAT: |
| case DOUBLE: |
| return defaultValue.isNumber(); |
| case BOOLEAN: |
| return defaultValue.isBoolean(); |
| case NULL: |
| return defaultValue.isNull(); |
| case ARRAY: |
| if (!defaultValue.isArray()) |
| return false; |
| for (JsonNode element : defaultValue) |
| if (!isValidDefault(schema.getElementType(), element)) |
| return false; |
| return true; |
| case MAP: |
| if (!defaultValue.isObject()) |
| return false; |
| for (JsonNode value : defaultValue) |
| if (!isValidDefault(schema.getValueType(), value)) |
| return false; |
| return true; |
| case UNION: // union default: first branch |
| return isValidDefault(schema.getTypes().get(0), defaultValue); |
| case RECORD: |
| if (!defaultValue.isObject()) |
| return false; |
| for (Field field : schema.getFields()) |
| if (!isValidDefault(field.schema(), |
| defaultValue.has(field.name()) |
| ? defaultValue.get(field.name()) |
| : field.defaultValue())) |
| return false; |
| return true; |
| default: |
| return false; |
| } |
| } |
| |
| /** @see #parse(String) */ |
| static Schema parse(JsonNode schema, Names names) { |
| if (schema.isTextual()) { // name |
| Schema result = names.get(schema.getTextValue()); |
| if (result == null) |
| throw new SchemaParseException("Undefined name: "+schema); |
| return result; |
| } else if (schema.isObject()) { |
| Schema result; |
| String type = getRequiredText(schema, "type", "No type"); |
| Name name = null; |
| String savedSpace = names.space(); |
| String doc = null; |
| if (type.equals("record") || type.equals("error") |
| || type.equals("enum") || type.equals("fixed")) { |
| String space = getOptionalText(schema, "namespace"); |
| doc = getOptionalText(schema, "doc"); |
| if (space == null) |
| space = names.space(); |
| name = new Name(getRequiredText(schema, "name", "No name in schema"), |
| space); |
| if (name.space != null) { // set default namespace |
| names.space(name.space); |
| } |
| } |
| if (PRIMITIVES.containsKey(type)) { // primitive |
| result = create(PRIMITIVES.get(type)); |
| } else if (type.equals("record") || type.equals("error")) { // record |
| List<Field> fields = new ArrayList<Field>(); |
| result = new RecordSchema(name, doc, type.equals("error")); |
| if (name != null) names.add(result); |
| JsonNode fieldsNode = schema.get("fields"); |
| if (fieldsNode == null || !fieldsNode.isArray()) |
| throw new SchemaParseException("Record has no fields: "+schema); |
| for (JsonNode field : fieldsNode) { |
| String fieldName = getRequiredText(field, "name", "No field name"); |
| String fieldDoc = getOptionalText(field, "doc"); |
| JsonNode fieldTypeNode = field.get("type"); |
| if (fieldTypeNode == null) |
| throw new SchemaParseException("No field type: "+field); |
| if (fieldTypeNode.isTextual() |
| && names.get(fieldTypeNode.getTextValue()) == null) |
| throw new SchemaParseException |
| (fieldTypeNode+" is not a defined name." |
| +" The type of the \""+fieldName+"\" field must be" |
| +" a defined name or a {\"type\": ...} expression."); |
| Schema fieldSchema = parse(fieldTypeNode, names); |
| Field.Order order = Field.Order.ASCENDING; |
| JsonNode orderNode = field.get("order"); |
| if (orderNode != null) |
| order = Field.Order.valueOf(orderNode.getTextValue().toUpperCase(Locale.ENGLISH)); |
| JsonNode defaultValue = field.get("default"); |
| if (defaultValue != null |
| && (Type.FLOAT.equals(fieldSchema.getType()) |
| || Type.DOUBLE.equals(fieldSchema.getType())) |
| && defaultValue.isTextual()) |
| defaultValue = |
| new DoubleNode(Double.valueOf(defaultValue.getTextValue())); |
| Field f = new Field(fieldName, fieldSchema, |
| fieldDoc, defaultValue, order); |
| Iterator<String> i = field.getFieldNames(); |
| while (i.hasNext()) { // add field props |
| String prop = i.next(); |
| if (!FIELD_RESERVED.contains(prop)) |
| f.addProp(prop, field.get(prop)); |
| } |
| f.aliases = parseAliases(field); |
| fields.add(f); |
| } |
| result.setFields(fields); |
| } else if (type.equals("enum")) { // enum |
| JsonNode symbolsNode = schema.get("symbols"); |
| if (symbolsNode == null || !symbolsNode.isArray()) |
| throw new SchemaParseException("Enum has no symbols: "+schema); |
| LockableArrayList<String> symbols = new LockableArrayList<String>(); |
| for (JsonNode n : symbolsNode) |
| symbols.add(n.getTextValue()); |
| result = new EnumSchema(name, doc, symbols); |
| if (name != null) names.add(result); |
| } else if (type.equals("array")) { // array |
| JsonNode itemsNode = schema.get("items"); |
| if (itemsNode == null) |
| throw new SchemaParseException("Array has no items type: "+schema); |
| result = new ArraySchema(parse(itemsNode, names)); |
| } else if (type.equals("map")) { // map |
| JsonNode valuesNode = schema.get("values"); |
| if (valuesNode == null) |
| throw new SchemaParseException("Map has no values type: "+schema); |
| result = new MapSchema(parse(valuesNode, names)); |
| } else if (type.equals("fixed")) { // fixed |
| JsonNode sizeNode = schema.get("size"); |
| if (sizeNode == null || !sizeNode.isInt()) |
| throw new SchemaParseException("Invalid or no size: "+schema); |
| result = new FixedSchema(name, doc, sizeNode.getIntValue()); |
| if (name != null) names.add(result); |
| } else |
| throw new SchemaParseException("Type not supported: "+type); |
| Iterator<String> i = schema.getFieldNames(); |
| while (i.hasNext()) { // add properties |
| String prop = i.next(); |
| if (!SCHEMA_RESERVED.contains(prop)) // ignore reserved |
| result.addProp(prop, schema.get(prop)); |
| } |
| // parse logical type if present |
| result.logicalType = LogicalTypes.fromSchemaIgnoreInvalid(result); |
| names.space(savedSpace); // restore space |
| if (result instanceof NamedSchema) { |
| Set<String> aliases = parseAliases(schema); |
| if (aliases != null) // add aliases |
| for (String alias : aliases) |
| result.addAlias(alias); |
| } |
| return result; |
| } else if (schema.isArray()) { // union |
| LockableArrayList<Schema> types = |
| new LockableArrayList<Schema>(schema.size()); |
| for (JsonNode typeNode : schema) |
| types.add(parse(typeNode, names)); |
| return new UnionSchema(types); |
| } else { |
| throw new SchemaParseException("Schema not yet supported: "+schema); |
| } |
| } |
| |
| static Set<String> parseAliases(JsonNode node) { |
| JsonNode aliasesNode = node.get("aliases"); |
| if (aliasesNode == null) |
| return null; |
| if (!aliasesNode.isArray()) |
| throw new SchemaParseException("aliases not an array: "+node); |
| Set<String> aliases = new LinkedHashSet<String>(); |
| for (JsonNode aliasNode : aliasesNode) { |
| if (!aliasNode.isTextual()) |
| throw new SchemaParseException("alias not a string: "+aliasNode); |
| aliases.add(aliasNode.getTextValue()); |
| } |
| return aliases; |
| } |
| |
| /** Extracts text value associated to key from the container JsonNode, |
| * and throws {@link SchemaParseException} if it doesn't exist. |
| * |
| * @param container Container where to find key. |
| * @param key Key to look for in container. |
| * @param error String to prepend to the SchemaParseException. |
| */ |
| private static String getRequiredText(JsonNode container, String key, |
| String error) { |
| String out = getOptionalText(container, key); |
| if (null == out) { |
| throw new SchemaParseException(error + ": " + container); |
| } |
| return out; |
| } |
| |
| /** Extracts text value associated to key from the container JsonNode. */ |
| private static String getOptionalText(JsonNode container, String key) { |
| JsonNode jsonNode = container.get(key); |
| return jsonNode != null ? jsonNode.getTextValue() : null; |
| } |
| |
| /** |
| * Parses a string as Json. |
| * @deprecated use {@link org.apache.avro.data.Json#parseJson(String)} |
| */ |
| @Deprecated |
| public static JsonNode parseJson(String s) { |
| try { |
| return MAPPER.readTree(FACTORY.createJsonParser(new StringReader(s))); |
| } catch (JsonParseException e) { |
| throw new RuntimeException(e); |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| /** Rewrite a writer's schema using the aliases from a reader's schema. This |
| * permits reading records, enums and fixed schemas whose names have changed, |
| * and records whose field names have changed. The returned schema always |
| * contains the same data elements in the same order, but with possibly |
| * different names. */ |
| public static Schema applyAliases(Schema writer, Schema reader) { |
| if (writer == reader) return writer; // same schema |
| |
| // create indexes of names |
| Map<Schema,Schema> seen = new IdentityHashMap<Schema,Schema>(1); |
| Map<Name,Name> aliases = new HashMap<Name, Name>(1); |
| Map<Name,Map<String,String>> fieldAliases = |
| new HashMap<Name, Map<String,String>>(1); |
| getAliases(reader, seen, aliases, fieldAliases); |
| |
| if (aliases.size() == 0 && fieldAliases.size() == 0) |
| return writer; // no aliases |
| |
| seen.clear(); |
| return applyAliases(writer, seen, aliases, fieldAliases); |
| } |
| |
| private static Schema applyAliases(Schema s, Map<Schema,Schema> seen, |
| Map<Name,Name> aliases, |
| Map<Name,Map<String,String>> fieldAliases){ |
| |
| Name name = s instanceof NamedSchema ? ((NamedSchema)s).name : null; |
| Schema result = s; |
| switch (s.getType()) { |
| case RECORD: |
| if (seen.containsKey(s)) return seen.get(s); // break loops |
| if (aliases.containsKey(name)) |
| name = aliases.get(name); |
| result = Schema.createRecord(name.full, s.getDoc(), null, s.isError()); |
| seen.put(s, result); |
| List<Field> newFields = new ArrayList<Field>(); |
| for (Field f : s.getFields()) { |
| Schema fSchema = applyAliases(f.schema, seen, aliases, fieldAliases); |
| String fName = getFieldAlias(name, f.name, fieldAliases); |
| Field newF = new Field(fName, fSchema, f.doc, f.defaultValue, f.order); |
| newF.props.putAll(f.props); // copy props |
| newFields.add(newF); |
| } |
| result.setFields(newFields); |
| break; |
| case ENUM: |
| if (aliases.containsKey(name)) |
| result = Schema.createEnum(aliases.get(name).full, s.getDoc(), null, |
| s.getEnumSymbols()); |
| break; |
| case ARRAY: |
| Schema e = applyAliases(s.getElementType(), seen, aliases, fieldAliases); |
| if (e != s.getElementType()) |
| result = Schema.createArray(e); |
| break; |
| case MAP: |
| Schema v = applyAliases(s.getValueType(), seen, aliases, fieldAliases); |
| if (v != s.getValueType()) |
| result = Schema.createMap(v); |
| break; |
| case UNION: |
| List<Schema> types = new ArrayList<Schema>(); |
| for (Schema branch : s.getTypes()) |
| types.add(applyAliases(branch, seen, aliases, fieldAliases)); |
| result = Schema.createUnion(types); |
| break; |
| case FIXED: |
| if (aliases.containsKey(name)) |
| result = Schema.createFixed(aliases.get(name).full, s.getDoc(), null, |
| s.getFixedSize()); |
| break; |
| } |
| if (result != s) |
| result.props.putAll(s.props); // copy props |
| return result; |
| } |
| |
| |
| private static void getAliases(Schema schema, |
| Map<Schema,Schema> seen, |
| Map<Name,Name> aliases, |
| Map<Name,Map<String,String>> fieldAliases) { |
| if (schema instanceof NamedSchema) { |
| NamedSchema namedSchema = (NamedSchema)schema; |
| if (namedSchema.aliases != null) |
| for (Name alias : namedSchema.aliases) |
| aliases.put(alias, namedSchema.name); |
| } |
| switch (schema.getType()) { |
| case RECORD: |
| if (seen.containsKey(schema)) return; // break loops |
| seen.put(schema, schema); |
| RecordSchema record = (RecordSchema)schema; |
| for (Field field : schema.getFields()) { |
| if (field.aliases != null) |
| for (String fieldAlias : field.aliases) { |
| Map<String,String> recordAliases = fieldAliases.get(record.name); |
| if (recordAliases == null) |
| fieldAliases.put(record.name, |
| recordAliases = new HashMap<String,String>()); |
| recordAliases.put(fieldAlias, field.name); |
| } |
| getAliases(field.schema, seen, aliases, fieldAliases); |
| } |
| if (record.aliases != null && fieldAliases.containsKey(record.name)) |
| for (Name recordAlias : record.aliases) |
| fieldAliases.put(recordAlias, fieldAliases.get(record.name)); |
| break; |
| case ARRAY: |
| getAliases(schema.getElementType(), seen, aliases, fieldAliases); |
| break; |
| case MAP: |
| getAliases(schema.getValueType(), seen, aliases, fieldAliases); |
| break; |
| case UNION: |
| for (Schema s : schema.getTypes()) |
| getAliases(s, seen, aliases, fieldAliases); |
| break; |
| } |
| } |
| |
| private static String getFieldAlias |
| (Name record, String field, Map<Name,Map<String,String>> fieldAliases) { |
| Map<String,String> recordAliases = fieldAliases.get(record); |
| if (recordAliases == null) |
| return field; |
| String alias = recordAliases.get(field); |
| if (alias == null) |
| return field; |
| return alias; |
| } |
| |
| /** |
| * No change is permitted on LockableArrayList once lock() has been |
| * called on it. |
| * @param <E> |
| */ |
| |
| /* |
| * This class keeps a boolean variable <tt>locked</tt> which is set |
| * to <tt>true</tt> in the lock() method. It's legal to call |
| * lock() any number of times. Any lock() other than the first one |
| * is a no-op. |
| * |
| * This class throws <tt>IllegalStateException</tt> if a mutating |
| * operation is performed after being locked. Since modifications through |
| * iterator also use the list's mutating operations, this effectively |
| * blocks all modifications. |
| */ |
| static class LockableArrayList<E> extends ArrayList<E> { |
| private static final long serialVersionUID = 1L; |
| private boolean locked = false; |
| |
| public LockableArrayList() { |
| } |
| |
| public LockableArrayList(int size) { |
| super(size); |
| } |
| |
| public LockableArrayList(List<E> types) { |
| super(types); |
| } |
| |
| public LockableArrayList(E... types) { |
| super(types.length); |
| Collections.addAll(this, types); |
| } |
| |
| public List<E> lock() { |
| locked = true; |
| return this; |
| } |
| |
| private void ensureUnlocked() { |
| if (locked) { |
| throw new IllegalStateException(); |
| } |
| } |
| |
| public boolean add(E e) { |
| ensureUnlocked(); |
| return super.add(e); |
| } |
| |
| public boolean remove(Object o) { |
| ensureUnlocked(); |
| return super.remove(o); |
| } |
| |
| public E remove(int index) { |
| ensureUnlocked(); |
| return super.remove(index); |
| } |
| |
| public boolean addAll(Collection<? extends E> c) { |
| ensureUnlocked(); |
| return super.addAll(c); |
| } |
| |
| public boolean addAll(int index, Collection<? extends E> c) { |
| ensureUnlocked(); |
| return super.addAll(index, c); |
| } |
| |
| public boolean removeAll(Collection<?> c) { |
| ensureUnlocked(); |
| return super.removeAll(c); |
| } |
| |
| public boolean retainAll(Collection<?> c) { |
| ensureUnlocked(); |
| return super.retainAll(c); |
| } |
| |
| public void clear() { |
| ensureUnlocked(); |
| super.clear(); |
| } |
| |
| } |
| |
| } |
