/*
 * Licensed to 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;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import io.crate.analyze.WhereClause;
import io.crate.metadata.Routing;
import io.crate.metadata.TableIdent;
import io.crate.metadata.table.TableInfo;
import io.crate.planner.fetch.IndexBaseBuilder;

import javax.annotation.Nullable;
import java.util.*;

final class RoutingBuilder {

    final Map<TableIdent, List<TableRouting>> routingListByTable = new HashMap<>();

    //index, shardId, node
    private Map<String, Map<Integer, String>> shardNodes;
    private ReaderAllocations readerAllocations;
    private HashMultimap<TableIdent, String> tableIndices;

    @VisibleForTesting
    final static class TableRouting {
        final WhereClause where;
        final String preference;
        final Routing routing;
        boolean nodesAllocated = false;

        TableRouting(WhereClause where, String preference, Routing routing) {
            this.where = where;
            this.preference = preference;
            this.routing = routing;
        }
    }

    Routing allocateRouting(TableInfo tableInfo, WhereClause where, @Nullable String preference) {
        List<TableRouting> existingRoutings = routingListByTable.get(tableInfo.ident());
        if (existingRoutings == null) {
            return allocateNewRouting(tableInfo, where, preference);
        }
        Routing existing = tryFindMatchInExisting(where, preference, existingRoutings);
        if (existing != null) return existing;

        Routing routing = tableInfo.getRouting(where, preference);
        existingRoutings.add(new TableRouting(where, preference, routing));
        // ensure all routings of this table are allocated
        // and update new routing by merging with existing ones
        for (TableRouting existingRouting : existingRoutings) {
            allocateIfRequired(tableInfo.ident(), existingRouting);
            // Merge locations with existing routing
            routing.mergeLocations(existingRouting.routing.locations());
        }
        if (!allocateRoutingNodes(tableInfo.ident(), routing.locations())) {
            throw new UnsupportedOperationException(
                "Nodes of existing routing are not allocated, routing rebuild needed");
        }
        return routing;
    }

    private static Routing tryFindMatchInExisting(WhereClause where,
                                                  @Nullable String preference,
                                                  Iterable<TableRouting> existingRoutings) {
        for (TableRouting existing : existingRoutings) {
            assert preference == null || preference.equals(existing.preference) :
                "preference must not be null or equals existing preference";
            if (Objects.equals(existing.where, where)) {
                return existing.routing;
            }
        }
        return null;
    }

    private Routing allocateNewRouting(TableInfo tableInfo, WhereClause where, @Nullable String preference) {
        List<TableRouting> existingRoutings = new ArrayList<>();
        routingListByTable.put(tableInfo.ident(), existingRoutings);
        Routing routing = tableInfo.getRouting(where, preference);
        existingRoutings.add(new TableRouting(where, preference, routing));
        return routing;
    }

    ReaderAllocations buildReaderAllocations() {
        if (readerAllocations != null) {
            return readerAllocations;
        }

        Multimap<TableIdent, String> indicesByTable = HashMultimap.create();
        IndexBaseBuilder indexBaseBuilder = new IndexBaseBuilder();

        for (final Map.Entry<TableIdent, List<TableRouting>> tableRoutingEntry : routingListByTable.entrySet()) {
            TableIdent table = tableRoutingEntry.getKey();
            List<TableRouting> routingList = tableRoutingEntry.getValue();
            for (TableRouting tr : routingList) {
                allocateIfRequired(table, tr);

                for (Map.Entry<String, Map<String, List<Integer>>> entry : tr.routing.locations().entrySet()) {
                    Map<String, List<Integer>> shardsByIndex = entry.getValue();
                    indicesByTable.putAll(table, shardsByIndex.keySet());

                    for (Map.Entry<String, List<Integer>> shardsByIndexEntry : shardsByIndex.entrySet()) {
                        indexBaseBuilder.allocate(shardsByIndexEntry.getKey(), shardsByIndexEntry.getValue());
                    }
                }
            }
        }
        readerAllocations = new ReaderAllocations(indexBaseBuilder.build(), shardNodes, indicesByTable);
        return readerAllocations;
    }

    private void allocateIfRequired(TableIdent table, TableRouting tr) {
        if (tr.nodesAllocated) {
            return;
        }
        tr.nodesAllocated = true;
        allocateRoutingNodes(table, tr.routing.locations());
    }

    private boolean allocateRoutingNodes(TableIdent tableIdent, Map<String, Map<String, List<Integer>>> locations) {
        boolean success = true;
        if (tableIndices == null) {
            tableIndices = HashMultimap.create();
        }
        if (shardNodes == null) {
            shardNodes = new HashMap<>();
        }
        for (Map.Entry<String, Map<String, List<Integer>>> indicesByNodeId : locations.entrySet()) {
            String nodeId = indicesByNodeId.getKey();
            for (Map.Entry<String, List<Integer>> shardsByIndexEntry : indicesByNodeId.getValue().entrySet()) {
                String index = shardsByIndexEntry.getKey();
                Map<Integer, String> shardsOnIndex = shardNodes.get(index);
                tableIndices.put(tableIdent, index);
                List<Integer> shards = shardsByIndexEntry.getValue();
                if (shardsOnIndex == null) {
                    shardsOnIndex = new HashMap<>(shards.size());
                    shardNodes.put(index, shardsOnIndex);
                    for (Integer id : shards) {
                        shardsOnIndex.put(id, nodeId);
                    }
                } else {
                    for (Integer id : shards) {
                        String allocatedNodeId = shardsOnIndex.get(id);
                        if (allocatedNodeId != null) {
                            if (!allocatedNodeId.equals(nodeId)) {
                                success = false;
                            }
                        } else {
                            shardsOnIndex.put(id, nodeId);
                        }
                    }
                }
            }
        }
        return success;
    }
}

@@ -1671,67 +1671,8 @@
 ons;
-%0A    private HashMultimap%3CTableIdent, String%3E tableIndices;
 %0A%0A  
@@ -2877,35 +2877,16 @@
 equired(
-tableInfo.ident(), 
 existing
@@ -3065,35 +3065,16 @@
 ngNodes(
-tableInfo.ident(), 
 routing.
@@ -4914,23 +4914,16 @@
 equired(
-table, 
 tr);%0A%0A  
@@ -5656,26 +5656,8 @@
 red(
-TableIdent table, 
 Tabl
@@ -5797,23 +5797,16 @@
 ngNodes(
-table, 
 tr.routi
@@ -5874,31 +5874,8 @@
 des(
-TableIdent tableIdent, 
 Map%3C
@@ -5959,104 +5959,8 @@
 ue;%0A
-        if (tableIndices == null) %7B%0A            tableIndices = HashMultimap.create();%0A        %7D%0A
     
@@ -6448,61 +6448,8 @@
 x);%0A
-                tableIndices.put(tableIdent, index);%0A
     

Remove unused tableIndices