/*
 *  Copyright 2016-2019 Netflix, Inc.
 *
 *     Licensed 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.
 *
 */
package com.netflix.hollow.tools.diff;

import com.netflix.hollow.core.index.HollowPrimaryKeyIndex;
import com.netflix.hollow.core.read.engine.PopulatedOrdinalListener;
import com.netflix.hollow.core.read.engine.object.HollowObjectTypeReadState;
import com.netflix.hollow.core.util.IntList;
import com.netflix.hollow.core.util.LongList;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Intended for use in the context of a HollowDiff.
 *
 * This class will match records of a specific type from two data states based on a user defined primary key.
 */
public class HollowDiffMatcher {

    private final List<String> matchPaths;

    private final HollowObjectTypeReadState fromTypeState;
    private final HollowObjectTypeReadState toTypeState;

    private final LongList matchedOrdinals;
    private final IntList extraInFrom;
    private final IntList extraInTo;

    private HollowPrimaryKeyIndex fromIdx;
    private HollowPrimaryKeyIndex toIdx;
    private static final Logger LOGGER = Logger.getLogger("HollowDiffMatcher.class");

    public HollowDiffMatcher(HollowObjectTypeReadState fromTypeState, HollowObjectTypeReadState toTypeState) {
        this.matchPaths = new ArrayList<>();
        this.fromTypeState = fromTypeState;
        this.toTypeState = toTypeState;
        this.matchedOrdinals = new LongList();
        this.extraInFrom = new IntList();
        this.extraInTo = new IntList();
    }

    public void addMatchPath(String path) {
        matchPaths.add(path);
    }

    public List<String> getMatchPaths() {
        return matchPaths;
    }

    public void calculateMatches() {
        if (fromTypeState==null) {
            toTypeState.getPopulatedOrdinals().stream().forEach(i -> extraInTo.add(i));
            return;
        }

        if (toTypeState==null) {
            fromTypeState.getPopulatedOrdinals().stream().forEach(i -> extraInFrom.add(i));
            return;
        }

        // No Primary Key so no matching will be done
        if (matchPaths==null || matchPaths.isEmpty()) {
            toTypeState.getPopulatedOrdinals().stream().forEach(i -> extraInTo.add(i));
            fromTypeState.getPopulatedOrdinals().stream().forEach(i -> extraInFrom.add(i));
            return;
        }

        fromIdx = new HollowPrimaryKeyIndex(fromTypeState.getStateEngine(), fromTypeState.getSchema().getName(), matchPaths.toArray(new String[matchPaths.size()]));
        toIdx = new HollowPrimaryKeyIndex(toTypeState.getStateEngine(), toTypeState.getSchema().getName(), matchPaths.toArray(new String[matchPaths.size()]));

        BitSet fromPopulatedOrdinals = fromTypeState.getListener(PopulatedOrdinalListener.class).getPopulatedOrdinals();
        BitSet fromUnmatchedOrdinals = new BitSet(fromPopulatedOrdinals.length());
        fromUnmatchedOrdinals.or(fromPopulatedOrdinals);

        BitSet toPopulatedOrdinals = toTypeState.getListener(PopulatedOrdinalListener.class).getPopulatedOrdinals();

        int candidateToMatchOrdinal = toPopulatedOrdinals.nextSetBit(0);
        while(candidateToMatchOrdinal != -1) {
            Object key[] = toIdx.getRecordKey(candidateToMatchOrdinal);
            int matchedOrdinal = -1;
            try {
                matchedOrdinal = fromIdx.getMatchingOrdinal(key);
            } catch(NullPointerException ex) {
                LOGGER.log(Level.SEVERE, "Null values found for type " + toTypeState.getSchema().getName()
                        + " with key field values " + Arrays.asList(key) + " at ordinal : " + candidateToMatchOrdinal);
                throw new RuntimeException("Error fetching matching ordinal for null type " + toTypeState.getSchema().getName()
                        + " with key field values " + Arrays.asList(key) + " at ordinal : " + candidateToMatchOrdinal
                        + "with stack trace " + ex.getCause());
            }

            if(matchedOrdinal != -1) {
                matchedOrdinals.add(((long)matchedOrdinal << 32) | candidateToMatchOrdinal);
                fromUnmatchedOrdinals.clear(matchedOrdinal);
            } else {
                extraInTo.add(candidateToMatchOrdinal);
            }

            candidateToMatchOrdinal = toPopulatedOrdinals.nextSetBit(candidateToMatchOrdinal + 1);
        }

        int unmatchedFromOrdinal = fromUnmatchedOrdinals.nextSetBit(0);
        while(unmatchedFromOrdinal != -1) {
            extraInFrom.add(unmatchedFromOrdinal);
            unmatchedFromOrdinal = fromUnmatchedOrdinals.nextSetBit(unmatchedFromOrdinal + 1);
        }
    }

    public LongList getMatchedOrdinals() {
        return matchedOrdinals;
    }

    public IntList getExtraInFrom() {
        return extraInFrom;
    }

    public IntList getExtraInTo() {
        return extraInTo;
    }

    public String getKeyDisplayString(HollowObjectTypeReadState state, int ordinal) {
        Object[] key = null;

        if(state == fromTypeState && fromIdx!=null) {
            key = fromIdx.getRecordKey(ordinal);
        } else if(state == toTypeState && toIdx!=null) {
            key = toIdx.getRecordKey(ordinal);
        }

        // Show Display similar to Hollow Explorer when there is no primary key
        if(key == null)
            return "ORDINAL:" + ordinal;

        return keyDisplayString(key);
    }

    private String keyDisplayString(Object[] key) {
        StringBuilder sb = new StringBuilder(key[0].toString());

        for(int i=1;i<key.length;i++) {
            sb.append(" ");
            sb.append(key[i].toString());
        }

        return sb.toString();
    }

}

@@ -675,16 +675,64 @@
 .diff;%0A%0A
+import com.netflix.hollow.core.HollowConstants;%0A
 import c
@@ -1124,73 +1124,8 @@
 ist;
-%0Aimport java.util.logging.Level;%0Aimport java.util.logging.Logger;
 %0A%0A/*
@@ -1698,94 +1698,8 @@
 Idx;
-%0A    private static final Logger LOGGER = Logger.getLogger(%22HollowDiffMatcher.class%22);
 %0A%0A  
@@ -3827,18 +3827,44 @@
 dinal = 
--1
+HollowConstants.ORDINAL_NONE
 ;%0A      
@@ -3992,235 +3992,8 @@
 ) %7B%0A
-                LOGGER.log(Level.SEVERE, %22Null values found for type %22 + toTypeState.getSchema().getName()%0A                        + %22 with key field values %22 + Arrays.asList(key) + %22 at ordinal : %22 + candidateToMatchOrdinal);%0A
     
@@ -4283,24 +4283,12 @@
 ce %22
- +
+,
  ex
-.getCause()
 );%0A 

Address comments