/*
 * Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
 * license agreements.  See the NOTICE file distributed with this work for
 * additional information regarding copyright ownership.  Crate 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.
 *
 * However, if you have executed another commercial license agreement
 * with Crate these terms will supersede the license and you may use the
 * software solely pursuant to the terms of the relevant commercial agreement.
 */

package io.crate.expression.reference.doc.lucene;

import com.google.common.base.Joiner;
import io.crate.execution.engine.collect.collectors.CollectorFieldsVisitor;
import io.crate.metadata.Reference;
import io.crate.metadata.doc.DocSysColumns;
import io.crate.types.DataType;
import org.apache.lucene.index.LeafReaderContext;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.lookup.SourceLookup;

import java.util.Map;

public class DocCollectorExpression extends LuceneCollectorExpression<Map<String, Object>> {

    public static final String COLUMN_NAME = DocSysColumns.DOC.name();

    private CollectorFieldsVisitor visitor;

    public DocCollectorExpression() {
        super();
    }

    @Override
    public void startCollect(CollectorContext context) {
        context.visitor().required(true);
        this.visitor = context.visitor();
    }

    @Override
    public Map<String, Object> value() {
        return XContentHelper.convertToMap(visitor.source(), false, XContentType.JSON).v2();
    }

    public static LuceneCollectorExpression<?> create(final Reference reference) {
        assert reference.column().name().equals(DocSysColumns.DOC.name()) :
            "column name must be " + DocSysColumns.DOC.name();
        if (reference.column().path().size() == 0) {
            return new DocCollectorExpression();
        }

        assert reference.column().path().size() > 0 : "column's path size must be > 0";
        final String fqn = Joiner.on(".").join(reference.column().path());
        return new ChildDocCollectorExpression(reference.valueType(), fqn);
    }

    static final class ChildDocCollectorExpression extends LuceneCollectorExpression<Object> {

        private final DataType returnType;
        private final String columnName;
        SourceLookup sourceLookup;
        private LeafReaderContext context;

        ChildDocCollectorExpression(DataType returnType, String columnName) {
            this.returnType = returnType;
            this.columnName = columnName;
        }

        @Override
        public void setNextDocId(int doc) {
            sourceLookup.setSegmentAndDocument(context, doc);
        }

        @Override
        public void setNextReader(LeafReaderContext context) {
            this.context = context;
        }

        @Override
        public void startCollect(CollectorContext context) {
            sourceLookup = context.sourceLookup();
        }

        @Override
        public Object value() {
            // need to make sure it has the correct type;
            // for example:
            //      sourceExtractor might read byte as int and
            //      then eq(byte, byte) would get eq(byte, int) and fail
            return returnType.value(sourceLookup.extractValue(columnName));
        }
    }
}

@@ -1071,46 +1071,8 @@
 e;%0A%0A
-import com.google.common.base.Joiner;%0A
 impo
@@ -1488,19 +1488,101 @@
 va.util.
-Map
+ArrayList;%0Aimport java.util.List;%0Aimport java.util.Map;%0Aimport java.util.RandomAccess
 ;%0A%0Apubli
@@ -2503,172 +2503,8 @@
    %7D
-%0A%0A        assert reference.column().path().size() %3E 0 : %22column's path size must be %3E 0%22;%0A        final String fqn = Joiner.on(%22.%22).join(reference.column().path());
 %0A   
@@ -2574,11 +2574,33 @@
 (), 
-fqn
+reference.column().path()
 );%0A 
@@ -2766,33 +2766,33 @@
 e final 
+List%3C
 String
- columnName
+%3E path
 ;%0A      
@@ -2793,16 +2793,24 @@
         
+private 
 SourceLo
@@ -2933,25 +2933,25 @@
 pe, 
+List%3C
 String
- columnName
+%3E path
 ) %7B%0A
@@ -3013,31 +3013,19 @@
 his.
-columnName = columnName
+path = path
 ;%0A  
@@ -3749,46 +3749,1149 @@
 lue(
-sourceLookup.extractValue(columnName))
+extractValue(sourceLookup, 0, path));%0A        %7D%0A%0A        @SuppressWarnings(%22unchecked%22)%0A        private static Object extractValue(final Map map, int index, List%3CString%3E path) %7B%0A            assert path instanceof RandomAccess : %22path should support RandomAccess for fast index optimized loop%22;%0A            Map m = map;%0A            Object tmp = null;%0A            for (int i = index; i %3C path.size(); i++) %7B%0A                tmp = m.get(path.get(i));%0A                if (tmp instanceof Map) %7B%0A                    m = (Map) tmp;%0A                %7D else if (tmp instanceof List) %7B%0A                    List list = (List) tmp;%0A                    List newList = new ArrayList(list.size());%0A                    for (Object o : list) %7B%0A                        if (o instanceof Map && i + 1 %3C path.size()) %7B%0A                            newList.add(extractValue((Map) o, i + 1, path));%0A                        %7D else %7B%0A                            newList.add(o);%0A                        %7D%0A                    %7D%0A                    return newList;%0A                %7D else %7B%0A                    break;%0A                %7D%0A            %7D%0A            return tmp
 ;%0A  

Optimize value lookup in ChildDocCollectorExpression