/*
 * 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.planner.node;

import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import io.crate.planner.symbol.Reference;
import io.crate.planner.symbol.Symbol;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;

import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

public class ESIndexNode extends PlanNode {

    private String index;

    private List<Reference> columns;
    private List<List<Symbol>> valuesLists;
    private int[] primaryKeyIndices;

    public ESIndexNode(String index,
                       List<Reference> columns,
                       List<List<Symbol>> valuesLists,
                       @Nullable int[] primaryKeyIndices) {
        Preconditions.checkNotNull(index, "index is null");
        this.index = index;
        this.columns = Objects.firstNonNull(columns, ImmutableList.<Reference>of());
        this.valuesLists = Objects.firstNonNull(valuesLists, ImmutableList.<List<Symbol>>of());
        this.primaryKeyIndices = Objects.firstNonNull(primaryKeyIndices, new int[0]);
    }

    public String index() {
        return index;
    }

    public List<Reference> columns() {
        return columns;
    }

    public List<List<Symbol>> valuesLists() {
        return valuesLists;
    }

    @Override
    public Set<String> executionNodes() {
        return null;
    }

    @Override
    public <C, R> R accept(PlanVisitor<C, R> visitor, C context) {
        return visitor.visitESIndexNode(this, context);
    }

    @Override
    public void readFrom(StreamInput in) throws IOException {
        super.readFrom(in);
        index = in.readString();

        int numColumns = in.readVInt();
        columns = new ArrayList<>(numColumns);
        for (int i=0; i<numColumns;i++) {
            Reference reference = new Reference();
            reference.readFrom(in);
            columns.add(reference);
        }

        int numValuesLists = in.readVInt();
        valuesLists = new ArrayList<>(numValuesLists);
        for (int i=0; i<numValuesLists;i++) {
            int numValues = in.readVInt();
            List<Symbol> values = new ArrayList<>(numValues);
            for (int j=0; j<numValues;j++) {
                Symbol symbol = Symbol.fromStream(in);
                values.add(symbol);
            }
            valuesLists.add(values);
        }
        int numIndices = in.readVInt();

        primaryKeyIndices = new int[numIndices];
        for (int i = 0; i< numIndices; i++) {
            primaryKeyIndices[i] = in.readVInt();
        }
    }

    @Override
    public void writeTo(StreamOutput out) throws IOException {
        super.writeTo(out);
        out.writeString(index);

        out.writeVInt(columns.size());
        for (Reference reference : columns) {
            reference.writeTo(out);
        }

        out.writeVInt(valuesLists.size());
        for (List<Symbol> values : valuesLists) {
            out.writeVInt(values.size());
            for (Symbol symbol : values) {
                Symbol.toStream(symbol, out);
            }
        }
        out.writeVInt(primaryKeyIndices.length);
        if (primaryKeyIndices.length > 0) {
            for (int i : primaryKeyIndices) {
                out.writeVInt(i);
            }
        }
    }

    public int[] primaryKeyIndices() {
        return primaryKeyIndices;
    }

    public boolean hasPrimaryKey() {
        return primaryKeyIndices.length > 0;
    }
}

@@ -1437,36 +1437,8 @@
 on;%0A
-import java.util.ArrayList;%0A
 impo
@@ -2756,1689 +2756,269 @@
     
-super.readFrom(in);%0A        index = in.readString();%0A%0A        int numColumns = in.readVInt();%0A        columns = new ArrayList%3C%3E(numColumns);%0A        for (int i=0; i%3CnumColumns;i++) %7B%0A            Reference reference = new Reference();%0A            reference.readFrom(in);%0A            columns.add(reference);%0A        %7D%0A%0A        int numValuesLists = in.readVInt();%0A        valuesLists = new ArrayList%3C%3E(numValuesLists);%0A        for (int i=0; i%3CnumValuesLists;i++) %7B%0A            int numValues = in.readVInt();%0A            List%3CSymbol%3E values = new ArrayList%3C%3E(numValues);%0A            for (int j=0; j%3CnumValues;j++) %7B%0A                Symbol symbol = Symbol.fromStream(in);%0A                values.add(symbol);%0A            %7D%0A            valuesLists.add(values);%0A        %7D%0A        int numIndices = in.readVInt();%0A%0A        primaryKeyIndices = new int%5BnumIndices%5D;%0A        for (int i = 0; i%3C numIndices; i++) %7B%0A            primaryKeyIndices%5Bi%5D = in.readVInt();%0A        %7D%0A    %7D%0A%0A    @Override%0A    public void writeTo(StreamOutput out) throws IOException %7B%0A        super.writeTo(out);%0A        out.writeString(index);%0A%0A        out.writeVInt(columns.size());%0A        for (Reference reference : columns) %7B%0A            reference.writeTo(out);%0A        %7D%0A%0A        out.writeVInt(valuesLists.size());%0A        for (List%3CSymbol%3E values : valuesLists) %7B%0A            out.writeVInt(values.size());%0A            for (Symbol symbol : values) %7B%0A                Symbol.toStream(symbol, out);%0A            %7D%0A        %7D%0A        out.writeVInt(primaryKeyIndices.length);%0A        if (primaryKeyIndices.length %3E 0) %7B%0A            for (int i : primaryKeyIndices) %7B%0A                out.writeVInt(i);%0A            %7D%0A        %7D
+throw new UnsupportedOperationException(%22ESIndexNode has no serialization support%22);%0A    %7D%0A%0A    @Override%0A    public void writeTo(StreamOutput out) throws IOException %7B%0A        throw new UnsupportedOperationException(%22ESIndexNode has no serialization support%22);
 %0A   

index nodes should not be serializable