public interface BatchContext { } public interface Path {

    /** Return a new Path consisting of this Path concatenated with another Path. */
    Path concat(Path other);

    /** Return a new Path consisting of this Path concatenated with the specified list indices as elements. */
    Path withIndices(int... indices);

    /** Return a new Path consisting of this Path concatenated with the specified elements. */
    Path withElements(String... elements);

    /** Get the encoded elements of this Path. */
    Iterable<String> getElements();

    /**
     * Returns the first element in this path, or null if this is an empty path.
     */
    String head();

    /**
     * Returns the rest of the path after the head (first) element.
     * If this is an empty path, throws IndexOutOfBoundsException,
     * if this path has only one element, return an empty path,
     * otherwise return a new path with elements starting after the head.
     */
    Path tail();

    /**
     * Return a new Path consisting of the rest of the path elements of this path starting with the specified index.
     * @param startIndex 0-based start index
     * @return new Path
     * @throws IndexOutOfBoundsException if path has insufficient size
     */
    Path tail(int startIndex);

    /**
     * Return true if this path starts with the specified path.
     */
    boolean startsWith(Path path);

    /** Get the number of elements in this path. */
    int size();

    /** Return true if this Path has zero elements. */
    boolean isEmpty();
} public interface StructuredDataSupport<K> {

    /**
     * Begin a batch operation. Returns a BatchContext object which must be passed to all mutating methods that
     * participate in this batch operation, and then is passed to applyBatch to execute the batch.
     * @return the BatchContext
     */
    BatchContext beginBatch();

    /**
     * Execute the batch.
     * @param batchContext The BatchContext for this set of mutating operations to execute
     */
    void applyBatch(BatchContext batchContext);

    /**
     * Read an object from a path in the database.
     * @param rowKey the row key for the object to be read
     * @param path the path to the object to be read
     * @param type a TypeReference for the type of object to be returned
     * @param <T> the type of the object to be returned
     * @return the object of type T, or null if not found
     * @throws IllegalArgumentException if path is empty or contains any of the special characters '@#'
     */
    <T> T readFromPath(K rowKey, Path path, TypeReference<T> type);

    /**
     * Write an object to a path in the database.
     * @param rowKey the row key for the object to be written
     * @param path the path to the object to be written
     * @param value the Object to be written
     */
    void writeToPath(K rowKey, Path path, Object value);

    /**
     * Write an object to a path in the database as part of a batch operation.
     * @param rowKey the row key for the object to be written
     * @param path the path to the object to be written
     * @param value the Object to be written
     * @param batchContext the BatchContext that this write operation should participate in
     */
    void writeToPath(K rowKey, Path path, Object value, BatchContext batchContext);

    /**
     * Delete the values found at the specified path.
     * @param rowKey the row key for the object to be deleted
     * @param path the path to the object to be deleted
     */
    void deletePath(K rowKey, Path path);

    /**
     * Delete the values found at the specified path as part of a batch operation.
     * @param rowKey the row key for the object to be deleted
     * @param path the path to the object to be deleted
     * @param batchContext the BatchContext that this delete operation should participate in
     */
    void deletePath(K rowKey, Path path, BatchContext batchContext);

    /**
     * Utility method for creating a Path from individual string elements.
     */
    Path createPath(String... elements);
} public abstract class TypeReference<T>
        implements Comparable<TypeReference<T>>
{
    protected final Type type;

    protected TypeReference()
    {
        Type superClass = getClass().getGenericSuperclass();
        if (superClass instanceof Class<?>) { // sanity check, should never happen
            throw new IllegalArgumentException("Internal error: TypeReference constructed without actual type information");
        }
        /* 22-Dec-2008, tatu: Not sure if this case is safe -- I suspect
         *   it is possible to make it fail?
         *   But let's deal with specific
         *   case when we know an actual use case, and thereby suitable
         *   workarounds for valid case(s) and/or error to throw
         *   on invalid one(s).
         */
        type = ((ParameterizedType) superClass).getActualTypeArguments()[0];
    }

    public Type getType() { return type; }

    /**
     * The only reason we define this method (and require implementation
     * of {@code Comparable}) is to prevent constructing a
     * reference without type information.
     */
    @Override
    public int compareTo(TypeReference<T> o) {
        // just need an implementation, not a good one... hence:
        return 0;
    }
} public class CustomTypeResolverBuilder extends StdTypeResolverBuilder {

    public CustomTypeResolverBuilder() {
        init(JsonTypeInfo.Id.CLASS, new CustomTypeIdResolver());
        inclusion(JsonTypeInfo.As.WRAPPER_ARRAY);
    }

    @Override
    public TypeDeserializer buildTypeDeserializer(DeserializationConfig config,
                                                  JavaType baseType,
                                                  Collection<NamedType> subtypes) {
        return useForType(baseType) ? super.buildTypeDeserializer(config, baseType, subtypes) : null;
    }

    @Override
    public TypeSerializer buildTypeSerializer(SerializationConfig config,
                                              JavaType baseType,
                                              Collection<NamedType> subtypes) {
        return useForType(baseType) ? super.buildTypeSerializer(config, baseType, subtypes) : null;
    }

    private boolean useForType(JavaType t) {
        return t.isArrayType() || Collection.class.isAssignableFrom(t.getRawClass());
    }

    private class CustomTypeIdResolver implements TypeIdResolver {
        private final TypeFactory typeFactory = TypeFactory.defaultInstance();
        private JavaType baseType;

        @Override
        public void init(JavaType bt) {
            baseType = bt;
        }

        @Override
        public String idFromValue(Object value) {
            return idFrom(value, value.getClass());
        }

        @Override
        public String idFromValueAndType(Object value, Class<?> type) {
            return idFrom(value, type);
        }

        @Override
        public JavaType typeFromId(String id)
        {
        /* 30-Jan-2010, tatu: Most ids are basic class names; so let's first
         *    check if any generics info is added; and only then ask factory
         *    to do translation when necessary
         */
            if (id.indexOf('<') > 0) {
                JavaType t = typeFactory.constructFromCanonical(id);
                // note: may want to try combining with specialization (esp for EnumMap)?
                return t;
            }
            try {
                Class<?> cls =  ClassUtil.findClass(id);
                return typeFactory.constructSpecializedType(baseType, cls);
            } catch (ClassNotFoundException ignored) {
                throw new IllegalArgumentException("Invalid type id '"+id+"' (for id type 'Id.class'): no such class found");
            } catch (Exception e) {
                throw new IllegalArgumentException("Invalid type id '"+id+"' (for id type 'Id.class'): "+e.getMessage(), e);
            }
        }

        @Override
        public JsonTypeInfo.Id getMechanism() {
            return JsonTypeInfo.Id.CUSTOM;
        }

        @Override
        public String idFromBaseType()
        {
        /* By default we will just defer to regular handling, handing out the
         * base type; and since there is no value, must just pass null here
         * assuming that implementations can deal with it.
         * Alternative would be to pass a bogus Object, but that does not seem right.
         */
            return idFromValueAndType(null, baseType.getRawClass());
        }


        @SuppressWarnings("ChainOfInstanceofChecks")
        protected final String idFrom(Object value, Class<?> cls)
        {
            // [JACKSON-380] Need to ensure that "enum subtypes" work too
            if (Enum.class.isAssignableFrom(cls)) {
                if (!cls.isEnum()) { // means that it's sub-class of base enum, so:
                    cls = cls.getSuperclass();
                }
            }

            // try to get element type if this is a Set (otherwise we don't care)
            Class<?> elementType = null;
            if (value instanceof Set) {
                // if we have baseType, use that
                if (baseType != null) {
                    JavaType contentType = baseType.getContentType();
                    elementType = contentType.getRawClass();
                } else {
                    Class<?> clazz = value.getClass();
                    do {
                        Type superclassType = clazz.getGenericSuperclass();
                        if (superclassType instanceof ParameterizedType) {
                            ParameterizedType parameterizedSuperclass = (ParameterizedType)superclassType;
                            Type[] elementTypes = parameterizedSuperclass.getActualTypeArguments();
                            if (elementTypes.length == 1 && elementTypes[0] instanceof Class) {
                                elementType = (Class<?>)elementTypes[0];
                            }
                        }
                        clazz = clazz.getSuperclass();
                    } while (elementType == null && clazz != null);
                }
            }

            String str = cls.getName();
            if (str.startsWith("java.util")) {
            /* 25-Jan-2009, tatu: There are some internal classes that
             *   we can not access as is. We need better mechanism; for
             *   now this has to do...
             */
            /* Enum sets and maps are problematic since we MUST know
             * type of contained enums, to be able to deserialize.
             * In addition, EnumSet is not a concrete type either
             */
                if (value instanceof EnumSet<?>) { // Regular- and JumboEnumSet...
                    Class<?> enumClass = ClassUtil.findEnumType((EnumSet<?>) value);
                    // not optimal: but EnumSet is not a customizable type so this is sort of ok
                    str = TypeFactory.defaultInstance().constructCollectionType(EnumSet.class, enumClass).toCanonical();
                } else if (value instanceof EnumMap<?,?>) {
                    Class<?> enumClass = ClassUtil.findEnumType((EnumMap<?,?>) value);
                    Class<?> valueClass = Object.class;
                    // not optimal: but EnumMap is not a customizable type so this is sort of ok
                    str = TypeFactory.defaultInstance().constructMapType(EnumMap.class, enumClass, valueClass).toCanonical();
                } else {
                    String end = str.substring(9);
                    if ((end.startsWith(".Arrays$") || end.startsWith(".Collections$"))
                            && str.contains("List")) {
                    /* 17-Feb-2010, tatus: Another such case: result of
                     *    Arrays.asList() is named like so in Sun JDK...
                     *   Let's just plain old ArrayList in its place
                     * NOTE: chances are there are plenty of similar cases
                     * for other wrappers... (immutable, singleton, synced etc)
                     */
                        str = "java.util.ArrayList";
                    }
                }
            } else if (str.indexOf('$') >= 0) {
            /* Other special handling may be needed for inner classes, [JACKSON-584].
             * The best way to handle would be to find 'hidden' constructor; pass parent
             * value etc (which is actually done for non-anonymous static classes!),
             * but that is just not possible due to various things. So, we will instead
             * try to generalize type into something we will be more likely to be able
             * construct.
             */
                Class<?> outer = ClassUtil.getOuterClass(cls);
                if (outer != null) {
                /* one more check: let's actually not worry if the declared
                 * static type is non-static as well; if so, deserializer does
                 * have a chance at figuring it all out.
                 */
                    Class<?> staticType = baseType.getRawClass();
                    if (ClassUtil.getOuterClass(staticType) == null) {
                        // Is this always correct? Seems like it should be...
                        cls = baseType.getRawClass();
                        str = cls.getName();
                    }
                }
            }

            if (elementType != null) {
                return str + "<" + elementType.getName() + ">";
            }
            return str;
        }
    }
} public class Composer {
    private static final Logger log = LoggerFactory.getLogger(Composer.class);

    private static final String INCONSISTENT_ROOT = "@ROOT";
    private static final Composer INSTANCE = new Composer();

    private Composer() { }

    public static Composer get() {
        return INSTANCE;
    }

    /**
     * Compose a map of simple objects keyed by paths into a single complex object, e.g. a map or list
     *
     * @param simpleObjects input map of decomposed objects, paths mapped to simple values (i.e. strings, numbers, or booleans)
     * @return a complex object such as a map or list decoded from the paths in decomposedObjects
     * @throws IllegalArgumentException if there are unsupported objects types in simpleObjects, or
     *                                  if there is a key that is an empty path
     */
    public Object compose(Map<Path,Object> simpleObjects) {
        if (simpleObjects == null) {
            throw new IllegalArgumentException("simpleObject is null");
        }

        if (simpleObjects.isEmpty()) {
            return Collections.emptyMap();
        }

        // if this is a singleton map with an empty path as the key, then this represents itself a
        // simple object, so just return the value
        if (simpleObjects.size() == 1 && simpleObjects.keySet().iterator().next().size() == 0) {
            return simpleObjects.values().iterator().next();
        }

        // decompose into nested maps by merging the partial map from each path.
        // After composing into nested maps, go through the tree structure and transform SortedMaps into Lists.
        // The reason for a two-pass approach is that the lists may be "sparse" due to deleted indexes, and
        // this is difficult to handle in one pass.
        return transformLists(composeMap(simpleObjects));
    }

    private Map<String,Object> composeMap(Map<Path,Object> simpleObjects) {
        Map<String,Object> composition = new LinkedHashMap<String,Object>(simpleObjects.size());
        for (Map.Entry<Path,Object> entry : simpleObjects.entrySet()) {
            merge(entry, composition);
        }
        return composition;
    }

    @SuppressWarnings("unchecked")
    private void merge(Map.Entry<Path,Object> simpleEntry, Map<String,Object> compositionMap) {
        Path path = simpleEntry.getKey();
        String head = path.head();
        assert head != null;
        Object nextLevelComposition = compositionMap.get(head);
        Path tail = path.tail();
        Object simpleValue = simpleEntry.getValue();
        if (nextLevelComposition == null) {
            mergeEntryIntoEmptySlot(compositionMap, head, tail, simpleValue);
        } else if (Types.isSimple(nextLevelComposition)) {
            mergeEntryWithSimple(compositionMap, nextLevelComposition, head, tail, simpleValue);
        } else {
            mergeEntryWithStructure((Map<String,Object>)nextLevelComposition, tail, simpleValue);
        }
    }

    private void mergeEntryWithStructure(Map<String,Object> nextLevelComposition, Path tail, Object simpleValue) {
        if (tail.isEmpty()) {
            // INCONSISTENCY!! there is a simple value at the same level as a complex object
            // Resolve this by putting this value at the special key "@ROOT" inside the complex object.
            Object previousValue = nextLevelComposition.put(INCONSISTENT_ROOT, simpleValue);
            // not possible to have a previous value, there can only be one key with this path
            if (previousValue != null) {
                // merging two simple values at same level, this cannot happen because map keys are unique
                throw new IllegalStateException("two simple values at same level?");
            }
        } else {
            // simply advance to next level since the head matches a key already there
            merge(new SimpleEntry<Path,Object>(tail, simpleValue), nextLevelComposition);
        }
    }

    private void mergeEntryIntoEmptySlot(Map<String,Object> composition, String head, Path tail, Object simpleValue) {
        if (tail.isEmpty()) {
            composition.put(head, simpleValue);
        } else {
            composition.put(head, composeMap(Collections.singletonMap(tail, simpleValue)));
        }
    }

    private void mergeEntryWithSimple(Map<String,Object> composition,
                                      Object nextLevelComposition,
                                      String head,
                                      Path tail,
                                      Object simpleValue) {
        if (tail.isEmpty()) {
            // merging two simple values at same level, this cannot happen because map keys are unique
            throw new IllegalStateException("two simple values at same level?");
        }

        // merging longer path with simple value
        Map<String,Object> map = composeMap(Collections.singletonMap(tail, simpleValue));
        map.put(INCONSISTENT_ROOT, nextLevelComposition);
        // replace the simple value with map containing simple value and inconsistent root
        composition.put(head, map);
    }

    @SuppressWarnings("unchecked")
    private Object transformLists(Map<String,Object> map) {
        // go through nested maps and transform maps into lists where possible
        if (DefaultPath.isList(map.keySet())) {
            return transformActualList(map);
        }
        // if not a list, then just recursively transform the structure, and also URL-Decode the keys
        Map<String,Object> newMap = new HashMap<String,Object>(map.size());
        for (Map.Entry<String,Object> entry : map.entrySet()) {
            Object value = entry.getValue();
            if (value instanceof Map) {
                Object transformedValue = transformLists((Map<String,Object>)value);
                newMap.put(urlDecode(entry.getKey()), transformedValue);
            } else if (Types.isSimple(value)) {
                newMap.put(urlDecode(entry.getKey()), entry.getValue());
            } else {
                throw new IllegalStateException("found strange object in structure: " + value);
            }
        }
        return newMap;
    }

    @SuppressWarnings("unchecked")
    private Object transformActualList(Map<String,Object> map) {
        SortedMap<Integer,Object> sortedMap = new TreeMap<Integer,Object>();
        List<Object> list = new ArrayList<Object>(map.size());

        int listSize = -1;

        // convert keys into integer indexes and sort
        for (Map.Entry<String,Object> entry : map.entrySet()) {
            Object value = entry.getValue();
            int listIndex = DefaultPath.getListIndex(entry.getKey());
            if (Types.isListTerminator(value)) {
                listSize = listSize == -1 ? listIndex : Math.min(listIndex, listSize);
                continue;
            }

            Object transformedValue;
            if (Types.isSimple(value)) {
                transformedValue = value;
            } else if (value instanceof Map) {
                transformedValue = transformLists((Map<String,Object>)value);
            } else {
                throw new IllegalStateException("found strange object in structure: " + value);
            }
            sortedMap.put(DefaultPath.getListIndex(entry.getKey()), transformedValue);
        }

        // if no listSize was found then something went wrong, but just warn and use whole list found
        if (listSize == -1) {
            log.warn("no list terminator found, using all list elements");
        }

        // copy values into list in sorted order
        int index = 0;
        for (Object value : sortedMap.values()) {
            if (index == listSize) {
                break;
            }
            list.add(value);
            index++;
        }
        return list;
    }

    private String urlDecode(String encodedString) {
        String decodedString;
        try {
            decodedString = URLDecoder.decode(encodedString, "UTF-8");
        } catch (UnsupportedEncodingException ignored) {
            throw new AssertionError("UTF-8 is unknown?");
        }
        return decodedString;
    }
} public class Decomposer {
    private static final Decomposer INSTANCE = new Decomposer();

    private Decomposer() { }

    public static Decomposer get() {
        return INSTANCE;
    }

    /**
     * Decompose a map of arbitrarily complex structured objects into a map of
     * simple objects keyed by paths.
     *
     * @param structures the input map of paths to objects
     * @return a map of simple object keyed by paths
     * @throws IllegalArgumentException if there is an object of unsupported type in structures
     *                                  or if structures is null
     */
    public Map<Path,Object> decompose(Map<Path,Object> structures) {
        if (structures == null) {
            throw new IllegalArgumentException("structures is null");
        }
        Map<Path,Object> decomposed = new HashMap<Path,Object>(structures.size());

        for (Map.Entry<Path,Object> entry : structures.entrySet()) {

            Object structure = entry.getValue();
            Path path = entry.getKey();

            // handle null specially by replacing with a Null token
            if (structure == null) {
                decomposed.put(path, NULL);
                continue;
            }

            if (Types.isSimple(structure)) {
                decomposed.put(path, structure);
                continue;
            }

            Map<Path,Object> decomposedMap = decomposeStructure(structure);

            for (Map.Entry<Path,Object> decomposedEntry : decomposedMap.entrySet()) {
                decomposed.put(path.concat(decomposedEntry.getKey()), decomposedEntry.getValue());
            }
        }
        return decomposed;
    }

    //////// Private Methods //////////

    @SuppressWarnings("ChainOfInstanceofChecks")
    private Map<Path,Object> decomposeStructure(Object structure) {
        Map<Path,Object> decomposedMap;
        if (structure instanceof Map) {
            decomposedMap = normalizeMap((Map<?,?>)structure);
        } else if (structure instanceof List) {
            decomposedMap = normalizeList((List<?>)structure);
        } else {
            throw new IllegalArgumentException("Unsupported data type: " + structure.getClass().getSimpleName());
        }
        return decompose(decomposedMap);
    }

    private Map<Path,Object> normalizeMap(Map<?,?> map) {
        Map<Path,Object> normalized = new HashMap<Path,Object>(map.size());
        for (Map.Entry<?,?> entry : map.entrySet()) {

            Object key = entry.getKey();
            if (!Types.isSimple(key)) {
                throw new IllegalArgumentException(String.format("map key of type %s not supported",
                                                                 key.getClass().getSimpleName()));
            }
            Path keyPath = key instanceof Path ? (Path)key : DefaultPath.fromStrings(key.toString());

            Object value = entry.getValue();
            normalized.put(keyPath, value);
        }
        return normalized;
    }

    private Map<Path,Object> normalizeList(List<?> list) {
        // get type info for list
        // TODO: if this is a set of simple types, then encode the set into the keys using #
        // String type = (String)list.get(0);
        /// get list itself
        List<?> listItself = (List<?>)list.get(1);

        Map<Path,Object> normalized = new HashMap<Path,Object>(listItself.size());
        for (int i = 0; i < listItself.size(); i++) {
            normalized.put(DefaultPath.fromIndex(i), listItself.get(i));
        }
        // add terminator column, issue #20
        normalized.put(DefaultPath.fromIndex(listItself.size()), Types.LIST_TERMINATOR_VALUE);

        return normalized;
    }
} public class DefaultPath implements Path {
    private static final char PATH_DELIMITER_CHAR = '/';
    private static final String LIST_INDEX_PREFIX = "@";

    private static final Function<String,String> urlEncodeFunction = new UrlEncode();

    private final Iterable<String> pathElements;


    /** Create a DefaultPath from an iterable of encoded path element strings. */
    private DefaultPath(Iterable<String> encodedPathElements) {
        pathElements = encodedPathElements;
    }

    /** Create a DefaultPath from un-encoded string elements. */
    public static DefaultPath fromStrings(String... elements) {
        return new DefaultPath(Lists.transform(Arrays.asList(elements), urlEncodeFunction));
    }

    @Override
    public Path concat(Path other) {
        Iterable<String> newPathElements = Iterables.concat(pathElements, other.getElements());
        return new DefaultPath(newPathElements);
    }

    /**
     * Paths always end with the delimiter character in order to facilitate
     * start/finish slice queries in Cassandra.
     * @return the (encoded) String representation of a path.
     */
    @Override
    public String toString() {
        StringBuilder builder = new StringBuilder();
        for (String pathElement : pathElements) {
            builder.append(pathElement);
            builder.append(PATH_DELIMITER_CHAR);
        }
        return builder.toString();
    }

    public static DefaultPath fromIndex(int i) {
        return new DefaultPath(Arrays.asList(LIST_INDEX_PREFIX + i));
    }

    @Override
    public String head() {
        return getFirst(pathElements, null);
    }

    @Override
    public DefaultPath tail() {
        return tail(1);
    }

    @Override
    public DefaultPath tail(int startIndex) {
        return new DefaultPath(skip(pathElements, startIndex));
    }

    @Override
    public boolean isEmpty() {
        return Iterables.isEmpty(pathElements);
    }

    @Override
    public boolean startsWith(Path path) {
        return elementsEqual(limit(pathElements, path.size()),
                             path.getElements());
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }

        return elementsEqual(pathElements, ((DefaultPath)o).pathElements);
    }

    @Override
    public int hashCode() {
        return Lists.newArrayList(pathElements).hashCode();
    }

    @Override
    public int size() {
        return Iterables.size(pathElements);
    }

    @Override
    public Iterable<String> getElements() {
        return unmodifiableIterable(pathElements);
    }

    @Override
    public Path withIndices(int... indices) {
        List<String> newPathElements = Lists.newArrayList(pathElements);
        for (int index : indices) {
            newPathElements.add(LIST_INDEX_PREFIX + index);
        }
        return new DefaultPath(newPathElements);
    }

    @Override
    public Path withElements(String... elements) {
        List<String> newPathElements = Lists.newArrayList(pathElements);
        newPathElements.addAll(Lists.transform(Arrays.asList(elements), urlEncodeFunction));
        return new DefaultPath(newPathElements);
    }

    /** Create a Path object from a String produced by the Path#toString method. */

    public static Path fromEncodedPathString(String pathString) {
        Iterable<String> parts = Splitter.on(PATH_DELIMITER_CHAR).omitEmptyStrings().split(pathString);
        return new DefaultPath(parts);
    }

    public static int getListIndex(String pathElement) {
        if (pathElement.isEmpty()) {
            throw new IllegalStateException("empty path");
        }

        if (!pathElement.startsWith(LIST_INDEX_PREFIX)) {
            throw new IllegalStateException("not a list index");
        }

        String rest = pathElement.substring(LIST_INDEX_PREFIX.length());
        try {
            return Integer.parseInt(rest);
        } catch (NumberFormatException ignored) {
            throw new IllegalStateException("bad format for list index");
        }
    }

    /**
     * Return true if all the given encoded path elements are list indexes.
     * @throws IllegalArgumentException if an empty path is found
     */
    public static boolean isList(Iterable<String> encodedElements) {
        for (String element : encodedElements)  {
            if (!isListIndex(element)) {
                return false;
            }
        }
        return true;
    }

    /**
     * Return true if the first element in this path is a list index.
     */
    private static boolean isListIndex(String pathElement) {
        if (!pathElement.startsWith(LIST_INDEX_PREFIX)) {
            return false;
        }
        String rest = pathElement.substring(LIST_INDEX_PREFIX.length());
        int index;
        try {
            index = Integer.parseInt(rest);
        } catch (NumberFormatException ignored) {
            return false;
        }
        return index >= 0;
    }

    private static class UrlEncode implements Function<String,String> {
        @Nullable
        @Override
        public String apply(@Nullable String s) {
            if (s == null) {
                return null;
            }
            String encodedString;
            try {
                encodedString = URLEncoder.encode(s, "UTF-8");
            } catch (UnsupportedEncodingException ignored) {
                throw new AssertionError("UTF-8 is unknown?");
            }
            return encodedString;
        }
    }
} public class JacksonTypeReference<T> extends TypeReference<T> {
    private final com.ebuddy.cassandra.TypeReference<T> providedType;

    public JacksonTypeReference(com.ebuddy.cassandra.TypeReference<T> providedType) {
        this.providedType = providedType;
    }

    @Override
    public Type getType() {
        return providedType.getType();
    }
} public class StructureConverter {
    private static final String UTF_8 = "UTF-8";
    private static final Charset UTF8_CHARSET = Charset.forName(UTF_8);
    private static final StructureConverter INSTANCE = new StructureConverter();
    protected static final ObjectMapper JSON_MAPPER = new ObjectMapper();

    /**
     * Header char, a unicode non-character, used to flag a JSON deserialized object.
     */
    private static final int HEADER_CHAR = '\uFFFE';
    /**
     * utf-8 encoded bytes for HEADER_CHAR.
     */
    private static final byte[] UTF8_HEADER_BYTES = {(byte)0xef, (byte)0xbf, (byte)0xbe};

    /**
     * Only instantiated once for the static singleton.
     */
    private StructureConverter() { }

    /**
     * Get the singleton instance of StructureConverter.
     *
     * @return the singleton StructureConverter
     */
    public static StructureConverter get() {
        return INSTANCE;
    }

    /**
     * @throws DataFormatException if format of the string is incorrect and could not be parsed as JSON
     */
    public Object fromString(String str) {
        if (str == null) {
            return null;
        }
        return decodeBytes(str.getBytes(UTF8_CHARSET));
    }

    /**
     * @throws DataFormatException if the object could not be encoded as JSON
     */
    public String toString(Object obj) {
        if (obj == null) {
            return null;
        }

        if (obj instanceof String) {
            return (String)obj;
        }

        // write as special header bytes followed by JSON
        // intercept the Null token which stands in for a real null
        if (obj == NULL) {
            obj = null;
        }

        String jsonString = encodeJson(obj);
        char[] chars = new char[jsonString.length() + 1];
        chars[0] = HEADER_CHAR;
        jsonString.getChars(0, jsonString.length(), chars, 1);
        return new String(chars);
    }

    /**
     * @throws DataFormatException if object cannot be encoded as JSON
     */
    public ByteBuffer toByteBuffer(Object obj) {
        if (obj == null) {
            return null;
        }

        if (obj instanceof String) {
            return ByteBuffer.wrap(((String)obj).getBytes(UTF8_CHARSET));
        }

        // write as special header bytes followed by JSON
        // intercept the Null token which stands in for a real null
        if (obj == NULL) {
            obj = null;
        }
        String jsonString = encodeJson(obj);
        byte[] jsonBytes = jsonString.getBytes(UTF8_CHARSET);
        byte[] result = new byte[jsonBytes.length + UTF8_HEADER_BYTES.length];
        System.arraycopy(UTF8_HEADER_BYTES, 0, result, 0, UTF8_HEADER_BYTES.length);
        System.arraycopy(jsonBytes, 0, result, UTF8_HEADER_BYTES.length, jsonBytes.length);
        return ByteBuffer.wrap(result);
    }

    /**
     * @throws DataFormatException is data in the byte buffer is incorrect and cannot be decoded
     */
    public Object fromByteBuffer(ByteBuffer byteBuffer) {
        if (byteBuffer == null) {
            return null;
        }
        byte[] bytes = new byte[byteBuffer.remaining()];
        byteBuffer.get(bytes, 0, bytes.length);
        return decodeBytes(bytes);
    }

    @SuppressWarnings("fallthrough")
    private Object decodeBytes(byte[] bytes) {
        if (bytes.length == 0) {
            return "";
        }

        // look for header char to determine if a JSON object or legacy NestedProperties
        PushbackReader reader = new PushbackReader(new InputStreamReader(new ByteArrayInputStream(bytes), UTF8_CHARSET));

        try {
            int firstChar = reader.read();
            switch (firstChar) {
                case '\uFFFF':
                    // legacy NestedProperties, obsolete and interpreted now as simply a JSON encoded Map or
                    // beginning of a list terminator

                    // if the second character is \uFFFF then this is a list terminator value and just return it
                    int secondChar = reader.read();
                    if (secondChar == '\uFFFF') {
                        return Types.LIST_TERMINATOR_VALUE;
                    }
                    if (secondChar == -1) {
                        throw new DataFormatException("Found header FFFF but no data");
                    }
                    reader.unread(secondChar);
                    // fall through and read as a JSON object

                case HEADER_CHAR:
                    try {
                        return JSON_MAPPER.readValue(reader, Object.class);
                    } catch (IOException e) {
                        throw new DataFormatException("Could not parse JSON", e);
                    }

                default:
                    // if no special header, then bytes are just a string
                    return new String(bytes, UTF8_CHARSET);
            }
        } catch (IOException ioe) {
            throw new DataFormatException("Could read data", ioe);
        }
    }

    private String encodeJson(Object value) {
        String jsonString;
        try {
            jsonString = JSON_MAPPER.writeValueAsString(value);
            return jsonString;
        } catch (IOException ioe) {
            throw new DataFormatException("Could not encode object as JSON: class=" + value.getClass().getName(), ioe);
        }
    }
}
import static com.datastax.driver.core.querybuilder.QueryBuilder.batch;
import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
import static com.datastax.driver.core.querybuilder.QueryBuilder.delete;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.gte;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.lte;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
import static com.datastax.driver.core.querybuilder.QueryBuilder.timestamp;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.lang3.Validate;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Query;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.querybuilder.Batch;
import com.datastax.driver.core.querybuilder.Delete;
import com.ebuddy.cassandra.BatchContext;
import com.ebuddy.cassandra.Path;
import com.ebuddy.cassandra.StructuredDataSupport;
import com.ebuddy.cassandra.TypeReference;
import com.ebuddy.cassandra.databind.CustomTypeResolverBuilder;
import com.ebuddy.cassandra.structure.Composer;
import com.ebuddy.cassandra.structure.Decomposer;
import com.ebuddy.cassandra.structure.DefaultPath;
import com.ebuddy.cassandra.structure.JacksonTypeReference;
import com.ebuddy.cassandra.structure.StructureConverter;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Optional;
/*
 * Copyright 2013 eBuddy B.V.
 *
 *    Licensed 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 com.ebuddy.cassandra.cql.dao;





/**
 * Implementation of StructuredDataSupport for CQL.
 *
 * To use structured data in CQL3, the following data modeling rules apply:
 * <ul>
 *     <li>There must be a designated path column and it must be the first clustering key, i.e. the next element of the
 *         primary key after the partition key.</li>
 *     <li>There must be a designated value column.</li>
 *     <li>There can only be one designated path and one designated value column per table.</li>
 *     <li>The designated path and value columns must be typed as a textual type.</li>
 * </ul>
 *
 * @author Eric Zoerner <a href="mailto:ezoerner@ebuddy.com">ezoerner@ebuddy.com</a>
 */
