<commit_before>package com.jivesoftware.os.miru.service.index.filer;

import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.io.ByteArrayDataOutput;
import com.google.common.io.ByteStreams;
import com.jivesoftware.os.filer.io.Filer;
import com.jivesoftware.os.filer.io.FilerDataInput;
import com.jivesoftware.os.filer.io.FilerIO;
import com.jivesoftware.os.filer.io.api.ChunkTransaction;
import com.jivesoftware.os.filer.io.api.KeyedFilerStore;
import com.jivesoftware.os.filer.io.api.StackBuffer;
import com.jivesoftware.os.filer.io.chunk.ChunkFiler;
import com.jivesoftware.os.miru.plugin.bitmap.MiruBitmaps;
import com.jivesoftware.os.miru.plugin.index.MiruInvertedIndex;
import com.jivesoftware.os.miru.service.index.BitmapAndLastId;
import com.jivesoftware.os.miru.service.partition.TrackError;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.io.DataInput;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * @author jonathan
 */
public class MiruFilerInvertedIndex<BM extends IBM, IBM> implements MiruInvertedIndex<IBM> {

    private static final MetricLogger log = MetricLoggerFactory.getLogger();

    private static final int LAST_ID_LENGTH = 4;

    private final MiruBitmaps<BM, IBM> bitmaps;
    private final TrackError trackError;
    private final byte[] indexKeyBytes;
    private final KeyedFilerStore<Long, Void> keyedFilerStore;
    private final int considerIfIndexIdGreaterThanN;
    private final Object mutationLock;
    private volatile int lastId = Integer.MIN_VALUE;

    private final ChunkTransaction<Void, BitmapAndLastId<BM>> getTransaction;

    public MiruFilerInvertedIndex(MiruBitmaps<BM, IBM> bitmaps,
        TrackError trackError,
        byte[] indexKeyBytes,
        KeyedFilerStore<Long, Void> keyedFilerStore,
        int considerIfIndexIdGreaterThanN,
        Object mutationLock) {
        this.bitmaps = bitmaps;
        this.trackError = trackError;
        this.indexKeyBytes = Preconditions.checkNotNull(indexKeyBytes);
        this.keyedFilerStore = Preconditions.checkNotNull(keyedFilerStore);
        this.considerIfIndexIdGreaterThanN = considerIfIndexIdGreaterThanN;
        this.mutationLock = mutationLock;
        //TODO pass in
        this.getTransaction = (monkey, filer, stackBuffer, lock) -> {
            if (filer != null) {
                synchronized (lock) {
                    filer.seek(0);
                    return deser(bitmaps, trackError, filer, stackBuffer);
                }
            }
            return null;
        };
    }

    @Override
    public Optional<IBM> getIndex(StackBuffer stackBuffer) throws Exception {
        if (lastId > Integer.MIN_VALUE && lastId <= considerIfIndexIdGreaterThanN) {
            return Optional.absent();
        }

        BitmapAndLastId<BM> bitmapAndLastId = keyedFilerStore.read(indexKeyBytes, null, getTransaction, stackBuffer);

        if (bitmapAndLastId != null) {
            log.inc("get>hit");
            if (lastId == Integer.MIN_VALUE) {
                lastId = bitmapAndLastId.lastId;
            }
            return Optional.of(bitmapAndLastId.bitmap);
        } else {
            log.inc("get>miss");
            lastId = -1;
            return Optional.absent();
        }
    }

    @Override
    public <R> R txIndex(IndexTx<R, IBM> tx, StackBuffer stackBuffer) throws Exception {
        if (lastId > Integer.MIN_VALUE && lastId <= considerIfIndexIdGreaterThanN) {
            return tx.tx(null, null, -1, null);
        }

        return keyedFilerStore.read(indexKeyBytes, null, (monkey, filer, stackBuffer1, lock) -> {
            try {
                if (filer != null) {
                    synchronized (lock) {
                        if (filer.length() < LAST_ID_LENGTH + 4) {
                            return tx.tx(null, null, -1, null);
                        } else {
                            return tx.tx(null, filer, LAST_ID_LENGTH, stackBuffer1);
                        }
                    }
                } else {
                    return tx.tx(null, null, -1, null);
                }
            } catch (Exception e) {
                throw new IOException(e);
            }
        }, stackBuffer);
    }

