/* Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF 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.
 */

#define C_LUCY_DELETIONSWRITER
#define C_LUCY_DEFAULTDELETIONSWRITER
#include "Lucy/Util/ToolSet.h"

#include <math.h>

#include "Lucy/Index/DeletionsWriter.h"
#include "Lucy/Index/DeletionsReader.h"
#include "Lucy/Index/IndexReader.h"
#include "Lucy/Index/PolyReader.h"
#include "Lucy/Index/PostingList.h"
#include "Lucy/Index/PostingListReader.h"
#include "Lucy/Index/Segment.h"
#include "Lucy/Index/SegReader.h"
#include "Lucy/Index/Snapshot.h"
#include "Lucy/Plan/Schema.h"
#include "Lucy/Search/BitVecMatcher.h"
#include "Lucy/Search/Compiler.h"
#include "Lucy/Search/IndexSearcher.h"
#include "Lucy/Search/Matcher.h"
#include "Lucy/Search/Query.h"
#include "Lucy/Store/Folder.h"
#include "Lucy/Store/OutStream.h"

DeletionsWriter*
DelWriter_init(DeletionsWriter *self, Schema *schema, Snapshot *snapshot,
               Segment *segment, PolyReader *polyreader) {
    DataWriter_init((DataWriter*)self, schema, snapshot, segment, polyreader);
    ABSTRACT_CLASS_CHECK(self, DELETIONSWRITER);
    return self;
}

I32Array*
DelWriter_generate_doc_map(DeletionsWriter *self, Matcher *deletions,
                           int32_t doc_max, int32_t offset) {
    int32_t *doc_map = (int32_t*)CALLOCATE(doc_max + 1, sizeof(int32_t));
    int32_t  next_deletion = deletions ? Matcher_Next(deletions) : I32_MAX;
    UNUSED_VAR(self);

    // 0 for a deleted doc, a new number otherwise
    for (int32_t i = 1, new_doc_id = 1; i <= doc_max; i++) {
        if (i == next_deletion) {
            next_deletion = Matcher_Next(deletions);
        }
        else {
            doc_map[i] = offset + new_doc_id++;
        }
    }

    return I32Arr_new_steal(doc_map, doc_max + 1);
}

int32_t DefDelWriter_current_file_format = 1;

DefaultDeletionsWriter*
DefDelWriter_new(Schema *schema, Snapshot *snapshot, Segment *segment,
                 PolyReader *polyreader) {
    DefaultDeletionsWriter *self
        = (DefaultDeletionsWriter*)VTable_Make_Obj(DEFAULTDELETIONSWRITER);
    return DefDelWriter_init(self, schema, snapshot, segment, polyreader);
}

DefaultDeletionsWriter*
DefDelWriter_init(DefaultDeletionsWriter *self, Schema *schema,
                  Snapshot *snapshot, Segment *segment,
                  PolyReader *polyreader) {

    DataWriter_init((DataWriter*)self, schema, snapshot, segment, polyreader);
    self->seg_readers           = PolyReader_Seg_Readers(polyreader);
    uint32_t num_seg_readers    = VA_Get_Size(self->seg_readers);
    self->seg_starts            = PolyReader_Offsets(polyreader);
    self->bit_vecs              = VA_new(num_seg_readers);
    self->updated               = (bool_t*)CALLOCATE(num_seg_readers, sizeof(bool_t));
    self->searcher              = IxSearcher_new((Obj*)polyreader);
    self->name_to_tick          = Hash_new(num_seg_readers);

    // Materialize a BitVector of deletions for each segment.
    for (uint32_t i = 0; i < num_seg_readers; i++) {
        SegReader *seg_reader = (SegReader*)VA_Fetch(self->seg_readers, i);
        BitVector *bit_vec    = BitVec_new(SegReader_Doc_Max(seg_reader));
        DeletionsReader *del_reader
            = (DeletionsReader*)SegReader_Fetch(
                  seg_reader, VTable_Get_Name(DELETIONSREADER));
        Matcher *seg_dels = del_reader
                            ? DelReader_Iterator(del_reader)
                            : NULL;

        if (seg_dels) {
            int32_t del;
            while (0 != (del = Matcher_Next(seg_dels))) {
                BitVec_Set(bit_vec, del);
            }
            DECREF(seg_dels);
        }
        VA_Store(self->bit_vecs, i, (Obj*)bit_vec);
        Hash_Store(self->name_to_tick,
                   (Obj*)SegReader_Get_Seg_Name(seg_reader),
                   (Obj*)Int32_new(i));
    }

    return self;
}

void
DefDelWriter_destroy(DefaultDeletionsWriter *self) {
    DECREF(self->seg_readers);
    DECREF(self->seg_starts);
    DECREF(self->bit_vecs);
    DECREF(self->searcher);
    DECREF(self->name_to_tick);
    FREEMEM(self->updated);
    SUPER_DESTROY(self, DEFAULTDELETIONSWRITER);
}