public class CqlStructuredDataSupport<K> implements StructuredDataSupport<K> {
    private static final String DEFAULT_VALUE_COLUMN = "value";
    private static final String DEFAULT_PATH_COLUMN = "column1";
    private static final String DEFAULT_PARTITION_KEY_COLUMN = "key";

    private static final int MAX_CODE_POINT = 0x10FFFF;

    private static final AtomicLong lastTime = new AtomicLong();

    private final Session session;
    private final String pathColumnName;
    private final String valueColumnName;
    private final ObjectMapper writeMapper;
    private final ObjectMapper readMapper;

    private final PreparedStatement readPathQuery;
    private final PreparedStatement readForDeleteQuery;

    private final String tableName;
    private final String partitionKeyColumnName;

    /** The default consistency level for all operations. */
    private final ConsistencyLevel defaultConsistencyLevel;

    /**
     * Used for tables that are upgraded from a thrift dynamic column family that still have the default column names.
     * @param session a Session configured with the keyspace
     */
    public CqlStructuredDataSupport(String tableName, ConsistencyLevel defaultConsistencyLevel, Session session) {
        this(tableName,
             DEFAULT_PARTITION_KEY_COLUMN,
             DEFAULT_PATH_COLUMN,
             DEFAULT_VALUE_COLUMN,
             defaultConsistencyLevel,
             session);
    }