    @Override
    public void replaceIndex(IBM index, int setLastId, StackBuffer stackBuffer) throws Exception {
        synchronized (mutationLock) {
            setIndex(index, setLastId, stackBuffer);
            lastId = Math.max(setLastId, lastId);
        }
    }

    private static <BM extends IBM, IBM> BitmapAndLastId<BM> deser(MiruBitmaps<BM, IBM> bitmaps,
        TrackError trackError,
        ChunkFiler filer,
        StackBuffer stackBuffer) throws IOException {

        if (filer.length() > LAST_ID_LENGTH + 4) {
            int lastId = filer.readInt();
            DataInput dataInput = new FilerDataInput(filer, stackBuffer);
            try {
                return new BitmapAndLastId<>(bitmaps.deserialize(dataInput), lastId);
            } catch (Exception e) {
                try {
                    trackError.error("Failed to deserialize a bitmap, length=" + filer.length());

                    byte[] raw = new byte[512];
                    filer.seek(0);
                    filer.read(raw);
                    log.error("Failed to deserialize, head bytes from filer: {}", Arrays.toString(raw));
                } catch (Exception e1) {
                    log.error("Failed to print debug info", e1);
                }
                throw new IOException("Failed to deserialize", e);
            }
        }
        return null;
    }

    private IBM getOrCreateIndex(StackBuffer stackBuffer) throws Exception {
        Optional<IBM> index = getIndex(stackBuffer);
        return index.isPresent() ? index.get() : bitmaps.create();
    }

    private static <BM extends IBM, IBM> long serializedSizeInBytes(MiruBitmaps<BM, IBM> bitmaps, IBM index) {
        return LAST_ID_LENGTH + bitmaps.serializedSizeInBytes(index);
    }

    private void setIndex(IBM index, int setLastId, StackBuffer stackBuffer) throws Exception {
        long filerSizeInBytes = serializedSizeInBytes(bitmaps, index);
        ByteArrayDataOutput dataOutput = ByteStreams.newDataOutput((int) filerSizeInBytes);
        dataOutput.write(FilerIO.intBytes(setLastId));
        bitmaps.serialize(index, dataOutput);
        final byte[] bytes = dataOutput.toByteArray();

        BM check = bitmaps.deserialize(ByteStreams.newDataInput(bytes));
        BM checked = bitmaps.or(Arrays.asList(index, check));
        if (bitmaps.cardinality(index) != bitmaps.cardinality(check) || bitmaps.cardinality(index) != bitmaps.cardinality(checked)) {
            log.error("BAD BITS index:{} check:{} checked:{}", index, check, checked);
            trackError.error("Bad bits" +
                " index:" + bitmaps.cardinality(index) +
                " check:" + bitmaps.cardinality(check) +
                " checked:" + bitmaps.cardinality(checked));
        }

        keyedFilerStore.writeNewReplace(indexKeyBytes, filerSizeInBytes, new SetTransaction(bytes), stackBuffer);
        log.inc("set>total");
        log.inc("set>bytes", bytes.length);
    }