static CharBuf*
S_del_filename(DefaultDeletionsWriter *self, SegReader *target_reader) {
    Segment *target_seg = SegReader_Get_Segment(target_reader);
    return CB_newf("%o/deletions-%o.bv", Seg_Get_Name(self->segment),
                   Seg_Get_Name(target_seg));
}

void
DefDelWriter_finish(DefaultDeletionsWriter *self) {
    Folder *const folder = self->folder;

    for (uint32_t i = 0, max = VA_Get_Size(self->seg_readers); i < max; i++) {
        SegReader *seg_reader = (SegReader*)VA_Fetch(self->seg_readers, i);
        if (self->updated[i]) {
            BitVector *deldocs   = (BitVector*)VA_Fetch(self->bit_vecs, i);
            int32_t    doc_max   = SegReader_Doc_Max(seg_reader);
            double     used      = (doc_max + 1) / 8.0;
            uint32_t   byte_size = (uint32_t)ceil(used);
            uint32_t   new_max   = byte_size * 8 - 1;
            CharBuf   *filename  = S_del_filename(self, seg_reader);
            OutStream *outstream = Folder_Open_Out(folder, filename);
            if (!outstream) { RETHROW(INCREF(Err_get_error())); }

            // Ensure that we have 1 bit for each doc in segment.
            BitVec_Grow(deldocs, new_max);

            // Write deletions data and clean up.
            OutStream_Write_Bytes(outstream,
                                  (char*)BitVec_Get_Raw_Bits(deldocs),
                                  byte_size);
            OutStream_Close(outstream);
            DECREF(outstream);
            DECREF(filename);
        }
    }

    Seg_Store_Metadata_Str(self->segment, "deletions", 9,
                           (Obj*)DefDelWriter_Metadata(self));
}

Hash*
DefDelWriter_metadata(DefaultDeletionsWriter *self) {
    Hash    *const metadata = DataWriter_metadata((DataWriter*)self);
    Hash    *const files    = Hash_new(0);

    for (uint32_t i = 0, max = VA_Get_Size(self->seg_readers); i < max; i++) {
        SegReader *seg_reader = (SegReader*)VA_Fetch(self->seg_readers, i);
        if (self->updated[i]) {
            BitVector *deldocs   = (BitVector*)VA_Fetch(self->bit_vecs, i);
            Segment   *segment   = SegReader_Get_Segment(seg_reader);
            Hash      *mini_meta = Hash_new(2);
            Hash_Store_Str(mini_meta, "count", 5,
                           (Obj*)CB_newf("%u32", (uint32_t)BitVec_Count(deldocs)));
            Hash_Store_Str(mini_meta, "filename", 8,
                           (Obj*)S_del_filename(self, seg_reader));
            Hash_Store(files, (Obj*)Seg_Get_Name(segment), (Obj*)mini_meta);
        }
    }
    Hash_Store_Str(metadata, "files", 5, (Obj*)files);

    return metadata;
}

int32_t
DefDelWriter_format(DefaultDeletionsWriter *self) {
    UNUSED_VAR(self);
    return DefDelWriter_current_file_format;
}

Matcher*
DefDelWriter_seg_deletions(DefaultDeletionsWriter *self,
                           SegReader *seg_reader) {
    Matcher *deletions    = NULL;
    Segment *segment      = SegReader_Get_Segment(seg_reader);
    CharBuf *seg_name     = Seg_Get_Name(segment);
    Integer32 *tick_obj   = (Integer32*)Hash_Fetch(self->name_to_tick,
                                                   (Obj*)seg_name);
    int32_t tick          = tick_obj ? Int32_Get_Value(tick_obj) : 0;
    SegReader *candidate  = tick_obj
                            ? (SegReader*)VA_Fetch(self->seg_readers, tick)
                            : NULL;

    if (tick_obj) {
        DeletionsReader *del_reader
            = (DeletionsReader*)SegReader_Obtain(
                  candidate, VTable_Get_Name(DELETIONSREADER));
        if (self->updated[tick] || DelReader_Del_Count(del_reader)) {
            BitVector *deldocs = (BitVector*)VA_Fetch(self->bit_vecs, tick);
            deletions = (Matcher*)BitVecMatcher_new(deldocs);
        }
    }
    else { // Sanity check.
        THROW(ERR, "Couldn't find SegReader %o", seg_reader);
    }

    return deletions;
}

int32_t
DefDelWriter_seg_del_count(DefaultDeletionsWriter *self,
                           const CharBuf *seg_name) {
    Integer32 *tick
        = (Integer32*)Hash_Fetch(self->name_to_tick, (Obj*)seg_name);
    BitVector *deldocs = tick
                         ? (BitVector*)VA_Fetch(self->bit_vecs, Int32_Get_Value(tick))
                         : NULL;
    return deldocs ? BitVec_Count(deldocs) : 0;
}