    /**
     * Construct an instance of CqlStructuredDataSupport with the specified table and column names.
     * @param session a Session configured with the keyspace
     */
    public CqlStructuredDataSupport(String tableName,
                                    String partitionKeyColumnName,
                                    String pathColumnName,
                                    String valueColumnName,
                                    ConsistencyLevel defaultConsistencyLevel,
                                    Session session) {
        Validate.notEmpty(tableName);
        this.session = session;
        this.pathColumnName = pathColumnName;
        this.valueColumnName = valueColumnName;
        this.defaultConsistencyLevel = defaultConsistencyLevel;

        writeMapper = new ObjectMapper();
        writeMapper.setDefaultTyping(new CustomTypeResolverBuilder());
        readMapper = new ObjectMapper();
        this.tableName = tableName;
        this.partitionKeyColumnName = partitionKeyColumnName;

        readPathQuery = session.prepare(select(pathColumnName, valueColumnName)
                .from(tableName)
                .where(eq(partitionKeyColumnName, bindMarker()))
                    .and(gte(pathColumnName, bindMarker()))
                    .and(lte(pathColumnName, bindMarker()))
                .getQueryString());
        readPathQuery.setConsistencyLevel(defaultConsistencyLevel);

        readForDeleteQuery = session.prepare(select(pathColumnName)
                                                     .from(tableName)
                                                     .where(eq(partitionKeyColumnName, bindMarker()))
                                                        .and(gte(pathColumnName, bindMarker()))
                                                        .and(lte(pathColumnName, bindMarker()))
                                                     .getQueryString());
        readForDeleteQuery.setConsistencyLevel(defaultConsistencyLevel);
    }

