Support nested fields in Elasticsearch Connector

[zhenxiao]

No description provided.

[martint]

Since we know the type of the column, it might be better to drive the "un-flattening" based on the fields that are actually needed. I'll think about how we might be able to do this and comment later.

[martint]

Use Integer.compare(). Comparing using subtraction can result in subtle issues due to overflow (unlikely to be an issue here, but let's use the more robust way).

[martint]

I'd just call the argument fields

[martint]

This can only happen if there are no fields with . left, but it's not immediately obvious. I have the feeling we should be able to structure it in such a way that it's easier to read, but at a minimum, it deserves a comment.

[martint]

@zhenxiao, here's my suggestion for an alternate implementation of the unflatten logic:

    private void extractFromSource(SearchHit hit)
    {
        List<Field> fields = new ArrayList<>();
        for (Map.Entry<String, Object> entry : hit.getSourceAsMap().entrySet()) {
            fields.add(new Field(entry.getKey(), entry.getValue()));
        }
        Collections.sort(fields, Comparator.comparing(Field::getName));

        for (Map.Entry<String, Object> entry : unflatten(fields).entrySet()) {
            setFieldIfExists(entry.getKey(), entry.getValue());
        }
    }

    private static Map<String, Object> unflatten(List<Field> fields)
    {
        return unflatten(fields, 0, 0, fields.size());
    }

    private static Map<String, Object> unflatten(List<Field> fields, int level, int start, int length)
    {
        checkArgument(length > 0, "length must be > 0");

        int limit = start + length;

        Map<String, Object> result = new HashMap<>();
        int anchor = start;
        int current = start;

        do {
            Field field = fields.get(anchor);
            String name = field.getPathElement(level);

            current++;
            if (current == limit || !name.equals(fields.get(current).getPathElement(level))) {
                // We assume that fields can't be both leaves and intermediate nodes
                Object value;
                if (level < field.getDepth() - 1) {
                    value = unflatten(fields, level + 1, anchor, current - anchor);
                }
                else {
                    value = field.getValue();
                }
                result.put(name, value);
                anchor = current;
            }
        }
        while (current < limit);

        return result;
    }

    private final static class Field
    {
        private final String name;
        private final List<String> path;
        private final Object value;

        public Field(String name, Object value)
        {
            this.name = name;
            this.path = Arrays.asList(name.split("\\."));
            this.value = value;
        }

        public String getName()
        {
            return name;
        }

        public int getDepth()
        {
            return path.size();
        }

        public String getPathElement(int level)
        {
            return path.get(level);
        }

        public Object getValue()
        {
            return value;
        }
    }

[zhenxiao]

thank you, @martint
get it updated. could you please take a look?

[martint]

Looks good. I'll merge it once travis passes