    @Override
    public Optional<IBM> getIndexUnsafe(StackBuffer stackBuffer) throws Exception {
        return getIndex(stackBuffer);
    }
<commit_msg>also guard against deserialization failure in term index consistency check<commit_after>package com.jivesoftware.os.miru.service.index.filer;

import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.io.ByteArrayDataOutput;
import com.google.common.io.ByteStreams;
import com.jivesoftware.os.filer.io.Filer;
import com.jivesoftware.os.filer.io.FilerDataInput;
import com.jivesoftware.os.filer.io.FilerIO;
import com.jivesoftware.os.filer.io.api.ChunkTransaction;
import com.jivesoftware.os.filer.io.api.KeyedFilerStore;
import com.jivesoftware.os.filer.io.api.StackBuffer;
import com.jivesoftware.os.filer.io.chunk.ChunkFiler;
import com.jivesoftware.os.miru.plugin.bitmap.MiruBitmaps;
import com.jivesoftware.os.miru.plugin.index.MiruInvertedIndex;
import com.jivesoftware.os.miru.service.index.BitmapAndLastId;
import com.jivesoftware.os.miru.service.partition.TrackError;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.io.DataInput;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * @author jonathan
 */
public class MiruFilerInvertedIndex<BM extends IBM, IBM> implements MiruInvertedIndex<IBM> {

    private static final MetricLogger log = MetricLoggerFactory.getLogger();

    private static final int LAST_ID_LENGTH = 4;

    private final MiruBitmaps<BM, IBM> bitmaps;
    private final TrackError trackError;
    private final byte[] indexKeyBytes;
    private final KeyedFilerStore<Long, Void> keyedFilerStore;
    private final int considerIfIndexIdGreaterThanN;
    private final Object mutationLock;
    private volatile int lastId = Integer.MIN_VALUE;

    private final ChunkTransaction<Void, BitmapAndLastId<BM>> getTransaction;

    public MiruFilerInvertedIndex(MiruBitmaps<BM, IBM> bitmaps,
        TrackError trackError,
        byte[] indexKeyBytes,
        KeyedFilerStore<Long, Void> keyedFilerStore,
        int considerIfIndexIdGreaterThanN,
        Object mutationLock) {
        this.bitmaps = bitmaps;
        this.trackError = trackError;
        this.indexKeyBytes = Preconditions.checkNotNull(indexKeyBytes);
        this.keyedFilerStore = Preconditions.checkNotNull(keyedFilerStore);
        this.considerIfIndexIdGreaterThanN = considerIfIndexIdGreaterThanN;
        this.mutationLock = mutationLock;
        //TODO pass in
        this.getTransaction = (monkey, filer, stackBuffer, lock) -> {
            if (filer != null) {
                synchronized (lock) {
                    filer.seek(0);
                    return deser(bitmaps, trackError, filer, stackBuffer);
                }
            }
            return null;
        };
    }

    @Override
    public Optional<IBM> getIndex(StackBuffer stackBuffer) throws Exception {
        if (lastId > Integer.MIN_VALUE && lastId <= considerIfIndexIdGreaterThanN) {
            return Optional.absent();
        }

        BitmapAndLastId<BM> bitmapAndLastId = keyedFilerStore.read(indexKeyBytes, null, getTransaction, stackBuffer);

        if (bitmapAndLastId != null) {
            log.inc("get>hit");
            if (lastId == Integer.MIN_VALUE) {
                lastId = bitmapAndLastId.lastId;
            }
            return Optional.of(bitmapAndLastId.bitmap);
        } else {
            log.inc("get>miss");
            lastId = -1;
            return Optional.absent();
        }
    }

    @Override
    public <R> R txIndex(IndexTx<R, IBM> tx, StackBuffer stackBuffer) throws Exception {
        if (lastId > Integer.MIN_VALUE && lastId <= considerIfIndexIdGreaterThanN) {
            return tx.tx(null, null, -1, null);
        }

        return keyedFilerStore.read(indexKeyBytes, null, (monkey, filer, stackBuffer1, lock) -> {
            try {
                if (filer != null) {
                    synchronized (lock) {
                        if (filer.length() < LAST_ID_LENGTH + 4) {
                            return tx.tx(null, null, -1, null);
                        } else {
                            return tx.tx(null, filer, LAST_ID_LENGTH, stackBuffer1);
                        }
                    }
                } else {
                    return tx.tx(null, null, -1, null);
                }
            } catch (Exception e) {
                throw new IOException(e);
            }
        }, stackBuffer);
    }