    @Override
    public BatchContext beginBatch() {
        return new CqlBatchContext();
    }

    @Override
    public void applyBatch(BatchContext batchContext) {
        Batch batch = validateAndGetBatch(batchContext);
        List<Object> bindArguments = ((CqlBatchContext)batchContext).getBindArguments();
        Query query;
        if (bindArguments.isEmpty()) {
            query = new SimpleStatement(batch.getQueryString());
        } else {
            query = session.prepare(batch.getQueryString()).bind(bindArguments.toArray());
        }
        query.setConsistencyLevel(defaultConsistencyLevel);
        session.execute(query);
        ((CqlBatchContext)batchContext).reset();
    }

    @Override
    public <T> T readFromPath(K rowKey, Path path, TypeReference<T> type) {
        validateArgs(rowKey, path);

        String start = path.toString();
        // use the maximum unicode code point to terminate the range
        String finish = getFinishString(start);

        // note: prepared statements should be cached and reused by the connection pooling component....

        Object[] args = {rowKey,start,finish};
        ResultSet resultSet = session.execute(readPathQuery.bind(args));

        Map<Path,Object> pathMap = getPathMap(path, resultSet);

        if (pathMap.isEmpty()) {
            // not found
            return null;
        }

        Object structure = Composer.get().compose(pathMap);

        // convert object structure into POJO of type referred to by TypeReference
        return readMapper.convertValue(structure, new JacksonTypeReference<T>(type));
    }