void
DefDelWriter_delete_by_term(DefaultDeletionsWriter *self,
                            const CharBuf *field, Obj *term) {
    for (uint32_t i = 0, max = VA_Get_Size(self->seg_readers); i < max; i++) {
        SegReader *seg_reader = (SegReader*)VA_Fetch(self->seg_readers, i);
        PostingListReader *plist_reader
            = (PostingListReader*)SegReader_Fetch(
                  seg_reader, VTable_Get_Name(POSTINGLISTREADER));
        BitVector *bit_vec = (BitVector*)VA_Fetch(self->bit_vecs, i);
        PostingList *plist = plist_reader
                             ? PListReader_Posting_List(plist_reader, field, term)
                             : NULL;
        int32_t doc_id;
        int32_t num_zapped = 0;

        // Iterate through postings, marking each doc as deleted.
        if (plist) {
            while (0 != (doc_id = PList_Next(plist))) {
                num_zapped += !BitVec_Get(bit_vec, doc_id);
                BitVec_Set(bit_vec, doc_id);
            }
            if (num_zapped) { self->updated[i] = true; }
            DECREF(plist);
        }
    }
}

void
DefDelWriter_delete_by_query(DefaultDeletionsWriter *self, Query *query) {
    Compiler *compiler = Query_Make_Compiler(query, (Searcher*)self->searcher,
                                             Query_Get_Boost(query), false);

    for (uint32_t i = 0, max = VA_Get_Size(self->seg_readers); i < max; i++) {
        SegReader *seg_reader = (SegReader*)VA_Fetch(self->seg_readers, i);
        BitVector *bit_vec = (BitVector*)VA_Fetch(self->bit_vecs, i);
        Matcher *matcher = Compiler_Make_Matcher(compiler, seg_reader, false);

        if (matcher) {
            int32_t doc_id;
            int32_t num_zapped = 0;

            // Iterate through matches, marking each doc as deleted.
            while (0 != (doc_id = Matcher_Next(matcher))) {
                num_zapped += !BitVec_Get(bit_vec, doc_id);
                BitVec_Set(bit_vec, doc_id);
            }
            if (num_zapped) { self->updated[i] = true; }

            DECREF(matcher);
        }
    }

    DECREF(compiler);
}

void
DefDelWriter_delete_by_doc_id(DefaultDeletionsWriter *self, int32_t doc_id) {
    uint32_t   sub_tick   = PolyReader_sub_tick(self->seg_starts, doc_id);
    BitVector *bit_vec    = (BitVector*)VA_Fetch(self->bit_vecs, sub_tick);
    uint32_t   offset     = I32Arr_Get(self->seg_starts, sub_tick);
    int32_t    seg_doc_id = doc_id - offset;

    if (!BitVec_Get(bit_vec, seg_doc_id)) {
        self->updated[sub_tick] = true;
        BitVec_Set(bit_vec, seg_doc_id);
    }
}

bool_t
DefDelWriter_updated(DefaultDeletionsWriter *self) {
    for (uint32_t i = 0, max = VA_Get_Size(self->seg_readers); i < max; i++) {
        if (self->updated[i]) { return true; }
    }
    return false;
}

void
DefDelWriter_add_segment(DefaultDeletionsWriter *self, SegReader *reader,
                         I32Array *doc_map) {
    // This method is a no-op, because the only reason it would be called is
    // if we are adding an entire index.  If that's the case, all deletes are
    // already being applied.
    UNUSED_VAR(self);
    UNUSED_VAR(reader);
    UNUSED_VAR(doc_map);
}

void
DefDelWriter_merge_segment(DefaultDeletionsWriter *self, SegReader *reader,
                           I32Array *doc_map) {
    UNUSED_VAR(doc_map);
    Segment *segment = SegReader_Get_Segment(reader);
    Hash *del_meta = (Hash*)Seg_Fetch_Metadata_Str(segment, "deletions", 9);

    if (del_meta) {
        VArray *seg_readers = self->seg_readers;
        Hash   *files = (Hash*)Hash_Fetch_Str(del_meta, "files", 5);
        if (files) {
            CharBuf *seg;
            Hash *mini_meta;
            Hash_Iterate(files);
            while (Hash_Next(files, (Obj**)&seg, (Obj**)&mini_meta)) {

                /* Find the segment the deletions from the SegReader
                 * we're adding correspond to.  If it's gone, we don't
                 * need to worry about losing deletions files that point
                 * at it. */
                for (uint32_t i = 0, max = VA_Get_Size(seg_readers); i < max; i++) {
                    SegReader *candidate
                        = (SegReader*)VA_Fetch(seg_readers, i);
                    CharBuf *candidate_name
                        = Seg_Get_Name(SegReader_Get_Segment(candidate));

                    if (CB_Equals(seg, (Obj*)candidate_name)) {
                        /* If the count hasn't changed, we're about to
                         * merge away the most recent deletions file
                         * pointing at this target segment -- so force a
                         * new file to be written out. */
                        int32_t count = (int32_t)Obj_To_I64(Hash_Fetch_Str(mini_meta, "count", 5));
                        DeletionsReader *del_reader
                            = (DeletionsReader*)SegReader_Obtain(
                                  candidate, VTable_Get_Name(DELETIONSREADER));
                        if (count == DelReader_Del_Count(del_reader)) {
                            self->updated[i] = true;
                        }
                        break;
                    }
                }
            }
        }
    }
}