/*
 * 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.operation.reference.file;

import com.google.common.collect.ImmutableMap;
import io.crate.metadata.ReferenceInfo;
import io.crate.operation.collect.files.LineCollectorExpression;
import io.crate.operation.reference.DocLevelReferenceResolver;

import java.util.Map;

public class FileLineReferenceResolver implements DocLevelReferenceResolver<LineCollectorExpression<?>> {

    public static final FileLineReferenceResolver INSTANCE = new FileLineReferenceResolver();

    private static final Map<String, LineCollectorExpression<?>> implementations =
            ImmutableMap.<String, LineCollectorExpression<?>>of(
                    SourceLineExpression.COLUMN_NAME, new SourceLineExpression());

    private FileLineReferenceResolver() {
    }

    public LineCollectorExpression<?> getImplementation(ReferenceInfo info) {
        LineCollectorExpression<?> expression = implementations.get(info.ident().columnIdent().name());
        if (expression != null) {
            return expression;
        }
        return new ColumnExtractingLineExpression(info.ident().columnIdent());
    }
}

@@ -1507,84 +1507,235 @@
     
-private static final Map%3CString, LineCollectorExpression%3C?%3E%3E implementations
+// need to create a new instance here so that each collector will have its own instance.%0A    // otherwise multiple collectors would share the same state.%0A    private static final Map%3CString, ExpressionBuilder%3E expressionBuilder
  =%0A 
@@ -1767,29 +1767,16 @@
 String, 
-LineCollector
 Expressi
@@ -1777,19 +1777,23 @@
 pression
-%3C?%3E
+Builder
 %3Eof(%0A   
@@ -1851,30 +1851,243 @@
 new 
-SourceLineExpression()
+ExpressionBuilder() %7B%0A                        @Override%0A                        public LineCollectorExpression%3C?%3E create() %7B%0A                            return new SourceLineExpression();%0A                        %7D%0A                    %7D
 );%0A%0A
@@ -2225,63 +2225,53 @@
     
-LineCollectorExpression%3C?%3E expression = implementations
+ExpressionBuilder builder = expressionBuilder
 .get
@@ -2319,26 +2319,23 @@
     if (
-expression
+builder
  != null
@@ -2357,26 +2357,32 @@
  return 
-expression
+builder.create()
 ;%0A      
@@ -2470,10 +2470,97 @@
 ;%0A    %7D%0A
+%0A%0A    interface ExpressionBuilder %7B%0A        LineCollectorExpression%3C?%3E create();%0A    %7D%0A
 %7D%0A

fix copy from shared state issue