    @Override
    public void writeToPath(K rowKey, Path path, Object value) {
        writeToPath(rowKey, path, value, null);
    }

    @Override
    public void writeToPath(K rowKey,
                            Path path,
                            Object structuredValue,
                            BatchContext batchContext) {
        Batch batch = validateAndGetBatch(batchContext);

        validateArgs(rowKey, path);
        Object simplifiedStructure = writeMapper.convertValue(structuredValue, Object.class);
        Map<Path,Object> pathMap = Collections.singletonMap(path, simplifiedStructure);
        Map<Path,Object> objectMap = Decomposer.get().decompose(pathMap);

        batch = batchContext == null ? batch() : batch;
        List<Object> bindArguments = batchContext == null ?
                                        new ArrayList<Object>() :
                                        ((CqlBatchContext)batchContext).getBindArguments();
        Statement insertStatement = insertInto(tableName)
                .value(partitionKeyColumnName, bindMarker())
                .value(pathColumnName, bindMarker())
                .value(valueColumnName, bindMarker())
                .using(timestamp(getCurrentMicros()));
        insertStatement.setConsistencyLevel(defaultConsistencyLevel);


        for (Map.Entry<Path,Object> entry : objectMap.entrySet()) {
            batch.add(insertStatement);

            String stringValue = StructureConverter.get().toString(entry.getValue());