    @Override
    public void replaceIndex(IBM index, int setLastId, StackBuffer stackBuffer) throws Exception {
        synchronized (mutationLock) {
            setIndex(index, setLastId, stackBuffer);
            lastId = Math.max(setLastId, lastId);
        }
    }

    private static <BM extends IBM, IBM> BitmapAndLastId<BM> deser(MiruBitmaps<BM, IBM> bitmaps,
        TrackError trackError,
        ChunkFiler filer,
        StackBuffer stackBuffer) throws IOException {

        if (filer.length() > LAST_ID_LENGTH + 4) {
            int lastId = filer.readInt();
            DataInput dataInput = new FilerDataInput(filer, stackBuffer);
            try {
                return new BitmapAndLastId<>(bitmaps.deserialize(dataInput), lastId);
            } catch (Exception e) {
                try {
                    trackError.error("Failed to deserialize a bitmap, length=" + filer.length());

                    byte[] raw = new byte[512];
                    filer.seek(0);
                    filer.read(raw);
                    log.error("Failed to deserialize, head bytes from filer: {}", Arrays.toString(raw));
                } catch (Exception e1) {
                    log.error("Failed to print debug info", e1);
                }
                throw new IOException("Failed to deserialize", e);
            }
        }
        return null;
    }

    private IBM getOrCreateIndex(StackBuffer stackBuffer) throws Exception {
        Optional<IBM> index = getIndex(stackBuffer);
        return index.isPresent() ? index.get() : bitmaps.create();
    }

    private static <BM extends IBM, IBM> long serializedSizeInBytes(MiruBitmaps<BM, IBM> bitmaps, IBM index) {
        return LAST_ID_LENGTH + bitmaps.serializedSizeInBytes(index);
    }

    private void setIndex(IBM index, int setLastId, StackBuffer stackBuffer) throws Exception {
        long filerSizeInBytes = serializedSizeInBytes(bitmaps, index);
        ByteArrayDataOutput dataOutput = ByteStreams.newDataOutput((int) filerSizeInBytes);
        dataOutput.write(FilerIO.intBytes(setLastId));
        bitmaps.serialize(index, dataOutput);
        final byte[] bytes = dataOutput.toByteArray();

        try {
            BM check = bitmaps.deserialize(ByteStreams.newDataInput(bytes));
            BM checked = bitmaps.or(Arrays.asList(index, check));
            if (bitmaps.cardinality(index) != bitmaps.cardinality(check) || bitmaps.cardinality(index) != bitmaps.cardinality(checked)) {
                log.error("BAD BITS bitmap mismatch index:{} check:{} checked:{}", index, check, checked);
                trackError.error("Bad bits bitmap mismatch" +
                    " index:" + bitmaps.cardinality(index) +
                    " check:" + bitmaps.cardinality(check) +
                    " checked:" + bitmaps.cardinality(checked));
            }
        } catch (Exception e) {
            log.error("BAD BITS failed to check index:{} bytes:{}", new Object[] { index, Arrays.toString(bytes) }, e);
            trackError.error("Bad bits failed to check index:" + bitmaps.cardinality(index) + " bytes:" + bytes.length);
        }

        keyedFilerStore.writeNewReplace(indexKeyBytes, filerSizeInBytes, new SetTransaction(bytes), stackBuffer);
        log.inc("set>total");
        log.inc("set>bytes", bytes.length);
    }

    @Override
    public Optional<IBM> getIndexUnsafe(StackBuffer stackBuffer) throws Exception {
        return getIndex(stackBuffer);
    }

{'fraction_non_alphanumeric': 0.10057284207785444, 'fraction_numerical': 0.0020830621012888945, 'mean_word_length': 2.449259092950157, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 61, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '836000', 'n_tokens_mistral': 4456, 'n_tokens_neox': 4373, 'n_words': 924}